Skip to content

Commit bd36df8

Browse files
ibihimibihim
committed
pkg/operator/encryption/kms/health: add KMS plugin health checker
Add Checker, which dials each co-located KMSv2 plugin's UDS Status endpoint and reports per-plugin health, and wire it into the reporter's probe loop. Naming follows openshift/api#2881 (the structured KMS health API) rather than the enhancement proposal, since the API PR is the newer and authoritative source: the type is PluginHealthReport (not Condition) to match KMSPluginHealthReport. Go-idiomatic field names (KeyID/KEKID) are kept; the writer will map them onto the API's KeyId/KEKId. keyIDFromSocket reuses the validation regex's capture group as the single source of truth for the socket shape, and the probe loop drops the redundant timeout context since each Status RPC already enforces ReadTimeout internally.
1 parent bae2d2e commit bd36df8

3 files changed

Lines changed: 281 additions & 8 deletions

File tree

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

Lines changed: 71 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -10,14 +10,29 @@ import (
1010
"github.com/spf13/cobra"
1111
"github.com/spf13/pflag"
1212

13+
"k8s.io/apimachinery/pkg/util/wait"
14+
"k8s.io/apiserver/pkg/server"
15+
k8senvelopekmsv2 "k8s.io/apiserver/pkg/storage/value/encrypt/envelope/kmsv2"
1316
"k8s.io/client-go/rest"
1417
"k8s.io/client-go/tools/clientcmd"
1518
"k8s.io/klog/v2"
1619
)
1720

21+
const providerName = "kms-health-reporter"
22+
1823
// kmsSocketPattern matches the socket path each co-located KMSv2 plugin is
1924
// mounted at, e.g. unix:///var/run/kmsplugin/kms-1.sock.
20-
var kmsSocketPattern = regexp.MustCompile(`^unix:///var/run/kmsplugin/kms-\d+\.sock$`)
25+
var kmsSocketPattern = regexp.MustCompile(`^unix:///var/run/kmsplugin/kms-(\d+)\.sock$`)
26+
27+
// keyIDFromSocket extracts the sequential key id captured by kmsSocketPattern,
28+
// e.g. "1" from unix:///var/run/kmsplugin/kms-1.sock.
29+
func keyIDFromSocket(socket string) (string, error) {
30+
m := kmsSocketPattern.FindStringSubmatch(socket)
31+
if m == nil {
32+
return "", fmt.Errorf("socket %q must match %s", socket, kmsSocketPattern)
33+
}
34+
return m[1], nil
35+
}
2136

2237
// options' flag-bound fields are exported so the struct can be logged as a
2338
// whole via klog.InfoS, which JSON-marshals its values.
@@ -45,7 +60,7 @@ func NewCommand(ctx context.Context, newOperatorClient func(*rest.Config) (v1hel
4560
if err := o.validate(); err != nil {
4661
return err
4762
}
48-
return o.run()
63+
return o.run(ctx)
4964
},
5065
}
5166
o.addFlags(cmd.Flags())
@@ -87,8 +102,12 @@ func (o *options) validate() error {
87102
return nil
88103
}
89104

