Merge pull request #2318 from ibihim/CNTRLPLANE-3234-health-reporter-writer

CNTRLPLANE-3234: health reporter writer

Author: openshift-merge-bot[bot]

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

@@ -6,50 +6,46 @@ import (
 	"regexp"
 	"time"
 
-	"github.com/openshift/library-go/pkg/operator/v1helpers"
 	"github.com/spf13/cobra"
-	"github.com/spf13/pflag"
 
-	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apiserver/pkg/server"
 	k8senvelopekmsv2 "k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2"
 	"k8s.io/client-go/rest"
-	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog/v2"
+
+	applyoperatorv1 "github.com/openshift/client-go/operator/applyconfigurations/operator/v1"
 )
 
-const Subcommand = "kms-health-reporter"
+const (
+	Subcommand = "kms-health-reporter"
+)
 
 // kmsSocketPattern matches the socket path each co-located KMSv2 plugin is
 // mounted at, e.g. unix:///var/run/kmsplugin/kms-1.sock.
 var kmsSocketPattern = regexp.MustCompile(`^unix:///var/run/kmsplugin/kms-(\d+)\.sock$`)
 
-// 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 {
-	KMSSockets   []string
-	Interval     time.Duration
-	ReadTimeout  time.Duration
-	WriteTimeout time.Duration
-	NodeName     string
-	Kubeconfig   string
-
-	newOperatorClient func(*rest.Config) (v1helpers.OperatorClient, error)
-}
+// NewEncryptionStatusWriterFunc builds the EncryptionStatusWriter for a target
+// apiserver operator status CR. fieldManager sets the owner in the
+// managedFields when doing SSA.
+type NewEncryptionStatusWriterFunc func(restConfig *rest.Config, fieldManager string) (EncryptionStatusWriter, error)
+
+// EncryptionStatusWriter is capable of applying the
+// KMSEncryptionStatusApplyConfiguration at the correct place in the operator's
+// status.
+type EncryptionStatusWriter func(ctx context.Context, status *applyoperatorv1.KMSEncryptionStatusApplyConfiguration) error
 
 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{
@@ -78,80 +74,17 @@ func NewCommand(ctx context.Context, newOperatorClient func(*rest.Config) (v1hel
 	return cmd
 }
 
-func (o *options) addFlags(fs *pflag.FlagSet) {
-	fs.StringSliceVar(&o.KMSSockets, "kms-sockets", nil, "KMS plugin endpoints in unix:// URI format (e.g. unix:///var/run/kmsplugin/kms-1.sock)")
-	fs.DurationVar(&o.Interval, "interval", 30*time.Second, "cadence between probe+emit cycles")
-	fs.DurationVar(&o.ReadTimeout, "read-timeout", 5*time.Second, "deadline for each Status RPC")
-	fs.DurationVar(&o.WriteTimeout, "write-timeout", 10*time.Second, "deadline for each condition update")
-	fs.StringVar(&o.NodeName, "node-name", "", "node name recorded in the condition used to help to identify the origin")
-	fs.StringVar(&o.Kubeconfig, "kubeconfig", "", "path to a kubeconfig; empty uses in-cluster config")
-}
-
-func (o *options) validate() error {
-	if len(o.KMSSockets) == 0 {
-		return fmt.Errorf("--kms-sockets is required, at least one")
-	}
-	socketSet := sets.New[string]()
-	for _, s := range o.KMSSockets {
-		if !kmsSocketPattern.MatchString(s) {
-			return fmt.Errorf("--kms-sockets entry %q must match %s", s, kmsSocketPattern)
-		}
-		if socketSet.Has(s) {
-			return fmt.Errorf("--kms-sockets entry %q is duplicated", s)
-		}
-		socketSet.Insert(s)
-	}
-
-	if o.Interval <= 0 {
-		return fmt.Errorf("--interval must be positive")
-	}
-	if o.ReadTimeout <= 0 {
-		return fmt.Errorf("--read-timeout must be positive")
-	}
-	if o.WriteTimeout <= 0 {
-		return fmt.Errorf("--write-timeout must be positive")
-	}
-	if o.NodeName == "" {
-		return fmt.Errorf("--node-name is required")
-	}
-
-	return nil
-}
-
-func (o *options) Config(ctx context.Context) (*Config, error) {
-	// Empty kubeconfig falls back to the in-cluster config (service account
-	// token + KUBERNETES_SERVICE_HOST), which is the deployed path.
-	restCfg, err := clientcmd.BuildConfigFromFlags("", o.Kubeconfig)
-	if err != nil {
-		return nil, fmt.Errorf("build rest config: %w", err)
-	}
-
-	operatorClient, err := o.newOperatorClient(restCfg)
-	if err != nil {
-		return nil, fmt.Errorf("build operator client: %w", err)
-	}
-
-	plugins, err := buildPlugins(ctx, o.KMSSockets, o.ReadTimeout)
-	if err != nil {
-		return nil, err
-	}
-
-	return &Config{
-		operatorClient: operatorClient,
-		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, reports); err != nil {
+			klog.ErrorS(err, "failed to publish kms plugin health")
+		}
 	}, c.interval, 0.1, false)
 
 	return nil
@@ -168,7 +101,7 @@ func buildPlugins(ctx context.Context, sockets []string, timeout time.Duration)
 
 		// Unique name per plugin so the gRPC client's KMS operation metrics
 		// don't merge both plugins into one series.
-		service, err := k8senvelopekmsv2.NewGRPCService(ctx, socket, Subcommand+"-"+keyID, timeout)
+		service, err := k8senvelopekmsv2.NewGRPCService(ctx, socket, Subcommand, timeout)
 		if err != nil {
 			// With the current dependency version this should never happen with a validated GRPC endpoint.
 			return nil, fmt.Errorf("setting up grpc service failed at %q: %w", socket, err)

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

@@ -1,121 +1,43 @@
 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
-// mutates a single field so the failure under test is unambiguous.
-func validOptions() *options {
-	return &options{
-		KMSSockets:   []string{"unix:///var/run/kmsplugin/kms-1.sock"},
-		Interval:     30 * time.Second,
-		ReadTimeout:  5 * time.Second,
-		WriteTimeout: 10 * time.Second,
-		NodeName:     "node-1",
-	}
-}
+// 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.WithTimeout(context.Background(), 5*time.Second)
+	t.Cleanup(cancel)
 
-func TestValidate(t *testing.T) {
-	tests := []struct {
-		name    string
-		mutate  func(*options)
-		wantErr bool
-	}{
-		{
-			name:   "valid",
-			mutate: func(*options) {},
-		},
-		{
-			name:    "no sockets",
-			mutate:  func(o *options) { o.KMSSockets = nil },
-			wantErr: true,
-		},
-		{
-			name:    "empty socket entry",
-			mutate:  func(o *options) { o.KMSSockets = []string{""} },
-			wantErr: true,
-		},
-		{
-			name:   "multiple valid sockets",
-			mutate: func(o *options) { o.KMSSockets = append(o.KMSSockets, "unix:///var/run/kmsplugin/kms-2.sock") },
-		},
-		{
-			name:    "duplicate sockets",
-			mutate:  func(o *options) { o.KMSSockets = append(o.KMSSockets, o.KMSSockets[0]) },
-			wantErr: true,
-		},
-		{
-			name:    "socket missing unix scheme",
-			mutate:  func(o *options) { o.KMSSockets = []string{"/var/run/kmsplugin/kms-1.sock"} },
-			wantErr: true,
-		},
-		{
-			name:    "socket scheme without path",
-			mutate:  func(o *options) { o.KMSSockets = []string{"unix://"} },
-			wantErr: true,
-		},
-		{
-			name:    "socket wrong directory",
-			mutate:  func(o *options) { o.KMSSockets = []string{"unix:///tmp/kms-1.sock"} },
-			wantErr: true,
-		},
-		{
-			name:    "socket non-numeric index",
-			mutate:  func(o *options) { o.KMSSockets = []string{"unix:///var/run/kmsplugin/kms-x.sock"} },
-			wantErr: true,
-		},
-		{
-			name:    "socket missing .sock suffix",
-			mutate:  func(o *options) { o.KMSSockets = []string{"unix:///var/run/kmsplugin/kms-1"} },
-			wantErr: true,
-		},
-		{
-			name:    "socket with surrounding whitespace",
-			mutate:  func(o *options) { o.KMSSockets = []string{" unix:///var/run/kmsplugin/kms-1.sock "} },
-			wantErr: true,
-		},
-		{
-			name:    "interval zero",
-			mutate:  func(o *options) { o.Interval = 0 },
-			wantErr: true,
-		},
-		{
-			name:    "interval negative",
-			mutate:  func(o *options) { o.Interval = -time.Second },
-			wantErr: true,
-		},
-		{
-			name:    "read timeout zero",
-			mutate:  func(o *options) { o.ReadTimeout = 0 },
-			wantErr: true,
-		},
-		{
-			name:    "write timeout zero",
-			mutate:  func(o *options) { o.WriteTimeout = 0 },
-			wantErr: true,
-		},
-		{
-			name:    "node name empty",
-			mutate:  func(o *options) { o.NodeName = "" },
-			wantErr: true,
+	var have *applyoperatorv1.KMSEncryptionStatusApplyConfiguration
+	c := &Config{
+		interval:     time.Hour, // never reached; cancelled after the first tick
+		writeTimeout: time.Second,
+		prober: &prober{
+			nodeName: "node-1",
+			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
 		},
 	}
 
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			o := validOptions()
-			tc.mutate(o)
-
-			err := o.validate()
-			if tc.wantErr {
-				require.Error(t, err)
-			} else {
-				require.NoError(t, err)
-			}
-		})
-	}
+	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)
 }