feat(nvca): emit events for MiniService abnormal status conditions

src/compute-plane-services/nvca/internal/miniservice/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "modelcache_storage_selection.go",
         "controller.go",
         "decode.go",
+        "events.go",
         "gvkcache.go",
         "metadata_configmap.go",
         "mutate.go",
@@ -125,6 +126,7 @@ go_test(
     name = "miniservice_test",
     srcs = [
         "controller_test.go",
+        "events_test.go",
         "gvkcache_test.go",
         "metadata_configmap_test.go",
         "mutate_test.go",

src/compute-plane-services/nvca/internal/miniservice/events.go

@@ -0,0 +1,80 @@
+/*
+SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+SPDX-License-Identifier: Apache-2.0
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package mscontroller
+
+import (
+	"strings"
+	"unicode/utf8"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1"
+)
+
+// maxEventMessageLen bounds recorded event messages so a verbose or multi-line
+// condition message can't bloat the Event object stored in etcd. Matches the
+// +kubebuilder:validation:MaxLength=1024 already enforced on metav1.Condition.Message.
+const maxEventMessageLen = 1024
+
+// recordEvent emits a Kubernetes event on ms. It is a no-op if the reconciler has
+// no event recorder configured, so callers don't need to guard against a nil recorder.
+func (r *Reconciler) recordEvent(ms *v1alpha1.MiniService, eventType, reason, messageFmt string, args ...any) {
+	if r.eventRecorder == nil {
+		return
+	}
+	r.eventRecorder.Eventf(ms, eventType, reason, messageFmt, args...)
+}
+
+// emitConditionEvents records a Warning event for each MiniService status condition that
+// transitioned to False during this reconcile, and a Normal recovery event for each
+// condition that transitioned back to True. Conditions whose Status and Reason are
+// unchanged are skipped, so an unresolved abnormal state is not re-announced on every
+// reconcile.
+func (r *Reconciler) emitConditionEvents(ms *v1alpha1.MiniService, oldConditions, newConditions []metav1.Condition) {
+	for _, cond := range newConditions {
+		oldCond := meta.FindStatusCondition(oldConditions, cond.Type)
+		if oldCond != nil && oldCond.Status == cond.Status && oldCond.Reason == cond.Reason {
+			continue
+		}
+
+		switch cond.Status {
+		case metav1.ConditionFalse:
+			r.recordEvent(ms, corev1.EventTypeWarning, cond.Reason, "%s", sanitizeEventMessage(cond.Message))
+		case metav1.ConditionTrue:
+			if oldCond != nil && oldCond.Status == metav1.ConditionFalse {
+				r.recordEvent(ms, corev1.EventTypeNormal, cond.Reason, "%s condition recovered", cond.Type)
+			}
+		}
+	}
+}
+
+// sanitizeEventMessage collapses a condition message to a single line and truncates it,
+// so it stays a concise, etcd-friendly event message rather than a raw multi-line error dump.
+func sanitizeEventMessage(msg string) string {
+	msg = strings.Join(strings.Fields(msg), " ")
+	if len(msg) > maxEventMessageLen {
+		cut := maxEventMessageLen - 3
+		for cut > 0 && !utf8.RuneStart(msg[cut]) {
+			cut--
+		}
+		msg = msg[:cut] + "..."
+	}
+	return msg
+}

src/compute-plane-services/nvca/internal/miniservice/events_test.go

@@ -0,0 +1,188 @@
+/*
+SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+SPDX-License-Identifier: Apache-2.0
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package mscontroller
+
+import (
+	"strings"
+	"testing"
+	"unicode/utf8"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/record"
+
+	"github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1"
+)
+
+// drainEvents reads any events currently buffered on rec.Events without blocking.
+func drainEvents(t *testing.T, rec *record.FakeRecorder) []string {
+	t.Helper()
+	var events []string
+	for {
+		select {
+		case e := <-rec.Events:
+			events = append(events, e)
+		default:
+			return events
+		}
+	}
+}
+
+func newEventTestMiniService() *v1alpha1.MiniService {
+	return &v1alpha1.MiniService{
+		ObjectMeta: metav1.ObjectMeta{Name: "test-ms", Namespace: "ns"},
+	}
+}
+
+func TestEmitConditionEvents(t *testing.T) {
+	t.Run("unset to false emits a warning event", func(t *testing.T) {
+		rec := record.NewFakeRecorder(10)
+		r := &Reconciler{eventRecorder: rec}
+		ms := newEventTestMiniService()
+
+		newConds := []metav1.Condition{{
+			Type:    v1alpha1.MiniServiceConditionObjectsHealthy,
+			Status:  metav1.ConditionFalse,
+			Reason:  v1alpha1.MiniServiceStatusReasonDegradedWorker,
+			Message: "worker pod worker-0 crash looping",
+		}}
+
+		r.emitConditionEvents(ms, nil, newConds)
+
+		events := drainEvents(t, rec)
+		require.Len(t, events, 1)
+		assert.Equal(t, "Warning DegradedWorker worker pod worker-0 crash looping", events[0])
+	})
+
+	t.Run("unchanged false condition does not re-emit", func(t *testing.T) {
+		rec := record.NewFakeRecorder(10)
+		r := &Reconciler{eventRecorder: rec}
+		ms := newEventTestMiniService()
+
+		cond := metav1.Condition{
+			Type:    v1alpha1.MiniServiceConditionObjectsHealthy,
+			Status:  metav1.ConditionFalse,
+			Reason:  v1alpha1.MiniServiceStatusReasonObjectsFailed,
+			Message: "objects failed to deploy",
+		}
+
+		r.emitConditionEvents(ms, []metav1.Condition{cond}, []metav1.Condition{cond})
+
+		assert.Empty(t, drainEvents(t, rec))
+	})
+
+	t.Run("false to false with a new reason emits a new warning event", func(t *testing.T) {
+		rec := record.NewFakeRecorder(10)
+		r := &Reconciler{eventRecorder: rec}
+		ms := newEventTestMiniService()
+
+		oldConds := []metav1.Condition{{
+			Type:    v1alpha1.MiniServiceConditionObjectsHealthy,
+			Status:  metav1.ConditionFalse,
+			Reason:  v1alpha1.MiniServiceStatusReasonWaitingObjectReadiness,
+			Message: "waiting on objects",
+		}}
+		newConds := []metav1.Condition{{
+			Type:    v1alpha1.MiniServiceConditionObjectsHealthy,
+			Status:  metav1.ConditionFalse,
+			Reason:  v1alpha1.MiniServiceStatusReasonPendingTimeout,
+			Message: "objects timed out pending",
+		}}
+
+		r.emitConditionEvents(ms, oldConds, newConds)
+
+		events := drainEvents(t, rec)
+		require.Len(t, events, 1)
+		assert.Equal(t, "Warning ObjectsTimedOutPending objects timed out pending", events[0])
+	})
+
+	t.Run("false to true emits a normal recovery event", func(t *testing.T) {
+		rec := record.NewFakeRecorder(10)
+		r := &Reconciler{eventRecorder: rec}
+		ms := newEventTestMiniService()
+
+		oldConds := []metav1.Condition{{
+			Type:   v1alpha1.MiniServiceConditionObjectsHealthy,
+			Status: metav1.ConditionFalse,
+			Reason: v1alpha1.MiniServiceStatusReasonDegradedWorker,
+		}}
+		newConds := []metav1.Condition{{
+			Type:   v1alpha1.MiniServiceConditionObjectsHealthy,
+			Status: metav1.ConditionTrue,
+			Reason: "ObjectsReady",
+		}}
+
+		r.emitConditionEvents(ms, oldConds, newConds)
+
+		events := drainEvents(t, rec)
+		require.Len(t, events, 1)
+		assert.Equal(t, "Normal ObjectsReady ObjectsHealthy condition recovered", events[0])
+	})
+
+	t.Run("unset to true does not emit a recovery event", func(t *testing.T) {
+		rec := record.NewFakeRecorder(10)
+		r := &Reconciler{eventRecorder: rec}
+		ms := newEventTestMiniService()
+
+		newConds := []metav1.Condition{{
+			Type:   v1alpha1.MiniServiceConditionObjectsHealthy,
+			Status: metav1.ConditionTrue,
+			Reason: "ObjectsReady",
+		}}
+
+		r.emitConditionEvents(ms, nil, newConds)
+
+		assert.Empty(t, drainEvents(t, rec))
+	})
+
+	t.Run("nil event recorder does not panic", func(t *testing.T) {
+		r := &Reconciler{}
+		ms := newEventTestMiniService()
+		newConds := []metav1.Condition{{
+			Type:   v1alpha1.MiniServiceConditionObjectsHealthy,
+			Status: metav1.ConditionFalse,
+			Reason: v1alpha1.MiniServiceStatusReasonObjectsFailed,
+		}}
+
+		assert.NotPanics(t, func() { r.emitConditionEvents(ms, nil, newConds) })
+	})
+}
+
+func TestSanitizeEventMessage(t *testing.T) {
+	t.Run("collapses newlines and extra whitespace", func(t *testing.T) {
+		msg := "line one\nline two\n\tline three"
+		assert.Equal(t, "line one line two line three", sanitizeEventMessage(msg))
+	})
+
+	t.Run("truncates overly long messages", func(t *testing.T) {
+		msg := strings.Repeat("a", maxEventMessageLen+50)
+		got := sanitizeEventMessage(msg)
+		assert.LessOrEqual(t, len(got), maxEventMessageLen)
+		assert.True(t, strings.HasSuffix(got, "..."))
+	})
+
+	t.Run("truncates without splitting a multi-byte rune", func(t *testing.T) {
+		// "e" is a 3-byte rune (U+00e9 encoded as UTF-8 wouldn't apply here, use a
+		// genuine multi-byte character so the naive byte cut lands mid-rune).
+		msg := strings.Repeat("a", maxEventMessageLen-4) + "中文" // two 3-byte CJK runes
+		got := sanitizeEventMessage(msg)
+		assert.True(t, utf8.ValidString(got))
+		assert.LessOrEqual(t, len(got), maxEventMessageLen)
+	})
+}

src/compute-plane-services/nvca/internal/miniservice/prereqs_test.go

@@ -57,7 +57,7 @@ func TestImageCredentialUpdater(t *testing.T) {
 			Metrics:            metrics.NewDefaultMetrics("test-nca-id", "test-cluster", "test-group", "test-version", metrics.WithRegisterer(prometheus.NewRegistry())),
 		},
 		Client:                crclient,
-		eventRecorder:         record.NewFakeRecorder(10),
+		eventRecorder:         record.NewFakeRecorder(256),
 		newPermissionsChecker: newFakePermissionsChecker,
 	}
 	configuredToleration := corev1.Toleration{

src/compute-plane-services/nvca/internal/miniservice/reconcile.go

@@ -289,6 +289,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 			return reconcile.Result{}, err
 		}
 	}
+	r.emitConditionEvents(ms, ms.Status.Conditions, msCopy.Status.Conditions)
+
 	if phaseChanged {
 		fromPhase := normalizeMiniServicePhase(ms.Status.Phase)
 		toPhase := normalizeMiniServicePhase(msCopy.Status.Phase)
@@ -298,12 +300,12 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco
 
 		if ms.Status.Phase == "" {
 			log.Info("MiniService changed phase", "new_status", msCopy.Status.Phase)
-			r.eventRecorder.Eventf(ms, "Normal", "PhaseChange", "phase changed to %s",
+			r.recordEvent(ms, corev1.EventTypeNormal, "PhaseChange", "phase changed to %s",
 				msCopy.Status.Phase)
 		} else {
 			log.Info("MiniService changed phase", "prev_status", ms.Status.Phase,
 				"new_status", msCopy.Status.Phase)
-			r.eventRecorder.Eventf(ms, "Normal", "PhaseChange", "phase changed from %s to %s",
+			r.recordEvent(ms, corev1.EventTypeNormal, "PhaseChange", "phase changed from %s to %s",
 				ms.Status.Phase, msCopy.Status.Phase)
 		}
 	}

src/compute-plane-services/nvca/internal/miniservice/reconcile_test.go

@@ -278,7 +278,7 @@ func TestReconcile_Function(t *testing.T) {
 		Decoder:               serializer.NewCodecFactory(testScheme).UniversalDeserializer(),
 		NFClient:              nfClient,
 		tracer:                otel.NewTracer(),
-		eventRecorder:         record.NewFakeRecorder(10),
+		eventRecorder:         record.NewFakeRecorder(256),
 		chartCache:            chartcache.New(t.TempDir()),
 		regITCache:            regITCache,
 		now:                   time.Now,
@@ -1336,7 +1336,7 @@ func TestReconcile_Function_SaveRevisionHistoryFails(t *testing.T) {
 		Decoder:               serializer.NewCodecFactory(testScheme).UniversalDeserializer(),
 		NFClient:              nfClient,
 		tracer:                otel.NewTracer(),
-		eventRecorder:         record.NewFakeRecorder(10),
+		eventRecorder:         record.NewFakeRecorder(256),
 		chartCache:            chartcache.New(t.TempDir()),
 		regITCache:            regITCache,
 		now:                   time.Now,
@@ -1847,7 +1847,7 @@ func testReconcileNVLinkOptimizedHelper(t *testing.T, helmObjs []client.Object,
 		Decoder:               serializer.NewCodecFactory(testScheme).UniversalDeserializer(),
 		NFClient:              nfClient,
 		tracer:                otel.NewTracer(),
-		eventRecorder:         record.NewFakeRecorder(10),
+		eventRecorder:         record.NewFakeRecorder(256),
 		chartCache:            chartcache.New(t.TempDir()),
 		regITCache:            regITCache,
 		now:                   time.Now,
@@ -2222,7 +2222,7 @@ func TestReconcile_Task(t *testing.T) {
 		Decoder:               serializer.NewCodecFactory(testScheme).UniversalDeserializer(),
 		NFClient:              nfClient,
 		tracer:                otel.NewTracer(),
-		eventRecorder:         record.NewFakeRecorder(10),
+		eventRecorder:         record.NewFakeRecorder(256),
 		chartCache:            chartcache.New(t.TempDir()),
 		regITCache:            regITCache,
 		now:                   time.Now,
@@ -2993,6 +2993,7 @@ func TestReconcile_TaskStatus(t *testing.T) {
 	regITCache := icms.NewRegistrationInstanceTypeCache()
 	regITCache.Put(nvcatypes.BackendGPUs(nfClient.BackendGPUs).ToRegistration(false, corev1.ResourceList{}))
 
+	fakeRecorder := record.NewFakeRecorder(256)
 	r := &Reconciler{
 		ControllerOptions: ControllerOptions{
 			SystemNamespace:      "nvca-system",
@@ -3019,7 +3020,7 @@ func TestReconcile_TaskStatus(t *testing.T) {
 		Decoder:               serializer.NewCodecFactory(testScheme).UniversalDeserializer(),
 		NFClient:              nfClient,
 		tracer:                otel.NewTracer(),
-		eventRecorder:         record.NewFakeRecorder(10),
+		eventRecorder:         fakeRecorder,
 		chartCache:            chartcache.New(t.TempDir()),
 		regITCache:            regITCache,
 		now:                   time.Now,
@@ -3588,6 +3589,11 @@ rules:
 			},
 		}, ms.Status.Conditions)
 	}
+	// The ObjectsHealthy condition flipped to False without a phase change (phase stayed
+	// Running), but a Warning event should still be recorded so this is visible via
+	// `kubectl describe`.
+	assert.Contains(t, drainEvents(t, fakeRecorder),
+		"Warning ObjectsFailedWithinBackoffTimeout batch/v1.Job foo: BackoffLimitExceeded v1.Pod job-pod: SomeContainersNotReadyReason")
 
 	// Sleep to allow backoff timeout to elapse (configured to 1ms in test)
 	time.Sleep(5 * time.Millisecond)

src/compute-plane-services/nvca/internal/miniservice/reconcile_update_test.go

@@ -206,7 +206,7 @@ func newUpdateTestReconciler(t *testing.T, c client.Client, scheme *runtime.Sche
 		},
 		Client:                c,
 		Decoder:               serializer.NewCodecFactory(scheme).UniversalDeserializer(),
-		eventRecorder:         record.NewFakeRecorder(20),
+		eventRecorder:         record.NewFakeRecorder(256),
 		tracer:                otel.NewTracer(),
 		chartCache:            chartcache.New(t.TempDir()),
 		newPermissionsChecker: newFakePermissionsChecker,