UPSTREAM: <carry>: Ensure service ca is mounted for projected tokens

OpenShift since 3.x has injected the service serving certificate
ca (service ca) bundle into service account token secrets. This was
intended to ensure that all pods would be able to easily verify
connections to endpoints secured with service serving
certificates. Since breaking customer workloads is not an option, and
there is no way to ensure that customers are not relying on the
service ca bundle being mounted at
/var/run/secrets/kubernetes.io/serviceaccount/service-ca.crt, it is
necessary to continue mounting the service ca bundle in the same
location in the bound token projected volumes that will be enabled by
the BoundServiceAccountTokenVolume feature (enabled by default in
1.21/4.8).

A new controller is added to create a configmap per namespace that is
annotated for service ca injection. The controller is derived from the
controller that creates configmaps for the root ca.

This new controller is being added in 4.7 to ensure that the service
ca configmaps will be present in all namespaces in the cluster before
an upgrade to 4.8 is attempted. An upgrade to 4.8 will enable the
BoundServiceAccountTokenVolume feature and all new pods will expect
the configmaps to already be present.

A similar strategy was pursued upstream, in that publication of root
ca configmaps was added in a release previous to enablement of
BoundServiceAccountTokenVolume.

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

@@ -27,6 +27,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/klog/v2"
+	"k8s.io/kubernetes/openshift-kube-controller-manager/servicecacertpublisher"
 	"k8s.io/kubernetes/pkg/controller/certificates/approver"
 	"k8s.io/kubernetes/pkg/controller/certificates/cleaner"
 	"k8s.io/kubernetes/pkg/controller/certificates/rootcacertpublisher"
@@ -221,3 +222,16 @@ func startRootCACertPublisher(ctx ControllerContext) (http.Handler, bool, error)
 	go sac.Run(1, ctx.Stop)
 	return nil, true, nil
 }
+
+func startServiceCACertPublisher(ctx ControllerContext) (http.Handler, bool, error) {
+	sac, err := servicecacertpublisher.NewPublisher(
+		ctx.InformerFactory.Core().V1().ConfigMaps(),
+		ctx.InformerFactory.Core().V1().Namespaces(),
+		ctx.ClientBuilder.ClientOrDie("service-ca-cert-publisher"),
+	)
+	if err != nil {
+		return nil, true, fmt.Errorf("error creating service CA certificate publisher: %v", err)
+	}
+	go sac.Run(1, ctx.Stop)
+	return nil, true, nil
+}

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

@@ -455,6 +455,7 @@ func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc
 	controllers["pv-protection"] = startPVProtectionController
 	controllers["ttl-after-finished"] = startTTLAfterFinishedController
 	controllers["root-ca-cert-publisher"] = startRootCACertPublisher
