Detect plugin process exit and reflect it in piped's /healthz

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.

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
+}

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())
+}

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.

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()