diff --git a/pkg/registry/apps/application/rest.go b/pkg/registry/apps/application/rest.go index 583a4a8c..0728ea13 100644 --- a/pkg/registry/apps/application/rest.go +++ b/pkg/registry/apps/application/rest.go @@ -704,13 +704,19 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio } customW.underlying = helmWatcher - // Start watch on WorkloadMonitor to detect pod readiness changes + // Start watch on WorkloadMonitor to detect pod readiness changes. + // For Tenant applications the WorkloadMonitor lives in a computed child + // namespace (see computeTenantNamespace), not in the HelmRelease namespace, + // so scoping the watch to `namespace` would miss all events. Use a + // cluster-wide watch in that case — label selectors still restrict the + // stream to the relevant kind/group. wmLabelSelector := labels.NewSelector().Add(*appKindReq, *appGroupReq) wmList := &cozyv1alpha1.WorkloadMonitorList{} - wmWatcher, err := r.w.Watch(ctx, wmList, &client.ListOptions{ - Namespace: namespace, - LabelSelector: wmLabelSelector, - }) + wmListOpts := &client.ListOptions{LabelSelector: wmLabelSelector} + if r.kindName != "Tenant" { + wmListOpts.Namespace = namespace + } + wmWatcher, err := r.w.Watch(ctx, wmList, wmListOpts) if err != nil { klog.Warningf("Failed to set up WorkloadMonitor watch, workload status changes won't trigger events: %v", err) // Non-fatal: proceed without WorkloadMonitor watch @@ -731,6 +737,16 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio initialEventsEndSent := !sendInitialEvents // If not sendInitialEvents, consider it already sent var lastResourceVersion string + // Buffer of WorkloadMonitor events that arrived before the initial + // snapshot finished. The watch-list contract requires the stream to + // deliver all ADDED events followed by the initial-events-end bookmark + // before any live updates, so we hold WM-triggered Modified events and + // replay them once the bookmark has been emitted. Without this, a + // workload whose status flips during the snapshot window (after the + // Application ADDED but before the bookmark) and then stops changing + // would leave the client with a stale WorkloadsReady forever. + var pendingWMEvents []watch.Event + // Get the starting resourceVersion from options // If client provides resourceVersion (e.g., from a previous List), we should skip // objects with resourceVersion <= startingRV (client already has them) @@ -741,6 +757,19 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio } } + drainPendingWMEvents := func() { + for _, ev := range pendingWMEvents { + select { + case customW.resultChan <- ev: + case <-customW.stopChan: + return + case <-ctx.Done(): + return + } + } + pendingWMEvents = nil + } + // Helper function to send initial-events-end bookmark sendInitialEventsEndBookmark := func() { if initialEventsEndSent { @@ -765,8 +794,11 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio select { case customW.resultChan <- bookmarkEvent: case <-customW.stopChan: + return case <-ctx.Done(): + return } + drainPendingWMEvents() } // Process watch events @@ -794,8 +826,10 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio APIVersion: appsv1alpha1.SchemeGroupVersion.String(), Kind: r.kindName, } + justFlipped := false if !initialEventsEndSent { initialEventsEndSent = true + justFlipped = true bookmarkApp.SetAnnotations(map[string]string{ "k8s.io/initial-events-end": "true", }) @@ -812,6 +846,9 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio case <-ctx.Done(): return } + if justFlipped { + drainPendingWMEvents() + } } continue } @@ -911,12 +948,6 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio } continue } - // Don't emit WM-triggered events until the initial snapshot is - // complete — the watch-list contract requires all ADDED events - // followed by the initial-events-end bookmark before any live updates. - if !initialEventsEndSent { - continue - } wm, ok := wmEvent.Object.(*cozyv1alpha1.WorkloadMonitor) if !ok { continue @@ -937,12 +968,37 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio if resourceName != "" && wmAppName != resourceName { continue } - hr := &helmv2.HelmRelease{} - if err := r.c.Get(ctx, client.ObjectKey{Namespace: wm.Namespace, Name: hrName}, hr); err != nil { - klog.V(4).Infof("Cannot find HelmRelease %s/%s for WorkloadMonitor event: %v", wm.Namespace, hrName, err) + + // Locate the owning HelmRelease. For most application kinds the + // WorkloadMonitor and its HelmRelease live in the same namespace, + // but Tenant workloads live in a child namespace (see + // computeTenantNamespace) while the HelmRelease remains in the + // parent/requested namespace — so the WM-to-HR namespace mapping + // differs. + hrNS := wm.Namespace + if r.kindName == "Tenant" { + hrNS = namespace + // Filter out WM events whose child namespace does not + // correspond to the Tenant in our watched namespace, since + // the cluster-wide WM watch delivers events for all tenants. + if r.computeTenantNamespace(namespace, wmAppName) != wm.Namespace { + continue + } + if !fieldFilter.MatchesNamespace(hrNS) { + continue + } + } else if !fieldFilter.MatchesNamespace(hrNS) { continue } - app, err := r.ConvertHelmReleaseToApplication(ctx, hr) + hr := &helmv2.HelmRelease{} + if err := r.c.Get(ctx, client.ObjectKey{Namespace: hrNS, Name: hrName}, hr); err != nil { + klog.V(4).Infof("Cannot find HelmRelease %s/%s for WorkloadMonitor event: %v", hrNS, hrName, err) + continue + } + // Pass the fresh WorkloadMonitor so conversion uses the latest + // operational status even if the cache (r.c) has not yet + // observed this watch event. + app, err := r.ConvertHelmReleaseToApplicationWithMonitor(ctx, hr, wm) if err != nil { klog.V(4).Infof("Error converting HelmRelease for WorkloadMonitor event: %v", err) continue @@ -961,9 +1017,17 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio // Use the WorkloadMonitor's ResourceVersion for the emitted event // so clients see a monotonically increasing RV and don't skip this update. app.SetResourceVersion(wm.GetResourceVersion()) + outEvent := watch.Event{Type: watch.Modified, Object: &app} + // Buffer WM-triggered events that arrive before the + // initial-events-end bookmark. They will be replayed in order + // immediately after the bookmark is emitted. + if !initialEventsEndSent { + pendingWMEvents = append(pendingWMEvents, outEvent) + continue + } lastResourceVersion = wm.GetResourceVersion() select { - case customW.resultChan <- watch.Event{Type: watch.Modified, Object: &app}: + case customW.resultChan <- outEvent: case <-customW.stopChan: return case <-ctx.Done(): @@ -1094,12 +1158,21 @@ func filterPrefixedMap(original map[string]string, prefix string) map[string]str return processed } -// ConvertHelmReleaseToApplication converts a HelmRelease to an Application +// ConvertHelmReleaseToApplication converts a HelmRelease to an Application. func (r *REST) ConvertHelmReleaseToApplication(ctx context.Context, hr *helmv2.HelmRelease) (appsv1alpha1.Application, error) { + return r.ConvertHelmReleaseToApplicationWithMonitor(ctx, hr, nil) +} + +// ConvertHelmReleaseToApplicationWithMonitor converts a HelmRelease to an +// Application, optionally overriding the cached copy of a WorkloadMonitor with +// a fresher version received from the watch client. This prevents the emitted +// Application object from carrying stale WorkloadsReady data when r.c (cache) +// lags behind r.w (watch). +func (r *REST) ConvertHelmReleaseToApplicationWithMonitor(ctx context.Context, hr *helmv2.HelmRelease, freshMonitor *cozyv1alpha1.WorkloadMonitor) (appsv1alpha1.Application, error) { klog.V(6).Infof("Converting HelmRelease to Application for resource %s", hr.GetName()) // Convert HelmRelease struct to Application struct - app, err := r.convertHelmReleaseToApplication(ctx, hr) + app, err := r.convertHelmReleaseToApplication(ctx, hr, freshMonitor) if err != nil { klog.Errorf("Error converting from HelmRelease to Application: %v", err) return appsv1alpha1.Application{}, err @@ -1212,8 +1285,11 @@ func (r *REST) validateTenantNamespaceLength(currentNamespace, tenantName string return allErrs } -// convertHelmReleaseToApplication implements the actual conversion logic -func (r *REST) convertHelmReleaseToApplication(ctx context.Context, hr *helmv2.HelmRelease) (appsv1alpha1.Application, error) { +// convertHelmReleaseToApplication implements the actual conversion logic. +// The optional freshMonitor is used to override the cache copy of a +// WorkloadMonitor when a newer version was delivered via the watch client — +// see ConvertHelmReleaseToApplicationWithMonitor for the rationale. +func (r *REST) convertHelmReleaseToApplication(ctx context.Context, hr *helmv2.HelmRelease, freshMonitor *cozyv1alpha1.WorkloadMonitor) (appsv1alpha1.Application, error) { // Filter out internal keys (starting with "_") from spec filteredSpec := filterInternalKeys(hr.Spec.Values) @@ -1250,28 +1326,49 @@ func (r *REST) convertHelmReleaseToApplication(ctx context.Context, hr *helmv2.H }) } } - // Enrich conditions with WorkloadMonitor operational status - ws, wsErr := r.getWorkloadsOperational(ctx, hr.Namespace, app.Name) + // Enrich conditions with WorkloadMonitor operational status. + // Tenant workloads live in a child namespace (computed from the Tenant name), + // not in the same namespace as the owning HelmRelease — look there instead. + workloadsNS := hr.Namespace + if r.kindName == "Tenant" { + workloadsNS = r.computeTenantNamespace(hr.Namespace, app.Name) + } + ws, wsErr := r.getWorkloadsOperational(ctx, workloadsNS, app.Name, freshMonitor) + // Derive a stable LastTransitionTime: use the owning HelmRelease's own + // condition update time (or CreationTimestamp as a floor) so that repeated + // conversions of the same underlying state produce identical timestamps. + wrTransition := hr.CreationTimestamp + for _, c := range hr.GetConditions() { + if c.LastTransitionTime.After(wrTransition.Time) { + wrTransition = c.LastTransitionTime + } + } + if ws.transitionTime.After(wrTransition.Time) { + wrTransition = ws.transitionTime + } + if wrTransition.IsZero() { + // Fallback for objects that somehow have no timestamps at all + // (e.g. hand-crafted test fixtures). In production HelmReleases + // always carry a CreationTimestamp, so the stable branch above + // is used. + wrTransition = metav1.Now() + } if wsErr != nil { // Fail-open: if we can't query WorkloadMonitors (e.g., informer cache not ready), // don't override Ready. Prefer operational availability over safety. // The WorkloadsReady=Unknown condition still signals the issue to the user. - klog.Warningf("Failed to check workload monitors for %s/%s: %v", hr.Namespace, app.Name, wsErr) + klog.Warningf("Failed to check workload monitors for %s/%s: %v", workloadsNS, app.Name, wsErr) conditions = append(conditions, metav1.Condition{ Type: "WorkloadsReady", Status: metav1.ConditionUnknown, - LastTransitionTime: metav1.Now(), + LastTransitionTime: wrTransition, Reason: "Error", Message: fmt.Sprintf("Failed to check workload status: %v", wsErr), }) } else if ws.found { - // LastTransitionTime is set to the current time because the Application - // resource is virtual (computed on-the-fly from HelmRelease). There is no - // persistent condition state to track actual transitions. This is consistent - // with how computed/virtual API resources work in Kubernetes. workloadsCondition := metav1.Condition{ Type: "WorkloadsReady", - LastTransitionTime: metav1.Now(), + LastTransitionTime: wrTransition, Reason: "WorkloadMonitorCheck", } switch { @@ -1321,11 +1418,20 @@ type workloadsStatus struct { operational bool found bool unknown bool // true when at least one monitor has nil Operational (not yet reconciled) + // transitionTime is the most recent metadata update time across the + // matching monitors. Used as WorkloadsReady.LastTransitionTime so that + // repeated conversions for the same underlying state produce stable + // timestamps (preserving the Kubernetes contract that identical + // resource versions represent identical content). + transitionTime metav1.Time } // getWorkloadsOperational checks WorkloadMonitor resources for an application and returns // aggregated operational status. If no monitors exist, returns found=false. -func (r *REST) getWorkloadsOperational(ctx context.Context, namespace, appName string) (workloadsStatus, error) { +// When freshOverride is non-nil, its status replaces the cached copy for the +// corresponding monitor — this keeps the result consistent with watch events +// when the cache (r.c) lags behind the watch client (r.w). +func (r *REST) getWorkloadsOperational(ctx context.Context, namespace, appName string, freshOverride *cozyv1alpha1.WorkloadMonitor) (workloadsStatus, error) { monitors := &cozyv1alpha1.WorkloadMonitorList{} if err := r.c.List(ctx, monitors, client.InNamespace(namespace), @@ -1337,19 +1443,47 @@ func (r *REST) getWorkloadsOperational(ctx context.Context, namespace, appName s ); err != nil { return workloadsStatus{}, err } + // Ensure the freshOverride is represented in the aggregation even when + // the cache has not yet observed it (brand-new resource) or is behind. + replaced := false + if freshOverride != nil { + for i := range monitors.Items { + if monitors.Items[i].UID == freshOverride.UID || + (monitors.Items[i].Name == freshOverride.Name && monitors.Items[i].Namespace == freshOverride.Namespace) { + monitors.Items[i] = *freshOverride + replaced = true + break + } + } + if !replaced { + monitors.Items = append(monitors.Items, *freshOverride) + } + } if len(monitors.Items) == 0 { return workloadsStatus{operational: true, found: false}, nil } operational := true unknown := false + var latest metav1.Time for _, m := range monitors.Items { if m.Status.Operational == nil { unknown = true } else if !*m.Status.Operational { operational = false } + // Pick the most recent monitor mtime as a stable transition time. + if t := latestMonitorTime(&m); t.After(latest.Time) { + latest = t + } } - return workloadsStatus{operational: operational, found: true, unknown: unknown}, nil + return workloadsStatus{operational: operational, found: true, unknown: unknown, transitionTime: latest}, nil +} + +// latestMonitorTime returns the most recent timestamp associated with a +// WorkloadMonitor — currently only the object creation time is guaranteed. +// Status does not carry a transition time, so we fall back to CreationTimestamp. +func latestMonitorTime(m *cozyv1alpha1.WorkloadMonitor) metav1.Time { + return m.CreationTimestamp } // convertApplicationToHelmRelease implements the actual conversion logic diff --git a/pkg/registry/apps/application/rest_conditions_test.go b/pkg/registry/apps/application/rest_conditions_test.go index dbb99d52..185f59b2 100644 --- a/pkg/registry/apps/application/rest_conditions_test.go +++ b/pkg/registry/apps/application/rest_conditions_test.go @@ -61,7 +61,7 @@ func TestConvertConditions_WorkloadsReadyAdded(t *testing.T) { {Type: "Released", Status: metav1.ConditionTrue, Reason: "Succeeded", Message: "Released"}, } - app, err := r.convertHelmReleaseToApplication(context.TODO(), hr) + app, err := r.convertHelmReleaseToApplication(context.TODO(), hr, nil) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -107,7 +107,7 @@ func TestConvertConditions_ReadyNotOverriddenWhenWorkloadsNotReady(t *testing.T) {Type: "Ready", Status: metav1.ConditionTrue, Reason: "Succeeded", Message: "Release applied"}, } - app, err := r.convertHelmReleaseToApplication(context.TODO(), hr) + app, err := r.convertHelmReleaseToApplication(context.TODO(), hr, nil) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -139,7 +139,7 @@ func TestConvertConditions_NoOverrideWhenNoMonitors(t *testing.T) { {Type: "Ready", Status: metav1.ConditionTrue, Reason: "Succeeded", Message: "Release applied"}, } - app, err := r.convertHelmReleaseToApplication(context.TODO(), hr) + app, err := r.convertHelmReleaseToApplication(context.TODO(), hr, nil) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -181,7 +181,7 @@ func TestConvertConditions_ReadyStaysTrue_WhenAllOperational(t *testing.T) { {Type: "Released", Status: metav1.ConditionTrue, Reason: "Succeeded", Message: "Released"}, } - app, err := r.convertHelmReleaseToApplication(context.TODO(), hr) + app, err := r.convertHelmReleaseToApplication(context.TODO(), hr, nil) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -219,7 +219,7 @@ func TestConvertConditions_WorkloadsReadyTimestampIsNonZero(t *testing.T) { // No Ready condition — HR still being reconciled hr.Status.Conditions = []metav1.Condition{} - app, err := r.convertHelmReleaseToApplication(context.TODO(), hr) + app, err := r.convertHelmReleaseToApplication(context.TODO(), hr, nil) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -255,7 +255,7 @@ func TestConvertConditions_WorkloadsReadyUnknownWhenNilOperational(t *testing.T) {Type: "Ready", Status: metav1.ConditionTrue, Reason: "Succeeded", Message: "ok"}, } - app, err := r.convertHelmReleaseToApplication(context.TODO(), hr) + app, err := r.convertHelmReleaseToApplication(context.TODO(), hr, nil) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -302,7 +302,7 @@ func TestConvertConditions_WorkloadsReadyUnknownOnError(t *testing.T) { {Type: "Ready", Status: metav1.ConditionTrue, Reason: "Succeeded", Message: "ok"}, } - app, err := r.convertHelmReleaseToApplication(context.TODO(), hr) + app, err := r.convertHelmReleaseToApplication(context.TODO(), hr, nil) if err != nil { t.Fatalf("unexpected error: %v", err) } diff --git a/pkg/registry/apps/application/rest_validation_test.go b/pkg/registry/apps/application/rest_validation_test.go index 088990ce..b806161c 100644 --- a/pkg/registry/apps/application/rest_validation_test.go +++ b/pkg/registry/apps/application/rest_validation_test.go @@ -190,7 +190,7 @@ func TestConvertHelmReleaseToApplication_TenantNamespaceKindGate(t *testing.T) { }, } - app, err := r.convertHelmReleaseToApplication(context.Background(), hr) + app, err := r.convertHelmReleaseToApplication(context.Background(), hr, nil) if err != nil { t.Fatalf("convertHelmReleaseToApplication: %v", err) } @@ -208,7 +208,7 @@ func TestConvertHelmReleaseToApplication_TenantNamespaceKindGate(t *testing.T) { }, } - app, err := r.convertHelmReleaseToApplication(context.Background(), hr) + app, err := r.convertHelmReleaseToApplication(context.Background(), hr, nil) if err != nil { t.Fatalf("convertHelmReleaseToApplication: %v", err) } diff --git a/pkg/registry/apps/application/rest_watch_test.go b/pkg/registry/apps/application/rest_watch_test.go index 8fe2a04a..93a51f3c 100644 --- a/pkg/registry/apps/application/rest_watch_test.go +++ b/pkg/registry/apps/application/rest_watch_test.go @@ -144,7 +144,7 @@ func TestWatchIntegration_MonitorDeletionDropsWorkloadsReady(t *testing.T) { r := newTestRESTWithSchemesFromClient(c) // Step 1: With non-operational monitor, WorkloadsReady=False, Ready=True - app1, err := r.convertHelmReleaseToApplication(context.TODO(), hr) + app1, err := r.convertHelmReleaseToApplication(context.TODO(), hr, nil) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -163,7 +163,7 @@ func TestWatchIntegration_MonitorDeletionDropsWorkloadsReady(t *testing.T) { } // Step 3: WorkloadsReady should disappear, Ready stays True - app2, err := r.convertHelmReleaseToApplication(context.TODO(), hr) + app2, err := r.convertHelmReleaseToApplication(context.TODO(), hr, nil) if err != nil { t.Fatalf("unexpected error: %v", err) } diff --git a/pkg/registry/apps/application/rest_workloads_test.go b/pkg/registry/apps/application/rest_workloads_test.go index 30dd85c6..b706c323 100644 --- a/pkg/registry/apps/application/rest_workloads_test.go +++ b/pkg/registry/apps/application/rest_workloads_test.go @@ -40,7 +40,7 @@ func newTestRESTWithSchemes(objs ...runtime.Object) *REST { func TestGetWorkloadsOperational_NoMonitors(t *testing.T) { r := newTestRESTWithSchemes() - ws, err := r.getWorkloadsOperational(context.TODO(), "default", "mydb") + ws, err := r.getWorkloadsOperational(context.TODO(), "default", "mydb", nil) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -83,7 +83,7 @@ func TestGetWorkloadsOperational_AllOperational(t *testing.T) { } r := newTestRESTWithSchemes(m1, m2) - ws, err := r.getWorkloadsOperational(context.TODO(), "default", "mydb") + ws, err := r.getWorkloadsOperational(context.TODO(), "default", "mydb", nil) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -126,7 +126,7 @@ func TestGetWorkloadsOperational_SomeNotOperational(t *testing.T) { } r := newTestRESTWithSchemes(m1, m2) - ws, err := r.getWorkloadsOperational(context.TODO(), "default", "mydb") + ws, err := r.getWorkloadsOperational(context.TODO(), "default", "mydb", nil) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -155,7 +155,7 @@ func TestGetWorkloadsOperational_OperationalNil(t *testing.T) { } r := newTestRESTWithSchemes(m) - ws, err := r.getWorkloadsOperational(context.TODO(), "default", "mydb") + ws, err := r.getWorkloadsOperational(context.TODO(), "default", "mydb", nil) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -198,7 +198,7 @@ func TestGetWorkloadsOperational_MixedNilAndOperational(t *testing.T) { } r := newTestRESTWithSchemes(m1, m2) - ws, err := r.getWorkloadsOperational(context.TODO(), "default", "mydb") + ws, err := r.getWorkloadsOperational(context.TODO(), "default", "mydb", nil) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -238,7 +238,7 @@ func TestGetWorkloadsOperational_MixedFailedAndPending(t *testing.T) { } r := newTestRESTWithSchemes(m1, m2) - ws, err := r.getWorkloadsOperational(context.TODO(), "default", "mydb") + ws, err := r.getWorkloadsOperational(context.TODO(), "default", "mydb", nil) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -296,7 +296,7 @@ func TestConvertConditions_MixedFailedAndPendingShowsFalse(t *testing.T) { {Type: "Ready", Status: metav1.ConditionTrue, Reason: "Succeeded", Message: "ok"}, } - app, err := r.convertHelmReleaseToApplication(context.TODO(), hr) + app, err := r.convertHelmReleaseToApplication(context.TODO(), hr, nil) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -328,7 +328,7 @@ func TestGetWorkloadsOperational_DifferentApp_NotFound(t *testing.T) { } r := newTestRESTWithSchemes(m) - ws, err := r.getWorkloadsOperational(context.TODO(), "default", "mydb") + ws, err := r.getWorkloadsOperational(context.TODO(), "default", "mydb", nil) if err != nil { t.Fatalf("unexpected error: %v", err) }