diff --git a/src/compute-plane-services/nvca/internal/miniservice/BUILD.bazel b/src/compute-plane-services/nvca/internal/miniservice/BUILD.bazel index 7a0aa34860..c307de6f78 100644 --- a/src/compute-plane-services/nvca/internal/miniservice/BUILD.bazel +++ b/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", diff --git a/src/compute-plane-services/nvca/internal/miniservice/events.go b/src/compute-plane-services/nvca/internal/miniservice/events.go new file mode 100644 index 0000000000..8c720f41b1 --- /dev/null +++ b/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 +} diff --git a/src/compute-plane-services/nvca/internal/miniservice/events_test.go b/src/compute-plane-services/nvca/internal/miniservice/events_test.go new file mode 100644 index 0000000000..d86207c418 --- /dev/null +++ b/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) + }) +} diff --git a/src/compute-plane-services/nvca/internal/miniservice/prereqs_test.go b/src/compute-plane-services/nvca/internal/miniservice/prereqs_test.go index a12054bc97..f34ba48e60 100644 --- a/src/compute-plane-services/nvca/internal/miniservice/prereqs_test.go +++ b/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{ diff --git a/src/compute-plane-services/nvca/internal/miniservice/reconcile.go b/src/compute-plane-services/nvca/internal/miniservice/reconcile.go index 8da0b3de5b..5792d1a5a5 100644 --- a/src/compute-plane-services/nvca/internal/miniservice/reconcile.go +++ b/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) } } diff --git a/src/compute-plane-services/nvca/internal/miniservice/reconcile_test.go b/src/compute-plane-services/nvca/internal/miniservice/reconcile_test.go index 4b9a99ad97..a58274f0b4 100644 --- a/src/compute-plane-services/nvca/internal/miniservice/reconcile_test.go +++ b/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) diff --git a/src/compute-plane-services/nvca/internal/miniservice/reconcile_update_test.go b/src/compute-plane-services/nvca/internal/miniservice/reconcile_update_test.go index a543db9234..20d0440f03 100644 --- a/src/compute-plane-services/nvca/internal/miniservice/reconcile_update_test.go +++ b/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,