kms/health: add report writer

Author: ibihim

pkg/operator/encryption/kms/health/cmd.go

@@ -6,7 +6,6 @@ import (
 	"regexp"
 	"time"
 
-	"github.com/openshift/library-go/pkg/operator/v1helpers"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 
@@ -25,6 +24,13 @@ const providerName = "kms-health-reporter"
 // mounted at, e.g. unix:///var/run/kmsplugin/kms-1.sock.
 var kmsSocketPattern = regexp.MustCompile(`^unix:///var/run/kmsplugin/kms-(\d+)\.sock$`)
 
+// NewEncryptionStatusWriterFunc builds the EncryptionStatusWriter for a target
+// CR. The caller binds the concrete CR; this package only supplies the rest
+// config and a reporterID. reporterID is a stable, per-node identifier the write
+// path uses to claim ownership of its own entries; with server-side apply, pass
+// it as the field manager.
+type NewEncryptionStatusWriterFunc func(restConfig *rest.Config, reporterID string) (EncryptionStatusWriter, error)
+
 // options' flag-bound fields are exported so the struct can be logged as a
 // whole via klog.InfoS, which JSON-marshals its values.
 type options struct {
@@ -35,21 +41,21 @@ type options struct {
 	NodeName     string
 	Kubeconfig   string
 
-	newOperatorClient func(*rest.Config) (v1helpers.OperatorClient, error)
+	newWriter NewEncryptionStatusWriterFunc
 }
 
 type Config struct {
-	operatorClient v1helpers.OperatorClient
-	prober         *prober
+	writeStatus EncryptionStatusWriter
+	prober      *prober
 
 	interval     time.Duration
 	writeTimeout time.Duration
 	nodeName     string
 }
 
-func NewCommand(ctx context.Context, newOperatorClient func(*rest.Config) (v1helpers.OperatorClient, error)) *cobra.Command {
+func NewCommand(ctx context.Context, newWriter NewEncryptionStatusWriterFunc) *cobra.Command {
 	o := &options{
-		newOperatorClient: newOperatorClient,
+		newWriter: newWriter,
 	}
 
 	cmd := &cobra.Command{
@@ -126,9 +132,12 @@ func (o *options) Config(ctx context.Context) (*Config, error) {
 		return nil, fmt.Errorf("build rest config: %w", err)
 	}
 
-	operatorClient, err := o.newOperatorClient(restCfg)
+	// reporterID is the per-node ownership identity. The naming convention lives
+	// here, not in the caller's writer, so all three operators stay uniform.
+	reporterID := "kms-health-reporter-" + o.NodeName
+	writeStatus, err := o.newWriter(restCfg, reporterID)
 	if err != nil {
-		return nil, fmt.Errorf("build operator client: %w", err)
+		return nil, fmt.Errorf("build encryption status writer: %w", err)
 	}
 
 	plugins, err := buildPlugins(ctx, o.KMSSockets, o.ReadTimeout)
@@ -137,21 +146,25 @@ func (o *options) Config(ctx context.Context) (*Config, error) {
 	}
 
 	return &Config{
-		operatorClient: operatorClient,
-		prober:         newProber(plugins),
-		interval:       o.Interval,
-		writeTimeout:   o.WriteTimeout,
-		nodeName:       o.NodeName,
+		writeStatus:  writeStatus,
+		prober:       newProber(plugins),
+		interval:     o.Interval,
+		writeTimeout: o.WriteTimeout,
+		nodeName:     o.NodeName,
 	}, nil
 }
 
 func (c *Config) Run(ctx context.Context) error {
 	wait.JitterUntilWithContext(ctx, func(ctx context.Context) {
 		// Each Status RPC enforces the read timeout internally (set at dial
 		// time); ctx here only carries shutdown cancellation.
-		conditions := c.prober.probeAll(ctx)
-		// TODO: hand conditions to the writer once it lands; logging is a placeholder.
-		klog.InfoS("kms plugin health", "conditions", conditions)
+		reports := c.prober.probeAll(ctx)
+
+		writeCtx, cancel := context.WithTimeout(ctx, c.writeTimeout)
+		defer cancel()
+		if err := c.writeStatus(writeCtx, buildEncryptionStatus(c.nodeName, reports)); err != nil {
+			klog.ErrorS(err, "failed to publish kms plugin health")
+		}
 	}, c.interval, 0.1, false)
 
 	return nil

pkg/operator/encryption/kms/health/cmd_test.go

@@ -1,10 +1,14 @@
 package health
 
 import (
+	"context"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/require"
+
+	applyoperatorv1 "github.com/openshift/client-go/operator/applyconfigurations/operator/v1"
+	kmsservice "k8s.io/kms/pkg/service"
 )
 
 // validOptions returns an options value that passes validate. Each test case
@@ -119,3 +123,33 @@ func TestValidate(t *testing.T) {
 		})
 	}
 }
+
+// TestRunReportsOnce checks the loop wiring: Run probes, builds the status, and
+// hands it to the reporter. The reporter cancels the context so the loop ends
+// after a single tick.
+func TestRunReportsOnce(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	var have *applyoperatorv1.KMSEncryptionStatusApplyConfiguration
+	c := &Config{
+		nodeName:     "node-1",
+		interval:     time.Hour, // never reached; cancelled after the first tick
+		writeTimeout: time.Second,
+		prober: &prober{
+			plugins: []pluginClient{
+				{keyID: "1", service: &fakeService{resp: &kmsservice.StatusResponse{Healthz: "ok", KeyID: "kek-abc"}}},
+			},
+			now: func() time.Time { return time.Unix(0, 0).UTC() },
+		},
+		writeStatus: func(_ context.Context, status *applyoperatorv1.KMSEncryptionStatusApplyConfiguration) error {
+			have = status
+			cancel()
+			return nil
+		},
+	}
+
+	require.NoError(t, c.Run(ctx))
+	require.Len(t, have.HealthReports, 1)
+	require.Equal(t, "node-1", *have.HealthReports[0].NodeName)
+	require.Equal(t, "1", *have.HealthReports[0].KeyId)
+}

pkg/operator/encryption/kms/health/prober.go

@@ -21,11 +21,11 @@ const (
 type pluginHealthReport struct {
 	// KeyID is the controller's sequential key id; KEKID is the KMS provider's
 	// encryption key id. Distinct identifiers, easy to confuse.
-	KeyID       string
-	KEKID       string
-	Status      string
-	LastChecked time.Time
-	Detail      string
+	KeyID       string    `json:"keyID"`
+	KEKID       string    `json:"kekID,omitempty"`
+	Status      string    `json:"status"`
+	LastChecked time.Time `json:"lastChecked"`
+	Detail      string    `json:"detail,omitempty"`
 }
 
 // pluginClient is the dialed handle to one co-located KMS plugin; the plugin

pkg/operator/encryption/kms/health/writer.go

@@ -0,0 +1,51 @@
+package health
+
+import (
+	"context"
+
+	operatorv1 "github.com/openshift/api/operator/v1"
+	applyoperatorv1 "github.com/openshift/client-go/operator/applyconfigurations/operator/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// EncryptionStatusWriter is injected by the operator. The operator knows where
+// to put it. Creating the status is the health reporter's concern.
+type EncryptionStatusWriter func(ctx context.Context, status *applyoperatorv1.KMSEncryptionStatusApplyConfiguration) error
+
+// buildEncryptionStatus builds the KMSEncryptionStatusApplyConfiguration to be
+// applied by the operator.
+func buildEncryptionStatus(nodeName string, reports []pluginHealthReport) *applyoperatorv1.KMSEncryptionStatusApplyConfiguration {
+	healthReports := make([]*applyoperatorv1.KMSPluginHealthReportApplyConfiguration, 0, len(reports))
+	for _, r := range reports {
+		hr := applyoperatorv1.KMSPluginHealthReport().
+			WithNodeName(nodeName).
+			WithKeyId(r.KeyID).
+			WithStatus(mapStatus(r.Status)).
+			WithLastCheckedTime(metav1.NewTime(r.LastChecked))
+
+		// kekId/detail have MinLength=1; setting "" would fail validation.
+		if r.KEKID != "" {
+			hr = hr.WithKEKId(r.KEKID)
+		}
+		if r.Detail != "" {
+			hr = hr.WithDetail(r.Detail)
+		}
+
+		healthReports = append(healthReports, hr)
+	}
+
+	return applyoperatorv1.KMSEncryptionStatus().WithHealthReports(healthReports...)
+}
+
+// mapStatus defaults to Error so an unknown value never becomes an empty,
+// invalid enum.
+func mapStatus(s string) operatorv1.KMSPluginHealthStatus {
+	switch s {
+	case statusHealthy:
+		return operatorv1.KMSPluginHealthStatusHealthy
+	case statusUnhealthy:
+		return operatorv1.KMSPluginHealthStatusUnhealthy
+	default:
+		return operatorv1.KMSPluginHealthStatusError
+	}
+}

pkg/operator/encryption/kms/health/writer_test.go

@@ -0,0 +1,67 @@
+package health
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	operatorv1 "github.com/openshift/api/operator/v1"
+	applyoperatorv1 "github.com/openshift/client-go/operator/applyconfigurations/operator/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestBuildEncryptionStatus(t *testing.T) {
+	// Fixed UTC time dodges Go's monotonic clock and timezone drift.
+	checked := time.Unix(0, 0).UTC()
+	reports := []pluginHealthReport{
+		{KeyID: "1", KEKID: "kek-abc", Status: statusHealthy, LastChecked: checked},
+		{KeyID: "2", Status: statusUnhealthy, Detail: "not ok", LastChecked: checked},
+		{KeyID: "3", Status: statusError, Detail: "DeadlineExceeded", LastChecked: checked},
+	}
+
+	have := buildEncryptionStatus("node-1", reports)
+
+	// Each entry stamps nodeName; kekId only on healthy, detail only on the
+	// unhealthy/error entries, status mapped to the API enum.
+	want := applyoperatorv1.KMSEncryptionStatus().WithHealthReports(
+		applyoperatorv1.KMSPluginHealthReport().
+			WithNodeName("node-1").
+			WithKeyId("1").
+			WithStatus(operatorv1.KMSPluginHealthStatusHealthy).
+			WithLastCheckedTime(metav1.NewTime(checked)).
+			WithKEKId("kek-abc"),
+		applyoperatorv1.KMSPluginHealthReport().
+			WithNodeName("node-1").
+			WithKeyId("2").
+			WithStatus(operatorv1.KMSPluginHealthStatusUnhealthy).
+			WithLastCheckedTime(metav1.NewTime(checked)).
+			WithDetail("not ok"),
+		applyoperatorv1.KMSPluginHealthReport().
+			WithNodeName("node-1").
+			WithKeyId("3").
+			WithStatus(operatorv1.KMSPluginHealthStatusError).
+			WithLastCheckedTime(metav1.NewTime(checked)).
+			WithDetail("DeadlineExceeded"),
+	)
+
+	require.Equal(t, want, have)
+}
+
+func TestMapStatus(t *testing.T) {
+	tests := []struct {
+		in   string
+		want operatorv1.KMSPluginHealthStatus
+	}{
+		{statusHealthy, operatorv1.KMSPluginHealthStatusHealthy},
+		{statusUnhealthy, operatorv1.KMSPluginHealthStatusUnhealthy},
+		{statusError, operatorv1.KMSPluginHealthStatusError},
+		{"unexpected", operatorv1.KMSPluginHealthStatusError},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.in, func(t *testing.T) {
+			require.Equal(t, tc.want, mapStatus(tc.in))
+		})
+	}
+}