Bug 1977383: [release-4.7] Ensure service ca configmap is created in all namespaces

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
+}