diff options
| author | Valery Piashchynski <[email protected]> | 2020-11-16 15:46:08 +0300 |
|---|---|---|
| committer | GitHub <[email protected]> | 2020-11-16 15:46:08 +0300 |
| commit | 6236aac37bd1661b20400689f66d1e92283c5111 (patch) | |
| tree | eb8a9a4e4717fb4cd6c971b5ce67c53b5f6a0f8c | |
| parent | 0874bcb2f6b284a940ba4f3507eb8c4619c27868 (diff) | |
| parent | 38f6925db27dd94cfbca873901bf932ed1456906 (diff) | |
Merge pull request #392 from spiral/plugin/metricsv2.0.0-alpha18
[RR2] Metrics plugin 2.0
| -rwxr-xr-x | .github/workflows/ci-build.yml | 5 | ||||
| -rwxr-xr-x | .golangci.yml | 13 | ||||
| -rw-r--r-- | Makefile | 5 | ||||
| -rw-r--r-- | codecov.yml | 4 | ||||
| -rwxr-xr-x | go.mod | 11 | ||||
| -rwxr-xr-x | go.sum | 110 | ||||
| -rw-r--r-- | interfaces/log/interface.go (renamed from log/interface.go) | 0 | ||||
| -rw-r--r-- | interfaces/log/zap_adapter.go (renamed from log/zap_adapter.go) | 0 | ||||
| -rw-r--r-- | interfaces/metrics/interface.go | 9 | ||||
| -rw-r--r-- | interfaces/rpc/interface.go | 7 | ||||
| -rw-r--r-- | interfaces/server/interface.go | 17 | ||||
| -rw-r--r-- | plugins/logger/plugin.go | 2 | ||||
| -rw-r--r-- | plugins/logger/tests/plugin.go | 2 | ||||
| -rw-r--r-- | plugins/metrics/config.go | 135 | ||||
| -rw-r--r-- | plugins/metrics/config_test.go | 87 | ||||
| -rw-r--r-- | plugins/metrics/doc.go | 1 | ||||
| -rw-r--r-- | plugins/metrics/plugin.go | 230 | ||||
| -rw-r--r-- | plugins/metrics/rpc.go | 294 | ||||
| -rw-r--r-- | plugins/metrics/tests/.rr-test.yaml | 13 | ||||
| -rw-r--r-- | plugins/metrics/tests/docker-compose.yml | 7 | ||||
| -rw-r--r-- | plugins/metrics/tests/metrics_test.go | 754 | ||||
| -rw-r--r-- | plugins/metrics/tests/plugin1.go | 46 | ||||
| -rwxr-xr-x | plugins/rpc/plugin.go | 36 | ||||
| -rw-r--r-- | plugins/rpc/tests/plugin1.go | 4 | ||||
| -rw-r--r-- | plugins/server/config.go (renamed from plugins/app/config.go) | 10 | ||||
| -rw-r--r-- | plugins/server/plugin.go (renamed from plugins/app/plugin.go) | 24 | ||||
| -rw-r--r-- | plugins/server/tests/configs/.rr-no-app-section.yaml (renamed from plugins/app/tests/configs/.rr-no-app-section.yaml) | 2 | ||||
| -rw-r--r-- | plugins/server/tests/configs/.rr-sockets.yaml (renamed from plugins/app/tests/configs/.rr-sockets.yaml) | 2 | ||||
| -rw-r--r-- | plugins/server/tests/configs/.rr-tcp.yaml (renamed from plugins/app/tests/configs/.rr-tcp.yaml) | 2 | ||||
| -rw-r--r-- | plugins/server/tests/configs/.rr-wrong-command.yaml (renamed from plugins/app/tests/configs/.rr-wrong-command.yaml) | 2 | ||||
| -rw-r--r-- | plugins/server/tests/configs/.rr-wrong-relay.yaml (renamed from plugins/app/tests/configs/.rr-wrong-relay.yaml) | 2 | ||||
| -rw-r--r-- | plugins/server/tests/configs/.rr.yaml (renamed from plugins/app/tests/configs/.rr.yaml) | 2 | ||||
| -rw-r--r-- | plugins/server/tests/plugin_pipes.go (renamed from plugins/app/tests/plugin_pipes.go) | 11 | ||||
| -rw-r--r-- | plugins/server/tests/plugin_sockets.go (renamed from plugins/app/tests/plugin_sockets.go) | 9 | ||||
| -rw-r--r-- | plugins/server/tests/plugin_tcp.go (renamed from plugins/app/tests/plugin_tcp.go) | 9 | ||||
| -rw-r--r-- | plugins/server/tests/server_test.go (renamed from plugins/app/tests/app_test.go) | 16 | ||||
| -rw-r--r-- | plugins/server/tests/socket.php (renamed from plugins/app/tests/socket.php) | 0 | ||||
| -rw-r--r-- | plugins/server/tests/tcp.php (renamed from plugins/app/tests/tcp.php) | 0 | ||||
| -rwxr-xr-x | pool.go | 6 | ||||
| -rwxr-xr-x | static_pool.go | 230 | ||||
| -rwxr-xr-x | static_pool_test.go | 26 | ||||
| -rwxr-xr-x | supervisor_pool.go | 6 | ||||
| -rw-r--r-- | supervisor_test.go | 6 | ||||
| -rwxr-xr-x | util/events.go | 10 | ||||
| -rwxr-xr-x | worker_watcher.go | 6 |
45 files changed, 1968 insertions, 205 deletions
diff --git a/.github/workflows/ci-build.yml b/.github/workflows/ci-build.yml index 99eb8834..06e260ce 100755 --- a/.github/workflows/ci-build.yml +++ b/.github/workflows/ci-build.yml @@ -70,13 +70,14 @@ jobs: go test -v -race ./plugins/rpc/tests -tags=debug -coverprofile=rpc.txt -covermode=atomic go test -v -race ./plugins/config/tests -tags=debug -coverprofile=plugin_config.txt -covermode=atomic go test -v -race ./plugins/logger/tests -tags=debug -coverprofile=logger.txt -covermode=atomic - go test -v -race ./plugins/app/tests -tags=debug -coverprofile=app.txt -covermode=atomic + go test -v -race ./plugins/server/tests -tags=debug -coverprofile=server.txt -covermode=atomic + go test -v -race ./plugins/metrics/tests -tags=debug -coverprofile=metrics.txt -covermode=atomic - name: Run code coverage uses: codecov/codecov-action@v1 with: token: ${{ secrets.CODECOV_TOKEN }} - files: lib.txt, rpc_config.txt, rpc.txt, plugin_config.txt, logger.txt, app.txt + files: lib.txt, rpc_config.txt, rpc.txt, plugin_config.txt, logger.txt, server.txt, metrics.txt flags: unittests name: codecov-umbrella fail_ci_if_error: false diff --git a/.golangci.yml b/.golangci.yml index 1d9eb013..38dd313b 100755 --- a/.golangci.yml +++ b/.golangci.yml @@ -1,14 +1,15 @@ linters: - # please, do not use `enable-all`: it's deprecated and will be removed soon. - # inverted configuration with `enable-all` and `disable` is not scalable during updates of golangci-lint + skip-files: + - ".*\\test\\.go$" + - lib/bad.go disable-all: true enable: - bodyclose - depguard - dogsled -# - dupl + # - dupl - gochecknoinits - - goconst + # - goconst - gocritic - gocyclo - gofmt @@ -27,10 +28,10 @@ linters: - scopelint - staticcheck - structcheck -# - stylecheck + # - stylecheck - typecheck - unconvert - - unparam + # - unparam # - unused - varcheck - whitespace @@ -3,5 +3,6 @@ test: go test -v -race -cover ./plugins/rpc -tags=debug go test -v -race -cover ./plugins/rpc/tests -tags=debug go test -v -race -cover ./plugins/config/tests -tags=debug - go test -v -race -cover ./plugins/app/tests -tags=debug - go test -v -race -cover ./plugins/logger/tests -tags=debug
\ No newline at end of file + go test -v -race -cover ./plugins/server/tests -tags=debug + go test -v -race -cover ./plugins/logger/tests -tags=debug + go test -v -race -cover ./plugins/metrics/tests -tags=debug
\ No newline at end of file diff --git a/codecov.yml b/codecov.yml new file mode 100644 index 00000000..3f7eef9a --- /dev/null +++ b/codecov.yml @@ -0,0 +1,4 @@ +coverage: + status: + project: on + patch: off
\ No newline at end of file @@ -3,19 +3,20 @@ module github.com/spiral/roadrunner/v2 go 1.15 require ( - github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d // indirect github.com/fatih/color v1.10.0 - github.com/go-ole/go-ole v1.2.4 // indirect github.com/json-iterator/go v1.1.10 github.com/pkg/errors v0.9.1 + github.com/prometheus/client_golang v1.7.1 github.com/shirou/gopsutil v3.20.10+incompatible github.com/spf13/viper v1.7.1 - github.com/spiral/endure v1.0.0-beta15 + github.com/spiral/endure v1.0.0-beta19 github.com/spiral/errors v1.0.4 github.com/spiral/goridge/v2 v2.4.6 + github.com/spiral/roadrunner v1.8.4 github.com/stretchr/testify v1.6.1 github.com/valyala/tcplisten v0.0.0-20161114210144-ceec8f93295a go.uber.org/multierr v1.6.0 go.uber.org/zap v1.16.0 - golang.org/x/sync v0.0.0-20190423024810-112230192c58 -) + golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e + golang.org/x/sys v0.0.0-20200615200032-f1bc736245b1 +)
\ No newline at end of file @@ -14,32 +14,48 @@ dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7 github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo= +github.com/NYTimes/gziphandler v1.1.1/go.mod h1:n/CVRwUEOgIxrgPvAQhUUr9oeUtvrhMomdKFjzJNB0c= github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU= github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d h1:G0m3OIz70MZUWq3EgK3CesDbo8upS2Vm9/P3FtgI+Jk= github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg= github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= +github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= +github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o= +github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8= github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY= github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8= github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q= +github.com/beorn7/perks v1.0.0 h1:HWo1m869IqiPhD389kmkxeTalrjNbbJTC8LXupb+sl0= github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8= +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs= github.com/bketelsen/crypt v0.0.3-0.20200106085610-5cbc8cc4026c/go.mod h1:MKsuJmJgSg28kpZDP6UIiPt0e0Oz0kqKNGyRaWEPv84= +github.com/buger/goterm v0.0.0-20181115115552-c206103e1f37/go.mod h1:u9UyCz2eTrSGy6fbupqJ54eY5c4IC8gREQ1053dK12U= +github.com/cenkalti/backoff/v4 v4.0.0/go.mod h1:eEew/i+1Q6OrCDZh3WiXYv3+nJwBASZ8Bog/87DQnVg= github.com/cenkalti/backoff/v4 v4.1.0 h1:c8LkOFQTzuO0WBM/ae5HdGQuZPfPxp7lqBRwQRm4fSc= github.com/cenkalti/backoff/v4 v4.1.0/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw= +github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko= github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc= +github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+qY= +github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw= github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk= +github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE= github.com/coreos/etcd v3.3.13+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE= +github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk= github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk= github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4= github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA= +github.com/cpuguy83/go-md2man/v2 v2.0.0/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU= 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/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ= github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no= +github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk= github.com/fatih/color v1.7.0 h1:DkWD4oS2D8LGGgTQ6IvwJJXSL5Vp2ffcQg58nFV38Ys= github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4= github.com/fatih/color v1.10.0 h1:s36xzo75JdqLaaWoiEHk767eHiwo0598uUxyfiPkDsg= @@ -48,11 +64,14 @@ github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04= github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU= +github.com/go-ini/ini v1.38.1/go.mod h1:ByCAeIL28uOIIG0E3PJtZPDL8WnHpFKFOtgjp+3Ies8= github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as= +github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as= github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE= github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk= github.com/go-ole/go-ole v1.2.4 h1:nNBDSCOigTSiarFpYE9J/KtEA1IOW4CNeqT9TQDqCxI= github.com/go-ole/go-ole v1.2.4/go.mod h1:XCwSNxSkXRo4vlyPy93sltvi/qJq0jqQhjqQNIwKuxM= +github.com/go-restit/lzjson v0.0.0-20161206095556-efe3c53acc68/go.mod h1:7vXSKQt83WmbPeyVjCfNT9YDJ5BUFmcwFsEjI9SCvYM= github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY= github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ= github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4= @@ -63,11 +82,22 @@ github.com/golang/mock v1.2.0/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfb github.com/golang/mock v1.3.1/go.mod h1:sBzyDLLjw3U8JLTeZvSv8jJB+tU5PVekmnlKIyFUx0Y= github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= +github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs= github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= +github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8= +github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA= +github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs= +github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:WU3c8KckQ9AFe+yFwt9sWVRKCVIyN9cPHBJSNnbL67w= +github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0= +github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0= +github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI= github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ= github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ= github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M= github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= +github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= +github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4= +github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs= github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc= @@ -75,8 +105,10 @@ github.com/google/pprof v0.0.0-20190515194954-54271f7e092f/go.mod h1:zfwlbNMJ+OI github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI= github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg= github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk= +github.com/gopherjs/gopherjs v0.0.0-20180825215210-0210a2f0f73c/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY= github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8= github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY= +github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ= github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs= github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk= @@ -102,37 +134,48 @@ github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO github.com/hashicorp/mdns v1.0.0/go.mod h1:tL+uN++7HEJ6SQLQ2/p+z2pH24WQKWjBPkE0mNTz8vQ= github.com/hashicorp/memberlist v0.1.3/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I= github.com/hashicorp/serf v0.8.2/go.mod h1:6hOLApaqBFA1NXqRQAsxw9QxuDEvNxSQRwA/JwenrHc= +github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8= github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo= github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU= github.com/json-iterator/go v1.1.10 h1:Kz6Cvnvv2wGdaG/V8yMvfkmNiXq9Ya2KUv4rouJJr68= github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4= github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU= +github.com/jtolds/gls v4.2.1+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU= github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo= github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU= github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w= github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGiHgQ4OO8tzTaLawm8vnODuwDk= github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= +github.com/konsorten/go-windows-terminal-sequences v1.0.3 h1:CE8S1cTafDpPvMhIxNJKvHsGVBgn1xWYf1NbHQhywc8= +github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc= github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ= github.com/magiconair/properties v1.8.1 h1:ZC2Vc7/ZFkGmsVC9KvOjumD+G5lXy2RtTKyzRKO2BQ4= github.com/magiconair/properties v1.8.1/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ= github.com/mattn/go-colorable v0.0.9 h1:UVL0vNpWh04HeJXV0KLcaT7r06gOH2l4OW6ddYRUIY4= github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU= +github.com/mattn/go-colorable v0.1.7/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc= github.com/mattn/go-colorable v0.1.8 h1:c1ghPdyEDarC70ftn0y+A/Ee++9zz8ljHG1b13eJ0s8= github.com/mattn/go-colorable v0.1.8/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc= github.com/mattn/go-isatty v0.0.3 h1:ns/ykhmWi7G9O+8a448SecJU3nSMBXJfqQkl0upE1jI= github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4= github.com/mattn/go-isatty v0.0.12 h1:wuysRhFDzyxgEmMf5xjvJ2M9dZoWAXNNr5LSBS7uHXY= github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU= +github.com/mattn/go-runewidth v0.0.7/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI= +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/mgutz/ansi v0.0.0-20200706080929-d51e80ef957d/go.mod h1:01TrycV0kFyexm33Z7vhZRXopbI8J3TDReVlkTgMUxE= github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg= github.com/mitchellh/cli v1.0.0/go.mod h1:hNIlj7HEI86fIcpObd7a0FcrxTWetlwJDGcceTlRvqc= github.com/mitchellh/go-homedir v1.0.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0= +github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0= github.com/mitchellh/go-testing-interface v1.0.0/go.mod h1:kRemZodwjscx+RGhAo8eIhFbs2+BFgRtFPeD/KE+zxI= github.com/mitchellh/gox v0.4.0/go.mod h1:Sd9lOJ0+aimLBi73mGofS1ycjY8lL3uZM3JPS42BGNg= github.com/mitchellh/iochan v1.0.0/go.mod h1:JwYml1nuB7xOzsp52dPpHFffvOCDupsG0QubkSMEySY= @@ -147,6 +190,7 @@ github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9 github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U= +github.com/olekukonko/tablewriter v0.0.4/go.mod h1:zq6QwlOf5SlnkVbMSr5EoBv3636FWnp+qbPhuoO21uA= github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc= github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181zc= github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic= @@ -158,23 +202,47 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndrE9hABlRI= github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw= +github.com/prometheus/client_golang v0.9.3 h1:9iH4JKXLzFbOAdtqv/a+j8aewx2Y8lAjAydhbaScPF8= github.com/prometheus/client_golang v0.9.3/go.mod h1:/TN21ttK/J9q6uSwhBd54HahCDft0ttaMvbicHlPoso= +github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo= +github.com/prometheus/client_golang v1.7.1 h1:NTGy1Ja9pByO+xAeH/qiWnLrKtr3hJPNjaVUwnjpdpA= +github.com/prometheus/client_golang v1.7.1/go.mod h1:PY5Wy2awLA44sXw4AOSfFBetzPP4j5+D6mVACh+pe2M= github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo= +github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90 h1:S/YWwWx/RA8rT8tKFRuGUZhuA90OyIBpPCXkcbwU8DE= github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= +github.com/prometheus/client_model v0.2.0 h1:uq5h0d+GuxiXLJLNABMgp2qUWDPiLvgCzz2dUR+/W/M= +github.com/prometheus/client_model v0.2.0/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= github.com/prometheus/common v0.0.0-20181113130724-41aa239b4cce/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro= +github.com/prometheus/common v0.4.0 h1:7etb9YClo3a6HjLzfl6rIQaU+FDfi0VSX39io3aQ+DM= github.com/prometheus/common v0.4.0/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4= +github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4= +github.com/prometheus/common v0.10.0 h1:RyRA7RzGXQZiW+tGMr7sxa85G1z0yOpM1qq5c8lNawc= +github.com/prometheus/common v0.10.0/go.mod h1:Tlit/dnDKsSWFlCLTWaA1cyBgKHSMdTB80sz/V91rCo= github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk= +github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084 h1:sofwID9zm4tzrgykg80hfFph1mryUeLRsUfoocVVmRY= github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA= +github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA= +github.com/prometheus/procfs v0.1.3 h1:F0+tqvhOksq22sc6iCHF5WGlWjdwj92p0udFh1VFBS8= +github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU= github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU= github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg= github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4= +github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts= github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc= +github.com/shirou/gopsutil v2.20.7+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA= github.com/shirou/gopsutil v3.20.10+incompatible h1:kQuRhh6h6y4luXvnmtu/lJEGtdJ3q8lbu9NQY99GP+o= github.com/shirou/gopsutil v3.20.10+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA= +github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc= +github.com/sirupsen/logrus v1.2.0 h1:juTguoYk5qI21pwyTXY3B3Y5cOTH3ZUyZCg1v/mihuo= github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo= +github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE= +github.com/sirupsen/logrus v1.6.0 h1:UBcNElsrwanuuMsnGSlYmtmgbb23qDR5dG+6X6Oo89I= +github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88= +github.com/smartystreets/assertions v0.0.0-20180820201707-7c9eb446e3cf/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc= github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d h1:zE9ykElWQ6/NYmHa3jpm/yHnI4xSofP+UP6SpjHcSeM= github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc= +github.com/smartystreets/goconvey v0.0.0-20180222194500-ef6db91d284a/go.mod h1:XDJAKZRPZ1CvBcN2aX5YOUTYGHki24fSF0Iv48Ibg0s= github.com/smartystreets/goconvey v1.6.4 h1:fv0U8FUIMPNf1L9lnHLvLhgicrIVChEkdzIKYqbNC9s= github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA= github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM= @@ -183,20 +251,25 @@ github.com/spf13/afero v1.1.2 h1:m8/z1t7/fwjysjQRYbP0RD+bUIF/8tJwPdEZsI83ACI= github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ= github.com/spf13/cast v1.3.0 h1:oget//CVOEoFewqQxwr0Ej5yjygnqGkvggSE/gB35Q8= github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE= +github.com/spf13/cobra v1.0.0/go.mod h1:/6GTrnGXV9HjY+aR4k0oJ5tcvakLuG6EuKReYlHNrgE= github.com/spf13/jwalterweatherman v1.0.0 h1:XHEdyB+EcvlqZamSM4ZOMGlc93t6AcsBEu9Gc1vn7yk= github.com/spf13/jwalterweatherman v1.0.0/go.mod h1:cQK4TGJAtQXfYWX+Ddv3mKDzgVb68N+wFjFa4jdeBTo= github.com/spf13/pflag v1.0.3 h1:zPAT6CGy6wXeQ7NtTnaTerfKOsV6V6F8agHXFiazDkg= github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4= +github.com/spf13/viper v1.4.0/go.mod h1:PTJ7Z/lr49W6bUbkmS1V3by4uWynFiR9p7+dSq/yZzE= github.com/spf13/viper v1.7.1 h1:pM5oEahlgWv/WnHXpgbKz7iLIxRf65tye2Ci+XFK5sk= github.com/spf13/viper v1.7.1/go.mod h1:8WkrPz2fc9jxqZNCJI/76HCieCp4Q8HaLFoCha5qpdg= -github.com/spiral/endure v1.0.0-beta15 h1:pBFn9LKQLPSzrG7kGE30T0VEjp2A6yT6p2BRRUpN978= -github.com/spiral/endure v1.0.0-beta15/go.mod h1:iGh1Zf1cckkJa5J9Obm8d/96kKYvlJBv/D0iHOuYWLQ= -github.com/spiral/errors v1.0.1 h1:OyKLwQH+42hhaRYuXGzfPKCFOmawA/PYXTY9wsK99n4= -github.com/spiral/errors v1.0.1/go.mod h1:SwMSZVdZkkJVgXNNafccqOaxWg0XPzVU/dEdUEInE0o= +github.com/spiral/endure v1.0.0-beta18 h1:SJOh8b6G6AfXg2RgKvKnLE03Ep8bXGFiEVcfc/F41WI= +github.com/spiral/endure v1.0.0-beta18/go.mod h1:qm3evrNggh26QQhwln2uH/1KJQInFZKJZeD5Yvm2k6Y= +github.com/spiral/endure v1.0.0-beta19 h1:dQpFfiFmeh9E+oErAlM9f4ojZJN1wjEy75xuuQmIdGM= +github.com/spiral/endure v1.0.0-beta19/go.mod h1:qm3evrNggh26QQhwln2uH/1KJQInFZKJZeD5Yvm2k6Y= +github.com/spiral/errors v1.0.2/go.mod h1:SwMSZVdZkkJVgXNNafccqOaxWg0XPzVU/dEdUEInE0o= github.com/spiral/errors v1.0.4 h1:Y6Bop9GszdDh+Dn3s5aqsGebNLydqZ1F6OdOIQ9EpU0= github.com/spiral/errors v1.0.4/go.mod h1:SwMSZVdZkkJVgXNNafccqOaxWg0XPzVU/dEdUEInE0o= github.com/spiral/goridge/v2 v2.4.6 h1:9u/mrxCtOSy0lnumrpPCSOlGBX/Vprid/hFsnzWrd6k= github.com/spiral/goridge/v2 v2.4.6/go.mod h1:mYjL+Ny7nVfLqjRwIYV2pUSQ61eazvVclHII6FfZfYc= +github.com/spiral/roadrunner v1.8.4 h1:ertz4272GMOf7R/br/GRhvC0zqXodOwstC26Zao4NoI= +github.com/spiral/roadrunner v1.8.4/go.mod h1:Q1al1YGjs7ZHVkAA7+gUKM0rwk6XWG07G0UjyjjuK+0= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= @@ -207,9 +280,12 @@ github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/ github.com/subosito/gotenv v1.2.0 h1:Slr1R9HxAlEKefgq5jn9U+DnETlIUa6HfgEzj0g5d7s= github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw= github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U= +github.com/ugorji/go v1.1.4/go.mod h1:uQMGLiO92mf5W77hV/PUCpI3pbzQx3CRekS0kk+RGrc= github.com/valyala/tcplisten v0.0.0-20161114210144-ceec8f93295a h1:0R4NLDRDZX6JcmhJgXi5E4b8Wg84ihbmUKp/GvSPEzc= github.com/valyala/tcplisten v0.0.0-20161114210144-ceec8f93295a/go.mod h1:v3UYOV9WzVtRmSR+PDvWpU/qWl4Wa5LApYYX4ZtKbio= github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU= +github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q= +github.com/yookoala/gofast v0.4.0/go.mod h1:rfbkoKaQG1bnuTUZcmV3vAlnfpF4FTq8WbQJf2vcpg8= go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU= go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU= go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8= @@ -229,6 +305,7 @@ golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnf golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= +golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5 h1:58fnuSXlxZmFdJyvtTFVmVhcMLU6v5fEb/ok4wyqtNU= golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= @@ -261,8 +338,11 @@ golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190501004415-9ce7a6920f09/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190503192946-f4e77d36d62c/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190522155817-f3200d17e092/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks= golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks= +golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20200222125558-5a598a2470a0/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= @@ -272,6 +352,8 @@ golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190423024810-112230192c58 h1:8gQV6CLnAEikrhgkHFbMAEhagSSnXWGV915qUMm9mrU= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e h1:vcxGaoTs7kV8m5Np9uUNQin4BrLOthgV7252N8V+FwY= +golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -281,14 +363,18 @@ golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5h golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190502145724-3ef323f4f1fd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190507160741-ecd444e8653b/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190606165138-5da285871e9c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0 h1:HyfiK1WMnHj5FXFXatD+Qs1A/xC2Run6RzeW1SyHxpc= golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200106162015-b016eb3dc98e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae h1:/WDfKMnPU+m5M4xB+6x4kaepxRw6jWvR5iDRdvjHgy8= golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200615200032-f1bc736245b1 h1:ogLJMz+qpzav7lGMh10LMvAkM/fAoGlaiiHYiFYdm80= +golang.org/x/sys v0.0.0-20200615200032-f1bc736245b1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs= @@ -296,6 +382,7 @@ golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20180726210403-bfb5194568d3/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY= @@ -316,6 +403,8 @@ golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtn golang.org/x/tools v0.0.0-20191112195655-aa38f8e97acc h1:NCy3Ohtk6Iny5V/reW2Ktypo4zIpWBdRJ1uFMjBxdg8= golang.org/x/tools v0.0.0-20191112195655-aa38f8e97acc/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= google.golang.org/api v0.4.0/go.mod h1:8k5glujaEP+g9n7WNsDg8QP6cUVNI86fCNMcbazEtwE= google.golang.org/api v0.7.0/go.mod h1:WtwebWUNSVBH/HAw79HIFXZNqEvBhG+Ra+ax0hx3E3M= google.golang.org/api v0.8.0/go.mod h1:o4eAsZoiT+ibD93RtjEohWalFOjRDx6CVaqeizhEnKg= @@ -336,12 +425,23 @@ google.golang.org/genproto v0.0.0-20190911173649-1774047e7e51/go.mod h1:IbNlFCBr google.golang.org/genproto v0.0.0-20191108220845-16a3f7862a1a/go.mod h1:n3cpQtvxv34hfy77yVDNjmbRyujviMdxYliBSkLhpCc= google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c= google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38= +google.golang.org/grpc v1.21.0/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM= google.golang.org/grpc v1.21.1/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM= +google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8= +google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0= +google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM= +google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miEFZTKqfCUM6K7xSMQL9OKL/b6hQv+e19PK+JZNE= +google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo= +google.golang.org/protobuf v1.23.0 h1:4MY060fB1DLGMB/7MBTLnwQUY6+F09GEiz6SsrNqyzM= +google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo= +gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI= +gopkg.in/ini.v1 v1.38.2/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= gopkg.in/ini.v1 v1.51.0 h1:AQvPpx3LzTDM0AjnIRlVFwFFGC+npRopjZxLJj6gdno= gopkg.in/ini.v1 v1.51.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= gopkg.in/resty.v1 v1.12.0/go.mod h1:mDo4pnntr5jdWRML875a/NmxYqAlA73dVijT2AXvQQo= @@ -350,6 +450,8 @@ gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.2.4 h1:/eiJrUcujPVeJ3xlSWaiNi3uSVmDGBK1pDHUHAnao1I= gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v2 v2.2.5 h1:ymVxjfMaHvXD8RqPRmzHHsB3VvucivSkIAvJFDI5O3c= +gopkg.in/yaml.v2 v2.2.5/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= diff --git a/log/interface.go b/interfaces/log/interface.go index e2f2ce31..e2f2ce31 100644 --- a/log/interface.go +++ b/interfaces/log/interface.go diff --git a/log/zap_adapter.go b/interfaces/log/zap_adapter.go index 65f8d04b..65f8d04b 100644 --- a/log/zap_adapter.go +++ b/interfaces/log/zap_adapter.go diff --git a/interfaces/metrics/interface.go b/interfaces/metrics/interface.go new file mode 100644 index 00000000..505c3d7b --- /dev/null +++ b/interfaces/metrics/interface.go @@ -0,0 +1,9 @@ +package metrics + +import ( + "github.com/prometheus/client_golang/prometheus" +) + +type StatProvider interface { + MetricsCollector() []prometheus.Collector +} diff --git a/interfaces/rpc/interface.go b/interfaces/rpc/interface.go new file mode 100644 index 00000000..683fd2ec --- /dev/null +++ b/interfaces/rpc/interface.go @@ -0,0 +1,7 @@ +package rpc + +// RPCer declares the ability to create set of public RPC methods. +type RPCer interface { + // Provides RPC methods for the given service. + RPC() interface{} +} diff --git a/interfaces/server/interface.go b/interfaces/server/interface.go new file mode 100644 index 00000000..51d172cb --- /dev/null +++ b/interfaces/server/interface.go @@ -0,0 +1,17 @@ +package server + +import ( + "context" + "os/exec" + + "github.com/spiral/roadrunner/v2" +) + +type Env map[string]string + +// WorkerFactory creates workers for the application. +type WorkerFactory interface { + CmdFactory(env Env) (func() *exec.Cmd, error) + NewWorker(ctx context.Context, env Env) (roadrunner.WorkerBase, error) + NewWorkerPool(ctx context.Context, opt roadrunner.PoolConfig, env Env) (roadrunner.Pool, error) +} diff --git a/plugins/logger/plugin.go b/plugins/logger/plugin.go index f05d0ff0..0a8485d9 100644 --- a/plugins/logger/plugin.go +++ b/plugins/logger/plugin.go @@ -2,7 +2,7 @@ package logger import ( "github.com/spiral/endure" - "github.com/spiral/roadrunner/v2/log" + "github.com/spiral/roadrunner/v2/interfaces/log" "github.com/spiral/roadrunner/v2/plugins/config" "go.uber.org/zap" ) diff --git a/plugins/logger/tests/plugin.go b/plugins/logger/tests/plugin.go index 75d2736d..32238f63 100644 --- a/plugins/logger/tests/plugin.go +++ b/plugins/logger/tests/plugin.go @@ -2,7 +2,7 @@ package tests import ( "github.com/spiral/errors" - "github.com/spiral/roadrunner/v2/log" + "github.com/spiral/roadrunner/v2/interfaces/log" "github.com/spiral/roadrunner/v2/plugins/config" ) diff --git a/plugins/metrics/config.go b/plugins/metrics/config.go new file mode 100644 index 00000000..933b7eb8 --- /dev/null +++ b/plugins/metrics/config.go @@ -0,0 +1,135 @@ +package metrics + +import ( + "fmt" + + "github.com/prometheus/client_golang/prometheus" +) + +// Config configures metrics service. +type Config struct { + // Address to listen + Address string + + // Collect define application specific metrics. + Collect map[string]Collector +} + +type NamedCollector struct { + // Name of the collector + Name string `json:"name"` + + // Collector structure + Collector `json:"collector"` +} + +// CollectorType represents prometheus collector types +type CollectorType string + +const ( + // Histogram type + Histogram CollectorType = "histogram" + + // Gauge type + Gauge CollectorType = "gauge" + + // Counter type + Counter CollectorType = "counter" + + // Summary type + Summary CollectorType = "summary" +) + +// Collector describes single application specific metric. +type Collector struct { + // Namespace of the metric. + Namespace string `json:"namespace"` + // Subsystem of the metric. + Subsystem string `json:"subsystem"` + // Collector type (histogram, gauge, counter, summary). + Type CollectorType `json:"type"` + // Help of collector. + Help string `json:"help"` + // Labels for vectorized metrics. + Labels []string `json:"labels"` + // Buckets for histogram metric. + Buckets []float64 `json:"buckets"` +} + +// register application specific metrics. +func (c *Config) getCollectors() (map[string]prometheus.Collector, error) { + if c.Collect == nil { + return nil, nil + } + + collectors := make(map[string]prometheus.Collector) + + for name, m := range c.Collect { + var collector prometheus.Collector + switch m.Type { + case Histogram: + opts := prometheus.HistogramOpts{ + Name: name, + Namespace: m.Namespace, + Subsystem: m.Subsystem, + Help: m.Help, + Buckets: m.Buckets, + } + + if len(m.Labels) != 0 { + collector = prometheus.NewHistogramVec(opts, m.Labels) + } else { + collector = prometheus.NewHistogram(opts) + } + case Gauge: + opts := prometheus.GaugeOpts{ + Name: name, + Namespace: m.Namespace, + Subsystem: m.Subsystem, + Help: m.Help, + } + + if len(m.Labels) != 0 { + collector = prometheus.NewGaugeVec(opts, m.Labels) + } else { + collector = prometheus.NewGauge(opts) + } + case Counter: + opts := prometheus.CounterOpts{ + Name: name, + Namespace: m.Namespace, + Subsystem: m.Subsystem, + Help: m.Help, + } + + if len(m.Labels) != 0 { + collector = prometheus.NewCounterVec(opts, m.Labels) + } else { + collector = prometheus.NewCounter(opts) + } + case Summary: + opts := prometheus.SummaryOpts{ + Name: name, + Namespace: m.Namespace, + Subsystem: m.Subsystem, + Help: m.Help, + } + + if len(m.Labels) != 0 { + collector = prometheus.NewSummaryVec(opts, m.Labels) + } else { + collector = prometheus.NewSummary(opts) + } + default: + return nil, fmt.Errorf("invalid metric type `%s` for `%s`", m.Type, name) + } + + collectors[name] = collector + } + + return collectors, nil +} + +func (c *Config) InitDefaults() { + +} diff --git a/plugins/metrics/config_test.go b/plugins/metrics/config_test.go new file mode 100644 index 00000000..24c8406c --- /dev/null +++ b/plugins/metrics/config_test.go @@ -0,0 +1,87 @@ +package metrics + +import ( + "bytes" + "testing" + + json "github.com/json-iterator/go" + "github.com/prometheus/client_golang/prometheus" + "github.com/stretchr/testify/assert" +) + +func Test_Config_Hydrate_Error1(t *testing.T) { + cfg := `{"request": {"From": "Something"}}` + c := &Config{} + f := new(bytes.Buffer) + f.WriteString(cfg) + + err := json.Unmarshal(f.Bytes(), &c) + if err != nil { + t.Fatal(err) + } +} + +func Test_Config_Hydrate_Error2(t *testing.T) { + cfg := `{"dir": "/dir/"` + c := &Config{} + + f := new(bytes.Buffer) + f.WriteString(cfg) + + err := json.Unmarshal(f.Bytes(), &c) + assert.Error(t, err) +} + +func Test_Config_Metrics(t *testing.T) { + cfg := `{ +"collect":{ + "metric1":{"type": "gauge"}, + "metric2":{ "type": "counter"}, + "metric3":{"type": "summary"}, + "metric4":{"type": "histogram"} +} +}` + c := &Config{} + f := new(bytes.Buffer) + f.WriteString(cfg) + + err := json.Unmarshal(f.Bytes(), &c) + if err != nil { + t.Fatal(err) + } + + m, err := c.getCollectors() + assert.NoError(t, err) + + assert.IsType(t, prometheus.NewGauge(prometheus.GaugeOpts{}), m["metric1"]) + assert.IsType(t, prometheus.NewCounter(prometheus.CounterOpts{}), m["metric2"]) + assert.IsType(t, prometheus.NewSummary(prometheus.SummaryOpts{}), m["metric3"]) + assert.IsType(t, prometheus.NewHistogram(prometheus.HistogramOpts{}), m["metric4"]) +} + +func Test_Config_MetricsVector(t *testing.T) { + cfg := `{ +"collect":{ + "metric1":{"type": "gauge","labels":["label"]}, + "metric2":{ "type": "counter","labels":["label"]}, + "metric3":{"type": "summary","labels":["label"]}, + "metric4":{"type": "histogram","labels":["label"]} +} +}` + c := &Config{} + f := new(bytes.Buffer) + f.WriteString(cfg) + + err := json.Unmarshal(f.Bytes(), &c) + if err != nil { + t.Fatal(err) + } + + m, err := c.getCollectors() + assert.NoError(t, err) + + assert.IsType(t, prometheus.NewGaugeVec(prometheus.GaugeOpts{}, []string{}), m["metric1"]) + assert.IsType(t, prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{}), m["metric2"]) + assert.IsType(t, prometheus.NewSummaryVec(prometheus.SummaryOpts{}, []string{}), m["metric3"]) + assert.IsType(t, prometheus.NewHistogramVec(prometheus.HistogramOpts{}, []string{}), m["metric4"]) +} diff --git a/plugins/metrics/doc.go b/plugins/metrics/doc.go new file mode 100644 index 00000000..1abe097a --- /dev/null +++ b/plugins/metrics/doc.go @@ -0,0 +1 @@ +package metrics diff --git a/plugins/metrics/plugin.go b/plugins/metrics/plugin.go new file mode 100644 index 00000000..3fd42ee4 --- /dev/null +++ b/plugins/metrics/plugin.go @@ -0,0 +1,230 @@ +package metrics + +import ( + "context" + "crypto/tls" + "net/http" + "sync" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promhttp" + "github.com/spiral/endure" + "github.com/spiral/errors" + "github.com/spiral/roadrunner/v2/interfaces/log" + "github.com/spiral/roadrunner/v2/interfaces/metrics" + "github.com/spiral/roadrunner/v2/plugins/config" + "golang.org/x/sys/cpu" +) + +const ( + // ID declares public service name. + ServiceName = "metrics" + // maxHeaderSize declares max header size for prometheus server + maxHeaderSize = 1024 * 1024 * 100 // 104MB +) + +type statsProvider struct { + collectors []prometheus.Collector + name string +} + +// Plugin to manage application metrics using Prometheus. +type Plugin struct { + cfg Config + log log.Logger + mu sync.Mutex // all receivers are pointers + http *http.Server + collectors sync.Map // all receivers are pointers + registry *prometheus.Registry +} + +// Init service. +func (m *Plugin) Init(cfg config.Configurer, log log.Logger) error { + const op = errors.Op("Metrics Init") + err := cfg.UnmarshalKey(ServiceName, &m.cfg) + if err != nil { + return err + } + + // TODO figure out what is Init + m.cfg.InitDefaults() + + m.log = log + m.registry = prometheus.NewRegistry() + + // Default + err = m.registry.Register(prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{})) + if err != nil { + return errors.E(op, err) + } + + // Default + err = m.registry.Register(prometheus.NewGoCollector()) + if err != nil { + return errors.E(op, err) + } + + collectors, err := m.cfg.getCollectors() + if err != nil { + return errors.E(op, err) + } + + // Register invocation will be later in the Serve method + for k, v := range collectors { + m.collectors.Store(k, statsProvider{ + collectors: []prometheus.Collector{v}, + name: k, + }) + } + return nil +} + +// Register new prometheus collector. +func (m *Plugin) Register(c prometheus.Collector) error { + return m.registry.Register(c) +} + +// Serve prometheus metrics service. +func (m *Plugin) Serve() chan error { + errCh := make(chan error, 1) + m.collectors.Range(func(key, value interface{}) bool { + // key - name + // value - statsProvider struct + c := value.(statsProvider) + for _, v := range c.collectors { + if err := m.registry.Register(v); err != nil { + errCh <- err + return false + } + } + + return true + }) + + var topCipherSuites []uint16 + var defaultCipherSuitesTLS13 []uint16 + + hasGCMAsmAMD64 := cpu.X86.HasAES && cpu.X86.HasPCLMULQDQ + hasGCMAsmARM64 := cpu.ARM64.HasAES && cpu.ARM64.HasPMULL + // Keep in sync with crypto/aes/cipher_s390x.go. + hasGCMAsmS390X := cpu.S390X.HasAES && cpu.S390X.HasAESCBC && cpu.S390X.HasAESCTR && (cpu.S390X.HasGHASH || cpu.S390X.HasAESGCM) + + hasGCMAsm := hasGCMAsmAMD64 || hasGCMAsmARM64 || hasGCMAsmS390X + + if hasGCMAsm { + // If AES-GCM hardware is provided then prioritise AES-GCM + // cipher suites. + topCipherSuites = []uint16{ + tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256, + tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384, + tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256, + tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384, + tls.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305, + tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305, + } + defaultCipherSuitesTLS13 = []uint16{ + tls.TLS_AES_128_GCM_SHA256, + tls.TLS_CHACHA20_POLY1305_SHA256, + tls.TLS_AES_256_GCM_SHA384, + } + } else { + // Without AES-GCM hardware, we put the ChaCha20-Poly1305 + // cipher suites first. + topCipherSuites = []uint16{ + tls.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305, + tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305, + tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256, + tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384, + tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256, + tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384, + } + defaultCipherSuitesTLS13 = []uint16{ + tls.TLS_CHACHA20_POLY1305_SHA256, + tls.TLS_AES_128_GCM_SHA256, + tls.TLS_AES_256_GCM_SHA384, + } + } + + DefaultCipherSuites := make([]uint16, 0, 22) + DefaultCipherSuites = append(DefaultCipherSuites, topCipherSuites...) + DefaultCipherSuites = append(DefaultCipherSuites, defaultCipherSuitesTLS13...) + + m.http = &http.Server{ + Addr: m.cfg.Address, + Handler: promhttp.HandlerFor(m.registry, promhttp.HandlerOpts{}), + IdleTimeout: time.Hour * 24, + ReadTimeout: time.Minute * 60, + MaxHeaderBytes: maxHeaderSize, + ReadHeaderTimeout: time.Minute * 60, + WriteTimeout: time.Minute * 60, + TLSConfig: &tls.Config{ + CurvePreferences: []tls.CurveID{ + tls.CurveP256, + tls.CurveP384, + tls.CurveP521, + tls.X25519, + }, + CipherSuites: DefaultCipherSuites, + MinVersion: tls.VersionTLS12, + PreferServerCipherSuites: true, + }, + } + + go func() { + err := m.http.ListenAndServe() + if err != nil && err != http.ErrServerClosed { + errCh <- err + return + } + }() + + return errCh +} + +// Stop prometheus metrics service. +func (m *Plugin) Stop() error { + m.mu.Lock() + defer m.mu.Unlock() + + if m.http != nil { + // timeout is 10 seconds + ctx, cancel := context.WithTimeout(context.Background(), time.Second*10) + defer cancel() + err := m.http.Shutdown(ctx) + if err != nil { + // Function should be Stop() error + m.log.Error("stop error", "error", errors.Errorf("error shutting down the metrics server: error %v", err)) + } + } + return nil +} + +// Collects used to collect all plugins which implement metrics.StatProvider interface (and Named) +func (m *Plugin) Collects() []interface{} { + return []interface{}{ + m.AddStatProvider, + } +} + +// Collector returns application specific collector by name or nil if collector not found. +func (m *Plugin) AddStatProvider(name endure.Named, stat metrics.StatProvider) error { + m.collectors.Store(name.Name(), statsProvider{ + collectors: stat.MetricsCollector(), + name: name.Name(), + }) + return nil +} + +// RPC interface satisfaction +func (m *Plugin) Name() string { + return ServiceName +} + +// RPC interface satisfaction +func (m *Plugin) RPC() interface{} { + return &rpcServer{ + svc: m, + log: m.log, + } +} diff --git a/plugins/metrics/rpc.go b/plugins/metrics/rpc.go new file mode 100644 index 00000000..b8897098 --- /dev/null +++ b/plugins/metrics/rpc.go @@ -0,0 +1,294 @@ +package metrics + +import ( + "github.com/prometheus/client_golang/prometheus" + "github.com/spiral/errors" + "github.com/spiral/roadrunner/v2/interfaces/log" +) + +type rpcServer struct { + svc *Plugin + log log.Logger +} + +// Metric represent single metric produced by the application. +type Metric struct { + // Collector name. + Name string + + // Collector value. + Value float64 + + // Labels associated with metric. Only for vector metrics. Must be provided in a form of label values. + Labels []string +} + +// Add new metric to the designated collector. +func (rpc *rpcServer) Add(m *Metric, ok *bool) error { + const op = errors.Op("Add metric") + rpc.log.Info("Adding metric", "name", m.Name, "value", m.Value, "labels", m.Labels) + c, exist := rpc.svc.collectors.Load(m.Name) + if !exist { + rpc.log.Error("undefined collector", "collector", m.Name) + return errors.E(op, errors.Errorf("undefined collector %s, try first Declare the desired collector", m.Name)) + } + + switch c := c.(type) { + case prometheus.Gauge: + c.Add(m.Value) + + case *prometheus.GaugeVec: + if len(m.Labels) == 0 { + rpc.log.Error("required labels for collector", "collector", m.Name) + return errors.E(op, errors.Errorf("required labels for collector %s", m.Name)) + } + + gauge, err := c.GetMetricWithLabelValues(m.Labels...) + if err != nil { + rpc.log.Error("failed to get metrics with label values", "collector", m.Name, "labels", m.Labels) + return errors.E(op, err) + } + gauge.Add(m.Value) + case prometheus.Counter: + c.Add(m.Value) + + case *prometheus.CounterVec: + if len(m.Labels) == 0 { + return errors.E(op, errors.Errorf("required labels for collector `%s`", m.Name)) + } + + gauge, err := c.GetMetricWithLabelValues(m.Labels...) + if err != nil { + rpc.log.Error("failed to get metrics with label values", "collector", m.Name, "labels", m.Labels) + return errors.E(op, err) + } + gauge.Add(m.Value) + + default: + return errors.E(op, errors.Errorf("collector %s does not support method `Add`", m.Name)) + } + + // RPC, set ok to true as return value. Need by rpc.Call reply argument + *ok = true + rpc.log.Info("new metric successfully added", "name", m.Name, "labels", m.Labels, "value", m.Value) + return nil +} + +// Sub subtract the value from the specific metric (gauge only). +func (rpc *rpcServer) Sub(m *Metric, ok *bool) error { + const op = errors.Op("Subtracting metric") + rpc.log.Info("Subtracting value from metric", "name", m.Name, "value", m.Value, "labels", m.Labels) + c, exist := rpc.svc.collectors.Load(m.Name) + if !exist { + rpc.log.Error("undefined collector", "name", m.Name, "value", m.Value, "labels", m.Labels) + return errors.E(op, errors.Errorf("undefined collector %s", m.Name)) + } + if c == nil { + // can it be nil ??? I guess can't + return errors.E(op, errors.Errorf("undefined collector %s", m.Name)) + } + + switch c := c.(type) { + case prometheus.Gauge: + c.Sub(m.Value) + + case *prometheus.GaugeVec: + if len(m.Labels) == 0 { + rpc.log.Error("required labels for collector, but none was provided", "name", m.Name, "value", m.Value) + return errors.E(op, errors.Errorf("required labels for collector %s", m.Name)) + } + + gauge, err := c.GetMetricWithLabelValues(m.Labels...) + if err != nil { + rpc.log.Error("failed to get metrics with label values", "collector", m.Name, "labels", m.Labels) + return errors.E(op, err) + } + gauge.Sub(m.Value) + default: + return errors.E(op, errors.Errorf("collector `%s` does not support method `Sub`", m.Name)) + } + rpc.log.Info("Subtracting operation applied successfully", "name", m.Name, "labels", m.Labels, "value", m.Value) + + *ok = true + return nil +} + +// Observe the value (histogram and summary only). +func (rpc *rpcServer) Observe(m *Metric, ok *bool) error { + const op = errors.Op("Observe metrics") + rpc.log.Info("Observing metric", "name", m.Name, "value", m.Value, "labels", m.Labels) + + c, exist := rpc.svc.collectors.Load(m.Name) + if !exist { + rpc.log.Error("undefined collector", "name", m.Name, "value", m.Value, "labels", m.Labels) + return errors.E(op, errors.Errorf("undefined collector %s", m.Name)) + } + if c == nil { + return errors.E(op, errors.Errorf("undefined collector %s", m.Name)) + } + + switch c := c.(type) { + case *prometheus.SummaryVec: + if len(m.Labels) == 0 { + return errors.E(op, errors.Errorf("required labels for collector `%s`", m.Name)) + } + + observer, err := c.GetMetricWithLabelValues(m.Labels...) + if err != nil { + return errors.E(op, err) + } + observer.Observe(m.Value) + + case prometheus.Histogram: + c.Observe(m.Value) + + case *prometheus.HistogramVec: + if len(m.Labels) == 0 { + return errors.E(op, errors.Errorf("required labels for collector `%s`", m.Name)) + } + + observer, err := c.GetMetricWithLabelValues(m.Labels...) + if err != nil { + rpc.log.Error("failed to get metrics with label values", "collector", m.Name, "labels", m.Labels) + return errors.E(op, err) + } + observer.Observe(m.Value) + default: + return errors.E(op, errors.Errorf("collector `%s` does not support method `Observe`", m.Name)) + } + + rpc.log.Info("observe operation finished successfully", "name", m.Name, "labels", m.Labels, "value", m.Value) + + *ok = true + return nil +} + +// Declare is used to register new collector in prometheus +// THE TYPES ARE: +// NamedCollector -> Collector with the name +// bool -> RPC reply value +// RETURNS: +// error +func (rpc *rpcServer) Declare(nc *NamedCollector, ok *bool) error { + const op = errors.Op("Declare metric") + rpc.log.Info("Declaring new metric", "name", nc.Name, "type", nc.Type, "namespace", nc.Namespace) + _, exist := rpc.svc.collectors.Load(nc.Name) + if exist { + rpc.log.Error("metric with provided name already exist", "name", nc.Name, "type", nc.Type, "namespace", nc.Namespace) + return errors.E(op, errors.Errorf("tried to register existing collector with the name `%s`", nc.Name)) + } + + var collector prometheus.Collector + switch nc.Type { + case Histogram: + opts := prometheus.HistogramOpts{ + Name: nc.Name, + Namespace: nc.Namespace, + Subsystem: nc.Subsystem, + Help: nc.Help, + Buckets: nc.Buckets, + } + + if len(nc.Labels) != 0 { + collector = prometheus.NewHistogramVec(opts, nc.Labels) + } else { + collector = prometheus.NewHistogram(opts) + } + case Gauge: + opts := prometheus.GaugeOpts{ + Name: nc.Name, + Namespace: nc.Namespace, + Subsystem: nc.Subsystem, + Help: nc.Help, + } + + if len(nc.Labels) != 0 { + collector = prometheus.NewGaugeVec(opts, nc.Labels) + } else { + collector = prometheus.NewGauge(opts) + } + case Counter: + opts := prometheus.CounterOpts{ + Name: nc.Name, + Namespace: nc.Namespace, + Subsystem: nc.Subsystem, + Help: nc.Help, + } + + if len(nc.Labels) != 0 { + collector = prometheus.NewCounterVec(opts, nc.Labels) + } else { + collector = prometheus.NewCounter(opts) + } + case Summary: + opts := prometheus.SummaryOpts{ + Name: nc.Name, + Namespace: nc.Namespace, + Subsystem: nc.Subsystem, + Help: nc.Help, + } + + if len(nc.Labels) != 0 { + collector = prometheus.NewSummaryVec(opts, nc.Labels) + } else { + collector = prometheus.NewSummary(opts) + } + + default: + return errors.E(op, errors.Errorf("unknown collector type %s", nc.Type)) + } + + // add collector to sync.Map + rpc.svc.collectors.Store(nc.Name, collector) + // that method might panic, we handle it by recover + err := rpc.svc.Register(collector) + if err != nil { + *ok = false + return errors.E(op, err) + } + + rpc.log.Info("metric successfully added", "name", nc.Name, "type", nc.Type, "namespace", nc.Namespace) + + *ok = true + return nil +} + +// Set the metric value (only for gaude). +func (rpc *rpcServer) Set(m *Metric, ok *bool) (err error) { + const op = errors.Op("Set metric") + rpc.log.Info("Observing metric", "name", m.Name, "value", m.Value, "labels", m.Labels) + + c, exist := rpc.svc.collectors.Load(m.Name) + if !exist { + return errors.E(op, errors.Errorf("undefined collector %s", m.Name)) + } + if c == nil { + return errors.E(op, errors.Errorf("undefined collector %s", m.Name)) + } + + switch c := c.(type) { + case prometheus.Gauge: + c.Set(m.Value) + + case *prometheus.GaugeVec: + if len(m.Labels) == 0 { + rpc.log.Error("required labels for collector", "collector", m.Name) + return errors.E(op, errors.Errorf("required labels for collector %s", m.Name)) + } + + gauge, err := c.GetMetricWithLabelValues(m.Labels...) + if err != nil { + rpc.log.Error("failed to get metrics with label values", "collector", m.Name, "labels", m.Labels) + return errors.E(op, err) + } + gauge.Set(m.Value) + + default: + return errors.E(op, errors.Errorf("collector `%s` does not support method Set", m.Name)) + } + + rpc.log.Info("set operation finished successfully", "name", m.Name, "labels", m.Labels, "value", m.Value) + + *ok = true + return nil +} diff --git a/plugins/metrics/tests/.rr-test.yaml b/plugins/metrics/tests/.rr-test.yaml new file mode 100644 index 00000000..79343e3c --- /dev/null +++ b/plugins/metrics/tests/.rr-test.yaml @@ -0,0 +1,13 @@ +rpc: + listen: tcp://127.0.0.1:6001 + disabled: false + +metrics: + # prometheus client address (path /metrics added automatically) + address: localhost:2112 + collect: + app_metric: + type: histogram + help: "Custom application metric" + labels: [ "type" ] + buckets: [ 0.1, 0.2, 0.3, 1.0 ]
\ No newline at end of file diff --git a/plugins/metrics/tests/docker-compose.yml b/plugins/metrics/tests/docker-compose.yml new file mode 100644 index 00000000..610633b4 --- /dev/null +++ b/plugins/metrics/tests/docker-compose.yml @@ -0,0 +1,7 @@ +version: '3.7' + +services: + prometheus: + image: prom/prometheus + ports: + - 9090:9090 diff --git a/plugins/metrics/tests/metrics_test.go b/plugins/metrics/tests/metrics_test.go new file mode 100644 index 00000000..ed5d085a --- /dev/null +++ b/plugins/metrics/tests/metrics_test.go @@ -0,0 +1,754 @@ +package tests + +import ( + "io/ioutil" + "net" + "net/http" + "net/rpc" + "os" + "os/signal" + "syscall" + "testing" + "time" + + "github.com/spiral/endure" + "github.com/spiral/goridge/v2" + "github.com/spiral/roadrunner/v2/plugins/config" + "github.com/spiral/roadrunner/v2/plugins/logger" + "github.com/spiral/roadrunner/v2/plugins/metrics" + rpcPlugin "github.com/spiral/roadrunner/v2/plugins/rpc" + "github.com/stretchr/testify/assert" +) + +const dialAddr = "127.0.0.1:6001" +const dialNetwork = "tcp" +const getAddr = "http://localhost:2112/metrics" + +// get request and return body +func get(url string) (string, error) { + r, err := http.Get(url) + if err != nil { + return "", err + } + + b, err := ioutil.ReadAll(r.Body) + if err != nil { + return "", err + } + + err = r.Body.Close() + if err != nil { + return "", err + } + // unsafe + return string(b), err +} + +func TestMetricsInit(t *testing.T) { + cont, err := endure.NewContainer(nil, endure.SetLogLevel(endure.DebugLevel)) + if err != nil { + t.Fatal(err) + } + + cfg := &config.Viper{} + cfg.Prefix = "rr" + cfg.Path = ".rr-test.yaml" + + err = cont.Register(cfg) + if err != nil { + t.Fatal(err) + } + + err = cont.Register(&metrics.Plugin{}) + if err != nil { + t.Fatal(err) + } + + err = cont.Register(&rpcPlugin.Plugin{}) + if err != nil { + t.Fatal(err) + } + + err = cont.Register(&logger.ZapLogger{}) + if err != nil { + t.Fatal(err) + } + err = cont.Register(&Plugin1{}) + if err != nil { + t.Fatal(err) + } + + err = cont.Init() + if err != nil { + t.Fatal(err) + } + + ch, err := cont.Serve() + assert.NoError(t, err) + + sig := make(chan os.Signal, 1) + signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) + + tt := time.NewTimer(time.Second * 5) + + out, err := get("http://localhost:2112/metrics") + assert.NoError(t, err) + + assert.Contains(t, out, "go_gc_duration_seconds") + + for { + select { + case e := <-ch: + assert.Fail(t, "error", e.Error.Error()) + err = cont.Stop() + if err != nil { + assert.FailNow(t, "error", err.Error()) + } + case <-sig: + err = cont.Stop() + if err != nil { + assert.FailNow(t, "error", err.Error()) + } + return + case <-tt.C: + // timeout + err = cont.Stop() + if err != nil { + assert.FailNow(t, "error", err.Error()) + } + return + } + } +} + +func TestMetricsGaugeCollector(t *testing.T) { + cont, err := endure.NewContainer(nil, endure.SetLogLevel(endure.DebugLevel), endure.Visualize(endure.StdOut, "")) + if err != nil { + t.Fatal(err) + } + + cfg := &config.Viper{} + cfg.Prefix = "rr" + cfg.Path = ".rr-test.yaml" + + err = cont.RegisterAll( + cfg, + &metrics.Plugin{}, + &rpcPlugin.Plugin{}, + &logger.ZapLogger{}, + &Plugin1{}, + ) + assert.NoError(t, err) + + err = cont.Init() + if err != nil { + t.Fatal(err) + } + + ch, err := cont.Serve() + assert.NoError(t, err) + + sig := make(chan os.Signal, 1) + signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) + + time.Sleep(time.Second) + tt := time.NewTimer(time.Second * 5) + + out, err := get("http://localhost:2112/metrics") + assert.NoError(t, err) + assert.Contains(t, out, "my_gauge 100") + assert.Contains(t, out, "my_gauge2 100") + + out, err = get("http://localhost:2112/metrics") + assert.NoError(t, err) + assert.Contains(t, out, "go_gc_duration_seconds") + + for { + select { + case e := <-ch: + assert.Fail(t, "error", e.Error.Error()) + err = cont.Stop() + if err != nil { + assert.FailNow(t, "error", err.Error()) + } + case <-sig: + err = cont.Stop() + if err != nil { + assert.FailNow(t, "error", err.Error()) + } + return + case <-tt.C: + // timeout + err = cont.Stop() + if err != nil { + assert.FailNow(t, "error", err.Error()) + } + return + } + } +} + +func TestMetricsDifferentRPCCalls(t *testing.T) { + cont, err := endure.NewContainer(nil, endure.SetLogLevel(endure.DebugLevel), endure.Visualize(endure.StdOut, "")) + if err != nil { + t.Fatal(err) + } + + cfg := &config.Viper{} + cfg.Prefix = "rr" + cfg.Path = ".rr-test.yaml" + + err = cont.RegisterAll( + cfg, + &metrics.Plugin{}, + &rpcPlugin.Plugin{}, + &logger.ZapLogger{}, + ) + assert.NoError(t, err) + + err = cont.Init() + if err != nil { + t.Fatal(err) + } + + ch, err := cont.Serve() + assert.NoError(t, err) + + sig := make(chan os.Signal, 1) + signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) + + go func() { + tt := time.NewTimer(time.Minute * 3) + for { + select { + case e := <-ch: + assert.Fail(t, "error", e.Error.Error()) + err = cont.Stop() + if err != nil { + assert.FailNow(t, "error", err.Error()) + } + case <-sig: + err = cont.Stop() + if err != nil { + assert.FailNow(t, "error", err.Error()) + } + return + case <-tt.C: + // timeout + err = cont.Stop() + if err != nil { + assert.FailNow(t, "error", err.Error()) + } + return + } + } + }() + + t.Run("DeclareMetric", declareMetricsTest) + genericOut, err := get(getAddr) + assert.NoError(t, err) + assert.Contains(t, genericOut, "test_metrics_named_collector") + + t.Run("AddMetric", addMetricsTest) + genericOut, err = get(getAddr) + assert.NoError(t, err) + assert.Contains(t, genericOut, "test_metrics_named_collector 10000") + + t.Run("SetMetric", setMetric) + genericOut, err = get(getAddr) + assert.NoError(t, err) + assert.Contains(t, genericOut, "user_gauge_collector 100") + + t.Run("VectorMetric", vectorMetric) + genericOut, err = get(getAddr) + assert.NoError(t, err) + assert.Contains(t, genericOut, "gauge_2_collector{section=\"first\",type=\"core\"} 100") + + t.Run("MissingSection", missingSection) + t.Run("SetWithoutLabels", setWithoutLabels) + t.Run("SetOnHistogram", setOnHistogram) + t.Run("MetricSub", subMetric) + genericOut, err = get(getAddr) + assert.NoError(t, err) + assert.Contains(t, genericOut, "sub_gauge_subMetric 1") + + t.Run("SubVector", subVector) + genericOut, err = get(getAddr) + assert.NoError(t, err) + assert.Contains(t, genericOut, "sub_gauge_subVector{section=\"first\",type=\"core\"} 1") + + t.Run("RegisterHistogram", registerHistogram) + + genericOut, err = get(getAddr) + assert.NoError(t, err) + assert.Contains(t, genericOut, `TYPE histogram_registerHistogram`) + + // check buckets + assert.Contains(t, genericOut, `histogram_registerHistogram_bucket{le="0.1"} 0`) + assert.Contains(t, genericOut, `histogram_registerHistogram_bucket{le="0.2"} 0`) + assert.Contains(t, genericOut, `histogram_registerHistogram_bucket{le="0.5"} 0`) + assert.Contains(t, genericOut, `histogram_registerHistogram_bucket{le="+Inf"} 0`) + assert.Contains(t, genericOut, `histogram_registerHistogram_sum 0`) + assert.Contains(t, genericOut, `histogram_registerHistogram_count 0`) + + t.Run("CounterMetric", counterMetric) + genericOut, err = get(getAddr) + assert.NoError(t, err) + assert.Contains(t, genericOut, "HELP default_default_counter_CounterMetric test_counter") + assert.Contains(t, genericOut, `default_default_counter_CounterMetric{section="section2",type="type2"}`) + + t.Run("ObserveMetric", observeMetric) + genericOut, err = get(getAddr) + assert.NoError(t, err) + assert.Contains(t, genericOut, "observe_observeMetric") + + t.Run("ObserveMetricNotEnoughLabels", observeMetricNotEnoughLabels) + + close(sig) +} + +func observeMetricNotEnoughLabels(t *testing.T) { + conn, err := net.Dial(dialNetwork, dialAddr) + assert.NoError(t, err) + defer func() { + _ = conn.Close() + }() + client := rpc.NewClientWithCodec(goridge.NewClientCodec(conn)) + var ret bool + + nc := metrics.NamedCollector{ + Name: "observe_observeMetricNotEnoughLabels", + Collector: metrics.Collector{ + Namespace: "default", + Subsystem: "default", + Help: "test_observe", + Type: metrics.Histogram, + Labels: []string{"type", "section"}, + }, + } + + err = client.Call("metrics.Declare", nc, &ret) + assert.NoError(t, err) + assert.True(t, ret) + ret = false + + assert.Error(t, client.Call("metrics.Observe", metrics.Metric{ + Name: "observe_observeMetric", + Value: 100.0, + Labels: []string{"test"}, + }, &ret)) + assert.False(t, ret) +} + +func observeMetric(t *testing.T) { + conn, err := net.Dial(dialNetwork, dialAddr) + assert.NoError(t, err) + defer func() { + _ = conn.Close() + }() + client := rpc.NewClientWithCodec(goridge.NewClientCodec(conn)) + var ret bool + + nc := metrics.NamedCollector{ + Name: "observe_observeMetric", + Collector: metrics.Collector{ + Namespace: "default", + Subsystem: "default", + Help: "test_observe", + Type: metrics.Histogram, + Labels: []string{"type", "section"}, + }, + } + + err = client.Call("metrics.Declare", nc, &ret) + assert.NoError(t, err) + assert.True(t, ret) + ret = false + + assert.NoError(t, client.Call("metrics.Observe", metrics.Metric{ + Name: "observe_observeMetric", + Value: 100.0, + Labels: []string{"test", "test2"}, + }, &ret)) + assert.True(t, ret) +} + +func counterMetric(t *testing.T) { + conn, err := net.Dial(dialNetwork, dialAddr) + assert.NoError(t, err) + defer func() { + _ = conn.Close() + }() + client := rpc.NewClientWithCodec(goridge.NewClientCodec(conn)) + var ret bool + + nc := metrics.NamedCollector{ + Name: "counter_CounterMetric", + Collector: metrics.Collector{ + Namespace: "default", + Subsystem: "default", + Help: "test_counter", + Type: metrics.Counter, + Labels: []string{"type", "section"}, + }, + } + + err = client.Call("metrics.Declare", nc, &ret) + assert.NoError(t, err) + assert.True(t, ret) + + ret = false + + assert.NoError(t, client.Call("metrics.Add", metrics.Metric{ + Name: "counter_CounterMetric", + Value: 100.0, + Labels: []string{"type2", "section2"}, + }, &ret)) + assert.True(t, ret) +} + +func registerHistogram(t *testing.T) { + conn, err := net.Dial(dialNetwork, dialAddr) + assert.NoError(t, err) + defer func() { + _ = conn.Close() + }() + client := rpc.NewClientWithCodec(goridge.NewClientCodec(conn)) + var ret bool + + nc := metrics.NamedCollector{ + Name: "histogram_registerHistogram", + Collector: metrics.Collector{ + Help: "test_histogram", + Type: metrics.Histogram, + Buckets: []float64{0.1, 0.2, 0.5}, + }, + } + + err = client.Call("metrics.Declare", nc, &ret) + assert.NoError(t, err) + assert.True(t, ret) + + ret = false + + m := metrics.Metric{ + Name: "histogram_registerHistogram", + Value: 10000, + Labels: nil, + } + + err = client.Call("metrics.Add", m, &ret) + assert.Error(t, err) + assert.False(t, ret) +} + +func subVector(t *testing.T) { + conn, err := net.Dial(dialNetwork, dialAddr) + assert.NoError(t, err) + defer func() { + _ = conn.Close() + }() + + client := rpc.NewClientWithCodec(goridge.NewClientCodec(conn)) + var ret bool + + nc := metrics.NamedCollector{ + Name: "sub_gauge_subVector", + Collector: metrics.Collector{ + Namespace: "default", + Subsystem: "default", + Type: metrics.Gauge, + Labels: []string{"type", "section"}, + }, + } + + err = client.Call("metrics.Declare", nc, &ret) + assert.NoError(t, err) + assert.True(t, ret) + ret = false + + m := metrics.Metric{ + Name: "sub_gauge_subVector", + Value: 100000, + Labels: []string{"core", "first"}, + } + + err = client.Call("metrics.Add", m, &ret) + assert.NoError(t, err) + assert.True(t, ret) + ret = false + + m = metrics.Metric{ + Name: "sub_gauge_subVector", + Value: 99999, + Labels: []string{"core", "first"}, + } + + err = client.Call("metrics.Sub", m, &ret) + assert.NoError(t, err) + assert.True(t, ret) +} + +func subMetric(t *testing.T) { + conn, err := net.Dial(dialNetwork, dialAddr) + assert.NoError(t, err) + defer func() { + _ = conn.Close() + }() + client := rpc.NewClientWithCodec(goridge.NewClientCodec(conn)) + var ret bool + + nc := metrics.NamedCollector{ + Name: "sub_gauge_subMetric", + Collector: metrics.Collector{ + Namespace: "default", + Subsystem: "default", + Type: metrics.Gauge, + }, + } + + err = client.Call("metrics.Declare", nc, &ret) + assert.NoError(t, err) + assert.True(t, ret) + ret = false + + m := metrics.Metric{ + Name: "sub_gauge_subMetric", + Value: 100000, + } + + err = client.Call("metrics.Add", m, &ret) + assert.NoError(t, err) + assert.True(t, ret) + ret = false + + m = metrics.Metric{ + Name: "sub_gauge_subMetric", + Value: 99999, + } + + err = client.Call("metrics.Sub", m, &ret) + assert.NoError(t, err) + assert.True(t, ret) +} + +func setOnHistogram(t *testing.T) { + conn, err := net.Dial(dialNetwork, dialAddr) + assert.NoError(t, err) + defer func() { + _ = conn.Close() + }() + client := rpc.NewClientWithCodec(goridge.NewClientCodec(conn)) + var ret bool + + nc := metrics.NamedCollector{ + Name: "histogram_setOnHistogram", + Collector: metrics.Collector{ + Namespace: "default", + Subsystem: "default", + Type: metrics.Histogram, + Labels: []string{"type", "section"}, + }, + } + + err = client.Call("metrics.Declare", nc, &ret) + assert.NoError(t, err) + assert.True(t, ret) + + ret = false + + m := metrics.Metric{ + Name: "gauge_setOnHistogram", + Value: 100.0, + } + + err = client.Call("metrics.Set", m, &ret) // expected 2 label values but got 1 in []string{"missing"} + assert.Error(t, err) + assert.False(t, ret) +} + +func setWithoutLabels(t *testing.T) { + conn, err := net.Dial(dialNetwork, dialAddr) + assert.NoError(t, err) + defer func() { + _ = conn.Close() + }() + client := rpc.NewClientWithCodec(goridge.NewClientCodec(conn)) + var ret bool + + nc := metrics.NamedCollector{ + Name: "gauge_setWithoutLabels", + Collector: metrics.Collector{ + Namespace: "default", + Subsystem: "default", + Type: metrics.Gauge, + Labels: []string{"type", "section"}, + }, + } + + err = client.Call("metrics.Declare", nc, &ret) + assert.NoError(t, err) + assert.True(t, ret) + + ret = false + + m := metrics.Metric{ + Name: "gauge_setWithoutLabels", + Value: 100.0, + } + + err = client.Call("metrics.Set", m, &ret) // expected 2 label values but got 1 in []string{"missing"} + assert.Error(t, err) + assert.False(t, ret) +} + +func missingSection(t *testing.T) { + conn, err := net.Dial(dialNetwork, dialAddr) + assert.NoError(t, err) + defer func() { + _ = conn.Close() + }() + client := rpc.NewClientWithCodec(goridge.NewClientCodec(conn)) + var ret bool + + nc := metrics.NamedCollector{ + Name: "gauge_missing_section_collector", + Collector: metrics.Collector{ + Namespace: "default", + Subsystem: "default", + Type: metrics.Gauge, + Labels: []string{"type", "section"}, + }, + } + + err = client.Call("metrics.Declare", nc, &ret) + assert.NoError(t, err) + assert.True(t, ret) + + ret = false + + m := metrics.Metric{ + Name: "gauge_missing_section_collector", + Value: 100.0, + Labels: []string{"missing"}, + } + + err = client.Call("metrics.Set", m, &ret) // expected 2 label values but got 1 in []string{"missing"} + assert.Error(t, err) + assert.False(t, ret) +} + +func vectorMetric(t *testing.T) { + conn, err := net.Dial(dialNetwork, dialAddr) + assert.NoError(t, err) + defer func() { + _ = conn.Close() + }() + client := rpc.NewClientWithCodec(goridge.NewClientCodec(conn)) + var ret bool + + nc := metrics.NamedCollector{ + Name: "gauge_2_collector", + Collector: metrics.Collector{ + Namespace: "default", + Subsystem: "default", + Type: metrics.Gauge, + Labels: []string{"type", "section"}, + }, + } + + err = client.Call("metrics.Declare", nc, &ret) + assert.NoError(t, err) + assert.True(t, ret) + + ret = false + + m := metrics.Metric{ + Name: "gauge_2_collector", + Value: 100.0, + Labels: []string{"core", "first"}, + } + + err = client.Call("metrics.Set", m, &ret) + assert.NoError(t, err) + assert.True(t, ret) +} + +func setMetric(t *testing.T) { + conn, err := net.Dial(dialNetwork, dialAddr) + assert.NoError(t, err) + defer func() { + _ = conn.Close() + }() + client := rpc.NewClientWithCodec(goridge.NewClientCodec(conn)) + var ret bool + + nc := metrics.NamedCollector{ + Name: "user_gauge_collector", + Collector: metrics.Collector{ + Namespace: "default", + Subsystem: "default", + Type: metrics.Gauge, + }, + } + + err = client.Call("metrics.Declare", nc, &ret) + assert.NoError(t, err) + assert.True(t, ret) + ret = false + + m := metrics.Metric{ + Name: "user_gauge_collector", + Value: 100.0, + } + + err = client.Call("metrics.Set", m, &ret) + assert.NoError(t, err) + assert.True(t, ret) +} + +func addMetricsTest(t *testing.T) { + conn, err := net.Dial(dialNetwork, dialAddr) + assert.NoError(t, err) + defer func() { + _ = conn.Close() + }() + client := rpc.NewClientWithCodec(goridge.NewClientCodec(conn)) + var ret bool + + m := metrics.Metric{ + Name: "test_metrics_named_collector", + Value: 10000, + Labels: nil, + } + + err = client.Call("metrics.Add", m, &ret) + assert.NoError(t, err) + assert.True(t, ret) +} + +func declareMetricsTest(t *testing.T) { + conn, err := net.Dial(dialNetwork, dialAddr) + assert.NoError(t, err) + defer func() { + _ = conn.Close() + }() + client := rpc.NewClientWithCodec(goridge.NewClientCodec(conn)) + var ret bool + + nc := metrics.NamedCollector{ + Name: "test_metrics_named_collector", + Collector: metrics.Collector{ + Namespace: "default", + Subsystem: "default", + Type: metrics.Counter, + Help: "NO HELP!", + Labels: nil, + Buckets: nil, + }, + } + + err = client.Call("metrics.Declare", nc, &ret) + assert.NoError(t, err) + assert.True(t, ret) +} diff --git a/plugins/metrics/tests/plugin1.go b/plugins/metrics/tests/plugin1.go new file mode 100644 index 00000000..b48c415d --- /dev/null +++ b/plugins/metrics/tests/plugin1.go @@ -0,0 +1,46 @@ +package tests + +import ( + "github.com/prometheus/client_golang/prometheus" + "github.com/spiral/roadrunner/v2/plugins/config" +) + +// Gauge ////////////// +type Plugin1 struct { + config config.Configurer +} + +func (p1 *Plugin1) Init(cfg config.Configurer) error { + p1.config = cfg + return nil +} + +func (p1 *Plugin1) Serve() chan error { + errCh := make(chan error, 1) + return errCh +} + +func (p1 *Plugin1) Stop() error { + return nil +} + +func (p1 *Plugin1) Name() string { + return "metrics_test.plugin1" +} + +func (p1 *Plugin1) MetricsCollector() []prometheus.Collector { + collector := prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "my_gauge", + Help: "My gauge value", + }) + + collector.Set(100) + + collector2 := prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "my_gauge2", + Help: "My gauge2 value", + }) + + collector2.Set(100) + return []prometheus.Collector{collector, collector2} +} diff --git a/plugins/rpc/plugin.go b/plugins/rpc/plugin.go index 6401c0e2..82b30563 100755 --- a/plugins/rpc/plugin.go +++ b/plugins/rpc/plugin.go @@ -8,27 +8,25 @@ import ( "github.com/spiral/endure" "github.com/spiral/errors" "github.com/spiral/goridge/v2" - "github.com/spiral/roadrunner/v2/log" + "github.com/spiral/roadrunner/v2/interfaces/log" + rpc_ "github.com/spiral/roadrunner/v2/interfaces/rpc" "github.com/spiral/roadrunner/v2/plugins/config" ) -// Pluggable declares the ability to create set of public RPC methods. -type Pluggable interface { - endure.Named - - // Provides RPC methods for the given service. - RPCService() (interface{}, error) -} - // ServiceName contains default service name. const ServiceName = "RPC" +type pluggable struct { + service rpc_.RPCer + name string +} + // Plugin is RPC service. type Plugin struct { cfg Config log log.Logger rpc *rpc.Server - services []Pluggable + services []pluggable listener net.Listener closed *uint32 } @@ -69,19 +67,13 @@ func (s *Plugin) Serve() chan error { // Attach all services for i := 0; i < len(s.services); i++ { - svc, err := s.services[i].RPCService() + err := s.Register(s.services[i].name, s.services[i].service.RPC()) if err != nil { errCh <- errors.E(op, err) return errCh } - err = s.Register(s.services[i].Name(), svc) - if err != nil { - errCh <- errors.E(op, err) - return errCh - } - - services = append(services, s.services[i].Name()) + services = append(services, s.services[i].name) } var err error @@ -139,9 +131,11 @@ func (s *Plugin) Collects() []interface{} { } // RegisterPlugin registers RPC service plugin. -func (s *Plugin) RegisterPlugin(p Pluggable) error { - s.services = append(s.services, p) - return nil +func (s *Plugin) RegisterPlugin(name endure.Named, p rpc_.RPCer) { + s.services = append(s.services, pluggable{ + service: p, + name: name.Name(), + }) } // Register publishes in the server the set of methods of the diff --git a/plugins/rpc/tests/plugin1.go b/plugins/rpc/tests/plugin1.go index 788e6a2c..a8d5c216 100644 --- a/plugins/rpc/tests/plugin1.go +++ b/plugins/rpc/tests/plugin1.go @@ -28,8 +28,8 @@ func (p1 *Plugin1) Name() string { return "rpc_test.plugin1" } -func (p1 *Plugin1) RPCService() (interface{}, error) { - return &PluginRpc{srv: p1}, nil +func (p1 *Plugin1) RPC() interface{} { + return &PluginRpc{srv: p1} } type PluginRpc struct { diff --git a/plugins/app/config.go b/plugins/server/config.go index eaa54e2d..147ae0f7 100644 --- a/plugins/app/config.go +++ b/plugins/server/config.go @@ -1,6 +1,10 @@ -package app +package server -import "time" +import ( + "time" + + "github.com/spiral/roadrunner/v2/interfaces/server" +) // Config config combines factory, pool and cmd configurations. type Config struct { @@ -14,7 +18,7 @@ type Config struct { Group string // Env represents application environment. - Env Env + Env server.Env // Listen defines connection method and factory to be used to connect to workers: // "pipes", "tcp://:6001", "unix://rr.sock" diff --git a/plugins/app/plugin.go b/plugins/server/plugin.go index d76961ca..e096708a 100644 --- a/plugins/app/plugin.go +++ b/plugins/server/plugin.go @@ -1,4 +1,4 @@ -package app +package server import ( "context" @@ -9,21 +9,13 @@ import ( "github.com/spiral/errors" "github.com/spiral/roadrunner/v2" - "github.com/spiral/roadrunner/v2/log" + "github.com/spiral/roadrunner/v2/interfaces/log" + "github.com/spiral/roadrunner/v2/interfaces/server" "github.com/spiral/roadrunner/v2/plugins/config" "github.com/spiral/roadrunner/v2/util" ) -const ServiceName = "app" - -type Env map[string]string - -// WorkerFactory creates workers for the application. -type WorkerFactory interface { - CmdFactory(env Env) (func() *exec.Cmd, error) - NewWorker(ctx context.Context, env Env) (roadrunner.WorkerBase, error) - NewWorkerPool(ctx context.Context, opt roadrunner.Config, env Env) (roadrunner.Pool, error) -} +const ServiceName = "server" // Plugin manages worker type Plugin struct { @@ -71,7 +63,7 @@ func (app *Plugin) Stop() error { } // CmdFactory provides worker command factory assocated with given context. -func (app *Plugin) CmdFactory(env Env) (func() *exec.Cmd, error) { +func (app *Plugin) CmdFactory(env server.Env) (func() *exec.Cmd, error) { var cmdArgs []string // create command according to the config @@ -97,7 +89,7 @@ func (app *Plugin) CmdFactory(env Env) (func() *exec.Cmd, error) { } // NewWorker issues new standalone worker. -func (app *Plugin) NewWorker(ctx context.Context, env Env) (roadrunner.WorkerBase, error) { +func (app *Plugin) NewWorker(ctx context.Context, env server.Env) (roadrunner.WorkerBase, error) { const op = errors.Op("new worker") spawnCmd, err := app.CmdFactory(env) if err != nil { @@ -115,7 +107,7 @@ func (app *Plugin) NewWorker(ctx context.Context, env Env) (roadrunner.WorkerBas } // NewWorkerPool issues new worker pool. -func (app *Plugin) NewWorkerPool(ctx context.Context, opt roadrunner.Config, env Env) (roadrunner.Pool, error) { +func (app *Plugin) NewWorkerPool(ctx context.Context, opt roadrunner.PoolConfig, env server.Env) (roadrunner.Pool, error) { spawnCmd, err := app.CmdFactory(env) if err != nil { return nil, err @@ -159,7 +151,7 @@ func (app *Plugin) initFactory() (roadrunner.Factory, error) { } } -func (app *Plugin) setEnv(e Env) []string { +func (app *Plugin) setEnv(e server.Env) []string { env := append(os.Environ(), fmt.Sprintf("RR_RELAY=%s", app.cfg.Relay)) for k, v := range e { env = append(env, fmt.Sprintf("%s=%s", strings.ToUpper(k), v)) diff --git a/plugins/app/tests/configs/.rr-no-app-section.yaml b/plugins/server/tests/configs/.rr-no-app-section.yaml index d129ae8a..b6e3ea93 100644 --- a/plugins/app/tests/configs/.rr-no-app-section.yaml +++ b/plugins/server/tests/configs/.rr-no-app-section.yaml @@ -1,4 +1,4 @@ -upp: +server: command: "php ../../../tests/client.php echo pipes" user: "" group: "" diff --git a/plugins/app/tests/configs/.rr-sockets.yaml b/plugins/server/tests/configs/.rr-sockets.yaml index 9bd62693..ab1239aa 100644 --- a/plugins/app/tests/configs/.rr-sockets.yaml +++ b/plugins/server/tests/configs/.rr-sockets.yaml @@ -1,4 +1,4 @@ -app: +server: command: "php socket.php" user: "" group: "" diff --git a/plugins/app/tests/configs/.rr-tcp.yaml b/plugins/server/tests/configs/.rr-tcp.yaml index c5a26d37..f53bffcc 100644 --- a/plugins/app/tests/configs/.rr-tcp.yaml +++ b/plugins/server/tests/configs/.rr-tcp.yaml @@ -1,4 +1,4 @@ -app: +server: command: "php tcp.php" user: "" group: "" diff --git a/plugins/app/tests/configs/.rr-wrong-command.yaml b/plugins/server/tests/configs/.rr-wrong-command.yaml index 4bd019d3..d2c087a6 100644 --- a/plugins/app/tests/configs/.rr-wrong-command.yaml +++ b/plugins/server/tests/configs/.rr-wrong-command.yaml @@ -1,4 +1,4 @@ -app: +server: command: "php some_absent_file.php" user: "" group: "" diff --git a/plugins/app/tests/configs/.rr-wrong-relay.yaml b/plugins/server/tests/configs/.rr-wrong-relay.yaml index d8ffe8f8..1dd73d73 100644 --- a/plugins/app/tests/configs/.rr-wrong-relay.yaml +++ b/plugins/server/tests/configs/.rr-wrong-relay.yaml @@ -1,4 +1,4 @@ -app: +server: command: "php ../../../tests/client.php echo pipes" user: "" group: "" diff --git a/plugins/app/tests/configs/.rr.yaml b/plugins/server/tests/configs/.rr.yaml index 221aff92..b6e3ea93 100644 --- a/plugins/app/tests/configs/.rr.yaml +++ b/plugins/server/tests/configs/.rr.yaml @@ -1,4 +1,4 @@ -app: +server: command: "php ../../../tests/client.php echo pipes" user: "" group: "" diff --git a/plugins/app/tests/plugin_pipes.go b/plugins/server/tests/plugin_pipes.go index fc999718..840021eb 100644 --- a/plugins/app/tests/plugin_pipes.go +++ b/plugins/server/tests/plugin_pipes.go @@ -6,14 +6,15 @@ import ( "github.com/spiral/errors" "github.com/spiral/roadrunner/v2" - "github.com/spiral/roadrunner/v2/plugins/app" + "github.com/spiral/roadrunner/v2/interfaces/server" "github.com/spiral/roadrunner/v2/plugins/config" + plugin "github.com/spiral/roadrunner/v2/plugins/server" ) const ConfigSection = "app" const Response = "test" -var testPoolConfig = roadrunner.Config{ +var testPoolConfig = roadrunner.PoolConfig{ NumWorkers: 10, MaxJobs: 100, AllocateTimeout: time.Second * 10, @@ -29,11 +30,11 @@ var testPoolConfig = roadrunner.Config{ type Foo struct { configProvider config.Configurer - wf app.WorkerFactory + wf server.WorkerFactory pool roadrunner.Pool } -func (f *Foo) Init(p config.Configurer, workerFactory app.WorkerFactory) error { +func (f *Foo) Init(p config.Configurer, workerFactory server.WorkerFactory) error { f.configProvider = p f.wf = workerFactory return nil @@ -50,7 +51,7 @@ func (f *Foo) Serve() chan error { errCh := make(chan error, 1) - conf := &app.Config{} + conf := &plugin.Config{} var err error err = f.configProvider.UnmarshalKey(ConfigSection, conf) if err != nil { diff --git a/plugins/app/tests/plugin_sockets.go b/plugins/server/tests/plugin_sockets.go index 585264f6..b12f4ead 100644 --- a/plugins/app/tests/plugin_sockets.go +++ b/plugins/server/tests/plugin_sockets.go @@ -5,17 +5,18 @@ import ( "github.com/spiral/errors" "github.com/spiral/roadrunner/v2" - "github.com/spiral/roadrunner/v2/plugins/app" + "github.com/spiral/roadrunner/v2/interfaces/server" "github.com/spiral/roadrunner/v2/plugins/config" + plugin "github.com/spiral/roadrunner/v2/plugins/server" ) type Foo2 struct { configProvider config.Configurer - wf app.WorkerFactory + wf server.WorkerFactory pool roadrunner.Pool } -func (f *Foo2) Init(p config.Configurer, workerFactory app.WorkerFactory) error { +func (f *Foo2) Init(p config.Configurer, workerFactory server.WorkerFactory) error { f.configProvider = p f.wf = workerFactory return nil @@ -25,7 +26,7 @@ func (f *Foo2) Serve() chan error { const op = errors.Op("serve") var err error errCh := make(chan error, 1) - conf := &app.Config{} + conf := &plugin.Config{} // test payload for echo r := roadrunner.Payload{ diff --git a/plugins/app/tests/plugin_tcp.go b/plugins/server/tests/plugin_tcp.go index 6abc533d..39044577 100644 --- a/plugins/app/tests/plugin_tcp.go +++ b/plugins/server/tests/plugin_tcp.go @@ -5,17 +5,18 @@ import ( "github.com/spiral/errors" "github.com/spiral/roadrunner/v2" - "github.com/spiral/roadrunner/v2/plugins/app" + "github.com/spiral/roadrunner/v2/interfaces/server" "github.com/spiral/roadrunner/v2/plugins/config" + plugin "github.com/spiral/roadrunner/v2/plugins/server" ) type Foo3 struct { configProvider config.Configurer - wf app.WorkerFactory + wf server.WorkerFactory pool roadrunner.Pool } -func (f *Foo3) Init(p config.Configurer, workerFactory app.WorkerFactory) error { +func (f *Foo3) Init(p config.Configurer, workerFactory server.WorkerFactory) error { f.configProvider = p f.wf = workerFactory return nil @@ -25,7 +26,7 @@ func (f *Foo3) Serve() chan error { const op = errors.Op("serve") var err error errCh := make(chan error, 1) - conf := &app.Config{} + conf := &plugin.Config{} // test payload for echo r := roadrunner.Payload{ diff --git a/plugins/app/tests/app_test.go b/plugins/server/tests/server_test.go index 3c416b59..f917df5d 100644 --- a/plugins/app/tests/app_test.go +++ b/plugins/server/tests/server_test.go @@ -7,9 +7,9 @@ import ( "time" "github.com/spiral/endure" - "github.com/spiral/roadrunner/v2/plugins/app" "github.com/spiral/roadrunner/v2/plugins/config" "github.com/spiral/roadrunner/v2/plugins/logger" + "github.com/spiral/roadrunner/v2/plugins/server" "github.com/stretchr/testify/assert" ) @@ -27,7 +27,7 @@ func TestAppPipes(t *testing.T) { t.Fatal(err) } - err = container.Register(&app.Plugin{}) + err = container.Register(&server.Plugin{}) if err != nil { t.Fatal(err) } @@ -93,7 +93,7 @@ func TestAppSockets(t *testing.T) { t.Fatal(err) } - err = container.Register(&app.Plugin{}) + err = container.Register(&server.Plugin{}) if err != nil { t.Fatal(err) } @@ -159,7 +159,7 @@ func TestAppTCP(t *testing.T) { t.Fatal(err) } - err = container.Register(&app.Plugin{}) + err = container.Register(&server.Plugin{}) if err != nil { t.Fatal(err) } @@ -225,7 +225,7 @@ func TestAppWrongConfig(t *testing.T) { t.Fatal(err) } - err = container.Register(&app.Plugin{}) + err = container.Register(&server.Plugin{}) if err != nil { t.Fatal(err) } @@ -257,7 +257,7 @@ func TestAppWrongRelay(t *testing.T) { t.Fatal(err) } - err = container.Register(&app.Plugin{}) + err = container.Register(&server.Plugin{}) if err != nil { t.Fatal(err) } @@ -295,7 +295,7 @@ func TestAppWrongCommand(t *testing.T) { t.Fatal(err) } - err = container.Register(&app.Plugin{}) + err = container.Register(&server.Plugin{}) if err != nil { t.Fatal(err) } @@ -333,7 +333,7 @@ func TestAppNoAppSectionInConfig(t *testing.T) { t.Fatal(err) } - err = container.Register(&app.Plugin{}) + err = container.Register(&server.Plugin{}) if err != nil { t.Fatal(err) } diff --git a/plugins/app/tests/socket.php b/plugins/server/tests/socket.php index 143c3ce4..143c3ce4 100644 --- a/plugins/app/tests/socket.php +++ b/plugins/server/tests/socket.php diff --git a/plugins/app/tests/tcp.php b/plugins/server/tests/tcp.php index 2d6fb00a..2d6fb00a 100644 --- a/plugins/app/tests/tcp.php +++ b/plugins/server/tests/tcp.php @@ -49,7 +49,7 @@ type Pool interface { AddListener(listener util.EventListener) // GetConfig returns pool configuration. - GetConfig() Config + GetConfig() PoolConfig // Exec Exec(rqs Payload) (Payload, error) @@ -67,7 +67,7 @@ type Pool interface { } // Configures the pool behaviour. -type Config struct { +type PoolConfig struct { // Debug flag creates new fresh worker before every request. Debug bool @@ -93,7 +93,7 @@ type Config struct { } // InitDefaults enables default config values. -func (cfg *Config) InitDefaults() { +func (cfg *PoolConfig) InitDefaults() { if cfg.NumWorkers == 0 { cfg.NumWorkers = int64(runtime.NumCPU()) } diff --git a/static_pool.go b/static_pool.go index f64a2c9a..c1dacd8d 100755 --- a/static_pool.go +++ b/static_pool.go @@ -2,7 +2,6 @@ package roadrunner import ( "context" - "fmt" "os/exec" "github.com/spiral/errors" @@ -14,9 +13,23 @@ const StopRequest = "{\"stop\":true}" var bCtx = context.Background() +// Allocator is responsible for worker allocation in the pool +type Allocator func() (WorkerBase, error) + +// ErrorEncoder encode error or make a decision based on the error type +type ErrorEncoder func(err error, w WorkerBase) (Payload, error) + +// PoolBefore is set of functions that executes BEFORE Exec +type Before func(req Payload) Payload + +// PoolAfter is set of functions that executes AFTER Exec +type After func(req Payload, resp Payload) Payload + +type PoolOptions func(p *StaticPool) + // StaticPool controls worker creation, destruction and task routing. Pool uses fixed amount of stack. type StaticPool struct { - cfg Config + cfg PoolConfig // worker command creator cmd func() *exec.Cmd @@ -25,14 +38,22 @@ type StaticPool struct { factory Factory // distributes the events - events *util.EventHandler + events util.EventsHandler // manages worker states and TTLs - ww *workerWatcher + ww WorkerWatcher + + // allocate new worker + allocator Allocator + + errEncoder ErrorEncoder + before []Before + after []After } // NewPool creates new worker pool and task multiplexer. StaticPool will initiate with one worker. -func NewPool(ctx context.Context, cmd func() *exec.Cmd, factory Factory, cfg Config) (Pool, error) { +func NewPool(ctx context.Context, cmd func() *exec.Cmd, factory Factory, cfg PoolConfig, options ...PoolOptions) (Pool, error) { + const op = errors.Op("NewPool") cfg.InitDefaults() if cfg.Debug { @@ -44,21 +65,13 @@ func NewPool(ctx context.Context, cmd func() *exec.Cmd, factory Factory, cfg Con cfg: cfg, cmd: cmd, factory: factory, - events: &util.EventHandler{}, + events: util.NewEventsHandler(), + after: make([]After, 0, 0), + before: make([]Before, 0, 0), } - p.ww = newWorkerWatcher(func(args ...interface{}) (WorkerBase, error) { - w, err := p.factory.SpawnWorkerWithContext(ctx, p.cmd()) - if err != nil { - return nil, err - } - - sw, err := NewSyncWorker(w) - if err != nil { - return nil, err - } - return sw, nil - }, p.cfg.NumWorkers, p.events) + p.allocator = newPoolAllocator(factory, cmd) + p.ww = newWorkerWatcher(p.allocator, p.cfg.NumWorkers, p.events) workers, err := p.allocateWorkers(ctx, p.cfg.NumWorkers) if err != nil { @@ -71,6 +84,13 @@ func NewPool(ctx context.Context, cmd func() *exec.Cmd, factory Factory, cfg Con return nil, err } + p.errEncoder = defaultErrEncoder(p) + + // add pool options + for i := 0; i < len(options); i++ { + options[i](p) + } + // if supervised config not nil, guess, that pool wanted to be supervised if cfg.Supervisor != nil { sp := newPoolWatcher(p, p.events, p.cfg.Supervisor) @@ -82,13 +102,25 @@ func NewPool(ctx context.Context, cmd func() *exec.Cmd, factory Factory, cfg Con return p, nil } +func PoolBefore(before ...Before) PoolOptions { + return func(p *StaticPool) { + p.before = append(p.before, before...) + } +} + +func PoolAfter(after ...After) PoolOptions { + return func(p *StaticPool) { + p.after = append(p.after, after...) + } +} + // AddListener connects event listener to the pool. func (sp *StaticPool) AddListener(listener util.EventListener) { sp.events.AddListener(listener) } -// Config returns associated pool configuration. Immutable. -func (sp *StaticPool) GetConfig() Config { +// PoolConfig returns associated pool configuration. Immutable. +func (sp *StaticPool) GetConfig() PoolConfig { return sp.cfg } @@ -107,86 +139,54 @@ func (sp *StaticPool) Exec(p Payload) (Payload, error) { return sp.execDebug(p) } w, err := sp.ww.GetFreeWorker(context.Background()) - if err != nil && errors.Is(errors.ErrWatcherStopped, err) { + if err != nil { return EmptyPayload, errors.E(op, err) - } else if err != nil { - return EmptyPayload, err } sw := w.(SyncWorker) - rsp, err := sw.Exec(p) - if err != nil { - // soft job errors are allowed - if errors.Is(errors.Exec, err) { - if sp.cfg.MaxJobs != 0 && w.State().NumExecs() >= sp.cfg.MaxJobs { - err = sp.ww.AllocateNew(bCtx) - if err != nil { - sp.events.Push(PoolEvent{Event: EventPoolError, Payload: err}) - } - - w.State().Set(StateInvalid) - err = w.Stop(bCtx) - if err != nil { - sp.events.Push(WorkerEvent{Event: EventWorkerError, Worker: w, Payload: err}) - } - } else { - sp.ww.PushWorker(w) - } - - return EmptyPayload, err - } - - sw.State().Set(StateInvalid) - sp.events.Push(PoolEvent{Event: EventWorkerDestruct, Payload: w}) - errS := w.Stop(bCtx) - - if errS != nil { - return EmptyPayload, fmt.Errorf("%v, %v", err, errS) + if len(sp.before) > 0 { + for i := 0; i < len(sp.before); i++ { + p = sp.before[i](p) } + } - return EmptyPayload, err + rsp, err := sw.Exec(p) + if err != nil { + return sp.errEncoder(err, sw) } // worker want's to be terminated if rsp.Body == nil && rsp.Context != nil && string(rsp.Context) == StopRequest { - w.State().Set(StateInvalid) - err = w.Stop(bCtx) + sw.State().Set(StateInvalid) + err = sw.Stop(bCtx) if err != nil { - sp.events.Push(WorkerEvent{Event: EventWorkerError, Worker: w, Payload: err}) + sp.events.Push(WorkerEvent{Event: EventWorkerError, Worker: sw, Payload: errors.E(op, err)}) } return sp.Exec(p) } - if sp.cfg.MaxJobs != 0 && w.State().NumExecs() >= sp.cfg.MaxJobs { + if sp.cfg.MaxJobs != 0 && sw.State().NumExecs() >= sp.cfg.MaxJobs { err = sp.ww.AllocateNew(bCtx) if err != nil { - return EmptyPayload, err + return EmptyPayload, errors.E(op, err) } } else { - sp.ww.PushWorker(w) + sp.ww.PushWorker(sw) } - return rsp, nil -} -func (sp *StaticPool) execDebug(p Payload) (Payload, error) { - sw, err := sp.ww.allocator() - if err != nil { - return EmptyPayload, err - } - - r, err := sw.(SyncWorker).Exec(p) - - if stopErr := sw.Stop(context.Background()); stopErr != nil { - sp.events.Push(WorkerEvent{Event: EventWorkerError, Worker: sw, Payload: err}) + if len(sp.after) > 0 { + for i := 0; i < len(sp.after); i++ { + rsp = sp.after[i](p, rsp) + } } - return r, err + return rsp, nil } func (sp *StaticPool) ExecWithContext(ctx context.Context, rqs Payload) (Payload, error) { - const op = errors.Op("Exec") + const op = errors.Op("Exec with context") w, err := sp.ww.GetFreeWorker(context.Background()) if err != nil { return EmptyPayload, errors.E(op, err) @@ -194,8 +194,54 @@ func (sp *StaticPool) ExecWithContext(ctx context.Context, rqs Payload) (Payload sw := w.(SyncWorker) + if len(sp.before) > 0 { + for i := 0; i < len(sp.before); i++ { + rqs = sp.before[i](rqs) + } + } + rsp, err := sw.ExecWithContext(ctx, rqs) if err != nil { + return sp.errEncoder(err, sw) + } + + // worker want's to be terminated + if rsp.Body == nil && rsp.Context != nil && string(rsp.Context) == StopRequest { + sw.State().Set(StateInvalid) + err = sw.Stop(bCtx) + if err != nil { + sp.events.Push(WorkerEvent{Event: EventWorkerError, Worker: sw, Payload: errors.E(op, err)}) + } + + return sp.Exec(rqs) + } + + if sp.cfg.MaxJobs != 0 && sw.State().NumExecs() >= sp.cfg.MaxJobs { + err = sp.ww.AllocateNew(bCtx) + if err != nil { + return EmptyPayload, errors.E(op, err) + } + } else { + sp.ww.PushWorker(sw) + } + + if len(sp.after) > 0 { + for i := 0; i < len(sp.after); i++ { + rsp = sp.after[i](rqs, rsp) + } + } + + return rsp, nil +} + +// Destroy all underlying stack (but let them to complete the task). +func (sp *StaticPool) Destroy(ctx context.Context) { + sp.ww.Destroy(ctx) +} + +func defaultErrEncoder(sp *StaticPool) ErrorEncoder { + return func(err error, w WorkerBase) (Payload, error) { + const op = errors.Op("error encoder") // soft job errors are allowed if errors.Is(errors.Exec, err) { if sp.cfg.MaxJobs != 0 && w.State().NumExecs() >= sp.cfg.MaxJobs { @@ -216,7 +262,7 @@ func (sp *StaticPool) ExecWithContext(ctx context.Context, rqs Payload) (Payload return EmptyPayload, errors.E(op, err) } - sw.State().Set(StateInvalid) + w.State().Set(StateInvalid) sp.events.Push(PoolEvent{Event: EventWorkerDestruct, Payload: w}) errS := w.Stop(bCtx) @@ -226,32 +272,36 @@ func (sp *StaticPool) ExecWithContext(ctx context.Context, rqs Payload) (Payload return EmptyPayload, errors.E(op, err) } +} - // worker want's to be terminated - if rsp.Body == nil && rsp.Context != nil && string(rsp.Context) == StopRequest { - w.State().Set(StateInvalid) - err = w.Stop(bCtx) +func newPoolAllocator(factory Factory, cmd func() *exec.Cmd) Allocator { + return func() (WorkerBase, error) { + w, err := factory.SpawnWorkerWithContext(bCtx, cmd()) if err != nil { - sp.events.Push(WorkerEvent{Event: EventWorkerError, Worker: w, Payload: errors.E(op, err)}) + return nil, err } - return sp.Exec(rqs) - } - - if sp.cfg.MaxJobs != 0 && w.State().NumExecs() >= sp.cfg.MaxJobs { - err = sp.ww.AllocateNew(bCtx) + sw, err := NewSyncWorker(w) if err != nil { - return EmptyPayload, errors.E(op, err) + return nil, err } - } else { - sp.ww.PushWorker(w) + return sw, nil } - return rsp, nil } -// Destroy all underlying stack (but let them to complete the task). -func (sp *StaticPool) Destroy(ctx context.Context) { - sp.ww.Destroy(ctx) +func (sp *StaticPool) execDebug(p Payload) (Payload, error) { + sw, err := sp.allocator() + if err != nil { + return EmptyPayload, err + } + + r, err := sw.(SyncWorker).Exec(p) + + if stopErr := sw.Stop(context.Background()); stopErr != nil { + sp.events.Push(WorkerEvent{Event: EventWorkerError, Worker: sw, Payload: err}) + } + + return r, err } // allocate required number of stack diff --git a/static_pool_test.go b/static_pool_test.go index 8f8a6f56..27907af5 100755 --- a/static_pool_test.go +++ b/static_pool_test.go @@ -14,7 +14,7 @@ import ( "github.com/stretchr/testify/assert" ) -var cfg = Config{ +var cfg = PoolConfig{ NumWorkers: int64(runtime.NumCPU()), AllocateTimeout: time.Second, DestroyTimeout: time.Second, @@ -52,7 +52,7 @@ func Test_ConfigNoErrorInitDefaults(t *testing.T) { context.Background(), func() *exec.Cmd { return exec.Command("php", "tests/client.php", "echo", "pipes") }, NewPipeFactory(), - Config{ + PoolConfig{ AllocateTimeout: time.Second, DestroyTimeout: time.Second, }, @@ -157,7 +157,7 @@ func Test_StaticPool_JobError(t *testing.T) { t.Fatal("error should be of type errors.Exec") } - assert.Contains(t, err.Error(), "exec payload: Exec: hello") + assert.Contains(t, err.Error(), "hello") } func Test_StaticPool_Broken_Replace(t *testing.T) { @@ -252,7 +252,7 @@ func Test_StaticPool_AllocateTimeout(t *testing.T) { context.Background(), func() *exec.Cmd { return exec.Command("php", "tests/client.php", "delay", "pipes") }, NewPipeFactory(), - Config{ + PoolConfig{ NumWorkers: 1, AllocateTimeout: time.Nanosecond * 1, DestroyTimeout: time.Second * 2, @@ -268,7 +268,7 @@ func Test_StaticPool_Replace_Worker(t *testing.T) { ctx, func() *exec.Cmd { return exec.Command("php", "tests/client.php", "pid", "pipes") }, NewPipeFactory(), - Config{ + PoolConfig{ NumWorkers: 1, MaxJobs: 1, AllocateTimeout: time.Second, @@ -305,7 +305,7 @@ func Test_StaticPool_Debug_Worker(t *testing.T) { ctx, func() *exec.Cmd { return exec.Command("php", "tests/client.php", "pid", "pipes") }, NewPipeFactory(), - Config{ + PoolConfig{ Debug: true, AllocateTimeout: time.Second, DestroyTimeout: time.Second, @@ -345,7 +345,7 @@ func Test_StaticPool_Stop_Worker(t *testing.T) { ctx, func() *exec.Cmd { return exec.Command("php", "tests/client.php", "stop", "pipes") }, NewPipeFactory(), - Config{ + PoolConfig{ NumWorkers: 1, AllocateTimeout: time.Second, DestroyTimeout: time.Second, @@ -385,7 +385,7 @@ func Test_Static_Pool_Destroy_And_Close(t *testing.T) { ctx, func() *exec.Cmd { return exec.Command("php", "tests/client.php", "delay", "pipes") }, NewPipeFactory(), - Config{ + PoolConfig{ NumWorkers: 1, AllocateTimeout: time.Second, DestroyTimeout: time.Second, @@ -407,7 +407,7 @@ func Test_Static_Pool_Destroy_And_Close_While_Wait(t *testing.T) { ctx, func() *exec.Cmd { return exec.Command("php", "tests/client.php", "delay", "pipes") }, NewPipeFactory(), - Config{ + PoolConfig{ NumWorkers: 1, AllocateTimeout: time.Second, DestroyTimeout: time.Second, @@ -437,7 +437,7 @@ func Test_Static_Pool_Handle_Dead(t *testing.T) { context.Background(), func() *exec.Cmd { return exec.Command("php", "tests/slow-destroy.php", "echo", "pipes") }, NewPipeFactory(), - Config{ + PoolConfig{ NumWorkers: 5, AllocateTimeout: time.Second, DestroyTimeout: time.Second, @@ -462,7 +462,7 @@ func Test_Static_Pool_Slow_Destroy(t *testing.T) { context.Background(), func() *exec.Cmd { return exec.Command("php", "tests/slow-destroy.php", "echo", "pipes") }, NewPipeFactory(), - Config{ + PoolConfig{ NumWorkers: 5, AllocateTimeout: time.Second, DestroyTimeout: time.Second, @@ -503,7 +503,7 @@ func Benchmark_Pool_Echo_Batched(b *testing.B) { ctx, func() *exec.Cmd { return exec.Command("php", "tests/client.php", "echo", "pipes") }, NewPipeFactory(), - Config{ + PoolConfig{ NumWorkers: int64(runtime.NumCPU()), AllocateTimeout: time.Second * 100, DestroyTimeout: time.Second, @@ -533,7 +533,7 @@ func Benchmark_Pool_Echo_Replaced(b *testing.B) { ctx, func() *exec.Cmd { return exec.Command("php", "tests/client.php", "echo", "pipes") }, NewPipeFactory(), - Config{ + PoolConfig{ NumWorkers: 1, MaxJobs: 1, AllocateTimeout: time.Second, diff --git a/supervisor_pool.go b/supervisor_pool.go index e23abdd1..b354b493 100755 --- a/supervisor_pool.go +++ b/supervisor_pool.go @@ -19,13 +19,13 @@ type SupervisedPool interface { type supervisedPool struct { cfg *SupervisorConfig - events *util.EventHandler + events util.EventsHandler pool Pool stopCh chan struct{} mu *sync.RWMutex } -func newPoolWatcher(pool Pool, events *util.EventHandler, cfg *SupervisorConfig) SupervisedPool { +func newPoolWatcher(pool Pool, events util.EventsHandler, cfg *SupervisorConfig) SupervisedPool { sp := &supervisedPool{ cfg: cfg, events: events, @@ -92,7 +92,7 @@ func (sp *supervisedPool) AddListener(listener util.EventListener) { sp.pool.AddListener(listener) } -func (sp *supervisedPool) GetConfig() Config { +func (sp *supervisedPool) GetConfig() PoolConfig { return sp.pool.GetConfig() } diff --git a/supervisor_test.go b/supervisor_test.go index 34172d7d..08ea356d 100644 --- a/supervisor_test.go +++ b/supervisor_test.go @@ -9,7 +9,7 @@ import ( "github.com/stretchr/testify/assert" ) -var cfgSupervised = Config{ +var cfgSupervised = PoolConfig{ NumWorkers: int64(1), AllocateTimeout: time.Second, DestroyTimeout: time.Second, @@ -69,7 +69,7 @@ func TestSupervisedPool_Exec(t *testing.T) { } func TestSupervisedPool_ExecTTL_TimedOut(t *testing.T) { - var cfgExecTTL = Config{ + var cfgExecTTL = PoolConfig{ NumWorkers: int64(1), AllocateTimeout: time.Second, DestroyTimeout: time.Second, @@ -109,7 +109,7 @@ func TestSupervisedPool_ExecTTL_TimedOut(t *testing.T) { } func TestSupervisedPool_ExecTTL_OK(t *testing.T) { - var cfgExecTTL = Config{ + var cfgExecTTL = PoolConfig{ NumWorkers: int64(1), AllocateTimeout: time.Second, DestroyTimeout: time.Second, diff --git a/util/events.go b/util/events.go index 9e12c4f7..21ebc29b 100755 --- a/util/events.go +++ b/util/events.go @@ -1,5 +1,11 @@ package util +type EventsHandler interface { + NumListeners() int + AddListener(listener EventListener) + Push(e interface{}) +} + // Event listener listens for the events produced by worker, worker pool or other servce. type EventListener func(event interface{}) @@ -8,6 +14,10 @@ type EventHandler struct { listeners []EventListener } +func NewEventsHandler() EventsHandler { + return &EventHandler{listeners: make([]EventListener, 0, 2)} +} + // NumListeners returns number of event listeners. func (eb *EventHandler) NumListeners() int { return len(eb.listeners) diff --git a/worker_watcher.go b/worker_watcher.go index 3a89554d..84be44f2 100755 --- a/worker_watcher.go +++ b/worker_watcher.go @@ -84,7 +84,7 @@ type WorkerWatcher interface { } // workerCreateFunc can be nil, but in that case, dead stack will not be replaced -func newWorkerWatcher(allocator func(args ...interface{}) (WorkerBase, error), numWorkers int64, events *util.EventHandler) *workerWatcher { +func newWorkerWatcher(allocator Allocator, numWorkers int64, events util.EventsHandler) WorkerWatcher { ww := &workerWatcher{ stack: NewWorkersStack(), allocator: allocator, @@ -99,10 +99,10 @@ func newWorkerWatcher(allocator func(args ...interface{}) (WorkerBase, error), n type workerWatcher struct { mutex sync.RWMutex stack *Stack - allocator func(args ...interface{}) (WorkerBase, error) + allocator Allocator initialNumWorkers int64 actualNumWorkers int64 - events *util.EventHandler + events util.EventsHandler } func (ww *workerWatcher) AddToWatch(ctx context.Context, workers []WorkerBase) error { |