pkg/metrics: Add ownerReference to metrics Service object (#1037)

* pkg/metrics: Add ownerRef to Service

This allows the Service cleanup to happen when owner is
deleted. Owner is detected by tracing up until last owner is
found. If no owner is found, the pod is set as owner.

* Gopkg.lock: Rerun dep ensure

* pkg/metrics: Update Service if it exists

When Service already exists but the OwnerRef might change,
we must make sure to the propagate those changes to the Service object
by updating the object.

Author: lilic

commands/operator-sdk/cmd/test/cluster.go

@@ -24,7 +24,6 @@ import (
 	k8sInternal "github.com/operator-framework/operator-sdk/internal/util/k8sutil"
 	"github.com/operator-framework/operator-sdk/internal/util/projutil"
 	"github.com/operator-framework/operator-sdk/pkg/k8sutil"
-	"github.com/operator-framework/operator-sdk/pkg/leader"
 	"github.com/operator-framework/operator-sdk/pkg/scaffold"
 	"github.com/operator-framework/operator-sdk/pkg/scaffold/ansible"
 	"github.com/operator-framework/operator-sdk/pkg/test"
@@ -113,7 +112,7 @@ func testClusterFunc(cmd *cobra.Command, args []string) error {
 					Name:  k8sutil.OperatorNameEnvVar,
 					Value: "test-operator",
 				}, {
-					Name:      leader.PodNameEnv,
+					Name:      k8sutil.PodNameEnvVar,
 					ValueFrom: &v1.EnvVarSource{FieldRef: &v1.ObjectFieldSelector{FieldPath: "metadata.name"}},
 				}},
 			}},

pkg/k8sutil/constants.go

@@ -25,6 +25,10 @@ const (
 	WatchNamespaceEnvVar = "WATCH_NAMESPACE"
 
 	// OperatorNameEnvVar is the constant for env variable OPERATOR_NAME
-	// wich is the name of the current operator
+	// which is the name of the current operator
 	OperatorNameEnvVar = "OPERATOR_NAME"
+
+	// PodNameEnvVar is the contast for env variable POD_NAME
+	// which is the name of the current pod.
+	PodNameEnvVar = "POD_NAME"
 )

pkg/k8sutil/k8sutil.go

@@ -15,12 +15,16 @@
 package k8sutil
 
 import (
+	"context"
 	"fmt"
 	"io/ioutil"
 	"os"
 	"strings"
 
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	discovery "k8s.io/client-go/discovery"
+	crclient "sigs.k8s.io/controller-runtime/pkg/client"
 	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
 )
 
@@ -83,3 +87,32 @@ func ResourceExists(dc discovery.DiscoveryInterface, apiGroupVersion, kind strin
 	}
 	return false, nil
 }
+
+// GetPod returns a Pod object that corresponds to the pod in which the code
+// is currently running.
+// It expects the environment variable POD_NAME to be set by the downwards API.
+func GetPod(ctx context.Context, client crclient.Client, ns string) (*corev1.Pod, error) {
+	podName := os.Getenv(PodNameEnvVar)
+	if podName == "" {
+		return nil, fmt.Errorf("required env %s not set, please configure downward API", PodNameEnvVar)
+	}
+
+	log.V(1).Info("Found podname", "Pod.Name", podName)
+
+	pod := &corev1.Pod{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "v1",
+			Kind:       "Pod",
+		},
+	}
+	key := crclient.ObjectKey{Namespace: ns, Name: podName}
+	err := client.Get(ctx, key, pod)
+	if err != nil {
+		log.Error(err, "Failed to get Pod", "Pod.Namespace", ns, "Pod.Name", podName)
+		return nil, err
+	}
+
+	log.V(1).Info("Found Pod", "Pod.Namespace", ns, "Pod.Name", pod.Name)
+
+	return pod, nil
+}

pkg/leader/leader.go

