Listen for secret changes in the operator namespace and trust TLS cer…

…tificates stored in secrets with the prefix "operator-ca-tls."

* No longer copy the secret `operator-ca-tls` from the operator namespace to the tenants namespace: Since [PR minio#1847](minio#1847), the secret `operator-ca-tls` is no longer mounted in the tenant, so there is no need to keep a copy.
* `queue.NewNamedRateLimitingQueue` is deprecated and has been replaced with the recommended `queue.NewRateLimitingQueueWithConfig`.
* Remove the duplicated method `getTLSSecret` and invoke `getCertificateSecret` instead.
* Rename [generateTLSCert](https://github.com/minio/operator/blob/1c2fa4f402cc2c91c9903e6da6e9a693c92b65e4/pkg/controller/tls.go#L108) to `generateTLSCertificateForService` for better understanding.
* Remove duplicated constants for 'public.crt', 'tls.crt', and 'ca.crt' in the `github.com/minio/operator/pkg/common` namespace.
* Replace hardcoded strings 'public.crt', 'tls.crt', and 'ca.crt' with constants in the `github.com/minio/operator/pkg/certs` namespace.

Signed-off-by: pjuarezd <[email protected]>

pkg/apis/minio.min.io/v2/helper.go

@@ -37,6 +37,8 @@ import (
 	"text/template"
 	"time"
 
+	"github.com/minio/operator/pkg/certs"
+
 	"github.com/miekg/dns"
 
 	appsv1 "k8s.io/api/apps/v1"
@@ -102,7 +104,7 @@ var (
 // GetPodCAFromFile assumes the operator is running inside a k8s pod and extract the
 // current ca certificate from /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
 func GetPodCAFromFile() []byte {
-	cert, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/ca.crt")
+	cert, err := os.ReadFile(fmt.Sprintf("/var/run/secrets/kubernetes.io/serviceaccount/%s", certs.CAPublicCertFile))
 	if err != nil {
 		return nil
 	}
@@ -120,7 +122,7 @@ func GetOpenshiftServiceCAFromFile() []byte {
 
 // GetOpenshiftCSRSignerCAFromFile extracts the tls.crt certificate in Openshift deployments coming from the mounted secret openshift-csr-signer-ca
 func GetOpenshiftCSRSignerCAFromFile() []byte {
-	cert, err := os.ReadFile("/tmp/csr-signer-ca/tls.crt")
+	cert, err := os.ReadFile(fmt.Sprintf("/tmp/csr-signer-ca/%s", certs.TLSCertFile))
 	if err != nil {
 		return nil
 	}
@@ -129,13 +131,13 @@ func GetOpenshiftCSRSignerCAFromFile() []byte {
 
 // GetPublicCertFilePath return the path to the certificate file based for the serviceName
 func GetPublicCertFilePath(serviceName string) string {
-	publicCertPath := fmt.Sprintf("/tmp/%s/public.crt", serviceName)
+	publicCertPath := fmt.Sprintf("/tmp/%s/%s", serviceName, certs.PublicCertFile)
 	return publicCertPath
 }
 
 // GetPrivateKeyFilePath return the path to the key file based for the serviceName
 func GetPrivateKeyFilePath(serviceName string) string {
-	privateKey := fmt.Sprintf("/tmp/%s/private.key", serviceName)
+	privateKey := fmt.Sprintf("/tmp/%s/%s", serviceName, certs.PrivateKeyFile)
 	return privateKey
 }
 

pkg/certs/const.go

@@ -35,6 +35,9 @@ const (
 	// PrivateKeyFile Private key file for HTTPS.
 	PrivateKeyFile = "private.key"
 
+	// CAPublicCertFile  Public certificate file for Certificate authority.
+	CAPublicCertFile = "ca.crt"
+
 	// TLSKeyFile Private key file for HTTPS.
 	TLSKeyFile = "tls.key"
 )

pkg/common/const.go

@@ -36,15 +36,6 @@ const (
 	OperatorRuntimeOpenshift Runtime = "OPENSHIFT"
 	// OperatorRuntimeRancher is the Rancher runtime flag
 	OperatorRuntimeRancher Runtime = "RANCHER"
-
-	// TLSCRT is  name of the field containing tls certificate in secret
-	TLSCRT = "tls.crt"
-
-	// CACRT name of the field containing ca certificate in secret
-	CACRT = "ca.crt"
-
-	// PublicCRT name of the field containing public certificate in secret
-	PublicCRT = "public.crt"
 )
 
 // Runtimes is a map of the supported Kubernetes runtimes

pkg/controller/console.go

@@ -100,12 +100,12 @@ func (c *Controller) checkConsoleSvc(ctx context.Context, tenant *miniov2.Tenant
 
 // generateConsoleTLSCert Issues the Operator Console TLS Certificate
 func (c *Controller) generateConsoleTLSCert() (*string, *string) {
-	return c.generateTLSCert("console", OperatorConsoleTLSSecretName, getConsoleDeploymentName())
+	return c.generateTLSCertificateForService("console", OperatorConsoleTLSSecretName, getConsoleDeploymentName())
 }
 
 func (c *Controller) recreateOperatorConsoleCertsIfRequired(ctx context.Context) error {
 	namespace := miniov2.GetNSFromFile()
-	operatorConsoleTLSSecret, err := c.getTLSSecret(ctx, namespace, OperatorConsoleTLSSecretName)
+	operatorConsoleTLSSecret, err := c.getCertificateSecret(ctx, namespace, OperatorConsoleTLSSecretName)
 	if err != nil {
 		if k8serrors.IsNotFound(err) {
 			klog.V(2).Info("TLS certificate not found. Generating one.")

pkg/controller/controller.go

@@ -148,6 +148,7 @@ func StartOperator(kubeconfig string) {
 		klog.Infof("Watching only namespaces: %s", strings.Join(namespaces.ToSlice(), ","))
 	}
 
+	// kubeInformerFactory := kubeinformers.NewSharedInformerFactoryWithOptions(kubeClient, time.Second*30, kubeinformers.WithNamespace(v2.GetNSFromFile()))
 	kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClient, time.Second*30)
 	minioInformerFactory := informers.NewSharedInformerFactory(controllerClient, time.Second*30)
 	podName := os.Getenv(HostnameEnv)
@@ -163,16 +164,12 @@ func StartOperator(kubeconfig string) {
 		k8sClient,
 		controllerClient,
 		promClient,
-		kubeInformerFactory.Apps().V1().StatefulSets(),
-		kubeInformerFactory.Apps().V1().Deployments(),
-		kubeInformerFactory.Core().V1().Pods(),
-		minioInformerFactory.Minio().V2().Tenants(),
-		minioInformerFactory.Sts().V1beta1().PolicyBindings(),
-		kubeInformerFactory.Core().V1().Services(),
 		hostsTemplate,
 		pkg.Version,
+		kubeInformerFactory,
+		minioInformerFactory.Minio().V2().Tenants(),
+		minioInformerFactory.Sts().V1beta1().PolicyBindings(),
 		minioInformerFactory.Job().V1alpha1().MinIOJobs(),
-		kubeInformerFactory.Batch().V1().Jobs(),
 	)
 
 	go kubeInformerFactory.Start(stopCh)

pkg/controller/csr.go

@@ -29,6 +29,8 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/minio/operator/pkg/certs"
+
 	"github.com/minio/operator/pkg/controller/certificates"
 
 	certificatesV1 "k8s.io/api/certificates/v1"
@@ -257,8 +259,8 @@ func (c *Controller) createSecret(ctx context.Context, tenant *miniov2.Tenant, l
 			},
 		},
 		Data: map[string][]byte{
-			"private.key": pkBytes,
-			"public.crt":  certBytes,
+			certs.PrivateKeyFile: pkBytes,
+			certs.PublicCertFile: certBytes,
 		},
 	}
 	_, err := c.kubeClientSet.CoreV1().Secrets(tenant.Namespace).Create(ctx, secret, metav1.CreateOptions{})

pkg/controller/custom.go

@@ -24,15 +24,17 @@ import (
 	"math"
 	"time"
 
+	"github.com/minio/operator/pkg/certs"
+
 	miniov2 "github.com/minio/operator/pkg/apis/minio.min.io/v2"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 var secretTypePublicKeyNameMap = map[string]string{
-	"kubernetes.io/tls":        "tls.crt",
-	"cert-manager.io/v1":       "tls.crt",
-	"cert-manager.io/v1alpha2": "tls.crt",
+	"kubernetes.io/tls":        certs.TLSCertFile,
+	"cert-manager.io/v1":       certs.TLSCertFile,
+	"cert-manager.io/v1alpha2": certs.TLSCertFile,
 	// Add newer secretTypes and their corresponding values in future
 }
 
@@ -51,7 +53,7 @@ func (c *Controller) getCustomCertificates(ctx context.Context, tenant *miniov2.
 
 	for certType, secrets := range secretsMap {
 		certificates = nil
-		publicKey := "public.crt"
+		publicKey := certs.PublicCertFile
 		// Iterate over TLS secrets and build array of CertificateInfo structure
 		// that will be used to display information about certs
 		for _, secret := range secrets {

pkg/controller/kes.go

@@ -26,6 +26,8 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/minio/operator/pkg/certs"
+
 	"github.com/minio/operator/pkg/controller/certificates"
 
 	corev1 "k8s.io/api/core/v1"
@@ -326,9 +328,9 @@ func (c *Controller) getCertIdentity(ns string, cert *miniov2.LocalCertificateRe
 	}
 	// Store the Identity to be used later during KES container creation
 	if secret.Type == "kubernetes.io/tls" || secret.Type == "cert-manager.io/v1alpha2" || secret.Type == "cert-manager.io/v1" {
-		certbytes = secret.Data["tls.crt"]
+		certbytes = secret.Data[certs.TLSCertFile]
 	} else {
-		certbytes = secret.Data["public.crt"]
+		certbytes = secret.Data[certs.PublicCertFile]
 	}
 
 	// parse the certificate here to generate the identity for this certifcate.

pkg/controller/main-controller.go

@@ -26,6 +26,8 @@ import (
 	"syscall"
 	"time"
 
+	kubeinformers "k8s.io/client-go/informers"
+
 	"github.com/minio/operator/pkg/utils"
 
 	"github.com/minio/madmin-go/v3"
@@ -54,9 +56,6 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
-	appsinformers "k8s.io/client-go/informers/apps/v1"
-	batchv1 "k8s.io/client-go/informers/batch/v1"
-	coreinformers "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
@@ -138,25 +137,25 @@ type Controller struct {
 	// statefulSetListerSynced returns true if the StatefulSet shared informer
 	// has synced at least once.
 	statefulSetListerSynced cache.InformerSynced
-
+	// secretLister returns list/get secrets from a shared informer
+	secretLister corelisters.SecretLister
+	// secretListerSynced returns true if Secret shared informer has synced at least once
+	secretListerSynced cache.InformerSynced
 	// deploymentLister is able to list/get Deployments from a shared
 	// informer's store.
 	deploymentLister appslisters.DeploymentLister
 	// deploymentListerSynced returns true if the Deployment shared informer
 	// has synced at least once.
 	deploymentListerSynced cache.InformerSynced
-
 	// tenantsSynced returns true if the StatefulSet shared informer
 	// has synced at least once.
 	tenantsSynced cache.InformerSynced
-
 	// serviceLister is able to list/get Services from a shared informer's
 	// store.
 	serviceLister corelisters.ServiceLister
 	// serviceListerSynced returns true if the Service shared informer
 	// has synced at least once.
 	serviceListerSynced cache.InformerSynced
-
 	// queue is a rate limited work queue. This is used to queue work to be
 	// processed instead of performing it as soon as a change happens. This
 	// means we can ensure we only process a fixed amount of resources at a
@@ -227,17 +226,20 @@ func NewController(
 	k8sClient client.Client,
 	minioClientSet clientset.Interface,
 	promClient promclientset.Interface,
-	statefulSetInformer appsinformers.StatefulSetInformer,
-	deploymentInformer appsinformers.DeploymentInformer,
-	podInformer coreinformers.PodInformer,
-	tenantInformer informers.TenantInformer,
-	policyBindingInformer stsInformers.PolicyBindingInformer,
-	serviceInformer coreinformers.ServiceInformer,
 	hostsTemplate,
 	operatorVersion string,
-	minioJobinformer jobinformers.MinIOJobInformer,
-	jobInformer batchv1.JobInformer,
+	kubeInformerFactory kubeinformers.SharedInformerFactory,
+	tenantInformer informers.TenantInformer,
+	policyBindingInformer stsInformers.PolicyBindingInformer,
+	minioJobInformer jobinformers.MinIOJobInformer,
 ) *Controller {
+	statefulSetInformer := kubeInformerFactory.Apps().V1().StatefulSets()
+	deploymentInformer := kubeInformerFactory.Apps().V1().Deployments()
+	podInformer := kubeInformerFactory.Core().V1().Pods()
+	serviceInformer := kubeInformerFactory.Core().V1().Services()
+	jobInformer := kubeInformerFactory.Batch().V1().Jobs()
+	secretInformer := kubeInformerFactory.Core().V1().Secrets()
+
 	// Create event broadcaster
 	// Add minio-controller types to the default Kubernetes Scheme so Events can be
 	// logged for minio-controller types.
@@ -263,20 +265,22 @@ func NewController(
 		tenantsSynced:             tenantInformer.Informer().HasSynced,
 		serviceLister:             serviceInformer.Lister(),
 		serviceListerSynced:       serviceInformer.Informer().HasSynced,
-		workqueue:                 queue.NewNamedRateLimitingQueue(MinIOControllerRateLimiter(), "Tenants"),
-		healthCheckQueue:          queue.NewNamedRateLimitingQueue(MinIOControllerRateLimiter(), "TenantsHealth"),
+		secretLister:              secretInformer.Lister(),
+		secretListerSynced:        secretInformer.Informer().HasSynced,
+		workqueue:                 queue.NewRateLimitingQueueWithConfig(MinIOControllerRateLimiter(), queue.RateLimitingQueueConfig{Name: "Tenants"}),
+		healthCheckQueue:          queue.NewRateLimitingQueueWithConfig(MinIOControllerRateLimiter(), queue.RateLimitingQueueConfig{Name: "TenantsHealth"}),
 		recorder:                  recorder,
 		hostsTemplate:             hostsTemplate,
 		operatorVersion:           operatorVersion,
 		policyBindingListerSynced: policyBindingInformer.Informer().HasSynced,
 		controllers: []*JobController{
 			NewJobController(
-				minioJobinformer,
+				minioJobInformer,
 				jobInformer,
 				namespacesToWatch,
 				kubeClientSet,
 				recorder,
-				queue.NewNamedRateLimitingQueue(MinIOControllerRateLimiter(), "MinioJobs"),
+				queue.NewRateLimitingQueueWithConfig(MinIOControllerRateLimiter(), queue.RateLimitingQueueConfig{Name: "MinioJobs"}),
 				k8sClient,
 			),
 		},
@@ -347,14 +351,30 @@ func NewController(
 			oldDepl := old.(*corev1.Pod)
 			if newDepl.ResourceVersion == oldDepl.ResourceVersion {
 				// Periodic resync will send update events for all known Deployments.
-				// Two different versions of the same Deployments will always have different RVs.
+				// Two different versions of the same Pods will always have different RVs.
 				return
 			}
 			controller.handlePodChange(new)
 		},
 		DeleteFunc: controller.handlePodChange,
 	})
 
+	secretInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			controller.handleSecret(obj, nil)
+		},
+		UpdateFunc: func(old, new interface{}) {
+			newSecret := new.(*corev1.Secret)
+			oldSecret := old.(*corev1.Secret)
+			if newSecret.ResourceVersion == oldSecret.ResourceVersion {
+				// Periodic resync will send update events for all known Deployments.
+				// Two different versions of the same secret will always have different RVs.
+				return
+			}
+			controller.handleSecret(new, old)
+		},
+	})
+
 	return controller
 }
 
@@ -415,7 +435,7 @@ func leaderRun(ctx context.Context, c *Controller, threadiness int, stopCh <-cha
 
 	// Wait for the caches to be synced before starting workers
 	klog.Info("Waiting for informer caches to sync")
-	if ok := cache.WaitForCacheSync(stopCh, c.statefulSetListerSynced, c.deploymentListerSynced, c.tenantsSynced, c.policyBindingListerSynced); !ok {
+	if ok := cache.WaitForCacheSync(stopCh, c.statefulSetListerSynced, c.deploymentListerSynced, c.tenantsSynced, c.policyBindingListerSynced, c.secretListerSynced); !ok {
 		panic("failed to wait for caches to sync")
 	}
 	// Wait for the caches to be synced before starting workers
@@ -896,13 +916,6 @@ func (c *Controller) syncHandler(key string) (Result, error) {
 		return WrapResult(Result{}, err)
 	}
 
-	// check if operator-ca-tls has to be updated or re-created in the tenant namespace
-	operatorCATLSExists, err := c.checkOperatorCAForTenant(ctx, tenant)
-	if err != nil {
-		// Don't return here as we get stuck when recreating the stateful set
-		klog.Infof("There was an error while updating the certificate %s", err)
-	}
-
 	// consolidate the status of all pools. this is meant to cover for legacy tenants
 	// this status value is zero only for new tenants or legacy tenants
 	if len(tenant.Status.Pools) == 0 {
@@ -974,7 +987,6 @@ func (c *Controller) syncHandler(key string) (Result, error) {
 				ServiceName:     tenant.MinIOHLServiceName(),
 				HostsTemplate:   c.hostsTemplate,
 				OperatorVersion: c.operatorVersion,
-				OperatorCATLS:   operatorCATLSExists,
 			})
 			ss, err = c.kubeClientSet.AppsV1().StatefulSets(tenant.Namespace).Create(ctx, ss, cOpts)
 			if err != nil {
@@ -1183,7 +1195,6 @@ func (c *Controller) syncHandler(key string) (Result, error) {
 				ServiceName:     tenant.MinIOHLServiceName(),
 				HostsTemplate:   c.hostsTemplate,
 				OperatorVersion: c.operatorVersion,
-				OperatorCATLS:   operatorCATLSExists,
 			})
 			if _, err = c.kubeClientSet.AppsV1().StatefulSets(tenant.Namespace).Update(ctx, ss, uOpts); err != nil {
 				return WrapResult(Result{}, err)
@@ -1232,7 +1243,6 @@ func (c *Controller) syncHandler(key string) (Result, error) {
 			ServiceName:     tenant.MinIOHLServiceName(),
 			HostsTemplate:   c.hostsTemplate,
 			OperatorVersion: c.operatorVersion,
-			OperatorCATLS:   operatorCATLSExists,
 		})
 		// Verify if this pool matches the spec on the tenant (resources, affinity, sidecars, etc)
 		poolMatchesSS, err := poolSSMatchesSpec(expectedStatefulSet, existingStatefulSet)
@@ -1390,6 +1400,29 @@ func (c *Controller) handleObject(obj interface{}) {
 	}
 }
 
+func (c *Controller) handleSecret(obj interface{}, oldObj interface{}) {
+	ns := miniov2.GetNSFromFile()
+	var secret *corev1.Secret
+	var ok bool
+	if secret, ok = obj.(*corev1.Secret); !ok {
+		runtime.HandleError(fmt.Errorf("error decoding object, invalid type"))
+		return
+	}
+	// Observe secrets in the Operator namespace
+	if secret.Namespace == ns {
+		// a secret with prefix "operator-ca-tls" changed, reload all trusted CA certificates
+		if strings.HasPrefix(secret.Name, OperatorCATLSSecretName) {
+			klog.Infof("secret '%s' found, adding TLS certs in it to trusted CA's", secret.Name)
+			var oldSecret *corev1.Secret
+			if oldSecret != nil {
+				oldSecret = oldObj.(*corev1.Secret)
+			}
+			// Add new certificates to Transport Certs if any changed
+			c.TrustTLSCertificatesInSecretIfChanged(secret, oldSecret)
+		}
+	}
+}
+
 // MinIOControllerRateLimiter is a no-arg constructor for a default rate limiter for a workqueue for our controller.
 // both overall and per-item rate limiting.  The overall is a token bucket and the per-item is exponential
 func MinIOControllerRateLimiter() queue.RateLimiter {