From 12fb141b8fe80b920b6395e1da1c781eae5482c3 Mon Sep 17 00:00:00 2001 From: Yashraj Shukla Date: Fri, 18 Sep 2026 18:02:39 +0000 Subject: [PATCH] fix(controller): populate Harness status Signed-off-by: Yashraj Shukla --- go/core/internal/controller/collections.go | 3 + go/core/internal/controller/reconciler.go | 48 ++++++++++- .../internal/controller/reconciler_test.go | 80 +++++++++++++++++ go/core/internal/controller/status.go | 41 +++++++++ go/core/internal/controller/status_test.go | 86 +++++++++++++++++++ 5 files changed, 257 insertions(+), 1 deletion(-) diff --git a/go/core/internal/controller/collections.go b/go/core/internal/controller/collections.go index bbab415892..9e3e49b50f 100644 --- a/go/core/internal/controller/collections.go +++ b/go/core/internal/controller/collections.go @@ -30,6 +30,7 @@ type Collections struct { ModelConfigStatuses krt.StatusCollection[*kagentv1alpha3.ModelConfig, kagentv1alpha3.ModelConfigStatus] ResolvedModelConfigs krt.Collection[v2translator.ResolvedModelConfig] AgentTemplateStatuses krt.StatusCollection[*kagentv1alpha3.AgentTemplate, kagentv1alpha3.AgentTemplateStatus] + HarnessStatuses krt.StatusCollection[*kagentv1alpha3.Harness, kagentv1alpha3.HarnessStatus] } // PairRuntimeObservation records a pair's preparation. The revision prevents a @@ -76,6 +77,7 @@ func NewCollections(client kube.Client, watchNamespaces []string, opts krt.Optio } reconciliations := newPairReconciliations(pairs, compilerCollections, pairRuntimeObservations, opts) statuses := newAgentTemplateStatuses(agentTemplates, reconciliations, opts) + harnessStatuses := newHarnessStatuses(harnesses, workerPools, opts) return Collections{ AgentTemplates: agentTemplates, @@ -91,6 +93,7 @@ func NewCollections(client kube.Client, watchNamespaces []string, opts krt.Optio ModelConfigStatuses: modelConfigStatuses, ResolvedModelConfigs: resolvedModelConfigs, AgentTemplateStatuses: statuses, + HarnessStatuses: harnessStatuses, } } diff --git a/go/core/internal/controller/reconciler.go b/go/core/internal/controller/reconciler.go index 09f6334d97..3b61773331 100644 --- a/go/core/internal/controller/reconciler.go +++ b/go/core/internal/controller/reconciler.go @@ -138,9 +138,11 @@ type Reconciler struct { pairs controllers.Queue agentTemplateStatuses controllers.Queue modelConfigStatuses controllers.Queue + harnessStatuses controllers.Queue pairHandler krt.HandlerRegistration agentTemplateStatusHandler krt.HandlerRegistration modelConfigStatusHandler krt.HandlerRegistration + harnessStatusHandler krt.HandlerRegistration } // NewReconciler creates the Kubernetes and database write boundary. Run starts @@ -174,6 +176,9 @@ func newReconciler( r.modelConfigStatuses = newReconciliationQueue("v2-model-config-status", func(item any) error { return r.reconcileModelConfigStatus(context.Background(), item.(string)) }) + r.harnessStatuses = newReconciliationQueue("v2-harness-status", func(item any) error { + return r.reconcileHarnessStatus(context.Background(), item.(string)) + }) r.pairHandler = collections.Reconciliations.Register(func(event krt.Event[PairReconciliation]) { r.pairs.Add(krt.GetKey(event.Latest())) @@ -192,6 +197,13 @@ func newReconciler( } r.modelConfigStatuses.Add(status.ResourceName()) }) + r.harnessStatusHandler = collections.HarnessStatuses.Register(func(event krt.Event[krt.ObjectWithStatus[*kagentv1alpha3.Harness, kagentv1alpha3.HarnessStatus]]) { + status := event.Latest() + if apiequality.Semantic.DeepEqual(harnessStatusWithTransitionTimes(status.Status, status.Obj.Status), status.Obj.Status) { + return + } + r.harnessStatuses.Add(status.ResourceName()) + }) return r } @@ -207,15 +219,17 @@ func newReconciliationQueue(name string, reconcile func(any) error) controllers. // Run waits for the graph boundary to observe initial state, then processes // pair and status writes until stop closes. func (r *Reconciler) Run(stop <-chan struct{}) { - if !r.pairHandler.WaitUntilSynced(stop) || !r.agentTemplateStatusHandler.WaitUntilSynced(stop) || !r.modelConfigStatusHandler.WaitUntilSynced(stop) { + if !r.pairHandler.WaitUntilSynced(stop) || !r.agentTemplateStatusHandler.WaitUntilSynced(stop) || !r.modelConfigStatusHandler.WaitUntilSynced(stop) || !r.harnessStatusHandler.WaitUntilSynced(stop) { r.pairs.ShutDownEarly() r.agentTemplateStatuses.ShutDownEarly() r.modelConfigStatuses.ShutDownEarly() + r.harnessStatuses.ShutDownEarly() return } go r.pollPendingTemplates(stop) go r.agentTemplateStatuses.Run(stop) go r.modelConfigStatuses.Run(stop) + go r.harnessStatuses.Run(stop) r.pairs.Run(stop) } @@ -370,6 +384,38 @@ func (r *Reconciler) reconcileModelConfigStatus(ctx context.Context, key string) return nil } +func (r *Reconciler) reconcileHarnessStatus(ctx context.Context, key string) error { + desired := r.collections.HarnessStatuses.GetKey(key) + harness := r.collections.Harnesses.GetKey(key) + if desired == nil || harness == nil { + return nil + } + updated := (*harness).DeepCopy() + updated.Status = harnessStatusWithTransitionTimes(desired.Status, updated.Status) + if apiequality.Semantic.DeepEqual(updated.Status, (*harness).Status) { + return nil + } + if _, err := r.status.Harnesses(updated.Namespace).UpdateStatus(ctx, updated, metav1.UpdateOptions{}); err != nil { + return fmt.Errorf("update Harness %s status: %w", key, err) + } + return nil +} + +func harnessStatusWithTransitionTimes(desired, current kagentv1alpha3.HarnessStatus) kagentv1alpha3.HarnessStatus { + desired.Conditions = append([]metav1.Condition(nil), desired.Conditions...) + for conditionIndex := range desired.Conditions { + condition := &desired.Conditions[conditionIndex] + if previous := apimeta.FindStatusCondition(current.Conditions, condition.Type); previous != nil && + previous.Status == condition.Status && previous.Reason == condition.Reason && + previous.Message == condition.Message && previous.ObservedGeneration == condition.ObservedGeneration { + condition.LastTransitionTime = previous.LastTransitionTime + continue + } + condition.LastTransitionTime = metav1.Now() + } + return desired +} + func statusWithTransitionTimes(desired, current kagentv1alpha3.AgentTemplateStatus) kagentv1alpha3.AgentTemplateStatus { desired.Harnesses = append([]kagentv1alpha3.AgentTemplateHarnessStatus(nil), desired.Harnesses...) for harnessIndex := range desired.Harnesses { diff --git a/go/core/internal/controller/reconciler_test.go b/go/core/internal/controller/reconciler_test.go index b052904571..b002d68420 100644 --- a/go/core/internal/controller/reconciler_test.go +++ b/go/core/internal/controller/reconciler_test.go @@ -9,6 +9,7 @@ import ( "time" a2apb "github.com/a2aproject/a2a-go/v2/a2apb/v1" + atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" "google.golang.org/protobuf/encoding/protowire" "google.golang.org/protobuf/proto" @@ -26,6 +27,7 @@ import ( "istio.io/istio/pkg/kube/krt" "istio.io/istio/pkg/kube/krt/krttest" corev1 "k8s.io/api/core/v1" + apimeta "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -314,6 +316,78 @@ func (s *fakeRuntimeRevisionStore) RetirePairIdentities(_ context.Context, names return nil } +func TestReconcilerUpdatesHarnessStatus(t *testing.T) { + stop := make(chan struct{}) + t.Cleanup(func() { close(stop) }) + opts := krt.NewOptionsBuilder(stop, "test-harness-status-reconciler", nil) + + harness := &kagentv1alpha3.Harness{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "team-a", + Name: "kagent", + Generation: 1, + }, + Spec: kagentv1alpha3.HarnessSpec{ + Substrate: kagentv1alpha3.HarnessSubstratePolicy{ + WorkerPoolRef: corev1.LocalObjectReference{Name: "default"}, + }, + }, + } + workerPool := &atev1alpha1.WorkerPool{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "team-a", + Name: "default", + }, + } + + mock := krttest.NewMock(t, []any{harness, workerPool}) + harnesses := krttest.GetMockCollection[*kagentv1alpha3.Harness](mock) + workerPools := krttest.GetMockCollection[*atev1alpha1.WorkerPool](mock) + harnessStatuses := newHarnessStatuses(harnesses, workerPools, opts) + + statusClient := kagentfake.NewSimpleClientset(harness.DeepCopy()).ApiV1alpha3() + + collections := Collections{ + Harnesses: harnesses, + WorkerPools: workerPools, + HarnessStatuses: harnessStatuses, + AgentTemplates: krttest.GetMockCollection[*kagentv1alpha3.AgentTemplate](mock), + Reconciliations: krttest.GetMockCollection[PairReconciliation](mock), + AgentTemplateStatuses: krttest.GetMockCollection[krt.ObjectWithStatus[*kagentv1alpha3.AgentTemplate, kagentv1alpha3.AgentTemplateStatus]](mock), + ModelConfigStatuses: krttest.GetMockCollection[krt.ObjectWithStatus[*kagentv1alpha3.ModelConfig, kagentv1alpha3.ModelConfigStatus]](mock), + } + + reconciler := newReconciler( + collections, + &fakeActorTemplates{}, + &fakeRuntimeRevisionStore{}, + statusClient, + ) + + go reconciler.Run(stop) + + var updated *kagentv1alpha3.Harness + var err error + require.Eventually(t, func() bool { + updated, err = statusClient.Harnesses(harness.Namespace).Get( + context.Background(), + harness.Name, + metav1.GetOptions{}, + ) + if err != nil { + return false + } + + condition := apimeta.FindStatusCondition( + updated.Status.Conditions, + kagentv1alpha3.HarnessConditionTypeReady, + ) + return condition != nil && + condition.Status == metav1.ConditionTrue && + !condition.LastTransitionTime.IsZero() + }, 3*time.Second, 10*time.Millisecond) +} + func TestReconcilerUpdatesModelConfigStatusOnSecretHashChange(t *testing.T) { stop := make(chan struct{}) t.Cleanup(func() { close(stop) }) @@ -337,6 +411,9 @@ func TestReconcilerUpdatesModelConfigStatusOnSecretHashChange(t *testing.T) { secrets := krt.NewStaticCollection(nil, []*corev1.Secret{secret}, opts.WithName("Secrets")...) configMaps := krttest.GetMockCollection[*corev1.ConfigMap](mock) modelConfigStatuses, resolvedModelConfigs := newModelConfigReconciliations(modelConfigs, configMaps, secrets, opts) + harnesses := krttest.GetMockCollection[*kagentv1alpha3.Harness](mock) + workerPools := krttest.GetMockCollection[*atev1alpha1.WorkerPool](mock) + harnessStatuses := newHarnessStatuses(harnesses, workerPools, opts) collections := Collections{ ModelConfigs: modelConfigs, @@ -347,6 +424,9 @@ func TestReconcilerUpdatesModelConfigStatusOnSecretHashChange(t *testing.T) { AgentTemplates: krttest.GetMockCollection[*kagentv1alpha3.AgentTemplate](mock), Reconciliations: krttest.GetMockCollection[PairReconciliation](mock), AgentTemplateStatuses: krttest.GetMockCollection[krt.ObjectWithStatus[*kagentv1alpha3.AgentTemplate, kagentv1alpha3.AgentTemplateStatus]](mock), + Harnesses: harnesses, + WorkerPools: workerPools, + HarnessStatuses: harnessStatuses, } statusClient := kagentfake.NewSimpleClientset(modelConfig.DeepCopy()).ApiV1alpha3() diff --git a/go/core/internal/controller/status.go b/go/core/internal/controller/status.go index 22698a6faf..c9a8562a7f 100644 --- a/go/core/internal/controller/status.go +++ b/go/core/internal/controller/status.go @@ -4,9 +4,11 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "fmt" "slices" "strings" + atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" kagentv1alpha3 "github.com/kagent-dev/kagent/go/api/v1alpha3" "istio.io/istio/pkg/kube/krt" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -35,6 +37,45 @@ func newAgentTemplateStatuses(templates krt.Collection[*kagentv1alpha3.AgentTemp return statuses } +func newHarnessStatuses( + harnesses krt.Collection[*kagentv1alpha3.Harness], + workerPools krt.Collection[*atev1alpha1.WorkerPool], + opts krt.OptionsBuilder, +) krt.StatusCollection[*kagentv1alpha3.Harness, kagentv1alpha3.HarnessStatus] { + statuses, _ := krt.NewStatusManyCollection(harnesses, func(ctx krt.HandlerContext, harness *kagentv1alpha3.Harness) (*kagentv1alpha3.HarnessStatus, []*atev1alpha1.WorkerPool) { + workerPoolKey := types.NamespacedName{ + Namespace: harness.Namespace, + Name: harness.Spec.Substrate.WorkerPoolRef.Name, + } + workerPool := krt.FetchOne(ctx, workerPools, krt.FilterObjectName(workerPoolKey)) + + status := &kagentv1alpha3.HarnessStatus{ + ObservedGeneration: harness.Generation, + } + + if workerPool == nil || *workerPool == nil { + status.Conditions = []metav1.Condition{{ + Type: kagentv1alpha3.HarnessConditionTypeReady, + Status: metav1.ConditionFalse, + Reason: "WorkerPoolNotFound", + Message: fmt.Sprintf("WorkerPool %q not found", workerPoolKey.String()), + ObservedGeneration: harness.Generation, + }} + return status, nil + } + + status.Conditions = []metav1.Condition{{ + Type: kagentv1alpha3.HarnessConditionTypeReady, + Status: metav1.ConditionTrue, + Reason: "WorkerPoolResolved", + Message: fmt.Sprintf("WorkerPool %q exists", workerPoolKey.String()), + ObservedGeneration: harness.Generation, + }} + return status, nil + }, opts.WithName("HarnessStatuses")...) + return statuses +} + func statusForPair(state PairReconciliation, generation int64, latestSuccessful string) kagentv1alpha3.AgentTemplateHarnessStatus { desired := state.RevisionID.String() if state.RevisionID.IsZero() { diff --git a/go/core/internal/controller/status_test.go b/go/core/internal/controller/status_test.go index a160da4a66..17707ca1ed 100644 --- a/go/core/internal/controller/status_test.go +++ b/go/core/internal/controller/status_test.go @@ -3,8 +3,14 @@ package controller import ( "testing" + atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" kagentv1alpha3 "github.com/kagent-dev/kagent/go/api/v1alpha3" v2translator "github.com/kagent-dev/kagent/go/core/internal/translator" + "istio.io/istio/pkg/kube/krt" + "istio.io/istio/pkg/kube/krt/krttest" + corev1 "k8s.io/api/core/v1" + apimeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) func TestStatusForPairPublishesCompilationWarnings(t *testing.T) { @@ -24,3 +30,83 @@ func TestStatusForPairPublishesCompilationWarnings(t *testing.T) { t.Fatal("status aliases mutable compilation warnings") } } + +func TestHarnessStatusReadyWhenWorkerPoolExists(t *testing.T) { + stop := make(chan struct{}) + t.Cleanup(func() { close(stop) }) + opts := krt.NewOptionsBuilder(stop, "test-harness-status", nil) + + harness := &kagentv1alpha3.Harness{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "team-a", + Name: "kagent", + }, + Spec: kagentv1alpha3.HarnessSpec{ + Substrate: kagentv1alpha3.HarnessSubstratePolicy{ + WorkerPoolRef: corev1.LocalObjectReference{Name: "default"}, + }, + }, + } + workerPool := &atev1alpha1.WorkerPool{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "team-a", + Name: "default", + }, + } + + mock := krttest.NewMock(t, []any{harness, workerPool}) + harnesses := krttest.GetMockCollection[*kagentv1alpha3.Harness](mock) + workerPools := krttest.GetMockCollection[*atev1alpha1.WorkerPool](mock) + + statuses := newHarnessStatuses(harnesses, workerPools, opts) + + waitFor(t, func() bool { + status := statuses.GetKey("team-a/kagent") + if status == nil { + return false + } + condition := apimeta.FindStatusCondition( + status.Status.Conditions, + kagentv1alpha3.HarnessConditionTypeReady, + ) + return condition != nil && condition.Status == metav1.ConditionTrue + }) +} + +func TestHarnessStatusNotReadyWhenWorkerPoolMissing(t *testing.T) { + stop := make(chan struct{}) + t.Cleanup(func() { close(stop) }) + opts := krt.NewOptionsBuilder(stop, "test-harness-status-missing", nil) + + harness := &kagentv1alpha3.Harness{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "team-a", + Name: "kagent", + }, + Spec: kagentv1alpha3.HarnessSpec{ + Substrate: kagentv1alpha3.HarnessSubstratePolicy{ + WorkerPoolRef: corev1.LocalObjectReference{Name: "missing"}, + }, + }, + } + + mock := krttest.NewMock(t, []any{harness}) + harnesses := krttest.GetMockCollection[*kagentv1alpha3.Harness](mock) + workerPools := krttest.GetMockCollection[*atev1alpha1.WorkerPool](mock) + + statuses := newHarnessStatuses(harnesses, workerPools, opts) + + waitFor(t, func() bool { + status := statuses.GetKey("team-a/kagent") + if status == nil { + return false + } + condition := apimeta.FindStatusCondition( + status.Status.Conditions, + kagentv1alpha3.HarnessConditionTypeReady, + ) + return condition != nil && + condition.Status == metav1.ConditionFalse && + condition.Reason == "WorkerPoolNotFound" + }) +}