diff --git a/.github/workflows/linux.yml b/.github/workflows/linux.yml index 3f16b49..b191580 100644 --- a/.github/workflows/linux.yml +++ b/.github/workflows/linux.yml @@ -63,21 +63,21 @@ jobs: echo fs.inotify.max_user_watches=524288 | sudo tee -a /etc/sysctl.conf && sudo sysctl -p ulimit -n 5000 + - name: Create folders + run: | + mkdir ./tests/coverage-ci + - name: Install Go dependencies run: go mod download - name: Run service unit tests with coverage run: | - mkdir ./tests/coverage-ci - go test -timeout 20m -v -race -cover -tags=debug -failfast -coverpkg=github.com/roadrunner-server/service/v6 -coverprofile=./tests/coverage-ci/service_u.out -covermode=atomic . + go test -timeout 20m -v -race -cover -tags=debug -coverpkg=./... -coverprofile=./tests/coverage-ci/service_u.out -covermode=atomic ./... - name: Run service e2e tests with coverage run: | cd tests - - chmod 777 php_test_files/test_binary - chmod +x php_test_files/test_binary - go test -timeout 20m -v -race -cover -tags=debug -failfast -coverpkg=github.com/roadrunner-server/service/v6 -coverprofile=./coverage-ci/service.out -covermode=atomic . + go test -timeout 20m -v -race -cover -tags=debug -coverpkg=github.com/roadrunner-server/service/v6/... -coverprofile=./coverage-ci/service.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..b5b3c2b --- /dev/null +++ b/config_test.go @@ -0,0 +1,37 @@ +package service + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestConfigInitDefault(t *testing.T) { + cfg := &Config{Services: map[string]*Service{ + "unset": {}, + "negative": {ProcessNum: -3}, + "explicit": {ProcessNum: 4, RestartSec: 7, TimeoutStopSec: 9}, + }} + + cfg.InitDefault() + + tests := []struct { + name string + processNum int + restartSec uint64 + timeoutStopSec uint64 + }{ + {name: "unset", processNum: 1, restartSec: 30, timeoutStopSec: 5}, + {name: "negative", processNum: 1, restartSec: 30, timeoutStopSec: 5}, + {name: "explicit", processNum: 4, restartSec: 7, timeoutStopSec: 9}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + svc := cfg.Services[tt.name] + require.Equal(t, tt.processNum, svc.ProcessNum) + require.Equal(t, tt.restartSec, svc.RestartSec) + require.Equal(t, tt.timeoutStopSec, svc.TimeoutStopSec) + }) + } +} diff --git a/go.mod b/go.mod index 3ef360e..5b8aaa2 100644 --- a/go.mod +++ b/go.mod @@ -5,20 +5,25 @@ 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/errors v1.5.0 github.com/roadrunner-server/pool/v2 v2.0.0-beta.1 github.com/shirou/gopsutil v3.21.11+incompatible + github.com/stretchr/testify v1.11.1 ) require ( github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/go-ole/go-ole v1.3.0 // indirect + github.com/kr/pretty v0.3.1 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/roadrunner-server/goridge/v4 v4.0.0-beta.3 // indirect + github.com/rogpeppe/go-internal v1.14.1 // indirect github.com/tklauser/go-sysconf v0.4.0 // indirect github.com/tklauser/numcpus v0.12.0 // indirect github.com/yusufpapurcu/wmi v1.2.4 // indirect golang.org/x/sys v0.47.0 // indirect google.golang.org/protobuf v1.36.12 // indirect + gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index b47739d..a00f80f 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,4 @@ +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0= @@ -5,16 +6,27 @@ github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE= github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +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/errors v1.5.0 h1:unG7LKIZrSzkCCF3YLRLA5VyqE0KKomofXVJUXJe00g= github.com/roadrunner-server/errors v1.5.0/go.mod h1:g9fo/T2C13cWRDR9PW1r0ZAOSQfNhWAZawyfkGiaHuI= 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/pool/v2 v2.0.0-beta.1 h1:jpYXFtdD6QGAdAGPgMxrNi3j1CegCRpb2y+A+3GnXFA= github.com/roadrunner-server/pool/v2 v2.0.0-beta.1/go.mod h1:Bo1wT7RtL3eyQHXBUohNhtj/yAmRt6Rq8smuBg5pWkY= +github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= +github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/shirou/gopsutil v3.21.11+incompatible h1:+1+c1VGhc88SSonWP6foOcLhvnKlUeu/erjjvaPEYiI= github.com/shirou/gopsutil v3.21.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= @@ -31,5 +43,8 @@ golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/logger_test.go b/logger_test.go new file mode 100644 index 0000000..84286fc --- /dev/null +++ b/logger_test.go @@ -0,0 +1,108 @@ +package service + +import ( + "context" + "log/slog" + "maps" + "sync" +) + +// capturedRecord is a log record with the attributes carried by its handler. +type capturedRecord struct { + message string + attrs map[string]string +} + +// captureStore collects the records written through a captureHandler. +type captureStore struct { + mu sync.Mutex + records []capturedRecord +} + +func (s *captureStore) add(r capturedRecord) { + s.mu.Lock() + s.records = append(s.records, r) + s.mu.Unlock() +} + +// messages returns the messages in the order they were written. +func (s *captureStore) messages() []string { + s.mu.Lock() + defer s.mu.Unlock() + + out := make([]string, 0, len(s.records)) + for _, r := range s.records { + out = append(out, r.message) + } + + return out +} + +// count returns how many records carry the given message. +func (s *captureStore) count(message string) int { + s.mu.Lock() + defer s.mu.Unlock() + + var n int + for _, r := range s.records { + if r.message == message { + n++ + } + } + + return n +} + +// find returns the first record with the given message. +func (s *captureStore) find(message string) (capturedRecord, bool) { + s.mu.Lock() + defer s.mu.Unlock() + + for _, r := range s.records { + if r.message == message { + return r, true + } + } + + return capturedRecord{}, false +} + +// captureHandler is an slog.Handler that keeps every record in memory. +type captureHandler struct { + store *captureStore + attrs map[string]string +} + +// newCaptureLogger returns a logger writing into the returned store. +func newCaptureLogger() (*slog.Logger, *captureStore) { + store := &captureStore{} + return slog.New(&captureHandler{store: store, attrs: map[string]string{}}), store +} + +func (h *captureHandler) Enabled(context.Context, slog.Level) bool { + return true +} + +func (h *captureHandler) Handle(_ context.Context, r slog.Record) error { + rec := capturedRecord{message: r.Message, attrs: maps.Clone(h.attrs)} + r.Attrs(func(a slog.Attr) bool { + rec.attrs[a.Key] = a.Value.String() + return true + }) + + h.store.add(rec) + return nil +} + +func (h *captureHandler) WithAttrs(attrs []slog.Attr) slog.Handler { + next := maps.Clone(h.attrs) + for _, a := range attrs { + next[a.Key] = a.Value.String() + } + + return &captureHandler{store: h.store, attrs: next} +} + +func (h *captureHandler) WithGroup(string) slog.Handler { + return h +} diff --git a/plugin_test.go b/plugin_test.go new file mode 100644 index 0000000..1f9f373 --- /dev/null +++ b/plugin_test.go @@ -0,0 +1,196 @@ +package service + +import ( + "context" + stderr "errors" + "fmt" + "log/slog" + "path/filepath" + "testing" + "time" + + "github.com/roadrunner-server/errors" + "github.com/roadrunner-server/pool/v2/state/process" + "github.com/stretchr/testify/require" +) + +// testConfigurer feeds Init a service section without going through viper. +type testConfigurer struct { + has bool + services map[string]*Service + err error +} + +func (c *testConfigurer) Has(string) bool { return c.has } + +func (c *testConfigurer) UnmarshalKey(_ string, out any) error { + if c.err != nil { + return c.err + } + + dst, ok := out.(*map[string]*Service) + if !ok { + return fmt.Errorf("unexpected destination type %T", out) + } + + *dst = c.services + return nil +} + +type testLogger struct { + l *slog.Logger +} + +func (l *testLogger) NamedLogger(string) *slog.Logger { return l.l } + +func TestPluginInitDisabled(t *testing.T) { + log, _ := newCaptureLogger() + p := &Plugin{} + + err := p.Init(&testConfigurer{has: false}, &testLogger{l: log}) + + require.Error(t, err) + require.True(t, errors.Is(errors.Disabled, err)) +} + +func TestPluginInitUnmarshalError(t *testing.T) { + log, _ := newCaptureLogger() + p := &Plugin{} + + err := p.Init(&testConfigurer{has: true, err: stderr.New("broken section")}, &testLogger{l: log}) + + require.ErrorContains(t, err, "service_plugin_init") + require.ErrorContains(t, err, "broken section") +} + +func TestPluginInitAppliesDefaults(t *testing.T) { + log, _ := newCaptureLogger() + p := &Plugin{} + + cfg := &testConfigurer{has: true, services: map[string]*Service{ + "some_service": {Command: "sleep 30"}, + }} + require.NoError(t, p.Init(cfg, &testLogger{l: log})) + + require.Equal(t, 1, p.cfg.Services["some_service"].ProcessNum) + require.EqualValues(t, 30, p.cfg.Services["some_service"].RestartSec) + require.EqualValues(t, 5, p.cfg.Services["some_service"].TimeoutStopSec) +} + +func TestPluginGetters(t *testing.T) { + p := &Plugin{} + + require.Equal(t, PluginName, p.Name()) + require.EqualValues(t, 10, p.Weight()) + require.IsType(t, &rpc{}, p.RPC()) +} + +func TestPluginServeAndStop(t *testing.T) { + p, store := newTestPlugin(t, map[string]*Service{ + "some_service": {Command: "sleep 30", ProcessNum: 2}, + }) + + errCh := p.Serve() + requireWorkers(t, p, 2) + require.Empty(t, errCh) + + for _, st := range p.Workers() { + require.NotZero(t, st.Pid) + require.Contains(t, st.Command, "sleep") + require.NotZero(t, st.MemoryUsage) + } + + require.NoError(t, p.Stop(context.Background())) + + require.Equal(t, 2, store.count("service was started")) + require.Equal(t, 2, store.count("service was stopped")) + require.Empty(t, p.Workers()) +} + +func TestPluginServeReportsStartError(t *testing.T) { + p, store := newTestPlugin(t, map[string]*Service{ + "some_service": {Command: filepath.Join(t.TempDir(), "no-such-binary"), ProcessNum: 1}, + }) + + errCh := p.Serve() + + select { + case err := <-errCh: + require.Error(t, err) + case <-time.After(time.Second * 10): + require.Fail(t, "the start error was not pushed to the error channel") + } + + // the process was stored before it was started, but it has no pid + require.Empty(t, p.Workers()) + require.Equal(t, 1, store.count("get process state")) +} + +func TestPluginResetReplacesProcesses(t *testing.T) { + p, _ := newTestPlugin(t, map[string]*Service{ + "some_service": {Command: "sleep 30", ProcessNum: 2}, + }) + + p.Serve() + requireWorkers(t, p, 2) + before := workerPids(p.Workers()) + + require.NoError(t, p.Reset()) + + after := workerPids(p.Workers()) + require.Len(t, after, 2) + for _, pid := range after { + require.NotContains(t, before, pid) + } +} + +func TestPluginResetLogsFailedRestart(t *testing.T) { + p, store := newTestPlugin(t, map[string]*Service{ + "some_service": {Command: "sleep 30", ProcessNum: 1}, + }) + + p.Serve() + requireWorkers(t, p, 1) + + // the replacement is built from the stored service, which now points at + // something that cannot be executed + v, ok := p.processes.Load("some_service") + require.True(t, ok) + v.([]*Process)[0].service.Command = filepath.Join(t.TempDir(), "no-such-binary") + + require.NoError(t, p.Reset()) + + require.Equal(t, 1, store.count("unable to start the service")) + require.Empty(t, p.Workers()) +} + +// newTestPlugin returns a plugin wired to an in-memory logger, stopped at the +// end of the test. +func newTestPlugin(t *testing.T, services map[string]*Service) (*Plugin, *captureStore) { + t.Helper() + + log, store := newCaptureLogger() + p := &Plugin{logger: log, cfg: Config{Services: services}} + t.Cleanup(func() { _ = p.Stop(context.Background()) }) + + return p, store +} + +// requireWorkers waits until the plugin reports n live processes. +func requireWorkers(t *testing.T, p *Plugin, n int) { + t.Helper() + + require.Eventually(t, func() bool { + return len(p.Workers()) == n + }, time.Second*10, time.Millisecond*20, "the plugin did not report %d processes", n) +} + +// workerPids collects the pids the plugin reports. +func workerPids(states []*process.State) []int64 { + pids := make([]int64, 0, len(states)) + for _, st := range states { + pids = append(pids, st.Pid) + } + + return pids +} diff --git a/process_test.go b/process_test.go index bc924f0..f91ec9e 100644 --- a/process_test.go +++ b/process_test.go @@ -1,24 +1,242 @@ package service import ( + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" "testing" + "time" + + "github.com/shirou/gopsutil/process" + "github.com/stretchr/testify/require" ) func TestSetEnv(t *testing.T) { - e := make(Env, 5) - e["foo"] = "bar" - e["bar"] = "baz" - p := &Process{} - out := p.setEnv(e) - val := out[len(out)-1] - val2 := out[len(out)-2] + out := p.setEnv(Env{"foo": "bar", "bar": "baz"}) + + require.Len(t, out, len(os.Environ())+2) + require.Subset(t, out, []string{"FOO=bar", "BAR=baz"}) +} + +func TestCreateProcess(t *testing.T) { + tests := []struct { + name string + command string + args []string + }{ + {name: "single token", command: "sleep", args: []string{"sleep"}}, + {name: "command with arguments", command: "sleep 30", args: []string{"sleep", "30"}}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + p := &Process{service: &Service{Command: tt.command}} + p.createProcess(strings.Split(tt.command, " ")) + + require.Equal(t, tt.args, p.command.Args) + require.Equal(t, "sleep", filepath.Base(p.command.Path)) + require.Nil(t, p.cancel) + }) + } +} + +func TestCreateProcessCtx(t *testing.T) { + tests := []struct { + name string + command string + args []string + }{ + {name: "single token", command: "sleep", args: []string{"sleep"}}, + {name: "command with arguments", command: "sleep 30", args: []string{"sleep", "30"}}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + p := &Process{service: &Service{Command: tt.command, ExecTimeout: time.Second}} + p.createProcessCtx(strings.Split(tt.command, " ")) + t.Cleanup(p.cancel) + + require.Equal(t, tt.args, p.command.Args) + require.Equal(t, "sleep", filepath.Base(p.command.Path)) + require.NotNil(t, p.cancel) + }) + } +} + +func TestConfigureUser(t *testing.T) { + t.Run("unset", func(t *testing.T) { + p := &Process{service: &Service{}} + require.NoError(t, p.configureUser()) + }) + + t.Run("unknown user", func(t *testing.T) { + p := &Process{ + service: &Service{User: "roadrunner-user-that-cannot-exist"}, + command: exec.CommandContext(t.Context(), "sleep", "30"), + } + + require.Error(t, p.configureUser()) + }) +} + +func TestProcessWriteTrimsOutput(t *testing.T) { + log, store := newCaptureLogger() + p := &Process{log: log} + + payload := []byte("service output \n\t") + n, err := p.Write(payload) + + require.NoError(t, err) + require.Len(t, payload, n) + require.Equal(t, []string{"service output"}, store.messages()) +} + +func TestNewServiceProcessDefaults(t *testing.T) { + log, store := newCaptureLogger() + + svc := &Service{} + p := NewServiceProcess(svc, "some_service", log) + + require.EqualValues(t, 30, svc.RestartSec) + require.EqualValues(t, 5, svc.TimeoutStopSec) + + p.log.Info("output") + rec, ok := store.find("output") + require.True(t, ok) + require.NotContains(t, rec.attrs, "service") +} + +func TestNewServiceProcessServiceNameInLog(t *testing.T) { + log, store := newCaptureLogger() + + svc := &Service{UseServiceName: true, RestartSec: 3, TimeoutStopSec: 7} + p := NewServiceProcess(svc, "some_service", log) + + require.EqualValues(t, 3, svc.RestartSec) + require.EqualValues(t, 7, svc.TimeoutStopSec) - if val != "BAR=baz" && val != "FOO=bar" { - t.Fail() + p.log.Info("output") + rec, ok := store.find("output") + require.True(t, ok) + require.Equal(t, "some_service", rec.attrs["service"]) +} + +func TestProcessStopWithoutStart(t *testing.T) { + log, _ := newCaptureLogger() + p := NewServiceProcess(&Service{Command: "sleep 30", TimeoutStopSec: 30}, "some_service", log) + + started := time.Now() + p.stop() + + // there is no child to signal, so stop returns instead of arming the kill timer + require.Less(t, time.Since(started), time.Second) +} + +func TestProcessStartConfigureUserError(t *testing.T) { + log, _ := newCaptureLogger() + p := NewServiceProcess(&Service{Command: "sleep 30", User: "roadrunner-user-that-cannot-exist"}, "some_service", log) + + require.Error(t, p.start()) + require.Zero(t, p.pid) +} + +func TestProcessStopKillsUnresponsiveChild(t *testing.T) { + tests := []struct { + name string + execTimeout time.Duration + }{ + {name: "without exec timeout", execTimeout: 0}, + {name: "with exec timeout", execTimeout: time.Minute}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + script := writeScript(t, "ignore-sigint.sh", "#!/bin/sh\ntrap '' INT\necho ready\nsleep 5\n") + + log, store := newCaptureLogger() + p := NewServiceProcess(&Service{ + Command: script, + ExecTimeout: tt.execTimeout, + TimeoutStopSec: 1, + }, "some_service", log) + + require.NoError(t, p.start()) + pid := p.pid + require.NotZero(t, pid) + + // the marker is written once the child ignores SIGINT + require.Eventually(t, func() bool { return store.count("ready") == 1 }, + time.Second*10, time.Millisecond*20) + + started := time.Now() + p.stop() + + // SIGINT is ignored, so stop falls through to the timer and kills the child + require.GreaterOrEqual(t, time.Since(started), time.Second) + require.Eventually(t, func() bool { return !processAlive(pid) }, + time.Second*10, time.Millisecond*20) + }) } +} + +func TestProcessRestartsAfterExit(t *testing.T) { + script := writeScript(t, "exit-at-once.sh", "#!/bin/sh\necho ready\n") + + log, store := newCaptureLogger() + p := NewServiceProcess(&Service{ + Command: script, + RemainAfterExit: true, + RestartSec: 1, + TimeoutStopSec: 1, + }, "some_service", log) + + require.NoError(t, p.start()) + t.Cleanup(p.stop) + + // the child exits at once and is started again restart_sec later + require.Eventually(t, func() bool { return store.count("ready") >= 2 }, + time.Second*15, time.Millisecond*20) +} + +func TestProcessRestartError(t *testing.T) { + script := writeScript(t, "delete-itself.sh", "#!/bin/sh\necho ready\nrm -- \"$0\"\n") + + log, store := newCaptureLogger() + p := NewServiceProcess(&Service{ + Command: script, + RemainAfterExit: true, + RestartSec: 1, + TimeoutStopSec: 1, + }, "some_service", log) - if val2 != "BAR=baz" && val2 != "FOO=bar" { - t.Fail() + require.NoError(t, p.start()) + t.Cleanup(p.stop) + + // the child removes its own command, so the restart cannot execute it + require.Eventually(t, func() bool { return store.count("process start error") == 1 }, + time.Second*15, time.Millisecond*20) +} + +// writeScript drops an executable posix shell script into the test temp dir. +func writeScript(t *testing.T, name, body string) string { + t.Helper() + + if runtime.GOOS == "windows" { + t.Skip("the fixture is a posix shell script") } + + path := filepath.Join(t.TempDir(), name) + require.NoError(t, os.WriteFile(path, []byte(body), 0o600)) + require.NoError(t, os.Chmod(path, 0o700)) + + return path +} + +// processAlive reports whether the pid still refers to a live process. +func processAlive(pid int64) bool { + alive, err := process.PidExists(int32(pid)) //nolint:gosec + return err == nil && alive } diff --git a/rpc.go b/rpc.go index 01ab46d..325d358 100644 --- a/rpc.go +++ b/rpc.go @@ -28,7 +28,7 @@ func (r *rpc) loadProcesses(name string) ([]*Process, error) { return v.([]*Process), nil } -func (r *rpc) CreateService(in *serviceV1.Create, out *serviceV1.Response) error { +func (r *rpc) Create(in *serviceV1.Create, out *serviceV1.Response) error { r.p.logger.Debug("create service", "name", in.GetName(), "restart_sec", in.GetRestartSec(), "command", in.GetCommand(), "process number", in.GetProcessNum()) if in.GetProcessNum() == 0 { @@ -133,8 +133,8 @@ func (r *rpc) Restart(in *serviceV1.Service, out *serviceV1.Response) error { return nil } -// Deprecated: use GetStatuses to get correct info. -func (r *rpc) GetStatus(in *serviceV1.Service, out *serviceV1.Status) error { +// Deprecated: use Statuses to get correct info. +func (r *rpc) Status(in *serviceV1.Service, out *serviceV1.Status) error { r.p.logger.Debug("service status", "name", in.GetName()) r.mu.RLock() @@ -160,7 +160,7 @@ func (r *rpc) GetStatus(in *serviceV1.Service, out *serviceV1.Status) error { return nil } -func (r *rpc) GetStatuses(in *serviceV1.Service, out *serviceV1.Statuses) error { +func (r *rpc) Statuses(in *serviceV1.Service, out *serviceV1.Statuses) error { r.p.logger.Debug("service status", "name", in.GetName()) r.mu.RLock() @@ -200,7 +200,7 @@ func (r *rpc) GetStatuses(in *serviceV1.Service, out *serviceV1.Statuses) error return nil } -func (r *rpc) ListServices(_ *serviceV1.Service, out *serviceV1.List) error { +func (r *rpc) List(_ *serviceV1.Service, out *serviceV1.List) error { r.mu.RLock() defer r.mu.RUnlock() diff --git a/rpc_test.go b/rpc_test.go new file mode 100644 index 0000000..ed4217a --- /dev/null +++ b/rpc_test.go @@ -0,0 +1,234 @@ +package service + +import ( + "context" + "path/filepath" + "testing" + "time" + + serviceV1 "github.com/roadrunner-server/api-go/v6/service/v1" + "github.com/stretchr/testify/require" +) + +const ( + testServiceName = "some_service" + otherServiceName = "other_service" +) + +func TestRPCCreateRejectsEmptyGroup(t *testing.T) { + r := newTestRPC(t) + + out := &serviceV1.Response{} + err := r.Create(&serviceV1.Create{Name: testServiceName, Command: "sleep 30"}, out) + + require.ErrorContains(t, err, "at least 1 process") + require.False(t, out.GetOk()) + + _, stored := r.p.processes.Load(testServiceName) + require.False(t, stored) +} + +func TestRPCCreateTwice(t *testing.T) { + r := newTestRPC(t) + + in := &serviceV1.Create{ + Name: testServiceName, + Command: "sleep 30", + ProcessNum: 1, + Env: map[string]string{"foo": "bar"}, + RestartSec: 1, + TimeoutStopSec: 1, + RemainAfterExit: false, + } + require.NoError(t, r.Create(in, &serviceV1.Response{})) + running := rpcPids(t, r) + + err := r.Create(in, &serviceV1.Response{}) + + require.ErrorIs(t, err, errServiceExists) + // the group that is already running is left alone + require.Equal(t, running, rpcPids(t, r)) +} + +func TestRPCCreateBrokenCommand(t *testing.T) { + r := newTestRPC(t) + + out := &serviceV1.Response{} + err := r.Create(&serviceV1.Create{ + Name: testServiceName, + Command: filepath.Join(t.TempDir(), "no-such-binary"), + ProcessNum: 2, + }, out) + + require.Error(t, err) + require.False(t, out.GetOk()) + + _, stored := r.p.processes.Load(testServiceName) + require.False(t, stored) +} + +func TestRPCUnknownService(t *testing.T) { + r := newTestRPC(t) + in := &serviceV1.Service{Name: testServiceName} + + require.ErrorIs(t, r.Terminate(in, &serviceV1.Response{}), errNoSuchService) + require.ErrorIs(t, r.Restart(in, &serviceV1.Response{}), errNoSuchService) + require.ErrorIs(t, r.Status(in, &serviceV1.Status{}), errNoSuchService) + require.ErrorIs(t, r.Statuses(in, &serviceV1.Statuses{}), errNoSuchService) +} + +func TestRPCRestartReplacesProcesses(t *testing.T) { + r := newTestRPC(t) + require.NoError(t, r.Create(newCreate(testServiceName, 2), &serviceV1.Response{})) + + before := rpcPids(t, r) + require.Len(t, before, 2) + + out := &serviceV1.Response{} + require.NoError(t, r.Restart(&serviceV1.Service{Name: testServiceName}, out)) + require.True(t, out.GetOk()) + + after := rpcPids(t, r) + require.Len(t, after, 2) + for _, pid := range after { + require.NotContains(t, before, pid) + } + + for _, pid := range before { + require.Eventually(t, func() bool { return !processAlive(pid) }, time.Second*10, time.Millisecond*20) + } +} + +func TestRPCRestartRollsBackBrokenReplacement(t *testing.T) { + r := newTestRPC(t) + require.NoError(t, r.Create(newCreate(testServiceName, 2), &serviceV1.Response{})) + + // the replacements are built from the stored services, the second of which + // now points at something that cannot be executed + procs := loadProcs(t, r) + procs[1].service.Command = filepath.Join(t.TempDir(), "no-such-binary") + + require.Error(t, r.Restart(&serviceV1.Service{Name: testServiceName}, &serviceV1.Response{})) + + _, stored := r.p.processes.Load(testServiceName) + require.False(t, stored) + + for i := range procs { + require.Eventually(t, func() bool { return !processAlive(procs[i].pid) }, + time.Second*10, time.Millisecond*20) + } +} + +func TestRPCTerminateStopsProcesses(t *testing.T) { + r := newTestRPC(t) + require.NoError(t, r.Create(newCreate(testServiceName, 2), &serviceV1.Response{})) + procs := loadProcs(t, r) + + out := &serviceV1.Response{} + require.NoError(t, r.Terminate(&serviceV1.Service{Name: testServiceName}, out)) + require.True(t, out.GetOk()) + + _, stored := r.p.processes.Load(testServiceName) + require.False(t, stored) + + for i := range procs { + require.Eventually(t, func() bool { return !processAlive(procs[i].pid) }, + time.Second*10, time.Millisecond*20) + } +} + +func TestRPCListAndStatuses(t *testing.T) { + r := newTestRPC(t) + require.NoError(t, r.Create(newCreate(testServiceName, 2), &serviceV1.Response{})) + require.NoError(t, r.Create(newCreate(otherServiceName, 1), &serviceV1.Response{})) + + list := &serviceV1.List{} + require.NoError(t, r.List(&serviceV1.Service{}, list)) + require.ElementsMatch(t, []string{testServiceName, otherServiceName}, list.GetServices()) + + statuses := &serviceV1.Statuses{} + require.NoError(t, r.Statuses(&serviceV1.Service{Name: testServiceName}, statuses)) + require.Len(t, statuses.GetStatus(), 2) + + for _, st := range statuses.GetStatus() { + require.Nil(t, st.GetStatus()) + require.NotZero(t, st.GetPid()) + require.NotZero(t, st.GetMemoryUsage()) + require.Contains(t, st.GetCommand(), "sleep") + } + + // Status keeps only the last process of the group + status := &serviceV1.Status{} + require.NoError(t, r.Status(&serviceV1.Service{Name: testServiceName}, status)) + require.Equal(t, statuses.GetStatus()[1].GetPid(), status.GetPid()) + require.Equal(t, statuses.GetStatus()[1].GetCommand(), status.GetCommand()) +} + +func TestRPCStatusesReportsDeadProcess(t *testing.T) { + r := newTestRPC(t) + require.NoError(t, r.Create(newCreate(testServiceName, 1), &serviceV1.Response{})) + + procs := loadProcs(t, r) + procs[0].stop() + require.Eventually(t, func() bool { return !processAlive(procs[0].pid) }, + time.Second*10, time.Millisecond*20) + + statuses := &serviceV1.Statuses{} + require.NoError(t, r.Statuses(&serviceV1.Service{Name: testServiceName}, statuses)) + require.Len(t, statuses.GetStatus(), 1) + + // the pid and the command are still reported, the state carries the error + require.NotZero(t, statuses.GetStatus()[0].GetPid()) + require.Contains(t, statuses.GetStatus()[0].GetCommand(), "sleep") + require.NotEmpty(t, statuses.GetStatus()[0].GetStatus().GetMessage()) + + // Status has no per-process error slot and fails instead + require.Error(t, r.Status(&serviceV1.Service{Name: testServiceName}, &serviceV1.Status{})) +} + +// newTestRPC returns an rpc handler over a plugin wired to an in-memory logger, +// stopped at the end of the test. +func newTestRPC(t *testing.T) *rpc { + t.Helper() + + log, _ := newCaptureLogger() + p := &Plugin{logger: log} + t.Cleanup(func() { _ = p.Stop(context.Background()) }) + + return &rpc{p: p} +} + +// newCreate builds a create request for a long-lived command. +func newCreate(name string, processNum int64) *serviceV1.Create { + return &serviceV1.Create{ + Name: name, + Command: "sleep 30", + ProcessNum: processNum, + RestartSec: 1, + TimeoutStopSec: 1, + } +} + +// loadProcs returns the processes stored under the test service name. +func loadProcs(t *testing.T, r *rpc) []*Process { + t.Helper() + + v, ok := r.p.processes.Load(testServiceName) + require.True(t, ok) + + return v.([]*Process) +} + +// rpcPids returns the pids of the processes stored under the test service name. +func rpcPids(t *testing.T, r *rpc) []int64 { + t.Helper() + + procs := loadProcs(t, r) + pids := make([]int64, 0, len(procs)) + for i := range procs { + require.NotZero(t, procs[i].pid) + pids = append(pids, procs[i].pid) + } + + return pids +} diff --git a/state_test.go b/state_test.go new file mode 100644 index 0000000..515e804 --- /dev/null +++ b/state_test.go @@ -0,0 +1,26 @@ +package service + +import ( + "os" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestGeneralProcessState(t *testing.T) { + st, err := generalProcessState(int64(os.Getpid()), "go test") + require.NoError(t, err) + + require.Equal(t, int64(os.Getpid()), st.Pid) + require.Equal(t, "go test", st.Command) + require.EqualValues(t, 1, st.Status) + require.NotZero(t, st.MemoryUsage) +} + +func TestGeneralProcessStateInvalidPid(t *testing.T) { + st, err := generalProcessState(-1, "go test") + + require.Error(t, err) + require.Nil(t, st) + require.ErrorContains(t, err, "process_state") +} diff --git a/tests/concurrency_test.go b/tests/concurrency_test.go new file mode 100644 index 0000000..ed95bb0 --- /dev/null +++ b/tests/concurrency_test.go @@ -0,0 +1,97 @@ +package tests + +import ( + "sync" + "testing" + "time" + + "tests/helpers" + + serviceV1 "github.com/roadrunner-server/api-go/v6/service/v1" + rpcPlugin "github.com/roadrunner-server/rpc/v6" + "github.com/roadrunner-server/service/v6" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestServiceConcurrentRestart(t *testing.T) { + const ( + addr = "127.0.0.1:6311" + routines = 200 + rounds = 2 + ) + + rr, stop := helpers.Start(t, "configs/.rr-service-create-empty.yaml", + []any{&service.Plugin{}, &rpcPlugin.Plugin{}}, helpers.WithTCPProbe(addr)) + + client := helpers.RPC(t, addr) + helpers.Create(t, client, &serviceV1.Create{ + Name: "foo", + Command: "php php_test_files/loop.php", + ProcessNum: 1, + }) + + wg := &sync.WaitGroup{} + for range routines { + wg.Go(func() { + for range rounds { + // assert, not require: a failed require would only end this goroutine + assert.NoError(t, helpers.RestartErr(client, "foo")) + } + }) + } + wg.Wait() + + // the group survives every restart and still has a live process + require.Equal(t, []string{"foo"}, helpers.List(t, client)) + require.Len(t, helpers.Statuses(t, client, "foo"), 1) + + helpers.Terminate(t, client, "foo") + + stop() + + require.Empty(t, rr.Errs()) +} + +func TestServiceConcurrentRestartAndList(t *testing.T) { + const ( + addr = "127.0.0.1:6311" + routines = 200 + rounds = 2 + ) + + rr, stop := helpers.Start(t, "configs/.rr-service-create-empty.yaml", + []any{&service.Plugin{}, &rpcPlugin.Plugin{}}, + helpers.WithTCPProbe(addr), helpers.WithGracefulTimeout(time.Second*5)) + + client := helpers.RPC(t, addr) + helpers.Create(t, client, &serviceV1.Create{ + Name: "foo", + Command: "php php_test_files/loop.php", + ProcessNum: 1, + }) + + wg := &sync.WaitGroup{} + for range routines { + wg.Go(func() { + for range rounds { + assert.NoError(t, helpers.RestartErr(client, "foo")) + + // the service is never observable as missing while it is replaced + names, err := helpers.ListErr(client) + if assert.NoError(t, err) { + assert.Equal(t, []string{"foo"}, names) + } + } + }) + } + wg.Wait() + + require.Equal(t, []string{"foo"}, helpers.List(t, client)) + + helpers.Terminate(t, client, "foo") + + stop() + + require.Empty(t, rr.Errs()) +} diff --git a/tests/configs/.rr-service-create-empty.yaml b/tests/configs/.rr-service-create-empty.yaml index 88b42d1..066abaa 100644 --- a/tests/configs/.rr-service-create-empty.yaml +++ b/tests/configs/.rr-service-create-empty.yaml @@ -1,10 +1,6 @@ version: '3' rpc: - listen: tcp://127.0.0.1:6001 + listen: tcp://127.0.0.1:6311 -logs: - mode: development - level: debug - -service: { } \ No newline at end of file +service: { } diff --git a/tests/configs/.rr-service-create.yaml b/tests/configs/.rr-service-create.yaml index 3e309a2..50eaf5f 100644 --- a/tests/configs/.rr-service-create.yaml +++ b/tests/configs/.rr-service-create.yaml @@ -1,11 +1,7 @@ version: '3' rpc: - listen: tcp://127.0.0.1:6001 - -logs: - mode: development - level: debug + listen: tcp://127.0.0.1:6312 service: some_service_1: @@ -15,13 +11,8 @@ service: remain_after_exit: true restart_sec: 1 some_service_2: - command: "php_test_files/test_binary" + command: "php php_test_files/hello.php" process_num: 1 remain_after_exit: true - restart_delay: 1s + restart_sec: 1 exec_timeout: 5s - -endure: - grace_period: 120s - print_graph: false - log_level: error diff --git a/tests/configs/.rr-service-reload-2.yaml b/tests/configs/.rr-service-exec-timeout.yaml similarity index 69% rename from tests/configs/.rr-service-reload-2.yaml rename to tests/configs/.rr-service-exec-timeout.yaml index 2d9b307..96a4ff8 100644 --- a/tests/configs/.rr-service-reload-2.yaml +++ b/tests/configs/.rr-service-exec-timeout.yaml @@ -1,8 +1,5 @@ version: '3' -rpc: - listen: tcp://127.0.0.1:6001 - service: some_service_1: command: "php php_test_files/loop.php" @@ -11,13 +8,8 @@ service: remain_after_exit: true restart_sec: 1 some_service_2: - command: "php_test_files/test_binary" + command: "php php_test_files/hello.php" process_num: 10 remain_after_exit: true restart_sec: 1 exec_timeout: 2s - -logs: - mode: development - level: debug - encoding: console \ No newline at end of file diff --git a/tests/configs/.rr-service-from-php.yaml b/tests/configs/.rr-service-from-php.yaml index 5afb2bb..4b9a5c9 100644 --- a/tests/configs/.rr-service-from-php.yaml +++ b/tests/configs/.rr-service-from-php.yaml @@ -1,7 +1,7 @@ version: '3' rpc: - listen: tcp://127.0.0.1:6001 + listen: tcp://127.0.0.1:6313 service: some_service_1: @@ -10,8 +10,3 @@ service: exec_timeout: 20s # s,m,h (seconds, minutes, hours) remain_after_exit: false restart_sec: 1 - -endure: - grace_period: 10s - print_graph: false - log_level: debug diff --git a/tests/configs/.rr-service-init-remain.yaml b/tests/configs/.rr-service-init-remain.yaml deleted file mode 100644 index a2c1197..0000000 --- a/tests/configs/.rr-service-init-remain.yaml +++ /dev/null @@ -1,20 +0,0 @@ -version: '3' - -service: - some_service_1: - command: "php php_test_files/loop.php" - process_num: 1 - exec_timeout: 5s # s,m,h (seconds, minutes, hours) - remain_after_exit: false - restart_sec: 1 - some_service_2: - command: "php_test_files/test_binary" - process_num: 1 - remain_after_exit: false - restart_delay: 1s - exec_timeout: 5s - -endure: - grace_period: 120s - print_graph: false - log_level: error diff --git a/tests/configs/.rr-service-init-stdout.yaml b/tests/configs/.rr-service-init-stdout.yaml index 035636a..75bfa22 100644 --- a/tests/configs/.rr-service-init-stdout.yaml +++ b/tests/configs/.rr-service-init-stdout.yaml @@ -8,5 +8,5 @@ service: remain_after_exit: false restart_sec: 1 env: - - "FOO": "BAR" - - "FOO2": "BAZ" + FOO: "BAR" + FOO2: "BAZ" diff --git a/tests/configs/.rr-service-init.yaml b/tests/configs/.rr-service-init.yaml index 9868c1a..e7eadd9 100644 --- a/tests/configs/.rr-service-init.yaml +++ b/tests/configs/.rr-service-init.yaml @@ -2,17 +2,11 @@ version: '3' service: some_service_1: - command: "php ./php_test_files/loop.php" + command: "php php_test_files/loop.php" process_num: 1 exec_timeout: 5s # s,m,h (seconds, minutes, hours) restart_sec: 1 some_service_2: - command: "./php_test_files/test_binary" + command: "php php_test_files/hello.php" process_num: 1 - restart_delay: 1s exec_timeout: 5s - -endure: - grace_period: 10s - print_graph: false - log_level: error diff --git a/tests/configs/.rr-service-init_win.yaml b/tests/configs/.rr-service-init_win.yaml deleted file mode 100644 index 00c1502..0000000 --- a/tests/configs/.rr-service-init_win.yaml +++ /dev/null @@ -1,14 +0,0 @@ -version: '3' - -service: - some_service_1: - command: "php php_test_files/loop.php" - process_num: 1 - exec_timeout: 5s # s,m,h (seconds, minutes, hours) - remain_after_exit: false - restart_sec: 1 - -endure: - grace_period: 120s - print_graph: false - log_level: error diff --git a/tests/configs/.rr-service-reload.yaml b/tests/configs/.rr-service-list-terminate.yaml similarity index 53% rename from tests/configs/.rr-service-reload.yaml rename to tests/configs/.rr-service-list-terminate.yaml index e50732c..8e4ae8e 100644 --- a/tests/configs/.rr-service-reload.yaml +++ b/tests/configs/.rr-service-list-terminate.yaml @@ -1,23 +1,16 @@ version: '3' rpc: - listen: tcp://127.0.0.1:6111 + listen: tcp://127.0.0.1:6316 service: some_service_1: command: "php php_test_files/loop.php" process_num: 10 -# exec_timeout: 5s # s,m,h (seconds, minutes, hours) remain_after_exit: false restart_sec: 1 some_service_2: - command: "php_test_files/test_binary" + command: "php php_test_files/hello.php" process_num: 10 remain_after_exit: false restart_sec: 1 -# exec_timeout: 5s - -logs: - mode: development - level: debug - encoding: console \ No newline at end of file diff --git a/tests/configs/.rr-service-reload-22.yaml b/tests/configs/.rr-service-reload-22.yaml deleted file mode 100644 index ffb1bb2..0000000 --- a/tests/configs/.rr-service-reload-22.yaml +++ /dev/null @@ -1,23 +0,0 @@ -version: '3' - -rpc: - listen: tcp://127.0.0.1:6112 - -service: - some_service_1: - command: "php php_test_files/loop.php" - process_num: 10 -# exec_timeout: 5s # s,m,h (seconds, minutes, hours) - remain_after_exit: false - restart_sec: 1 - some_service_2: - command: "php_test_files/test_binary" - process_num: 10 - remain_after_exit: false - restart_sec: 1 -# exec_timeout: 5s - -logs: - mode: development - level: debug - encoding: console \ No newline at end of file diff --git a/tests/configs/.rr-service-reset.yaml b/tests/configs/.rr-service-reset.yaml index 04bd39e..06d0dc8 100644 --- a/tests/configs/.rr-service-reset.yaml +++ b/tests/configs/.rr-service-reset.yaml @@ -1,23 +1,16 @@ version: '3' rpc: - listen: tcp://127.0.0.1:6001 + listen: tcp://127.0.0.1:6314 service: some_service_1: command: "php php_test_files/loop.php" process_num: 10 -# exec_timeout: 5s # s,m,h (seconds, minutes, hours) remain_after_exit: false restart_sec: 1 some_service_2: - command: "php_test_files/test_binary" + command: "php php_test_files/hello.php" process_num: 10 remain_after_exit: false - restart_delay: 1s -# exec_timeout: 5s - -logs: - mode: development - level: debug - encoding: console \ No newline at end of file + restart_sec: 1 diff --git a/tests/configs/.rr-service-restarts.yaml b/tests/configs/.rr-service-restarts.yaml index 701b9a1..3410337 100644 --- a/tests/configs/.rr-service-restarts.yaml +++ b/tests/configs/.rr-service-restarts.yaml @@ -1,18 +1,9 @@ version: '3' service: - some_service_2: - command: "php_test_files/test_binary" + some_service_1: + command: "php php_test_files/hello.php" process_num: 1 - remain_after_exit: false - restart_delay: 1s - exec_timeout: 2s - -logs: - level: error - mode: raw - -endure: - grace_period: 120s - print_graph: false - log_level: error + remain_after_exit: true + restart_sec: 1 + exec_timeout: 1s diff --git a/tests/configs/.rr-service-workers.yaml b/tests/configs/.rr-service-workers.yaml index 99f83a3..68286f0 100644 --- a/tests/configs/.rr-service-workers.yaml +++ b/tests/configs/.rr-service-workers.yaml @@ -1,7 +1,7 @@ version: '3' rpc: - listen: tcp://127.0.0.1:6001 + listen: tcp://127.0.0.1:6315 service: some_service_1: @@ -11,12 +11,7 @@ service: remain_after_exit: false restart_sec: 1 some_service_2: - command: "php_test_files/test_binary" + command: "php php_test_files/hello.php" process_num: 10 remain_after_exit: false - restart_delay: 1s exec_timeout: 5s - -logs: - mode: development - level: debug diff --git a/tests/foo2.txt b/tests/foo2.txt deleted file mode 100644 index e69de29..0000000 diff --git a/tests/go.mod b/tests/go.mod index b39352c..007121e 100644 --- a/tests/go.mod +++ b/tests/go.mod @@ -5,13 +5,13 @@ 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/informer/v6 v6.0.0-beta.3 - github.com/roadrunner-server/logger/v6 v6.0.0-beta.3 - github.com/roadrunner-server/resetter/v6 v6.0.0-beta.4 + github.com/roadrunner-server/informer/v6 v6.0.0-beta.3.0.20260812194959-55c516b2a365 + github.com/roadrunner-server/pool/v2 v2.0.0-beta.1 + github.com/roadrunner-server/resetter/v6 v6.0.0-beta.4.0.20260812195002-d8956d0fde3c github.com/roadrunner-server/rpc/v6 v6.0.0-beta.5 github.com/roadrunner-server/service/v6 v6.0.0 github.com/stretchr/testify v1.11.1 @@ -32,7 +32,6 @@ require ( github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/roadrunner-server/api-plugins/v6 v6.0.0-beta.2 // indirect github.com/roadrunner-server/errors v1.5.0 // indirect - github.com/roadrunner-server/pool/v2 v2.0.0-beta.1 // indirect github.com/roadrunner-server/tcplisten v1.5.2 // indirect github.com/sagikazarmark/locafero v0.12.0 // indirect github.com/shirou/gopsutil v3.21.11+incompatible // indirect diff --git a/tests/go.sum b/tests/go.sum index 21d25a9..ea6c0f9 100644 --- a/tests/go.sum +++ b/tests/go.sum @@ -27,8 +27,8 @@ github.com/pelletier/go-toml/v2 v2.4.3 h1:GTRvJQutkOSftxIFD5xw9aepkYNuPWmVJpffdD github.com/pelletier/go-toml/v2 v2.4.3/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= 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/config/v6 v6.0.0-beta.3 h1:G0EUzJ6Yw4UnleM6BhnOBbYPXKDHRmCJiGhC3nXDBwI= @@ -39,14 +39,12 @@ github.com/roadrunner-server/errors v1.5.0 h1:unG7LKIZrSzkCCF3YLRLA5VyqE0KKomofX github.com/roadrunner-server/errors v1.5.0/go.mod h1:g9fo/T2C13cWRDR9PW1r0ZAOSQfNhWAZawyfkGiaHuI= 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/informer/v6 v6.0.0-beta.3 h1:Ca8rZynig3lsK2P9aaC93Wu9OJV7Hz6kxCfDumewWNw= -github.com/roadrunner-server/informer/v6 v6.0.0-beta.3/go.mod h1:Hl44YSlLszBtIIE98ynNFJp4T1iJDkgeBdMOieMuGZw= -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/informer/v6 v6.0.0-beta.3.0.20260812194959-55c516b2a365 h1:+e49oU2q2kUpC/40dXf+m+ja9ksaXj0jjTU7u5fifu8= +github.com/roadrunner-server/informer/v6 v6.0.0-beta.3.0.20260812194959-55c516b2a365/go.mod h1:JYSmwh7uts0NO/FOt311ml/rfL3OH+5ww3R4bQ2PvBM= github.com/roadrunner-server/pool/v2 v2.0.0-beta.1 h1:jpYXFtdD6QGAdAGPgMxrNi3j1CegCRpb2y+A+3GnXFA= github.com/roadrunner-server/pool/v2 v2.0.0-beta.1/go.mod h1:Bo1wT7RtL3eyQHXBUohNhtj/yAmRt6Rq8smuBg5pWkY= -github.com/roadrunner-server/resetter/v6 v6.0.0-beta.4 h1:zwCUQkb29WULIdftOtfz0Owlvqi7wU8u0U3IAqdGM2Q= -github.com/roadrunner-server/resetter/v6 v6.0.0-beta.4/go.mod h1:C6wq5K+jg4IYjYcME+Jiys1PHGlaDstnT5dg88inqi8= +github.com/roadrunner-server/resetter/v6 v6.0.0-beta.4.0.20260812195002-d8956d0fde3c h1:Ml5YNa4vF24bo8LRr+YhhzlcGvZ+a0Y960X73NQhhoo= +github.com/roadrunner-server/resetter/v6 v6.0.0-beta.4.0.20260812195002-d8956d0fde3c/go.mod h1:Qw4QCpDqht//IcqJ2K/pPQCHS9TRfqtjkFODg8eD6uE= github.com/roadrunner-server/rpc/v6 v6.0.0-beta.5 h1:FjwXfznbmyCEKFUHkxnvK8yo1HQKAQk+7fiPhSYC27E= github.com/roadrunner-server/rpc/v6 v6.0.0-beta.5/go.mod h1:z387hZZOEJ3+bB8iW1PEAEXF3jUUb/dDLDnVSc1CNNQ= github.com/roadrunner-server/tcplisten v1.5.2 h1:nn8yXYrhRDkfQ9AAu4V075uT4fZRmOnpxkawgE+bWPA= diff --git a/tests/helpers/rpc.go b/tests/helpers/rpc.go new file mode 100644 index 0000000..47a79cd --- /dev/null +++ b/tests/helpers/rpc.go @@ -0,0 +1,117 @@ +package helpers + +import ( + "net" + "net/rpc" + "testing" + + serviceV1 "github.com/roadrunner-server/api-go/v6/service/v1" + goridgeRpc "github.com/roadrunner-server/goridge/v4/pkg/rpc" + "github.com/roadrunner-server/pool/v2/state/process" + "github.com/stretchr/testify/require" +) + +// RPC dials the goridge endpoint at addr and returns a client closed at the end +// of the test. net/rpc clients are safe for concurrent use, so one client per +// test is enough even for the concurrency cases. +func RPC(t *testing.T, addr string) *rpc.Client { + t.Helper() + + var conn net.Conn + require.Eventually(t, func() bool { + d := net.Dialer{Timeout: probeDial} + c, err := d.DialContext(t.Context(), "tcp", addr) + if err != nil { + return false + } + + conn = c + return true + }, probeTimeout, probeTick, "rpc endpoint %s did not accept a connection", addr) + + client := rpc.NewClientWithCodec(goridgeRpc.NewClientCodec(conn)) + t.Cleanup(func() { _ = client.Close() }) + + return client +} + +// Create starts a new service group. +func Create(t *testing.T, c *rpc.Client, in *serviceV1.Create) { + t.Helper() + + out := &serviceV1.Response{} + require.NoError(t, c.Call("service.Create", in, out)) + require.True(t, out.GetOk()) +} + +// Terminate stops the named service group and drops it from the plugin. +func Terminate(t *testing.T, c *rpc.Client, name string) { + t.Helper() + + out := &serviceV1.Response{} + require.NoError(t, c.Call("service.Terminate", &serviceV1.Service{Name: name}, out)) + require.True(t, out.GetOk()) +} + +// Restart replaces every process of the named service group. +func Restart(t *testing.T, c *rpc.Client, name string) { + t.Helper() + require.NoError(t, RestartErr(c, name)) +} + +// RestartErr is Restart without a *testing.T, for calls made from goroutines. +func RestartErr(c *rpc.Client, name string) error { + out := &serviceV1.Response{} + return c.Call("service.Restart", &serviceV1.Service{Name: name}, out) +} + +// Statuses returns the per-process state of the named service group. +func Statuses(t *testing.T, c *rpc.Client, name string) []*serviceV1.Status { + t.Helper() + + out := &serviceV1.Statuses{} + require.NoError(t, c.Call("service.Statuses", &serviceV1.Service{Name: name}, out)) + + return out.GetStatus() +} + +// List returns the names of the service groups the plugin knows about. +func List(t *testing.T, c *rpc.Client) []string { + t.Helper() + + names, err := ListErr(c) + require.NoError(t, err) + + return names +} + +// ListErr is List without a *testing.T, for calls made from goroutines. +func ListErr(c *rpc.Client) ([]string, error) { + out := &serviceV1.List{} + if err := c.Call("service.List", &serviceV1.Service{}, out); err != nil { + return nil, err + } + + return out.GetServices(), nil +} + +// Reset asks the resetter plugin to reset the named plugin. +func Reset(t *testing.T, c *rpc.Client, plugin string) { + t.Helper() + + var ok bool + require.NoError(t, c.Call("resetter.Reset", plugin, &ok)) + require.True(t, ok) +} + +// Workers returns the process states the informer plugin reports for the plugin. +func Workers(t *testing.T, c *rpc.Client, plugin string) []process.State { + t.Helper() + + out := struct { + Workers []process.State `json:"workers"` + }{} + require.NoError(t, c.Call("informer.Workers", plugin, &out)) + + return out.Workers +} diff --git a/tests/helpers/rr.go b/tests/helpers/rr.go new file mode 100644 index 0000000..a1537a1 --- /dev/null +++ b/tests/helpers/rr.go @@ -0,0 +1,232 @@ +package helpers + +import ( + "context" + "log/slog" + "net" + "sync" + "testing" + "time" + + mocklogger "tests/mock" + + "github.com/roadrunner-server/config/v6" + "github.com/roadrunner-server/endure/v2" + "github.com/stretchr/testify/require" +) + +const ( + // defaultConfigVersion is the config schema version used by the test configs. + defaultConfigVersion = "2024.2.0" + // serviceStarted is written once per process, right after it is spawned. + serviceStarted = "service was started" + + // probeTimeout caps how long Start waits for the rpc listener to accept. + probeTimeout = time.Second * 15 + probeTick = time.Millisecond * 20 + probeDial = time.Second + + // logsTimeout caps how long WaitLogs waits for the expected records. + logsTimeout = time.Second * 30 + logsTick = time.Millisecond * 20 +) + +// bootCfg holds the options applied to a container before it is started. +type bootCfg struct { + inline string + graceful time.Duration + servicesStarted int + probeAddr string +} + +// Option customizes the container built by Start and its error-path variants. +type Option func(*bootCfg) + +// WithInlineConfig feeds the container YAML from memory; the cfgPath argument is ignored. +func WithInlineConfig(yaml string) Option { + return func(b *bootCfg) { b.inline = yaml } +} + +// WithGracefulTimeout sets the endure graceful shutdown timeout. +func WithGracefulTimeout(d time.Duration) Option { + return func(b *bootCfg) { b.graceful = d } +} + +// WithServicesStarted makes Start return only once n processes have been spawned. +func WithServicesStarted(n int) Option { + return func(b *bootCfg) { b.servicesStarted = n } +} + +// WithTCPProbe makes Start return only once addr accepts a connection, which is +// what the tests talking to the rpc plugin need. +func WithTCPProbe(addr string) Option { + return func(b *bootCfg) { b.probeAddr = addr } +} + +// RR is a running container. +type RR struct { + // Logs holds the records captured by the in-memory logger. + Logs *mocklogger.ObservedLogs + + mu sync.Mutex + errs []error +} + +// Errs returns the errors the container reported on its error channel so far. +func (rr *RR) Errs() []error { + rr.mu.Lock() + defer rr.mu.Unlock() + return append([]error(nil), rr.errs...) +} + +func (rr *RR) addErr(err error) { + rr.mu.Lock() + rr.errs = append(rr.errs, err) + rr.mu.Unlock() +} + +// Count returns the number of captured records whose message contains the snippet. +func (rr *RR) Count(snippet string) int { + return rr.Logs.FilterMessageSnippet(snippet).Len() +} + +// WaitLogs blocks until at least n captured records contain the snippet. +func (rr *RR) WaitLogs(t *testing.T, snippet string, n int) { + t.Helper() + + require.Eventually(t, func() bool { + return rr.Count(snippet) >= n + }, logsTimeout, logsTick, "expected at least %d log records containing %q", n, snippet) +} + +// CountExact returns the number of captured records whose message is exactly msg. +func (rr *RR) CountExact(msg string) int { + return rr.Logs.FilterMessage(msg).Len() +} + +// WaitLogsExact blocks until at least n captured records have exactly the message msg. +func (rr *RR) WaitLogsExact(t *testing.T, msg string, n int) { + t.Helper() + + require.Eventually(t, func() bool { + return rr.CountExact(msg) >= n + }, logsTimeout, logsTick, "expected at least %d log records with the message %q", n, msg) +} + +// Start registers the plugins, boots the container and waits for the readiness +// options, if any, to be satisfied. 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 logs written during shutdown can stop the container mid-test. +func Start(t *testing.T, cfgPath string, plugins []any, opts ...Option) (*RR, func()) { + t.Helper() + + cont, rr, bc := newContainer(t, cfgPath, plugins, opts) + 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 + } + rr.addErr(res.Error) + 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.servicesStarted > 0 { + rr.WaitLogs(t, serviceStarted, bc.servicesStarted) + } + + if bc.probeAddr != "" { + require.Eventually(t, func() bool { return dialOK(t.Context(), bc.probeAddr) }, + probeTimeout, probeTick, "rpc listener %s did not become ready", bc.probeAddr) + } + + return rr, stop +} + +// StartExpectInitError registers the plugins and requires Init to fail, returning its error. +func StartExpectInitError(t *testing.T, cfgPath string, plugins []any, opts ...Option) error { + t.Helper() + + cont, _, _ := newContainer(t, cfgPath, plugins, opts) + + err := cont.Init() + require.Error(t, err) + + return err +} + +// newContainer builds the container and registers the config, the in-memory +// logger and the caller's plugins. The container is not initialized yet. +func newContainer(t *testing.T, cfgPath string, plugins []any, opts []Option) (*endure.Endure, *RR, *bootCfg) { + t.Helper() + + bc := &bootCfg{} + for _, o := range opts { + o(bc) + } + + cfg := &config.Plugin{Version: defaultConfigVersion} + if bc.inline != "" { + cfg.Type = "yaml" + cfg.ReadInCfg = []byte(bc.inline) + } else { + cfg.Path = cfgPath + } + + var endureOpts []endure.Options + if bc.graceful != 0 { + endureOpts = append(endureOpts, endure.GracefulShutdownTimeout(bc.graceful)) + } + + // The service plugin owns no listener, so the log stream is the only place + // its behavior shows up: every test gets the observed logger. + l, obs := mocklogger.SlogTestLogger(slog.LevelDebug) + rr := &RR{Logs: obs} + + cont := endure.New(slog.LevelDebug, endureOpts...) + require.NoError(t, cont.RegisterAll(append([]any{cfg, l}, plugins...)...)) + + return cont, rr, bc +} + +// dialOK reports whether addr accepts a tcp connection. +func dialOK(ctx context.Context, addr string) bool { + d := net.Dialer{Timeout: probeDial} + conn, err := d.DialContext(ctx, "tcp", addr) + if err != nil { + return false + } + + _ = conn.Close() + return true +} diff --git a/tests/lifecycle_test.go b/tests/lifecycle_test.go new file mode 100644 index 0000000..a82a269 --- /dev/null +++ b/tests/lifecycle_test.go @@ -0,0 +1,112 @@ +package tests + +import ( + "testing" + "time" + + "tests/helpers" + + "github.com/roadrunner-server/service/v6" + "github.com/stretchr/testify/require" +) + +func TestServiceInit(t *testing.T) { + rr, stop := helpers.Start(t, "configs/.rr-service-init.yaml", []any{&service.Plugin{}}, + helpers.WithServicesStarted(2)) + + // both commands really execute: each writes its first line on startup + rr.WaitLogs(t, "The number is: 0", 1) + rr.WaitLogs(t, "Hello 0", 1) + + stop() + + require.Equal(t, 2, rr.Count("service was stopped")) + require.Empty(t, rr.Errs()) +} + +func TestServiceStdout(t *testing.T) { + rr, stop := helpers.Start(t, "configs/.rr-service-init-stdout.yaml", []any{&service.Plugin{}}, + helpers.WithServicesStarted(1)) + + // the child writes to stdout, which the plugin redirects into the log, and + // the env entries from the config reach it + rr.WaitLogs(t, "stdout write FOO=BAR FOO2=BAZ", 1) + + stop() + + require.Equal(t, 1, rr.Count("service was stopped")) + require.Empty(t, rr.Errs()) +} + +func TestServiceEnv(t *testing.T) { + rr, stop := helpers.Start(t, "configs/.rr-service-env.yaml", []any{&service.Plugin{}}, + helpers.WithServicesStarted(1)) + + // the script writes the marker only when both env entries arrive uppercased + rr.WaitLogs(t, "env ok", 1) + + stop() + + require.Zero(t, rr.Count("faillll")) + require.Empty(t, rr.Errs()) +} + +func TestServiceTrimOutput(t *testing.T) { + rr, stop := helpers.Start(t, "configs/.rr-service-newlines.yaml", []any{&service.Plugin{}}, + helpers.WithServicesStarted(1)) + + rr.WaitLogs(t, "stdout write", 1) + + stop() + + require.Equal(t, 1, rr.Count("stdout write")) + // the child writes "stdout write \n\t" and the record keeps none of the + // trailing whitespace, so the raw payload never shows up + require.Zero(t, rr.Count("stdout write ")) + require.Empty(t, rr.Errs()) +} + +func TestServiceCommandNotFound(t *testing.T) { + rr, stop := helpers.Start(t, "configs/.rr-service-error.yaml", []any{&service.Plugin{}}, + helpers.WithServicesStarted(1)) + + // php cannot open the script, reports it and exits non-zero + rr.WaitLogs(t, "Could not open input file", 1) + // the non-zero exit surfaces as a wait error + rr.WaitLogsExact(t, "wait", 1) + + // remain_after_exit is false, so the process is not brought back + require.Never(t, func() bool { return rr.Count("Could not open input file") > 1 }, + time.Second*2, time.Millisecond*100) + + stop() + + require.Empty(t, rr.Errs()) +} + +func TestServiceExitRestart(t *testing.T) { + rr, stop := helpers.Start(t, "configs/.rr-service-restarts.yaml", []any{&service.Plugin{}}, + helpers.WithServicesStarted(1)) + + // exec_timeout kills the process, remain_after_exit brings it back after + // restart_sec, and the fresh process writes its first line again + rr.WaitLogs(t, "Hello 0", 2) + + stop() + + require.NotZero(t, rr.CountExact("wait")) + require.Empty(t, rr.Errs()) +} + +func TestServiceBrokenConfig(t *testing.T) { + err := helpers.StartExpectInitError(t, "", []any{&service.Plugin{}}, helpers.WithInlineConfig(` +version: '3' + +service: + some_service_1: + command: "php php_test_files/loop.php" + exec_timeout: not-a-duration +`)) + + require.ErrorContains(t, err, "service_plugin_init") +} diff --git a/tests/mock/observer.go b/tests/mock/observer.go index 88f3c68..f553336 100644 --- a/tests/mock/observer.go +++ b/tests/mock/observer.go @@ -3,19 +3,14 @@ package mocklogger import ( "context" "log/slog" - "maps" - "reflect" "strings" "sync" - "time" ) // LoggedEntry is a representation of a log record captured by the observer. type LoggedEntry struct { Level slog.Level Message string - Time time.Time - Attrs map[string]any } // ObservedLogs is a concurrency-safe, ordered collection of observed logs. @@ -32,86 +27,34 @@ func (o *ObservedLogs) Len() int { return n } -// All returns a copy of all the observed logs. -func (o *ObservedLogs) All() []LoggedEntry { +// FilterMessage returns the entries whose message is exactly msg. Short messages +// such as "wait" need this instead of a snippet match. +func (o *ObservedLogs) FilterMessage(msg string) *ObservedLogs { o.mu.RLock() - ret := make([]LoggedEntry, len(o.logs)) - copy(ret, o.logs) - o.mu.RUnlock() - return ret -} - -// TakeAll returns a copy of all the observed logs, and truncates the observed -// slice. -func (o *ObservedLogs) TakeAll() []LoggedEntry { - o.mu.Lock() - ret := o.logs - o.logs = nil - o.mu.Unlock() - return ret -} + defer o.mu.RUnlock() -// AllUntimed returns a copy of all the observed logs, but overwrites the -// observed timestamps with time.Time's zero value. This is useful when making -// assertions in tests. -func (o *ObservedLogs) AllUntimed() []LoggedEntry { - ret := o.All() - for i := range ret { - ret[i].Time = time.Time{} + var filtered []LoggedEntry + for _, entry := range o.logs { + if entry.Message == msg { + filtered = append(filtered, entry) + } } - return ret -} -// FilterLevelExact filters entries to those logged at exactly the given level. -func (o *ObservedLogs) FilterLevelExact(level slog.Level) *ObservedLogs { - return o.Filter(func(e LoggedEntry) bool { - return e.Level == level - }) -} - -// FilterMessage filters entries to those that have the specified message. -func (o *ObservedLogs) FilterMessage(msg string) *ObservedLogs { - return o.Filter(func(e LoggedEntry) bool { - return e.Message == msg - }) + return &ObservedLogs{logs: filtered} } -// FilterMessageSnippet filters entries to those that have a message containing the specified snippet. +// FilterMessageSnippet returns the entries whose message contains the snippet. func (o *ObservedLogs) FilterMessageSnippet(snippet string) *ObservedLogs { - return o.Filter(func(e LoggedEntry) bool { - return strings.Contains(e.Message, snippet) - }) -} - -// FilterAttrKey filters entries to those that have the specified attribute key. -func (o *ObservedLogs) FilterAttrKey(key string) *ObservedLogs { - return o.Filter(func(e LoggedEntry) bool { - _, ok := e.Attrs[key] - return ok - }) -} - -// FilterAttr filters entries to those that have the specified attribute key -// AND value (compared with reflect.DeepEqual). -func (o *ObservedLogs) FilterAttr(key string, value any) *ObservedLogs { - return o.Filter(func(e LoggedEntry) bool { - v, ok := e.Attrs[key] - return ok && reflect.DeepEqual(v, value) - }) -} - -// Filter returns a copy of this ObservedLogs containing only those entries -// for which the provided function returns true. -func (o *ObservedLogs) Filter(keep func(LoggedEntry) bool) *ObservedLogs { o.mu.RLock() defer o.mu.RUnlock() var filtered []LoggedEntry for _, entry := range o.logs { - if keep(entry) { + if strings.Contains(entry.Message, snippet) { filtered = append(filtered, entry) } } + return &ObservedLogs{logs: filtered} } @@ -121,11 +64,11 @@ func (o *ObservedLogs) add(entry LoggedEntry) { o.mu.Unlock() } -// observerHandler is an slog.Handler that captures log records. +// observerHandler is an slog.Handler that captures the level and the message of +// every record. Attributes are not recorded, nothing asserts on them. type observerHandler struct { level slog.Level logs *ObservedLogs - attrs map[string]any } // NewObserverHandler creates a new slog.Handler that buffers logs in memory. @@ -134,7 +77,6 @@ func NewObserverHandler(level slog.Level) (slog.Handler, *ObservedLogs) { return &observerHandler{ level: level, logs: ol, - attrs: make(map[string]any), }, ol } @@ -143,36 +85,17 @@ func (h *observerHandler) Enabled(_ context.Context, level slog.Level) bool { } func (h *observerHandler) Handle(_ context.Context, r slog.Record) error { - attrs := make(map[string]any, len(h.attrs)) - maps.Copy(attrs, h.attrs) - r.Attrs(func(a slog.Attr) bool { - attrs[a.Key] = a.Value.Any() - return true - }) - h.logs.add(LoggedEntry{ Level: r.Level, Message: r.Message, - Time: r.Time, - Attrs: attrs, }) return nil } -func (h *observerHandler) WithAttrs(attrs []slog.Attr) slog.Handler { - newAttrs := make(map[string]any, len(h.attrs)+len(attrs)) - maps.Copy(newAttrs, h.attrs) - for _, a := range attrs { - newAttrs[a.Key] = a.Value.Any() - } - return &observerHandler{ - level: h.level, - logs: h.logs, - attrs: newAttrs, - } +func (h *observerHandler) WithAttrs(_ []slog.Attr) slog.Handler { + return h } -func (h *observerHandler) WithGroup(name string) slog.Handler { - _ = name +func (h *observerHandler) WithGroup(_ string) slog.Handler { return h } diff --git a/tests/php_test_files/create_3_services.php b/tests/php_test_files/create_3_services.php index 1dbf9a3..2b10dd5 100644 --- a/tests/php_test_files/create_3_services.php +++ b/tests/php_test_files/create_3_services.php @@ -6,7 +6,7 @@ ini_set('display_errors', 'stderr'); require __DIR__ . "/vendor/autoload.php"; -$rpc = RPC::create('tcp://127.0.0.1:6001'); +$rpc = RPC::create('tcp://127.0.0.1:6313'); $manager = new Manager($rpc); diff --git a/tests/php_test_files/env.php b/tests/php_test_files/env.php index fb2f41d..041d40a 100644 --- a/tests/php_test_files/env.php +++ b/tests/php_test_files/env.php @@ -1,13 +1,12 @@ bar -// bar -> foo -//var_dump($_SERVER); -if ($_SERVER["FOO"] !== "BAR" ) { +// The config maps foo -> BAR and bar -> FOO; the plugin uppercases the keys. +if ($_SERVER["FOO"] !== "BAR") { die("faillll"); } -if ($_SERVER["BAR"] !== "FOO" ) { +if ($_SERVER["BAR"] !== "FOO") { die("faillll"); } -exit(); \ No newline at end of file +error_log("env ok", 0); +exit(); diff --git a/tests/php_test_files/hello.php b/tests/php_test_files/hello.php new file mode 100644 index 0000000..5b00029 --- /dev/null +++ b/tests/php_test_files/hello.php @@ -0,0 +1,8 @@ + diff --git a/tests/php_test_files/loop_stdout.php b/tests/php_test_files/loop_stdout.php index e6a10ec..62ab7c9 100644 --- a/tests/php_test_files/loop_stdout.php +++ b/tests/php_test_files/loop_stdout.php @@ -1,8 +1,8 @@ diff --git a/tests/php_test_files/test_binary b/tests/php_test_files/test_binary deleted file mode 100644 index 480fb7e..0000000 Binary files a/tests/php_test_files/test_binary and /dev/null differ diff --git a/tests/reset_test.go b/tests/reset_test.go new file mode 100644 index 0000000..27d50d9 --- /dev/null +++ b/tests/reset_test.go @@ -0,0 +1,54 @@ +package tests + +import ( + "testing" + + "tests/helpers" + + "github.com/roadrunner-server/resetter/v6" + rpcPlugin "github.com/roadrunner-server/rpc/v6" + "github.com/roadrunner-server/service/v6" + "github.com/stretchr/testify/require" +) + +func TestServiceResetterRestartsProcesses(t *testing.T) { + const addr = "127.0.0.1:6314" + + rr, stop := helpers.Start(t, "configs/.rr-service-reset.yaml", + []any{&service.Plugin{}, &rpcPlugin.Plugin{}, &resetter.Plugin{}}, + helpers.WithServicesStarted(20), helpers.WithTCPProbe(addr)) + + // each of the 10+10 processes writes its first line once per start + rr.WaitLogs(t, "The number is: 0", 10) + rr.WaitLogs(t, "Hello 0", 10) + + helpers.Reset(t, helpers.RPC(t, addr), "service") + + rr.WaitLogs(t, "The number is: 0", 20) + rr.WaitLogs(t, "Hello 0", 20) + + stop() + + // remain_after_exit is false, so the reset is the only thing that can start + // a second generation of processes + require.Equal(t, 20, rr.Count("The number is: 0")) + require.Equal(t, 20, rr.Count("Hello 0")) + require.Equal(t, 20, rr.Count("service was stopped")) + require.Empty(t, rr.Errs()) +} + +func TestServiceExecTimeoutRestart(t *testing.T) { + rr, stop := helpers.Start(t, "configs/.rr-service-exec-timeout.yaml", []any{&service.Plugin{}}, + helpers.WithServicesStarted(20)) + + // exec_timeout kills every process and remain_after_exit brings it back + // restart_sec later, so the first line of each command is written again + rr.WaitLogs(t, "The number is: 0", 20) + rr.WaitLogs(t, "Hello 0", 20) + + stop() + + // every kill surfaces as a wait error before the process is started again + require.NotZero(t, rr.CountExact("wait")) + require.Empty(t, rr.Errs()) +} diff --git a/tests/rpc_test.go b/tests/rpc_test.go new file mode 100644 index 0000000..ecd0183 --- /dev/null +++ b/tests/rpc_test.go @@ -0,0 +1,208 @@ +package tests + +import ( + "os" + "os/exec" + "slices" + "testing" + "time" + + "tests/helpers" + + serviceV1 "github.com/roadrunner-server/api-go/v6/service/v1" + rpcPlugin "github.com/roadrunner-server/rpc/v6" + "github.com/roadrunner-server/service/v6" + "github.com/stretchr/testify/require" +) + +func TestServiceRPCCreate(t *testing.T) { + const addr = "127.0.0.1:6312" + + rr, stop := helpers.Start(t, "configs/.rr-service-create.yaml", + []any{&service.Plugin{}, &rpcPlugin.Plugin{}}, + helpers.WithServicesStarted(2), helpers.WithTCPProbe(addr)) + + client := helpers.RPC(t, addr) + helpers.Create(t, client, &serviceV1.Create{ + Name: "foo", + Command: "php php_test_files/loop.php", + ProcessNum: 1, + ExecTimeout: 10, + Env: map[string]string{"foo": "bar"}, + }) + + require.ElementsMatch(t, []string{"some_service_1", "some_service_2", "foo"}, helpers.List(t, client)) + + helpers.Terminate(t, client, "foo") + require.ElementsMatch(t, []string{"some_service_1", "some_service_2"}, helpers.List(t, client)) + + stop() + + require.Empty(t, rr.Errs()) +} + +func TestServiceRPCCreateOnEmptyConfig(t *testing.T) { + const addr = "127.0.0.1:6311" + + rr, stop := helpers.Start(t, "configs/.rr-service-create-empty.yaml", + []any{&service.Plugin{}, &rpcPlugin.Plugin{}}, helpers.WithTCPProbe(addr)) + + client := helpers.RPC(t, addr) + require.Empty(t, helpers.List(t, client)) + + helpers.Create(t, client, &serviceV1.Create{ + Name: "foo", + Command: "php php_test_files/loop.php", + ProcessNum: 1, + ExecTimeout: 10, + }) + + require.Equal(t, []string{"foo"}, helpers.List(t, client)) + rr.WaitLogs(t, "The number is: 0", 1) + + helpers.Terminate(t, client, "foo") + require.Empty(t, helpers.List(t, client)) + + stop() + + require.Empty(t, rr.Errs()) +} + +func TestServiceRPCRestart(t *testing.T) { + const addr = "127.0.0.1:6311" + + rr, stop := helpers.Start(t, "configs/.rr-service-create-empty.yaml", + []any{&service.Plugin{}, &rpcPlugin.Plugin{}}, helpers.WithTCPProbe(addr)) + + client := helpers.RPC(t, addr) + helpers.Create(t, client, &serviceV1.Create{ + Name: "foo", + Command: "php php_test_files/loop.php", + ProcessNum: 2, + ExecTimeout: 10, + }) + + before := statusPids(t, helpers.Statuses(t, client, "foo")) + require.Len(t, before, 2) + + helpers.Restart(t, client, "foo") + + after := statusPids(t, helpers.Statuses(t, client, "foo")) + require.Len(t, after, 2) + for _, pid := range after { + require.NotContains(t, before, pid) + } + + helpers.Terminate(t, client, "foo") + + stop() + + require.Empty(t, rr.Errs()) +} + +func TestServiceRPCStatuses(t *testing.T) { + const addr = "127.0.0.1:6311" + + rr, stop := helpers.Start(t, "configs/.rr-service-create-empty.yaml", + []any{&service.Plugin{}, &rpcPlugin.Plugin{}}, helpers.WithTCPProbe(addr)) + + client := helpers.RPC(t, addr) + helpers.Create(t, client, &serviceV1.Create{ + Name: "foo", + Command: "php php_test_files/loop.php", + ProcessNum: 2, + ExecTimeout: 10, + }) + + statuses := helpers.Statuses(t, client, "foo") + require.Len(t, statuses, 2) + + for _, st := range statuses { + require.Nil(t, st.GetStatus()) + require.NotZero(t, st.GetPid()) + require.NotZero(t, st.GetMemoryUsage()) + require.Contains(t, st.GetCommand(), "loop.php") + } + + helpers.Terminate(t, client, "foo") + + stop() + + require.Empty(t, rr.Errs()) +} + +func TestServiceRPCListAndTerminate(t *testing.T) { + const addr = "127.0.0.1:6316" + + rr, stop := helpers.Start(t, "configs/.rr-service-list-terminate.yaml", + []any{&service.Plugin{}, &rpcPlugin.Plugin{}}, + helpers.WithServicesStarted(20), helpers.WithTCPProbe(addr)) + + client := helpers.RPC(t, addr) + names := helpers.List(t, client) + require.ElementsMatch(t, []string{"some_service_1", "some_service_2"}, names) + + for _, name := range names { + helpers.Terminate(t, client, name) + } + + require.Empty(t, helpers.List(t, client)) + require.Equal(t, 20, rr.Count("service was started")) + + stop() + + require.Empty(t, rr.Errs()) +} + +func TestServiceRPCCreateFromPHP(t *testing.T) { + requirePHPRPCClient(t) + + const addr = "127.0.0.1:6313" + + rr, stop := helpers.Start(t, "configs/.rr-service-from-php.yaml", + []any{&service.Plugin{}, &rpcPlugin.Plugin{}}, + helpers.WithServicesStarted(1), helpers.WithTCPProbe(addr)) + + client := helpers.RPC(t, addr) + + // the php script creates a group of three processes over rpc + require.Eventually(t, func() bool { + names, err := helpers.ListErr(client) + return err == nil && slices.Contains(names, "listen-jobs") + }, time.Second*20, time.Millisecond*50, "the php side did not create its service") + + require.Len(t, helpers.Statuses(t, client, "listen-jobs"), 3) + + stop() + + // the configured service plus the three processes the php side created + require.Equal(t, 4, rr.Count("service was stopped")) + require.Empty(t, rr.Errs()) +} + +// statusPids collects the pids reported for a service group. +func statusPids(t *testing.T, statuses []*serviceV1.Status) []int32 { + t.Helper() + + pids := make([]int32, 0, len(statuses)) + for _, st := range statuses { + require.NotZero(t, st.GetPid()) + pids = append(pids, st.GetPid()) + } + + return pids +} + +// requirePHPRPCClient skips unless the php side can open an rpc connection: the +// composer dependencies have to be installed and php needs the sockets extension. +func requirePHPRPCClient(t *testing.T) { + t.Helper() + + if _, err := os.Stat("php_test_files/vendor/autoload.php"); err != nil { + t.Skip("php_test_files/vendor is not installed") + } + + if err := exec.CommandContext(t.Context(), "php", "-r", `exit(extension_loaded("sockets") ? 0 : 1);`).Run(); err != nil { + t.Skip("php is built without the sockets extension") + } +} diff --git a/tests/service_plugin_test.go b/tests/service_plugin_test.go deleted file mode 100644 index 9ae28c3..0000000 --- a/tests/service_plugin_test.go +++ /dev/null @@ -1,1626 +0,0 @@ -package service - -import ( - "log/slog" - "net" - "net/rpc" - "os" - "os/signal" - "sync" - "syscall" - "testing" - "time" - - mocklogger "tests/mock" - - informerProto "github.com/roadrunner-server/api-go/v6/informer/v1" - resetterProto "github.com/roadrunner-server/api-go/v6/resetter/v1" - serviceProto "github.com/roadrunner-server/api-go/v6/service/v1" - "github.com/roadrunner-server/config/v6" - "github.com/roadrunner-server/endure/v2" - goridgeRpc "github.com/roadrunner-server/goridge/v4/pkg/rpc" - "github.com/roadrunner-server/informer/v6" - "github.com/roadrunner-server/logger/v6" - "github.com/roadrunner-server/resetter/v6" - rpcPlugin "github.com/roadrunner-server/rpc/v6" - "github.com/roadrunner-server/service/v6" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -const serviceRPCAddr = "127.0.0.1:6001" - -func newRPCClient(t *testing.T, address string) *rpc.Client { - t.Helper() - var ( - conn net.Conn - err error - ) - d := net.Dialer{Timeout: 5 * time.Second} - // simple dial retry loop: the goridge RPC server may not be accepting yet - // right after Serve, or during the port handoff between sequential tests. - for range 10 { - conn, err = d.DialContext(t.Context(), "tcp", address) - if err == nil { - break - } - time.Sleep(100 * time.Millisecond) - } - require.NoError(t, err) - return rpc.NewClientWithCodec(goridgeRpc.NewClientCodec(conn)) -} - -func TestServiceInit(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-init.yaml", - } - - err := cont.RegisterAll( - cfg, - &logger.Plugin{}, - &service.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()) - } - return - 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 * 10) - stopCh <- struct{}{} - wg.Wait() -} - -func TestServicePHPCreate(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.2.0", - Path: "configs/.rr-service-from-php.yaml", - } - - l, oLogger := mocklogger.SlogTestLogger(slog.LevelDebug) - err := cont.RegisterAll( - l, - cfg, - &rpcPlugin.Plugin{}, - &service.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()) - } - return - 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 * 15) - stopCh <- struct{}{} - wg.Wait() - time.Sleep(time.Second) - - assert.GreaterOrEqual(t, oLogger.FilterMessageSnippet("service was stopped").Len(), 4) -} - -func TestServiceTrimOutput(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-newlines.yaml", - } - - l, oLogger := mocklogger.SlogTestLogger(slog.LevelDebug) - err := cont.RegisterAll( - cfg, - l, - &service.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()) - } - return - 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 * 5) - stopCh <- struct{}{} - wg.Wait() - - require.Equal(t, 1, oLogger.FilterMessageSnippet("stdout write").Len()) -} - -func TestServiceWorkers(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-workers.yaml", - } - - err := cont.RegisterAll( - cfg, - &logger.Plugin{}, - &service.Plugin{}, - &informer.Plugin{}, - &rpcPlugin.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()) - } - return - 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("workers", workers("service")) - time.Sleep(time.Second) - stopCh <- struct{}{} - wg.Wait() -} - -func TestServiceInitStdout(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-init-stdout.yaml", - } - - err := cont.RegisterAll( - cfg, - &logger.Plugin{}, - &service.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()) - } - return - 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 * 5) - stopCh <- struct{}{} - wg.Wait() -} - -func TestServiceEnv(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-env.yaml", - } - - l, oLogger := mocklogger.SlogTestLogger(slog.LevelDebug) - err := cont.RegisterAll( - cfg, - l, - &service.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()) - } - return - 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 * 5) - stopCh <- struct{}{} - wg.Wait() - require.Equal(t, 0, oLogger.FilterMessageSnippet("faillll").Len()) -} - -func TestServiceError(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-error.yaml", - } - - err := cont.RegisterAll( - cfg, - &logger.Plugin{}, - &service.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - assert.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()) - } - return - 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 * 10) - stopCh <- struct{}{} - wg.Wait() -} - -func TestServiceRestarts(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-restarts.yaml", - } - - err := cont.RegisterAll( - cfg, - &logger.Plugin{}, - &service.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()) - } - return - 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 * 10) - stopCh <- struct{}{} - wg.Wait() -} - -func TestServiceCreate(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-create.yaml", - } - - err := cont.RegisterAll( - cfg, - &logger.Plugin{}, - &service.Plugin{}, - &rpcPlugin.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()) - } - return - 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) - - in := &serviceProto.Create{ - Name: "foo", - Command: "php php_test_files/loop.php", - ProcessNum: 1, - ExecTimeout: 3, - RemainAfterExit: false, - Env: map[string]string{"foo": "bar"}, - RestartSec: 0, - } - - out := &serviceProto.Response{} - - t.Run("create", create(in, out)) - - time.Sleep(time.Second * 2) - - out = &serviceProto.Response{} - t.Run("terminate", terminate("127.0.0.1:6001", &serviceProto.Service{Name: "foo"}, out)) - - stopCh <- struct{}{} - wg.Wait() -} - -func TestServiceCreateEmptyConfig(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-create-empty.yaml", - } - - err := cont.RegisterAll( - cfg, - &logger.Plugin{}, - &service.Plugin{}, - &rpcPlugin.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()) - } - return - 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) - - in := &serviceProto.Create{ - Name: "foo", - Command: "php php_test_files/loop.php", - ProcessNum: 1, - ExecTimeout: 3, - RemainAfterExit: false, - Env: map[string]string{"foo": "bar"}, - RestartSec: 0, - } - - out := &serviceProto.Response{} - - t.Run("create", create(in, out)) - - time.Sleep(time.Second * 3) - l := &serviceProto.List{} - t.Run("list", list("127.0.0.1:6001", &serviceProto.Service{}, l)) - - for i := range len(l.GetServices()) { - cmd := &serviceProto.Service{ - Name: l.GetServices()[i], - } - - out = &serviceProto.Response{} - - t.Run("terminate", terminate("127.0.0.1:6001", cmd, out)) - } - - time.Sleep(time.Second * 2) - stopCh <- struct{}{} - wg.Wait() -} - -func TestServiceRestart(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-create-empty.yaml", - } - - err := cont.RegisterAll( - cfg, - &logger.Plugin{}, - &service.Plugin{}, - &rpcPlugin.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()) - } - return - 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) - - in := &serviceProto.Create{ - Name: "foo", - Command: "php php_test_files/loop.php", - ProcessNum: 1, - ExecTimeout: 3, - RemainAfterExit: false, - Env: map[string]string{"foo": "bar"}, - RestartSec: 0, - } - - out := &serviceProto.Response{} - - t.Run("create", create(in, out)) - - time.Sleep(time.Second * 3) - l := &serviceProto.List{} - t.Run("list", list("127.0.0.1:6001", &serviceProto.Service{}, l)) - - for i := range len(l.GetServices()) { - cmd := &serviceProto.Service{ - Name: l.GetServices()[i], - } - - out = &serviceProto.Response{} - - t.Run("restart", restart(cmd, out)) - - time.Sleep(time.Second * 5) - } - - time.Sleep(time.Second * 2) - out = &serviceProto.Response{} - t.Run("terminate", terminate("127.0.0.1:6001", &serviceProto.Service{Name: "foo"}, out)) - stopCh <- struct{}{} - wg.Wait() -} - -func TestServiceRestartConcurrent(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-create-empty.yaml", - } - - err := cont.RegisterAll( - cfg, - &logger.Plugin{}, - &service.Plugin{}, - &rpcPlugin.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()) - } - return - 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) - - in := &serviceProto.Create{ - Name: "foo", - Command: "php php_test_files/loop.php", - ProcessNum: 1, - ExecTimeout: 3, - RemainAfterExit: false, - Env: map[string]string{"foo": "bar"}, - RestartSec: 0, - } - - out := &serviceProto.Response{} - - t.Run("create", create(in, out)) - - time.Sleep(time.Second) - - l := &serviceProto.List{} - t.Run("list", list("127.0.0.1:6001", nil, l)) - - for range 100 { - go func() { - for i := range len(l.GetServices()) { - cmd := &serviceProto.Service{ - Name: l.GetServices()[i], - } - - out1 := &serviceProto.Response{} - - t.Run("restart", restart(cmd, out1)) - - time.Sleep(time.Second) - } - }() - - go func() { - for i := range len(l.GetServices()) { - cmd := &serviceProto.Service{ - Name: l.GetServices()[i], - } - - out2 := &serviceProto.Response{} - - t.Run("restart", restart(cmd, out2)) - - time.Sleep(time.Second) - } - }() - } - - time.Sleep(time.Second * 10) - out = &serviceProto.Response{} - t.Run("terminate", terminate("127.0.0.1:6001", &serviceProto.Service{Name: "foo"}, out)) - stopCh <- struct{}{} - wg.Wait() -} - -func TestServiceListConcurrent(t *testing.T) { - cont := endure.New(slog.LevelDebug, endure.GracefulShutdownTimeout(time.Second*5)) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-create-empty.yaml", - } - - err := cont.RegisterAll( - cfg, - &logger.Plugin{}, - &service.Plugin{}, - &rpcPlugin.Plugin{}, - ) - assert.NoError(t, err) - - require.NoError(t, cont.Init()) - - 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()) - } - return - 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) - - in := &serviceProto.Create{ - Name: "foo", - Command: "php php_test_files/loop.php", - ProcessNum: 1, - ExecTimeout: 3, - RemainAfterExit: false, - Env: map[string]string{"foo": "bar"}, - RestartSec: 0, - } - - out := &serviceProto.Response{} - - t.Run("create", create(in, out)) - - time.Sleep(time.Second) - - l := &serviceProto.List{} - t.Run("list", list("127.0.0.1:6001", nil, l)) - - for range 100 { - go func() { - for i := range len(l.GetServices()) { - cmd := &serviceProto.Service{ - Name: l.GetServices()[i], - } - - out1 := &serviceProto.Response{} - - t.Run("restart", restart(cmd, out1)) - ll := &serviceProto.List{} - t.Run("list", list("127.0.0.1:6001", nil, ll)) - require.Len(t, ll.GetServices(), 1) - - time.Sleep(time.Millisecond * 100) - } - }() - - go func() { - for i := range len(l.GetServices()) { - cmd := &serviceProto.Service{ - Name: l.GetServices()[i], - } - - out2 := &serviceProto.Response{} - t.Run("restart", restart(cmd, out2)) - ll := &serviceProto.List{} - t.Run("list", list("127.0.0.1:6001", nil, ll)) - require.Len(t, ll.GetServices(), 1) - - time.Sleep(time.Millisecond * 100) - } - }() - } - - time.Sleep(time.Second * 15) - out = &serviceProto.Response{} - t.Run("terminate", terminate("127.0.0.1:6001", &serviceProto.Service{Name: "foo"}, out)) - stopCh <- struct{}{} - wg.Wait() -} - -func TestServiceStatus(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-create-empty.yaml", - } - - err := cont.RegisterAll( - cfg, - &logger.Plugin{}, - &service.Plugin{}, - &rpcPlugin.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()) - } - return - 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) - - in := &serviceProto.Create{ - Name: "foo", - Command: "php php_test_files/loop.php", - ProcessNum: 1, - ExecTimeout: 10, - RemainAfterExit: false, - Env: map[string]string{"foo": "bar"}, - RestartSec: 0, - } - - out := &serviceProto.Response{} - - t.Run("create", create(in, out)) - - time.Sleep(time.Second) - - l := &serviceProto.List{} - t.Run("list", list("127.0.0.1:6001", nil, l)) - require.Len(t, l.GetServices(), 1) - - inStat := &serviceProto.Service{ - Name: l.GetServices()[0], - } - - outStat := &serviceProto.Statuses{} - t.Run("stats", status(inStat, outStat)) - require.NotEmpty(t, outStat.Status[0].GetCommand()) - require.NotZero(t, outStat.Status[0].GetMemoryUsage()) - require.NotZero(t, outStat.Status[0].GetPid()) - - out = &serviceProto.Response{} - t.Run("terminate", terminate("127.0.0.1:6001", &serviceProto.Service{Name: l.GetServices()[0]}, out)) - - time.Sleep(time.Second * 2) - stopCh <- struct{}{} - wg.Wait() -} - -func TestServiceInitRemain(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-init-remain.yaml", - } - - err := cont.RegisterAll( - cfg, - &logger.Plugin{}, - &service.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()) - } - return - 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 * 10) - stopCh <- struct{}{} - wg.Wait() -} - -func TestServiceReset(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-reset.yaml", - } - - l, oLogger := mocklogger.SlogTestLogger(slog.LevelDebug) - err := cont.RegisterAll( - cfg, - &rpcPlugin.Plugin{}, - &resetter.Plugin{}, - l, - &service.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()) - } - return - 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 * 2) - - client := newRPCClient(t, serviceRPCAddr) - defer func() { _ = client.Close() }() - resp := &resetterProto.Response{} - require.NoError(t, client.Call("resetter.Reset", &resetterProto.ResetRequest{Plugin: "service"}, resp)) - require.True(t, resp.GetOk()) - - time.Sleep(time.Second * 5) - stopCh <- struct{}{} - wg.Wait() - - require.Equal(t, 20, oLogger.FilterMessageSnippet("The number is: 0").Len()) - require.Equal(t, 20, oLogger.FilterMessageSnippet("Hello 0").Len()) -} - -func TestServiceReset2(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-reload-22.yaml", - } - - l, oLogger := mocklogger.SlogTestLogger(slog.LevelDebug) - err := cont.RegisterAll( - cfg, - l, - &rpcPlugin.Plugin{}, - &resetter.Plugin{}, - &service.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()) - } - return - 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 * 2) - - file, err := os.Create("foo.txt") - require.NoError(t, err) - - go func() { - d := net.Dialer{Timeout: 5 * time.Second} - conn, dialErr := d.DialContext(t.Context(), "tcp", "127.0.0.1:6112") - // assert (not require) inside goroutine: assert routes through t.Errorf, - // which is documented safe for concurrent use; require.FailNow would only - // terminate this goroutine and silently let the test pass. - if !assert.NoError(t, dialErr) { - return - } - client := rpc.NewClientWithCodec(goridgeRpc.NewClientCodec(conn)) - defer func() { _ = client.Close() }() - resp := &resetterProto.Response{} - callErr := client.Call("resetter.Reset", &resetterProto.ResetRequest{Plugin: "service"}, resp) - assert.NoError(t, callErr) - if callErr == nil { - assert.True(t, resp.GetOk()) - } - }() - - time.Sleep(time.Second * 10) - stopCh <- struct{}{} - wg.Wait() - - assert.LessOrEqual(t, oLogger.FilterMessageSnippet("The number is: 0").Len(), 30) - assert.LessOrEqual(t, oLogger.FilterMessageSnippet("Hello 0").Len(), 30) - - t.Cleanup(func() { - _ = file.Close() - _ = os.Remove("foo.txt") - }) -} - -func TestServiceReset3(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-reload-2.yaml", - } - - l, oLogger := mocklogger.SlogTestLogger(slog.LevelDebug) - err := cont.RegisterAll( - cfg, - l, - &rpcPlugin.Plugin{}, - &resetter.Plugin{}, - &service.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()) - } - return - 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 * 2) - - file, err := os.Create("foo.txt") - require.NoError(t, err) - - time.Sleep(time.Second * 4) - stopCh <- struct{}{} - wg.Wait() - - assert.Equal(t, 20, oLogger.FilterMessageSnippet("The number is: 0").Len()) - assert.Equal(t, 20, oLogger.FilterMessageSnippet("Hello 0").Len()) - - t.Cleanup(func() { - _ = file.Close() - _ = os.Remove("foo.txt") - }) -} - -func TestServiceReset4(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-service-reload.yaml", - } - - l, oLogger := mocklogger.SlogTestLogger(slog.LevelDebug) - err := cont.RegisterAll( - cfg, - l, - &rpcPlugin.Plugin{}, - &resetter.Plugin{}, - &service.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()) - } - return - 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 * 2) - - file, err := os.Create("foo2.txt") - require.NoError(t, err) - - go func() { - l1 := &serviceProto.List{} - t.Run("list", list("127.0.0.1:6111", nil, l1)) - require.Len(t, l1.GetServices(), 2) - }() - - go func() { - l2 := &serviceProto.List{} - t.Run("list", list("127.0.0.1:6111", nil, l2)) - require.Len(t, l2.GetServices(), 2) - for i := range len(l2.GetServices()) { - cmd := &serviceProto.Service{ - Name: l2.GetServices()[i], - } - - out := &serviceProto.Response{} - - t.Run("terminate", terminate("127.0.0.1:6111", cmd, out)) - } - }() - - time.Sleep(time.Second * 10) - stopCh <- struct{}{} - wg.Wait() - - require.Equal(t, 20, oLogger.FilterMessageSnippet("service was started").Len()) - t.Cleanup(func() { - _ = file.Close() - _ = os.Remove("foo2.txt") - }) -} - -func create(in *serviceProto.Create, out *serviceProto.Response) func(t *testing.T) { - return func(t *testing.T) { - client := newRPCClient(t, serviceRPCAddr) - defer func() { _ = client.Close() }() - require.NoError(t, client.Call("service.CreateService", in, out)) - } -} - -func terminate(address string, in *serviceProto.Service, out *serviceProto.Response) func(t *testing.T) { - return func(t *testing.T) { - client := newRPCClient(t, address) - defer func() { _ = client.Close() }() - require.NoError(t, client.Call("service.Terminate", in, out)) - } -} - -func restart(in *serviceProto.Service, out *serviceProto.Response) func(t *testing.T) { - return func(t *testing.T) { - client := newRPCClient(t, serviceRPCAddr) - defer func() { _ = client.Close() }() - require.NoError(t, client.Call("service.Restart", in, out)) - } -} - -func status(in *serviceProto.Service, out *serviceProto.Statuses) func(t *testing.T) { - return func(t *testing.T) { - client := newRPCClient(t, serviceRPCAddr) - defer func() { _ = client.Close() }() - require.NoError(t, client.Call("service.GetStatuses", in, out)) - } -} - -func list(address string, in *serviceProto.Service, out *serviceProto.List) func(t *testing.T) { - return func(t *testing.T) { - if in == nil { - in = &serviceProto.Service{} - } - client := newRPCClient(t, address) - defer func() { _ = client.Close() }() - require.NoError(t, client.Call("service.ListServices", in, out)) - } -} - -func workers(service string) func(t *testing.T) { - return func(t *testing.T) { - client := newRPCClient(t, serviceRPCAddr) - defer func() { _ = client.Close() }() - resp := &informerProto.WorkersList{} - require.NoError(t, client.Call("informer.GetWorkers", &informerProto.GetWorkersRequest{Plugin: service}, resp)) - require.Len(t, resp.GetWorkers(), 20) - } -} diff --git a/tests/workers_test.go b/tests/workers_test.go new file mode 100644 index 0000000..8f575af --- /dev/null +++ b/tests/workers_test.go @@ -0,0 +1,33 @@ +package tests + +import ( + "testing" + + "tests/helpers" + + "github.com/roadrunner-server/informer/v6" + rpcPlugin "github.com/roadrunner-server/rpc/v6" + "github.com/roadrunner-server/service/v6" + "github.com/stretchr/testify/require" +) + +func TestServiceInformerWorkers(t *testing.T) { + const addr = "127.0.0.1:6315" + + rr, stop := helpers.Start(t, "configs/.rr-service-workers.yaml", + []any{&service.Plugin{}, &informer.Plugin{}, &rpcPlugin.Plugin{}}, + helpers.WithServicesStarted(20), helpers.WithTCPProbe(addr)) + + states := helpers.Workers(t, helpers.RPC(t, addr), "service") + require.Len(t, states, 20) + + for _, st := range states { + require.NotZero(t, st.Pid) + require.NotZero(t, st.MemoryUsage) + require.Contains(t, st.Command, "php") + } + + stop() + + require.Empty(t, rr.Errs()) +}