add AggregatePodMetadata to track metadata of pod and its owners, add runtime risk tracking

Author: piceri

.gitignore

@@ -1,2 +1,3 @@
 *~
 /deployment-tracker
+.idea/

README.md

@@ -23,6 +23,7 @@ deployment records to GitHub's artifact metadata API.
 - **Real-time tracking**: Sends deployment records when pods are
   created or deleted
 - **Graceful shutdown**: Properly drains work queue before terminating
+- **Runtime risks**: Track runtime risks through annotations
 
 ## How It Works
 
@@ -82,6 +83,17 @@ The `DN_TEMPLATE` supports the following placeholders:
 - `{{deploymentName}}` - Name of the owning Deployment
 - `{{containerName}}` - Container name
 
+## Runtime Risks
+
+You can track runtime risks through annotations. Add the annotation `github.com/runtime-risks`, with a comma-separated list of supported runtime risk values. Annotations are aggregated from the pod and its owner reference objects. 
+
+Currently supported runtime risks:
+- `critical-resource`
+- `lateral-movement`
+- `internet-exposed`
+- `sensitive-data`
+
+
 ## Kubernetes Deployment
 
 A complete deployment manifest is provided in `deploy/manifest.yaml`

deploy/manifest.yaml

@@ -17,9 +17,12 @@ rules:
   - apiGroups: [""]
     resources: ["pods"]
     verbs: ["get", "list", "watch"]
-  - apiGroups: ["apps"]
-    resources: ["deployments"]
-    verbs: ["get"]
+  - apiGroups: [ "apps" ]
+    resources: [ "deployments" ]
+    verbs: [ "get" ]
+  - apiGroups: [ "apps" ]
+    resources: [ "replicasets" ]
+    verbs: [ "get" ]
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
@@ -66,7 +69,7 @@ spec:
             - name: PHYSICAL_ENVIRONMENT
               value: "iad-moda1"
             - name: CLUSTER
-              value: "kommendorkapten"
+              value: "test-cluster"
             - name: BASE_URL
               value: "http://artifact-registry.artifact-registry.svc.cluster.local:9090"
           resources:

internal/controller/controller.go

@@ -29,6 +29,8 @@ const (
 	EventCreated = "CREATED"
 	// EventDeleted indicates that a pod has been deleted.
 	EventDeleted = "DELETED"
+	// RuntimeRiskAnnotationKey represents the annotation key for runtime risks.
+	RuntimeRiskAnnotationKey = "github.com/runtime-risks"
 )
 
 // PodEvent represents a pod event to be processed.
@@ -38,6 +40,11 @@ type PodEvent struct {
 	DeletedPod *corev1.Pod // Only populated for delete events
 }
 
