add metadata client, switch to using PartialObjectMetadata

Author: piceri

cmd/deployment-tracker/main.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/github/deployment-tracker/internal/controller"
+	"k8s.io/client-go/metadata"
 
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"k8s.io/client-go/kubernetes"
@@ -112,6 +113,14 @@ func main() {
 		os.Exit(1)
 	}
 
+	// Create metadata client
+	metadataClient, err := metadata.NewForConfig(k8sCfg)
+	if err != nil {
+		slog.Error("Error creating Kubernetes metadata client",
+			"error", err)
+		os.Exit(1)
+	}
+
 	// Start the metrics server
 	var promSrv = &http.Server{
 		Addr:              ":" + metricsPort,
@@ -151,7 +160,7 @@ func main() {
 		cancel()
 	}()
 
-	cntrl, err := controller.New(clientset, namespace, excludeNamespaces, &cntrlCfg)
+	cntrl, err := controller.New(clientset, metadataClient, namespace, excludeNamespaces, &cntrlCfg)
 	if err != nil {
 		slog.Error("Failed to create controller",
 			"error", err)

internal/controller/controller.go

@@ -12,7 +12,9 @@ import (
 	"github.com/github/deployment-tracker/pkg/deploymentrecord"
 	"github.com/github/deployment-tracker/pkg/image"
 	"github.com/github/deployment-tracker/pkg/metrics"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/metadata"
 
 	corev1 "k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
@@ -48,19 +50,20 @@ type AggregatePodMetadata struct {
 
 // Controller is the Kubernetes controller for tracking deployments.
 type Controller struct {
-	clientset   kubernetes.Interface
-	podInformer cache.SharedIndexInformer
-	workqueue   workqueue.TypedRateLimitingInterface[PodEvent]
-	apiClient   *deploymentrecord.Client
-	cfg         *Config
+	clientset      kubernetes.Interface
+	metadataClient metadata.Interface
+	podInformer    cache.SharedIndexInformer
+	workqueue      workqueue.TypedRateLimitingInterface[PodEvent]
+	apiClient      *deploymentrecord.Client
+	cfg            *Config
 	// best effort cache to avoid redundant posts
 	// post requests are idempotent, so if this cache fails due to
 	// restarts or other events, nothing will break.
 	observedDeployments sync.Map
 }
 
 // New creates a new deployment tracker controller.
-func New(clientset kubernetes.Interface, namespace string, excludeNamespaces string, cfg *Config) (*Controller, error) {
+func New(clientset kubernetes.Interface, metadataClient metadata.Interface, namespace string, excludeNamespaces string, cfg *Config) (*Controller, error) {
 	// Create informer factory
 	factory := createInformerFactory(clientset, namespace, excludeNamespaces)
 
@@ -92,11 +95,12 @@ func New(clientset kubernetes.Interface, namespace string, excludeNamespaces str
 	}
 
 	cntrl := &Controller{
-		clientset:   clientset,
-		podInformer: podInformer,
-		workqueue:   queue,
-		apiClient:   apiClient,
-		cfg:         cfg,
+		clientset:      clientset,
+		metadataClient: metadataClient,
+		podInformer:    podInformer,
+		workqueue:      queue,
+		apiClient:      apiClient,
+		cfg:            cfg,
 	}
 
 	// Add event handlers to the informer
@@ -425,7 +429,7 @@ func (c *Controller) recordContainer(ctx context.Context, pod *corev1.Pod, conta
 	// Gather aggregate metadata for adds/updates
 	var runtimeRisks []deploymentrecord.RuntimeRisk
 	if status != deploymentrecord.StatusDecommissioned {
-		metadata := c.aggregateMetadata(ctx, pod)
+		metadata := c.aggregateMetadata(ctx, podToPartialMetadata(pod))
 		for risk := range metadata.RuntimeRisks {
 			runtimeRisks = append(runtimeRisks, risk)
 		}
@@ -493,11 +497,11 @@ func (c *Controller) recordContainer(ctx context.Context, pod *corev1.Pod, conta
 }
 
 // aggregateRuntimeRisks aggregates metadata for a pod and its owners.
-func (c *Controller) aggregateMetadata(ctx context.Context, obj metav1.Object) AggregatePodMetadata {
+func (c *Controller) aggregateMetadata(ctx context.Context, obj *metav1.PartialObjectMetadata) AggregatePodMetadata {
 	metadata := AggregatePodMetadata{
 		RuntimeRisks: make(map[deploymentrecord.RuntimeRisk]bool),
 	}
-	queue := []metav1.Object{obj}
+	queue := []*metav1.PartialObjectMetadata{obj}
 	visited := make(map[types.UID]bool)
 
 	for len(queue) > 0 {
@@ -522,11 +526,11 @@ func (c *Controller) aggregateMetadata(ctx context.Context, obj metav1.Object) A
 
 // collectRuntimeRisksFromOwners takes a current object and looks up its owners, adding them to the queue for processing
 // to collect their metadata.
-func (c *Controller) addOwnersToQueue(ctx context.Context, current metav1.Object, queue *[]metav1.Object) {
+func (c *Controller) addOwnersToQueue(ctx context.Context, current *metav1.PartialObjectMetadata, queue *[]*metav1.PartialObjectMetadata) {
 	ownerRefs := current.GetOwnerReferences()
 
 	for _, owner := range ownerRefs {
-		ownerObj, err := c.getOwnerObject(ctx, current.GetNamespace(), owner)
+		ownerObj, err := c.getOwnerMetadata(ctx, current.GetNamespace(), owner)
 		if err != nil {
 			slog.Warn("Failed to get owner object for metadata collection",
 				"namespace", current.GetNamespace(),
@@ -545,29 +549,36 @@ func (c *Controller) addOwnersToQueue(ctx context.Context, current metav1.Object
 	}
 }
 
-// getOwnerObject retrieves the owner object based on its kind, namespace, and name.
-func (c *Controller) getOwnerObject(ctx context.Context, namespace string, owner metav1.OwnerReference) (metav1.Object, error) {
+// getOwnerMetadata retrieves partial object metadata for an owner ref.
+func (c *Controller) getOwnerMetadata(ctx context.Context, namespace string, owner metav1.OwnerReference) (*metav1.PartialObjectMetadata, error) {
+	var gvr schema.GroupVersionResource
+
 	switch owner.Kind {
 	case "ReplicaSet":
-		rs, err := c.clientset.AppsV1().ReplicaSets(namespace).Get(ctx, owner.Name, metav1.GetOptions{})
-		if err != nil {
-			return nil, err
+		gvr = schema.GroupVersionResource{
+			Group:    "apps",
+			Version:  "v1",
+			Resource: "replicasets",
 		}
-		return rs, nil
 	case "Deployment":
-		deployment, err := c.clientset.AppsV1().Deployments(namespace).Get(ctx, owner.Name, metav1.GetOptions{})
-		if err != nil {
-			return nil, err
+		gvr = schema.GroupVersionResource{
+			Group:    "apps",
+			Version:  "v1",
+			Resource: "deployments",
 		}
-		return deployment, nil
 	default:
-		// Unsupported kinds
 		slog.Debug("Unsupported owner kind for runtime risk collection",
 			"kind", owner.Kind,
 			"name", owner.Name,
 		)
 		return nil, nil
 	}
+
+	obj, err := c.metadataClient.Resource(gvr).Namespace(namespace).Get(ctx, owner.Name, metav1.GetOptions{})
+	if err != nil {
+		return nil, err
+	}
+	return obj, nil
 }
 
 func getCacheKey(dn, digest string) string {
@@ -679,7 +690,7 @@ func getDeploymentName(pod *corev1.Pod) string {
 }
 
 // getMetadataFromObject extracts metadata from an object.
-func getMetadataFromObject(obj metav1.Object, metadata AggregatePodMetadata) {
+func getMetadataFromObject(obj *metav1.PartialObjectMetadata, metadata AggregatePodMetadata) {
 	annotations := obj.GetAnnotations()
 	if risks, exists := annotations[RuntimeRiskAnnotationKey]; exists {
 		for _, risk := range strings.Split(risks, ",") {
@@ -690,3 +701,13 @@ func getMetadataFromObject(obj metav1.Object, metadata AggregatePodMetadata) {
 		}
 	}
 }
+
+func podToPartialMetadata(pod *corev1.Pod) *metav1.PartialObjectMetadata {
+	return &metav1.PartialObjectMetadata{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "v1",
+			Kind:       "Pod",
+		},
+		ObjectMeta: pod.ObjectMeta,
+	}
+}