90-
func (o *options) run() error {
91-
cfg, err := buildRESTConfig(o.Kubeconfig)
105+
func (o *options) run(ctx context.Context) error {
106+
ctx = setupSignalContext(ctx)
107+
108+
// Empty kubeconfig falls back to the in-cluster config (service account
109+
// token + KUBERNETES_SERVICE_HOST), which is the deployed path.
110+
cfg, err := clientcmd.BuildConfigFromFlags("", o.Kubeconfig)
92111
if err != nil {
93112
return fmt.Errorf("build rest config: %w", err)
94113
}
@@ -97,14 +116,58 @@ func (o *options) run() error {
97116
return fmt.Errorf("build operator client: %w", err)
98117
}
99118

119+
plugins, err := buildPlugins(ctx, o.KMSSockets, o.ReadTimeout)
120+
if err != nil {
121+
return err
122+
}
123+
prober := newProber(plugins)
124+
100125
klog.InfoS("kms-health-reporter starting", "config", o)
101126

127+
wait.JitterUntilWithContext(ctx, func(ctx context.Context) {
128+
// Each Status RPC enforces o.ReadTimeout internally (set at dial time);
129+
// ctx here only carries shutdown cancellation.
130+
conditions := prober.probeAll(ctx)
131+
// TODO: hand conditions to the writer once it lands; logging is a placeholder.
132+
klog.InfoS("kms plugin health", "conditions", conditions)
133+
}, o.Interval, 0.1, false)
134+
102135
return nil
103136
}
104137

105-
func buildRESTConfig(kubeconfig string) (*rest.Config, error) {
106-
if kubeconfig != "" {
107-
return clientcmd.BuildConfigFromFlags("", kubeconfig)
138+
func buildPlugins(ctx context.Context, sockets []string, timeout time.Duration) ([]pluginClient, error) {
139+
plugins := make([]pluginClient, 0, len(sockets))
140+
141+
for _, socket := range sockets {
142+
keyID, err := keyIDFromSocket(socket)
143+
if err != nil {
144+
return nil, err
145+
}
146+
147+
// Unique name per plugin so the gRPC client's KMS operation metrics
148+
// don't merge both plugins into one series.
149+
service, err := k8senvelopekmsv2.NewGRPCService(ctx, socket, providerName+"-"+keyID, timeout)
150+
if err != nil {
151+
// With the current dependency version this should never happen with a validated GRPC endpoint.
152+
return nil, fmt.Errorf("setting up grpc service failed at %q: %w", socket, err)
153+
}
154+
155+
plugins = append(plugins, pluginClient{keyID: keyID, service: service})
108156
}
109-
return rest.InClusterConfig()
157+
158+
return plugins, nil
159+
}
160+
161+
// setupSignalContext registers for SIGTERM and SIGINT and returns a context
162+
// that will be cancelled once a signal is received. Compare startupmonitor's
163+
// setupSignalContext.
164+
func setupSignalContext(baseCtx context.Context) context.Context {
165+
shutdownCtx, cancel := context.WithCancel(baseCtx)
166+
shutdownHandler := server.SetupSignalHandler()
167+
go func() {
168+
defer cancel()
169+
<-shutdownHandler
170+
klog.Infof("Received SIGTERM or SIGINT signal, shutting down the process.")
171+
}()
172+
return shutdownCtx
110173
}

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

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
package health
2+
3+
import (
4+
"context"
5+
"sync"
6+
"time"
7+
8+
kmsservice "k8s.io/kms/pkg/service"
9+
)
10+
11+
// healthzOK is the value the KMS plugin returns when healthy.
12+
// See https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/kms/apis/v2/api.proto#L39
13+
const healthzOK = "ok"
14+
15+
const (
16+
statusHealthy = "healthy"
17+
statusUnhealthy = "unhealthy"
18+
statusError = "error"
19+
)
20+
21+
type pluginHealthReport struct {
22+
// KeyID is the controller's sequential key id; KEKID is the KMS provider's
23+
// encryption key id. Distinct identifiers, easy to confuse.
24+
KeyID string
25+
KEKID string
26+
Status string
27+
LastChecked time.Time
28+
Detail string
29+
}
30+
31+
// pluginClient is the dialed handle to one co-located KMS plugin; the plugin
32+
// itself is a separate process behind the unix socket.
33+
type pluginClient struct {
34+
keyID string
35+
service kmsservice.Service
36+
}
37+
38+
type prober struct {
39+
plugins []pluginClient
40+
now func() time.Time
41+
}
42+
43+
func newProber(plugins []pluginClient) *prober {
44+
return &prober{
45+
plugins: plugins,
46+
now: time.Now,
47+
}
48+
}
49+
50+
// probeAll never returns an error: a failed probe is encoded as a report
51+
// with Status "error" so the caller always gets one entry per plugin.
52+
// Probes run concurrently so one hung plugin doesn't delay the others;
53+
// worst-case duration is one read-timeout, not the sum.
54+
func (p *prober) probeAll(ctx context.Context) []pluginHealthReport {
55+
reports := make([]pluginHealthReport, len(p.plugins))
56+
57+
var wg sync.WaitGroup
58+
for i, plugin := range p.plugins {
59+
wg.Go(func() {
60+
reports[i] = p.probe(ctx, plugin)
61+
})
62+
}
63+
wg.Wait()
64+
65+
return reports
66+
}
67+
68+
func (p *prober) probe(ctx context.Context, plugin pluginClient) pluginHealthReport {
69+
report := pluginHealthReport{
70+
KeyID: plugin.keyID,
71+
LastChecked: p.now(),
72+
}
73+
74+
resp, err := plugin.service.Status(ctx)
75+
switch {
76+
case err != nil:
77+
report.Status = statusError
78+
report.Detail = err.Error()
79+
case resp.Healthz == healthzOK:
80+
report.Status = statusHealthy
81+
report.KEKID = resp.KeyID
82+
default:
83+
report.Status = statusUnhealthy
84+
report.Detail = resp.Healthz
85+
}
86+
87+
return report
88+
}

‎pkg/operator/encryption/kms/health/prober_test.go‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,122 @@
1+
package health
2+
3+
import (
4+
"context"
5+
"fmt"
6+
"reflect"
7+
"strconv"
8+
"sync"
9+
"testing"
10+
"time"
11+
12+
"k8s.io/apimachinery/pkg/util/wait"
13+
kmsservice "k8s.io/kms/pkg/service"
14+
)
15+
16+
type fakeService struct {
17+
resp *kmsservice.StatusResponse
18+
err error
19+
}
20+
21+
func (f *fakeService) Status(context.Context) (*kmsservice.StatusResponse, error) {
22+
return f.resp, f.err
23+
}
24+
func (f *fakeService) Encrypt(context.Context, string, []byte) (*kmsservice.EncryptResponse, error) {
25+
return nil, nil
26+
}
27+
func (f *fakeService) Decrypt(context.Context, string, *kmsservice.DecryptRequest) ([]byte, error) {
28+
return nil, nil
29+
}
30+
31+
func TestProber_ProbeAll(t *testing.T) {
32+
fixed := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
33+
p := &prober{
34+
plugins: []pluginClient{
35+
{keyID: "1", service: &fakeService{resp: &kmsservice.StatusResponse{Healthz: "ok", KeyID: "kek-abc"}}},
36+
{keyID: "2", service: &fakeService{err: fmt.Errorf("connection refused")}},
37+
{keyID: "3", service: &fakeService{resp: &kmsservice.StatusResponse{Healthz: "degraded"}}},
38+
},
39+
now: func() time.Time { return fixed },
40+
}
41+
42+
have := p.probeAll(context.Background())
43+
want := []pluginHealthReport{
44+
{KeyID: "1", KEKID: "kek-abc", Status: "healthy", LastChecked: fixed},
45+
{KeyID: "2", Status: "error", Detail: "connection refused", LastChecked: fixed},
46+
{KeyID: "3", Status: "unhealthy", Detail: "degraded", LastChecked: fixed},
47+
}
48+
if !reflect.DeepEqual(have, want) {
49+
t.Errorf("probeAll():\n have: %+v\n want: %+v", have, want)
50+
}
51+
}
52+
53+
// blockingService releases Status only once all expected probes have
54+
// arrived, so the test passes only if probeAll runs them concurrently.
55+
type blockingService struct {
56+
*fakeService
57+
barrier *sync.WaitGroup
58+
}
59+
60+
func (b *blockingService) Status(ctx context.Context) (*kmsservice.StatusResponse, error) {
61+
b.barrier.Done()
62+
b.barrier.Wait()
63+
return b.fakeService.Status(ctx)
64+
}
65+
66+
func TestProber_ProbeAllFansOut(t *testing.T) {
67+
const n = 3
68+
var barrier sync.WaitGroup
69+
barrier.Add(n)
70+
71+
plugins := make([]pluginClient, 0, n)
72+
for i := range n {
73+
keyID := strconv.Itoa(i + 1)
74+
plugins = append(plugins, pluginClient{
75+
keyID: keyID,
76+
service: &blockingService{
77+
fakeService: &fakeService{resp: &kmsservice.StatusResponse{Healthz: "ok", KeyID: "kek-" + keyID}},
78+
barrier: &barrier,
79+
},
80+
})
81+
}
82+
p := newProber(plugins)
83+
84+
done := make(chan []pluginHealthReport, 1)
85+
go func() { done <- p.probeAll(context.Background()) }()
86+
87+
select {
88+
case have := <-done:
89+
for i, report := range have {
90+
want := strconv.Itoa(i + 1)
91+
if report.KeyID != want || report.KEKID != "kek-"+want {
92+
t.Errorf("reports[%d] = {KeyID:%q KEKID:%q}, want {KeyID:%q KEKID:%q}",
93+
i, report.KeyID, report.KEKID, want, "kek-"+want)
94+
}
95+
}
96+
case <-time.After(wait.ForeverTestTimeout):
97+
t.Fatal("probeAll timed out: probes ran sequentially or deadlocked")
98+
}
99+
}
100+
101+
func Test_keyIDFromSocket(t *testing.T) {
102+
tests := []struct {
103+
socket string
104+
want string
105+
wantErr bool
106+
}{
107+
{socket: "unix:///var/run/kmsplugin/kms-1.sock", want: "1"},
108+
{socket: "unix:///var/run/kmsplugin/kms-42.sock", want: "42"},
109+
{socket: "unix:///var/run/kmsplugin/plugin.sock", wantErr: true},
110+
}
111+
for _, tt := range tests {
112+
t.Run(tt.socket, func(t *testing.T) {
113+
have, err := keyIDFromSocket(tt.socket)
114+
if (err != nil) != tt.wantErr {
115+
t.Fatalf("keyIDFromSocket(%q) err = %v, wantErr %v", tt.socket, err, tt.wantErr)
116+
}
117+
if have != tt.want {
118+
t.Errorf("keyIDFromSocket(%q) = %q, want %q", tt.socket, have, tt.want)
119+
}
120+
})
121+
}
122+
}

0 commit comments

Comments
 (0)