Merge branch 'main' into niharikag09/otel-url-template-resource-name

Author: niharikag09

pkg/collector/corechecks/containerlifecycle/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "handler_container_creation.go",
         "handler_container_termination.go",
         "handler_pod_creation.go",
+        "handler_pod_state.go",
         "handler_pod_termination.go",
         "handler_task_termination.go",
         "processor.go",

pkg/collector/corechecks/containerlifecycle/handler_pod_state.go

@@ -0,0 +1,191 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+package containerlifecycle
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/DataDog/agent-payload/v5/contlcycle"
+
+	workloadmeta "github.com/DataDog/datadog-agent/comp/core/workloadmeta/def"
+	types "github.com/DataDog/datadog-agent/pkg/containerlifecycle"
+)
+
+// podShadow is the last-observed state of a pod's phase and conditions
+type podShadow struct {
+	mu         sync.Mutex
+	phase      string
+	conditions map[string]workloadmeta.KubernetesPodCondition
+}
+
+// PodStateHandler handles pod phase and condition transition events.
+type PodStateHandler struct {
+	mu sync.Mutex
+	// shadow maps pod UID to its last-observed phase and conditions.
+	shadow map[string]*podShadow
+}
+
+// NewPodStateHandler returns an initialized PodStateHandler.
+func NewPodStateHandler() *PodStateHandler {
+	return &PodStateHandler{shadow: make(map[string]*podShadow)}
+}
+
+// String returns a human-readable name for the handler.
+func (h *PodStateHandler) String() string {
+	return "PodStateHandler"
+}
+
+// CanHandle reports whether this handler processes the given event.
+func (h *PodStateHandler) CanHandle(ev workloadmeta.Event, _ workloadmeta.Source) bool {
+	return ev.Entity.GetID().Kind == workloadmeta.KindKubernetesPod
+}
+
+// Handle builds a LifecycleEvent per detected phase or condition transition for the pod,
+// and clears the pod's shadow state once workloadmeta reports it as unset.
+func (h *PodStateHandler) Handle(ev workloadmeta.Event) ([]LifecycleEvent, error) {
+	podUID := ev.Entity.GetID().ID
+
+	if ev.Type == workloadmeta.EventTypeUnset {
+		h.mu.Lock()
+		delete(h.shadow, podUID)
+		h.mu.Unlock()
+		return nil, nil
+	}
+
+	pod, ok := ev.Entity.(*workloadmeta.KubernetesPod)
+	if !ok {
+		return nil, fmt.Errorf("expected *workloadmeta.KubernetesPod, got %T", ev.Entity)
+	}
+
+	h.mu.Lock()
+	shadow, ok := h.shadow[podUID]
+	if !ok {
+		shadow = &podShadow{conditions: make(map[string]workloadmeta.KubernetesPodCondition)}
+		h.shadow[podUID] = shadow
+	}
+	h.mu.Unlock()
+
+	shadow.mu.Lock()
+	defer shadow.mu.Unlock()
+
+	var transitions []*contlcycle.PodStateTransition
+
+	// If the pod phase has changed, add a transition event.
+	if phaseChanged(shadow.phase, pod.Phase) {
+		var lastObserved *contlcycle.PodStatusValue
+		if shadow.phase != "" {
+			lastObserved = &contlcycle.PodStatusValue{Value: &contlcycle.PodStatusValue_Phase{Phase: shadow.phase}}
+		}
+
+		transitions = append(transitions, &contlcycle.PodStateTransition{
+			Field:               contlcycle.PodStatusField_POD_STATUS_FIELD_PHASE,
+			LastObservedState:   lastObserved,
+			NewState:            &contlcycle.PodStatusValue{Value: &contlcycle.PodStatusValue_Phase{Phase: pod.Phase}},
+			TransitionTimestamp: time.Now().Unix(),
+			Precision:           contlcycle.Precision_PRECISION_APPROXIMATE,
+			MissedIntermediate:  contlcycle.MissedIntermediate_MISSED_INTERMEDIATE_UNKNOWABLE,
+		})
+
+		shadow.phase = pod.Phase
+	}
+
+	// If any condition has changed, add a transition event.
+	for _, condition := range pod.Conditions {
+		priorCondition, seen := shadow.conditions[condition.Type]
+		if conditionUnchanged(seen, priorCondition, condition) {
+			shadow.conditions[condition.Type] = condition
+			continue
+		}
+
+		missedIntermediate := conditionMissedIntermediate(seen, priorCondition, condition)
+
+		var lastObserved *contlcycle.PodStatusValue
+		if seen {
+			lastObserved = &contlcycle.PodStatusValue{Value: &contlcycle.PodStatusValue_Condition{
+				Condition: conditionToModel(priorCondition),
+			}}
+		}
+
+		// If the condition has no last transition time, it's indicative of a bug.
+		// We should probably still report it, but with PRECISION_APPROXIMATE.
+		if condition.LastTransitionTime.IsZero() {
+			transitions = append(transitions, &contlcycle.PodStateTransition{
+				Field:               contlcycle.PodStatusField_POD_STATUS_FIELD_CONDITION,
+				LastObservedState:   lastObserved,
+				NewState:            &contlcycle.PodStatusValue{Value: &contlcycle.PodStatusValue_Condition{Condition: conditionToModel(condition)}},
+				TransitionTimestamp: time.Now().Unix(),
+				Precision:           contlcycle.Precision_PRECISION_APPROXIMATE,
+				MissedIntermediate:  contlcycle.MissedIntermediate_MISSED_INTERMEDIATE_UNKNOWABLE,
+			})
+		} else {
+			transitions = append(transitions, &contlcycle.PodStateTransition{
+				Field:             contlcycle.PodStatusField_POD_STATUS_FIELD_CONDITION,
+				LastObservedState: lastObserved,
+				NewState: &contlcycle.PodStatusValue{Value: &contlcycle.PodStatusValue_Condition{
+					Condition: conditionToModel(condition),
+				}},
+				TransitionTimestamp: condition.LastTransitionTime.Unix(),
+				Precision:           contlcycle.Precision_PRECISION_EXACT,
+				MissedIntermediate:  missedIntermediate,
+			})
+		}
+
+		shadow.conditions[condition.Type] = condition
+	}
+
+	if len(transitions) == 0 {
+		return nil, nil
+	}
+
+	les := make([]LifecycleEvent, 0, len(transitions))
+	for _, transition := range transitions {
+		les = append(les, LifecycleEvent{
+			ObjectKind: types.ObjectKindPod,
+			ProtoEvent: &contlcycle.Event{
+				EventType: contlcycle.Event_Transition,
+				TypedEvent: &contlcycle.Event_Pod{Pod: &contlcycle.PodEvent{
+					PodUID:     podUID,
+					Source:     string(workloadmeta.SourceNodeOrchestrator),
+					Transition: transition,
+				}},
+			},
+		})
+	}
+
+	return les, nil
+}
+
+// phaseChanged reports whether current is a real, newly-observed phase value.
+func phaseChanged(prior, current string) bool {
+	return current != "" && current != prior
+}
+
+// conditionUnchanged reports whether a condition has already been observed and is identical to its prior value, barring reason changes.
+func conditionUnchanged(seen bool, prior, current workloadmeta.KubernetesPodCondition) bool {
+	return seen && prior.Status == current.Status && prior.LastTransitionTime.Equal(current.LastTransitionTime)
+}
+
+// conditionMissedIntermediate reports PROVEN when status is unchanged but LastTransitionTime advanced,
+// since that only happens on a real status flip.
+func conditionMissedIntermediate(seen bool, prior, current workloadmeta.KubernetesPodCondition) contlcycle.MissedIntermediate {
+	if seen && prior.Status == current.Status &&
+		!prior.LastTransitionTime.IsZero() && !current.LastTransitionTime.IsZero() &&
+		current.LastTransitionTime.After(prior.LastTransitionTime) {
+		return contlcycle.MissedIntermediate_MISSED_INTERMEDIATE_PROVEN
+	}
+	return contlcycle.MissedIntermediate_MISSED_INTERMEDIATE_UNKNOWABLE
+}
+
+// conditionToModel converts a workloadmeta pod condition into its contlcycle proto representation.
+func conditionToModel(c workloadmeta.KubernetesPodCondition) *contlcycle.ConditionValue {
+	return &contlcycle.ConditionValue{
+		Type:   c.Type,
+		Status: c.Status,
+		Reason: &c.Reason,
+	}
+}