+	controllers["service-ca-cert-publisher"] = startServiceCACertPublisher
 	controllers["ephemeral-volume"] = startEphemeralVolumeController
 	if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.APIServerIdentity) &&
 		utilfeature.DefaultFeatureGate.Enabled(genericfeatures.StorageVersionAPI) {

openshift-kube-controller-manager/servicecacertpublisher/metrics.go

@@ -0,0 +1,56 @@
+package servicecacertpublisher
+
+import (
+	"strconv"
+	"sync"
+	"time"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/component-base/metrics"
+	"k8s.io/component-base/metrics/legacyregistry"
+)
+
+// ServiceCACertPublisher - subsystem name used by service_ca_cert_publisher
+const ServiceCACertPublisher = "service_ca_cert_publisher"
+
+var (
+	syncCounter = metrics.NewCounterVec(
+		&metrics.CounterOpts{
+			Subsystem:      ServiceCACertPublisher,
+			Name:           "sync_total",
+			Help:           "Number of namespace syncs happened in service ca cert publisher.",
+			StabilityLevel: metrics.ALPHA,
+		},
+		[]string{"code"},
+	)
+	syncLatency = metrics.NewHistogramVec(
+		&metrics.HistogramOpts{
+			Subsystem:      ServiceCACertPublisher,
+			Name:           "sync_duration_seconds",
+			Help:           "Number of namespace syncs happened in service ca cert publisher.",
+			Buckets:        metrics.ExponentialBuckets(0.001, 2, 15),
+			StabilityLevel: metrics.ALPHA,
+		},
+		[]string{"code"},
+	)
+)
+
+func recordMetrics(start time.Time, ns string, err error) {
+	code := "500"
+	if err == nil {
+		code = "200"
+	} else if se, ok := err.(*apierrors.StatusError); ok && se.Status().Code != 0 {
+		code = strconv.Itoa(int(se.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)
+	})
+}

openshift-kube-controller-manager/servicecacertpublisher/metrics_test.go

@@ -0,0 +1,81 @@
+package servicecacertpublisher
+
+import (
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	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{
+				"service_ca_cert_publisher_sync_total",
+			},
+			want: `
+# HELP service_ca_cert_publisher_sync_total [ALPHA] Number of namespace syncs happened in service ca cert publisher.
+# TYPE service_ca_cert_publisher_sync_total counter
+service_ca_cert_publisher_sync_total{code="200"} 1
+				`,
+		},
+		{
+			desc: "kube api error",
+			err:  apierrors.NewNotFound(corev1.Resource("configmap"), "test-configmap"),
+			metrics: []string{
+				"service_ca_cert_publisher_sync_total",
+			},
+			want: `
+# HELP service_ca_cert_publisher_sync_total [ALPHA] Number of namespace syncs happened in service ca cert publisher.
+# TYPE service_ca_cert_publisher_sync_total counter
+service_ca_cert_publisher_sync_total{code="404"} 1
+				`,
+		},
+		{
+			desc: "kube api error without code",
+			err:  &apierrors.StatusError{},
+			metrics: []string{
+				"service_ca_cert_publisher_sync_total",
+			},
+			want: `
+# HELP service_ca_cert_publisher_sync_total [ALPHA] Number of namespace syncs happened in service ca cert publisher.
+# TYPE service_ca_cert_publisher_sync_total counter
+service_ca_cert_publisher_sync_total{code="500"} 1
+				`,
+		},
+		{
+			desc: "general error",
+			err:  errors.New("test"),
+			metrics: []string{
+				"service_ca_cert_publisher_sync_total",
+			},
+			want: `
+# HELP service_ca_cert_publisher_sync_total [ALPHA] Number of namespace syncs happened in service ca cert publisher.
+# TYPE service_ca_cert_publisher_sync_total counter
+service_ca_cert_publisher_sync_total{code="500"} 1
+				`,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			recordMetrics(time.Now(), "test-ns", tc.err)
+			defer syncCounter.Reset()
+			if err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, strings.NewReader(tc.want), tc.metrics...); err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}

openshift-kube-controller-manager/servicecacertpublisher/publisher.go

