trustbundles: add a new kube-apiserver-serving signer

cmd/kube-controller-manager/app/certificates.go

@@ -23,14 +23,21 @@ import (
 	"context"
 	"fmt"
 
+	certificatesv1alpha1 "k8s.io/api/certificates/v1alpha1"
+	"k8s.io/apiserver/pkg/server/dynamiccertificates"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/component-base/featuregate"
 	"k8s.io/controller-manager/controller"
 	"k8s.io/klog/v2"
 	"k8s.io/kubernetes/cmd/kube-controller-manager/names"
 	"k8s.io/kubernetes/pkg/controller/certificates/approver"
 	"k8s.io/kubernetes/pkg/controller/certificates/cleaner"
+	ctbpublisher "k8s.io/kubernetes/pkg/controller/certificates/clustertrustbundlepublisher"
 	"k8s.io/kubernetes/pkg/controller/certificates/rootcacertpublisher"
 	"k8s.io/kubernetes/pkg/controller/certificates/signer"
 	csrsigningconfig "k8s.io/kubernetes/pkg/controller/certificates/signer/config"
+	"k8s.io/kubernetes/pkg/features"
 )
 
 func newCertificateSigningRequestSigningControllerDescriptor() *ControllerDescriptor {
@@ -200,16 +207,9 @@ func newRootCACertificatePublisherControllerDescriptor() *ControllerDescriptor {
 }
 
 func startRootCACertificatePublisherController(ctx context.Context, controllerContext ControllerContext, controllerName string) (controller.Interface, bool, error) {
-	var (
+	rootCA, err := getKubeAPIServerCAFileContents(controllerContext)
-		rootCA []byte
+	if err != nil {
-		err    error
+		return nil, true, err
-	)
-	if controllerContext.ComponentConfig.SAController.RootCAFile != "" {
-		if rootCA, err = readCA(controllerContext.ComponentConfig.SAController.RootCAFile); err != nil {
-			return nil, true, fmt.Errorf("error parsing root-ca-file at %s: %v", controllerContext.ComponentConfig.SAController.RootCAFile, err)
-		}
-	} else {
-		rootCA = controllerContext.ClientBuilder.ConfigOrDie("root-ca-cert-publisher").CAData
 	}
 
 	sac, err := rootcacertpublisher.NewPublisher(
@@ -224,3 +224,77 @@ func startRootCACertificatePublisherController(ctx context.Context, controllerCo
 	go sac.Run(ctx, 1)
 	return nil, true, nil
 }
+
+func newKubeAPIServerSignerClusterTrustBundledPublisherDescriptor() *ControllerDescriptor {
+	return &ControllerDescriptor{
+		name:                 names.KubeAPIServerClusterTrustBundlePublisherController,
+		initFunc:             newKubeAPIServerSignerClusterTrustBundledPublisherController,
+		requiredFeatureGates: []featuregate.Feature{features.ClusterTrustBundle},
+	}
+}
+
+func newKubeAPIServerSignerClusterTrustBundledPublisherController(ctx context.Context, controllerContext ControllerContext, controllerName string) (controller.Interface, bool, error) {
+	rootCA, err := getKubeAPIServerCAFileContents(controllerContext)
+	if err != nil {
+		return nil, false, err
+	}
+
+	if len(rootCA) == 0 || !utilfeature.DefaultFeatureGate.Enabled(features.ClusterTrustBundle) {
+		return nil, false, nil
+	}
+
+	apiserverSignerClient := controllerContext.ClientBuilder.ClientOrDie("kube-apiserver-serving-clustertrustbundle-publisher")
+	ctbAvailable, err := clusterTrustBundlesAvailable(apiserverSignerClient)
+	if err != nil {
+		return nil, false, fmt.Errorf("discovery failed for ClusterTrustBundle: %w", err)
+	}
+
+	if !ctbAvailable {
+		return nil, false, nil
+	}
+
+	servingSigners, err := dynamiccertificates.NewStaticCAContent("kube-apiserver-serving", rootCA)
+	if err != nil {
+		return nil, false, fmt.Errorf("failed to create a static CA content provider for the kube-apiserver-serving signer: %w", err)
+	}
+
+	ctbPublisher, err := ctbpublisher.NewClusterTrustBundlePublisher(
+		"kubernetes.io/kube-apiserver-serving",
+		servingSigners,
+		apiserverSignerClient,
+	)
+	if err != nil {
+		return nil, false, fmt.Errorf("error creating kube-apiserver-serving signer certificates publisher: %w", err)
+	}
+
+	go ctbPublisher.Run(ctx)
+	return nil, true, nil
+}
+
+func clusterTrustBundlesAvailable(client kubernetes.Interface) (bool, error) {
+	resList, err := client.Discovery().ServerResourcesForGroupVersion(certificatesv1alpha1.SchemeGroupVersion.String())
+
+	if resList != nil {
+		// even in case of an error above there might be a partial list for APIs that
+		// were already successfully discovered
+		for _, r := range resList.APIResources {
+			if r.Name == "clustertrustbundles" {
+				return true, nil
+			}
+		}
+	}
+	return false, err
+}
+
+func getKubeAPIServerCAFileContents(controllerContext ControllerContext) ([]byte, error) {
+	if controllerContext.ComponentConfig.SAController.RootCAFile == "" {
+		return controllerContext.ClientBuilder.ConfigOrDie("root-ca-cert-publisher").CAData, nil
+	}
+
+	rootCA, err := readCA(controllerContext.ComponentConfig.SAController.RootCAFile)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing root-ca-file at %s: %w", controllerContext.ComponentConfig.SAController.RootCAFile, err)
+	}
+	return rootCA, nil
+
+}

cmd/kube-controller-manager/app/controllermanager.go

@@ -570,6 +570,7 @@ func NewControllerDescriptors() map[string]*ControllerDescriptor {
 	register(newVolumeAttributesClassProtectionControllerDescriptor())
 	register(newTTLAfterFinishedControllerDescriptor())
 	register(newRootCACertificatePublisherControllerDescriptor())
+	register(newKubeAPIServerSignerClusterTrustBundledPublisherDescriptor())
 	register(newEphemeralVolumeControllerDescriptor())
 
 	// feature gated

cmd/kube-controller-manager/app/controllermanager_test.go

@@ -89,6 +89,7 @@ func TestControllerNamesDeclaration(t *testing.T) {
 		names.VolumeAttributesClassProtectionController,
 		names.TTLAfterFinishedController,
 		names.RootCACertificatePublisherController,
+		names.KubeAPIServerClusterTrustBundlePublisherController,
 		names.EphemeralVolumeController,
 		names.StorageVersionGarbageCollectorController,
 		names.ResourceClaimController,

cmd/kube-controller-manager/names/controller_names.go

@@ -77,6 +77,7 @@ const (
 	PersistentVolumeProtectionController               = "persistentvolume-protection-controller"
 	TTLAfterFinishedController                         = "ttl-after-finished-controller"
 	RootCACertificatePublisherController               = "root-ca-certificate-publisher-controller"
+	KubeAPIServerClusterTrustBundlePublisherController = "kube-apiserver-serving-clustertrustbundle-publisher-controller"
 	EphemeralVolumeController                          = "ephemeral-volume-controller"
 	StorageVersionGarbageCollectorController           = "storageversion-garbage-collector-controller"
 	ResourceClaimController                            = "resourceclaim-controller"

pkg/controller/certificates/clustertrustbundlepublisher/metrics.go

@@ -0,0 +1,78 @@
+/*
+Copyright 2024 The Kubernetes 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 clustertrustbundlepublisher
+
+import (
+	"errors"
+	"strconv"
+	"sync"
+	"time"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/component-base/metrics"
+	"k8s.io/component-base/metrics/legacyregistry"
+)
+
+// clustertrustbundlePublisher - subsystem name used by clustertrustbundle_publisher
+const (
+	clustertrustbundlePublisher = "clustertrustbundle_publisher"
+)
+
+var (
+	syncCounter = metrics.NewCounterVec(
+		&metrics.CounterOpts{
+			Subsystem:      clustertrustbundlePublisher,
+			Name:           "sync_total",
+			Help:           "Number of syncs that occurred in cluster trust bundle publisher.",
+			StabilityLevel: metrics.ALPHA,
+		},
+		[]string{"code"},
+	)
+	syncLatency = metrics.NewHistogramVec(
+		&metrics.HistogramOpts{
+			Subsystem:      clustertrustbundlePublisher,
+			Name:           "sync_duration_seconds",
+			Help:           "The time it took to sync a cluster trust bundle.",
+			Buckets:        metrics.ExponentialBuckets(0.001, 2, 15),
+			StabilityLevel: metrics.ALPHA,
+		},
+		[]string{"code"},
+	)
+)
+
+func recordMetrics(start time.Time, err error) {
+	code := "500"
+	if err == nil {
+		code = "200"
+	} else {
+		var statusErr apierrors.APIStatus
+		if errors.As(err, &statusErr) && statusErr.Status().Code != 0 {
+			code = strconv.Itoa(int(statusErr.Status().Code))
+		}
+	}
+	syncLatency.WithLabelValues(code).Observe(time.Since(start).Seconds())
+	syncCounter.WithLabelValues(code).Inc()
+}
+
+var once sync.Once
+
+func registerMetrics() {
+	once.Do(func() {
+		legacyregistry.MustRegister(syncCounter)
+		legacyregistry.MustRegister(syncLatency)
+	})
+}

pkg/controller/certificates/clustertrustbundlepublisher/metrics_test.go

@@ -0,0 +1,110 @@
+/*
+Copyright 2024 The Kubernetes 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 clustertrustbundlepublisher
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	certificatesv1alpha1 "k8s.io/api/certificates/v1alpha1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/component-base/metrics/legacyregistry"
+	"k8s.io/component-base/metrics/testutil"
+)
+
+func TestSyncCounter(t *testing.T) {
+	testCases := []struct {
+		desc    string
+		err     error
+		metrics []string
+		want    string
+	}{
+		{
+			desc: "nil error",
+			err:  nil,
+			metrics: []string{
+				"clustertrustbundle_publisher_sync_total",
+			},
+			want: `
+# HELP clustertrustbundle_publisher_sync_total [ALPHA] Number of syncs that occurred in cluster trust bundle publisher.
+# TYPE clustertrustbundle_publisher_sync_total counter
+clustertrustbundle_publisher_sync_total{code="200"} 1
+				`,
+		},
+		{
+			desc: "kube api error",
+			err:  apierrors.NewNotFound(certificatesv1alpha1.Resource("clustertrustbundle"), "test.test:testSigner:something"),
+			metrics: []string{
+				"clustertrustbundle_publisher_sync_total",
+			},
+			want: `
+# HELP clustertrustbundle_publisher_sync_total [ALPHA] Number of syncs that occurred in cluster trust bundle publisher.
+# TYPE clustertrustbundle_publisher_sync_total counter
+clustertrustbundle_publisher_sync_total{code="404"} 1
+				`,
+		},
+		{
+			desc: "nested kube api error",
+			err:  fmt.Errorf("oh noes: %w", apierrors.NewBadRequest("bad request!")),
+			metrics: []string{
+				"clustertrustbundle_publisher_sync_total",
+			},
+			want: `
+# HELP clustertrustbundle_publisher_sync_total [ALPHA] Number of syncs that occurred in cluster trust bundle publisher.
+# TYPE clustertrustbundle_publisher_sync_total counter
+clustertrustbundle_publisher_sync_total{code="400"} 1
+				`,
+		},
+		{
+			desc: "kube api error without code",
+			err:  &apierrors.StatusError{},
+			metrics: []string{
+				"clustertrustbundle_publisher_sync_total",
+			},
+			want: `
+# HELP clustertrustbundle_publisher_sync_total [ALPHA] Number of syncs that occurred in cluster trust bundle publisher.
+# TYPE clustertrustbundle_publisher_sync_total counter
+clustertrustbundle_publisher_sync_total{code="500"} 1
+				`,
+		},
+		{
+			desc: "general error",
+			err:  errors.New("test"),
+			metrics: []string{
+				"clustertrustbundle_publisher_sync_total",
+			},
+			want: `
+# HELP clustertrustbundle_publisher_sync_total [ALPHA] Number of syncs that occurred in cluster trust bundle publisher.
+# TYPE clustertrustbundle_publisher_sync_total counter
+clustertrustbundle_publisher_sync_total{code="500"} 1
+				`,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			recordMetrics(time.Now(), tc.err)
+			defer syncCounter.Reset()
+			if err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, strings.NewReader(tc.want), tc.metrics...); err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}

pkg/controller/certificates/clustertrustbundlepublisher/publisher.go

@@ -0,0 +1,229 @@
+/*
+Copyright 2024 The Kubernetes 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 clustertrustbundlepublisher
+
+import (
+	"context"
+	"crypto/sha256"
+	"fmt"
+	"strings"
+	"time"
+
+	certificatesv1alpha1 "k8s.io/api/certificates/v1alpha1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/labels"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apiserver/pkg/server/dynamiccertificates"
+	certinformers "k8s.io/client-go/informers/certificates/v1alpha1"
+	clientset "k8s.io/client-go/kubernetes"
+	certlisters "k8s.io/client-go/listers/certificates/v1alpha1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog/v2"
+)
+
+func init() {
+	registerMetrics()
+}
+
+type ClusterTrustBundlePublisher struct {
+	signerName string
+	ca         dynamiccertificates.CAContentProvider
+
+	client clientset.Interface
+
+	ctbInformer     cache.SharedIndexInformer
+	ctbLister       certlisters.ClusterTrustBundleLister
+	ctbListerSynced cache.InformerSynced
+
+	queue workqueue.TypedRateLimitingInterface[string]
+}
+
+type caContentListener func()
+
+func (f caContentListener) Enqueue() {
+	f()
+}
+
+// NewClusterTrustBundlePublisher creates and maintains a cluster trust bundle object
+// for a signer named `signerName`. The cluster trust bundle object contains the
+// CA from the `caProvider` in its .spec.TrustBundle.
+func NewClusterTrustBundlePublisher(
+	signerName string,
+	caProvider dynamiccertificates.CAContentProvider,
+	kubeClient clientset.Interface,
+) (*ClusterTrustBundlePublisher, error) {
+	if len(signerName) == 0 {
+		return nil, fmt.Errorf("signerName cannot be empty")
+	}
+
+	p := &ClusterTrustBundlePublisher{
+		signerName: signerName,
+		ca:         caProvider,
+		client:     kubeClient,
+
+		queue: workqueue.NewTypedRateLimitingQueueWithConfig(
+			workqueue.DefaultTypedControllerRateLimiter[string](),
+			workqueue.TypedRateLimitingQueueConfig[string]{
+				Name: "ca_cert_publisher_cluster_trust_bundles",
+			},
+		),
+	}
+	p.ctbInformer = setupSignerNameFilteredCTBInformer(p.client, p.signerName)
+	p.ctbLister = certlisters.NewClusterTrustBundleLister(p.ctbInformer.GetIndexer())
+	p.ctbListerSynced = p.ctbInformer.HasSynced
+
+	_, err := p.ctbInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			p.queue.Add("")
+		},
+		UpdateFunc: func(_, _ interface{}) {
+			p.queue.Add("")
+		},
+		DeleteFunc: func(_ interface{}) {
+			p.queue.Add("")
+		},
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to register ClusterTrustBundle event handler: %w", err)
+	}
+	p.ca.AddListener(p.caContentChangedListener())
+
+	return p, nil
+}
+
+func (p *ClusterTrustBundlePublisher) caContentChangedListener() dynamiccertificates.Listener {
+	return caContentListener(func() {
+		p.queue.Add("")
+	})
+}
+
+func (p *ClusterTrustBundlePublisher) Run(ctx context.Context) {
+	defer utilruntime.HandleCrash()
+	defer p.queue.ShutDown()
+
+	logger := klog.FromContext(ctx)
+	logger.Info("Starting ClusterTrustBundle CA cert publisher controller")
+	defer logger.Info("Shutting down ClusterTrustBundle CA cert publisher controller")
+
+	go p.ctbInformer.Run(ctx.Done())
+
+	if !cache.WaitForNamedCacheSync("cluster trust bundle", ctx.Done(), p.ctbListerSynced) {
+		return
+	}
+
+	// init the signer syncer
+	p.queue.Add("")
+	go wait.UntilWithContext(ctx, p.runWorker(), time.Second)
+
+	<-ctx.Done()
+}
+
+func (p *ClusterTrustBundlePublisher) runWorker() func(context.Context) {
+	return func(ctx context.Context) {
+		for p.processNextWorkItem(ctx) {
+		}
+	}
+}
+
+// processNextWorkItem deals with one key off the queue. It returns false when
+// it's time to quit.
+func (p *ClusterTrustBundlePublisher) processNextWorkItem(ctx context.Context) bool {
+	key, quit := p.queue.Get()
+	if quit {
+		return false
+	}
+	defer p.queue.Done(key)
+
+	if err := p.syncClusterTrustBundle(ctx); err != nil {
+		utilruntime.HandleError(fmt.Errorf("syncing %q failed: %w", key, err))
+		p.queue.AddRateLimited(key)
+		return true
+	}
+
+	p.queue.Forget(key)
+	return true
+}
+
+func (p *ClusterTrustBundlePublisher) syncClusterTrustBundle(ctx context.Context) (err error) {
+	startTime := time.Now()
+	defer func() {
+		recordMetrics(startTime, err)
+		klog.FromContext(ctx).V(4).Info("Finished syncing ClusterTrustBundle", "signerName", p.signerName, "elapsedTime", time.Since(startTime))
+	}()
+
+	caBundle := string(p.ca.CurrentCABundleContent())
+	bundleName := constructBundleName(p.signerName, []byte(caBundle))
+
+	bundle, err := p.ctbLister.Get(bundleName)
+	if apierrors.IsNotFound(err) {
+		_, err = p.client.CertificatesV1alpha1().ClusterTrustBundles().Create(ctx, &certificatesv1alpha1.ClusterTrustBundle{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: bundleName,
+			},
+			Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+				SignerName:  p.signerName,
+				TrustBundle: caBundle,
+			},
+		}, metav1.CreateOptions{})
+	} else if err == nil && bundle.Spec.TrustBundle != caBundle {
+		bundle = bundle.DeepCopy()
+		bundle.Spec.TrustBundle = caBundle
+		_, err = p.client.CertificatesV1alpha1().ClusterTrustBundles().Update(ctx, bundle, metav1.UpdateOptions{})
+	}
+
+	if err != nil {
+		return err
+	}
+
+	signerTrustBundles, err := p.ctbLister.List(labels.Everything())
+	if err != nil {
+		return err
+	}
+
+	// keep the deletion error to be returned in the end in order to retrigger the reconciliation loop
+	var deletionError error
+	for _, bundleObject := range signerTrustBundles {
+		if bundleObject.Name == bundleName {
+			continue
+		}
+
+		if err := p.client.CertificatesV1alpha1().ClusterTrustBundles().Delete(ctx, bundleObject.Name, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
+			klog.FromContext(ctx).Error(err, "failed to remove a cluster trust bundle", "bundleName", bundleObject.Name)
+			deletionError = err
+		}
+	}
+
+	return deletionError
+}
+
+func setupSignerNameFilteredCTBInformer(client clientset.Interface, signerName string) cache.SharedIndexInformer {
+	return certinformers.NewFilteredClusterTrustBundleInformer(client, 0, cache.Indexers{},
+		func(options *metav1.ListOptions) {
+			options.FieldSelector = fields.OneTermEqualSelector("spec.signerName", signerName).String()
+		})
+}
+
+func constructBundleName(signerName string, bundleBytes []byte) string {
+	namePrefix := strings.ReplaceAll(signerName, "/", ":") + ":"
+	bundleHash := sha256.Sum256(bundleBytes)
+	return fmt.Sprintf("%s%x", namePrefix, bundleHash[:12])
+
+}

pkg/controller/certificates/clustertrustbundlepublisher/publisher_test.go

@@ -0,0 +1,362 @@
+/*
+Copyright 2024 The Kubernetes 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 clustertrustbundlepublisher
+
+import (
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	cryptorand "crypto/rand"
+	"testing"
+
+	certificatesv1alpha1 "k8s.io/api/certificates/v1alpha1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apiserver/pkg/server/dynamiccertificates"
+	"k8s.io/client-go/kubernetes/fake"
+	clienttesting "k8s.io/client-go/testing"
+	"k8s.io/client-go/tools/cache"
+	certutil "k8s.io/client-go/util/cert"
+	"k8s.io/kubernetes/test/utils/ktesting"
+)
+
+const testSignerName = "test.test/testSigner"
+
+func TestCTBPublisherSync(t *testing.T) {
+	checkCreatedTestSignerBundle := func(t *testing.T, actions []clienttesting.Action) {
+		filteredActions := filterOutListWatch(actions)
+		if len(filteredActions) != 1 {
+			t.Fatalf("expected 1 action, got %v", filteredActions)
+		}
+
+		createAction := expectAction[clienttesting.CreateAction](t, filteredActions[0], "create")
+
+		ctb, ok := createAction.GetObject().(*certificatesv1alpha1.ClusterTrustBundle)
+		if !ok {
+			t.Fatalf("expected ClusterTrustBundle create, got %v", createAction.GetObject())
+		}
+
+		if ctb.Spec.SignerName != testSignerName {
+			t.Fatalf("expected signer name %q, got %q", testSignerName, ctb.Spec.SignerName)
+		}
+	}
+
+	checkUpdatedTestSignerBundle := func(expectedCABytes []byte) func(t *testing.T, actions []clienttesting.Action) {
+		return func(t *testing.T, actions []clienttesting.Action) {
+			filteredActions := filterOutListWatch(actions)
+			if len(filteredActions) != 1 {
+				t.Fatalf("expected 1 action, got %v", filteredActions)
+			}
+
+			updateAction := expectAction[clienttesting.UpdateAction](t, filteredActions[0], "update")
+
+			ctb, ok := updateAction.GetObject().(*certificatesv1alpha1.ClusterTrustBundle)
+			if !ok {
+				t.Fatalf("expected ClusterTrustBundle update, got %v", updateAction.GetObject())
+			}
+
+			if ctb.Spec.SignerName != testSignerName {
+				t.Fatalf("expected signer name %q, got %q", testSignerName, ctb.Spec.SignerName)
+			}
+
+			if ctb.Spec.TrustBundle != string(expectedCABytes) {
+				t.Fatalf("expected trust bundle payload:\n%s\n, got %q", expectedCABytes, ctb.Spec.TrustBundle)
+			}
+		}
+	}
+
+	checkDeletedTestSignerBundle := func(name string) func(t *testing.T, actions []clienttesting.Action) {
+		return func(t *testing.T, actions []clienttesting.Action) {
+			filteredActions := filterOutListWatch(actions)
+			if len(filteredActions) != 1 {
+				t.Fatalf("expected 1 action, got %v", filteredActions)
+			}
+
+			deleteAction := expectAction[clienttesting.DeleteAction](t, filteredActions[0], "delete")
+
+			if actionName := deleteAction.GetName(); actionName != name {
+				t.Fatalf("expected deleted CTB name %q, got %q", name, actionName)
+			}
+		}
+	}
+
+	testCAProvider := testingCABundlleProvider(t)
+	testBundleName := constructBundleName(testSignerName, testCAProvider.CurrentCABundleContent())
+
+	for _, tt := range []struct {
+		name          string
+		existingCTBs  []runtime.Object
+		expectActions func(t *testing.T, actions []clienttesting.Action)
+		wantErr       bool
+	}{
+		{
+			name:          "no CTBs exist",
+			expectActions: checkCreatedTestSignerBundle,
+		},
+		{
+			name: "no CTBs for the current signer exist",
+			existingCTBs: []runtime.Object{
+				&certificatesv1alpha1.ClusterTrustBundle{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "nosigner",
+					},
+					Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+						TrustBundle: "somedatahere",
+					},
+				},
+				&certificatesv1alpha1.ClusterTrustBundle{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "signer:one",
+					},
+					Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+						SignerName:  "signer",
+						TrustBundle: "signerdata",
+					},
+				},
+			},
+			expectActions: checkCreatedTestSignerBundle,
+		},
+		{
+			name: "CTB for the signer exists with different content",
+			existingCTBs: []runtime.Object{
+				&certificatesv1alpha1.ClusterTrustBundle{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: testBundleName,
+					},
+					Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+						SignerName:  testSignerName,
+						TrustBundle: "olddata",
+					},
+				},
+			},
+			expectActions: checkUpdatedTestSignerBundle(testCAProvider.CurrentCABundleContent()),
+		},
+		{
+			name: "multiple CTBs for the signer",
+			existingCTBs: []runtime.Object{
+				&certificatesv1alpha1.ClusterTrustBundle{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: testBundleName,
+					},
+					Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+						SignerName:  testSignerName,
+						TrustBundle: string(testCAProvider.CurrentCABundleContent()),
+					},
+				},
+				&certificatesv1alpha1.ClusterTrustBundle{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "test.test/testSigner:name2",
+					},
+					Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+						SignerName:  testSignerName,
+						TrustBundle: string(testCAProvider.CurrentCABundleContent()),
+					},
+				},
+			},
+			expectActions: checkDeletedTestSignerBundle("test.test/testSigner:name2"),
+		},
+		{
+			name: "multiple CTBs for the signer - the one with the proper name needs changing",
+			existingCTBs: []runtime.Object{
+				&certificatesv1alpha1.ClusterTrustBundle{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: testBundleName,
+					},
+					Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+						SignerName:  testSignerName,
+						TrustBundle: "olddata",
+					},
+				},
+				&certificatesv1alpha1.ClusterTrustBundle{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "test.test/testSigner:name2",
+					},
+					Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+						SignerName:  testSignerName,
+						TrustBundle: string(testCAProvider.CurrentCABundleContent()),
+					},
+				},
+			},
+			expectActions: func(t *testing.T, actions []clienttesting.Action) {
+				filteredActions := filterOutListWatch(actions)
+				if len(filteredActions) != 2 {
+					t.Fatalf("expected 2 actions, got %v", filteredActions)
+				}
+				checkUpdatedTestSignerBundle(testCAProvider.CurrentCABundleContent())(t, filteredActions[:1])
+				checkDeletedTestSignerBundle("test.test/testSigner:name2")(t, filteredActions[1:])
+			},
+		},
+		{
+			name: "another CTB with a different name exists for the signer",
+			existingCTBs: []runtime.Object{
+				&certificatesv1alpha1.ClusterTrustBundle{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "test.test/testSigner:preexisting",
+					},
+					Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+						SignerName:  testSignerName,
+						TrustBundle: string(testCAProvider.CurrentCABundleContent()),
+					},
+				},
+			},
+			expectActions: func(t *testing.T, actions []clienttesting.Action) {
+				filteredActions := filterOutListWatch(actions)
+				if len(filteredActions) != 2 {
+					t.Fatalf("expected 2 actions, got %v", filteredActions)
+				}
+				checkCreatedTestSignerBundle(t, filteredActions[:1])
+				checkDeletedTestSignerBundle("test.test/testSigner:preexisting")(t, filteredActions[1:])
+			},
+		},
+		{
+			name: "CTB at the correct state - noop",
+			existingCTBs: []runtime.Object{
+				&certificatesv1alpha1.ClusterTrustBundle{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "nosigner",
+					},
+					Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+						TrustBundle: "somedatahere",
+					},
+				},
+				&certificatesv1alpha1.ClusterTrustBundle{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "signer:one",
+					},
+					Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+						SignerName:  "signer",
+						TrustBundle: "signerdata",
+					},
+				},
+				&certificatesv1alpha1.ClusterTrustBundle{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: testBundleName,
+					},
+					Spec: certificatesv1alpha1.ClusterTrustBundleSpec{
+						SignerName:  testSignerName,
+						TrustBundle: string(testCAProvider.CurrentCABundleContent()),
+					},
+				},
+			},
+			expectActions: func(t *testing.T, actions []clienttesting.Action) {
+				actions = filterOutListWatch(actions)
+				if len(actions) != 0 {
+					t.Fatalf("expected no actions, got %v", actions)
+				}
+			},
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			testCtx := ktesting.Init(t)
+
+			fakeClient := fakeKubeClientSetWithCTBList(t, testSignerName, tt.existingCTBs...)
+
+			p, err := NewClusterTrustBundlePublisher(testSignerName, testCAProvider, fakeClient)
+			if err != nil {
+				t.Fatalf("failed to set up a new cluster trust bundle publisher: %v", err)
+			}
+
+			go p.ctbInformer.Run(testCtx.Done())
+			if !cache.WaitForCacheSync(testCtx.Done(), p.ctbInformer.HasSynced) {
+				t.Fatal("timed out waiting for informer to sync")
+			}
+
+			if err := p.syncClusterTrustBundle(testCtx); (err != nil) != tt.wantErr {
+				t.Errorf("syncClusterTrustBundle() error = %v, wantErr %v", err, tt.wantErr)
+			}
+
+			tt.expectActions(t, fakeClient.Actions())
+		})
+	}
+}
+
+func fakeKubeClientSetWithCTBList(t *testing.T, signerName string, ctbs ...runtime.Object) *fake.Clientset {
+	fakeClient := fake.NewSimpleClientset(ctbs...)
+	fakeClient.PrependReactor("list", "clustertrustbundles", func(action clienttesting.Action) (handled bool, ret runtime.Object, err error) {
+		listAction, ok := action.(clienttesting.ListAction)
+		if !ok {
+			t.Fatalf("expected list action, got %v", action)
+		}
+
+		// fakeClient does not implement field selector, we have to do it manually
+		listRestrictions := listAction.GetListRestrictions()
+		if listRestrictions.Fields == nil || listRestrictions.Fields.String() != ("spec.signerName="+signerName) {
+			return false, nil, nil
+		}
+
+		retList := &certificatesv1alpha1.ClusterTrustBundleList{}
+		for _, ctb := range ctbs {
+			ctbObj, ok := ctb.(*certificatesv1alpha1.ClusterTrustBundle)
+			if !ok {
+				continue
+			}
+			if ctbObj.Spec.SignerName == testSignerName {
+				retList.Items = append(retList.Items, *ctbObj)
+			}
+		}
+
+		return true, retList, nil
+	})
+
+	return fakeClient
+}
+
+func expectAction[A clienttesting.Action](t *testing.T, action clienttesting.Action, verb string) A {
+	if action.GetVerb() != verb {
+		t.Fatalf("expected action with verb %q, got %q", verb, action.GetVerb())
+	}
+
+	retAction, ok := action.(A)
+	if !ok {
+		t.Fatalf("expected %T action, got %v", *new(A), action)
+	}
+
+	return retAction
+}
+
+func filterOutListWatch(actions []clienttesting.Action) []clienttesting.Action {
+	var filtered []clienttesting.Action
+	for _, a := range actions {
+		if a.Matches("list", "clustertrustbundles") || a.Matches("watch", "clustertrustbundles") {
+			continue
+		}
+		filtered = append(filtered, a)
+	}
+	return filtered
+}
+
+func testingCABundlleProvider(t *testing.T) dynamiccertificates.CAContentProvider {
+	key, err := ecdsa.GenerateKey(elliptic.P256(), cryptorand.Reader)
+	if err != nil {
+		t.Fatalf("failed to create a private key: %v", err)
+	}
+	caCert, err := certutil.NewSelfSignedCACert(certutil.Config{CommonName: "test-ca"}, key)
+	if err != nil {
+		t.Fatalf("failed to create a self-signed CA cert: %v", err)
+	}
+
+	caPEM, err := certutil.EncodeCertificates(caCert)
+	if err != nil {
+		t.Fatalf("failed to PEM-encode cert: %v", err)
+	}
+
+	caProvider, err := dynamiccertificates.NewStaticCAContent("testca", caPEM)
+	if err != nil {
+		t.Fatalf("failed to create a static CA provider: %v", err)
+	}
+
+	return caProvider
+}

plugin/pkg/auth/authorizer/rbac/bootstrappolicy/controller_policy.go

@@ -453,6 +453,18 @@ func buildControllerRoles() ([]rbacv1.ClusterRole, []rbacv1.ClusterRoleBinding)
 			eventsRule(),
 		},
 	})
+
+	if utilfeature.DefaultFeatureGate.Enabled(features.ClusterTrustBundle) {
+		addControllerRole(&controllerRoles, &controllerRoleBindings, rbacv1.ClusterRole{
+			ObjectMeta: metav1.ObjectMeta{Name: saRolePrefix + "kube-apiserver-serving-clustertrustbundle-publisher"},
+			Rules: []rbacv1.PolicyRule{
+				rbacv1helpers.NewRule("attest").Groups(certificatesGroup).Resources("signers").Names("kubernetes.io/kube-apiserver-serving").RuleOrDie(),
+				rbacv1helpers.NewRule("create", "update", "delete", "list", "watch").Groups(certificatesGroup).Resources("clustertrustbundles").RuleOrDie(),
+				eventsRule(),
+			},
+		})
+	}
+
 	addControllerRole(&controllerRoles, &controllerRoleBindings, rbacv1.ClusterRole{
 		ObjectMeta: metav1.ObjectMeta{Name: saRolePrefix + "validatingadmissionpolicy-status-controller"},
 		Rules: []rbacv1.PolicyRule{

test/integration/clustertrustbundles/apiserversigner_test.go

@@ -0,0 +1,257 @@
+/*
+Copyright 2024 The Kubernetes 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 clustertrustbundles
+
+import (
+	"context"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"fmt"
+	"math/big"
+	"path/filepath"
+	"regexp"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"k8s.io/api/certificates/v1alpha1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured/unstructuredscheme"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/wait"
+	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	certutil "k8s.io/client-go/util/cert"
+	kubeapiservertesting "k8s.io/kubernetes/cmd/kube-apiserver/app/testing"
+	kubecontrollermanagertesting "k8s.io/kubernetes/cmd/kube-controller-manager/app/testing"
+	"k8s.io/kubernetes/test/integration/framework"
+	"k8s.io/kubernetes/test/utils/ktesting"
+	"k8s.io/kubernetes/test/utils/kubeconfig"
+)
+
+func TestClusterTrustBundlesPublisherController(t *testing.T) {
+	// KUBE_APISERVER_SERVE_REMOVED_APIS_FOR_ONE_RELEASE allows for APIs pending removal to not block tests
+	// TODO: Remove this line once certificates v1alpha1 types to be removed in 1.32 are fully removed
+	t.Setenv("KUBE_APISERVER_SERVE_REMOVED_APIS_FOR_ONE_RELEASE", "true")
+	ctx := ktesting.Init(t)
+
+	certBytes := mustMakeCertificate(t, &x509.Certificate{
+		SerialNumber: big.NewInt(0),
+		Subject: pkix.Name{
+			CommonName: "testsigner-kas",
+		},
+		IsCA:                  true,
+		BasicConstraintsValid: true,
+	})
+
+	certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certBytes})
+
+	tmpDir := t.TempDir()
+	cacertPath := filepath.Join(tmpDir, "kube-apiserver-serving.crt")
+	if err := certutil.WriteCert(cacertPath, certPEM); err != nil {
+		t.Fatalf("failed to write the CA cert into a file: %v", err)
+	}
+
+	apiServerFlags := []string{
+		"--disable-admission-plugins", "ServiceAccount",
+		"--authorization-mode=RBAC",
+		"--feature-gates", "ClusterTrustBundle=true",
+	}
+	storageConfig := framework.SharedEtcd()
+	server := kubeapiservertesting.StartTestServerOrDie(t, nil, apiServerFlags, storageConfig)
+	defer server.TearDownFn()
+
+	kubeConfigFile := createKubeConfigFileForRestConfig(t, server.ClientConfig)
+
+	kcm := kubecontrollermanagertesting.StartTestServerOrDie(ctx, []string{
+		"--kubeconfig=" + kubeConfigFile,
+		"--controllers=kube-apiserver-serving-clustertrustbundle-publisher-controller", // these are the only controllers needed for this test
+		"--use-service-account-credentials=true",                                       // exercise RBAC of kube-apiserver-serving-clustertrustbundle-publisher controller
+		"--leader-elect=false",                                                         // KCM leader election calls os.Exit when it ends, so it is easier to just turn it off altogether
+		"--root-ca-file=" + cacertPath,
+		"--feature-gates=ClusterTrustBundle=true",
+	})
+	defer kcm.TearDownFn()
+
+	// setup finished, tests follow
+	clientSet, err := clientset.NewForConfig(server.ClientConfig)
+	if err != nil {
+		t.Fatalf("error in create clientset: %v", err)
+	}
+
+	unrelatedSigner := mustMakeCertificate(t, &x509.Certificate{
+		SerialNumber: big.NewInt(0),
+		Subject: pkix.Name{
+			CommonName: "testsigner-kas",
+		},
+		IsCA:                  true,
+		BasicConstraintsValid: true,
+	})
+	unrelatedPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: unrelatedSigner})
+	// set up a signer that's completely unrelated to the controller to check
+	// it's not anyhow handled by it
+	unrelatedCTB, err := clientSet.CertificatesV1alpha1().ClusterTrustBundles().Create(ctx,
+		&v1alpha1.ClusterTrustBundle{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test.test:unrelated:0",
+			},
+			Spec: v1alpha1.ClusterTrustBundleSpec{
+				SignerName:  "test.test/unrelated",
+				TrustBundle: string(unrelatedPEM),
+			},
+		},
+		metav1.CreateOptions{})
+	if err != nil {
+		t.Fatalf("failed to set up an unrelated signer CTB: %v", err)
+	}
+
+	t.Log("check that the controller creates a single buundle with expected PEM content")
+	waitUntilSingleKASSignerCTB(ctx, t, clientSet, certPEM)
+
+	t.Log("check that the controller deletes any additional bundles for the same signer")
+	if _, err := clientSet.CertificatesV1alpha1().ClusterTrustBundles().Create(ctx, &v1alpha1.ClusterTrustBundle{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "kubernetes.io:kube-apiserver-serving:testname",
+		},
+		Spec: v1alpha1.ClusterTrustBundleSpec{
+			SignerName:  "kubernetes.io/kube-apiserver-serving",
+			TrustBundle: string(certPEM),
+		},
+	}, metav1.CreateOptions{}); err != nil {
+		t.Fatalf("failed to create an additional cluster trust bundle: %v", err)
+	}
+
+	waitUntilSingleKASSignerCTB(ctx, t, clientSet, certPEM)
+
+	t.Log("check that the controller reconciles the bundle back to its original state if changed")
+	differentSigner := mustMakeCertificate(t, &x509.Certificate{
+		SerialNumber: big.NewInt(0),
+		Subject: pkix.Name{
+			CommonName: "testsigner-kas-different",
+		},
+		IsCA:                  true,
+		BasicConstraintsValid: true,
+	})
+	differentSignerPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: differentSigner})
+
+	ctbList, err := clientSet.CertificatesV1alpha1().ClusterTrustBundles().List(ctx, metav1.ListOptions{
+		FieldSelector: "spec.signerName=kubernetes.io/kube-apiserver-serving",
+	})
+	if err != nil || len(ctbList.Items) != 1 {
+		t.Fatalf("failed to retrieve CTB list containing the single CTB for the KAS serving signer: %v", err)
+	}
+
+	ctbToUpdate := ctbList.Items[0].DeepCopy()
+	ctbToUpdate.Spec.TrustBundle = string(differentSignerPEM)
+
+	if _, err = clientSet.CertificatesV1alpha1().ClusterTrustBundles().Update(ctx, ctbToUpdate, metav1.UpdateOptions{}); err != nil {
+		t.Fatalf("failed to update ctb with new PEM bundle: %v", err)
+	}
+
+	waitUntilSingleKASSignerCTB(ctx, t, clientSet, certPEM)
+
+	unrelatedCTB, err = clientSet.CertificatesV1alpha1().ClusterTrustBundles().Get(ctx, unrelatedCTB.Name, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("failed to get the unrelated CTB back: %v", err)
+	}
+	if unrelatedCTB.Spec.TrustBundle != string(unrelatedPEM) {
+		t.Fatalf("the PEM content changed for the unrelated CTB:\n%s\n", unrelatedCTB.Spec.TrustBundle)
+	}
+
+	totalSynncs := getTotalSyncMetric(ctx, t, server.ClientConfig, "clustertrustbundle_publisher_sync_total")
+	if totalSynncs <= 0 {
+		t.Fatalf("expected non-zero total syncs: %d", totalSynncs)
+	}
+}
+
+func waitUntilSingleKASSignerCTB(ctx context.Context, t *testing.T, clientSet *clientset.Clientset, caPEM []byte) {
+	err := wait.PollUntilContextTimeout(ctx, 200*time.Millisecond, 30*time.Second, true, func(ctx context.Context) (done bool, err error) {
+		ctbList, err := clientSet.CertificatesV1alpha1().ClusterTrustBundles().List(ctx, metav1.ListOptions{
+			FieldSelector: "spec.signerName=kubernetes.io/kube-apiserver-serving",
+		})
+
+		if err != nil {
+			t.Logf("failed to list kube-apiserver-signer trust bundles: %v", err)
+			return false, nil
+		}
+
+		if len(ctbList.Items) != 1 {
+			t.Logf("expected a single CTB, got %v", ctbList.Items)
+			return false, nil
+		}
+
+		if ctbList.Items[0].Spec.TrustBundle != string(caPEM) {
+			t.Logf("CTB trustBundles are different")
+			return false, nil
+		}
+		return true, nil
+	})
+
+	if err != nil {
+		t.Fatalf("there has always been a wrong number of trust bundles: %v", err)
+	}
+
+}
+
+func createKubeConfigFileForRestConfig(t *testing.T, restConfig *rest.Config) string {
+	t.Helper()
+
+	clientConfig := kubeconfig.CreateKubeConfig(restConfig)
+
+	kubeConfigFile := filepath.Join(t.TempDir(), "kubeconfig.yaml")
+	if err := clientcmd.WriteToFile(*clientConfig, kubeConfigFile); err != nil {
+		t.Fatal(err)
+	}
+	return kubeConfigFile
+}
+
+func getTotalSyncMetric(ctx context.Context, t *testing.T, clientConfig *rest.Config, metric string) int {
+	t.Helper()
+
+	copyConfig := rest.CopyConfig(clientConfig)
+	copyConfig.GroupVersion = &schema.GroupVersion{}
+	copyConfig.NegotiatedSerializer = unstructuredscheme.NewUnstructuredNegotiatedSerializer()
+	rc, err := rest.RESTClientFor(copyConfig)
+	if err != nil {
+		t.Fatalf("Failed to create REST client: %v", err)
+	}
+
+	body, err := rc.Get().AbsPath("/metrics").DoRaw(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	metricRegex := regexp.MustCompile(fmt.Sprintf(`%s{.*} (\d+)`, metric))
+	for _, line := range strings.Split(string(body), "\n") {
+		if strings.HasPrefix(line, metric) {
+			matches := metricRegex.FindStringSubmatch(line)
+			if len(matches) == 2 {
+				metricValue, err := strconv.Atoi(matches[1])
+				if err != nil {
+					t.Fatalf("Failed to convert metric value to integer: %v", err)
+				}
+				return metricValue
+			}
+		}
+	}
+
+	t.Fatalf("metric %q not seen in body:\n%s\n", metric, string(body))
+	return 0
+}