diff --git a/pkg/app/pipedv1/cmd/piped/piped.go b/pkg/app/pipedv1/cmd/piped/piped.go index eb10091c82..130fffed7d 100644 --- a/pkg/app/pipedv1/cmd/piped/piped.go +++ b/pkg/app/pipedv1/cmd/piped/piped.go @@ -30,7 +30,6 @@ import ( "path/filepath" "strconv" "strings" - "sync" "time" secretmanager "cloud.google.com/go/secretmanager/apiv1" @@ -217,6 +216,10 @@ func (p *piped) run(ctx context.Context, input cli.Input) (runErr error) { return notifier.Run(ctx) }) + // Tracks plugins whose process has exited on its own, so that /healthz + // below can reflect it instead of always reporting "ok". + pluginHealth := newPluginHealth() + // Start running admin server. { var ( @@ -228,6 +231,11 @@ func (p *piped) run(ctx context.Context, input cli.Input) (runErr error) { w.Write(ver) }) admin.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { + if names := pluginHealth.UnhealthyNames(); len(names) > 0 { + w.WriteHeader(http.StatusServiceUnavailable) + fmt.Fprintf(w, "unhealthy plugins: %s", strings.Join(names, ", ")) + return + } w.Write([]byte("ok")) }) admin.Handle("/metrics", input.PrometheusMetricsHandlerFor(registry)) @@ -336,28 +344,38 @@ func (p *piped) run(ctx context.Context, input cli.Input) (runErr error) { // Start plugins that registered in the configuration. { - // Start all plugins and keep their commands to stop them later. + // Start all plugins and keep their commands to stop or watch them later. plugins, err := p.runPlugins(ctx, cfg.Plugins, input.Logger) if err != nil { input.Logger.Error("failed to run plugins", zap.Error(err)) return err } - group.Go(func() error { - <-ctx.Done() - wg := &sync.WaitGroup{} - for _, plg := range plugins { - wg.Add(1) - go func() { - defer wg.Done() + // Each plugin gets its own goroutine that either stops it when piped + // is shutting down, or notices when the plugin's process has exited + // by itself (crash, OOM kill, etc.) and reports it through + // pluginHealth instead of leaving it unnoticed. + for i, plg := range plugins { + name := cfg.Plugins[i].Name + plg := plg + group.Go(func() error { + select { + case <-ctx.Done(): if err := plg.GracefulStop(p.gracePeriod); err != nil { - input.Logger.Error("failed to stop plugin", zap.Error(err)) + input.Logger.Error("failed to stop plugin", zap.String("plugin", name), zap.Error(err)) } - }() - } - wg.Wait() - return nil - }) + return nil + case <-plg.Done(): + if ctx.Err() != nil { + // piped is already shutting down, this is expected. + return nil + } + input.Logger.Error("plugin exited unexpectedly", zap.String("plugin", name), zap.Error(plg.Err())) + pluginHealth.MarkUnhealthy(name, plg.Err()) + return nil + } + }) + } } // Make grpc clients to connect to plugins. diff --git a/pkg/app/pipedv1/cmd/piped/pluginhealth.go b/pkg/app/pipedv1/cmd/piped/pluginhealth.go new file mode 100644 index 0000000000..5b72c2f296 --- /dev/null +++ b/pkg/app/pipedv1/cmd/piped/pluginhealth.go @@ -0,0 +1,59 @@ +// Copyright 2024 The PipeCD Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package piped + +import "sync" + +// pluginHealth keeps track of plugins whose process has exited on its own, +// outside of piped's own shutdown. The admin /healthz handler reads this +// so that a plugin crashing can actually be observed from the outside, +// instead of piped silently staying "ok" forever. +type pluginHealth struct { + mu sync.Mutex + unhealthy map[string]error +} + +func newPluginHealth() *pluginHealth { + return &pluginHealth{ + unhealthy: make(map[string]error), + } +} + +// MarkUnhealthy records that the named plugin's process has exited +// unexpectedly. +func (h *pluginHealth) MarkUnhealthy(name string, err error) { + h.mu.Lock() + defer h.mu.Unlock() + h.unhealthy[name] = err +} + +// Healthy reports whether every known plugin is still considered alive. +func (h *pluginHealth) Healthy() bool { + h.mu.Lock() + defer h.mu.Unlock() + return len(h.unhealthy) == 0 +} + +// UnhealthyNames returns the names of plugins currently marked unhealthy, +// in no particular order. +func (h *pluginHealth) UnhealthyNames() []string { + h.mu.Lock() + defer h.mu.Unlock() + names := make([]string, 0, len(h.unhealthy)) + for name := range h.unhealthy { + names = append(names, name) + } + return names +} diff --git a/pkg/app/pipedv1/cmd/piped/pluginhealth_test.go b/pkg/app/pipedv1/cmd/piped/pluginhealth_test.go new file mode 100644 index 0000000000..f67a41a485 --- /dev/null +++ b/pkg/app/pipedv1/cmd/piped/pluginhealth_test.go @@ -0,0 +1,62 @@ +// Copyright 2024 The PipeCD Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package piped + +import ( + "errors" + "sync" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestPluginHealth(t *testing.T) { + h := newPluginHealth() + + assert.True(t, h.Healthy()) + assert.Empty(t, h.UnhealthyNames()) + + h.MarkUnhealthy("kubernetes", errors.New("exit status 1")) + + assert.False(t, h.Healthy()) + assert.Equal(t, []string{"kubernetes"}, h.UnhealthyNames()) + + // Marking the same plugin again must not create a duplicate entry. + h.MarkUnhealthy("kubernetes", errors.New("exit status 2")) + assert.Equal(t, []string{"kubernetes"}, h.UnhealthyNames()) + + h.MarkUnhealthy("terraform", errors.New("killed")) + assert.False(t, h.Healthy()) + assert.ElementsMatch(t, []string{"kubernetes", "terraform"}, h.UnhealthyNames()) +} + +func TestPluginHealthConcurrentAccess(t *testing.T) { + h := newPluginHealth() + + var wg sync.WaitGroup + for i := 0; i < 50; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + h.MarkUnhealthy("plugin", errors.New("boom")) + h.Healthy() + h.UnhealthyNames() + }(i) + } + wg.Wait() + + assert.False(t, h.Healthy()) + assert.Equal(t, []string{"plugin"}, h.UnhealthyNames()) +} diff --git a/pkg/lifecycle/binary.go b/pkg/lifecycle/binary.go index 5986a8361a..3e381a033c 100644 --- a/pkg/lifecycle/binary.go +++ b/pkg/lifecycle/binary.go @@ -51,6 +51,22 @@ func (c *Command) IsRunning() bool { } } +// Done returns a channel that is closed once the process has exited, +// no matter whether it was stopped on purpose (GracefulStop) or it +// exited/crashed on its own. +func (c *Command) Done() <-chan struct{} { + return c.stoppedCh +} + +// Err returns the error the process exited with, if any. It is only +// meaningful after the channel returned by Done has been closed. +func (c *Command) Err() error { + if perr := c.result.Load(); perr != nil { + return *perr + } + return nil +} + func (c *Command) GracefulStop(period time.Duration) error { // For graceful shutdown, we send SIGTERM signal to old Piped process // and wait grace-period of time before force killing it. diff --git a/pkg/lifecycle/binary_test.go b/pkg/lifecycle/binary_test.go index 572d644eb8..422024b42a 100644 --- a/pkg/lifecycle/binary_test.go +++ b/pkg/lifecycle/binary_test.go @@ -88,6 +88,48 @@ func TestGracefulStopCommandResult(t *testing.T) { } } +func TestCommandDone(t *testing.T) { + t.Run("Done is closed when the process exits on its own", func(t *testing.T) { + cmd, err := RunBinary(context.TODO(), "sh", []string{"-c", "exit 0"}) + require.NoError(t, err) + require.NotNil(t, cmd) + + select { + case <-cmd.Done(): + case <-time.After(time.Second): + t.Fatal("Done was not closed after the process exited") + } + assert.NoError(t, cmd.Err()) + }) + + t.Run("Err reflects a non-zero exit", func(t *testing.T) { + cmd, err := RunBinary(context.TODO(), "sh", []string{"-c", "exit 1"}) + require.NoError(t, err) + require.NotNil(t, cmd) + + select { + case <-cmd.Done(): + case <-time.After(time.Second): + t.Fatal("Done was not closed after the process exited") + } + assert.Error(t, cmd.Err()) + }) + + t.Run("Done is already closed once GracefulStop returns", func(t *testing.T) { + cmd, err := RunBinary(context.TODO(), "sh", []string{"/bin/sleep", "1m"}) + require.NoError(t, err) + require.NotNil(t, cmd) + + cmd.GracefulStop(time.Nanosecond) + + select { + case <-cmd.Done(): + default: + t.Fatal("Done should already be closed once GracefulStop returns") + } + }) +} + func TestDownloadBinary(t *testing.T) { server := httpTestServer() defer server.Close()