@@ -0,0 +1,222 @@
+package servicecacertpublisher
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"time"
+
+	v1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	coreinformers "k8s.io/client-go/informers/core/v1"
+	clientset "k8s.io/client-go/kubernetes"
+	corelisters "k8s.io/client-go/listers/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/component-base/metrics/prometheus/ratelimiter"
+	"k8s.io/klog/v2"
+)
+
+// ServiceCACertConfigMapName is name of the configmap which stores certificates
+// to validate service serving certificates issued by the service ca operator.
+const ServiceCACertConfigMapName = "openshift-service-ca.crt"
+
+func init() {
+	registerMetrics()
+}
+
+// NewPublisher construct a new controller which would manage the configmap
+// which stores certificates in each namespace. It will make sure certificate
+// configmap exists in each namespace.
+func NewPublisher(cmInformer coreinformers.ConfigMapInformer, nsInformer coreinformers.NamespaceInformer, cl clientset.Interface) (*Publisher, error) {
+	e := &Publisher{
+		client: cl,
+		queue:  workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "service_ca_cert_publisher"),
+	}
+	if cl.CoreV1().RESTClient().GetRateLimiter() != nil {
+		if err := ratelimiter.RegisterMetricAndTrackRateLimiterUsage("service_ca_cert_publisher", cl.CoreV1().RESTClient().GetRateLimiter()); err != nil {
+			return nil, err
+		}
+	}
+
+	cmInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		DeleteFunc: e.configMapDeleted,
+		UpdateFunc: e.configMapUpdated,
+	})
+	e.cmLister = cmInformer.Lister()
+	e.cmListerSynced = cmInformer.Informer().HasSynced
+
+	nsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    e.namespaceAdded,
+		UpdateFunc: e.namespaceUpdated,
+	})
+	e.nsListerSynced = nsInformer.Informer().HasSynced
+
+	e.syncHandler = e.syncNamespace
+
+	return e, nil
+}
+
+// Publisher manages certificate ConfigMap objects inside Namespaces
+type Publisher struct {
+	client clientset.Interface
+
+	// To allow injection for testing.
+	syncHandler func(key string) error
+
+	cmLister       corelisters.ConfigMapLister
+	cmListerSynced cache.InformerSynced
+
+	nsListerSynced cache.InformerSynced
+
+	queue workqueue.RateLimitingInterface
+}
+
+// Run starts process
+func (c *Publisher) Run(workers int, stopCh <-chan struct{}) {
+	defer utilruntime.HandleCrash()
+	defer c.queue.ShutDown()
+
+	klog.Infof("Starting service CA certificate configmap publisher")
+	defer klog.Infof("Shutting down service CA certificate configmap publisher")
+
+	if !cache.WaitForNamedCacheSync("crt configmap", stopCh, c.cmListerSynced) {
+		return
+	}
+
+	for i := 0; i < workers; i++ {
+		go wait.Until(c.runWorker, time.Second, stopCh)
+	}
+
+	<-stopCh
+}
+
+func (c *Publisher) configMapDeleted(obj interface{}) {
+	cm, err := convertToCM(obj)
+	if err != nil {
+		utilruntime.HandleError(err)
+		return
+	}
+	if cm.Name != ServiceCACertConfigMapName {
+		return
+	}
+	c.queue.Add(cm.Namespace)
+}
+
+func (c *Publisher) configMapUpdated(_, newObj interface{}) {
+	cm, err := convertToCM(newObj)
+	if err != nil {
+		utilruntime.HandleError(err)
+		return
+	}
+	if cm.Name != ServiceCACertConfigMapName {
+		return
+	}
+	c.queue.Add(cm.Namespace)
+}
+
+func (c *Publisher) namespaceAdded(obj interface{}) {
+	namespace := obj.(*v1.Namespace)
+	c.queue.Add(namespace.Name)
+}
+
+func (c *Publisher) namespaceUpdated(oldObj interface{}, newObj interface{}) {
+	newNamespace := newObj.(*v1.Namespace)
+	if newNamespace.Status.Phase != v1.NamespaceActive {
+		return
+	}
+	c.queue.Add(newNamespace.Name)
+}
+
+func (c *Publisher) runWorker() {
+	for c.processNextWorkItem() {
+	}
+}
+
+// processNextWorkItem deals with one key off the queue. It returns false when
+// it's time to quit.
+func (c *Publisher) processNextWorkItem() bool {
+	key, quit := c.queue.Get()
+	if quit {
+		return false
+	}
+	defer c.queue.Done(key)
+
+	if err := c.syncHandler(key.(string)); err != nil {
+		utilruntime.HandleError(fmt.Errorf("syncing %q failed: %v", key, err))
+		c.queue.AddRateLimited(key)
+		return true
+	}
+
+	c.queue.Forget(key)
+	return true
+}
+
+func (c *Publisher) syncNamespace(ns string) (err error) {
+	startTime := time.Now()
+	defer func() {
+		recordMetrics(startTime, ns, err)
+		klog.V(4).Infof("Finished syncing namespace %q (%v)", ns, time.Since(startTime))
+	}()
+
+	annotations := map[string]string{
+		// This annotation prompts the service ca operator to inject
+		// the service ca bundle into the configmap.
+		"service.beta.openshift.io/inject-cabundle": "true",
+	}
+
+	cm, err := c.cmLister.ConfigMaps(ns).Get(ServiceCACertConfigMapName)
+	switch {
+	case apierrors.IsNotFound(err):
+		_, err = c.client.CoreV1().ConfigMaps(ns).Create(context.TODO(), &v1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:        ServiceCACertConfigMapName,
+				Annotations: annotations,
+			},
+			// Create new configmaps with the field referenced by the default
+			// projected volume. This ensures that pods - including the pod for
+			// service ca operator - will be able to start during initial
+			// deployment before the service ca operator has responded to the
+			// injection annotation.
+			Data: map[string]string{
+				"service-ca.crt": "",
+			},
+		}, metav1.CreateOptions{})
+		// don't retry a create if the namespace doesn't exist or is terminating
+		if apierrors.IsNotFound(err) || apierrors.HasStatusCause(err, v1.NamespaceTerminatingCause) {
+			return nil
+		}
+		return err
+	case err != nil:
+		return err
+	}
+
+	if reflect.DeepEqual(cm.Annotations, annotations) {
+		return nil
+	}
+
+	// copy so we don't modify the cache's instance of the configmap
+	cm = cm.DeepCopy()
+	cm.Annotations = annotations
+
+	_, err = c.client.CoreV1().ConfigMaps(ns).Update(context.TODO(), cm, metav1.UpdateOptions{})
+	return err
+}
+
+func convertToCM(obj interface{}) (*v1.ConfigMap, error) {
+	cm, ok := obj.(*v1.ConfigMap)
+	if !ok {
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			return nil, fmt.Errorf("couldn't get object from tombstone %#v", obj)
+		}
+		cm, ok = tombstone.Obj.(*v1.ConfigMap)
+		if !ok {
+			return nil, fmt.Errorf("tombstone contained object that is not a ConfigMap %#v", obj)
+		}
+	}
+	return cm, nil
+}