diff --git a/.github/workflows/linux.yml b/.github/workflows/linux.yml index 0155a4a..5e25ebd 100644 --- a/.github/workflows/linux.yml +++ b/.github/workflows/linux.yml @@ -80,7 +80,7 @@ jobs: run: | mkdir -p coverage-ci cd tests - go test -timeout 20m -v -race -cover -tags=debug -failfast -coverpkg=github.com/roadrunner-server/status/v6/... -coverprofile=../coverage-ci/e2e.out -covermode=atomic ./... + go test -timeout 20m -v -race -cover -coverpkg=github.com/roadrunner-server/status/v6/... -coverprofile=../coverage-ci/e2e.out -covermode=atomic ./... - name: Archive code coverage results uses: actions/upload-artifact@v7 diff --git a/config_test.go b/config_test.go new file mode 100644 index 0000000..4927314 --- /dev/null +++ b/config_test.go @@ -0,0 +1,44 @@ +package status + +import ( + "net/http" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestConfigInitDefaults(t *testing.T) { + for _, tt := range []struct { + name string + cfg Config + wantCode int + wantTimeoutS int + }{ + { + name: "zero value", + cfg: Config{}, + wantCode: http.StatusServiceUnavailable, + wantTimeoutS: 60, + }, + { + name: "negative check timeout", + cfg: Config{CheckTimeout: -1}, + wantCode: http.StatusServiceUnavailable, + wantTimeoutS: 60, + }, + { + name: "configured values are kept", + cfg: Config{CheckTimeout: 5, UnavailableStatusCode: http.StatusInternalServerError}, + wantCode: http.StatusInternalServerError, + wantTimeoutS: 5, + }, + } { + t.Run(tt.name, func(t *testing.T) { + cfg := tt.cfg + cfg.InitDefaults() + + assert.Equal(t, tt.wantCode, cfg.UnavailableStatusCode) + assert.Equal(t, tt.wantTimeoutS, cfg.CheckTimeout) + }) + } +} diff --git a/go.mod b/go.mod index 932fb80..68898f0 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,7 @@ go 1.26 toolchain go1.26.5 require ( - github.com/roadrunner-server/api-go/v6 v6.0.0-beta.13 + github.com/roadrunner-server/api-go/v6 v6.0.0-beta.14 github.com/roadrunner-server/api-plugins/v6 v6.0.0-beta.2 github.com/roadrunner-server/endure/v2 v2.6.2 github.com/roadrunner-server/errors v1.5.0 diff --git a/go.sum b/go.sum index fe4825e..daf7adf 100644 --- a/go.sum +++ b/go.sum @@ -13,8 +13,8 @@ github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/roadrunner-server/api-go/v6 v6.0.0-beta.13 h1:BAV1aKkRp51C1OXDfEYZXgfrXqn4O7bpr6Z/m5otwd8= -github.com/roadrunner-server/api-go/v6 v6.0.0-beta.13/go.mod h1:Y4rsabWjr4Y10Jg6H8J5NDitQqlnXmGhCdgR+zyLYkI= +github.com/roadrunner-server/api-go/v6 v6.0.0-beta.14 h1:sTskv/3ImOZlUdtHuj9uT24gm1gQl/qU8rFNvn3MzhU= +github.com/roadrunner-server/api-go/v6 v6.0.0-beta.14/go.mod h1:Y4rsabWjr4Y10Jg6H8J5NDitQqlnXmGhCdgR+zyLYkI= github.com/roadrunner-server/api-plugins/v6 v6.0.0-beta.2 h1:GqsZzWQ5jMXRF1O/b8IqFz9PLpS7Ui0K4OyACLql2MI= github.com/roadrunner-server/api-plugins/v6 v6.0.0-beta.2/go.mod h1:2v4yUK5Kvbvq8C3IkDoBkuamq9h+7i/JLjyf7k1j5JM= github.com/roadrunner-server/endure/v2 v2.6.2 h1:sIB4kTyE7gtT3fDhuYWUYn6Vt/dcPtiA6FoNS1eS+84= diff --git a/handler_test.go b/handler_test.go index df68be6..ca02d0a 100644 --- a/handler_test.go +++ b/handler_test.go @@ -47,6 +47,38 @@ func (m *mockJobsChecker) JobsState(_ context.Context) ([]*jobsApi.State, error) } func (m *mockJobsChecker) Name() string { return "jobs" } +// failingWriter is a ResponseWriter whose body write always fails, which is how +// a client that hangs up mid-response looks to a handler. +type failingWriter struct { + header http.Header +} + +func newFailingWriter() *failingWriter { + return &failingWriter{header: make(http.Header)} +} + +func (w *failingWriter) Header() http.Header { return w.header } +func (w *failingWriter) WriteHeader(int) {} + +func (w *failingWriter) Write([]byte) (int, error) { + return 0, errors.New("broken pipe") +} + +// logCapture keeps the messages the handler logged. +type logCapture struct { + messages []string +} + +func (c *logCapture) Enabled(context.Context, slog.Level) bool { return true } + +func (c *logCapture) Handle(_ context.Context, r slog.Record) error { + c.messages = append(c.messages, r.Message) + return nil +} + +func (c *logCapture) WithAttrs([]slog.Attr) slog.Handler { return c } +func (c *logCapture) WithGroup(string) slog.Handler { return c } + // --- Helpers --- func newShutdownPtr(val bool) *atomic.Bool { @@ -228,6 +260,21 @@ func TestHealthHandler(t *testing.T) { assert.Empty(t, reports) }) + t.Run("Filtered_NilPlugin", func(t *testing.T) { + registry := map[string]Checker{ + "http": nil, + } + h := NewHealthHandler(registry, newShutdownPtr(false), log, http.StatusServiceUnavailable) + rec := httptest.NewRecorder() + req := httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/health?plugin=http", nil) + h.ServeHTTP(rec, req) + + // Filtered path skips a nil plugin, while the all-plugins path reports it + assert.Equal(t, http.StatusOK, rec.Code) + reports := parseReports(t, rec.Body.Bytes()) + assert.Empty(t, reports) + }) + t.Run("Filtered_Error", func(t *testing.T) { registry := map[string]Checker{ "http": &mockChecker{name: "http", err: errors.New("connection refused")}, @@ -470,6 +517,21 @@ func TestReadyHandler(t *testing.T) { assert.Empty(t, reports) }) + t.Run("Filtered_NilPlugin", func(t *testing.T) { + registry := map[string]Readiness{ + "http": nil, + } + h := NewReadyHandler(registry, newShutdownPtr(false), log, http.StatusServiceUnavailable) + rec := httptest.NewRecorder() + req := httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/ready?plugin=http", nil) + h.ServeHTTP(rec, req) + + // Filtered path skips a nil plugin, while the all-plugins path reports it + assert.Equal(t, http.StatusOK, rec.Code) + reports := parseReports(t, rec.Body.Bytes()) + assert.Empty(t, reports) + }) + t.Run("Filtered_Error", func(t *testing.T) { registry := map[string]Readiness{ "http": &mockReadiness{name: "http", err: errors.New("not ready")}, @@ -631,6 +693,78 @@ func TestJobsHandler(t *testing.T) { }) } +// --- Response Write Failures --- + +// TestHandlerWriteError checks that a handler whose response body cannot be +// written logs the failure instead of panicking. +func TestHandlerWriteError(t *testing.T) { + healthRegistry := map[string]Checker{ + "http": &mockChecker{name: "http", st: &apiStatus.Status{Code: 200}}, + } + readyRegistry := map[string]Readiness{ + "http": &mockReadiness{name: "http", st: &apiStatus.Status{Code: 200}}, + } + + for _, tt := range []struct { + newHandler func(log *slog.Logger) http.Handler + name string + target string + wantLog string + }{ + { + name: "HealthAllPlugins", + target: "/health", + newHandler: func(log *slog.Logger) http.Handler { + return NewHealthHandler(healthRegistry, newShutdownPtr(false), log, http.StatusServiceUnavailable) + }, + wantLog: "failed to write response", + }, + { + name: "HealthFiltered", + target: "/health?plugin=http", + newHandler: func(log *slog.Logger) http.Handler { + return NewHealthHandler(healthRegistry, newShutdownPtr(false), log, http.StatusServiceUnavailable) + }, + wantLog: "failed to write response", + }, + { + name: "ReadyAllPlugins", + target: "/ready", + newHandler: func(log *slog.Logger) http.Handler { + return NewReadyHandler(readyRegistry, newShutdownPtr(false), log, http.StatusServiceUnavailable) + }, + wantLog: "failed to write response", + }, + { + name: "ReadyFiltered", + target: "/ready?plugin=http", + newHandler: func(log *slog.Logger) http.Handler { + return NewReadyHandler(readyRegistry, newShutdownPtr(false), log, http.StatusServiceUnavailable) + }, + wantLog: "failed to write response", + }, + { + name: "Jobs", + target: "/jobs", + newHandler: func(log *slog.Logger) http.Handler { + jc := &mockJobsChecker{states: []*jobsApi.State{{Pipeline: "pipe1", Driver: "memory"}}} + return NewJobsHandler(jc, newShutdownPtr(false), log, http.StatusServiceUnavailable) + }, + wantLog: "failed to write jobs state report", + }, + } { + t.Run(tt.name, func(t *testing.T) { + capture := &logCapture{} + w := newFailingWriter() + req := httptest.NewRequestWithContext(context.Background(), http.MethodGet, tt.target, nil) + + tt.newHandler(slog.New(capture)).ServeHTTP(w, req) + + assert.Contains(t, capture.messages, tt.wantLog) + }) + } +} + // --- Fuzz Tests --- func FuzzHealthPluginQuery(f *testing.F) { diff --git a/plugin_test.go b/plugin_test.go index 4c550fb..9bd3589 100644 --- a/plugin_test.go +++ b/plugin_test.go @@ -3,22 +3,45 @@ package status import ( stderr "errors" "log/slog" + "net" "testing" + "time" "github.com/roadrunner-server/errors" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) -// initConfigurer is a minimal Configurer for exercising Plugin.Init's early -// error returns without standing up a full container. +// initConfigurer is a minimal Configurer for exercising Plugin.Init without +// standing up a full container. type initConfigurer struct { - has bool + cfg *Config unmarshalErr error + has bool } -func (c *initConfigurer) Has(string) bool { return c.has } -func (c *initConfigurer) UnmarshalKey(string, any) error { return c.unmarshalErr } +func (c *initConfigurer) Has(string) bool { return c.has } + +func (c *initConfigurer) UnmarshalKey(_ string, out any) error { + if c.unmarshalErr != nil { + return c.unmarshalErr + } + + // the plugin passes a **Config, which the config plugin fills in + dst, ok := out.(**Config) + if !ok { + return stderr.New("unexpected destination type") + } + + cfg := c.cfg + if cfg == nil { + cfg = &Config{} + } + + *dst = cfg + + return nil +} type initLogger struct{} @@ -37,3 +60,39 @@ func TestPluginInit(t *testing.T) { assert.True(t, errors.Is(errors.Disabled, err)) }) } + +// TestPluginServeAddressTaken checks that a listener the plugin cannot open is +// reported on the channel Serve returns. +func TestPluginServeAddressTaken(t *testing.T) { + var lc net.ListenConfig + + ln, err := lc.Listen(t.Context(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = ln.Close() }) + + p := &Plugin{} + require.NoError(t, p.Init(&initConfigurer{has: true, cfg: &Config{Address: ln.Addr().String()}}, initLogger{})) + + errCh := p.Serve() + t.Cleanup(p.StopHTTPServer) + + select { + case serveErr := <-errCh: + require.Error(t, serveErr) + case <-time.After(time.Second * 10): + t.Fatal("the address is taken, but the plugin reported no error") + } +} + +// TestPluginUnknownPlugin pins the sentinel both lookups wrap, which the rpc +// service turns into the "no such plugin" reply. +func TestPluginUnknownPlugin(t *testing.T) { + p := &Plugin{} + require.NoError(t, p.Init(&initConfigurer{has: true}, initLogger{})) + + _, err := p.status("nonexistent") + require.ErrorIs(t, err, errPluginNotFound) + + _, err = p.ready("nonexistent") + require.ErrorIs(t, err, errPluginNotFound) +} diff --git a/rpc.go b/rpc.go index 2a42581..bb9841a 100644 --- a/rpc.go +++ b/rpc.go @@ -3,7 +3,7 @@ package status import ( "log/slog" - statusV2 "github.com/roadrunner-server/api-go/v6/status/v2" + statusV1 "github.com/roadrunner-server/api-go/v6/status/v1" "github.com/roadrunner-server/errors" ) @@ -13,7 +13,7 @@ type rpc struct { } // Status returns the current status of the provided plugin. -func (r *rpc) Status(in *statusV2.StatusRequest, out *statusV2.StatusResponse) error { +func (r *rpc) Status(in *statusV1.Request, out *statusV1.Response) error { const op = errors.Op("checker_rpc_status") plugin := in.GetPlugin() r.log.Debug("Status method was invoked", "plugin", plugin) @@ -33,7 +33,7 @@ func (r *rpc) Status(in *statusV2.StatusRequest, out *statusV2.StatusResponse) e } // Ready returns the readiness check of the provided plugin. -func (r *rpc) Ready(in *statusV2.StatusRequest, out *statusV2.StatusResponse) error { +func (r *rpc) Ready(in *statusV1.Request, out *statusV1.Response) error { const op = errors.Op("checker_rpc_ready") plugin := in.GetPlugin() r.log.Debug("Ready method was invoked", "plugin", plugin) diff --git a/tests/configs/.rr-ready-init.yaml b/tests/configs/.rr-ready-init.yaml index 2f4770e..7106961 100644 --- a/tests/configs/.rr-ready-init.yaml +++ b/tests/configs/.rr-ready-init.yaml @@ -20,7 +20,6 @@ http: max_request_size: 1024 uploads: forbid: [ ".php", ".exe", ".bat" ] - trusted_subnets: [ "10.0.0.0/8", "127.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "::1/128", "fc00::/7", "fe80::/10" ] pool: num_workers: 1 allocate_timeout: 5s diff --git a/tests/configs/.rr-status-init.yaml b/tests/configs/.rr-status-init.yaml index f79cebd..3ef3633 100644 --- a/tests/configs/.rr-status-init.yaml +++ b/tests/configs/.rr-status-init.yaml @@ -20,7 +20,6 @@ http: max_request_size: 1024 uploads: forbid: [ ".php", ".exe", ".bat" ] - trusted_subnets: [ "10.0.0.0/8", "127.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "::1/128", "fc00::/7", "fe80::/10" ] pool: num_workers: 2 allocate_timeout: 60s diff --git a/tests/doc.go b/tests/doc.go index 757882e..8e4e1bc 100644 --- a/tests/doc.go +++ b/tests/doc.go @@ -1,4 +1,4 @@ -// Package status contains integration tests for the status plugin. The tests -// exercise HTTP endpoints (/health, /ready, /jobs) and RPC methods using a -// full plugin lifecycle managed by the endure framework. -package status +// Package tests contains the integration tests of the status plugin. They +// drive the /health, /ready and /jobs endpoints and the status rpc methods +// through a full plugin lifecycle managed by the endure framework. +package tests diff --git a/tests/endpoints_test.go b/tests/endpoints_test.go new file mode 100644 index 0000000..c150267 --- /dev/null +++ b/tests/endpoints_test.go @@ -0,0 +1,169 @@ +package tests + +import ( + "context" + "net/http" + "testing" + "time" + + "tests/helpers" + + statusV1 "github.com/roadrunner-server/api-go/v6/status/v1" + httpPlugin "github.com/roadrunner-server/http/v6" + rpcPlugin "github.com/roadrunner-server/rpc/v6" + "github.com/roadrunner-server/server/v6" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + // sharedHTTPAddr is the http address of both the status-init and the + // ready-init config, so the tests using them have to run one after another: + // no t.Parallel here. + sharedHTTPAddr = "127.0.0.1:11933" + + statusInitCfg = "configs/.rr-status-init.yaml" + statusInitAddr = "127.0.0.1:34333" + statusInitURL = "http://" + statusInitAddr + statusInitRPC = "127.0.0.1:6005" + + readyInitCfg = "configs/.rr-ready-init.yaml" + readyInitAddr = "127.0.0.1:34334" + readyInitURL = "http://" + readyInitAddr + readyInitRPC = "127.0.0.1:6006" +) + +// TestStatusEndpoints drives /health, /ready and /jobs against a container +// running the http plugin. The rpc plugin is deliberately left out: the +// ?plugin=http&plugin=rpc queries then prove that a name missing from the +// registry is skipped instead of reported. +func TestStatusEndpoints(t *testing.T) { + helpers.Start(t, statusInitCfg, []any{ + &server.Plugin{}, + &httpPlugin.Plugin{}, + helpers.NewStatusPlugin(t), + }, helpers.WithTCPProbe(statusInitAddr)) + + // the http plugin binds its listener after the pool is allocated, which is + // what makes its status and readiness meaningful + helpers.WaitListener(t, "tcp", sharedHTTPAddr) + + t.Run("HealthFiltered", func(t *testing.T) { + code, reports := helpers.GetReports(t, statusInitURL+"/health?plugin=http&plugin=rpc") + assert.Equal(t, http.StatusOK, code) + + require.Len(t, reports, 1) + assert.Equal(t, "http", reports[0].PluginName) + assert.Empty(t, reports[0].ErrorMessage) + assert.Equal(t, http.StatusOK, reports[0].StatusCode) + }) + + t.Run("HealthAll", func(t *testing.T) { + code, reports := helpers.GetReports(t, statusInitURL+"/health") + assert.Equal(t, http.StatusOK, code) + + require.Len(t, reports, 1) + assert.Equal(t, "http", reports[0].PluginName) + assert.Empty(t, reports[0].ErrorMessage) + assert.Equal(t, http.StatusOK, reports[0].StatusCode) + }) + + t.Run("ReadyFiltered", func(t *testing.T) { + code, reports := helpers.GetReports(t, statusInitURL+"/ready?plugin=http&plugin=rpc") + assert.Equal(t, http.StatusOK, code) + + require.Len(t, reports, 1) + assert.Equal(t, "http", reports[0].PluginName) + assert.Empty(t, reports[0].ErrorMessage) + assert.Equal(t, http.StatusOK, reports[0].StatusCode) + }) + + t.Run("ReadyAll", func(t *testing.T) { + code, reports := helpers.GetReports(t, statusInitURL+"/ready") + assert.Equal(t, http.StatusOK, code) + + require.Len(t, reports, 1) + assert.Equal(t, "http", reports[0].PluginName) + assert.Empty(t, reports[0].ErrorMessage) + assert.Equal(t, http.StatusOK, reports[0].StatusCode) + }) + + t.Run("JobsWithoutJobsPlugin", func(t *testing.T) { + code, body := helpers.GetBody(t, statusInitURL+"/jobs") + assert.Equal(t, http.StatusServiceUnavailable, code) + assert.Contains(t, body, "jobs plugin not found") + }) +} + +// TestReadinessWorkerBusy occupies the only worker of the pool with a request +// the worker never answers, which leaves the http plugin without a ready +// worker. The rpc assertion lives here rather than in rpc_test.go to avoid a +// second boot of this container. +func TestReadinessWorkerBusy(t *testing.T) { + helpers.Start(t, readyInitCfg, []any{ + &rpcPlugin.Plugin{}, + &server.Plugin{}, + &httpPlugin.Plugin{}, + helpers.NewStatusPlugin(t), + }, helpers.WithTCPProbe(readyInitAddr)) + + helpers.WaitListener(t, "tcp", sharedHTTPAddr) + + occupyWorker(t, "http://"+sharedHTTPAddr) + + readyURL := readyInitURL + "/ready?plugin=http&plugin=rpc" + require.Eventually(t, func() bool { + return statusCode(t.Context(), readyURL) == http.StatusServiceUnavailable + }, time.Second*15, time.Millisecond*20, "the pool kept a ready worker") + + code, reports := helpers.GetReports(t, readyURL) + assert.Equal(t, http.StatusServiceUnavailable, code) + + require.Len(t, reports, 1) + assert.Equal(t, "http", reports[0].PluginName) + assert.Equal(t, "internal server error, see logs", reports[0].ErrorMessage) + assert.Equal(t, http.StatusServiceUnavailable, reports[0].StatusCode) + + rsp := &statusV1.Response{} + require.NoError(t, helpers.RPC(t, readyInitRPC).Call("status.Ready", &statusV1.Request{Plugin: "http"}, rsp)) + assert.Equal(t, int64(http.StatusServiceUnavailable), rsp.GetCode()) +} + +// occupyWorker sends a request the worker answers only after a sleep longer +// than the test, so the worker stays busy. The request is canceled before the +// container is stopped. +func occupyWorker(t *testing.T, url string) { + t.Helper() + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + require.NoError(t, err) + + go func() { + rsp, errR := http.DefaultClient.Do(req) + if errR == nil { + _ = rsp.Body.Close() + } + }() +} + +// statusCode returns the response code of a GET, or -1 if the request failed. +// It reports nothing to the testing framework, so it is safe to call from a +// require.Eventually condition, which runs in its own goroutine. +func statusCode(ctx context.Context, url string) int { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return -1 + } + + rsp, err := http.DefaultClient.Do(req) + if err != nil { + return -1 + } + + _ = rsp.Body.Close() + + return rsp.StatusCode +} diff --git a/tests/go.mod b/tests/go.mod index 97cfc6c..fbb890b 100644 --- a/tests/go.mod +++ b/tests/go.mod @@ -5,12 +5,12 @@ go 1.26 toolchain go1.26.5 require ( - github.com/roadrunner-server/api-go/v6 v6.0.0-beta.13 + github.com/roadrunner-server/api-go/v6 v6.0.0-beta.14 github.com/roadrunner-server/config/v6 v6.0.0-beta.3 github.com/roadrunner-server/endure/v2 v2.6.2 github.com/roadrunner-server/goridge/v4 v4.0.0-beta.3 - github.com/roadrunner-server/http/v6 v6.0.0-beta.8 - github.com/roadrunner-server/jobs/v6 v6.0.0-beta.8 + github.com/roadrunner-server/http/v6 v6.0.0-beta.8.0.20260812194928-5567f9e0d93a + github.com/roadrunner-server/jobs/v6 v6.0.0-beta.8.0.20260812194934-8622d89d2a7e github.com/roadrunner-server/logger/v6 v6.0.0-beta.3 github.com/roadrunner-server/memory/v6 v6.0.0-beta.4 github.com/roadrunner-server/rpc/v6 v6.0.0-beta.5 diff --git a/tests/go.sum b/tests/go.sum index 5a46c66..c585b0c 100644 --- a/tests/go.sum +++ b/tests/go.sum @@ -78,8 +78,8 @@ github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= github.com/quic-go/quic-go v0.61.0 h1:ui88A53s8MSVYLC56en0KQ17HARk+9986Dn0SBfKNvA= github.com/quic-go/quic-go v0.61.0/go.mod h1:9So2anK4Tp22URSQq00k+Vo2PNkle96ycDPDHL4s9vs= -github.com/roadrunner-server/api-go/v6 v6.0.0-beta.13 h1:BAV1aKkRp51C1OXDfEYZXgfrXqn4O7bpr6Z/m5otwd8= -github.com/roadrunner-server/api-go/v6 v6.0.0-beta.13/go.mod h1:Y4rsabWjr4Y10Jg6H8J5NDitQqlnXmGhCdgR+zyLYkI= +github.com/roadrunner-server/api-go/v6 v6.0.0-beta.14 h1:sTskv/3ImOZlUdtHuj9uT24gm1gQl/qU8rFNvn3MzhU= +github.com/roadrunner-server/api-go/v6 v6.0.0-beta.14/go.mod h1:Y4rsabWjr4Y10Jg6H8J5NDitQqlnXmGhCdgR+zyLYkI= github.com/roadrunner-server/api-plugins/v6 v6.0.0-beta.2 h1:GqsZzWQ5jMXRF1O/b8IqFz9PLpS7Ui0K4OyACLql2MI= github.com/roadrunner-server/api-plugins/v6 v6.0.0-beta.2/go.mod h1:2v4yUK5Kvbvq8C3IkDoBkuamq9h+7i/JLjyf7k1j5JM= github.com/roadrunner-server/config/v6 v6.0.0-beta.3 h1:G0EUzJ6Yw4UnleM6BhnOBbYPXKDHRmCJiGhC3nXDBwI= @@ -94,10 +94,10 @@ github.com/roadrunner-server/events v1.0.1 h1:waCkKhxhzdK3VcI1xG22l+h+0J+Nfdpxjh github.com/roadrunner-server/events v1.0.1/go.mod h1:WZRqoEVaFm209t52EuoT7ISUtvX6BrCi6bI/7pjkVC0= github.com/roadrunner-server/goridge/v4 v4.0.0-beta.3 h1:+kUw00/fpqwdMWrPMYW+OZH3O4gEar8hqrY7I+nAztA= github.com/roadrunner-server/goridge/v4 v4.0.0-beta.3/go.mod h1:1aHppV68y/VqRED/AsfNg59sft9aQOhqgr5Z5n49jbM= -github.com/roadrunner-server/http/v6 v6.0.0-beta.8 h1:habLZdPLG57XATjBLiTmDujnVgkOc4bgkHK39lMwYU8= -github.com/roadrunner-server/http/v6 v6.0.0-beta.8/go.mod h1:tv/QMqbNcKbqtdt/iRH4kRwvCWdhRLWwGsfAoMBC9F8= -github.com/roadrunner-server/jobs/v6 v6.0.0-beta.8 h1:kzt/0wepfVKkjsFf24cBwdc7lan9FHAdU2AoyG9Xen4= -github.com/roadrunner-server/jobs/v6 v6.0.0-beta.8/go.mod h1:6QiZfsJCMHYW8p36SQlRRdCZmad/4q7OBNezWWfg/AY= +github.com/roadrunner-server/http/v6 v6.0.0-beta.8.0.20260812194928-5567f9e0d93a h1:vNJjSucO763ihyb+9fw/+TZ0e9ZJdB4QF6Z5Ny6EkQQ= +github.com/roadrunner-server/http/v6 v6.0.0-beta.8.0.20260812194928-5567f9e0d93a/go.mod h1:5BAslY4ApCpDUliijJAkARgepnZreTkXl3tB8MP8Cio= +github.com/roadrunner-server/jobs/v6 v6.0.0-beta.8.0.20260812194934-8622d89d2a7e h1:Y63uBwjrChd1MsEaLCjfAi/r9pVckW4mt9PQ+/XJ9dE= +github.com/roadrunner-server/jobs/v6 v6.0.0-beta.8.0.20260812194934-8622d89d2a7e/go.mod h1:y7Uv5RtzYQpMcfvkhzdSw54DGUb9VRJzoS66b8WLkno= github.com/roadrunner-server/logger/v6 v6.0.0-beta.3 h1:eoJKXAUSyykDfVX6eTUhmAn6Y8pS/LyI5fDP4H+G5rQ= github.com/roadrunner-server/logger/v6 v6.0.0-beta.3/go.mod h1:MwHb3AbltHYtu7nRpml5NeYu7O+W8rCpDBeNTTEoE1M= github.com/roadrunner-server/memory/v6 v6.0.0-beta.4 h1:jeqC7VH/FBuBcXTfOhPU/GafVUMS+OWNE3YkMKCXnQw= diff --git a/tests/helpers/http.go b/tests/helpers/http.go new file mode 100644 index 0000000..4eed8b0 --- /dev/null +++ b/tests/helpers/http.go @@ -0,0 +1,65 @@ +package helpers + +import ( + "encoding/json" + "io" + "net/http" + "testing" + "time" + + "github.com/roadrunner-server/status/v6" + "github.com/stretchr/testify/require" +) + +// requestTimeout bounds a single request, so a listener that accepts the +// connection but never answers fails the test instead of hanging until the +// go test timeout. +const requestTimeout = time.Second * 10 + +// GetBody issues a GET and returns the status code and the body. The body is +// read and closed before returning. +func GetBody(t *testing.T, url string) (int, string) { + t.Helper() + + req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, url, nil) + require.NoError(t, err) + + client := &http.Client{Timeout: requestTimeout} + + r, err := client.Do(req) + require.NoError(t, err) + + defer func() { + _ = r.Body.Close() + }() + + b, err := io.ReadAll(r.Body) + require.NoError(t, err) + + return r.StatusCode, string(b) +} + +// GetReports issues a GET against /health or /ready and decodes the plugin reports. +func GetReports(t *testing.T, url string) (int, []*status.Report) { + t.Helper() + + return getJSONList[status.Report](t, url) +} + +// GetJobsReports issues a GET against /jobs and decodes the pipeline reports. +func GetJobsReports(t *testing.T, url string) (int, []*status.JobsReport) { + t.Helper() + + return getJSONList[status.JobsReport](t, url) +} + +func getJSONList[T any](t *testing.T, url string) (int, []*T) { + t.Helper() + + code, body := GetBody(t, url) + + var list []*T + require.NoError(t, json.Unmarshal([]byte(body), &list), "body: %s", body) + + return code, list +} diff --git a/tests/helpers/listener.go b/tests/helpers/listener.go new file mode 100644 index 0000000..280dabc --- /dev/null +++ b/tests/helpers/listener.go @@ -0,0 +1,34 @@ +package helpers + +import ( + "net" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +const ( + // listenerTimeout caps how long WaitListener waits for an address to accept. + listenerTimeout = time.Second * 15 + // listenerTick is the interval between two connection attempts. + listenerTick = time.Millisecond * 20 +) + +// WaitListener waits until addr accepts a connection. The http plugin binds its +// listener only after the worker pool is allocated, so the readiness of the +// status listener says nothing about it. +func WaitListener(t *testing.T, network, addr string) { + t.Helper() + + require.Eventually(t, func() bool { + var d net.Dialer + + conn, err := d.DialContext(t.Context(), network, addr) + if err != nil { + return false + } + + return conn.Close() == nil + }, listenerTimeout, listenerTick, "listener %s did not start", addr) +} diff --git a/tests/helpers/rpc.go b/tests/helpers/rpc.go new file mode 100644 index 0000000..bde96d0 --- /dev/null +++ b/tests/helpers/rpc.go @@ -0,0 +1,26 @@ +package helpers + +import ( + "net" + "net/rpc" + "testing" + + goridgeRpc "github.com/roadrunner-server/goridge/v4/pkg/rpc" + "github.com/stretchr/testify/require" +) + +// RPC dials the RoadRunner rpc plugin at addr. The client, and with it the +// connection, is closed by t.Cleanup. +func RPC(t *testing.T, addr string) *rpc.Client { + t.Helper() + + var d net.Dialer + + conn, err := d.DialContext(t.Context(), "tcp", addr) + require.NoError(t, err) + + client := rpc.NewClientWithCodec(goridgeRpc.NewClientCodec(conn)) + t.Cleanup(func() { _ = client.Close() }) + + return client +} diff --git a/tests/helpers/rr.go b/tests/helpers/rr.go new file mode 100644 index 0000000..8d699aa --- /dev/null +++ b/tests/helpers/rr.go @@ -0,0 +1,129 @@ +package helpers + +import ( + "context" + "log/slog" + "net" + "sync" + "testing" + "time" + + "github.com/roadrunner-server/config/v6" + "github.com/roadrunner-server/endure/v2" + "github.com/roadrunner-server/logger/v6" + "github.com/stretchr/testify/require" +) + +const ( + // configVersion is the config schema version used by the test configs. + configVersion = "2024.1.0" + // defaultGraceful is the endure graceful shutdown timeout. + defaultGraceful = time.Minute + // probeTimeout caps how long Start waits for the server to answer the probe. + probeTimeout = time.Second * 15 + probeTick = time.Millisecond * 20 + probeDial = time.Second +) + +// bootCfg holds the options applied to a container before it is started. +type bootCfg struct { + graceful time.Duration + cfgTimeout time.Duration + probe func(ctx context.Context) bool +} + +// Option customizes the container built by Start. +type Option func(*bootCfg) + +// WithGracefulTimeout sets the endure graceful shutdown timeout. +func WithGracefulTimeout(d time.Duration) Option { + return func(b *bootCfg) { b.graceful = d } +} + +// WithConfigTimeout sets the timeout the config plugin hands to the plugins +// that drain on shutdown. +func WithConfigTimeout(d time.Duration) Option { + return func(b *bootCfg) { b.cfgTimeout = d } +} + +// WithTCPProbe makes Start return only once addr accepts a connection. +func WithTCPProbe(addr string) Option { + return func(b *bootCfg) { + b.probe = func(ctx context.Context) bool { + d := net.Dialer{Timeout: probeDial} + + conn, err := d.DialContext(ctx, "tcp", addr) + if err != nil { + return false + } + + return conn.Close() == nil + } + } +} + +// Start registers the plugins, boots the container and waits for the probe, if +// any, to answer. Errors arriving on the container channel are reported through +// t.Errorf and stop the container, but they do not abort the test. +// +// The returned stop is idempotent and also registered with t.Cleanup, so tests +// asserting on the behavior during shutdown can stop the container mid-test. It +// returns only after cont.Stop has completed. +func Start(t *testing.T, cfgPath string, plugins []any, opts ...Option) func() { + t.Helper() + + bc := &bootCfg{graceful: defaultGraceful} + for _, o := range opts { + o(bc) + } + + cont := endure.New(slog.LevelDebug, endure.GracefulShutdownTimeout(bc.graceful)) + cfg := &config.Plugin{Version: configVersion, Path: cfgPath, Timeout: bc.cfgTimeout} + + require.NoError(t, cont.RegisterAll(append([]any{cfg, &logger.Plugin{}}, plugins...)...)) + require.NoError(t, cont.Init()) + + ch, err := cont.Serve() + require.NoError(t, err) + + stopCont := sync.OnceValue(cont.Stop) + done := make(chan struct{}) + wg := &sync.WaitGroup{} + + wg.Go(func() { + for { + select { + case res := <-ch: + if res == nil { + return + } + + t.Errorf("plugin %s reported an error: %v", res.VertexID, res.Error) + + if errS := stopCont(); errS != nil { + t.Errorf("container stop: %v", errS) + } + case <-done: + if errS := stopCont(); errS != nil { + t.Errorf("container stop: %v", errS) + } + + return + } + } + }) + + // The drain goroutine calls t.Errorf, so it has to be joined while the test + // is still running. + stop := sync.OnceFunc(func() { + close(done) + wg.Wait() + }) + t.Cleanup(stop) + + if bc.probe != nil { + require.Eventually(t, func() bool { return bc.probe(t.Context()) }, probeTimeout, probeTick, "server did not become ready") + } + + return stop +} diff --git a/tests/helpers/status.go b/tests/helpers/status.go new file mode 100644 index 0000000..d996f88 --- /dev/null +++ b/tests/helpers/status.go @@ -0,0 +1,21 @@ +package helpers + +import ( + "testing" + + "github.com/roadrunner-server/status/v6" +) + +// NewStatusPlugin returns a status plugin whose http listener is closed by +// t.Cleanup. Plugin.Stop only raises the shutdown flag, so without the explicit +// close the address stays bound and the next test on it fails to listen. The +// cleanup is registered before the container is started, so it runs after the +// container has been stopped. +func NewStatusPlugin(t *testing.T) *status.Plugin { + t.Helper() + + p := &status.Plugin{} + t.Cleanup(p.StopHTTPServer) + + return p +} diff --git a/tests/jobs_test.go b/tests/jobs_test.go new file mode 100644 index 0000000..231306c --- /dev/null +++ b/tests/jobs_test.go @@ -0,0 +1,80 @@ +package tests + +import ( + "net/http" + "testing" + + "tests/helpers" + + statusV1 "github.com/roadrunner-server/api-go/v6/status/v1" + "github.com/roadrunner-server/jobs/v6" + "github.com/roadrunner-server/memory/v6" + rpcPlugin "github.com/roadrunner-server/rpc/v6" + "github.com/roadrunner-server/server/v6" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + jobsCfg = "configs/.rr-jobs-status.yaml" + jobsAddr = "127.0.0.1:35544" + jobsURL = "http://" + jobsAddr + jobsRPC = "127.0.0.1:6007" +) + +// jobsPlugins is the stack the two memory pipelines of the jobs config need. +func jobsPlugins(t *testing.T) []any { + t.Helper() + + return []any{ + &rpcPlugin.Plugin{}, + &server.Plugin{}, + &jobs.Plugin{}, + &memory.Plugin{}, + helpers.NewStatusPlugin(t), + } +} + +// TestJobsStatus checks that /jobs reports both pipelines of the config with +// the priority they were declared with and an empty queue. +func TestJobsStatus(t *testing.T) { + helpers.Start(t, jobsCfg, jobsPlugins(t), helpers.WithTCPProbe(jobsAddr)) + + code, reports := helpers.GetJobsReports(t, jobsURL+"/jobs") + assert.Equal(t, http.StatusOK, code) + require.Len(t, reports, 2) + + for _, report := range reports { + assert.Equal(t, uint64(13), report.Priority) + assert.True(t, report.Ready) + assert.Equal(t, int64(0), report.Active) + assert.Equal(t, int64(0), report.Delayed) + assert.Equal(t, int64(0), report.Reserved) + assert.Equal(t, "memory", report.Driver) + assert.Empty(t, report.ErrorMessage) + } +} + +// TestJobsReadiness checks the readiness of the jobs plugin over both the http +// endpoint and the rpc service. +func TestJobsReadiness(t *testing.T) { + helpers.Start(t, jobsCfg, jobsPlugins(t), helpers.WithTCPProbe(jobsAddr)) + + code, reports := helpers.GetReports(t, jobsURL+"/ready?plugin=jobs") + assert.Equal(t, http.StatusOK, code) + + require.Len(t, reports, 1) + assert.Equal(t, "jobs", reports[0].PluginName) + assert.Empty(t, reports[0].ErrorMessage) + assert.Equal(t, http.StatusOK, reports[0].StatusCode) + + client := helpers.RPC(t, jobsRPC) + + for _, method := range []string{"status.Ready", "status.Status"} { + t.Run(method, func(t *testing.T) { + rsp := &statusV1.Response{} + require.NoError(t, client.Call(method, &statusV1.Request{Plugin: "jobs"}, rsp)) + assert.Equal(t, int64(http.StatusOK), rsp.GetCode()) + }) + } +} diff --git a/tests/php_test_files/jobs/jobs_ok.php b/tests/php_test_files/jobs/jobs_ok.php index a2176c6..8485067 100644 --- a/tests/php_test_files/jobs/jobs_ok.php +++ b/tests/php_test_files/jobs/jobs_ok.php @@ -1,24 +1,16 @@ waitTask()) { try { $task->complete(); } catch (\Throwable $e) { - $rr->error((string)$e); + $task->fail((string)$e); } } diff --git a/tests/plugin_test.go b/tests/plugin_test.go deleted file mode 100644 index ee86558..0000000 --- a/tests/plugin_test.go +++ /dev/null @@ -1,879 +0,0 @@ -package status - -import ( - "context" - "encoding/json" - "io" - "log/slog" - "net" - "net/http" - "net/rpc" - "os" - "os/signal" - "sync" - "syscall" - "testing" - "time" - - statusV2 "github.com/roadrunner-server/api-go/v6/status/v2" - "github.com/roadrunner-server/config/v6" - "github.com/roadrunner-server/endure/v2" - goridgeRpc "github.com/roadrunner-server/goridge/v4/pkg/rpc" - httpPlugin "github.com/roadrunner-server/http/v6" - "github.com/roadrunner-server/jobs/v6" - "github.com/roadrunner-server/logger/v6" - "github.com/roadrunner-server/memory/v6" - rpcPlugin "github.com/roadrunner-server/rpc/v6" - "github.com/roadrunner-server/server/v6" - "github.com/roadrunner-server/status/v6" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -// newStatusRPCClient dials the rpc plugin and returns a net/rpc client speaking -// the goridge codec for the status.v2.StatusService methods. -func newStatusRPCClient(t *testing.T, address string) *rpc.Client { - t.Helper() - conn, err := net.Dial("tcp", address) - require.NoError(t, err) - client := rpc.NewClientWithCodec(goridgeRpc.NewClientCodec(conn)) - t.Cleanup(func() { _ = client.Close() }) - return client -} - -func TestStatusHttp(t *testing.T) { - cont := endure.New(slog.LevelDebug, endure.GracefulShutdownTimeout(time.Minute)) - - cfg := &config.Plugin{ - Version: "2024.1.0", - Path: "configs/.rr-status-init.yaml", - } - - sp := &status.Plugin{} - err := cont.RegisterAll( - cfg, - &logger.Plugin{}, - &server.Plugin{}, - &httpPlugin.Plugin{}, - sp, - ) - 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) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - 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 <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second) - t.Run("CheckerGetStatus", checkHTTPStatus) - t.Run("CheckerGetStatusAll", checkHTTPStatusAll) - t.Run("JobsEndpointWithoutPlugin", func(t *testing.T) { - req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, "http://127.0.0.1:34333/jobs", nil) - assert.NoError(t, err) - - r, err := http.DefaultClient.Do(req) - assert.NoError(t, err) - require.NotNil(t, r) - - b, err := io.ReadAll(r.Body) - assert.NoError(t, err) - assert.Equal(t, http.StatusServiceUnavailable, r.StatusCode) - assert.Contains(t, string(b), "jobs plugin not found") - - err = r.Body.Close() - assert.NoError(t, err) - }) - - stopCh <- struct{}{} - wg.Wait() - - t.Cleanup(func() { - sp.StopHTTPServer() - }) -} - -func TestStatusRPC(t *testing.T) { - cont := endure.New(slog.LevelDebug, endure.GracefulShutdownTimeout(time.Minute)) - - cfg := &config.Plugin{ - Version: "2024.1.0", - Path: "configs/.rr-status-init.yaml", - } - - sp := &status.Plugin{} - err := cont.RegisterAll( - cfg, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - sp, - &httpPlugin.Plugin{}, - ) - 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) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - 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 <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second) - t.Run("CheckerGetStatusRpc", func(t *testing.T) { - checkRPCStatus(t, "http", 200, "6005") - }) - stopCh <- struct{}{} - wg.Wait() - - t.Cleanup(func() { - sp.StopHTTPServer() - }) -} - -func TestReadyHttp(t *testing.T) { - cont := endure.New(slog.LevelDebug, endure.GracefulShutdownTimeout(time.Minute)) - - cfg := &config.Plugin{ - Version: "2024.1.0", - Path: "configs/.rr-status-init.yaml", - } - - sp := &status.Plugin{} - err := cont.RegisterAll( - cfg, - &logger.Plugin{}, - &server.Plugin{}, - &httpPlugin.Plugin{}, - sp, - ) - 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) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - 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 <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second) - t.Run("CheckerGetReadiness", checkHTTPReadiness) - t.Run("CheckerGetReadinessAll", checkHTTPReadinessAll) - - stopCh <- struct{}{} - wg.Wait() - - t.Cleanup(func() { - sp.StopHTTPServer() - }) -} - -func TestReadinessRPCWorkerNotReady(t *testing.T) { - cont := endure.New(slog.LevelDebug, endure.GracefulShutdownTimeout(time.Minute)) - - cfg := &config.Plugin{ - Version: "2024.1.0", - Path: "configs/.rr-ready-init.yaml", - } - - sp := &status.Plugin{} - err := cont.RegisterAll( - cfg, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - &httpPlugin.Plugin{}, - sp, - ) - 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) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - 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 <-stopCh: - // timeout, error here is OK, because in the PHP we are sleeping for the 300s - _ = cont.Stop() - return - } - } - }) - - time.Sleep(time.Second) - t.Run("DoHttpReq", doHTTPReq) - time.Sleep(time.Second * 5) - t.Run("CheckerGetReadiness2", checkHTTPReadiness2) - t.Run("CheckerGetRpcReadiness", func(t *testing.T) { - checkRPCReadiness(t, "http", 503, "6006") - }) - stopCh <- struct{}{} - wg.Wait() - - t.Cleanup(func() { - sp.StopHTTPServer() - }) -} - -func TestJobsStatus(t *testing.T) { - cont := endure.New(slog.LevelDebug, endure.GracefulShutdownTimeout(time.Minute)) - - cfg := &config.Plugin{ - Version: "2023.1.0", - Path: "configs/.rr-jobs-status.yaml", - } - - sp := &status.Plugin{} - err := cont.RegisterAll( - cfg, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - sp, - &jobs.Plugin{}, - &memory.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - require.NoError(t, err) - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - 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 <-stopCh: - // timeout, error here is OK, because in the PHP we are sleeping for the 300s - _ = cont.Stop() - return - } - } - }) - - time.Sleep(time.Second) - - req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, "http://127.0.0.1:35544/jobs", nil) - assert.NoError(t, err) - - r, err := http.DefaultClient.Do(req) - assert.NoError(t, err) - require.NotNil(t, r) - - b, err := io.ReadAll(r.Body) - assert.NoError(t, err) - assert.Equal(t, http.StatusOK, r.StatusCode) - - jr := make([]*status.JobsReport, 0, 2) - - err = json.Unmarshal(b, &jr) - require.NoError(t, err) - - require.Len(t, jr, 2) - require.Equal(t, jr[0].Priority, uint64(13)) - require.Equal(t, jr[0].Ready, true) - require.Equal(t, jr[0].Active, int64(0)) - require.Equal(t, jr[0].Delayed, int64(0)) - require.Equal(t, jr[0].Reserved, int64(0)) - require.Equal(t, jr[0].Driver, "memory") - require.Equal(t, jr[0].ErrorMessage, "") - - require.Equal(t, jr[1].Priority, uint64(13)) - require.Equal(t, jr[1].Ready, true) - require.Equal(t, jr[1].Active, int64(0)) - require.Equal(t, jr[1].Delayed, int64(0)) - require.Equal(t, jr[1].Reserved, int64(0)) - require.Equal(t, jr[1].Driver, "memory") - require.Equal(t, jr[1].ErrorMessage, "") - - err = r.Body.Close() - assert.NoError(t, err) - - stopCh <- struct{}{} - wg.Wait() - - t.Cleanup(func() { - sp.StopHTTPServer() - }) -} - -func TestJobsReadiness(t *testing.T) { - cont := endure.New(slog.LevelDebug, endure.GracefulShutdownTimeout(time.Minute)) - - cfg := &config.Plugin{ - Version: "2023.2.0", - Path: "configs/.rr-jobs-status.yaml", - } - - sp := &status.Plugin{} - err := cont.RegisterAll( - cfg, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - sp, - &jobs.Plugin{}, - &memory.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - require.NoError(t, err) - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - 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 <-stopCh: - // timeout, error here is OK, because in the PHP we are sleeping for the 300s - _ = cont.Stop() - return - } - } - }) - - time.Sleep(time.Second) - t.Run("checkJobsReadiness", checkJobsReadiness) - t.Run("checkJobsRPC", func(t *testing.T) { - checkRPCReadiness(t, "jobs", 200, "6007") - checkRPCStatus(t, "jobs", 200, "6007") - }) - - stopCh <- struct{}{} - wg.Wait() - - t.Cleanup(func() { - sp.StopHTTPServer() - }) -} - -func TestShutdown503(t *testing.T) { - cont := endure.New(slog.LevelDebug, endure.GracefulShutdownTimeout(time.Second*10)) - - cfg := &config.Plugin{ - Version: "2024.1.0", - Timeout: time.Second * 10, - Path: "configs/.rr-status-503.yaml", - } - - sp := &status.Plugin{} - err := cont.RegisterAll( - cfg, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - sp, - &jobs.Plugin{}, - ) - 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) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - 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 <-stopCh: - // timeout, error here is OK, because in the PHP we are sleeping for the 300s - _ = cont.Stop() - return - } - } - }) - - time.Sleep(time.Second) - go func() { - httpClient := &http.Client{ - Timeout: time.Second * 10, - } - - req, err2 := http.NewRequestWithContext(context.Background(), http.MethodGet, "http://127.0.0.1:11934", nil) - assert.NoError(t, err2) - rsp, _ := httpClient.Do(req) - if rsp != nil { - _ = rsp.Body.Close() - } - }() - - time.Sleep(time.Second) - stopCh <- struct{}{} - - httpClient := &http.Client{ - Timeout: time.Second * 10, - } - - req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, "http://127.0.0.1:34711/health", nil) - assert.NoError(t, err) - require.NotNil(t, req) - - rsp, err := httpClient.Do(req) - require.NoError(t, err) - - assert.Equal(t, http.StatusOK, rsp.StatusCode) - - req, err = http.NewRequestWithContext(context.Background(), http.MethodGet, "http://127.0.0.1:34711/ready", nil) - assert.NoError(t, err) - require.NotNil(t, req) - - rsp, err = httpClient.Do(req) - require.NoError(t, err) - require.NotNil(t, rsp) - - assert.Equal(t, http.StatusServiceUnavailable, rsp.StatusCode) - _ = rsp.Body.Close() - - // Also verify /jobs returns 503 during shutdown - req, err = http.NewRequestWithContext(context.Background(), http.MethodGet, "http://127.0.0.1:34711/jobs", nil) - assert.NoError(t, err) - require.NotNil(t, req) - - rsp, err = httpClient.Do(req) - require.NoError(t, err) - - b, err := io.ReadAll(rsp.Body) - assert.NoError(t, err) - assert.Equal(t, http.StatusServiceUnavailable, rsp.StatusCode) - assert.Contains(t, string(b), "service is shutting down") - - wg.Wait() - - t.Cleanup(func() { - if rsp != nil { - _ = rsp.Body.Close() - } - - sp.StopHTTPServer() - }) -} - -func TestRPCNonExistentPlugin(t *testing.T) { - cont := endure.New(slog.LevelDebug, endure.GracefulShutdownTimeout(time.Minute)) - - cfg := &config.Plugin{ - Version: "2024.1.0", - Path: "configs/.rr-status-init.yaml", - } - - // Minimal stack: this test only verifies the "no such plugin" path of the - // status RPC handler, so no Checker/Readiness providers are needed and - // http/server are dropped to keep the test PHP-independent. - sp := &status.Plugin{} - err := cont.RegisterAll( - cfg, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - sp, - ) - 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) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - 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 <-stopCh: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second) - - t.Run("StatusNonExistent", func(t *testing.T) { - client := newStatusRPCClient(t, "127.0.0.1:6005") - err := client.Call("status.Status", &statusV2.StatusRequest{Plugin: "nonexistent"}, &statusV2.StatusResponse{}) - require.Error(t, err) - assert.Contains(t, err.Error(), "no such plugin") - }) - - t.Run("ReadyNonExistent", func(t *testing.T) { - client := newStatusRPCClient(t, "127.0.0.1:6005") - err := client.Call("status.Ready", &statusV2.StatusRequest{Plugin: "nonexistent"}, &statusV2.StatusResponse{}) - require.Error(t, err) - assert.Contains(t, err.Error(), "no such plugin") - }) - - stopCh <- struct{}{} - wg.Wait() - - t.Cleanup(func() { - sp.StopHTTPServer() - }) -} - -func checkJobsReadiness(t *testing.T) { - req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, "http://127.0.0.1:35544/ready?plugin=jobs", nil) - assert.NoError(t, err) - - r, err := http.DefaultClient.Do(req) - assert.NoError(t, err) - b, err := io.ReadAll(r.Body) - assert.NoError(t, err) - - rep := make([]*status.Report, 0, 2) - err = json.Unmarshal(b, &rep) - require.NoError(t, err) - - assert.Equal(t, 200, r.StatusCode) - - assert.Len(t, rep, 1) - assert.Equal(t, "jobs", rep[0].PluginName) - assert.Equal(t, "", rep[0].ErrorMessage) - assert.Equal(t, 200, rep[0].StatusCode) - - err = r.Body.Close() - assert.NoError(t, err) -} - -func checkHTTPStatus(t *testing.T) { - req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, "http://127.0.0.1:34333/health?plugin=http&plugin=rpc", nil) - assert.NoError(t, err) - - r, err := http.DefaultClient.Do(req) - assert.NoError(t, err) - b, err := io.ReadAll(r.Body) - assert.NoError(t, err) - assert.Equal(t, 200, r.StatusCode) - - rep := make([]*status.Report, 0, 2) - err = json.Unmarshal(b, &rep) - require.NoError(t, err) - - assert.Len(t, rep, 1) - assert.Equal(t, "http", rep[0].PluginName) - assert.Equal(t, "", rep[0].ErrorMessage) - assert.Equal(t, 200, rep[0].StatusCode) - - err = r.Body.Close() - assert.NoError(t, err) -} - -func checkHTTPStatusAll(t *testing.T) { - req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, "http://127.0.0.1:34333/health", nil) - assert.NoError(t, err) - - r, err := http.DefaultClient.Do(req) - assert.NoError(t, err) - b, err := io.ReadAll(r.Body) - assert.NoError(t, err) - assert.Equal(t, 200, r.StatusCode) - - rep := make([]*status.Report, 0, 2) - err = json.Unmarshal(b, &rep) - require.NoError(t, err) - - assert.Len(t, rep, 1) - assert.Equal(t, "http", rep[0].PluginName) - assert.Equal(t, "", rep[0].ErrorMessage) - assert.Equal(t, 200, rep[0].StatusCode) - - err = r.Body.Close() - assert.NoError(t, err) -} - -func doHTTPReq(t *testing.T) { - go func() { - client := &http.Client{ - Timeout: time.Second * 10, - } - req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, "http://127.0.0.1:11933", nil) - assert.NoError(t, err) - - _, err = client.Do(req) //nolint:bodyclose - assert.Error(t, err) - }() -} - -func checkHTTPReadiness2(t *testing.T) { - req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, "http://127.0.0.1:34334/ready?plugin=http&plugin=rpc", nil) - assert.NoError(t, err) - - r, err := http.DefaultClient.Do(req) - assert.NoError(t, err) - b, err := io.ReadAll(r.Body) - require.NoError(t, err) - - res := make([]*status.Report, 0, 2) - err = json.Unmarshal(b, &res) - require.NoError(t, err) - assert.Len(t, res, 1) - assert.Equal(t, "http", res[0].PluginName) - assert.Equal(t, "internal server error, see logs", res[0].ErrorMessage) - assert.Equal(t, 503, res[0].StatusCode) - assert.Equal(t, 503, r.StatusCode) - - err = r.Body.Close() - assert.NoError(t, err) -} - -func checkHTTPReadinessAll(t *testing.T) { - req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, "http://127.0.0.1:34333/ready", nil) - assert.NoError(t, err) - - r, err := http.DefaultClient.Do(req) - assert.NoError(t, err) - b, err := io.ReadAll(r.Body) - assert.NoError(t, err) - assert.Equal(t, 200, r.StatusCode) - res := make([]*status.Report, 0, 2) - err = json.Unmarshal(b, &res) - assert.NoError(t, err) - - err = r.Body.Close() - assert.NoError(t, err) -} - -func checkHTTPReadiness(t *testing.T) { - req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, "http://127.0.0.1:34333/ready?plugin=http&plugin=rpc", nil) - assert.NoError(t, err) - - r, err := http.DefaultClient.Do(req) - assert.NoError(t, err) - b, err := io.ReadAll(r.Body) - assert.NoError(t, err) - assert.Equal(t, 200, r.StatusCode) - res := make([]*status.Report, 0, 2) - err = json.Unmarshal(b, &res) - assert.NoError(t, err) - - err = r.Body.Close() - assert.NoError(t, err) -} - -func checkRPCReadiness(t *testing.T, plugin string, code int64, port string) { - client := newStatusRPCClient(t, "127.0.0.1:"+port) - rsp := &statusV2.StatusResponse{} - err := client.Call("status.Ready", &statusV2.StatusRequest{Plugin: plugin}, rsp) - require.NoError(t, err) - assert.Equal(t, code, rsp.GetCode()) -} - -func checkRPCStatus(t *testing.T, plugin string, code int64, port string) { - client := newStatusRPCClient(t, "127.0.0.1:"+port) - rsp := &statusV2.StatusResponse{} - err := client.Call("status.Status", &statusV2.StatusRequest{Plugin: plugin}, rsp) - require.NoError(t, err) - assert.Equal(t, code, rsp.GetCode()) -} diff --git a/tests/rpc_test.go b/tests/rpc_test.go new file mode 100644 index 0000000..b572122 --- /dev/null +++ b/tests/rpc_test.go @@ -0,0 +1,51 @@ +package tests + +import ( + "net/http" + "testing" + + "tests/helpers" + + statusV1 "github.com/roadrunner-server/api-go/v6/status/v1" + httpPlugin "github.com/roadrunner-server/http/v6" + rpcPlugin "github.com/roadrunner-server/rpc/v6" + "github.com/roadrunner-server/server/v6" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestStatusRPC asks the status rpc service for the state of a running plugin. +func TestStatusRPC(t *testing.T) { + helpers.Start(t, statusInitCfg, []any{ + &rpcPlugin.Plugin{}, + &server.Plugin{}, + &httpPlugin.Plugin{}, + helpers.NewStatusPlugin(t), + }, helpers.WithTCPProbe(statusInitAddr)) + + helpers.WaitListener(t, "tcp", sharedHTTPAddr) + + rsp := &statusV1.Response{} + require.NoError(t, helpers.RPC(t, statusInitRPC).Call("status.Status", &statusV1.Request{Plugin: "http"}, rsp)) + assert.Equal(t, int64(http.StatusOK), rsp.GetCode()) +} + +// TestRPCUnknownPlugin runs the minimal stack the rpc service needs: no +// Checker or Readiness provider is registered, so every name is unknown and the +// test needs no PHP worker. +func TestRPCUnknownPlugin(t *testing.T) { + helpers.Start(t, statusInitCfg, []any{ + &rpcPlugin.Plugin{}, + helpers.NewStatusPlugin(t), + }, helpers.WithTCPProbe(statusInitAddr)) + + client := helpers.RPC(t, statusInitRPC) + + for _, method := range []string{"status.Status", "status.Ready"} { + t.Run(method, func(t *testing.T) { + err := client.Call(method, &statusV1.Request{Plugin: "nonexistent"}, &statusV1.Response{}) + require.Error(t, err) + assert.Contains(t, err.Error(), "no such plugin") + }) + } +} diff --git a/tests/shutdown_test.go b/tests/shutdown_test.go new file mode 100644 index 0000000..99d2ecd --- /dev/null +++ b/tests/shutdown_test.go @@ -0,0 +1,53 @@ +package tests + +import ( + "net/http" + "testing" + "time" + + "tests/helpers" + + "github.com/roadrunner-server/jobs/v6" + rpcPlugin "github.com/roadrunner-server/rpc/v6" + "github.com/roadrunner-server/server/v6" + "github.com/stretchr/testify/assert" +) + +const ( + shutdownCfg = "configs/.rr-status-503.yaml" + shutdownAddr = "127.0.0.1:34711" + shutdownURL = "http://" + shutdownAddr + shutdownGrace = time.Second * 10 +) + +// TestShutdown503 checks the endpoints of a stopped container: Plugin.Stop only +// raises the shutdown flag, so the status listener still answers while the rest +// of the container drains. +func TestShutdown503(t *testing.T) { + stop := helpers.Start(t, shutdownCfg, []any{ + &rpcPlugin.Plugin{}, + &server.Plugin{}, + &jobs.Plugin{}, + helpers.NewStatusPlugin(t), + }, + helpers.WithTCPProbe(shutdownAddr), + helpers.WithGracefulTimeout(shutdownGrace), + helpers.WithConfigTimeout(shutdownGrace), + ) + + // returns once the container has been stopped + stop() + + // liveness stays 200 so the orchestrator does not kill the draining process + code, body := helpers.GetBody(t, shutdownURL+"/health") + assert.Equal(t, http.StatusOK, code) + assert.Contains(t, body, "service is shutting down") + + code, body = helpers.GetBody(t, shutdownURL+"/ready") + assert.Equal(t, http.StatusServiceUnavailable, code) + assert.Contains(t, body, "service is shutting down") + + code, body = helpers.GetBody(t, shutdownURL+"/jobs") + assert.Equal(t, http.StatusServiceUnavailable, code) + assert.Contains(t, body, "service is shutting down") +}