Skip to content

Commit bcda55d

Browse files
hexfusionhexfusion
committed
datalayer: hot-reload the scrape client certificate on rotation
The metrics/models scrape client loaded its TLS client certificate once, at data source construction, and pinned it into tls.Config.Certificates, so a rotated certificate broke mTLS until the EPP was restarted. Rotation was already handled in the serving direction -- the ext-proc server via common.CertReloader, the metrics server via the certwatcher that controller-runtime builds for CertDir -- but nothing covered the outbound scrape. Serve the client certificate through a controller-runtime certwatcher: wire GetClientCertificate to watcher.GetCertificate so new handshakes present the current cert, and close idle connections on the reload callback so pooled keep-alives re-handshake rather than lingering on the old cert until they idle out. The manager runs the watcher (Runtime.Start adds it), so its fsnotify watch and periodic re-read are active, and the source needs no reload logic of its own. The CA is read once. Part of #2045. Signed-off-by: Sam Batschelet <sbatsche@redhat.com>
1 parent 44828e2 commit bcda55d

8 files changed

Lines changed: 322 additions & 47 deletions

File tree

‎pkg/epp/datalayer/runtime.go‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ import (
2626
"github.com/go-logr/logr"
2727
"k8s.io/apimachinery/pkg/runtime/schema"
2828
ctrl "sigs.k8s.io/controller-runtime"
29+
"sigs.k8s.io/controller-runtime/pkg/certwatcher"
2930

3031
"github.com/llm-d/llm-d-router/pkg/common/observability/logging"
3132
fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer"
@@ -351,6 +352,9 @@ func (r *Runtime) findSourceByType(sourceType string, gvkFilter *schema.GroupVer
351352
// Start is called to enable the Runtime to start processing data collection. It wires
352353
// Kubernetes notifications into the manager.
353354
func (r *Runtime) Start(ctx context.Context, mgr ctrl.Manager) error {
355+
if err := r.addCertWatchers(mgr); err != nil {
356+
return err
357+
}
354358
return r.notification.ForEach(func(srcName string, src fwkdl.NotificationSource) error {
355359
var extractors []fwkdl.NotificationExtractor
356360
if rawExts, ok := r.extractors.Get(srcName); ok {
@@ -366,6 +370,31 @@ func (r *Runtime) Start(ctx context.Context, mgr ctrl.Manager) error {
366370
})
367371
}
368372

373+
// certWatcherSource is a polling source that runs a client-certificate watcher,
374+
// for example the https metrics/models scrape client.
375+
type certWatcherSource interface {
376+
CertWatcher() *certwatcher.CertWatcher
377+
}
378+
379+
// addCertWatchers hands each source's certificate watcher to the manager, which
380+
// owns its lifecycle. certwatcher needs no leader election, so every replica runs it.
381+
func (r *Runtime) addCertWatchers(mgr ctrl.Manager) error {
382+
for name, disp := range r.dispatchers.Dispatchers() {
383+
src, ok := disp.(certWatcherSource)
384+
if !ok {
385+
continue
386+
}
387+
w := src.CertWatcher()
388+
if w == nil {
389+
continue
390+
}
391+
if err := mgr.Add(w); err != nil {
392+
return fmt.Errorf("failed to add cert watcher for source %s: %w", name, err)
393+
}
394+
}
395+
return nil
396+
}
397+
369398
// NewEndpoint sets up data polling on the provided endpoint.
370399
func (r *Runtime) NewEndpoint(ctx context.Context, endpointMetadata *fwkdl.EndpointMetadata) fwkdl.Endpoint {
371400
logger, _ := logr.FromContext(ctx)

‎pkg/epp/datalayer/runtime_certwatcher_test.go‎

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
/*
2+
Copyright 2026 The Kubernetes Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package datalayer
18+
19+
import (
20+
"context"
21+
"crypto/ecdsa"
22+
"crypto/elliptic"
23+
"crypto/rand"
24+
"crypto/x509"
25+
"crypto/x509/pkix"
26+
"encoding/pem"
27+
"math/big"
28+
"os"
29+
"path/filepath"
30+
"testing"
31+
"time"
32+
33+
"github.com/stretchr/testify/assert"
34+
"github.com/stretchr/testify/require"
35+
"sigs.k8s.io/controller-runtime/pkg/certwatcher"
36+
"sigs.k8s.io/controller-runtime/pkg/manager"
37+
38+
fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer"
39+
fwkplugin "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin"
40+
)
41+
42+
// capturingManager records the runnables Start would hand to the manager.
43+
type capturingManager struct {
44+
manager.Manager
45+
added []manager.Runnable
46+
}
47+
48+
func (m *capturingManager) Add(r manager.Runnable) error {
49+
m.added = append(m.added, r)
50+
return nil
51+
}
52+
53+
// certSource is a polling dispatcher that optionally carries a cert watcher, the
54+
// shape addCertWatchers looks for.
55+
type certSource struct {
56+
name string
57+
watcher *certwatcher.CertWatcher
58+
}
59+
60+
func (s *certSource) TypedName() fwkplugin.TypedName {
61+
return fwkplugin.TypedName{Type: "test", Name: s.name}
62+
}
63+
func (s *certSource) Dispatch(context.Context, fwkdl.Endpoint) error { return nil }
64+
func (s *certSource) AppendExtractor(fwkplugin.Plugin) error { return nil }
65+
func (s *certSource) CertWatcher() *certwatcher.CertWatcher { return s.watcher }
66+
67+
func writeCertWatcher(t *testing.T) *certwatcher.CertWatcher {
68+
t.Helper()
69+
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
70+
require.NoError(t, err)
71+
tmpl := &x509.Certificate{
72+
SerialNumber: big.NewInt(1),
73+
Subject: pkix.Name{CommonName: "test-client"},
74+
NotBefore: time.Now().Add(-time.Hour),
75+
NotAfter: time.Now().Add(time.Hour),
76+
KeyUsage: x509.KeyUsageDigitalSignature,
77+
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth},
78+
}
79+
der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
80+
require.NoError(t, err)
81+
keyDER, err := x509.MarshalECPrivateKey(key)
82+
require.NoError(t, err)
83+
dir := t.TempDir()
84+
certPath := filepath.Join(dir, "tls.crt")
85+
keyPath := filepath.Join(dir, "tls.key")
86+
require.NoError(t, os.WriteFile(certPath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o600))
87+
require.NoError(t, os.WriteFile(keyPath, pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), 0o600))
88+
w, err := certwatcher.New(certPath, keyPath)
89+
require.NoError(t, err)
90+
return w
91+
}
92+
93+
// TestAddCertWatchers proves the seam the manager relies on: a source that carries
94+
// a watcher is handed to the manager (which then runs its Start), and a source
95+
// without one is skipped.
96+
func TestAddCertWatchers(t *testing.T) {
97+
watcher := writeCertWatcher(t)
98+
tests := []struct {
99+
name string
100+
sources []*certSource
101+
wantAdded int
102+
}{
103+
{name: "source with a watcher is added", sources: []*certSource{{name: "https-mtls", watcher: watcher}}, wantAdded: 1},
104+
{name: "source without a watcher is skipped", sources: []*certSource{{name: "plain-http"}}, wantAdded: 0},
105+
{name: "only the watcher-carrying source is added", sources: []*certSource{
106+
{name: "plain-http"}, {name: "https-mtls", watcher: watcher},
107+
}, wantAdded: 1},
108+
}
109+
for _, tt := range tests {
110+
t.Run(tt.name, func(t *testing.T) {
111+
r := NewRuntime(0)
112+
for _, s := range tt.sources {
113+
require.NoError(t, r.dispatchers.Register(s))
114+
}
115+
mgr := &capturingManager{}
116+
require.NoError(t, r.addCertWatchers(mgr))
117+
assert.Len(t, mgr.added, tt.wantAdded)
118+
})
119+
}
120+
}

‎pkg/epp/framework/plugins/datalayer/source/http/certreload_test.go‎

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
/*
2+
Copyright 2026 The Kubernetes Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package http
18+
19+
import (
20+
"crypto/tls"
21+
"io"
22+
"math/big"
23+
"net/http"
24+
"path/filepath"
25+
"sync/atomic"
26+
"testing"
27+
"time"
28+
29+
"github.com/stretchr/testify/assert"
30+
"github.com/stretchr/testify/require"
31+
)
32+
33+
// servedSerial returns the serial of the cert the config would present.
34+
func servedSerial(t *testing.T, cfg *tls.Config) *big.Int {
35+
t.Helper()
36+
cert, err := cfg.GetClientCertificate(nil)
37+
require.NoError(t, err)
38+
require.NotNil(t, cert)
39+
return cert.Leaf.SerialNumber
40+
}
41+
42+
// tlsConfigOf returns the TLS config the source's transport uses.
43+
func tlsConfigOf(t *testing.T, c Client) *tls.Config {
44+
t.Helper()
45+
tr, ok := c.(*client).Transport.(*http.Transport)
46+
require.True(t, ok)
47+
return tr.TLSClientConfig
48+
}
49+
50+
func TestTLSClientConfig_ReloadServesNewCertAndCyclesConnections(t *testing.T) {
51+
dir := t.TempDir()
52+
certPath, keyPath := writeCertKeyAt(t, dir, 2), filepath.Join(dir, "key.pem")
53+
54+
var cycled atomic.Int32
55+
cfg, watcher, err := tlsClientConfig(
56+
TLSOptions{ClientCertPath: certPath, ClientKeyPath: keyPath},
57+
func() { cycled.Add(1) })
58+
require.NoError(t, err)
59+
require.NotNil(t, watcher)
60+
61+
before := servedSerial(t, cfg)
62+
63+
writeCertKeyAt(t, dir, 3) // rotate in place
64+
require.NoError(t, watcher.ReadCertificate())
65+
66+
// certwatcher invokes the reload callback in its own goroutine.
67+
require.Eventually(t, func() bool { return servedSerial(t, cfg).Cmp(before) != 0 },
68+
10*time.Second, 10*time.Millisecond, "rotated certificate was never served")
69+
assert.Eventually(t, func() bool { return cycled.Load() > 0 },
70+
10*time.Second, 10*time.Millisecond, "idle connections were never cycled on rotation")
71+
}
72+
73+
func TestNewHTTPDataSource_CertWatcher(t *testing.T) {
74+
parser := func(r io.Reader) (any, error) { return struct{}{}, nil }
75+
dir := t.TempDir()
76+
certPath, keyPath := writeCertKeyAt(t, dir, 2), filepath.Join(dir, "key.pem")
77+
78+
tests := []struct {
79+
name string
80+
scheme string
81+
opts TLSOptions
82+
wantWatch bool
83+
}{
84+
{name: "http has no watcher", scheme: "http", opts: TLSOptions{}},
85+
{name: "https without client cert has no watcher", scheme: "https", opts: TLSOptions{}},
86+
{name: "https with client cert has a watcher", scheme: "https",
87+
opts: TLSOptions{ClientCertPath: certPath, ClientKeyPath: keyPath}, wantWatch: true},
88+
}
89+
for _, tt := range tests {
90+
t.Run(tt.name, func(t *testing.T) {
91+
ds, err := NewHTTPDataSource[any](tt.scheme, "/metrics", tt.opts, "test-type", "ds", parser)
92+
require.NoError(t, err)
93+
if tt.wantWatch {
94+
assert.NotNil(t, ds.CertWatcher())
95+
} else {
96+
assert.Nil(t, ds.CertWatcher())
97+
}
98+
})
99+
}
100+
}

‎pkg/epp/framework/plugins/datalayer/source/http/datasource.go‎

Lines changed: 38 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@ import (
3434
"sync"
3535
"time"
3636

37+
"sigs.k8s.io/controller-runtime/pkg/certwatcher"
3738
"sigs.k8s.io/controller-runtime/pkg/log"
3839

3940
"github.com/llm-d/llm-d-router/pkg/common/observability/logging"
@@ -68,10 +69,21 @@ type HTTPDataSource[T any] struct {
6869
// the dispatcher does not validate.
6970
parser func(io.Reader) (T, error)
7071

72+
// certWatcher watches the mTLS client cert for rotation. Nil for http or when
73+
// no client cert is configured. Run by the manager (see Runtime.Start).
74+
certWatcher *certwatcher.CertWatcher
75+
7176
mu sync.RWMutex
7277
exts []fwkdl.PollingExtractor[T]
7378
}
7479

80+
// CertWatcher returns the client-certificate watcher for the manager to run, or
81+
// nil when there is nothing to watch. Runtime.Start uses it to add the watcher to
82+
// the manager without knowing the source's concrete type.
83+
func (s *HTTPDataSource[T]) CertWatcher() *certwatcher.CertWatcher {
84+
return s.certWatcher
85+
}
86+
7587
// TLSOptions configures the https transport. The zero value verifies the target
7688
// against the system CA pool with no client certificate.
7789
type TLSOptions struct {
@@ -131,14 +143,18 @@ func NewHTTPDataSource[T any](scheme, path string, tlsOpts TLSOptions,
131143
Transport: baseTransport,
132144
},
133145
}
146+
var watcher *certwatcher.CertWatcher
134147
if scheme == "https" {
135-
tlsCfg, err := tlsClientConfig(tlsOpts)
148+
httpsTransport := baseTransport.Clone()
149+
// Close idle connections on rotation so pooled keep-alives re-handshake and
150+
// pick up the new cert instead of lingering on the old one until they idle out.
151+
tlsCfg, w, err := tlsClientConfig(tlsOpts, httpsTransport.CloseIdleConnections)
136152
if err != nil {
137153
return nil, err
138154
}
139-
httpsTransport := baseTransport.Clone()
140155
httpsTransport.TLSClientConfig = tlsCfg
141156
cl.Transport = httpsTransport
157+
watcher = w
142158
}
143159
return &HTTPDataSource[T]{
144160
typedName: fwkplugin.TypedName{Type: pluginType, Name: pluginName},
@@ -148,6 +164,7 @@ func NewHTTPDataSource[T any](scheme, path string, tlsOpts TLSOptions,
148164
useNodeAddress: cfg.useNodeAddress,
149165
client: cl,
150166
parser: parser,
167+
certWatcher: watcher,
151168
}, nil
152169
}
153170

@@ -171,24 +188,33 @@ func caCertPool(path string) (*x509.CertPool, error) {
171188
}
172189

173190
// tlsClientConfig builds a tls.Config: server verification via CACertPath (or the
174-
// system pool), plus an mTLS client certificate when ClientCertPath is set.
175-
func tlsClientConfig(opts TLSOptions) (*tls.Config, error) {
191+
// system pool), plus an mTLS client certificate when ClientCertPath is set. The
192+
// returned watcher serves the client cert and reloads it on rotation, calling
193+
// onReload after each reload. It is nil when no client cert is configured. The CA
194+
// is read once. The caller runs the watcher.
195+
func tlsClientConfig(opts TLSOptions, onReload func()) (*tls.Config, *certwatcher.CertWatcher, error) {
176196
cfg := &tls.Config{InsecureSkipVerify: opts.SkipVerify}
177197
if !opts.SkipVerify && opts.CACertPath != "" {
178198
pool, err := caCertPool(opts.CACertPath)
179199
if err != nil {
180-
return nil, err
200+
return nil, nil, err
181201
}
182202
cfg.RootCAs = pool
183203
}
184-
if opts.ClientCertPath != "" || opts.ClientKeyPath != "" {
185-
cert, err := tls.LoadX509KeyPair(opts.ClientCertPath, opts.ClientKeyPath)
186-
if err != nil {
187-
return nil, fmt.Errorf("%w: %w", ErrLoadClientCert, err)
188-
}
189-
cfg.Certificates = []tls.Certificate{cert}
204+
if opts.ClientCertPath == "" && opts.ClientKeyPath == "" {
205+
return cfg, nil, nil
206+
}
207+
watcher, err := certwatcher.New(opts.ClientCertPath, opts.ClientKeyPath)
208+
if err != nil {
209+
return nil, nil, fmt.Errorf("%w: %w", ErrLoadClientCert, err)
210+
}
211+
watcher.RegisterCallback(func(tls.Certificate) { onReload() })
212+
// certwatcher serves the current cert from its own cache. The CertificateRequestInfo
213+
// arg is unused, matching GetCertificate's ignored ClientHelloInfo arg.
214+
cfg.GetClientCertificate = func(*tls.CertificateRequestInfo) (*tls.Certificate, error) {
215+
return watcher.GetCertificate(nil)
190216
}
191-
return cfg, nil
217+
return cfg, watcher, nil
192218
}
193219

194220
func (s *HTTPDataSource[T]) TypedName() fwkplugin.TypedName { return s.typedName }

0 commit comments

Comments
 (0)