From 8273044585fd57aeb6ff4148f6cccb66312d3fab Mon Sep 17 00:00:00 2001 From: JuanmaBM Date: Wed, 2 Sep 2026 13:11:18 +0200 Subject: [PATCH 1/2] feat(control-plane): implement GatewayRelease reconciliation Replace the no-op GatewayReleaseReconciler stub with a real reconciler that validates the release image reference, writes back a deterministic status (Available/Invalid), and fans out to referencing gateways when a release image changes. Fan-out reuses a shared gateway reconcile queue via EnqueueForced so the gateway phase gate (which skips Running/Provisioning/Degraded gateways) is bypassed on release-driven re-reconciles. The GatewayRelease watch now runs through a reconcile queue for per-release serialization and retry. An invalid image is not propagated and retains the last valid image baseline, so a later correction to a different image is still detected as a genuine change and fans out. Adds unit tests for the release reconciler and watcher-level tests proving the phase-gate bypass wiring. Adds the behavior spec. Refs HYPERSHELL-173 Co-Authored-By: Claude Opus 4.8 --- .../cmd/hypershell-controller/main.go | 11 +- .../reconciler/gateway_release_test.go | 273 ++++++++++++++++++ .../internal/reconciler/reconciler.go | 205 ++++++++++++- .../internal/reconciler/reconciler_test.go | 4 +- .../watcher/gateway_reconcile_queue_test.go | 97 +++++++ .../control-plane/internal/watcher/watcher.go | 76 +++-- specs/platform/control-plane.spec.md | 1 + .../gateway-release-reconciliation.spec.md | 175 +++++++++++ 8 files changed, 815 insertions(+), 27 deletions(-) create mode 100644 components/control-plane/internal/reconciler/gateway_release_test.go create mode 100644 components/control-plane/internal/watcher/gateway_reconcile_queue_test.go create mode 100644 specs/platform/gateway-release-reconciliation.spec.md diff --git a/components/control-plane/cmd/hypershell-controller/main.go b/components/control-plane/cmd/hypershell-controller/main.go index 2e9b68d5..44e31b97 100644 --- a/components/control-plane/cmd/hypershell-controller/main.go +++ b/components/control-plane/cmd/hypershell-controller/main.go @@ -182,7 +182,6 @@ func main() { } else { log.Printf("WARN ManagedDatabase watch disabled: both Kubernetes typed and dynamic clients are required") } - releaseReconciler := reconciler.NewGatewayReleaseReconciler() networkReconciler := reconciler.NewGatewayNetworkReconciler() manifestsDir := os.Getenv("GATEWAY_MANIFESTS_DIR") @@ -240,6 +239,14 @@ func main() { gatewayReconciler = reconciler.NewStubGatewayReconciler() } + // The gateway reconcile queue is shared: the gateway watch stream drives it, + // and the GatewayRelease reconciler enqueues referencing gateways into it when + // a release image changes. It is created here (not inside WatchGateways) so the + // release reconciler can hold the same instance. + gatewayQueue := watcher.NewGatewayReconcileQueue(ctx, gatewayReconciler, cfg.GatewayReconcileWorkers) + defer gatewayQueue.Stop() + releaseReconciler := reconciler.NewGatewayReleaseReconciler(conn, gatewayQueue) + watchCount := 4 // managed clusters, gateway releases, gateways, networks if databaseReconciler != nil { watchCount++ @@ -286,7 +293,7 @@ func main() { return watcher.WatchGatewayReleases(ctx, conn, releaseReconciler) }) supervise("Gateway watch", func(ctx context.Context) error { - return watcher.WatchGateways(ctx, conn, gatewayReconciler, cfg.ClusterID, cfg.GatewayReconcileWorkers) + return watcher.WatchGateways(ctx, conn, gatewayQueue, cfg.ClusterID) }) supervise("GatewayNetwork watch", func(ctx context.Context) error { return watcher.WatchGatewayNetworks(ctx, conn, networkReconciler) diff --git a/components/control-plane/internal/reconciler/gateway_release_test.go b/components/control-plane/internal/reconciler/gateway_release_test.go new file mode 100644 index 00000000..9e0af5f6 --- /dev/null +++ b/components/control-plane/internal/reconciler/gateway_release_test.go @@ -0,0 +1,273 @@ +package reconciler + +import ( + "context" + "fmt" + "strings" + "testing" + + pb "github.com/openshift-online/hypershell/components/api-server/pkg/api/grpc/hypershell/v1" + "github.com/openshift-online/hypershell/components/control-plane/internal/watcher" + "google.golang.org/grpc" +) + +// fakeReleaseClient records UpdateGatewayRelease calls and can inject an error. +type fakeReleaseClient struct { + pb.GatewayReleaseServiceClient + updates []*pb.UpdateGatewayReleaseRequest + updateErr error +} + +func (f *fakeReleaseClient) UpdateGatewayRelease(ctx context.Context, in *pb.UpdateGatewayReleaseRequest, opts ...grpc.CallOption) (*pb.UpdateGatewayReleaseResponse, error) { + f.updates = append(f.updates, in) + if f.updateErr != nil { + return nil, f.updateErr + } + return &pb.UpdateGatewayReleaseResponse{}, nil +} + +// fakeReleaseGatewayClient serves a fixed gateway inventory to ListGateways. +type fakeReleaseGatewayClient struct { + pb.GatewayServiceClient + gateways []*pb.Gateway + listErr error +} + +func (f *fakeReleaseGatewayClient) ListGateways(ctx context.Context, in *pb.ListGatewaysRequest, opts ...grpc.CallOption) (*pb.ListGatewaysResponse, error) { + if f.listErr != nil { + return nil, f.listErr + } + return &pb.ListGatewaysResponse{ + Items: f.gateways, + Metadata: &pb.ListMeta{Page: in.Page, Size: in.Size, Total: int32(len(f.gateways))}, + }, nil +} + +// recordingEnqueuer captures the gateways enqueued for reconciliation. +type recordingEnqueuer struct { + enqueued []string +} + +func (r *recordingEnqueuer) EnqueueForced(ev watcher.Event[*pb.Gateway]) { + r.enqueued = append(r.enqueued, ev.ResourceID) +} + +func newTestReleaseReconciler(gw pb.GatewayServiceClient, rel pb.GatewayReleaseServiceClient, q gatewayEnqueuer) *GatewayReleaseReconciler { + return &GatewayReleaseReconciler{ + active: make(map[string]struct{}), + lastImage: make(map[string]string), + gateways: gw, + releases: rel, + gwQueue: q, + } +} + +func releaseEvent(t watcher.EventType, id, image, status string) watcher.Event[*pb.GatewayRelease] { + rel := &pb.GatewayRelease{ + Metadata: &pb.ObjectReference{Id: id}, + Name: "rel-" + id, + Image: image, + } + if status != "" { + rel.Status = &status + } + return watcher.Event[*pb.GatewayRelease]{Type: t, ResourceID: id, Resource: rel} +} + +func gatewayWithRelease(id, releaseID string) *pb.Gateway { + return &pb.Gateway{ + Metadata: &pb.ObjectReference{Id: id}, + Name: "gw-" + id, + ReleaseId: releaseID, + } +} + +func TestGatewayRelease_ValidImageSetsAvailableWithoutFanOut(t *testing.T) { + rel := &fakeReleaseClient{} + gw := &fakeReleaseGatewayClient{gateways: []*pb.Gateway{gatewayWithRelease("g1", "r1")}} + q := &recordingEnqueuer{} + r := newTestReleaseReconciler(gw, rel, q) + + err := r.Handle(context.Background(), releaseEvent(watcher.EventCreated, "r1", "registry.redhat.io/openshell/gateway:v1", "")) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(rel.updates) != 1 || rel.updates[0].GetStatus() != releaseStatusAvailable { + t.Fatalf("expected status Available, got updates=%v", rel.updates) + } + // First observation is a baseline: no fan-out even though g1 references r1. + if len(q.enqueued) != 0 { + t.Fatalf("expected no fan-out on first observation, got %v", q.enqueued) + } +} + +func TestGatewayRelease_MalformedImageSetsInvalidAndSkipsFanOut(t *testing.T) { + rel := &fakeReleaseClient{} + gw := &fakeReleaseGatewayClient{gateways: []*pb.Gateway{gatewayWithRelease("g1", "r1")}} + q := &recordingEnqueuer{} + r := newTestReleaseReconciler(gw, rel, q) + + // Seed a baseline with a valid image so a subsequent invalid update would + // otherwise be a "change". + if err := r.Handle(context.Background(), releaseEvent(watcher.EventCreated, "r1", "registry.redhat.io/openshell/gateway:v1", "")); err != nil { + t.Fatalf("seed: %v", err) + } + rel.updates = nil + q.enqueued = nil + + err := r.Handle(context.Background(), releaseEvent(watcher.EventUpdated, "r1", "gateway:v1; rm -rf /", "")) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(rel.updates) != 1 || !strings.HasPrefix(rel.updates[0].GetStatus(), releaseStatusInvalid) { + t.Fatalf("expected Invalid status with reason, got %v", rel.updates) + } + if len(q.enqueued) != 0 { + t.Fatalf("invalid release must not fan out, got %v", q.enqueued) + } +} + +func TestGatewayRelease_CorrectionAfterInvalidFansOut(t *testing.T) { + rel := &fakeReleaseClient{} + gw := &fakeReleaseGatewayClient{gateways: []*pb.Gateway{gatewayWithRelease("g1", "r1")}} + q := &recordingEnqueuer{} + r := newTestReleaseReconciler(gw, rel, q) + + // Baseline established at v1. + if err := r.Handle(context.Background(), releaseEvent(watcher.EventCreated, "r1", "registry.redhat.io/openshell/gateway:v1", releaseStatusAvailable)); err != nil { + t.Fatalf("seed: %v", err) + } + // An invalid update must not fan out and must not drop the v1 baseline. + if err := r.Handle(context.Background(), releaseEvent(watcher.EventUpdated, "r1", "gateway:v1; rm -rf /", releaseStatusAvailable)); err != nil { + t.Fatalf("invalid update: %v", err) + } + q.enqueued = nil + + // Correcting to a different valid image (v3) is a genuine change from the + // retained v1 baseline and must fan out to referencing gateways. + if err := r.Handle(context.Background(), releaseEvent(watcher.EventUpdated, "r1", "registry.redhat.io/openshell/gateway:v3", releaseStatusAvailable)); err != nil { + t.Fatalf("correction: %v", err) + } + if len(q.enqueued) != 1 || q.enqueued[0] != "g1" { + t.Fatalf("expected g1 to fan out after correction, got %v", q.enqueued) + } +} + +func TestGatewayRelease_NoRedundantStatusWrite(t *testing.T) { + rel := &fakeReleaseClient{} + r := newTestReleaseReconciler(&fakeReleaseGatewayClient{}, rel, &recordingEnqueuer{}) + + // Persisted status already Available and image valid: no update expected. + err := r.Handle(context.Background(), releaseEvent(watcher.EventUpdated, "r1", "registry.redhat.io/openshell/gateway:v1", releaseStatusAvailable)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(rel.updates) != 0 { + t.Fatalf("expected no status write, got %v", rel.updates) + } +} + +func TestGatewayRelease_ImageChangeFansOutToReferencingGatewaysOnly(t *testing.T) { + rel := &fakeReleaseClient{} + gw := &fakeReleaseGatewayClient{gateways: []*pb.Gateway{ + gatewayWithRelease("g1", "r1"), + gatewayWithRelease("g2", "r1"), + gatewayWithRelease("g3", "other"), + }} + q := &recordingEnqueuer{} + r := newTestReleaseReconciler(gw, rel, q) + + // Establish baseline. + if err := r.Handle(context.Background(), releaseEvent(watcher.EventCreated, "r1", "registry.redhat.io/openshell/gateway:v1", releaseStatusAvailable)); err != nil { + t.Fatalf("seed: %v", err) + } + q.enqueued = nil + + // Image changes -> fan out to g1 and g2 only. + err := r.Handle(context.Background(), releaseEvent(watcher.EventUpdated, "r1", "registry.redhat.io/openshell/gateway:v2", releaseStatusAvailable)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(q.enqueued) != 2 { + t.Fatalf("expected 2 gateways enqueued, got %v", q.enqueued) + } + got := map[string]bool{} + for _, id := range q.enqueued { + got[id] = true + } + if !got["g1"] || !got["g2"] || got["g3"] { + t.Fatalf("fan-out targeted wrong gateways: %v", q.enqueued) + } +} + +func TestGatewayRelease_RenameDoesNotFanOut(t *testing.T) { + rel := &fakeReleaseClient{} + gw := &fakeReleaseGatewayClient{gateways: []*pb.Gateway{gatewayWithRelease("g1", "r1")}} + q := &recordingEnqueuer{} + r := newTestReleaseReconciler(gw, rel, q) + + if err := r.Handle(context.Background(), releaseEvent(watcher.EventCreated, "r1", "registry.redhat.io/openshell/gateway:v1", releaseStatusAvailable)); err != nil { + t.Fatalf("seed: %v", err) + } + q.enqueued = nil + + // Same image, unchanged: no fan-out. + err := r.Handle(context.Background(), releaseEvent(watcher.EventUpdated, "r1", "registry.redhat.io/openshell/gateway:v1", releaseStatusAvailable)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(q.enqueued) != 0 { + t.Fatalf("expected no fan-out on unchanged image, got %v", q.enqueued) + } +} + +func TestGatewayRelease_DeleteIsNoOp(t *testing.T) { + rel := &fakeReleaseClient{} + q := &recordingEnqueuer{} + r := newTestReleaseReconciler(&fakeReleaseGatewayClient{}, rel, q) + + err := r.Handle(context.Background(), releaseEvent(watcher.EventDeleted, "r1", "registry.redhat.io/openshell/gateway:v1", "")) + if err != nil { + t.Fatalf("expected delete to succeed, got %v", err) + } + if len(rel.updates) != 0 || len(q.enqueued) != 0 { + t.Fatalf("delete must not write status or fan out: updates=%v enqueued=%v", rel.updates, q.enqueued) + } +} + +func TestGatewayRelease_StatusWriteFailureIsRetried(t *testing.T) { + rel := &fakeReleaseClient{updateErr: fmt.Errorf("api server unavailable")} + r := newTestReleaseReconciler(&fakeReleaseGatewayClient{}, rel, &recordingEnqueuer{}) + + err := r.Handle(context.Background(), releaseEvent(watcher.EventCreated, "r1", "registry.redhat.io/openshell/gateway:v1", "")) + if err == nil { + t.Fatalf("expected error to be returned so the reconcile is requeued") + } +} + +func TestGatewayRelease_FanOutFailureIsRetriedAndReDetected(t *testing.T) { + rel := &fakeReleaseClient{} + gw := &fakeReleaseGatewayClient{listErr: fmt.Errorf("list unavailable")} + q := &recordingEnqueuer{} + r := newTestReleaseReconciler(gw, rel, q) + + // Baseline with a valid image. + if err := r.Handle(context.Background(), releaseEvent(watcher.EventCreated, "r1", "v1.example/img:1", releaseStatusAvailable)); err != nil { + t.Fatalf("seed: %v", err) + } + + // Image change but listing fails -> error returned, baseline NOT advanced. + if err := r.Handle(context.Background(), releaseEvent(watcher.EventUpdated, "r1", "v1.example/img:2", releaseStatusAvailable)); err == nil { + t.Fatalf("expected fan-out error to be returned") + } + + // Listing recovers; the change must still be detected on retry. + gw.listErr = nil + gw.gateways = []*pb.Gateway{gatewayWithRelease("g1", "r1")} + if err := r.Handle(context.Background(), releaseEvent(watcher.EventUpdated, "r1", "v1.example/img:2", releaseStatusAvailable)); err != nil { + t.Fatalf("retry: %v", err) + } + if len(q.enqueued) != 1 || q.enqueued[0] != "g1" { + t.Fatalf("expected g1 enqueued on retry, got %v", q.enqueued) + } +} diff --git a/components/control-plane/internal/reconciler/reconciler.go b/components/control-plane/internal/reconciler/reconciler.go index 98eebce8..eee71940 100644 --- a/components/control-plane/internal/reconciler/reconciler.go +++ b/components/control-plane/internal/reconciler/reconciler.go @@ -1159,13 +1159,70 @@ func cnpgClusterGVR() schema.GroupVersionResource { } } +// Control-plane-owned GatewayRelease status values (see +// gateway-release-reconciliation.spec.md). The reconciler settles a release's +// status to reflect the reconciled validation outcome. Available means the image +// reference is well-formed and the release may be used by gateways; Invalid is +// prefixed onto a short reason describing why validation failed. +const ( + releaseStatusAvailable = "Available" + releaseStatusInvalid = "Invalid" + // releaseFanOutPageSize is the page size used when listing gateways to find + // the ones that reference a changed release. It matches the other reconcilers' + // list page size so a typical fleet is covered in a single request. + releaseFanOutPageSize = 500 +) + +// gatewayEnqueuer requests a gateway be re-reconciled through the shared gateway +// reconcile queue. The release reconciler uses it to propagate an image change to +// referencing gateways. EnqueueForced bypasses the gateway reconciler's phase +// gate (as recovery seeds do) so a gateway already Running is re-reconciled to +// pick up the new desired image rather than being skipped. *watcher.GatewayReconcileQueue +// satisfies it. +type gatewayEnqueuer interface { + EnqueueForced(watcher.Event[*pb.Gateway]) +} + +// GatewayReleaseReconciler reconciles GatewayRelease resources. A release owns no +// Kubernetes resources, so reconciliation means: validate the release image, +// write a deterministic status back to the API server, and -- when a known +// release's effective image changes -- request reconciliation of every gateway +// that references the release so the cluster converges toward the new version. +// Resolving release_id -> image at gateway deploy time and rollout safety are +// owned by sibling specs; this reconciler only guarantees the referencing +// gateways are re-reconciled. type GatewayReleaseReconciler struct { mu sync.Mutex active map[string]struct{} + // lastImage records the last validated image observed per release ID so an + // update that does not change the effective image does not fan out, and so the + // first observation of a release (e.g. on controller start or a fresh create) + // establishes a baseline without re-provisioning gateways that are already + // running. Guarded by mu. + lastImage map[string]string + + gateways pb.GatewayServiceClient + releases pb.GatewayReleaseServiceClient + gwQueue gatewayEnqueuer } -func NewGatewayReleaseReconciler() *GatewayReleaseReconciler { - return &GatewayReleaseReconciler{active: make(map[string]struct{})} +// NewGatewayReleaseReconciler builds the release reconciler. conn is the API +// server gRPC connection used to write release status and list referencing +// gateways; gwQueue is the shared gateway reconcile queue used to propagate image +// changes. Either dependency may be nil (e.g. when the controller runs without a +// Kubernetes client), in which case propagation is skipped but validation and +// status write-back still run. +func NewGatewayReleaseReconciler(conn *grpc.ClientConn, gwQueue gatewayEnqueuer) *GatewayReleaseReconciler { + r := &GatewayReleaseReconciler{ + active: make(map[string]struct{}), + lastImage: make(map[string]string), + gwQueue: gwQueue, + } + if conn != nil { + r.gateways = pb.NewGatewayServiceClient(conn) + r.releases = pb.NewGatewayReleaseServiceClient(conn) + } + return r } func (r *GatewayReleaseReconciler) Handle(ctx context.Context, event watcher.Event[*pb.GatewayRelease]) error { @@ -1183,12 +1240,152 @@ func (r *GatewayReleaseReconciler) Handle(ctx context.Context, event watcher.Eve }() _, endSpan := cpotel.StartReconcileSpan(ctx, "GatewayRelease", event.Type.String(), event.Resource.GetMetadata().GetTraceparent()) - defer func() { endSpan(nil) }() + var reconcileErr error + defer func() { endSpan(reconcileErr) }() - log.Printf("INFO reconciling GatewayRelease %s (event=%d)", event.ResourceID, event.Type) + // A release owns no cluster resources, so a delete is a terminal, idempotent + // no-op with respect to Kubernetes: running gateways deployed from the release + // are left untouched. Forget the baseline so a later create of a new release + // (KSUIDs are never reused, but be defensive) starts clean. + if event.Type == watcher.EventDeleted { + r.forget(event.ResourceID) + log.Printf("INFO gateway release %s deleted; no cluster resources to remove", event.ResourceID) + return nil + } + + rel := event.Resource + if rel == nil { + log.Printf("WARN gateway release event %s has nil resource, skipping", event.ResourceID) + return nil + } + + // Validate the image reference using the same rules applied to gateway + // workloads (well-formed reference, no shell-injection metacharacters). An + // empty image is rejected too. + image := rel.GetImage() + validationErr := gateway.ValidateImageReference(image) + + desiredStatus := releaseStatusAvailable + if validationErr != nil { + desiredStatus = fmt.Sprintf("%s: %s", releaseStatusInvalid, validationErr) + } + + // Deterministic, idempotent status write-back: only update when the persisted + // status differs from the reconciled outcome. + if rel.GetStatus() != desiredStatus { + if err := r.updateStatus(ctx, event.ResourceID, desiredStatus); err != nil { + reconcileErr = fmt.Errorf("update gateway release %s status: %w", event.ResourceID, err) + return reconcileErr + } + } + + if validationErr != nil { + // An invalid release is not propagated to any gateway. The last valid image + // baseline is intentionally retained (not forgotten): a later correction to + // an image different from that baseline is then detected as a genuine change + // and fans out, while a correction back to the same image correctly no-ops. + // Forgetting here would reclassify the correction as a first observation and + // silently skip the fan-out. + log.Printf("INFO gateway release %s invalid image: %v", event.ResourceID, validationErr) + return nil + } + + // Fan out only when a previously-observed release's effective image changed. + // The first observation records a baseline without fanning out: a brand-new + // release has no referencing gateways yet, and on controller restart every + // release would otherwise force-reconcile every running gateway. + prev, seen := r.lastImageFor(event.ResourceID) + if seen && prev != image { + if err := r.propagateToGateways(ctx, event.ResourceID, image); err != nil { + reconcileErr = fmt.Errorf("propagate gateway release %s to referencing gateways: %w", event.ResourceID, err) + // Leave the baseline unchanged so the retry re-detects the change and + // re-attempts the fan-out. + return reconcileErr + } + } + r.rememberImage(event.ResourceID, image) return nil } +// updateStatus writes the release's reconciled status back to the API server. It +// is a no-op when the release client is not configured. +func (r *GatewayReleaseReconciler) updateStatus(ctx context.Context, id, status string) error { + if r.releases == nil { + return nil + } + _, err := r.releases.UpdateGatewayRelease(ctx, &pb.UpdateGatewayReleaseRequest{ + Id: id, + Status: &status, + }) + return err +} + +// propagateToGateways enqueues every gateway that references the release for +// reconciliation. It is a no-op when the gateway client or the shared queue is +// not configured (e.g. the controller has no Kubernetes client). +func (r *GatewayReleaseReconciler) propagateToGateways(ctx context.Context, releaseID, image string) error { + if r.gateways == nil || r.gwQueue == nil { + return nil + } + gws, err := r.listGatewaysForRelease(ctx, releaseID) + if err != nil { + return err + } + for _, gw := range gws { + r.gwQueue.EnqueueForced(watcher.Event[*pb.Gateway]{ + Type: watcher.EventUpdated, + ResourceID: gw.GetMetadata().GetId(), + Resource: gw, + }) + } + log.Printf("INFO gateway release %s image changed to %s; enqueued %d referencing gateway(s) for reconciliation", releaseID, image, len(gws)) + return nil +} + +// listGatewaysForRelease returns every gateway whose release_id references the +// given release, paginating through the API server. +func (r *GatewayReleaseReconciler) listGatewaysForRelease(ctx context.Context, releaseID string) ([]*pb.Gateway, error) { + var matching []*pb.Gateway + for page := int32(1); ; page++ { + resp, err := r.gateways.ListGateways(ctx, &pb.ListGatewaysRequest{ + Page: page, + Size: releaseFanOutPageSize, + }) + if err != nil { + return nil, err + } + items := resp.GetItems() + for _, gw := range items { + if gw.GetReleaseId() == releaseID { + matching = append(matching, gw) + } + } + total := int(resp.GetMetadata().GetTotal()) + if len(items) == 0 || len(items) < releaseFanOutPageSize || (total > 0 && page*releaseFanOutPageSize >= int32(total)) { + return matching, nil + } + } +} + +func (r *GatewayReleaseReconciler) lastImageFor(id string) (string, bool) { + r.mu.Lock() + defer r.mu.Unlock() + img, ok := r.lastImage[id] + return img, ok +} + +func (r *GatewayReleaseReconciler) rememberImage(id, image string) { + r.mu.Lock() + defer r.mu.Unlock() + r.lastImage[id] = image +} + +func (r *GatewayReleaseReconciler) forget(id string) { + r.mu.Lock() + defer r.mu.Unlock() + delete(r.lastImage, id) +} + type GatewayReconciler struct { mu sync.Mutex active map[string]struct{} diff --git a/components/control-plane/internal/reconciler/reconciler_test.go b/components/control-plane/internal/reconciler/reconciler_test.go index e9fe42bf..4240ec64 100644 --- a/components/control-plane/internal/reconciler/reconciler_test.go +++ b/components/control-plane/internal/reconciler/reconciler_test.go @@ -717,9 +717,11 @@ func TestWatchGateways_KeycloakRetryPreservesGatedPayload(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) + gatewayQueue := watcher.NewGatewayReconcileQueue(ctx, r, config.DefaultGatewayReconcileWorkers) + defer gatewayQueue.Stop() watchErr := make(chan error, 1) go func() { - watchErr <- watcher.WatchGateways(ctx, grpcConn, r, "", config.DefaultGatewayReconcileWorkers) + watchErr <- watcher.WatchGateways(ctx, grpcConn, gatewayQueue, "") }() deadline := time.NewTimer(8 * time.Second) diff --git a/components/control-plane/internal/watcher/gateway_reconcile_queue_test.go b/components/control-plane/internal/watcher/gateway_reconcile_queue_test.go new file mode 100644 index 00000000..7d191678 --- /dev/null +++ b/components/control-plane/internal/watcher/gateway_reconcile_queue_test.go @@ -0,0 +1,97 @@ +package watcher + +import ( + "context" + "sync" + "testing" + "time" + + pb "github.com/openshift-online/hypershell/components/api-server/pkg/api/grpc/hypershell/v1" +) + +// gatewayRecordingHandler records the gateway payloads it is invoked with so a +// test can assert on the phase the reconciler would observe. +type gatewayRecordingHandler struct { + mu sync.Mutex + seen []*pb.Gateway +} + +func (h *gatewayRecordingHandler) Handle(_ context.Context, ev Event[*pb.Gateway]) error { + h.mu.Lock() + h.seen = append(h.seen, ev.Resource) + h.mu.Unlock() + return nil +} + +func (h *gatewayRecordingHandler) snapshot() []*pb.Gateway { + h.mu.Lock() + defer h.mu.Unlock() + return append([]*pb.Gateway(nil), h.seen...) +} + +func waitForGatewayCalls(t *testing.T, h *gatewayRecordingHandler, want int) []*pb.Gateway { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if got := h.snapshot(); len(got) >= want { + return got + } + time.Sleep(time.Millisecond) + } + t.Fatalf("handler called %d times, want >= %d", len(h.snapshot()), want) + return nil +} + +// EnqueueForced must drive a gateway through the shared queue with its phase +// cleared, so the gateway reconciler's phase gate (which skips Running/ +// Provisioning/Degraded gateways) does not silently drop a release-driven +// re-reconcile. This proves the wiring the release fan-out relies on: without the +// phase-clear a Running gateway would never pick up a new release image. +func TestGatewayReconcileQueue_EnqueueForcedClearsPhase(t *testing.T) { + h := &gatewayRecordingHandler{} + q := NewGatewayReconcileQueue(context.Background(), h, 1) + defer q.Stop() + + running := "Running" + q.EnqueueForced(Event[*pb.Gateway]{ + Type: EventUpdated, + ResourceID: "gw-1", + Resource: &pb.Gateway{ + Metadata: &pb.ObjectReference{Id: "gw-1"}, + Phase: &running, + }, + }) + + seen := waitForGatewayCalls(t, h, 1) + if seen[0].Phase != nil { + t.Fatalf("expected phase cleared on forced enqueue, got %q", seen[0].GetPhase()) + } + if seen[0].GetMetadata().GetId() != "gw-1" { + t.Fatalf("expected gw-1, got %q", seen[0].GetMetadata().GetId()) + } +} + +// A plain (non-forced) enqueue preserves the payload verbatim, so the normal +// watch-stream path still lets the reconciler's phase gate govern create/update +// traffic. This guards the boundary between the fan-out path (forced, gate- +// bypassing) and ordinary reconciliation. +func TestGatewayReconcileQueue_EnqueuePreservesPhase(t *testing.T) { + h := &gatewayRecordingHandler{} + q := NewGatewayReconcileQueue(context.Background(), h, 1) + defer q.Stop() + + running := "Running" + q.q.enqueue(Event[*pb.Gateway]{ + Type: EventUpdated, + ResourceID: "gw-2", + Resource: &pb.Gateway{ + Metadata: &pb.ObjectReference{Id: "gw-2"}, + Phase: &running, + }, + }) + + seen := waitForGatewayCalls(t, h, 1) + if seen[0].GetPhase() != "Running" { + t.Fatalf("expected phase preserved on plain enqueue, got %q", seen[0].GetPhase()) + } +} diff --git a/components/control-plane/internal/watcher/watcher.go b/components/control-plane/internal/watcher/watcher.go index abafa02c..3f0081c4 100644 --- a/components/control-plane/internal/watcher/watcher.go +++ b/components/control-plane/internal/watcher/watcher.go @@ -342,6 +342,15 @@ func listManagedDatabasesOnce(ctx context.Context, client pb.ManagedDatabaseServ func WatchGatewayReleases(ctx context.Context, conn *grpc.ClientConn, handler Handler[*pb.GatewayRelease]) error { client := pb.NewGatewayReleaseServiceClient(conn) + // Drive release reconciliation through a per-resource reconcile queue rather + // than inline so a failed reconcile (e.g. a transient API-server error writing + // the release status, or listing the referencing gateways for fan-out) is + // retried with capped backoff instead of being logged and dropped. Releases + // own no cluster resources, so -- unlike gateways -- no startup seed or + // recovery is needed; the queue exists purely for retry and per-release + // serialization. + rq := newReconcileQueue(ctx, "GatewayRelease", handler) + defer rq.stop() return watchLoop(ctx, "GatewayRelease", func(ctx context.Context) error { stream, err := client.WatchGatewayReleases(ctx, &pb.WatchGatewayReleasesRequest{}) if err != nil { @@ -355,13 +364,11 @@ func WatchGatewayReleases(ctx context.Context, conn *grpc.ClientConn, handler Ha if err != nil { return fmt.Errorf("receiving gateway release event: %w", err) } - if err := handler.Handle(ctx, Event[*pb.GatewayRelease]{ + rq.enqueue(Event[*pb.GatewayRelease]{ Type: toEventType(event.Type), ResourceID: event.ResourceId, Resource: event.GatewayRelease, - }); err != nil { - log.Printf("ERROR handling gateway release %s: %v", event.ResourceId, err) - } + }) } }) } @@ -390,16 +397,47 @@ func gatewayWorkerCount(configured int) int { return configured } -// WatchGateways streams gateway events and drives them through a per-resource -// reconcile queue. When clusterID is non-empty the watch and its seed lists are -// scoped server-side to gateways with that cluster_id, so a managed-cluster -// spoke only ever reconciles its own gateways (the pull model); empty watches -// every gateway. workers bounds how many distinct gateways reconcile -// concurrently (see gateway-reconcile-concurrency.spec.md); a value below 1 -// falls back to the queue's built-in default so the pool always has at least -// one worker. -func WatchGateways(ctx context.Context, conn *grpc.ClientConn, handler Handler[*pb.Gateway], clusterID string, workers int) error { - workers = gatewayWorkerCount(workers) +// GatewayReconcileQueue is a shareable handle to the gateway reconcile queue. It +// lets an out-of-band reconciler -- e.g. the GatewayRelease reconciler on an +// image change -- request a gateway be re-reconciled through the same serialized, +// retrying, phase-gate-bypassing path the gateway watch stream uses, without +// blocking the caller on the (potentially multi-minute) reconcile itself. +type GatewayReconcileQueue struct { + q *reconcileQueue[*pb.Gateway] +} + +// NewGatewayReconcileQueue builds and starts the shared gateway reconcile queue. +// The caller owns its lifecycle and must call Stop (or cancel ctx) to drain it. +// Pass the returned queue to both WatchGateways and any out-of-band enqueuer. +// workers bounds how many distinct gateways reconcile concurrently (see +// gateway-reconcile-concurrency.spec.md); a value below 1 falls back to the +// built-in default so the pool always has at least one worker. +func NewGatewayReconcileQueue(ctx context.Context, handler Handler[*pb.Gateway], workers int) *GatewayReconcileQueue { + return &GatewayReconcileQueue{ + q: newReconcileQueue(ctx, "Gateway", handler, + withRetryTransform(clearGatewayPhaseForRetry), + withVersion(gatewayEventVersion), + withWorkers[*pb.Gateway](gatewayWorkerCount(workers))), + } +} + +// EnqueueForced requests reconciliation of the given gateway, marking it so the +// next handler attempt bypasses the reconciler phase gate (as recovery seeds do). +// This is what lets a release image change re-reconcile a gateway that is already +// Running, which the phase gate would otherwise skip. +func (g *GatewayReconcileQueue) EnqueueForced(ev Event[*pb.Gateway]) { g.q.enqueueForced(ev) } + +// Stop drains and shuts the queue down. +func (g *GatewayReconcileQueue) Stop() { g.q.stop() } + +// WatchGateways streams gateway events and drives them through the caller-owned +// per-resource reconcile queue. When clusterID is non-empty the watch and its +// seed lists are scoped server-side to gateways with that cluster_id, so a +// managed-cluster spoke only ever reconciles its own gateways (the pull model); +// empty watches every gateway. The queue is owned and stopped by the caller +// (main) and shared with out-of-band enqueuers such as the GatewayRelease +// reconciler, so it is neither created nor stopped here. +func WatchGateways(ctx context.Context, conn *grpc.ClientConn, queue *GatewayReconcileQueue, clusterID string) error { client := pb.NewGatewayServiceClient(conn) // Gateway reconciliation is driven through a per-resource reconcile queue rather // than invoked inline: the watch stream does not replay state on reconnect, so a @@ -407,12 +445,10 @@ func WatchGateways(ctx context.Context, conn *grpc.ClientConn, handler Handler[* // Failed phase) would otherwise strand the gateway until its spec next changes. // The queue serializes work per gateway, coalesces to the latest observed state, // and retries failures indefinitely with capped backoff -- all on the watcher - // lifetime context so recovery survives a stream reconnect. - rq := newReconcileQueue(ctx, "Gateway", handler, - withRetryTransform(clearGatewayPhaseForRetry), - withVersion(gatewayEventVersion), - withWorkers[*pb.Gateway](workers)) - defer rq.stop() + // lifetime context so recovery survives a stream reconnect. The queue is owned + // by the caller (main) and shared with out-of-band enqueuers such as the + // GatewayRelease reconciler, so it is neither created nor stopped here. + rq := queue.q return watchLoop(ctx, "Gateway", func(ctx context.Context) error { // Derive a cancelable child before creating the stream so either the // receiver or the seed can cancel and join the other without waiting diff --git a/specs/platform/control-plane.spec.md b/specs/platform/control-plane.spec.md index 7ce6c231..146ff273 100644 --- a/specs/platform/control-plane.spec.md +++ b/specs/platform/control-plane.spec.md @@ -52,6 +52,7 @@ Gateway reconciliation is defined in detail across dedicated sub-specs: | [`openshell-gateway-oidc.spec.md`](./openshell-gateway-oidc.spec.md) | OIDC authentication, role validation, gateway.toml injection | | [`openshell-gateway-health.spec.md`](./openshell-gateway-health.spec.md) | Phase lifecycle, workload-readiness gating, continuous health reconciliation | | [`gateway-version-selection.spec.md`](./gateway-version-selection.spec.md) | Database-backed version selection: resolving `release_id` to a GatewayRelease image and its precedence over a direct image | +| [`gateway-release-reconciliation.spec.md`](./gateway-release-reconciliation.spec.md) | GatewayRelease reconciliation: image validation, deterministic release status, change propagation to referencing gateways | ### Config diff --git a/specs/platform/gateway-release-reconciliation.spec.md b/specs/platform/gateway-release-reconciliation.spec.md new file mode 100644 index 00000000..6d265209 --- /dev/null +++ b/specs/platform/gateway-release-reconciliation.spec.md @@ -0,0 +1,175 @@ +# GatewayRelease Reconciliation + +**Date:** 2026-09-02 +**Status:** Active + +## Purpose + +This spec defines how the HyperShell control plane reconciles `GatewayRelease` +resources. A `GatewayRelease` is a database-backed record of a versioned gateway +container image; it has **no direct Kubernetes footprint** of its own. Reconciling +a release therefore means (1) validating the release's image reference, +(2) recording a deterministic, observable `status` back on the release, and +(3) propagating an effective image change to every Gateway that references the +release so the cluster converges toward the new desired version. + +Today the `GatewayReleaseReconciler` is a no-op: it dedups, opens a trace span, +logs, and returns `nil` without validating the release or affecting any gateway. +This spec replaces that behavior with a deterministic reconciliation contract. + +This spec is a sub-spec of [`control-plane.spec.md`](./control-plane.spec.md) and +refines its "Manage release rollouts" and "Update resource status back to the API +server" responsibilities for the release resource specifically. + +### Scope Boundary + +- **In scope:** validating a release, writing back its deterministic status, and + requesting reconciliation of Gateways that reference the release when its + effective image changes. +- **Out of scope (sibling tickets):** how a Gateway resolves its `release_id` to + a concrete image at deploy time is defined by database-backed gateway version + selection; safe/progressive rollout ordering and canary traffic strategies are + defined by the release-rollout and canary specs. This spec only guarantees that + a release change causes the referencing Gateways to be re-reconciled; the + resulting deploy behavior is owned by those specs. + +## Domain Vocabulary + +A `GatewayRelease` carries a `status` field that the control plane owns and keeps +current to reflect the reconciled validation outcome. Allowed control-plane-owned +values: + +- **`Available`** - the release's image reference is well-formed and the release + is eligible to be used by Gateways. +- **`Invalid`** - the release's image reference failed validation; the value + includes a short human-readable reason. Gateways SHOULD NOT be driven toward an + `Invalid` release's image. + +The `status` string is a short, human-readable descriptor surfaced in the console +and CLI alongside the release. + +## Requirements + +### Requirement: Release Image Validation + +The control plane SHALL validate a `GatewayRelease`'s `image` reference on every +create and update event, using the same image-reference rules applied to gateway +workloads (well-formed reference format; rejection of shell-injection +metacharacters). A release whose `image` is empty or malformed SHALL be treated as +invalid and SHALL NOT be propagated to any Gateway. + +#### Scenario: Well-formed image passes validation + +- GIVEN a `GatewayRelease` with `image: registry.redhat.io/openshell/gateway:v1.2.0` +- WHEN the control plane reconciles the release +- THEN validation succeeds +- AND the release becomes eligible for propagation + +#### Scenario: Malformed image fails validation + +- GIVEN a `GatewayRelease` with `image: "gateway:v1; rm -rf /"` +- WHEN the control plane reconciles the release +- THEN validation fails with a reason describing the invalid reference +- AND no Gateway is reconciled as a result of this release + +### Requirement: Deterministic Release Status Write-Back + +The control plane SHALL write the reconciled release's status back to the API +server so the persisted `status` deterministically reflects the reconcile outcome: +`Available` on successful validation, or `Invalid` with a reason on failed +validation. The write-back SHALL be idempotent: the control plane SHALL NOT issue +a status update when the persisted `status` already equals the desired value. + +#### Scenario: Status settles to Available + +- GIVEN a `GatewayRelease` whose image passes validation and whose persisted + `status` is unset or not `Available` +- WHEN the control plane reconciles the release +- THEN the control plane updates the release `status` to `Available` + +#### Scenario: Status settles to Invalid with a reason + +- GIVEN a `GatewayRelease` whose image fails validation +- WHEN the control plane reconciles the release +- THEN the control plane updates the release `status` to `Invalid` including the + validation reason + +#### Scenario: No redundant status write + +- GIVEN a `GatewayRelease` whose persisted `status` is already `Available` +- WHEN the control plane reconciles the release and validation still passes +- THEN the control plane makes no status update call for the release + +### Requirement: Change Propagation to Referencing Gateways + +When a reconcile determines that a valid release's effective image has changed, the +control plane SHALL request reconciliation of every Gateway whose `release_id` +references that release, so each referencing Gateway converges toward the new +desired version. Propagation SHALL be limited to Gateways that reference the +release by `release_id`; Gateways that pin an explicit `image` and do not +reference the release SHALL NOT be disturbed. + +#### Scenario: Image change fans out to referencing gateways + +- GIVEN a valid `GatewayRelease` `r1` referenced by Gateways `g1` and `g2` via + `release_id` +- AND a Gateway `g3` that does not reference `r1` +- WHEN `r1`'s image is updated to a new valid reference +- THEN the control plane requests reconciliation of `g1` and `g2` +- AND the control plane does not request reconciliation of `g3` + +#### Scenario: Invalid release does not fan out + +- GIVEN a `GatewayRelease` `r1` referenced by Gateway `g1` +- WHEN `r1` is updated to a malformed image +- THEN the release status settles to `Invalid` +- AND `g1` is not driven toward the invalid image + +#### Scenario: No image change does not fan out + +- GIVEN a valid `GatewayRelease` `r1` referenced by Gateway `g1` +- WHEN `r1` is updated in a way that does not change its effective image + (for example, a rename) +- THEN the control plane does not request reconciliation of `g1` on account of the + image + +### Requirement: Release Deletion Has No Cluster Footprint + +A `GatewayRelease` delete event SHALL NOT remove or disrupt any running Gateway +workload, because a release owns no Kubernetes resources. The control plane SHALL +treat a release delete as a terminal, idempotent no-op with respect to cluster +state, and SHALL NOT error when the release is already absent. + +#### Scenario: Deleting a release leaves running gateways untouched + +- GIVEN a `GatewayRelease` `r1` that Gateway `g1` was deployed from +- WHEN `r1` is deleted +- THEN `g1`'s running workload is unchanged +- AND the control plane reports the release reconcile as successful + +### Requirement: Idempotent, Serialized Reconciliation + +Release reconciliation SHALL be idempotent and SHALL be serialized per release so +that a retry never runs concurrently with a live event for the same release. +Re-reconciling an unchanged release SHALL converge to the same status and SHALL NOT +produce redundant status writes or redundant gateway fan-out. + +#### Scenario: Repeated reconciles are stable + +- GIVEN a valid `GatewayRelease` already reconciled to `Available` +- WHEN the control plane reconciles the same release again with no change +- THEN no status update and no gateway reconciliation are requested + +### Requirement: Failure Handling Is Retried, Not Swallowed + +When a release reconcile cannot complete because a dependency is transiently +unavailable (for example, the API server rejects the status write, or the set of +referencing Gateways cannot be listed), the control plane SHALL return an error so +the reconcile is requeued and retried, rather than silently succeeding. Partial +failures SHALL NOT be silently swallowed. + +#### Scenario: Status write failure is retried + +- GIVEN a valid `GatewayRelease` whose status must be updated to `Available` +- WHEN the status write to the API server fails transiently +- THEN the reconcile returns an error and is requeued for retry From 7102a59e4ced9f7683e034bb17a79165a5386606 Mon Sep 17 00:00:00 2001 From: JuanmaBM Date: Thu, 10 Sep 2026 10:17:55 +0200 Subject: [PATCH 2/2] fix(control-plane): cluster-scope GatewayRelease fan-out Scope the release fan-out's gateway listing to the control plane's own managed cluster. On a managed-cluster spoke the GatewayRelease watch runs on every control plane, so an unscoped ListGateways matched and force-enqueued gateways owned by other clusters, breaking pull-model isolation. Reuse listAllGateways(ctx, client, clusterID) so the request carries the server-side cluster filter, and filter by release_id client-side; this also removes the duplicated pagination loop. Also drop the now-redundant per-release active-set guard: the reconcile queue that drives the handler already serializes per release, and the guard returned nil (success) on a spurious skip, which could mask a dropped reconcile from the queue's retry/backoff. Add regression tests for the cluster-scoped and single-cluster fan-out listings, and document the cluster-scoping requirement plus the known restart-window change-detection gap in the spec. Co-Authored-By: Claude Opus 4.8 --- .../cmd/hypershell-controller/main.go | 2 +- .../reconciler/gateway_release_test.go | 67 ++++++++++++++++-- .../internal/reconciler/reconciler.go | 69 ++++++++----------- .../gateway-release-reconciliation.spec.md | 25 +++++++ 4 files changed, 118 insertions(+), 45 deletions(-) diff --git a/components/control-plane/cmd/hypershell-controller/main.go b/components/control-plane/cmd/hypershell-controller/main.go index 44e31b97..a142e98c 100644 --- a/components/control-plane/cmd/hypershell-controller/main.go +++ b/components/control-plane/cmd/hypershell-controller/main.go @@ -245,7 +245,7 @@ func main() { // release reconciler can hold the same instance. gatewayQueue := watcher.NewGatewayReconcileQueue(ctx, gatewayReconciler, cfg.GatewayReconcileWorkers) defer gatewayQueue.Stop() - releaseReconciler := reconciler.NewGatewayReleaseReconciler(conn, gatewayQueue) + releaseReconciler := reconciler.NewGatewayReleaseReconciler(conn, gatewayQueue, cfg.ClusterID) watchCount := 4 // managed clusters, gateway releases, gateways, networks if databaseReconciler != nil { diff --git a/components/control-plane/internal/reconciler/gateway_release_test.go b/components/control-plane/internal/reconciler/gateway_release_test.go index 9e0af5f6..61f2e37f 100644 --- a/components/control-plane/internal/reconciler/gateway_release_test.go +++ b/components/control-plane/internal/reconciler/gateway_release_test.go @@ -26,14 +26,18 @@ func (f *fakeReleaseClient) UpdateGatewayRelease(ctx context.Context, in *pb.Upd return &pb.UpdateGatewayReleaseResponse{}, nil } -// fakeReleaseGatewayClient serves a fixed gateway inventory to ListGateways. +// fakeReleaseGatewayClient serves a fixed gateway inventory to ListGateways and +// records the ClusterId filter each call carried so tests can assert the fan-out +// is cluster-scoped. type fakeReleaseGatewayClient struct { pb.GatewayServiceClient - gateways []*pb.Gateway - listErr error + gateways []*pb.Gateway + listErr error + gotClusterIDs []*string } func (f *fakeReleaseGatewayClient) ListGateways(ctx context.Context, in *pb.ListGatewaysRequest, opts ...grpc.CallOption) (*pb.ListGatewaysResponse, error) { + f.gotClusterIDs = append(f.gotClusterIDs, in.ClusterId) if f.listErr != nil { return nil, f.listErr } @@ -54,7 +58,6 @@ func (r *recordingEnqueuer) EnqueueForced(ev watcher.Event[*pb.Gateway]) { func newTestReleaseReconciler(gw pb.GatewayServiceClient, rel pb.GatewayReleaseServiceClient, q gatewayEnqueuer) *GatewayReleaseReconciler { return &GatewayReleaseReconciler{ - active: make(map[string]struct{}), lastImage: make(map[string]string), gateways: gw, releases: rel, @@ -200,6 +203,62 @@ func TestGatewayRelease_ImageChangeFansOutToReferencingGatewaysOnly(t *testing.T } } +// On a managed-cluster spoke (clusterID set) the release fan-out MUST scope its +// gateway listing server-side to its own cluster; otherwise it would match and +// force-enqueue gateways owned by other clusters, breaking pull-model isolation. +func TestGatewayRelease_FanOutIsClusterScoped(t *testing.T) { + rel := &fakeReleaseClient{} + gw := &fakeReleaseGatewayClient{gateways: []*pb.Gateway{gatewayWithRelease("g1", "r1")}} + q := &recordingEnqueuer{} + r := newTestReleaseReconciler(gw, rel, q) + r.clusterID = "spoke-a" + + // Baseline, then an image change to trigger the fan-out list. + if err := r.Handle(context.Background(), releaseEvent(watcher.EventCreated, "r1", "registry.redhat.io/openshell/gateway:v1", releaseStatusAvailable)); err != nil { + t.Fatalf("seed: %v", err) + } + if err := r.Handle(context.Background(), releaseEvent(watcher.EventUpdated, "r1", "registry.redhat.io/openshell/gateway:v2", releaseStatusAvailable)); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if len(gw.gotClusterIDs) == 0 { + t.Fatalf("expected at least one ListGateways call") + } + for i, got := range gw.gotClusterIDs { + if got == nil { + t.Fatalf("ListGateways call %d was not cluster-scoped: ClusterId=nil", i) + } + if *got != "spoke-a" { + t.Fatalf("ListGateways call %d scoped to %q, want %q", i, *got, "spoke-a") + } + } +} + +// In single-cluster mode (empty clusterID) the fan-out list carries no ClusterId +// filter, so every gateway in the fleet is a candidate. +func TestGatewayRelease_FanOutSingleClusterHasNoFilter(t *testing.T) { + rel := &fakeReleaseClient{} + gw := &fakeReleaseGatewayClient{gateways: []*pb.Gateway{gatewayWithRelease("g1", "r1")}} + q := &recordingEnqueuer{} + r := newTestReleaseReconciler(gw, rel, q) + + if err := r.Handle(context.Background(), releaseEvent(watcher.EventCreated, "r1", "registry.redhat.io/openshell/gateway:v1", releaseStatusAvailable)); err != nil { + t.Fatalf("seed: %v", err) + } + if err := r.Handle(context.Background(), releaseEvent(watcher.EventUpdated, "r1", "registry.redhat.io/openshell/gateway:v2", releaseStatusAvailable)); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if len(gw.gotClusterIDs) == 0 { + t.Fatalf("expected at least one ListGateways call") + } + for i, got := range gw.gotClusterIDs { + if got != nil { + t.Fatalf("ListGateways call %d carried ClusterId=%q, want nil (single-cluster)", i, *got) + } + } +} + func TestGatewayRelease_RenameDoesNotFanOut(t *testing.T) { rel := &fakeReleaseClient{} gw := &fakeReleaseGatewayClient{gateways: []*pb.Gateway{gatewayWithRelease("g1", "r1")}} diff --git a/components/control-plane/internal/reconciler/reconciler.go b/components/control-plane/internal/reconciler/reconciler.go index eee71940..3931d2df 100644 --- a/components/control-plane/internal/reconciler/reconciler.go +++ b/components/control-plane/internal/reconciler/reconciler.go @@ -1167,10 +1167,6 @@ func cnpgClusterGVR() schema.GroupVersionResource { const ( releaseStatusAvailable = "Available" releaseStatusInvalid = "Invalid" - // releaseFanOutPageSize is the page size used when listing gateways to find - // the ones that reference a changed release. It matches the other reconcilers' - // list page size so a typical fleet is covered in a single request. - releaseFanOutPageSize = 500 ) // gatewayEnqueuer requests a gateway be re-reconciled through the shared gateway @@ -1192,8 +1188,7 @@ type gatewayEnqueuer interface { // owned by sibling specs; this reconciler only guarantees the referencing // gateways are re-reconciled. type GatewayReleaseReconciler struct { - mu sync.Mutex - active map[string]struct{} + mu sync.Mutex // lastImage records the last validated image observed per release ID so an // update that does not change the effective image does not fan out, and so the // first observation of a release (e.g. on controller start or a fresh create) @@ -1204,6 +1199,12 @@ type GatewayReleaseReconciler struct { gateways pb.GatewayServiceClient releases pb.GatewayReleaseServiceClient gwQueue gatewayEnqueuer + // clusterID scopes the release fan-out's gateway listing to this control + // plane's own cluster. On a managed-cluster spoke (non-empty) the GatewayRelease + // watch runs on every control plane, so an unscoped list would match and force + // foreign gateways into the local reconcile queue, breaking pull-model + // isolation; empty means single-cluster (no server-side filter). + clusterID string } // NewGatewayReleaseReconciler builds the release reconciler. conn is the API @@ -1211,12 +1212,14 @@ type GatewayReleaseReconciler struct { // gateways; gwQueue is the shared gateway reconcile queue used to propagate image // changes. Either dependency may be nil (e.g. when the controller runs without a // Kubernetes client), in which case propagation is skipped but validation and -// status write-back still run. -func NewGatewayReleaseReconciler(conn *grpc.ClientConn, gwQueue gatewayEnqueuer) *GatewayReleaseReconciler { +// status write-back still run. clusterID is this control plane's managed-cluster +// identity (empty in single-cluster mode); it scopes the fan-out's gateway +// listing so a spoke never force-reconciles another cluster's gateways. +func NewGatewayReleaseReconciler(conn *grpc.ClientConn, gwQueue gatewayEnqueuer, clusterID string) *GatewayReleaseReconciler { r := &GatewayReleaseReconciler{ - active: make(map[string]struct{}), lastImage: make(map[string]string), gwQueue: gwQueue, + clusterID: clusterID, } if conn != nil { r.gateways = pb.NewGatewayServiceClient(conn) @@ -1226,19 +1229,10 @@ func NewGatewayReleaseReconciler(conn *grpc.ClientConn, gwQueue gatewayEnqueuer) } func (r *GatewayReleaseReconciler) Handle(ctx context.Context, event watcher.Event[*pb.GatewayRelease]) error { - r.mu.Lock() - if _, ok := r.active[event.ResourceID]; ok { - r.mu.Unlock() - return nil - } - r.active[event.ResourceID] = struct{}{} - r.mu.Unlock() - defer func() { - r.mu.Lock() - delete(r.active, event.ResourceID) - r.mu.Unlock() - }() - + // Per-release serialization is owned by the reconcile queue that drives this + // handler (WatchGatewayReleases), so no in-handler active-set guard is needed; + // adding one back would risk returning nil (success) on a spurious skip and + // masking a dropped reconcile from the queue's retry/backoff. _, endSpan := cpotel.StartReconcileSpan(ctx, "GatewayRelease", event.Type.String(), event.Resource.GetMetadata().GetTraceparent()) var reconcileErr error defer func() { endSpan(reconcileErr) }() @@ -1343,28 +1337,23 @@ func (r *GatewayReleaseReconciler) propagateToGateways(ctx context.Context, rele } // listGatewaysForRelease returns every gateway whose release_id references the -// given release, paginating through the API server. +// given release. It reuses listAllGateways so the listing is scoped to this +// control plane's cluster (via r.clusterID): on a managed-cluster spoke this +// prevents matching and force-enqueuing gateways owned by other clusters, which +// would violate pull-model isolation. Filtering by release_id is done +// client-side; a server-side filter is a scale follow-up. func (r *GatewayReleaseReconciler) listGatewaysForRelease(ctx context.Context, releaseID string) ([]*pb.Gateway, error) { + all, err := listAllGateways(ctx, r.gateways, r.clusterID) + if err != nil { + return nil, err + } var matching []*pb.Gateway - for page := int32(1); ; page++ { - resp, err := r.gateways.ListGateways(ctx, &pb.ListGatewaysRequest{ - Page: page, - Size: releaseFanOutPageSize, - }) - if err != nil { - return nil, err - } - items := resp.GetItems() - for _, gw := range items { - if gw.GetReleaseId() == releaseID { - matching = append(matching, gw) - } - } - total := int(resp.GetMetadata().GetTotal()) - if len(items) == 0 || len(items) < releaseFanOutPageSize || (total > 0 && page*releaseFanOutPageSize >= int32(total)) { - return matching, nil + for _, gw := range all { + if gw.GetReleaseId() == releaseID { + matching = append(matching, gw) } } + return matching, nil } func (r *GatewayReleaseReconciler) lastImageFor(id string) (string, bool) { diff --git a/specs/platform/gateway-release-reconciliation.spec.md b/specs/platform/gateway-release-reconciliation.spec.md index 6d265209..754fc16e 100644 --- a/specs/platform/gateway-release-reconciliation.spec.md +++ b/specs/platform/gateway-release-reconciliation.spec.md @@ -32,6 +32,14 @@ server" responsibilities for the release resource specifically. defined by the release-rollout and canary specs. This spec only guarantees that a release change causes the referencing Gateways to be re-reconciled; the resulting deploy behavior is owned by those specs. +- **Known limitation (out of scope here):** effective-image change detection is + anchored to a control-plane-local baseline of the last observed image per + release, and the release watch has no startup seed. A release image edited while + the control plane is down is therefore not re-observed on restart, so already + running referencing Gateways are not fanned out until the release changes again. + Closing this restart-window gap (durable/generation-anchored change detection or + a periodic release resync) is owned by the control-plane world-synchronization + and reconciliation-contract specs. ## Domain Vocabulary @@ -109,6 +117,13 @@ desired version. Propagation SHALL be limited to Gateways that reference the release by `release_id`; Gateways that pin an explicit `image` and do not reference the release SHALL NOT be disturbed. +The fan-out's Gateway listing SHALL be scoped to the control plane's own managed +cluster: on a managed-cluster spoke (a non-empty cluster identity, the pull model) +the release watch runs on every control plane, so the listing SHALL apply the +server-side cluster filter and SHALL NOT match, or request reconciliation of, +Gateways owned by another cluster. In single-cluster mode (empty cluster identity) +no cluster filter is applied and the whole fleet is eligible. + #### Scenario: Image change fans out to referencing gateways - GIVEN a valid `GatewayRelease` `r1` referenced by Gateways `g1` and `g2` via @@ -125,6 +140,16 @@ reference the release SHALL NOT be disturbed. - THEN the release status settles to `Invalid` - AND `g1` is not driven toward the invalid image +#### Scenario: Fan-out on a spoke is scoped to its own cluster + +- GIVEN a managed-cluster spoke with cluster identity `spoke-a` +- AND a `GatewayRelease` `r1` referenced by Gateway `g1` on `spoke-a` and by + Gateway `g2` on another cluster +- WHEN `r1`'s image changes and the spoke reconciles it +- THEN the control plane lists Gateways scoped to `spoke-a` +- AND it requests reconciliation of `g1` +- AND it does not match or request reconciliation of `g2` + #### Scenario: No image change does not fan out - GIVEN a valid `GatewayRelease` `r1` referenced by Gateway `g1`