diff --git a/api/v1beta1/kafkacluster_types.go b/api/v1beta1/kafkacluster_types.go index 1042ec9df..697f361e8 100644 --- a/api/v1beta1/kafkacluster_types.go +++ b/api/v1beta1/kafkacluster_types.go @@ -148,6 +148,15 @@ type KafkaClusterSpec struct { // +optional KRaftMode bool `json:"kRaft"` HeadlessServiceEnabled bool `json:"headlessServiceEnabled"` + // externalResourceManagementEnabled allows an external controller (for example a mutating + // admission webhook such as ScaleOps) to manage CPU/memory requests and scheduling + // affinities on Kafka broker pods. When enabled, Koperator will not revert requests or + // preferred affinity terms that it did not itself declare, so that a genuine change to + // this KafkaCluster's own resource/affinity configuration still takes effect while + // externally-applied overrides are preserved across reconciles. + // +kubebuilder:default=false + // +optional + ExternalResourceManagementEnabled bool `json:"externalResourceManagementEnabled,omitempty"` // localDebugEnabled is used to decide whether to create a separate loadbalancer services for the // Kafka and Cruise Control Pods. These services will expose the internal listener ports of the Kafka // cluster with LoadBalancer type, which can be used for running Koperator on a local machine against diff --git a/pkg/resources/kafka/kafka.go b/pkg/resources/kafka/kafka.go index d737754f8..4f5a14e2f 100644 --- a/pkg/resources/kafka/kafka.go +++ b/pkg/resources/kafka/kafka.go @@ -956,6 +956,24 @@ func (r *Reconciler) handleRollingUpgrade(log logr.Logger, desiredPod, currentPo } desiredPod.Spec.Tolerations = uniqueTolerations } + + // Capture the pure CR-derived pod spec before any external-resource sync below mutates it, + // so the baseline we record further down always reflects Koperator's own intent, never + // drift merged in from currentPod. + pureDesired := desiredPod.DeepCopy() + + if r.KafkaCluster.Spec.ExternalResourceManagementEnabled { + baseline, err := getExternalResourceBaseline(currentPod) + if err != nil { + log.Error(err, "could not read external resource baseline annotation, skipping external resource/affinity sync") + } else { + // If resource requests or affinities were changed by an external controller (e.g. + // ScaleOps), sync them into desiredPod so they aren't discarded as drift below. + syncResourceRequests(desiredPod, currentPod, baseline) + syncAffinities(desiredPod, currentPod, baseline) + } + } + // Check if the resource actually updated or if labels match TaintedBrokersSelector patchResult, err := patch.DefaultPatchMaker.Calculate(currentPod, desiredPod) switch { @@ -983,6 +1001,12 @@ func (r *Reconciler) handleRollingUpgrade(log logr.Logger, desiredPod, currentPo return errors.WrapIf(err, "could not apply last state to annotation") } + if r.KafkaCluster.Spec.ExternalResourceManagementEnabled { + if err := setExternalResourceBaseline(desiredPod, buildExternalResourceBaseline(pureDesired)); err != nil { + return errors.WrapIf(err, "could not apply external resource baseline annotation") + } + } + if !k8sutil.IsPodContainsTerminatedContainer(currentPod) { if r.KafkaCluster.Status.State != banzaiv1beta1.KafkaClusterRollingUpgrading { if err := k8sutil.UpdateCRStatus(r.Client, r.KafkaCluster, banzaiv1beta1.KafkaClusterRollingUpgrading, log); err != nil { diff --git a/pkg/resources/kafka/util.go b/pkg/resources/kafka/util.go index cfafbae14..2c3d0c238 100644 --- a/pkg/resources/kafka/util.go +++ b/pkg/resources/kafka/util.go @@ -17,14 +17,35 @@ package kafka import ( "encoding/base64" + "encoding/json" "fmt" + "reflect" "sort" "github.com/google/uuid" + corev1 "k8s.io/api/core/v1" "github.com/banzaicloud/koperator/api/v1beta1" ) +// externalResourceBaselineAnnotation records, on each broker Pod, the CPU/memory requests and +// preferred affinity terms that Koperator itself last declared for that Pod - as opposed to +// whatever an external controller (for example a mutating admission webhook such as ScaleOps) +// may have layered on top of the live Pod afterwards. It intentionally does not reuse +// k8s-objectmatcher's own "last-applied" annotation: that annotation is snapshotted from +// desiredPod *after* syncResourceRequests/syncAffinities run, so it would end up recording the +// externally-applied values instead of Koperator's own intent, making it useless for telling +// the two apart on the next reconcile. +const externalResourceBaselineAnnotation = "banzaicloud.io/external-resource-baseline" + +// externalResourceBaseline is the payload stored under externalResourceBaselineAnnotation. +type externalResourceBaseline struct { + ContainerRequests map[string]corev1.ResourceList `json:"containerRequests,omitempty"` + InitContainerRequests map[string]corev1.ResourceList `json:"initContainerRequests,omitempty"` + PodAffinityTerms []corev1.WeightedPodAffinityTerm `json:"podAffinityTerms,omitempty"` + NodeAffinityTerms []corev1.PreferredSchedulingTerm `json:"nodeAffinityTerms,omitempty"` +} + // generateQuorumVoters generates the quorum voters in the format of brokerID@nodeAddress:listenerPort // The generated quorum voters are guaranteed in ascending order by broker IDs to ensure same quorum voters configurations are returned // regardless of the order of brokers and controllerListenerStatuses are passed in - this is needed to avoid triggering @@ -73,3 +94,216 @@ func generateRandomClusterID() string { randomUUID := uuid.New() return base64.URLEncoding.EncodeToString(randomUUID[:]) } + +// buildExternalResourceBaseline captures the CPU/memory requests and preferred affinity terms +// Koperator itself declared on pod, for later comparison against a freshly-computed desired pod. +// Callers must pass a pod that has not been touched by syncResourceRequests/syncAffinities, so +// the recorded baseline reflects only Koperator's own intent. +func buildExternalResourceBaseline(pod *corev1.Pod) *externalResourceBaseline { + baseline := &externalResourceBaseline{ + ContainerRequests: containerRequestsByName(pod.Spec.Containers), + InitContainerRequests: containerRequestsByName(pod.Spec.InitContainers), + PodAffinityTerms: podAffinityTerms(pod), + NodeAffinityTerms: nodeAffinityTerms(pod), + } + return baseline +} + +func containerRequestsByName(containers []corev1.Container) map[string]corev1.ResourceList { + if len(containers) == 0 { + return nil + } + index := make(map[string]corev1.ResourceList, len(containers)) + for _, c := range containers { + index[c.Name] = c.Resources.Requests + } + return index +} + +func podAffinityTerms(pod *corev1.Pod) []corev1.WeightedPodAffinityTerm { + if pod.Spec.Affinity == nil || pod.Spec.Affinity.PodAffinity == nil { + return nil + } + return pod.Spec.Affinity.PodAffinity.PreferredDuringSchedulingIgnoredDuringExecution +} + +func nodeAffinityTerms(pod *corev1.Pod) []corev1.PreferredSchedulingTerm { + if pod.Spec.Affinity == nil || pod.Spec.Affinity.NodeAffinity == nil { + return nil + } + return pod.Spec.Affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution +} + +// getExternalResourceBaseline reads back the baseline last recorded on pod by +// setExternalResourceBaseline, or returns nil if none is present yet (for example before +// externalResourceManagementEnabled was turned on, or on a pod created before this feature +// existed). +func getExternalResourceBaseline(pod *corev1.Pod) (*externalResourceBaseline, error) { + raw, ok := pod.Annotations[externalResourceBaselineAnnotation] + if !ok { + return nil, nil + } + baseline := &externalResourceBaseline{} + if err := json.Unmarshal([]byte(raw), baseline); err != nil { + return nil, err + } + return baseline, nil +} + +// setExternalResourceBaseline stamps baseline onto pod's annotations. +func setExternalResourceBaseline(pod *corev1.Pod, baseline *externalResourceBaseline) error { + raw, err := json.Marshal(baseline) + if err != nil { + return err + } + if pod.Annotations == nil { + pod.Annotations = make(map[string]string) + } + pod.Annotations[externalResourceBaselineAnnotation] = string(raw) + return nil +} + +// syncResourceRequests overwrites CPU and memory requests in desiredPod's containers with the +// values from currentPod, but only for requests that Koperator's own baseline shows as +// unchanged - so a genuine change to the KafkaCluster's resource requests still takes effect +// instead of being silently discarded in favor of externally-applied drift. +func syncResourceRequests(desiredPod, currentPod *corev1.Pod, baseline *externalResourceBaseline) { + var baselineContainers, baselineInitContainers map[string]corev1.ResourceList + if baseline != nil { + baselineContainers = baseline.ContainerRequests + baselineInitContainers = baseline.InitContainerRequests + } + syncContainerResourceRequests(desiredPod.Spec.Containers, currentPod.Spec.Containers, baselineContainers) + syncContainerResourceRequests(desiredPod.Spec.InitContainers, currentPod.Spec.InitContainers, baselineInitContainers) +} + +func syncContainerResourceRequests(desired, current []corev1.Container, baseline map[string]corev1.ResourceList) { + currentIndex := containerRequestsByName(current) + + for i := range desired { + c := &desired[i] + currentReqs, ok := currentIndex[c.Name] + if !ok { + continue + } + baselineReqs := baseline[c.Name] + + if c.Resources.Requests == nil { + c.Resources.Requests = make(corev1.ResourceList) + } + for _, res := range []corev1.ResourceName{corev1.ResourceCPU, corev1.ResourceMemory} { + desiredVal, desiredHas := c.Resources.Requests[res] + baselineVal, baselineHas := baselineReqs[res] + if desiredHas != baselineHas || (desiredHas && !desiredVal.Equal(baselineVal)) { + // The CR declares a different value than our recorded baseline, meaning it + // changed since we last recorded it. Let that new value take effect instead + // of preserving whatever is live on currentPod. + continue + } + if currentVal, ok := currentReqs[res]; ok { + c.Resources.Requests[res] = currentVal + } else { + delete(c.Resources.Requests, res) + } + } + } +} + +// syncAffinities preserves preferred pod/node affinity terms that an external controller added +// to currentPod on top of whatever Koperator itself declared, while still letting the +// KafkaCluster CR remove terms it previously declared (tracked via baseline) without them being +// resurrected from currentPod on the next reconcile. +func syncAffinities(desiredPod, currentPod *corev1.Pod, baseline *externalResourceBaseline) { + syncPodAffinities(desiredPod, currentPod, baseline) + syncNodeAffinities(desiredPod, currentPod, baseline) +} + +func syncPodAffinities(desiredPod, currentPod *corev1.Pod, baseline *externalResourceBaseline) { + currentTerms := podAffinityTerms(currentPod) + if len(currentTerms) == 0 { + return + } + var baselineTerms []corev1.WeightedPodAffinityTerm + if baseline != nil { + baselineTerms = baseline.PodAffinityTerms + } + + var externallyAddedTerms []corev1.WeightedPodAffinityTerm + for _, term := range currentTerms { + if !containsWeightedPodAffinityTerm(baselineTerms, term) { + externallyAddedTerms = append(externallyAddedTerms, term) + } + } + if len(externallyAddedTerms) == 0 { + return + } + + if desiredPod.Spec.Affinity == nil { + desiredPod.Spec.Affinity = &corev1.Affinity{} + } + if desiredPod.Spec.Affinity.PodAffinity == nil { + desiredPod.Spec.Affinity.PodAffinity = &corev1.PodAffinity{} + } + + existingTerms := desiredPod.Spec.Affinity.PodAffinity.PreferredDuringSchedulingIgnoredDuringExecution + for _, term := range externallyAddedTerms { + if !containsWeightedPodAffinityTerm(existingTerms, term) { + existingTerms = append(existingTerms, term) + } + } + desiredPod.Spec.Affinity.PodAffinity.PreferredDuringSchedulingIgnoredDuringExecution = existingTerms +} + +func containsWeightedPodAffinityTerm(terms []corev1.WeightedPodAffinityTerm, target corev1.WeightedPodAffinityTerm) bool { + for _, t := range terms { + if t.Weight == target.Weight && reflect.DeepEqual(t.PodAffinityTerm, target.PodAffinityTerm) { + return true + } + } + return false +} + +func syncNodeAffinities(desiredPod, currentPod *corev1.Pod, baseline *externalResourceBaseline) { + currentTerms := nodeAffinityTerms(currentPod) + if len(currentTerms) == 0 { + return + } + var baselineTerms []corev1.PreferredSchedulingTerm + if baseline != nil { + baselineTerms = baseline.NodeAffinityTerms + } + + var externallyAddedTerms []corev1.PreferredSchedulingTerm + for _, term := range currentTerms { + if !containsPreferredSchedulingTerm(baselineTerms, term) { + externallyAddedTerms = append(externallyAddedTerms, term) + } + } + if len(externallyAddedTerms) == 0 { + return + } + + if desiredPod.Spec.Affinity == nil { + desiredPod.Spec.Affinity = &corev1.Affinity{} + } + if desiredPod.Spec.Affinity.NodeAffinity == nil { + desiredPod.Spec.Affinity.NodeAffinity = &corev1.NodeAffinity{} + } + + existingTerms := desiredPod.Spec.Affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution + for _, term := range externallyAddedTerms { + if !containsPreferredSchedulingTerm(existingTerms, term) { + existingTerms = append(existingTerms, term) + } + } + desiredPod.Spec.Affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution = existingTerms +} + +func containsPreferredSchedulingTerm(terms []corev1.PreferredSchedulingTerm, target corev1.PreferredSchedulingTerm) bool { + for _, t := range terms { + if t.Weight == target.Weight && reflect.DeepEqual(t.Preference, target.Preference) { + return true + } + } + return false +} diff --git a/pkg/resources/kafka/util_test.go b/pkg/resources/kafka/util_test.go index d4c04045e..9fadd7ba3 100644 --- a/pkg/resources/kafka/util_test.go +++ b/pkg/resources/kafka/util_test.go @@ -20,6 +20,10 @@ import ( "reflect" "testing" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "github.com/banzaicloud/koperator/api/v1beta1" ) @@ -402,3 +406,237 @@ func TestGenerateQuorumVoters(t *testing.T) { }) } } + +func resourceListWithCPU(cpu string) corev1.ResourceList { + return corev1.ResourceList{corev1.ResourceCPU: resource.MustParse(cpu)} +} + +func TestSyncResourceRequests(t *testing.T) { + cpu100m := resource.MustParse("100m") + cpu200m := resource.MustParse("200m") + cpu300m := resource.MustParse("300m") + + tests := []struct { + name string + desired []corev1.Container + current []corev1.Container + baseline map[string]corev1.ResourceList + want corev1.ResourceList + }{ + { + name: "no baseline recorded: CR value wins over live drift", + desired: []corev1.Container{{Name: "kafka", Resources: corev1.ResourceRequirements{Requests: resourceListWithCPU("100m")}}}, + current: []corev1.Container{{Name: "kafka", Resources: corev1.ResourceRequirements{Requests: resourceListWithCPU("300m")}}}, + baseline: nil, + want: corev1.ResourceList{corev1.ResourceCPU: cpu100m}, + }, + { + name: "current matches baseline: CR hasn't changed, external drift preserved", + desired: []corev1.Container{{Name: "kafka", Resources: corev1.ResourceRequirements{Requests: resourceListWithCPU("100m")}}}, + current: []corev1.Container{{Name: "kafka", Resources: corev1.ResourceRequirements{Requests: resourceListWithCPU("300m")}}}, + baseline: map[string]corev1.ResourceList{"kafka": resourceListWithCPU("100m")}, + want: corev1.ResourceList{corev1.ResourceCPU: cpu300m}, + }, + { + name: "CR value differs from baseline: CR's new value wins over drift", + desired: []corev1.Container{{Name: "kafka", Resources: corev1.ResourceRequirements{Requests: resourceListWithCPU("200m")}}}, + current: []corev1.Container{{Name: "kafka", Resources: corev1.ResourceRequirements{Requests: resourceListWithCPU("300m")}}}, + baseline: map[string]corev1.ResourceList{"kafka": resourceListWithCPU("100m")}, + want: corev1.ResourceList{corev1.ResourceCPU: cpu200m}, + }, + { + name: "container absent from current: left unchanged", + desired: []corev1.Container{{Name: "kafka", Resources: corev1.ResourceRequirements{Requests: resourceListWithCPU("100m")}}}, + current: []corev1.Container{{Name: "other", Resources: corev1.ResourceRequirements{Requests: resourceListWithCPU("300m")}}}, + baseline: nil, + want: corev1.ResourceList{corev1.ResourceCPU: cpu100m}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + desiredPod := &corev1.Pod{Spec: corev1.PodSpec{Containers: tt.desired}} + currentPod := &corev1.Pod{Spec: corev1.PodSpec{Containers: tt.current}} + var baseline *externalResourceBaseline + if tt.baseline != nil { + baseline = &externalResourceBaseline{ContainerRequests: tt.baseline} + } + + syncResourceRequests(desiredPod, currentPod, baseline) + + got := desiredPod.Spec.Containers[0].Resources.Requests + gotCPU := got[corev1.ResourceCPU] + wantCPU := tt.want[corev1.ResourceCPU] + if !gotCPU.Equal(wantCPU) { + t.Errorf("expected cpu %s, got %s", wantCPU.String(), gotCPU.String()) + } + }) + } + + t.Run("init containers synced independently from regular containers", func(t *testing.T) { + desiredPod := &corev1.Pod{Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "kafka", Resources: corev1.ResourceRequirements{Requests: resourceListWithCPU("100m")}}}, + InitContainers: []corev1.Container{{Name: "init-certs", Resources: corev1.ResourceRequirements{Requests: resourceListWithCPU("100m")}}}, + }} + currentPod := &corev1.Pod{Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "kafka", Resources: corev1.ResourceRequirements{Requests: resourceListWithCPU("300m")}}}, + InitContainers: []corev1.Container{{Name: "init-certs", Resources: corev1.ResourceRequirements{Requests: resourceListWithCPU("300m")}}}, + }} + baseline := &externalResourceBaseline{ + ContainerRequests: map[string]corev1.ResourceList{"kafka": resourceListWithCPU("100m")}, + InitContainerRequests: map[string]corev1.ResourceList{"init-certs": resourceListWithCPU("999m")}, + } + + syncResourceRequests(desiredPod, currentPod, baseline) + + gotContainerCPU := desiredPod.Spec.Containers[0].Resources.Requests[corev1.ResourceCPU] + if !gotContainerCPU.Equal(cpu300m) { + t.Errorf("expected container cpu 300m (matches baseline, drift preserved), got %s", gotContainerCPU.String()) + } + gotInitCPU := desiredPod.Spec.InitContainers[0].Resources.Requests[corev1.ResourceCPU] + if !gotInitCPU.Equal(cpu100m) { + t.Errorf("expected init container cpu 100m (baseline mismatch, CR wins), got %s", gotInitCPU.String()) + } + }) +} + +func newPodWithPreferredPodAffinity(terms ...corev1.WeightedPodAffinityTerm) *corev1.Pod { + return &corev1.Pod{Spec: corev1.PodSpec{Affinity: &corev1.Affinity{ + PodAffinity: &corev1.PodAffinity{PreferredDuringSchedulingIgnoredDuringExecution: terms}, + }}} +} + +func weightedPodAffinityTerm(weight int32, key string) corev1.WeightedPodAffinityTerm { + return corev1.WeightedPodAffinityTerm{ + Weight: weight, + PodAffinityTerm: corev1.PodAffinityTerm{ + LabelSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"key": key}}, + TopologyKey: "kubernetes.io/hostname", + }, + } +} + +func TestSyncPodAffinities(t *testing.T) { + crTerm := weightedPodAffinityTerm(10, "cr-declared") + externalTerm := weightedPodAffinityTerm(20, "externally-added") + + t.Run("term added externally (absent from baseline) is preserved into desired", func(t *testing.T) { + desiredPod := newPodWithPreferredPodAffinity(crTerm) + currentPod := newPodWithPreferredPodAffinity(crTerm, externalTerm) + baseline := &externalResourceBaseline{PodAffinityTerms: []corev1.WeightedPodAffinityTerm{crTerm}} + + syncPodAffinities(desiredPod, currentPod, baseline) + + got := desiredPod.Spec.Affinity.PodAffinity.PreferredDuringSchedulingIgnoredDuringExecution + if len(got) != 2 { + t.Fatalf("expected 2 terms after sync, got %d", len(got)) + } + }) + + t.Run("term the CR removed (present in baseline and current) is not resurrected", func(t *testing.T) { + desiredPod := newPodWithPreferredPodAffinity() // CR no longer declares crTerm + currentPod := newPodWithPreferredPodAffinity(crTerm) + baseline := &externalResourceBaseline{PodAffinityTerms: []corev1.WeightedPodAffinityTerm{crTerm}} + + syncPodAffinities(desiredPod, currentPod, baseline) + + if desiredPod.Spec.Affinity != nil && desiredPod.Spec.Affinity.PodAffinity != nil && + len(desiredPod.Spec.Affinity.PodAffinity.PreferredDuringSchedulingIgnoredDuringExecution) != 0 { + t.Errorf("expected crTerm to stay removed, got %+v", desiredPod.Spec.Affinity.PodAffinity.PreferredDuringSchedulingIgnoredDuringExecution) + } + }) + + t.Run("no current affinity: no-op", func(t *testing.T) { + desiredPod := &corev1.Pod{} + currentPod := &corev1.Pod{} + + syncPodAffinities(desiredPod, currentPod, nil) + + if desiredPod.Spec.Affinity != nil { + t.Errorf("expected no affinity to be set, got %+v", desiredPod.Spec.Affinity) + } + }) +} + +func newPodWithPreferredNodeAffinity(terms ...corev1.PreferredSchedulingTerm) *corev1.Pod { + return &corev1.Pod{Spec: corev1.PodSpec{Affinity: &corev1.Affinity{ + NodeAffinity: &corev1.NodeAffinity{PreferredDuringSchedulingIgnoredDuringExecution: terms}, + }}} +} + +func preferredSchedulingTerm(weight int32, key string) corev1.PreferredSchedulingTerm { + return corev1.PreferredSchedulingTerm{ + Weight: weight, + Preference: corev1.NodeSelectorTerm{ + MatchExpressions: []corev1.NodeSelectorRequirement{ + {Key: key, Operator: corev1.NodeSelectorOpExists}, + }, + }, + } +} + +func TestSyncNodeAffinities(t *testing.T) { + crTerm := preferredSchedulingTerm(10, "cr-declared") + externalTerm := preferredSchedulingTerm(20, "scaleops.sh/node-packing") + + t.Run("term added externally (absent from baseline) is preserved into desired", func(t *testing.T) { + desiredPod := newPodWithPreferredNodeAffinity(crTerm) + currentPod := newPodWithPreferredNodeAffinity(crTerm, externalTerm) + baseline := &externalResourceBaseline{NodeAffinityTerms: []corev1.PreferredSchedulingTerm{crTerm}} + + syncNodeAffinities(desiredPod, currentPod, baseline) + + got := desiredPod.Spec.Affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution + if len(got) != 2 { + t.Fatalf("expected 2 terms after sync, got %d", len(got)) + } + }) + + t.Run("term the CR removed (present in baseline and current) is not resurrected", func(t *testing.T) { + desiredPod := newPodWithPreferredNodeAffinity() // CR no longer declares crTerm + currentPod := newPodWithPreferredNodeAffinity(crTerm) + baseline := &externalResourceBaseline{NodeAffinityTerms: []corev1.PreferredSchedulingTerm{crTerm}} + + syncNodeAffinities(desiredPod, currentPod, baseline) + + if desiredPod.Spec.Affinity != nil && desiredPod.Spec.Affinity.NodeAffinity != nil && + len(desiredPod.Spec.Affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution) != 0 { + t.Errorf("expected crTerm to stay removed, got %+v", desiredPod.Spec.Affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution) + } + }) +} + +func TestExternalResourceBaselineRoundTrip(t *testing.T) { + original := &externalResourceBaseline{ + ContainerRequests: map[string]corev1.ResourceList{"kafka": resourceListWithCPU("100m")}, + } + pod := &corev1.Pod{} + + if err := setExternalResourceBaseline(pod, original); err != nil { + t.Fatalf("setExternalResourceBaseline failed: %v", err) + } + + got, err := getExternalResourceBaseline(pod) + if err != nil { + t.Fatalf("getExternalResourceBaseline failed: %v", err) + } + if got == nil { + t.Fatal("expected non-nil baseline") + } + gotCPU := got.ContainerRequests["kafka"][corev1.ResourceCPU] + wantCPU := original.ContainerRequests["kafka"][corev1.ResourceCPU] + if !gotCPU.Equal(wantCPU) { + t.Errorf("expected cpu %s, got %s", wantCPU.String(), gotCPU.String()) + } +} + +func TestGetExternalResourceBaselineMissingAnnotation(t *testing.T) { + pod := &corev1.Pod{} + got, err := getExternalResourceBaseline(pod) + if err != nil { + t.Fatalf("expected no error for missing annotation, got %v", err) + } + if got != nil { + t.Errorf("expected nil baseline when annotation is absent, got %+v", got) + } +}