diff --git a/cmd/pd-server/main.go b/cmd/pd-server/main.go index 0cb8466..a6c13ef 100644 --- a/cmd/pd-server/main.go +++ b/cmd/pd-server/main.go @@ -10,6 +10,7 @@ import ( "github.com/b3fs/wfs/pkg/logutil" "github.com/b3fs/wfs/server" + "go.uber.org/zap" ) func main() { @@ -21,6 +22,25 @@ func main() { os.Exit(0) } + defer logutil.LogPanic() + + switch errors.Cause(err) { + case nil: + case flag.ErrHelp: + exit(0) + default: + log.Fatal("parse cmd flags error", zap.Error(err)) + } + // New zap logger + err = cfg.SetupLogger() + if err == nil { + log.ReplaceGlobals(cfg.GetZapLogger(), cfg.GetZapLogProperties()) + } else { + log.Fatal("initialize logger error", zap.Error(err)) + } + // Flushing any buffered log entries + defer log.Sync() + err = logutil.InitLogger(&cfg.Log) if err != nil { log.Fatalf("initialize logger error: %s\n", fmt.Sprintf("%+v", err)) diff --git a/go.mod b/go.mod index 8928bcd..1718d3c 100644 --- a/go.mod +++ b/go.mod @@ -2,7 +2,7 @@ module github.com/b3fs/wfs require ( github.com/BurntSushi/toml v0.3.1 - github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973 // indirect + github.com/boltdb/bolt v1.3.1 // indirect github.com/coreos/bbolt v1.3.0 // indirect github.com/coreos/etcd v3.3.9+incompatible github.com/coreos/go-semver v0.2.0 @@ -11,30 +11,32 @@ require ( github.com/dgrijalva/jwt-go v3.2.0+incompatible // indirect github.com/dustin/go-humanize v0.0.0-20180713052910-9f541cc9db5d // indirect github.com/ghodss/yaml v1.0.0 // indirect - github.com/gogo/protobuf v1.1.1 // indirect + github.com/gogo/protobuf v1.1.1 + github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef // indirect github.com/golang/protobuf v1.2.0 - github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c // indirect + github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c github.com/gorilla/websocket v1.4.0 // indirect github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect github.com/grpc-ecosystem/grpc-gateway v1.5.1 // indirect - github.com/hawkingrei/wfs v0.0.0-20181005145148-b215acd11928 // indirect github.com/jonboulle/clockwork v0.1.0 // indirect - github.com/juju/errors v0.0.0-20180806074554-22422dad46e1 // indirect - github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect + github.com/juju/ratelimit v1.0.1 + github.com/montanaflynn/stats v0.5.0 // indirect + github.com/pingcap/check v0.0.0-20190102082844-67f458068fc8 github.com/pingcap/kvproto v0.0.0-20180930052200-fae11119f066 - github.com/pingcap/pd v2.0.5+incompatible - github.com/pkg/errors v0.8.0 - github.com/prometheus/client_golang v0.8.0 // indirect - github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910 // indirect - github.com/prometheus/common v0.0.0-20180801064454-c7de2306084e // indirect - github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d // indirect + github.com/pingcap/log v0.0.0-20190307075452-bd41d9273596 + github.com/pingcap/pd v2.1.8+incompatible + github.com/pkg/errors v0.8.1 + github.com/prometheus/client_golang v0.9.2 // indirect github.com/sirupsen/logrus v1.1.0 github.com/soheilhy/cmux v0.1.4 // indirect + github.com/syndtr/goleveldb v1.0.0 github.com/tmc/grpc-websocket-proxy v0.0.0-20171017195756-830351dc03c6 // indirect - github.com/ugorji/go/codec v0.0.0-20180927125128-99ea80c8b19a // indirect - github.com/xiang90/probing v0.0.0-20160813154853-07dd2e8dfe18 // indirect - golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2 // indirect + github.com/ugorji/go v1.1.4 // indirect + github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2 // indirect + go.etcd.io/etcd v3.3.12+incompatible + go.uber.org/atomic v1.4.0 // indirect + go.uber.org/zap v1.10.0 + golang.org/x/time v0.0.0-20190308202827-9d24e82272b4 // indirect google.golang.org/grpc v1.15.0 gopkg.in/natefinch/lumberjack.v2 v2.0.0 - gopkg.in/yaml.v2 v2.2.1 // indirect ) diff --git a/go.sum b/go.sum index 7a948cf..2466bc2 100644 --- a/go.sum +++ b/go.sum @@ -3,98 +3,207 @@ github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973 h1:xJ4a3vCFaGF/jqvzLMYoU8P317H5OQ+Via4RmuPwCS0= github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q= +github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs= +github.com/boltdb/bolt v1.3.1 h1:JQmyP4ZBrce+ZQu0dY660FMfatumYDLun9hBCUVIkF4= +github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps= github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw= github.com/coreos/bbolt v1.3.0 h1:HIgH5xUWXT914HCI671AxuTTqjj64UOFr7pHn48LUTI= github.com/coreos/bbolt v1.3.0/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk= -github.com/coreos/etcd v3.3.9+incompatible h1:/pWnp1yEff0z+vBEOBFLZZ22Ux5xoVozEe7X0VFyRNo= +github.com/coreos/etcd v3.3.9+incompatible h1:iKSVPXGNGqroBx4+RmUXv8emeU7y+ucRZSzTYgzLZwM= github.com/coreos/etcd v3.3.9+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE= github.com/coreos/go-semver v0.2.0 h1:3Jm3tLmsgAYcjC+4Up7hJrFBPr+n7rAqYeSw/SZazuY= github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk= +github.com/coreos/go-systemd v0.0.0-20180511133405-39ca1b05acc7/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4= github.com/coreos/go-systemd v0.0.0-20181004131557-b14d3eb023cc h1:EDHXXD9U9N6r96eKHqu2Q4Zstkhlyv7RGN4Gwew+jVk= github.com/coreos/go-systemd v0.0.0-20181004131557-b14d3eb023cc/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4= +github.com/coreos/pkg v0.0.0-20160727233714-3ac0863d7acf/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA= github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f h1:lBNOc5arjvs8E5mO2tbpBpLoyyu8B6e44T7hJy6potg= github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumCAMpl/TFQ4/5kLM= github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ= +github.com/dustin/go-humanize v0.0.0-20171111073723-bb3d318650d4/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk= github.com/dustin/go-humanize v0.0.0-20180713052910-9f541cc9db5d h1:lDrio3iIdNb0Gw9CgH7cQF+iuB5mOOjdJ9ERNJCBgb4= github.com/dustin/go-humanize v0.0.0-20180713052910-9f541cc9db5d/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk= +github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4= +github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I= +github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk= github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04= +github.com/ghodss/yaml v1.0.1-0.20190212211648-25d852aebe32 h1:Mn26/9ZMNWSw9C9ERFA1PUxfmGpolnw2v0bKOREu5ew= +github.com/ghodss/yaml v1.0.1-0.20190212211648-25d852aebe32/go.mod h1:GIjDIg/heH5DOkXY3YJ/wNhfHsQHoXGjl8G8amsYQ1I= +github.com/gogo/protobuf v1.0.0/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ= github.com/gogo/protobuf v1.1.1 h1:72R+M5VuhED/KujmZVcIquuo8mBgX4oVda//DQb3PXo= github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ= +github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b h1:VKtxabqXZkF25pY9ekfRL6a582T4P37/31XEstQ5p58= github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= +github.com/golang/groupcache v0.0.0-20160516000752-02826c3e7903/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= +github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef h1:veQD95Isof8w9/WXiA+pa3tz3fJXkt5B7QaRBrM62gk= +github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= github.com/golang/lint v0.0.0-20180702182130-06c8688daad7/go.mod h1:tluoj9z5200jBnyusfRPU2LqT6J+DAorxEvtC7LHB+E= github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A= github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM= github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= +github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db h1:woRePGFeVFfLKN/pOkfl+p/TAqKOfFu+7KPlMVpok/w= +github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= +github.com/google/btree v0.0.0-20180124185431-e89373fe6b4a/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ= github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c h1:964Od4U6p2jUkFxvCydnIczKteheJEzHRToSGK3Bnlw= github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ= +github.com/google/uuid v1.0.0 h1:b4Gk+7WdP/d3HZH8EJsZpvV7EtDOgaZLtnaNGIu1adA= +github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gorilla/websocket v0.0.0-20170926233335-4201258b820c/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ= github.com/gorilla/websocket v1.4.0 h1:WDFjx/TMzVgy9VdMMQi2K2Emtwi2QcUQsztZ/zLaH/Q= github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ= +github.com/grpc-ecosystem/go-grpc-middleware v1.0.1-0.20190118093823-f849b5445de4 h1:z53tR0945TRRQO/fLEVPI6SMv7ZflF0TEaTAoU7tOzg= +github.com/grpc-ecosystem/go-grpc-middleware v1.0.1-0.20190118093823-f849b5445de4/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs= github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 h1:Ovs26xHkKqVztRpIrF/92BcuyuQ/YW4NSIpoGtfXNho= github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk= +github.com/grpc-ecosystem/grpc-gateway v1.4.1/go.mod h1:RSKVYQBd5MCa4OVpNdGskqpgL2+G+NZTnrVHpWWfpdw= github.com/grpc-ecosystem/grpc-gateway v1.5.1 h1:3scN4iuXkNOyP98jF55Lv8a9j1o/IwvnDIZ0LHJK1nk= github.com/grpc-ecosystem/grpc-gateway v1.5.1/go.mod h1:RSKVYQBd5MCa4OVpNdGskqpgL2+G+NZTnrVHpWWfpdw= -github.com/hawkingrei/wfs v0.0.0-20181005145148-b215acd11928 h1:YHy2PixxnEg+baBFL12oxskKTEDUCPH5sfO/rXWdvIo= -github.com/hawkingrei/wfs v0.0.0-20181005145148-b215acd11928/go.mod h1:3+XccHsdw6AWuOOZ6+QB9snpYiqY6EbqGejZzDtYJow= +github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI= +github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= +github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8= github.com/jonboulle/clockwork v0.1.0 h1:VKV+ZcuP6l3yW9doeqz6ziZGgcynBVQO+obU0+0hcPo= github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo= -github.com/juju/errors v0.0.0-20180806074554-22422dad46e1 h1:wnhMXidtb70kDZCeLt/EfsVtkXS5c8zLnE9y/6DIRAU= -github.com/juju/errors v0.0.0-20180806074554-22422dad46e1/go.mod h1:W54LbzXuIE0boCoNJfwqpmkKJ1O4TCTZMetAt6jGk7Q= +github.com/juju/ratelimit v1.0.1 h1:+7AIFJVQ0EQgq/K9+0Krm7m530Du7tIz0METWzN0RgY= +github.com/juju/ratelimit v1.0.1/go.mod h1:qapgC/Gy+xNh9UxzV13HGGl/6UXNN+ct+vwSgWNm/qk= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/konsorten/go-windows-terminal-sequences v0.0.0-20180402223658-b729f2633dfe h1:CHRGQ8V7OlCYtwaKPJi3iA7J+YdNKdo8j7nG5IgDhjs= github.com/konsorten/go-windows-terminal-sequences v0.0.0-20180402223658-b729f2633dfe/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= +github.com/kr/pty v1.0.0/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU= +github.com/mattn/go-isatty v0.0.4/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4= +github.com/mattn/go-runewidth v0.0.2/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU= +github.com/matttproud/golang_protobuf_extensions v1.0.0/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU= github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= -github.com/pingcap/kvproto v0.0.0-20180930052200-fae11119f066 h1:ulo0ph8sxCzY3GYI+5wh2x3h/gTO3BeIvCFy6U/EKks= +github.com/montanaflynn/stats v0.5.0 h1:2EkzeTSqBB4V4bJwWrt5gIIrZmpJBcoIRGS2kWLgzmk= +github.com/montanaflynn/stats v0.5.0/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc= +github.com/olekukonko/tablewriter v0.0.0-20170122224234-a0225b3f23b5/go.mod h1:vsDQFd/mU46D+Z4whnwzcISnGGzXWMclvtLoiIKAKIo= +github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= +github.com/onsi/ginkgo v1.7.0 h1:WSHQ+IS43OoUrWtD1/bbclrwK8TTH5hzp+umCiuxHgs= +github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= +github.com/onsi/gomega v1.4.2/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY= +github.com/onsi/gomega v1.4.3 h1:RE1xgDvH7imwFD45h+u2SgIfERHlS2yNG4DObb5BSKU= +github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY= +github.com/pingcap/check v0.0.0-20190102082844-67f458068fc8 h1:USx2/E1bX46VG32FIw034Au6seQ2fY9NEILmNh/UlQg= +github.com/pingcap/check v0.0.0-20190102082844-67f458068fc8/go.mod h1:B1+S9LNcuMyLH/4HMTViQOJevkGiik3wW2AN9zb2fNQ= +github.com/pingcap/errors v0.11.0 h1:DCJQB8jrHbQ1VVlMFIrbj2ApScNNotVmkSNplu2yUt4= +github.com/pingcap/errors v0.11.0/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8= +github.com/pingcap/kvproto v0.0.0-20180930052200-fae11119f066 h1:WsYymkD98E5CAiROBH1gH20xFcfgwTLtbA/wcZGGIOs= github.com/pingcap/kvproto v0.0.0-20180930052200-fae11119f066/go.mod h1:0gwbe1F2iBIjuQ9AH0DbQhL+Dpr5GofU8fgYyXk+ykk= -github.com/pingcap/pd v2.0.5+incompatible h1:nUTetzzM/tLbd03DoMB76AAvI57oYftNib9sUE3KPFk= -github.com/pingcap/pd v2.0.5+incompatible/go.mod h1:nD3+EoYes4+aNNODO99ES59V83MZSI+dFbhyr667a0E= -github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw= +github.com/pingcap/log v0.0.0-20190307075452-bd41d9273596 h1:t2OQTpPJnrPDGlvA+3FwJptMTt6MEPdzK1Wt99oaefQ= +github.com/pingcap/log v0.0.0-20190307075452-bd41d9273596/go.mod h1:WpHUKhNZ18v116SvGrmjkA9CBhYmuUTKL+p8JC9ANEw= +github.com/pingcap/pd v2.1.8+incompatible h1:nXOUkTCY3knCSyZoic9huHQj/5N9Z6gnyZ4ma3yjQ7I= +github.com/pingcap/pd v2.1.8+incompatible/go.mod h1:nD3+EoYes4+aNNODO99ES59V83MZSI+dFbhyr667a0E= github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I= +github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/prometheus/client_golang v0.8.0 h1:1921Yw9Gc3iSc4VQh3PIoOqgPCZS7G/4xQNVUp8Mda8= github.com/prometheus/client_golang v0.8.0/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw= +github.com/prometheus/client_golang v0.9.2 h1:awm861/B8OKDd2I/6o1dy3ra4BamzKhYOiGItCeZ740= +github.com/prometheus/client_golang v0.9.2/go.mod h1:OsXs2jCmiKlQ1lTBmv21f2mNfw4xf/QclQDMrYNZzcM= +github.com/prometheus/client_model v0.0.0-20170216185247-6f3806018612/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo= github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910 h1:idejC8f05m9MGOsuEi1ATq9shN03HrxNkD/luQvxCv8= github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo= +github.com/prometheus/common v0.0.0-20180518154759-7600349dcfe1/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro= github.com/prometheus/common v0.0.0-20180801064454-c7de2306084e h1:n/3MEhJQjQxrOUCzh1Y3Re6aJUUWRp2M9+Oc3eVn/54= github.com/prometheus/common v0.0.0-20180801064454-c7de2306084e/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro= +github.com/prometheus/common v0.0.0-20181126121408-4724e9255275 h1:PnBWHBf+6L0jOqq0gIVUe6Yk0/QMZ640k6NvkxcBf+8= +github.com/prometheus/common v0.0.0-20181126121408-4724e9255275/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro= +github.com/prometheus/procfs v0.0.0-20180612222113-7d6f385de8be/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk= github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d h1:GoAlyOgbOEIFdaDqxJVlbOQ1DtGmZWs/Qau0hIlk+WQ= github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk= +github.com/prometheus/procfs v0.0.0-20181204211112-1dc9a6cbc91a h1:9a8MnZMP0X2nLJdBg+pBmGgkJlSaKC2KaQmTCk1XDtE= +github.com/prometheus/procfs v0.0.0-20181204211112-1dc9a6cbc91a/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk= +github.com/sirupsen/logrus v1.0.5/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc= github.com/sirupsen/logrus v1.1.0 h1:65VZabgUiV9ktjGM5nTq0+YurgTyX+YI2lSSfDjI+qU= github.com/sirupsen/logrus v1.1.0/go.mod h1:zrgwTnHtNr00buQ1vSptGe8m1f/BbgsPukg8qsT7A+A= github.com/soheilhy/cmux v0.1.4 h1:0HKaf1o97UwFjHH9o5XsHUOF+tqmdA7KEzXLpiyaw0E= github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM= +github.com/spf13/cobra v0.0.3/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ= +github.com/spf13/pflag v1.0.1 h1:aCvUg6QPl3ibpQUxyLkrEkCHtPqYJL4x9AuhqVqFis4= +github.com/spf13/pflag v1.0.1/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= +github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/syndtr/goleveldb v1.0.0 h1:fBdIW9lB4Iz0n9khmH8w27SJ3QEJ7+IgjPEwGSZiFdE= +github.com/syndtr/goleveldb v1.0.0/go.mod h1:ZVVdQEZoIme9iO1Ch2Jdy24qqXrMMOU6lpPAyBWyWuQ= +github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U= github.com/tmc/grpc-websocket-proxy v0.0.0-20171017195756-830351dc03c6 h1:lYIiVDtZnyTWlNwiAxLj0bbpTcx1BWCFhXjfsvmPdNc= github.com/tmc/grpc-websocket-proxy v0.0.0-20171017195756-830351dc03c6/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U= -github.com/ugorji/go/codec v0.0.0-20180927125128-99ea80c8b19a h1:BgdofUvNP/srMxiUUpGyZm+WjX/qXpMXdl3edRf1Ta0= -github.com/ugorji/go/codec v0.0.0-20180927125128-99ea80c8b19a/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0= -github.com/xiang90/probing v0.0.0-20160813154853-07dd2e8dfe18 h1:MPPkRncZLN9Kh4MEFmbnK4h3BD7AUmskWv2+EeZJCCs= -github.com/xiang90/probing v0.0.0-20160813154853-07dd2e8dfe18/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU= +github.com/ugorji/go v1.1.2 h1:JON3E2/GPW2iDNGoSAusl1KDf5TRQ8k8q7Tp097pZGs= +github.com/ugorji/go v1.1.2/go.mod h1:hnLbHMwcvSihnDhEfx2/BzKp2xb0Y+ErdfYcrs9tkJQ= +github.com/ugorji/go v1.1.4 h1:j4s+tAvLfL3bZyefP2SEWmhBzmuIlH/eqNuPdFPgngw= +github.com/ugorji/go v1.1.4/go.mod h1:uQMGLiO92mf5W77hV/PUCpI3pbzQx3CRekS0kk+RGrc= +github.com/ugorji/go/codec v0.0.0-20190204201341-e444a5086c43 h1:BasDe+IErOQKrMVXab7UayvSlIpiyGwRvuX3EKYY7UA= +github.com/ugorji/go/codec v0.0.0-20190204201341-e444a5086c43/go.mod h1:iT03XoTwV7xq/+UGwKO3UbC1nNNlopQiY61beSdrtOA= +github.com/urfave/cli v1.20.0/go.mod h1:70zkFmudgCuE/ngEzBv17Jvp/497gISqfk5gWijbERA= +github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2 h1:eY9dn8+vbi4tKz5Qo6v2eYzo7kUS51QINcR5jNpbZS8= +github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU= +go.etcd.io/bbolt v1.3.2 h1:Z/90sZLPOeCy2PwprqkFa25PdkusRzaj9P8zm/KNyvk= +go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU= +go.etcd.io/etcd v0.0.0-20190320044326-77d4b742cdbf h1:rmttwKPEgG/l4UscTDYtaJgeUsedKPKSyFfNQLI6q+I= +go.etcd.io/etcd v0.0.0-20190320044326-77d4b742cdbf/go.mod h1:KSGwdbiFchh5KIC9My2+ZVl5/3ANcwohw50dpPwa2cw= +go.etcd.io/etcd v3.3.12+incompatible h1:V6PRYRGpU4k5EajJaaj/GL3hqIdzyPnBU8aPUp+35yw= +go.etcd.io/etcd v3.3.12+incompatible/go.mod h1:yaeTdrJi5lOmYerz05bd8+V7KubZs8YSFZfzsF9A6aI= +go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= +go.uber.org/atomic v1.4.0 h1:cxzIVoETapQEqDhQu3QfnvXAV4AlzcvUCxkVUFw3+EU= +go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= +go.uber.org/multierr v1.1.0 h1:HoEmRHQPVSqub6w2z2d2EOVs2fjyFRGyofhKuyDq0QI= +go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0= +go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q= +go.uber.org/zap v1.10.0 h1:ORx85nbTijNz8ljznvCMR1ZBIPKFn3jQrag10X2AsuM= +go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q= +golang.org/x/crypto v0.0.0-20180608092829-8ac0e0d97ce4/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20180904163835-0709b304e793 h1:u+LnwYTOOW7Ukr/fppxEb1Nwz0AtPflrblfvUudpo+I= golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/lint v0.0.0-20180702182130-06c8688daad7/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE= -golang.org/x/net v0.0.0-20180826012351-8a410e7b638d h1:g9qWBGx4puODJTMVyoPrpoxPFgVGd+z1DZwjfRu4d0I= golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= +golang.org/x/net v0.0.0-20180906233101-161cd47e91fd h1:nTDtHvHSdCn1m6ITfMRqtOd/9+7a3s8RBNOZ3eYZzJA= +golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= +golang.org/x/net v0.0.0-20181201002055-351d144fa1fc h1:a3CU5tJYVj92DY2LaA1kUkrsqD5/3mLDhx2NcNqyW+0= +golang.org/x/net v0.0.0-20181201002055-351d144fa1fc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= +golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f h1:wMNYb4v58l5UBM7MYRLPG6ZhfOqbKu7X5eyFl8ZhKvA= golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= -golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33 h1:I6FyU15t786LL7oL/hn43zqTuEGr4PN7F4XJ1p4E3Y8= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e h1:o3PsSEY8E4eXWkXrIP9YJALUkVZqzHJT5DOasTyn8Vs= +golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2 h1:+DCIGbF/swA92ohVg0//6X2IVY3KZs6p9mix0ziNYJM= golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= +golang.org/x/time v0.0.0-20190308202827-9d24e82272b4 h1:SvFZT6jyqRaOeXpc5h/JSfZenJ2O330aBsf7JfSUXmQ= +golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/tools v0.0.0-20180828015842-6cd1fcedba52/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM= +google.golang.org/genproto v0.0.0-20180608181217-32ee49c4dd80/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc= google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8 h1:Nw54tB0rB7hY/N0NQvRW8DG4Yk3Q6T9cu9RcFQDu1tc= google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc= +google.golang.org/grpc v1.14.0/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw= google.golang.org/grpc v1.15.0 h1:Az/KuahOM4NAidTEuJCv/RonAA7rYsTPkqXVjr+8OOw= google.golang.org/grpc v1.15.0/go.mod h1:0JHn/cJsOMiMfNA9+DeHDlAU7KAAB5GDlYFpa9MZMio= +gopkg.in/airbrake/gobrake.v2 v2.0.9/go.mod h1:/h5ZAUhDkGaJfjzjKLSjv6zCL6O0LLBxU4K+aSYdM/U= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/cheggaaa/pb.v1 v1.0.25/go.mod h1:V/YB90LKu/1FcN3WVnfiiE5oMCibMjukxqG/qStrOgw= +gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4= +gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys= +gopkg.in/gemnasium/logrus-airbrake-hook.v2 v2.1.2/go.mod h1:Xk6kEKp8OKb+X14hQBKWaSkCsqBpgog8nAV2xsGOxlo= gopkg.in/natefinch/lumberjack.v2 v2.0.0 h1:1Lc07Kr7qY4U2YPouBjpCLxpiyxIVoxqXgkXLknAOE8= gopkg.in/natefinch/lumberjack.v2 v2.0.0/go.mod h1:l0ndWWf7gzL7RNwBG7wST/UCcT4T24xpD6X8LsfU/+k= -gopkg.in/yaml.v2 v2.2.1 h1:mUhvW9EsL+naU5Q3cakzfE91YhliOondGd6ZrsDBHQE= +gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ= +gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw= gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw= +gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= honnef.co/go/tools v0.0.0-20180728063816-88497007e858/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= diff --git a/pkg/logutil/log.go b/pkg/logutil/log.go index 5ae1bf9..2013ebb 100644 --- a/pkg/logutil/log.go +++ b/pkg/logutil/log.go @@ -10,7 +10,6 @@ // distributed under the License is distributed on an "AS IS" BASIS, // See the License for the specific language governing permissions and // limitations under the License. - package logutil import ( @@ -19,14 +18,16 @@ import ( "os" "path" "runtime" - "runtime/debug" "strings" "sync" - "github.com/coreos/etcd/raft" "github.com/coreos/pkg/capnslog" + zaplog "github.com/pingcap/log" "github.com/pkg/errors" log "github.com/sirupsen/logrus" + "go.etcd.io/etcd/raft" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" "google.golang.org/grpc/grpclog" lumberjack "gopkg.in/natefinch/lumberjack.v2" ) @@ -144,6 +145,23 @@ func StringToLogLevel(level string) log.Level { return defaultLogLevel } +// StringToZapLogLevel translates log level string to log level. +func StringToZapLogLevel(level string) zapcore.Level { + switch strings.ToLower(level) { + case "fatal": + return zapcore.FatalLevel + case "error": + return zapcore.ErrorLevel + case "warn", "warning": + return zapcore.WarnLevel + case "debug": + return zapcore.DebugLevel + case "info": + return zapcore.InfoLevel + } + return zapcore.InfoLevel +} + // textFormatter is for compatibility with ngaut/log type textFormatter struct { DisableTimestamp bool @@ -173,7 +191,8 @@ func (f *textFormatter) Format(entry *log.Entry) ([]byte, error) { return b.Bytes(), nil } -func stringToLogFormatter(format string, disableTimestamp bool) log.Formatter { +// StringToLogFormatter uses the different log formatter according to a given format name. +func StringToLogFormatter(format string, disableTimestamp bool) log.Formatter { switch strings.ToLower(format) { case "text": return &textFormatter{ @@ -196,7 +215,7 @@ func stringToLogFormatter(format string, disableTimestamp bool) log.Formatter { } // InitFileLog initializes file based logging options. -func InitFileLog(cfg *FileLogConfig) error { +func InitFileLog(cfg *zaplog.FileLogConfig) error { if st, err := os.Stat(cfg.Filename); err == nil { if st.IsDir() { return errors.New("can't use directory as log file name") @@ -238,7 +257,7 @@ func (lg *wrapLogrus) V(l int) bool { var once sync.Once // InitLogger initializes PD's logger. -func InitLogger(cfg *LogConfig) error { +func InitLogger(cfg *zaplog.Config) error { var err error once.Do(func() { @@ -248,7 +267,7 @@ func InitLogger(cfg *LogConfig) error { if cfg.Format == "" { cfg.Format = defaultLogFormat } - log.SetFormatter(stringToLogFormatter(cfg.Format, cfg.DisableTimestamp)) + log.SetFormatter(StringToLogFormatter(cfg.Format, cfg.DisableTimestamp)) // etcd log capnslog.SetFormatter(&redirectFormatter{}) @@ -271,6 +290,6 @@ func InitLogger(cfg *LogConfig) error { // Commonly used with a `defer`. func LogPanic() { if e := recover(); e != nil { - log.Fatalf("panic: %v, stack: %s", e, string(debug.Stack())) + zaplog.Fatal("panic", zap.Reflect("recover", e)) } } diff --git a/pkg/logutil/log_test.go b/pkg/logutil/log_test.go index 07b8815..7d397f5 100644 --- a/pkg/logutil/log_test.go +++ b/pkg/logutil/log_test.go @@ -20,7 +20,9 @@ import ( "github.com/coreos/pkg/capnslog" . "github.com/pingcap/check" + zaplog "github.com/pingcap/log" log "github.com/sirupsen/logrus" + "go.uber.org/zap/zapcore" ) const ( @@ -51,9 +53,35 @@ func (s *testLogSuite) TestStringToLogLevel(c *C) { c.Assert(StringToLogLevel("whatever"), Equals, log.InfoLevel) } +func (s *testLogSuite) TestStringToZapLogLevel(c *C) { + c.Assert(StringToZapLogLevel("fatal"), Equals, zapcore.FatalLevel) + c.Assert(StringToZapLogLevel("ERROR"), Equals, zapcore.ErrorLevel) + c.Assert(StringToZapLogLevel("warn"), Equals, zapcore.WarnLevel) + c.Assert(StringToZapLogLevel("warning"), Equals, zapcore.WarnLevel) + c.Assert(StringToZapLogLevel("debug"), Equals, zapcore.DebugLevel) + c.Assert(StringToZapLogLevel("info"), Equals, zapcore.InfoLevel) + c.Assert(StringToZapLogLevel("whatever"), Equals, zapcore.InfoLevel) +} + +func (s *testLogSuite) TestStringToLogFormatter(c *C) { + c.Assert(StringToLogFormatter("text", true), DeepEquals, &textFormatter{ + DisableTimestamp: true, + }) + c.Assert(StringToLogFormatter("json", true), DeepEquals, &log.JSONFormatter{ + DisableTimestamp: true, + TimestampFormat: defaultLogTimeFormat, + }) + c.Assert(StringToLogFormatter("console", true), DeepEquals, &log.TextFormatter{ + DisableTimestamp: true, + FullTimestamp: true, + TimestampFormat: defaultLogTimeFormat, + }) + c.Assert(StringToLogFormatter("", true), DeepEquals, &textFormatter{}) +} + // TestLogging assure log format and log redirection works. func (s *testLogSuite) TestLogging(c *C) { - conf := &LogConfig{Level: "warn", File: FileLogConfig{}} + conf := &zaplog.Config{Level: "warn", File: zaplog.FileLogConfig{}} c.Assert(InitLogger(conf), IsNil) log.SetOutput(s.buf) @@ -76,3 +104,8 @@ func (s *testLogSuite) TestLogging(c *C) { c.Assert(entry, Matches, logPattern) c.Assert(strings.Contains(entry, "log_test.go"), IsTrue) } + +func (s *testLogSuite) TestFileLog(c *C) { + c.Assert(InitFileLog(&zaplog.FileLogConfig{Filename: "/tmp"}), NotNil) + c.Assert(InitFileLog(&zaplog.FileLogConfig{Filename: "/tmp/test_file_log", MaxSize: 0}), IsNil) +} diff --git a/server/cluster.go b/server/cluster.go new file mode 100644 index 0000000..77df332 --- /dev/null +++ b/server/cluster.go @@ -0,0 +1,50 @@ +// Copyright 2016 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "sync" + "time" + + syncer "github.com/b3fs/wfs/server/region_syncer" +) + +var backgroundJobInterval = time.Minute + +// RaftCluster is used for cluster config management. +// Raft cluster key format: +// cluster 1 -> /1/raft, value is metapb.Cluster +// cluster 2 -> /2/raft +// For cluster 1 +// store 1 -> /1/raft/s/1, value is metapb.Store +// region 1 -> /1/raft/r/1, value is metapb.Region +type RaftCluster struct { + sync.RWMutex + + s *Server + + running bool + + clusterID uint64 + clusterRoot string + + // cached cluster info + cachedCluster *clusterInfo + + coordinator *coordinator + + wg sync.WaitGroup + quit chan struct{} + regionSyncer *syncer.RegionSyncer +} diff --git a/server/cluster_info.go b/server/cluster_info.go new file mode 100644 index 0000000..56a7ac9 --- /dev/null +++ b/server/cluster_info.go @@ -0,0 +1,831 @@ +// Copyright 2016 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "sync" + "time" + + "github.com/coreos/go-semver/semver" + "github.com/gogo/protobuf/proto" + "github.com/pingcap/kvproto/pkg/metapb" + "github.com/pingcap/kvproto/pkg/pdpb" + log "github.com/pingcap/log" + "github.com/pingcap/pd/server/core" + "github.com/pingcap/pd/server/namespace" + "github.com/pingcap/pd/server/schedule" + "go.uber.org/zap" +) + +type clusterInfo struct { + sync.RWMutex + core *schedule.BasicCluster + + id core.IDAllocator + kv *core.KV + meta *metapb.Cluster + opt *scheduleOption + regionStats *regionStatistics + labelLevelStats *labelLevelStatistics + prepareChecker *prepareChecker + changedRegions chan *core.RegionInfo +} + +var defaultChangedRegionsLimit = 10000 + +func newClusterInfo(id core.IDAllocator, opt *scheduleOption, kv *core.KV) *clusterInfo { + return &clusterInfo{ + core: schedule.NewBasicCluster(), + id: id, + opt: opt, + kv: kv, + labelLevelStats: newLabelLevelStatistics(), + prepareChecker: newPrepareChecker(), + changedRegions: make(chan *core.RegionInfo, defaultChangedRegionsLimit), + } +} + +// Return nil if cluster is not bootstrapped. +func loadClusterInfo(id core.IDAllocator, kv *core.KV, opt *scheduleOption) (*clusterInfo, error) { + c := newClusterInfo(id, opt, kv) + + c.meta = &metapb.Cluster{} + ok, err := kv.LoadMeta(c.meta) + if err != nil { + return nil, err + } + if !ok { + return nil, nil + } + + start := time.Now() + if err := kv.LoadStores(c.core.Stores); err != nil { + return nil, err + } + log.Info("load stores", + zap.Int("count", c.core.Stores.GetStoreCount()), + zap.Duration("cost", time.Since(start)), + ) + + start = time.Now() + if err := kv.LoadRegions(c.core.Regions); err != nil { + return nil, err + } + log.Info("load regions", + zap.Int("count", c.core.Regions.GetRegionCount()), + zap.Duration("cost", time.Since(start)), + ) + + return c, nil +} + +func (c *clusterInfo) OnStoreVersionChange() { + var ( + minVersion *semver.Version + clusterVersion semver.Version + ) + + clusterVersion = c.opt.loadClusterVersion() + stores := c.GetStores() + for _, s := range stores { + if s.IsTombstone() { + continue + } + v := MustParseVersion(s.GetVersion()) + + if minVersion == nil || v.LessThan(*minVersion) { + minVersion = v + } + } + // If the cluster version of PD is less than the minimum version of all stores, + // it will update the cluster version. + if clusterVersion.LessThan(*minVersion) { + c.opt.SetClusterVersion(*minVersion) + err := c.opt.persist(c.kv) + if err != nil { + log.Error("persist cluster version meet error", zap.Error(err)) + } + log.Info("cluster version changed", + zap.Stringer("old-cluster-version", clusterVersion), + zap.Stringer("new-cluster-version", minVersion)) + CheckPDVersion(c.opt) + } +} + +func (c *clusterInfo) changedRegionNotifier() <-chan *core.RegionInfo { + return c.changedRegions +} + +// IsFeatureSupported checks if the feature is supported by current cluster. +func (c *clusterInfo) IsFeatureSupported(f Feature) bool { + clusterVersion := c.opt.loadClusterVersion() + minSupportVersion := MinSupportedVersion(f) + return !clusterVersion.LessThan(minSupportVersion) +} + +func (c *clusterInfo) allocID() (uint64, error) { + return c.id.Alloc() +} + +// AllocPeer allocs a new peer on a store. +func (c *clusterInfo) AllocPeer(storeID uint64) (*metapb.Peer, error) { + peerID, err := c.allocID() + if err != nil { + log.Error("failed to alloc peer", zap.Error(err)) + return nil, err + } + peer := &metapb.Peer{ + Id: peerID, + StoreId: storeID, + } + return peer, nil +} + +func (c *clusterInfo) getClusterID() uint64 { + c.RLock() + defer c.RUnlock() + return c.meta.GetId() +} + +func (c *clusterInfo) getMeta() *metapb.Cluster { + c.RLock() + defer c.RUnlock() + return proto.Clone(c.meta).(*metapb.Cluster) +} + +func (c *clusterInfo) putMeta(meta *metapb.Cluster) error { + c.Lock() + defer c.Unlock() + return c.putMetaLocked(proto.Clone(meta).(*metapb.Cluster)) +} + +func (c *clusterInfo) putMetaLocked(meta *metapb.Cluster) error { + if c.kv != nil { + if err := c.kv.SaveMeta(meta); err != nil { + return err + } + } + c.meta = meta + return nil +} + +// GetStore searches for a store by ID. +func (c *clusterInfo) GetStore(storeID uint64) *core.StoreInfo { + c.RLock() + defer c.RUnlock() + return c.core.GetStore(storeID) +} + +func (c *clusterInfo) putStore(store *core.StoreInfo) error { + c.Lock() + defer c.Unlock() + return c.putStoreLocked(store) +} + +func (c *clusterInfo) putStoreLocked(store *core.StoreInfo) error { + if c.kv != nil { + if err := c.kv.SaveStore(store.GetMeta()); err != nil { + return err + } + } + c.core.PutStore(store) + return nil +} + +func (c *clusterInfo) deleteStore(store *core.StoreInfo) error { + c.Lock() + defer c.Unlock() + return c.deleteStoreLocked(store) +} + +func (c *clusterInfo) deleteStoreLocked(store *core.StoreInfo) error { + if c.kv != nil { + if err := c.kv.DeleteStore(store.GetMeta()); err != nil { + return err + } + } + c.core.DeleteStore(store) + return nil +} + +// BlockStore stops balancer from selecting the store. +func (c *clusterInfo) BlockStore(storeID uint64) error { + c.Lock() + defer c.Unlock() + return c.core.BlockStore(storeID) +} + +// UnblockStore allows balancer to select the store. +func (c *clusterInfo) UnblockStore(storeID uint64) { + c.Lock() + defer c.Unlock() + c.core.UnblockStore(storeID) +} + +// GetStores returns all stores in the cluster. +func (c *clusterInfo) GetStores() []*core.StoreInfo { + c.RLock() + defer c.RUnlock() + return c.core.GetStores() +} + +func (c *clusterInfo) getMetaStores() []*metapb.Store { + c.RLock() + defer c.RUnlock() + return c.core.Stores.GetMetaStores() +} + +func (c *clusterInfo) getStoreCount() int { + c.RLock() + defer c.RUnlock() + return c.core.Stores.GetStoreCount() +} + +func (c *clusterInfo) getStoresBytesWriteStat() map[uint64]uint64 { + c.RLock() + defer c.RUnlock() + return c.core.Stores.GetStoresBytesWriteStat() +} + +func (c *clusterInfo) getStoresBytesReadStat() map[uint64]uint64 { + c.RLock() + defer c.RUnlock() + return c.core.Stores.GetStoresBytesReadStat() +} + +func (c *clusterInfo) getStoresKeysWriteStat() map[uint64]uint64 { + c.RLock() + defer c.RUnlock() + return c.core.Stores.GetStoresKeysWriteStat() +} + +func (c *clusterInfo) getStoresKeysReadStat() map[uint64]uint64 { + c.RLock() + defer c.RUnlock() + return c.core.Stores.GetStoresKeysReadStat() +} + +// ScanRegions scans region with start key, until number greater than limit. +func (c *clusterInfo) ScanRegions(startKey []byte, limit int) []*core.RegionInfo { + c.RLock() + defer c.RUnlock() + return c.core.Regions.ScanRange(startKey, limit) +} + +// GetAdjacentRegions returns region's info that is adjacent with specific region +func (c *clusterInfo) GetAdjacentRegions(region *core.RegionInfo) (*core.RegionInfo, *core.RegionInfo) { + c.RLock() + defer c.RUnlock() + return c.core.GetAdjacentRegions(region) +} + +// GetRegion searches for a region by ID. +func (c *clusterInfo) GetRegion(regionID uint64) *core.RegionInfo { + c.RLock() + defer c.RUnlock() + return c.core.GetRegion(regionID) +} + +// IsRegionHot checks if a region is in hot state. +func (c *clusterInfo) IsRegionHot(id uint64) bool { + c.RLock() + defer c.RUnlock() + return c.core.IsRegionHot(id, c.GetHotRegionCacheHitsThreshold()) +} + +// RandHotRegionFromStore randomly picks a hot region in specified store. +func (c *clusterInfo) RandHotRegionFromStore(store uint64, kind schedule.FlowKind) *core.RegionInfo { + c.RLock() + defer c.RUnlock() + r := c.core.HotCache.RandHotRegionFromStore(store, kind, c.GetHotRegionCacheHitsThreshold()) + if r == nil { + return nil + } + return c.core.GetRegion(r.RegionID) +} + +func (c *clusterInfo) searchRegion(regionKey []byte) *core.RegionInfo { + c.RLock() + defer c.RUnlock() + return c.core.Regions.SearchRegion(regionKey) +} + +func (c *clusterInfo) searchPrevRegion(regionKey []byte) *core.RegionInfo { + c.RLock() + defer c.RUnlock() + return c.core.Regions.SearchPrevRegion(regionKey) +} + +func (c *clusterInfo) putRegion(region *core.RegionInfo) error { + c.Lock() + defer c.Unlock() + return c.putRegionLocked(region) +} + +func (c *clusterInfo) putRegionLocked(region *core.RegionInfo) error { + if c.kv != nil { + if err := c.kv.SaveRegion(region.GetMeta()); err != nil { + return err + } + } + c.core.PutRegion(region) + return nil +} + +func (c *clusterInfo) getRegions() []*core.RegionInfo { + c.RLock() + defer c.RUnlock() + return c.core.Regions.GetRegions() +} + +func (c *clusterInfo) getStoreRegions(storeID uint64) []*core.RegionInfo { + c.RLock() + defer c.RUnlock() + return c.core.Regions.GetStoreRegions(storeID) +} + +func (c *clusterInfo) getMetaRegions() []*metapb.Region { + c.RLock() + defer c.RUnlock() + return c.core.Regions.GetMetaRegions() +} + +func (c *clusterInfo) getRegionCount() int { + c.RLock() + defer c.RUnlock() + return c.core.Regions.GetRegionCount() +} + +func (c *clusterInfo) getRegionStats(startKey, endKey []byte) *core.RegionStats { + c.RLock() + defer c.RUnlock() + return c.core.Regions.GetRegionStats(startKey, endKey) +} + +func (c *clusterInfo) dropRegion(id uint64) { + c.Lock() + defer c.Unlock() + if region := c.core.GetRegion(id); region != nil { + c.core.Regions.RemoveRegion(region) + } +} + +func (c *clusterInfo) getStoreRegionCount(storeID uint64) int { + c.RLock() + defer c.RUnlock() + return c.core.Regions.GetStoreRegionCount(storeID) +} + +// RandLeaderRegion returns a random region that has leader on the store. +func (c *clusterInfo) RandLeaderRegion(storeID uint64, opts ...core.RegionOption) *core.RegionInfo { + c.RLock() + defer c.RUnlock() + return c.core.RandLeaderRegion(storeID, opts...) +} + +// RandFollowerRegion returns a random region that has a follower on the store. +func (c *clusterInfo) RandFollowerRegion(storeID uint64, opts ...core.RegionOption) *core.RegionInfo { + c.RLock() + defer c.RUnlock() + return c.core.RandFollowerRegion(storeID, opts...) +} + +// GetAverageRegionSize returns the average region approximate size. +func (c *clusterInfo) GetAverageRegionSize() int64 { + c.RLock() + defer c.RUnlock() + return c.core.GetAverageRegionSize() +} + +// GetRegionStores returns all stores that contains the region's peer. +func (c *clusterInfo) GetRegionStores(region *core.RegionInfo) []*core.StoreInfo { + c.RLock() + defer c.RUnlock() + return c.getRegionStoresLocked(region) +} + +func (c *clusterInfo) getRegionStoresLocked(region *core.RegionInfo) []*core.StoreInfo { + stores := make([]*core.StoreInfo, 0, len(region.GetPeers())) + for _, p := range region.GetPeers() { + if store := c.core.Stores.GetStore(p.StoreId); store != nil { + stores = append(stores, store) + } + } + return stores +} + +func (c *clusterInfo) takeRegionStoresLocked(region *core.RegionInfo) []*core.StoreInfo { + stores := make([]*core.StoreInfo, 0, len(region.GetPeers())) + for _, p := range region.GetPeers() { + if store := c.core.Stores.TakeStore(p.StoreId); store != nil { + stores = append(stores, store) + } + } + return stores +} + +// GetLeaderStore returns all stores that contains the region's leader peer. +func (c *clusterInfo) GetLeaderStore(region *core.RegionInfo) *core.StoreInfo { + c.RLock() + defer c.RUnlock() + return c.core.Stores.GetStore(region.GetLeader().GetStoreId()) +} + +// GetFollowerStores returns all stores that contains the region's follower peer. +func (c *clusterInfo) GetFollowerStores(region *core.RegionInfo) []*core.StoreInfo { + c.RLock() + defer c.RUnlock() + var stores []*core.StoreInfo + for id := range region.GetFollowers() { + if store := c.core.Stores.GetStore(id); store != nil { + stores = append(stores, store) + } + } + return stores +} + +// isPrepared if the cluster information is collected +func (c *clusterInfo) isPrepared() bool { + c.RLock() + defer c.RUnlock() + return c.prepareChecker.check(c) +} + +// handleStoreHeartbeat updates the store status. +func (c *clusterInfo) handleStoreHeartbeat(stats *pdpb.StoreStats) error { + c.Lock() + defer c.Unlock() + + storeID := stats.GetStoreId() + store := c.core.Stores.GetStore(storeID) + if store == nil { + return core.NewStoreNotFoundErr(storeID) + } + newStore := store.Clone(core.SetStoreStats(stats), core.SetLastHeartbeatTS(time.Now())) + c.core.Stores.SetStore(newStore) + return nil +} + +func (c *clusterInfo) updateStoreStatusLocked(id uint64) { + leaderCount := c.core.Regions.GetStoreLeaderCount(id) + regionCount := c.core.Regions.GetStoreRegionCount(id) + pendingPeerCount := c.core.Regions.GetStorePendingPeerCount(id) + leaderRegionSize := c.core.Regions.GetStoreLeaderRegionSize(id) + regionSize := c.core.Regions.GetStoreRegionSize(id) + c.core.Stores.UpdateStoreStatusLocked(id, leaderCount, regionCount, pendingPeerCount, leaderRegionSize, regionSize) +} + +// handleRegionHeartbeat updates the region information. +func (c *clusterInfo) handleRegionHeartbeat(region *core.RegionInfo) error { + c.RLock() + origin := c.core.Regions.GetRegion(region.GetID()) + if origin == nil { + for _, item := range c.core.Regions.GetOverlaps(region) { + if region.GetRegionEpoch().GetVersion() < item.GetRegionEpoch().GetVersion() { + c.RUnlock() + return ErrRegionIsStale(region.GetMeta(), item) + } + } + } + isWriteUpdate, writeItem := c.core.CheckWriteStatus(region) + isReadUpdate, readItem := c.core.CheckReadStatus(region) + c.RUnlock() + + // Save to KV if meta is updated. + // Save to cache if meta or leader is updated, or contains any down/pending peer. + // Mark isNew if the region in cache does not have leader. + var saveKV, saveCache, isNew bool + if origin == nil { + log.Debug("insert new region", + zap.Uint64("region-id", region.GetID()), + zap.Reflect("meta-region", core.HexRegionMeta(region.GetMeta())), + ) + saveKV, saveCache, isNew = true, true, true + } else { + r := region.GetRegionEpoch() + o := origin.GetRegionEpoch() + // Region meta is stale, return an error. + if r.GetVersion() < o.GetVersion() || r.GetConfVer() < o.GetConfVer() { + return ErrRegionIsStale(region.GetMeta(), origin.GetMeta()) + } + if r.GetVersion() > o.GetVersion() { + log.Info("region Version changed", + zap.Uint64("region-id", region.GetID()), + zap.String("detail", core.DiffRegionKeyInfo(origin, region)), + zap.Uint64("old-version", o.GetVersion()), + zap.Uint64("new-version", r.GetVersion()), + ) + saveKV, saveCache = true, true + } + if r.GetConfVer() > o.GetConfVer() { + log.Info("region ConfVer changed", + zap.Uint64("region-id", region.GetID()), + zap.String("detail", core.DiffRegionPeersInfo(origin, region)), + zap.Uint64("old-confver", o.GetConfVer()), + zap.Uint64("new-confver", r.GetConfVer()), + ) + saveKV, saveCache = true, true + } + if region.GetLeader().GetId() != origin.GetLeader().GetId() { + if origin.GetLeader().GetId() == 0 { + isNew = true + } else { + log.Info("leader changed", + zap.Uint64("region-id", region.GetID()), + zap.Uint64("from", origin.GetLeader().GetStoreId()), + zap.Uint64("to", region.GetLeader().GetStoreId()), + ) + } + saveCache = true + } + if len(region.GetDownPeers()) > 0 || len(region.GetPendingPeers()) > 0 { + saveCache = true + } + if len(origin.GetDownPeers()) > 0 || len(origin.GetPendingPeers()) > 0 { + saveCache = true + } + if len(region.GetPeers()) != len(origin.GetPeers()) { + saveKV, saveCache = true, true + } + if region.GetApproximateSize() != origin.GetApproximateSize() { + saveCache = true + } + if region.GetApproximateKeys() != origin.GetApproximateKeys() { + saveCache = true + } + } + + if saveKV && c.kv != nil { + if err := c.kv.SaveRegion(region.GetMeta()); err != nil { + // Not successfully saved to kv is not fatal, it only leads to longer warm-up + // after restart. Here we only log the error then go on updating cache. + log.Error("fail to save region to kv", + zap.Uint64("region-id", region.GetID()), + zap.Reflect("region-meta", core.HexRegionMeta(region.GetMeta())), + zap.Error(err)) + } + select { + case c.changedRegions <- region: + default: + } + } + if !isWriteUpdate && !isReadUpdate && !saveCache && !isNew { + return nil + } + + c.Lock() + defer c.Unlock() + if isNew { + c.prepareChecker.collect(region) + } + + if saveCache { + overlaps := c.core.Regions.SetRegion(region) + if c.kv != nil { + for _, item := range overlaps { + if err := c.kv.DeleteRegion(item); err != nil { + log.Error("fail to delete region from kv", + zap.Uint64("region-id", item.GetId()), + zap.Reflect("region-meta", core.HexRegionMeta(item)), + zap.Error(err)) + } + } + } + for _, item := range overlaps { + if c.regionStats != nil { + c.regionStats.clearDefunctRegion(item.GetId()) + } + c.labelLevelStats.clearDefunctRegion(item.GetId()) + } + + // Update related stores. + if origin != nil { + for _, p := range origin.GetPeers() { + c.updateStoreStatusLocked(p.GetStoreId()) + } + } + for _, p := range region.GetPeers() { + c.updateStoreStatusLocked(p.GetStoreId()) + } + } + + if c.regionStats != nil { + c.regionStats.Observe(region, c.takeRegionStoresLocked(region)) + } + + key := region.GetID() + if isWriteUpdate { + c.core.HotCache.Update(key, writeItem, schedule.WriteFlow) + } + if isReadUpdate { + c.core.HotCache.Update(key, readItem, schedule.ReadFlow) + } + return nil +} + +func (c *clusterInfo) updateRegionsLabelLevelStats(regions []*core.RegionInfo) { + c.Lock() + defer c.Unlock() + for _, region := range regions { + c.labelLevelStats.Observe(region, c.takeRegionStoresLocked(region), c.GetLocationLabels()) + } +} + +func (c *clusterInfo) collectMetrics() { + if c.regionStats == nil { + return + } + c.RLock() + defer c.RUnlock() + c.regionStats.Collect() + c.labelLevelStats.Collect() + // collect hot cache metrics + c.core.HotCache.CollectMetrics(c.core.Stores) +} + +func (c *clusterInfo) GetRegionStatsByType(typ regionStatisticType) []*core.RegionInfo { + if c.regionStats == nil { + return nil + } + c.RLock() + defer c.RUnlock() + return c.regionStats.getRegionStatsByType(typ) +} + +func (c *clusterInfo) GetOpt() schedule.NamespaceOptions { + return c.opt +} + +func (c *clusterInfo) GetLeaderScheduleLimit() uint64 { + return c.opt.GetLeaderScheduleLimit(namespace.DefaultNamespace) +} + +func (c *clusterInfo) GetRegionScheduleLimit() uint64 { + return c.opt.GetRegionScheduleLimit(namespace.DefaultNamespace) +} + +func (c *clusterInfo) GetReplicaScheduleLimit() uint64 { + return c.opt.GetReplicaScheduleLimit(namespace.DefaultNamespace) +} + +func (c *clusterInfo) GetMergeScheduleLimit() uint64 { + return c.opt.GetMergeScheduleLimit(namespace.DefaultNamespace) +} + +func (c *clusterInfo) GetHotRegionScheduleLimit() uint64 { + return c.opt.GetHotRegionScheduleLimit(namespace.DefaultNamespace) +} + +func (c *clusterInfo) GetTolerantSizeRatio() float64 { + return c.opt.GetTolerantSizeRatio() +} + +func (c *clusterInfo) GetLowSpaceRatio() float64 { + return c.opt.GetLowSpaceRatio() +} + +func (c *clusterInfo) GetHighSpaceRatio() float64 { + return c.opt.GetHighSpaceRatio() +} + +func (c *clusterInfo) GetMaxSnapshotCount() uint64 { + return c.opt.GetMaxSnapshotCount() +} + +func (c *clusterInfo) GetMaxPendingPeerCount() uint64 { + return c.opt.GetMaxPendingPeerCount() +} + +func (c *clusterInfo) GetMaxMergeRegionSize() uint64 { + return c.opt.GetMaxMergeRegionSize() +} + +func (c *clusterInfo) GetMaxMergeRegionKeys() uint64 { + return c.opt.GetMaxMergeRegionKeys() +} + +func (c *clusterInfo) GetSplitMergeInterval() time.Duration { + return c.opt.GetSplitMergeInterval() +} + +func (c *clusterInfo) GetPatrolRegionInterval() time.Duration { + return c.opt.GetPatrolRegionInterval() +} + +func (c *clusterInfo) GetMaxStoreDownTime() time.Duration { + return c.opt.GetMaxStoreDownTime() +} + +func (c *clusterInfo) GetMaxReplicas() int { + return c.opt.GetMaxReplicas(namespace.DefaultNamespace) +} + +func (c *clusterInfo) GetLocationLabels() []string { + return c.opt.GetLocationLabels() +} + +func (c *clusterInfo) GetHotRegionCacheHitsThreshold() int { + return c.opt.GetHotRegionCacheHitsThreshold() +} + +func (c *clusterInfo) IsRaftLearnerEnabled() bool { + if !c.IsFeatureSupported(RaftLearner) { + return false + } + return c.opt.IsRaftLearnerEnabled() +} + +func (c *clusterInfo) IsRemoveDownReplicaEnabled() bool { + return c.opt.IsRemoveDownReplicaEnabled() +} + +func (c *clusterInfo) IsReplaceOfflineReplicaEnabled() bool { + return c.opt.IsReplaceOfflineReplicaEnabled() +} + +func (c *clusterInfo) IsMakeUpReplicaEnabled() bool { + return c.opt.IsMakeUpReplicaEnabled() +} + +func (c *clusterInfo) IsRemoveExtraReplicaEnabled() bool { + return c.opt.IsRemoveExtraReplicaEnabled() +} + +func (c *clusterInfo) IsLocationReplacementEnabled() bool { + return c.opt.IsLocationReplacementEnabled() +} + +func (c *clusterInfo) IsNamespaceRelocationEnabled() bool { + return c.opt.IsNamespaceRelocationEnabled() +} + +func (c *clusterInfo) CheckLabelProperty(typ string, labels []*metapb.StoreLabel) bool { + return c.opt.CheckLabelProperty(typ, labels) +} + +// RegionReadStats returns hot region's read stats. +func (c *clusterInfo) RegionReadStats() []*core.RegionStat { + // RegionStats is a thread-safe method + return c.core.HotCache.RegionStats(schedule.ReadFlow) +} + +// RegionWriteStats returns hot region's write stats. +func (c *clusterInfo) RegionWriteStats() []*core.RegionStat { + // RegionStats is a thread-safe method + return c.core.HotCache.RegionStats(schedule.WriteFlow) +} + +type prepareChecker struct { + reactiveRegions map[uint64]int + start time.Time + sum int + isPrepared bool +} + +func newPrepareChecker() *prepareChecker { + return &prepareChecker{ + start: time.Now(), + reactiveRegions: make(map[uint64]int), + } +} + +// Before starting up the scheduler, we need to take the proportion of the regions on each store into consideration. +func (checker *prepareChecker) check(c *clusterInfo) bool { + if checker.isPrepared || time.Since(checker.start) > collectTimeout { + return true + } + // The number of active regions should be more than total region of all stores * collectFactor + if float64(c.core.Regions.Length())*collectFactor > float64(checker.sum) { + return false + } + for _, store := range c.core.GetStores() { + if !store.IsUp() { + continue + } + storeID := store.GetID() + // For each store, the number of active regions should be more than total region of the store * collectFactor + if float64(c.core.Regions.GetStoreRegionCount(storeID))*collectFactor > float64(checker.reactiveRegions[storeID]) { + return false + } + } + checker.isPrepared = true + return true +} + +func (checker *prepareChecker) collect(region *core.RegionInfo) { + for _, p := range region.GetPeers() { + checker.reactiveRegions[p.GetStoreId()]++ + } + checker.sum++ +} diff --git a/server/config.go b/server/config.go index aa7aab4..2f86e1b 100644 --- a/server/config.go +++ b/server/config.go @@ -65,6 +65,8 @@ type Config struct { Namespace map[string]NamespaceConfig `json:"namespace"` + PDServerCfg PDServerConfig `toml:"pd-server" json:"pd-server"` + ClusterVersion semver.Version `json:"cluster-version"` // QuotaBackendBytes Raise alarms when backend size exceeds the given quota. 0 means use the default quota. @@ -621,6 +623,12 @@ func (s SecurityConfig) ToTLSConfig() (*tls.Config, error) { return tlsConfig, nil } +// PDServerConfig is the configuration for pd server. +type PDServerConfig struct { + // UseRegionStorage enables the independent region storage. + UseRegionStorage bool `toml:"use-region-storage" json:"use-region-storage,string"` +} + // StoreLabel is the config item of LabelPropertyConfig. type StoreLabel struct { Key string `toml:"key" json:"key"` diff --git a/server/coordinator.go b/server/coordinator.go new file mode 100644 index 0000000..b8b75cb --- /dev/null +++ b/server/coordinator.go @@ -0,0 +1,479 @@ +// Copyright 2016 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "context" + "fmt" + "sync" + "time" + + log "github.com/pingcap/log" + "github.com/pingcap/pd/pkg/logutil" + "github.com/pingcap/pd/server/core" + "github.com/pingcap/pd/server/namespace" + "github.com/pingcap/pd/server/schedule" + "github.com/pkg/errors" + "go.uber.org/zap" +) + +const ( + runSchedulerCheckInterval = 3 * time.Second + collectFactor = 0.8 + collectTimeout = 5 * time.Minute + maxScheduleRetries = 10 + + regionheartbeatSendChanCap = 1024 + hotRegionScheduleName = "balance-hot-region-scheduler" + + patrolScanRegionLimit = 128 // It takes about 14 minutes to iterate 1 million regions. +) + +var ( + errSchedulerExisted = errors.New("scheduler existed") + errSchedulerNotFound = errors.New("scheduler not found") +) + +// coordinator is used to manage all schedulers and checkers to decide if the region needs to be scheduled. +type coordinator struct { + sync.RWMutex + + wg sync.WaitGroup + ctx context.Context + cancel context.CancelFunc + + cluster *clusterInfo + replicaChecker *schedule.ReplicaChecker + regionScatterer *schedule.RegionScatterer + namespaceChecker *schedule.NamespaceChecker + mergeChecker *schedule.MergeChecker + schedulers map[string]*scheduleController + opController *schedule.OperatorController + classifier namespace.Classifier + hbStreams *heartbeatStreams +} + +// newCoordinator creates a new coordinator. +func newCoordinator(cluster *clusterInfo, hbStreams *heartbeatStreams, classifier namespace.Classifier) *coordinator { + ctx, cancel := context.WithCancel(context.Background()) + return &coordinator{ + ctx: ctx, + cancel: cancel, + cluster: cluster, + replicaChecker: schedule.NewReplicaChecker(cluster, classifier), + regionScatterer: schedule.NewRegionScatterer(cluster, classifier), + namespaceChecker: schedule.NewNamespaceChecker(cluster, classifier), + mergeChecker: schedule.NewMergeChecker(cluster, classifier), + schedulers: make(map[string]*scheduleController), + opController: schedule.NewOperatorController(cluster, hbStreams), + classifier: classifier, + hbStreams: hbStreams, + } +} + +// patrolRegions is used to scan regions. +// The checkers will check these regions to decide if they need to do some operations. +func (c *coordinator) patrolRegions() { + defer logutil.LogPanic() + + defer c.wg.Done() + timer := time.NewTimer(c.cluster.GetPatrolRegionInterval()) + defer timer.Stop() + + log.Info("coordinator: start patrol regions") + start := time.Now() + var key []byte + for { + select { + case <-timer.C: + timer.Reset(c.cluster.GetPatrolRegionInterval()) + case <-c.ctx.Done(): + log.Info("patrol regions has been stopped") + return + } + + regions := c.cluster.ScanRegions(key, patrolScanRegionLimit) + if len(regions) == 0 { + // Resets the scan key. + key = nil + continue + } + + for _, region := range regions { + // Skips the region if there is already a pending operator. + if c.opController.GetOperator(region.GetID()) != nil { + continue + } + + key = region.GetEndKey() + + if c.checkRegion(region) { + break + } + } + // Updates the label level isolation statistics. + c.cluster.updateRegionsLabelLevelStats(regions) + if len(key) == 0 { + patrolCheckRegionsHistogram.Observe(time.Since(start).Seconds()) + start = time.Now() + } + } +} + +func (c *coordinator) checkRegion(region *core.RegionInfo) bool { + // If PD has restarted, it need to check learners added before and promote them. + // Don't check isRaftLearnerEnabled cause it maybe disable learner feature but there are still some learners to promote. + opController := c.opController + for _, p := range region.GetLearners() { + if region.GetPendingLearner(p.GetId()) != nil { + continue + } + step := schedule.PromoteLearner{ + ToStore: p.GetStoreId(), + PeerID: p.GetId(), + } + op := schedule.NewOperator("promote-learner", region.GetID(), region.GetRegionEpoch(), schedule.OpRegion, step) + if opController.AddOperator(op) { + return true + } + } + + if opController.OperatorCount(schedule.OpLeader) < c.cluster.GetLeaderScheduleLimit() && + opController.OperatorCount(schedule.OpRegion) < c.cluster.GetRegionScheduleLimit() && + opController.OperatorCount(schedule.OpReplica) < c.cluster.GetReplicaScheduleLimit() { + if op := c.namespaceChecker.Check(region); op != nil { + if opController.AddOperator(op) { + return true + } + } + } + + if opController.OperatorCount(schedule.OpReplica) < c.cluster.GetReplicaScheduleLimit() { + if op := c.replicaChecker.Check(region); op != nil { + if opController.AddOperator(op) { + return true + } + } + } + if c.cluster.IsFeatureSupported(RegionMerge) && opController.OperatorCount(schedule.OpMerge) < c.cluster.GetMergeScheduleLimit() { + if ops := c.mergeChecker.Check(region); ops != nil { + // It makes sure that two operators can be added successfully altogether. + if opController.AddOperator(ops...) { + return true + } + } + } + return false +} + +func (c *coordinator) run() { + ticker := time.NewTicker(runSchedulerCheckInterval) + defer ticker.Stop() + log.Info("coordinator: Start collect cluster information") + for { + if c.shouldRun() { + log.Info("coordinator: Cluster information is prepared") + break + } + select { + case <-ticker.C: + case <-c.ctx.Done(): + return + } + } + log.Info("coordinator: Run scheduler") + + k := 0 + scheduleCfg := c.cluster.opt.load().clone() + for _, schedulerCfg := range scheduleCfg.Schedulers { + if schedulerCfg.Disable { + scheduleCfg.Schedulers[k] = schedulerCfg + k++ + log.Info("skip create scheduler", zap.String("scheduler-type", schedulerCfg.Type)) + continue + } + s, err := schedule.CreateScheduler(schedulerCfg.Type, c.opController, schedulerCfg.Args...) + if err != nil { + log.Error("can not create scheduler", zap.String("scheduler-type", schedulerCfg.Type), zap.Error(err)) + continue + } + log.Info("create scheduler", zap.String("scheduler-name", s.GetName())) + if err = c.addScheduler(s, schedulerCfg.Args...); err != nil { + log.Error("can not add scheduler", zap.String("scheduler-name", s.GetName()), zap.Error(err)) + } + + // Only records the valid scheduler config. + if err == nil { + scheduleCfg.Schedulers[k] = schedulerCfg + k++ + } + } + + // Removes the invalid scheduler config and persist. + scheduleCfg.Schedulers = scheduleCfg.Schedulers[:k] + c.cluster.opt.store(scheduleCfg) + if err := c.cluster.opt.persist(c.cluster.kv); err != nil { + log.Error("cannot persist schedule config", zap.Error(err)) + } + + c.wg.Add(1) + // Starts to patrol regions. + go c.patrolRegions() +} + +func (c *coordinator) stop() { + c.cancel() +} + +// Hack to retrieve info from scheduler. +// TODO: remove it. +type hasHotStatus interface { + GetHotReadStatus() *core.StoreHotRegionInfos + GetHotWriteStatus() *core.StoreHotRegionInfos +} + +func (c *coordinator) getHotWriteRegions() *core.StoreHotRegionInfos { + c.RLock() + defer c.RUnlock() + s, ok := c.schedulers[hotRegionScheduleName] + if !ok { + return nil + } + if h, ok := s.Scheduler.(hasHotStatus); ok { + return h.GetHotWriteStatus() + } + return nil +} + +func (c *coordinator) getHotReadRegions() *core.StoreHotRegionInfos { + c.RLock() + defer c.RUnlock() + s, ok := c.schedulers[hotRegionScheduleName] + if !ok { + return nil + } + if h, ok := s.Scheduler.(hasHotStatus); ok { + return h.GetHotReadStatus() + } + return nil +} + +func (c *coordinator) getSchedulers() []string { + c.RLock() + defer c.RUnlock() + + names := make([]string, 0, len(c.schedulers)) + for name := range c.schedulers { + names = append(names, name) + } + return names +} + +func (c *coordinator) collectSchedulerMetrics() { + c.RLock() + defer c.RUnlock() + for _, s := range c.schedulers { + var allowScheduler float64 + // If the scheduler is not allowed to schedule, it will disappear in Grafana panel. + // See issue #1341. + if s.AllowSchedule() { + allowScheduler = 1 + } + schedulerStatusGauge.WithLabelValues(s.GetName(), "allow").Set(allowScheduler) + } +} + +func (c *coordinator) collectHotSpotMetrics() { + c.RLock() + defer c.RUnlock() + // Collects hot write region metrics. + s, ok := c.schedulers[hotRegionScheduleName] + if !ok { + return + } + stores := c.cluster.GetStores() + status := s.Scheduler.(hasHotStatus).GetHotWriteStatus() + for _, s := range stores { + storeAddress := s.GetAddress() + storeID := s.GetID() + storeLabel := fmt.Sprintf("%d", storeID) + stat, ok := status.AsPeer[storeID] + if ok { + totalWriteBytes := float64(stat.TotalFlowBytes) + hotWriteRegionCount := float64(stat.RegionsCount) + + hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "total_written_bytes_as_peer").Set(totalWriteBytes) + hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "hot_write_region_as_peer").Set(hotWriteRegionCount) + } else { + hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "total_written_bytes_as_peer").Set(0) + hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "hot_write_region_as_peer").Set(0) + } + + stat, ok = status.AsLeader[storeID] + if ok { + totalWriteBytes := float64(stat.TotalFlowBytes) + hotWriteRegionCount := float64(stat.RegionsCount) + + hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "total_written_bytes_as_leader").Set(totalWriteBytes) + hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "hot_write_region_as_leader").Set(hotWriteRegionCount) + } else { + hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "total_written_bytes_as_leader").Set(0) + hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "hot_write_region_as_leader").Set(0) + } + } + + // Collects hot read region metrics. + status = s.Scheduler.(hasHotStatus).GetHotReadStatus() + for _, s := range stores { + storeAddress := s.GetAddress() + storeID := s.GetID() + storeLabel := fmt.Sprintf("%d", storeID) + stat, ok := status.AsLeader[storeID] + if ok { + totalReadBytes := float64(stat.TotalFlowBytes) + hotReadRegionCount := float64(stat.RegionsCount) + + hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "total_read_bytes_as_leader").Set(totalReadBytes) + hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "hot_read_region_as_leader").Set(hotReadRegionCount) + } else { + hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "total_read_bytes_as_leader").Set(0) + hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "hot_read_region_as_leader").Set(0) + } + } + +} + +func (c *coordinator) shouldRun() bool { + return c.cluster.isPrepared() +} + +func (c *coordinator) addScheduler(scheduler schedule.Scheduler, args ...string) error { + c.Lock() + defer c.Unlock() + + if _, ok := c.schedulers[scheduler.GetName()]; ok { + return errSchedulerExisted + } + + s := newScheduleController(c, scheduler) + if err := s.Prepare(c.cluster); err != nil { + return err + } + + c.wg.Add(1) + go c.runScheduler(s) + c.schedulers[s.GetName()] = s + c.cluster.opt.AddSchedulerCfg(s.GetType(), args) + + return nil +} + +func (c *coordinator) removeScheduler(name string) error { + c.Lock() + defer c.Unlock() + + s, ok := c.schedulers[name] + if !ok { + return errSchedulerNotFound + } + + s.Stop() + schedulerStatusGauge.WithLabelValues(name, "allow").Set(0) + delete(c.schedulers, name) + + return c.cluster.opt.RemoveSchedulerCfg(name) +} + +func (c *coordinator) runScheduler(s *scheduleController) { + defer logutil.LogPanic() + defer c.wg.Done() + defer s.Cleanup(c.cluster) + + timer := time.NewTimer(s.GetInterval()) + defer timer.Stop() + + for { + select { + case <-timer.C: + timer.Reset(s.GetInterval()) + if !s.AllowSchedule() { + continue + } + if op := s.Schedule(); op != nil { + c.opController.AddOperator(op...) + } + + case <-s.Ctx().Done(): + log.Info("scheduler has been stopped", + zap.String("scheduler-name", s.GetName()), + zap.Error(s.Ctx().Err())) + return + } + } +} + +// scheduleController is used to manage a scheduler to schedule. +type scheduleController struct { + schedule.Scheduler + cluster *clusterInfo + opController *schedule.OperatorController + classifier namespace.Classifier + nextInterval time.Duration + ctx context.Context + cancel context.CancelFunc +} + +// newScheduleController creates a new scheduleController. +func newScheduleController(c *coordinator, s schedule.Scheduler) *scheduleController { + ctx, cancel := context.WithCancel(c.ctx) + return &scheduleController{ + Scheduler: s, + cluster: c.cluster, + opController: c.opController, + nextInterval: s.GetMinInterval(), + classifier: c.classifier, + ctx: ctx, + cancel: cancel, + } +} + +func (s *scheduleController) Ctx() context.Context { + return s.ctx +} + +func (s *scheduleController) Stop() { + s.cancel() +} + +func (s *scheduleController) Schedule() []*schedule.Operator { + for i := 0; i < maxScheduleRetries; i++ { + // If we have schedule, reset interval to the minimal interval. + if op := scheduleByNamespace(s.cluster, s.classifier, s.Scheduler); op != nil { + s.nextInterval = s.Scheduler.GetMinInterval() + return op + } + } + s.nextInterval = s.Scheduler.GetNextInterval(s.nextInterval) + return nil +} + +// GetInterval returns the interval of scheduling for a scheduler. +func (s *scheduleController) GetInterval() time.Duration { + return s.nextInterval +} + +// AllowSchedule returns if a scheduler is allowed to schedule. +func (s *scheduleController) AllowSchedule() bool { + return s.Scheduler.IsScheduleAllowed(s.cluster) +} diff --git a/server/core/kv.go b/server/core/kv.go new file mode 100644 index 0000000..35ebe6d --- /dev/null +++ b/server/core/kv.go @@ -0,0 +1,33 @@ +// Copyright 2017 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package core + +const ( + clusterPath = "raft" + configPath = "config" + schedulePath = "schedule" + gcPath = "gc" +) + +const ( + maxKVRangeLimit = 10000 + minKVRangeLimit = 100 +) + +// KV wraps all kv operations, keep it stateless. +type KV struct { + KVBase + regionKV *RegionKV + useRegionKV int32 +} diff --git a/server/core/kv_base.go b/server/core/kv_base.go new file mode 100644 index 0000000..a6a637b --- /dev/null +++ b/server/core/kv_base.go @@ -0,0 +1,86 @@ +// Copyright 2017 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package core + +import ( + "sync" + + "github.com/google/btree" +) + +// KVBase is an abstract interface for load/save pd cluster data. +type KVBase interface { + Load(key string) (string, error) + LoadRange(key, endKey string, limit int) (keys []string, values []string, err error) + Save(key, value string) error + Delete(key string) error +} + +type memoryKV struct { + sync.RWMutex + tree *btree.BTree +} + +// NewMemoryKV returns an in-memory kvBase for testing. +func NewMemoryKV() KVBase { + return &memoryKV{ + tree: btree.New(2), + } +} + +type memoryKVItem struct { + key, value string +} + +func (s memoryKVItem) Less(than btree.Item) bool { + return s.key < than.(memoryKVItem).key +} + +func (kv *memoryKV) Load(key string) (string, error) { + kv.RLock() + defer kv.RUnlock() + item := kv.tree.Get(memoryKVItem{key, ""}) + if item == nil { + return "", nil + } + return item.(memoryKVItem).value, nil +} + +func (kv *memoryKV) LoadRange(key, endKey string, limit int) ([]string, []string, error) { + kv.RLock() + defer kv.RUnlock() + keys := make([]string, 0, limit) + values := make([]string, 0, limit) + kv.tree.AscendRange(memoryKVItem{key, ""}, memoryKVItem{endKey, ""}, func(item btree.Item) bool { + keys = append(keys, item.(memoryKVItem).key) + values = append(values, item.(memoryKVItem).value) + return len(keys) < limit + }) + return keys, values, nil +} + +func (kv *memoryKV) Save(key, value string) error { + kv.Lock() + defer kv.Unlock() + kv.tree.ReplaceOrInsert(memoryKVItem{key, value}) + return nil +} + +func (kv *memoryKV) Delete(key string) error { + kv.Lock() + defer kv.Unlock() + + kv.tree.Delete(memoryKVItem{key, ""}) + return nil +} diff --git a/server/core/levedb_kv.go b/server/core/levedb_kv.go new file mode 100644 index 0000000..e17291c --- /dev/null +++ b/server/core/levedb_kv.go @@ -0,0 +1,9 @@ +package core + +import ( + "github.com/syndtr/goleveldb/leveldb" +) + +type leveldbKV struct { + db *leveldb.DB +} diff --git a/server/core/region_kv.go b/server/core/region_kv.go new file mode 100644 index 0000000..935a9f0 --- /dev/null +++ b/server/core/region_kv.go @@ -0,0 +1,24 @@ +package core + +import ( + "context" + "sync" + "time" + + "github.com/pingcap/kvproto/pkg/metapb" +) + +var dirtyFlushTick = time.Second + +// RegionKV is used to save regions. +type RegionKV struct { + *leveldbKV + mu sync.RWMutex + batchRegions map[string]*metapb.Region + batchSize int + cacheSize int + flushRate time.Duration + flushTime time.Time + ctx context.Context + cancel context.CancelFunc +} diff --git a/server/handler.go b/server/handler.go new file mode 100644 index 0000000..825d846 --- /dev/null +++ b/server/handler.go @@ -0,0 +1,52 @@ +// Copyright 2016 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "github.com/pingcap/kvproto/pkg/metapb" + "github.com/pkg/errors" +) + +var ( + // ErrNotBootstrapped is error info for cluster not bootstrapped + ErrNotBootstrapped = errors.New("TiKV cluster not bootstrapped, please start TiKV first") + // ErrOperatorNotFound is error info for operator not found + ErrOperatorNotFound = errors.New("operator not found") + // ErrAddOperator is error info for already have an operator when adding operator + ErrAddOperator = errors.New("failed to add operator, maybe already have one") + // ErrRegionNotAdjacent is error info for region not adjacent + ErrRegionNotAdjacent = errors.New("two regions are not adjacent") + // ErrRegionNotFound is error info for region not found + ErrRegionNotFound = func(regionID uint64) error { + return errors.Errorf("region %v not found", regionID) + } + // ErrRegionAbnormalPeer is error info for region has abonormal peer + ErrRegionAbnormalPeer = func(regionID uint64) error { + return errors.Errorf("region %v has abnormal peer", regionID) + } + // ErrRegionIsStale is error info for region is stale + ErrRegionIsStale = func(region *metapb.Region, origin *metapb.Region) error { + return errors.Errorf("region is stale: region %v origin %v", region, origin) + } +) + +// Handler is a helper to export methods to handle API/RPC requests. +type Handler struct { + s *Server + opt *scheduleOption +} + +func newHandler(s *Server) *Handler { + return &Handler{s: s, opt: s.scheduleOpt} +} diff --git a/server/heartbeat_streams.go b/server/heartbeat_streams.go new file mode 100644 index 0000000..3161f6d --- /dev/null +++ b/server/heartbeat_streams.go @@ -0,0 +1,186 @@ +// Copyright 2017 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "context" + "strconv" + "sync" + "time" + + "github.com/pingcap/kvproto/pkg/metapb" + "github.com/pingcap/kvproto/pkg/pdpb" + log "github.com/pingcap/log" + "github.com/pingcap/pd/pkg/logutil" + "github.com/pingcap/pd/server/core" + "go.uber.org/zap" +) + +const heartbeatStreamKeepAliveInterval = time.Minute + +type heartbeatStream interface { + Send(*pdpb.RegionHeartbeatResponse) error +} + +type streamUpdate struct { + storeID uint64 + stream heartbeatStream +} + +type heartbeatStreams struct { + wg sync.WaitGroup + ctx context.Context + cancel context.CancelFunc + clusterID uint64 + streams map[uint64]heartbeatStream + msgCh chan *pdpb.RegionHeartbeatResponse + streamCh chan streamUpdate + cluster *RaftCluster +} + +func newHeartbeatStreams(clusterID uint64, cluster *RaftCluster) *heartbeatStreams { + ctx, cancel := context.WithCancel(context.Background()) + hs := &heartbeatStreams{ + ctx: ctx, + cancel: cancel, + clusterID: clusterID, + streams: make(map[uint64]heartbeatStream), + msgCh: make(chan *pdpb.RegionHeartbeatResponse, regionheartbeatSendChanCap), + streamCh: make(chan streamUpdate, 1), + cluster: cluster, + } + hs.wg.Add(1) + go hs.run() + return hs +} + +func (s *heartbeatStreams) run() { + defer logutil.LogPanic() + + defer s.wg.Done() + + keepAliveTicker := time.NewTicker(heartbeatStreamKeepAliveInterval) + defer keepAliveTicker.Stop() + + keepAlive := &pdpb.RegionHeartbeatResponse{Header: &pdpb.ResponseHeader{ClusterId: s.clusterID}} + + for { + select { + case update := <-s.streamCh: + s.streams[update.storeID] = update.stream + case msg := <-s.msgCh: + storeID := msg.GetTargetPeer().GetStoreId() + storeLabel := strconv.FormatUint(storeID, 10) + store, err := s.cluster.GetStore(storeID) + if err != nil { + log.Error("fail to get store", + zap.Uint64("region-id", msg.RegionId), + zap.Uint64("store-id", storeID), + zap.Error(err)) + delete(s.streams, storeID) + continue + } + storeAddress := store.GetAddress() + if stream, ok := s.streams[storeID]; ok { + if err := stream.Send(msg); err != nil { + log.Error("send heartbeat message fail", + zap.Uint64("region-id", msg.RegionId), zap.Error(err)) + delete(s.streams, storeID) + regionHeartbeatCounter.WithLabelValues(storeAddress, storeLabel, "push", "err").Inc() + } else { + regionHeartbeatCounter.WithLabelValues(storeAddress, storeLabel, "push", "ok").Inc() + } + } else { + log.Debug("heartbeat stream not found, skip send message", + zap.Uint64("region-id", msg.RegionId), + zap.Uint64("store-id", storeID)) + regionHeartbeatCounter.WithLabelValues(storeAddress, storeLabel, "push", "skip").Inc() + } + case <-keepAliveTicker.C: + for storeID, stream := range s.streams { + store, err := s.cluster.GetStore(storeID) + if err != nil { + log.Error("fail to get store", zap.Uint64("store-id", storeID), zap.Error(err)) + delete(s.streams, storeID) + continue + } + storeAddress := store.GetAddress() + storeLabel := strconv.FormatUint(storeID, 10) + if err := stream.Send(keepAlive); err != nil { + log.Error("send keepalive message fail", + zap.Uint64("target-store-id", storeID), + zap.Error(err)) + delete(s.streams, storeID) + regionHeartbeatCounter.WithLabelValues(storeAddress, storeLabel, "keepalive", "err").Inc() + } else { + regionHeartbeatCounter.WithLabelValues(storeAddress, storeLabel, "keepalive", "ok").Inc() + } + } + case <-s.ctx.Done(): + return + } + } +} + +func (s *heartbeatStreams) Close() { + s.cancel() + s.wg.Wait() +} + +func (s *heartbeatStreams) bindStream(storeID uint64, stream heartbeatStream) { + update := streamUpdate{ + storeID: storeID, + stream: stream, + } + select { + case s.streamCh <- update: + case <-s.ctx.Done(): + } +} + +func (s *heartbeatStreams) SendMsg(region *core.RegionInfo, msg *pdpb.RegionHeartbeatResponse) { + if region.GetLeader() == nil { + return + } + + msg.Header = &pdpb.ResponseHeader{ClusterId: s.clusterID} + msg.RegionId = region.GetID() + msg.RegionEpoch = region.GetRegionEpoch() + msg.TargetPeer = region.GetLeader() + + select { + case s.msgCh <- msg: + case <-s.ctx.Done(): + } +} + +func (s *heartbeatStreams) sendErr(errType pdpb.ErrorType, errMsg string, targetPeer *metapb.Peer, storeAddress, storeLabel string) { + regionHeartbeatCounter.WithLabelValues(storeAddress, storeLabel, "report", "err").Inc() + + msg := &pdpb.RegionHeartbeatResponse{ + Header: &pdpb.ResponseHeader{ + ClusterId: s.clusterID, + Error: &pdpb.Error{ + Type: errType, + Message: errMsg, + }, + }, + TargetPeer: targetPeer, + } + + select { + case s.msgCh <- msg: + case <-s.ctx.Done(): + } +} diff --git a/server/id.go b/server/id.go new file mode 100644 index 0000000..047d96a --- /dev/null +++ b/server/id.go @@ -0,0 +1,94 @@ +// Copyright 2016 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "sync" + + log "github.com/pingcap/log" + "github.com/pkg/errors" + "go.etcd.io/etcd/clientv3" + "go.uber.org/zap" +) + +const ( + allocStep = uint64(1000) +) + +type idAllocator struct { + mu sync.Mutex + base uint64 + end uint64 + + s *Server +} + +func (alloc *idAllocator) Alloc() (uint64, error) { + alloc.mu.Lock() + defer alloc.mu.Unlock() + + if alloc.base == alloc.end { + end, err := alloc.generate() + if err != nil { + return 0, err + } + + alloc.end = end + alloc.base = alloc.end - allocStep + } + + alloc.base++ + + return alloc.base, nil +} + +func (alloc *idAllocator) generate() (uint64, error) { + key := alloc.s.getAllocIDPath() + value, err := getValue(alloc.s.client, key) + if err != nil { + return 0, err + } + + var ( + cmp clientv3.Cmp + end uint64 + ) + + if value == nil { + // create the key + cmp = clientv3.Compare(clientv3.CreateRevision(key), "=", 0) + } else { + // update the key + end, err = bytesToUint64(value) + if err != nil { + return 0, err + } + + cmp = clientv3.Compare(clientv3.Value(key), "=", string(value)) + } + + end += allocStep + value = uint64ToBytes(end) + resp, err := alloc.s.leaderTxn(cmp).Then(clientv3.OpPut(key, string(value))).Commit() + if err != nil { + return 0, err + } + if !resp.Succeeded { + return 0, errors.New("generate id failed, we may not leader") + } + + log.Info("idAllocator allocates a new id", zap.Uint64("alloc-id", end)) + metadataGauge.WithLabelValues("idalloc").Set(float64(end)) + return end, nil +} diff --git a/server/namespace/classifier.go b/server/namespace/classifier.go new file mode 100644 index 0000000..2c063c4 --- /dev/null +++ b/server/namespace/classifier.go @@ -0,0 +1,112 @@ +// Copyright 2017 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package namespace + +import ( + "fmt" + + "github.com/pingcap/pd/server/core" + "github.com/pkg/errors" +) + +// DefaultNamespace is the namespace all the store and region belong to by +// default. +const DefaultNamespace = "global" + +// DefaultClassifier is a classifier that classifies all regions and stores to +// DefaultNamespace. +var DefaultClassifier = defaultClassifier{} + +// Classifier is used to determine the namespace which the store or region +// belongs. +type Classifier interface { + GetAllNamespaces() []string + GetStoreNamespace(*core.StoreInfo) string + GetRegionNamespace(*core.RegionInfo) string + IsNamespaceExist(name string) bool + AllowMerge(*core.RegionInfo, *core.RegionInfo) bool + // Reload underlying namespaces + ReloadNamespaces() error + // These function below are only for tests + IsMetaExist() bool + IsTableIDExist(int64) bool + IsStoreIDExist(uint64) bool +} + +type defaultClassifier struct{} + +func (c defaultClassifier) GetAllNamespaces() []string { + return []string{DefaultNamespace} +} + +func (c defaultClassifier) GetStoreNamespace(*core.StoreInfo) string { + return DefaultNamespace +} + +func (c defaultClassifier) GetRegionNamespace(*core.RegionInfo) string { + return DefaultNamespace +} + +func (c defaultClassifier) IsNamespaceExist(name string) bool { + return name == DefaultNamespace +} + +func (c defaultClassifier) AllowMerge(one *core.RegionInfo, other *core.RegionInfo) bool { + return true +} + +func (c defaultClassifier) ReloadNamespaces() error { + return nil +} + +func (c defaultClassifier) IsMetaExist() bool { + return false +} + +func (c defaultClassifier) IsTableIDExist(tableID int64) bool { + return false +} + +func (c defaultClassifier) IsStoreIDExist(storeID uint64) bool { + return false +} + +// CreateClassifierFunc is for creating namespace classifier. +type CreateClassifierFunc func(*core.KV, core.IDAllocator) (Classifier, error) + +var classifierMap = make(map[string]CreateClassifierFunc) + +// RegisterClassifier binds a classifier creator. It should be called in init() +// func of a package. +func RegisterClassifier(name string, createFn CreateClassifierFunc) { + if _, ok := classifierMap[name]; ok { + panic(fmt.Sprintf("duplicated classifier name: %v", name)) + } + classifierMap[name] = createFn +} + +// CreateClassifier creates a namespace classifier with registered creator func. +func CreateClassifier(name string, kv *core.KV, idAlloc core.IDAllocator) (Classifier, error) { + fn, ok := classifierMap[name] + if !ok { + return nil, errors.Errorf("create func of %v is not registered", name) + } + return fn(kv, idAlloc) +} + +func init() { + RegisterClassifier("default", func(*core.KV, core.IDAllocator) (Classifier, error) { + return DefaultClassifier, nil + }) +} diff --git a/server/option.go b/server/option.go new file mode 100644 index 0000000..e51626c --- /dev/null +++ b/server/option.go @@ -0,0 +1,451 @@ +// Copyright 2017 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "reflect" + "sync/atomic" + "time" + + "github.com/coreos/go-semver/semver" + "github.com/pingcap/kvproto/pkg/metapb" + "github.com/pingcap/pd/server/core" + "github.com/pingcap/pd/server/schedule" +) + +// scheduleOption is a wrapper to access the configuration safely. +type scheduleOption struct { + v atomic.Value + rep *Replication + ns map[string]*namespaceOption + labelProperty atomic.Value + clusterVersion atomic.Value + pdServerConfig atomic.Value +} + +func newScheduleOption(cfg *Config) *scheduleOption { + o := &scheduleOption{} + o.store(&cfg.Schedule) + o.ns = make(map[string]*namespaceOption) + for name, nsCfg := range cfg.Namespace { + nsCfg := nsCfg + o.ns[name] = newNamespaceOption(&nsCfg) + } + o.rep = newReplication(&cfg.Replication) + o.pdServerConfig.Store(&cfg.PDServerCfg) + o.labelProperty.Store(cfg.LabelProperty) + o.clusterVersion.Store(cfg.ClusterVersion) + return o +} + +func (o *scheduleOption) load() *ScheduleConfig { + return o.v.Load().(*ScheduleConfig) +} + +func (o *scheduleOption) store(cfg *ScheduleConfig) { + o.v.Store(cfg) +} + +func (o *scheduleOption) GetReplication() *Replication { + return o.rep +} + +func (o *scheduleOption) GetMaxReplicas(name string) int { + if n, ok := o.ns[name]; ok { + return n.GetMaxReplicas() + } + return o.rep.GetMaxReplicas() +} + +func (o *scheduleOption) SetMaxReplicas(replicas int) { + o.rep.SetMaxReplicas(replicas) +} + +func (o *scheduleOption) GetLocationLabels() []string { + return o.rep.GetLocationLabels() +} + +func (o *scheduleOption) GetMaxSnapshotCount() uint64 { + return o.load().MaxSnapshotCount +} + +func (o *scheduleOption) GetMaxPendingPeerCount() uint64 { + return o.load().MaxPendingPeerCount +} + +func (o *scheduleOption) GetMaxMergeRegionSize() uint64 { + return o.load().MaxMergeRegionSize +} + +func (o *scheduleOption) GetMaxMergeRegionKeys() uint64 { + return o.load().MaxMergeRegionKeys +} + +func (o *scheduleOption) GetSplitMergeInterval() time.Duration { + return o.load().SplitMergeInterval.Duration +} + +func (o *scheduleOption) GetPatrolRegionInterval() time.Duration { + return o.load().PatrolRegionInterval.Duration +} + +func (o *scheduleOption) GetMaxStoreDownTime() time.Duration { + return o.load().MaxStoreDownTime.Duration +} + +func (o *scheduleOption) GetLeaderScheduleLimit(name string) uint64 { + if n, ok := o.ns[name]; ok { + return n.GetLeaderScheduleLimit() + } + return o.load().LeaderScheduleLimit +} + +func (o *scheduleOption) GetRegionScheduleLimit(name string) uint64 { + if n, ok := o.ns[name]; ok { + return n.GetRegionScheduleLimit() + } + return o.load().RegionScheduleLimit +} + +func (o *scheduleOption) GetReplicaScheduleLimit(name string) uint64 { + if n, ok := o.ns[name]; ok { + return n.GetReplicaScheduleLimit() + } + return o.load().ReplicaScheduleLimit +} + +func (o *scheduleOption) GetMergeScheduleLimit(name string) uint64 { + if n, ok := o.ns[name]; ok { + return n.GetMergeScheduleLimit() + } + return o.load().MergeScheduleLimit +} + +func (o *scheduleOption) GetHotRegionScheduleLimit(name string) uint64 { + if n, ok := o.ns[name]; ok { + return n.GetHotRegionScheduleLimit() + } + return o.load().HotRegionScheduleLimit +} + +func (o *scheduleOption) GetTolerantSizeRatio() float64 { + return o.load().TolerantSizeRatio +} + +func (o *scheduleOption) GetLowSpaceRatio() float64 { + return o.load().LowSpaceRatio +} + +func (o *scheduleOption) GetHighSpaceRatio() float64 { + return o.load().HighSpaceRatio +} + +func (o *scheduleOption) IsRaftLearnerEnabled() bool { + return !o.load().DisableLearner +} + +func (o *scheduleOption) IsRemoveDownReplicaEnabled() bool { + return !o.load().DisableRemoveDownReplica +} + +func (o *scheduleOption) IsReplaceOfflineReplicaEnabled() bool { + return !o.load().DisableReplaceOfflineReplica +} + +func (o *scheduleOption) IsMakeUpReplicaEnabled() bool { + return !o.load().DisableMakeUpReplica +} + +func (o *scheduleOption) IsRemoveExtraReplicaEnabled() bool { + return !o.load().DisableRemoveExtraReplica +} + +func (o *scheduleOption) IsLocationReplacementEnabled() bool { + return !o.load().DisableLocationReplacement +} + +func (o *scheduleOption) IsNamespaceRelocationEnabled() bool { + return !o.load().DisableNamespaceRelocation +} + +func (o *scheduleOption) GetSchedulers() SchedulerConfigs { + return o.load().Schedulers +} + +func (o *scheduleOption) AddSchedulerCfg(tp string, args []string) { + c := o.load() + v := c.clone() + for i, schedulerCfg := range v.Schedulers { + // comparing args is to cover the case that there are schedulers in same type but not with same name + // such as two schedulers of type "evict-leader", + // one name is "evict-leader-scheduler-1" and the other is "evict-leader-scheduler-2" + if reflect.DeepEqual(schedulerCfg, SchedulerConfig{Type: tp, Args: args, Disable: false}) { + return + } + + if reflect.DeepEqual(schedulerCfg, SchedulerConfig{Type: tp, Args: args, Disable: true}) { + schedulerCfg.Disable = false + v.Schedulers[i] = schedulerCfg + o.store(v) + return + } + } + v.Schedulers = append(v.Schedulers, SchedulerConfig{Type: tp, Args: args, Disable: false}) + o.store(v) +} + +func (o *scheduleOption) RemoveSchedulerCfg(name string) error { + c := o.load() + v := c.clone() + for i, schedulerCfg := range v.Schedulers { + // To create a temporary scheduler is just used to get scheduler's name + tmp, err := schedule.CreateScheduler(schedulerCfg.Type, schedule.NewOperatorController(nil, nil), schedulerCfg.Args...) + if err != nil { + return err + } + if tmp.GetName() == name { + if IsDefaultScheduler(tmp.GetType()) { + schedulerCfg.Disable = true + v.Schedulers[i] = schedulerCfg + } else { + v.Schedulers = append(v.Schedulers[:i], v.Schedulers[i+1:]...) + } + o.store(v) + return nil + } + } + return nil +} + +func (o *scheduleOption) SetLabelProperty(typ, labelKey, labelValue string) { + cfg := o.loadLabelPropertyConfig().clone() + for _, l := range cfg[typ] { + if l.Key == labelKey && l.Value == labelValue { + return + } + } + cfg[typ] = append(cfg[typ], StoreLabel{Key: labelKey, Value: labelValue}) + o.labelProperty.Store(cfg) +} + +func (o *scheduleOption) DeleteLabelProperty(typ, labelKey, labelValue string) { + cfg := o.loadLabelPropertyConfig().clone() + oldLabels := cfg[typ] + cfg[typ] = []StoreLabel{} + for _, l := range oldLabels { + if l.Key == labelKey && l.Value == labelValue { + continue + } + cfg[typ] = append(cfg[typ], l) + } + if len(cfg[typ]) == 0 { + delete(cfg, typ) + } + o.labelProperty.Store(cfg) +} + +func (o *scheduleOption) loadLabelPropertyConfig() LabelPropertyConfig { + return o.labelProperty.Load().(LabelPropertyConfig) +} + +func (o *scheduleOption) SetClusterVersion(v semver.Version) { + o.clusterVersion.Store(v) +} + +func (o *scheduleOption) loadClusterVersion() semver.Version { + return o.clusterVersion.Load().(semver.Version) +} + +func (o *scheduleOption) loadPDServerConfig() *PDServerConfig { + return o.pdServerConfig.Load().(*PDServerConfig) +} + +func (o *scheduleOption) persist(kv *core.KV) error { + namespaces := make(map[string]NamespaceConfig) + for name, ns := range o.ns { + namespaces[name] = *ns.load() + } + cfg := &Config{ + Schedule: *o.load(), + Replication: *o.rep.load(), + Namespace: namespaces, + LabelProperty: o.loadLabelPropertyConfig(), + ClusterVersion: o.loadClusterVersion(), + PDServerCfg: *o.loadPDServerConfig(), + } + err := kv.SaveConfig(cfg) + return err +} + +func (o *scheduleOption) reload(kv *core.KV) error { + namespaces := make(map[string]NamespaceConfig) + for name, ns := range o.ns { + namespaces[name] = *ns.load() + } + cfg := &Config{ + Schedule: *o.load().clone(), + Replication: *o.rep.load(), + Namespace: namespaces, + LabelProperty: o.loadLabelPropertyConfig().clone(), + ClusterVersion: o.loadClusterVersion(), + PDServerCfg: *o.loadPDServerConfig(), + } + isExist, err := kv.LoadConfig(cfg) + if err != nil { + return err + } + o.adjustScheduleCfg(cfg) + if isExist { + o.store(&cfg.Schedule) + o.rep.store(&cfg.Replication) + for name, nsCfg := range cfg.Namespace { + nsCfg := nsCfg + o.ns[name] = newNamespaceOption(&nsCfg) + } + o.labelProperty.Store(cfg.LabelProperty) + o.clusterVersion.Store(cfg.ClusterVersion) + o.pdServerConfig.Store(&cfg.PDServerCfg) + } + return nil +} + +func (o *scheduleOption) adjustScheduleCfg(persistentCfg *Config) { + scheduleCfg := o.load().clone() + for i, s := range scheduleCfg.Schedulers { + for _, ps := range persistentCfg.Schedule.Schedulers { + if s.Type == ps.Type && reflect.DeepEqual(s.Args, ps.Args) { + scheduleCfg.Schedulers[i].Disable = ps.Disable + break + } + } + } + restoredSchedulers := make([]SchedulerConfig, 0, len(persistentCfg.Schedule.Schedulers)) + for _, ps := range persistentCfg.Schedule.Schedulers { + needRestore := true + for _, s := range scheduleCfg.Schedulers { + if s.Type == ps.Type && reflect.DeepEqual(s.Args, ps.Args) { + needRestore = false + break + } + } + if needRestore { + restoredSchedulers = append(restoredSchedulers, ps) + } + } + scheduleCfg.Schedulers = append(scheduleCfg.Schedulers, restoredSchedulers...) + persistentCfg.Schedule.Schedulers = scheduleCfg.Schedulers + o.store(scheduleCfg) +} + +func (o *scheduleOption) GetHotRegionCacheHitsThreshold() int { + return int(o.load().HotRegionCacheHitsThreshold) +} + +func (o *scheduleOption) CheckLabelProperty(typ string, labels []*metapb.StoreLabel) bool { + pc := o.labelProperty.Load().(LabelPropertyConfig) + for _, cfg := range pc[typ] { + for _, l := range labels { + if l.Key == cfg.Key && l.Value == cfg.Value { + return true + } + } + } + return false +} + +// Replication provides some help to do replication. +type Replication struct { + replicateCfg atomic.Value +} + +func newReplication(cfg *ReplicationConfig) *Replication { + r := &Replication{} + r.store(cfg) + return r +} + +func (r *Replication) load() *ReplicationConfig { + return r.replicateCfg.Load().(*ReplicationConfig) +} + +func (r *Replication) store(cfg *ReplicationConfig) { + r.replicateCfg.Store(cfg) +} + +// GetMaxReplicas returns the number of replicas for each region. +func (r *Replication) GetMaxReplicas() int { + return int(r.load().MaxReplicas) +} + +// SetMaxReplicas set the replicas for each region. +func (r *Replication) SetMaxReplicas(replicas int) { + c := r.load() + v := c.clone() + v.MaxReplicas = uint64(replicas) + r.store(v) +} + +// GetLocationLabels returns the location labels for each region +func (r *Replication) GetLocationLabels() []string { + return r.load().LocationLabels +} + +// namespaceOption is a wrapper to access the configuration safely. +type namespaceOption struct { + namespaceCfg atomic.Value +} + +func newNamespaceOption(cfg *NamespaceConfig) *namespaceOption { + n := &namespaceOption{} + n.store(cfg) + return n +} + +func (n *namespaceOption) load() *NamespaceConfig { + return n.namespaceCfg.Load().(*NamespaceConfig) +} + +func (n *namespaceOption) store(cfg *NamespaceConfig) { + n.namespaceCfg.Store(cfg) +} + +// GetMaxReplicas returns the number of replicas for each region. +func (n *namespaceOption) GetMaxReplicas() int { + return int(n.load().MaxReplicas) +} + +// GetLeaderScheduleLimit returns the limit for leader schedule. +func (n *namespaceOption) GetLeaderScheduleLimit() uint64 { + return n.load().LeaderScheduleLimit +} + +// GetRegionScheduleLimit returns the limit for region schedule. +func (n *namespaceOption) GetRegionScheduleLimit() uint64 { + return n.load().RegionScheduleLimit +} + +// GetReplicaScheduleLimit returns the limit for replica schedule. +func (n *namespaceOption) GetReplicaScheduleLimit() uint64 { + return n.load().ReplicaScheduleLimit +} + +// GetMergeScheduleLimit returns the limit for merge schedule. +func (n *namespaceOption) GetMergeScheduleLimit() uint64 { + return n.load().MergeScheduleLimit +} + +// GetHotRegionScheduleLimit returns the limit for hot region schedule. +func (n *namespaceOption) GetHotRegionScheduleLimit() uint64 { + return n.load().HotRegionScheduleLimit +} diff --git a/server/region_statistics.go b/server/region_statistics.go new file mode 100644 index 0000000..fc0d8c6 --- /dev/null +++ b/server/region_statistics.go @@ -0,0 +1,222 @@ +// Copyright 2018 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "fmt" + + "github.com/pingcap/pd/server/core" + "github.com/pingcap/pd/server/namespace" +) + +type regionStatisticType uint32 + +const ( + missPeer regionStatisticType = 1 << iota + extraPeer + downPeer + pendingPeer + offlinePeer + incorrectNamespace + learnerPeer +) + +type regionStatistics struct { + opt *scheduleOption + classifier namespace.Classifier + stats map[regionStatisticType]map[uint64]*core.RegionInfo + index map[uint64]regionStatisticType +} + +func newRegionStatistics(opt *scheduleOption, classifier namespace.Classifier) *regionStatistics { + r := ®ionStatistics{ + opt: opt, + classifier: classifier, + stats: make(map[regionStatisticType]map[uint64]*core.RegionInfo), + index: make(map[uint64]regionStatisticType), + } + r.stats[missPeer] = make(map[uint64]*core.RegionInfo) + r.stats[extraPeer] = make(map[uint64]*core.RegionInfo) + r.stats[downPeer] = make(map[uint64]*core.RegionInfo) + r.stats[pendingPeer] = make(map[uint64]*core.RegionInfo) + r.stats[offlinePeer] = make(map[uint64]*core.RegionInfo) + r.stats[incorrectNamespace] = make(map[uint64]*core.RegionInfo) + r.stats[learnerPeer] = make(map[uint64]*core.RegionInfo) + return r +} + +func (r *regionStatistics) getRegionStatsByType(typ regionStatisticType) []*core.RegionInfo { + res := make([]*core.RegionInfo, 0, len(r.stats[typ])) + for _, r := range r.stats[typ] { + res = append(res, r) + } + return res +} + +func (r *regionStatistics) deleteEntry(deleteIndex regionStatisticType, regionID uint64) { + for typ := regionStatisticType(1); typ <= deleteIndex; typ <<= 1 { + if deleteIndex&typ != 0 { + delete(r.stats[typ], regionID) + } + } +} + +func (r *regionStatistics) Observe(region *core.RegionInfo, stores []*core.StoreInfo) { + // Region state. + regionID := region.GetID() + namespace := r.classifier.GetRegionNamespace(region) + var ( + peerTypeIndex regionStatisticType + deleteIndex regionStatisticType + ) + if len(region.GetPeers()) < r.opt.GetMaxReplicas(namespace) { + r.stats[missPeer][regionID] = region + peerTypeIndex |= missPeer + } else if len(region.GetPeers()) > r.opt.GetMaxReplicas(namespace) { + r.stats[extraPeer][regionID] = region + peerTypeIndex |= extraPeer + } + + if len(region.GetDownPeers()) > 0 { + r.stats[downPeer][regionID] = region + peerTypeIndex |= downPeer + } + + if len(region.GetPendingPeers()) > 0 { + r.stats[pendingPeer][regionID] = region + peerTypeIndex |= pendingPeer + } + + if len(region.GetLearners()) > 0 { + r.stats[learnerPeer][regionID] = region + peerTypeIndex |= learnerPeer + } + + for _, store := range stores { + if store.IsOffline() { + peer := region.GetStorePeer(store.GetID()) + if peer != nil { + r.stats[offlinePeer][regionID] = region + peerTypeIndex |= offlinePeer + } + } + ns := r.classifier.GetStoreNamespace(store) + if ns == namespace { + continue + } + r.stats[incorrectNamespace][regionID] = region + peerTypeIndex |= incorrectNamespace + break + } + + if oldIndex, ok := r.index[regionID]; ok { + deleteIndex = oldIndex &^ peerTypeIndex + } + r.deleteEntry(deleteIndex, regionID) + r.index[regionID] = peerTypeIndex +} + +func (r *regionStatistics) clearDefunctRegion(regionID uint64) { + if oldIndex, ok := r.index[regionID]; ok { + r.deleteEntry(oldIndex, regionID) + } +} + +func (r *regionStatistics) Collect() { + regionStatusGauge.WithLabelValues("miss_peer_region_count").Set(float64(len(r.stats[missPeer]))) + regionStatusGauge.WithLabelValues("extra_peer_region_count").Set(float64(len(r.stats[extraPeer]))) + regionStatusGauge.WithLabelValues("down_peer_region_count").Set(float64(len(r.stats[downPeer]))) + regionStatusGauge.WithLabelValues("pending_peer_region_count").Set(float64(len(r.stats[pendingPeer]))) + regionStatusGauge.WithLabelValues("offline_peer_region_count").Set(float64(len(r.stats[offlinePeer]))) + regionStatusGauge.WithLabelValues("incorrect_namespace_region_count").Set(float64(len(r.stats[incorrectNamespace]))) + regionStatusGauge.WithLabelValues("learner_peer_region_count").Set(float64(len(r.stats[learnerPeer]))) +} + +type labelLevelStatistics struct { + regionLabelLevelStats map[uint64]int + labelLevelCounter map[int]int +} + +func newLabelLevelStatistics() *labelLevelStatistics { + return &labelLevelStatistics{ + regionLabelLevelStats: make(map[uint64]int), + labelLevelCounter: make(map[int]int), + } +} + +func (l *labelLevelStatistics) Observe(region *core.RegionInfo, stores []*core.StoreInfo, labels []string) { + regionID := region.GetID() + regionLabelLevel := getRegionLabelIsolationLevel(stores, labels) + if level, ok := l.regionLabelLevelStats[regionID]; ok { + if level == regionLabelLevel { + return + } + l.labelLevelCounter[level]-- + } + l.regionLabelLevelStats[regionID] = regionLabelLevel + l.labelLevelCounter[regionLabelLevel]++ +} + +func (l *labelLevelStatistics) Collect() { + for level, count := range l.labelLevelCounter { + typ := fmt.Sprintf("level_%d", level) + regionLabelLevelGauge.WithLabelValues(typ).Set(float64(count)) + } +} + +func (l *labelLevelStatistics) clearDefunctRegion(regionID uint64) { + if level, ok := l.regionLabelLevelStats[regionID]; ok { + l.labelLevelCounter[level]-- + delete(l.regionLabelLevelStats, regionID) + } +} + +func getRegionLabelIsolationLevel(stores []*core.StoreInfo, labels []string) int { + if len(stores) == 0 || len(labels) == 0 { + return 0 + } + queueStores := [][]*core.StoreInfo{stores} + for level, label := range labels { + newQueueStores := make([][]*core.StoreInfo, 0, len(stores)) + for _, stores := range queueStores { + notIsolatedStores := notIsolatedStoresWithLabel(stores, label) + if len(notIsolatedStores) > 0 { + newQueueStores = append(newQueueStores, notIsolatedStores...) + } + } + queueStores = newQueueStores + if len(queueStores) == 0 { + return level + 1 + } + } + return 0 +} + +func notIsolatedStoresWithLabel(stores []*core.StoreInfo, label string) [][]*core.StoreInfo { + m := make(map[string][]*core.StoreInfo) + for _, s := range stores { + labelValue := s.GetLabelValue(label) + if labelValue == "" { + continue + } + m[labelValue] = append(m[labelValue], s) + } + var res [][]*core.StoreInfo + for _, stores := range m { + if len(stores) > 1 { + res = append(res, stores) + } + } + return res +} diff --git a/server/region_syncer/history_buffer.go b/server/region_syncer/history_buffer.go new file mode 100644 index 0000000..b528638 --- /dev/null +++ b/server/region_syncer/history_buffer.go @@ -0,0 +1,36 @@ +// Copyright 2018 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package syncer + +import ( + "sync" + + "github.com/pingcap/pd/server/core" +) + +const ( + historyKey = "historyIndex" + defaultFlushCount = 100 +) + +type historyBuffer struct { + sync.RWMutex + index uint64 + records []*core.RegionInfo + head int + tail int + size int + kv core.KVBase + flushCount int +} diff --git a/server/region_syncer/server.go b/server/region_syncer/server.go new file mode 100644 index 0000000..7127647 --- /dev/null +++ b/server/region_syncer/server.go @@ -0,0 +1,69 @@ +// Copyright 2018 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package syncer + +import ( + "context" + "sync" + "time" + + "github.com/juju/ratelimit" + "github.com/pingcap/kvproto/pkg/metapb" + "github.com/pingcap/kvproto/pkg/pdpb" + "github.com/pingcap/pd/server/core" +) + +const ( + msgSize = 8 * 1024 * 1024 + defaultBucketRate = 20 * 1024 * 1024 // 20MB/s + defaultBucketCapacity = 20 * 1024 * 1024 // 20MB + maxSyncRegionBatchSize = 100 + syncerKeepAliveInterval = 10 * time.Second + defaultHistoryBufferSize = 10000 +) + +// ClientStream is the client side of the region syncer. +type ClientStream interface { + Recv() (*pdpb.SyncRegionResponse, error) + CloseSend() error +} + +// ServerStream is the server side of the region syncer. +type ServerStream interface { + Send(regions *pdpb.SyncRegionResponse) error +} + +// Server is the abstraction of the syncer storage server. +type Server interface { + Context() context.Context + ClusterID() uint64 + GetMemberInfo() *pdpb.Member + GetLeader() *pdpb.Member + GetStorage() *core.KV + Name() string + GetMetaRegions() []*metapb.Region +} + +// RegionSyncer is used to sync the region information without raft. +type RegionSyncer struct { + sync.RWMutex + streams map[string]ServerStream + ctx context.Context + cancel context.CancelFunc + server Server + closed chan struct{} + wg sync.WaitGroup + history *historyBuffer + limit *ratelimit.Bucket +} diff --git a/server/server.go b/server/server.go index f581a44..5a1261b 100644 --- a/server/server.go +++ b/server/server.go @@ -9,6 +9,9 @@ import ( "github.com/coreos/etcd/clientv3" "github.com/coreos/etcd/embed" "github.com/pingcap/kvproto/pkg/pdpb" + + "github.com/b3fs/wfs/server/core" + "github.com/b3fs/wfs/server/namespace" ) const ( @@ -69,7 +72,6 @@ type Server struct { hbStreams *heartbeatStreams } - // CreateServer creates the UNINITIALIZED pd server with given configuration. func CreateServer(cfg *Config, apiRegister func(*Server) http.Handler) (*Server, error) { log.Infof("PD config - %v", cfg) @@ -691,4 +693,4 @@ func (s *Server) CheckHealth(members []*pdpb.Member) map[uint64]*pdpb.Member { } } return unhealthMembers -} \ No newline at end of file +}