@@ -16,8 +16,6 @@ package leader
 
 import (
 	"context"
-	"fmt"
-	"os"
 	"time"
 
 	"github.com/operator-framework/operator-sdk/pkg/k8sutil"
@@ -37,8 +35,6 @@ var log = logf.Log.WithName("leader")
 // attempts to become the leader.
 const maxBackoffInterval = time.Second * 16
 
-const PodNameEnv = "POD_NAME"
-
 // Become ensures that the current pod is the leader within its namespace. If
 // run outside a cluster, it will skip leader election and return nil. It
 // continuously tries to create a ConfigMap with the provided name and the
@@ -143,24 +139,8 @@ func Become(ctx context.Context, lockName string) error {
 // this code is currently running.
 // It expects the environment variable POD_NAME to be set by the downwards API
 func myOwnerRef(ctx context.Context, client crclient.Client, ns string) (*metav1.OwnerReference, error) {
-	podName := os.Getenv(PodNameEnv)
-	if podName == "" {
-		return nil, fmt.Errorf("required env %s not set, please configure downward API", PodNameEnv)
-	}
-
-	log.V(1).Info("Found podname", "Pod.Name", podName)
-
-	myPod := &corev1.Pod{
-		TypeMeta: metav1.TypeMeta{
-			APIVersion: "v1",
-			Kind:       "Pod",
-		},
-	}
-
-	key := crclient.ObjectKey{Namespace: ns, Name: podName}
-	err := client.Get(ctx, key, myPod)
+	myPod, err := k8sutil.GetPod(ctx, client, ns)
 	if err != nil {
-		log.Error(err, "Failed to get pod", "Pod.Namespace", ns, "Pod.Name", podName)
 		return nil, err
 	}
 

pkg/metrics/metrics.go

@@ -23,72 +23,78 @@ import (
 	v1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
-	"k8s.io/client-go/rest"
 	crclient "sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/config"
 	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
 )
 
 var log = logf.Log.WithName("metrics")
 
-// PrometheusPortName defines the port name used in kubernetes deployment and service resources
-const PrometheusPortName = "metrics"
+var trueVar = true
+
+const (
+	// PrometheusPortName defines the port name used in the metrics Service.
+	PrometheusPortName = "metrics"
+)
 
 // ExposeMetricsPort creates a Kubernetes Service to expose the passed metrics port.
 func ExposeMetricsPort(ctx context.Context, port int32) (*v1.Service, error) {
+	client, err := createClient()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create new client: %v", err)
+	}
 	// We do not need to check the validity of the port, as controller-runtime
 	// would error out and we would never get to this stage.
-	s, err := initOperatorService(port, PrometheusPortName)
+	s, err := initOperatorService(ctx, client, port, PrometheusPortName)
 	if err != nil {
 		if err == k8sutil.ErrNoNamespace {
 			log.Info("Skipping metrics Service creation; not running in a cluster.")
 			return nil, nil
 		}
 		return nil, fmt.Errorf("failed to initialize service object for metrics: %v", err)
 	}
-	service, err := createService(ctx, s)
+	service, err := createOrUpdateService(ctx, client, s)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create or get service for metrics: %v", err)
 	}
 
 	return service, nil
 }
 
-func createService(ctx context.Context, s *v1.Service) (*v1.Service, error) {
-	config, err := rest.InClusterConfig()
-	if err != nil {
-		return nil, err
-	}
-
-	client, err := crclient.New(config, crclient.Options{})
-	if err != nil {
-		return nil, err
-	}
-
+func createOrUpdateService(ctx context.Context, client crclient.Client, s *v1.Service) (*v1.Service, error) {
 	if err := client.Create(ctx, s); err != nil {
 		if !apierrors.IsAlreadyExists(err) {
 			return nil, err
 		}
-		// Get existing Service and return it
+		// Service already exists, we want to update it
+		// as we do not know if any fields might have changed.
 		existingService := &v1.Service{}
 		err := client.Get(ctx, types.NamespacedName{
 			Name:      s.Name,
 			Namespace: s.Namespace,
 		}, existingService)
+
+		s.ResourceVersion = existingService.ResourceVersion
+		if existingService.Spec.Type == v1.ServiceTypeClusterIP {
+			s.Spec.ClusterIP = existingService.Spec.ClusterIP
+		}
+		err = client.Update(ctx, s)
 		if err != nil {
 			return nil, err
 		}
-		log.Info("Metrics Service object already exists", "name", existingService.Name)
+		log.V(1).Info("Metrics Service object updated", "Service.Name", s.Name, "Service.Namespace", s.Namespace)
 		return existingService, nil
 	}
 
-	log.Info("Metrics Service object created", "name", s.Name)
+	log.Info("Metrics Service object created", "Service.Name", s.Name, "Service.Namespace", s.Namespace)
 	return s, nil
 }
 
 // initOperatorService returns the static service which exposes specifed port.
-func initOperatorService(port int32, portName string) (*v1.Service, error) {
+func initOperatorService(ctx context.Context, client crclient.Client, port int32, portName string) (*v1.Service, error) {
 	operatorName, err := k8sutil.GetOperatorName()
 	if err != nil {
 		return nil, err
@@ -97,11 +103,14 @@ func initOperatorService(port int32, portName string) (*v1.Service, error) {
 	if err != nil {
 		return nil, err
 	}
+
+	label := map[string]string{"name": operatorName}
+
 	service := &v1.Service{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      operatorName,
 			Namespace: namespace,
-			Labels:    map[string]string{"name": operatorName},
+			Labels:    label,
 		},
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "Service",
@@ -119,8 +128,72 @@ func initOperatorService(port int32, portName string) (*v1.Service, error) {
 					Name: portName,
 				},
 			},
-			Selector: map[string]string{"name": operatorName},
+			Selector: label,
 		},
 	}
+
+	ownRef, err := getPodOwnerRef(ctx, client, namespace)
+	if err != nil {
+		return nil, err
+	}
+	service.SetOwnerReferences([]metav1.OwnerReference{*ownRef})
+
 	return service, nil
 }
+
+func getPodOwnerRef(ctx context.Context, client crclient.Client, ns string) (*metav1.OwnerReference, error) {
+	// Get current Pod the operator is running in
+	pod, err := k8sutil.GetPod(ctx, client, ns)
+	if err != nil {
+		return nil, err
+	}
+	podOwnerRefs := metav1.NewControllerRef(pod, pod.GroupVersionKind())
+	// Get Owner that the Pod belongs to
+	ownerRef := metav1.GetControllerOf(pod)
+	finalOwnerRef, err := findFinalOwnerRef(ctx, client, ns, ownerRef)
+	if err != nil {
+		return nil, err
+	}
+	if finalOwnerRef != nil {
+		return finalOwnerRef, nil
+	}
+
+	// Default to returning Pod as the Owner
+	return podOwnerRefs, nil
+}
+
+// findFinalOwnerRef tries to locate the final controller/owner based on the owner reference provided.
+func findFinalOwnerRef(ctx context.Context, client crclient.Client, ns string, ownerRef *metav1.OwnerReference) (*metav1.OwnerReference, error) {
+	if ownerRef == nil {
+		return nil, nil
+	}
+
+	obj := &unstructured.Unstructured{}
+	obj.SetAPIVersion(ownerRef.APIVersion)
+	obj.SetKind(ownerRef.Kind)
+	err := client.Get(ctx, types.NamespacedName{Namespace: ns, Name: ownerRef.Name}, obj)
+	if err != nil {
+		return nil, err
+	}
+	newOwnerRef := metav1.GetControllerOf(obj)
+	if newOwnerRef != nil {
+		return findFinalOwnerRef(ctx, client, ns, newOwnerRef)
+	}
+
+	log.V(1).Info("Pods owner found", "Kind", ownerRef.Kind, "Name", ownerRef.Name, "Namespace", ns)
+	return ownerRef, nil
+}
+
+func createClient() (crclient.Client, error) {
+	config, err := config.GetConfig()
+	if err != nil {
+		return nil, err
+	}
+
+	client, err := crclient.New(config, crclient.Options{})
+	if err != nil {
+		return nil, err
+	}
+
+	return client, nil
+}