From 8575114ba4bd74e90f2c4e5264aac531896aa447 Mon Sep 17 00:00:00 2001 From: Vipul Subhash Pandey Date: Mon, 3 Aug 2026 18:26:20 +0530 Subject: [PATCH] Detect when a plugin process exits and reflect it in /healthz Piped launches each plugin as a separate process and never checks on it again until its own shutdown. If a plugin process dies on its own (crash, OOM kill, etc.), piped keeps running, /healthz keeps saying ok, and every deployment needing that plugin fails until someone notices and restarts the pod by hand. lifecycle.Command already tracks process exit internally through stoppedCh, it was just never exposed or watched outside of shutdown. This adds Done()/Err() accessors to read that signal, watches each plugin for an unexpected exit alongside the existing shutdown path, and lets /healthz report unhealthy plugins so the livenessProbe and readinessProbe already pointed at /healthz in the shipped manifests can actually restart the pod. Signed-off-by: Vipul Subhash Pandey --- pkg/app/pipedv1/cmd/piped/piped.go | 48 +++++++++----- pkg/app/pipedv1/cmd/piped/pluginhealth.go | 59 ++++++++++++++++++ .../pipedv1/cmd/piped/pluginhealth_test.go | 62 +++++++++++++++++++ pkg/lifecycle/binary.go | 16 +++++ pkg/lifecycle/binary_test.go | 42 +++++++++++++ 5 files changed, 212 insertions(+), 15 deletions(-) create mode 100644 pkg/app/pipedv1/cmd/piped/pluginhealth.go create mode 100644 pkg/app/pipedv1/cmd/piped/pluginhealth_test.go 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()