+// AggregatePodMetadata represents combined metadata for a pod and its ownership hierarchy.
+type AggregatePodMetadata struct {
+	RuntimeRisks map[deploymentrecord.RuntimeRisk]bool
+}
+
 // Controller is the Kubernetes controller for tracking deployments.
 type Controller struct {
 	clientset   kubernetes.Interface
@@ -414,6 +421,15 @@ func (c *Controller) recordContainer(ctx context.Context, pod *corev1.Pod, conta
 	// Extract image name and tag
 	imageName, version := image.ExtractName(container.Image)
 
+	// Gather aggregate metadata
+	metadata := c.aggregateMetadata(ctx, pod)
+	var runtimeRisks []deploymentrecord.RuntimeRisk
+	if status != deploymentrecord.StatusDecommissioned {
+		for risk := range metadata.RuntimeRisks {
+			runtimeRisks = append(runtimeRisks, risk)
+		}
+	}
+
 	// Create deployment record
 	record := deploymentrecord.NewDeploymentRecord(
 		imageName,
@@ -424,6 +440,7 @@ func (c *Controller) recordContainer(ctx context.Context, pod *corev1.Pod, conta
 		c.cfg.Cluster,
 		status,
 		dn,
+		runtimeRisks,
 	)
 
 	if err := c.apiClient.PostOne(ctx, record); err != nil {
@@ -457,6 +474,7 @@ func (c *Controller) recordContainer(ctx context.Context, pod *corev1.Pod, conta
 		"name", record.Name,
 		"deployment_name", record.DeploymentName,
 		"status", record.Status,
+		"runtime_risks", record.RuntimeRisks,
 		"digest", record.Digest,
 	)
 
@@ -473,6 +491,82 @@ func (c *Controller) recordContainer(ctx context.Context, pod *corev1.Pod, conta
 	return nil
 }
 
+// aggregateRuntimeRisks aggregates metadata for a pod and its owners.
+func (c *Controller) aggregateMetadata(ctx context.Context, obj metav1.Object) AggregatePodMetadata {
+	metadata := AggregatePodMetadata{
+		RuntimeRisks: make(map[deploymentrecord.RuntimeRisk]bool),
+	}
+	visited := make(map[string]bool)
+
+	getMetadataFromObject(obj, metadata)
+	c.getMetadataFromOwners(ctx, obj, metadata, visited)
+
+	return metadata
+}
+
+// collectRuntimeRisksFromOwners recursively collects metadata from owner references
+// in the ownership chain
+// Visited map prevents infinite recursion on circular ownership references.
+func (c *Controller) getMetadataFromOwners(ctx context.Context, obj metav1.Object, metadata AggregatePodMetadata, visited map[string]bool) {
+	ownerRefs := obj.GetOwnerReferences()
+
+	for _, owner := range ownerRefs {
+		ownerKey := fmt.Sprintf("%s/%s/%s", obj.GetNamespace(), owner.Kind, owner.Name)
+		if visited[ownerKey] {
+			slog.Debug("Cycle detected in ownership chain, skipping",
+				"namespace", obj.GetNamespace(),
+				"owner_kind", owner.Kind,
+				"owner_name", owner.Name,
+			)
+			continue
+		}
+		visited[ownerKey] = true
+
+		ownerObj, err := c.getOwnerObject(ctx, obj.GetNamespace(), owner)
+		if err != nil {
+			slog.Warn("Failed to get owner object for metadata collection",
+				"namespace", obj.GetNamespace(),
+				"owner_kind", owner.Kind,
+				"owner_name", owner.Name,
+				"error", err,
+			)
+			continue
+		}
+
+		if ownerObj == nil {
+			continue
+		}
+
+		getMetadataFromObject(ownerObj, metadata)
+		c.getMetadataFromOwners(ctx, ownerObj, metadata, visited)
+	}
+}
+
+// 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) {
+	switch owner.Kind {
+	case "ReplicaSet":
+		rs, err := c.clientset.AppsV1().ReplicaSets(namespace).Get(ctx, owner.Name, metav1.GetOptions{})
+		if err != nil {
+			return nil, err
+		}
+		return rs, nil
+	case "Deployment":
+		deployment, err := c.clientset.AppsV1().Deployments(namespace).Get(ctx, owner.Name, metav1.GetOptions{})
+		if err != nil {
+			return nil, err
+		}
+		return deployment, nil
+	default:
+		// Unsupported kinds
+		slog.Debug("Unsupported owner kind for runtime risk collection",
+			"kind", owner.Kind,
+			"name", owner.Name,
+		)
+		return nil, nil
+	}
+}
+
 func getCacheKey(dn, digest string) string {
 	return dn + "||" + digest
 }
@@ -580,3 +674,16 @@ func getDeploymentName(pod *corev1.Pod) string {
 	}
 	return ""
 }
+
+// getMetadataFromObject extracts metadata from an object.
+func getMetadataFromObject(obj metav1.Object, metadata AggregatePodMetadata) {
+	annotations := obj.GetAnnotations()
+	if risks, exists := annotations[RuntimeRiskAnnotationKey]; exists {
+		for _, risk := range strings.Split(risks, ",") {
+			r := deploymentrecord.ValidateRuntimeRisk(strings.TrimSpace(risk))
+			if r != "" {
+				metadata.RuntimeRisks[r] = true
+			}
+		}
+	}
+}

pkg/deploymentrecord/record.go

@@ -6,24 +6,35 @@ const (
 	StatusDecommissioned = "decommissioned"
 )
 
+// Runtime risks for deployment records.
+type RuntimeRisk string
+
+const (
+	CriticalResource RuntimeRisk = "critical-resource"
+	InternetExposed  RuntimeRisk = "internet-exposed"
+	LateralMovement  RuntimeRisk = "lateral-movement"
+	SensitiveData    RuntimeRisk = "sensitive-data"
+)
+
 // DeploymentRecord represents a deployment event record.
 type DeploymentRecord struct {
-	Name                string `json:"name"`
-	Digest              string `json:"digest"`
-	Version             string `json:"version,omitempty"`
-	LogicalEnvironment  string `json:"logical_environment"`
-	PhysicalEnvironment string `json:"physical_environment"`
-	Cluster             string `json:"cluster"`
-	Status              string `json:"status"`
-	DeploymentName      string `json:"deployment_name"`
+	Name                string        `json:"name"`
+	Digest              string        `json:"digest"`
+	Version             string        `json:"version,omitempty"`
+	LogicalEnvironment  string        `json:"logical_environment"`
+	PhysicalEnvironment string        `json:"physical_environment"`
+	Cluster             string        `json:"cluster"`
+	Status              string        `json:"status"`
+	DeploymentName      string        `json:"deployment_name"`
+	RuntimeRisks        []RuntimeRisk `json:"runtime_risks,omitempty"`
 }
 
 // NewDeploymentRecord creates a new DeploymentRecord with the given status.
 // Status must be either StatusDeployed or StatusDecommissioned.
 //
 //nolint:revive
 func NewDeploymentRecord(name, digest, version, logicalEnv, physicalEnv,
-	cluster, status, deploymentName string) *DeploymentRecord {
+	cluster, status, deploymentName string, runtimeRisks []RuntimeRisk) *DeploymentRecord {
 	// Validate status
 	if status != StatusDeployed && status != StatusDecommissioned {
 		status = StatusDeployed // default to deployed if invalid
@@ -38,5 +49,18 @@ func NewDeploymentRecord(name, digest, version, logicalEnv, physicalEnv,
 		Cluster:             cluster,
 		Status:              status,
 		DeploymentName:      deploymentName,
+		RuntimeRisks:        runtimeRisks,
+	}
+}
+
+// ValidateRuntimeRisk confirms is string is a valid runtime risk,
+// then returns the canonical runtime risk constant if valid, empty string otherwise.
+func ValidateRuntimeRisk(risk string) RuntimeRisk {
+	r := RuntimeRisk(risk)
+	switch r {
+	case CriticalResource, InternetExposed, LateralMovement, SensitiveData:
+		return r
+	default:
+		return ""
 	}
 }