From d5194ce3c22f85b5ea388c5aedd215e233338ec7 Mon Sep 17 00:00:00 2001 From: Maxence Maireaux Date: Mon, 31 Aug 2026 21:39:10 +0200 Subject: [PATCH 1/4] feat(operator): configure Ledger v3 NATS sinks --- .../formance.com_ledgerconfigurations.yaml | 241 +++++++++++++----- .../settings.catalog.json | 2 +- go.mod | 2 +- go.sum | 4 +- ...ion_ledgerconfigurations.formance.com.yaml | 241 +++++++++++++----- .../resources/brokerconsumers/controller.go | 74 ++++-- .../brokerconsumers/controller_test.go | 52 ++++ internal/resources/brokers/reconcile.go | 42 +-- internal/resources/brokers/utils.go | 27 +- internal/resources/brokers/utils_test.go | 29 +++ internal/resources/ledgers/init.go | 2 + internal/resources/ledgers/v3.go | 5 + internal/resources/ledgers/v3_sinks.go | 82 ++++++ internal/resources/ledgers/v3_sinks_test.go | 42 +++ internal/resources/ledgers/v3_spec.go | 4 + internal/resources/ledgers/v3_spec_test.go | 19 ++ tools/kubectl-stacks/go.mod | 2 +- tools/kubectl-stacks/go.sum | 4 +- 18 files changed, 693 insertions(+), 181 deletions(-) create mode 100644 internal/resources/brokerconsumers/controller_test.go create mode 100644 internal/resources/ledgers/v3_sinks.go create mode 100644 internal/resources/ledgers/v3_sinks_test.go diff --git a/config/crd/bases/formance.com_ledgerconfigurations.yaml b/config/crd/bases/formance.com_ledgerconfigurations.yaml index c4e87596..76ef2015 100644 --- a/config/crd/bases/formance.com_ledgerconfigurations.yaml +++ b/config/crd/bases/formance.com_ledgerconfigurations.yaml @@ -1234,67 +1234,90 @@ spec: LogLevel takes precedence when both are set; prefer LogLevel for new manifests since it also unlocks the trace level. type: boolean - dnsEndpoint: - description: DNSEndpoint configuration for ExternalDNS. - properties: - annotations: - additionalProperties: - type: string - description: Annotations to add to the DNSEndpoint resource. - type: object - enabled: - description: Enabled enables the DNSEndpoint resource. - type: boolean - endpoints: - description: Endpoints is the list of DNS endpoint entries. - items: - description: DNSEndpointEntry defines a single DNS endpoint. - properties: - dnsName: - description: DNSName is the hostname for the DNS record. - type: string - providerSpecific: - description: ProviderSpecific holds provider-specific - properties. - items: - description: ProviderSpecificProperty defines a provider-specific - key-value pair. - properties: - name: - description: Name is the property name. - type: string - value: - description: Value is the property value. - type: string - required: - - name - - value - type: object - type: array - recordTTL: - description: RecordTTL is the TTL in seconds for the - DNS record. - format: int64 - type: integer - recordType: - description: RecordType is the DNS record type (e.g., - CNAME, A). Defaults to CNAME. - type: string - targets: - description: Targets is the list of target hostnames - or IPs. - items: - type: string - type: array - required: - - dnsName - - targets + dnsEndpoints: + description: |- + DNSEndpoints configures one or more ExternalDNS DNSEndpoint resources. + Each entry is reconciled into its own DNSEndpoint object, so distinct + endpoints (e.g. a public and a private one) can carry different + annotations. Entries are keyed by their unique name. + items: + description: DNSEndpointSpec defines a single ExternalDNS DNSEndpoint + configuration. + properties: + annotations: + additionalProperties: + type: string + description: Annotations to add to the DNSEndpoint resource. type: object - type: array - type: object - x-kubernetes-validations: - - message: endpoints are required when dnsEndpoint is enabled - rule: '!self.enabled || size(self.endpoints) > 0' + enabled: + description: Enabled enables the DNSEndpoint resource. + type: boolean + endpoints: + description: Endpoints is the list of DNS endpoint entries. + items: + description: DNSEndpointEntry defines a single DNS endpoint. + properties: + dnsName: + description: DNSName is the hostname for the DNS record. + type: string + providerSpecific: + description: ProviderSpecific holds provider-specific + properties. + items: + description: ProviderSpecificProperty defines a + provider-specific key-value pair. + properties: + name: + description: Name is the property name. + type: string + value: + description: Value is the property value. + type: string + required: + - name + - value + type: object + type: array + recordTTL: + description: RecordTTL is the TTL in seconds for the + DNS record. + format: int64 + type: integer + recordType: + description: RecordType is the DNS record type (e.g., + CNAME, A). Defaults to CNAME. + type: string + targets: + description: Targets is the list of target hostnames + or IPs. + items: + type: string + type: array + required: + - dnsName + - targets + type: object + type: array + name: + description: |- + Name is a unique, stable identifier for this DNSEndpoint within the + Cluster. It is used as the suffix of the generated DNSEndpoint object's + name (e.g. "public", "private"), so it must be unique across entries and + a valid DNS-1123 label (it is concatenated into the object's metadata.name). + maxLength: 63 + minLength: 1 + pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?$ + type: string + required: + - name + type: object + x-kubernetes-validations: + - message: endpoints are required when the DNSEndpoint is enabled + rule: '!self.enabled || size(self.endpoints) > 0' + type: array + x-kubernetes-list-map-keys: + - name + x-kubernetes-list-type: map extraEnv: description: ExtraEnv is a list of additional environment variables to inject into ledger containers. @@ -2879,8 +2902,10 @@ spec: type: object type: object queryProfileThreshold: - description: QueryProfileThreshold logs and emits OTel attributes - for queries exceeding this duration (0 to disable). + description: |- + QueryProfileThreshold logs and emits OTel attributes for reads whose total + server-side handling (excluding the caller-requested read barrier) exceeds + this duration. 0 disables the slow-read log. type: string raft: description: Raft consensus configuration. @@ -3609,6 +3634,98 @@ spec: description: Name overrides the service account name. type: string type: object + sinks: + description: |- + Sinks declares the NATS event sinks the operator maintains in the + Ledger cluster's Raft-replicated runtime configuration. + + Ownership semantics: + - Field ABSENT (nil) => unmanaged. The operator never lists, creates, + updates, or removes event sinks. This preserves sinks configured + directly with ledgerctl before the CRD field existed. + - Field PRESENT (even empty {}) => managed. The operator creates the + declared sinks and removes only sinks it previously created (tracked + in status.appliedSinks). Externally-created sinks are never adopted or + removed. An empty object therefore means "remove the sinks previously + managed by this Cluster", not "remove every sink". + + Sink changes are applied at runtime through Ledger's Raft API and do not + roll the StatefulSet. + properties: + nats: + description: |- + NATS is the set of NATS JetStream sinks to maintain. Names are unique and + stable: Ledger uses them to persist each sink's delivery cursor and status. + items: + description: NATSEventSinkSpec configures one NATS JetStream + event sink. + properties: + batchDelayMs: + description: |- + BatchDelayMs is the maximum delay before a partial batch is published. + Zero lets Ledger use its default (10ms). + format: int64 + minimum: 0 + type: integer + batchSize: + description: |- + BatchSize is the maximum number of events published per batch. Zero lets + Ledger use its default (64). + format: int32 + maximum: 100000 + minimum: 0 + type: integer + eventTypes: + description: EventTypes filters the emitted events. + Empty means all event types. + items: + enum: + - COMMITTED_TRANSACTION + - REVERTED_TRANSACTION + - SAVED_METADATA + - DELETED_METADATA + - CREATED_LEDGER + - DELETED_LEDGER + - SKIPPED_ORDER + type: string + type: array + x-kubernetes-list-type: set + format: + default: json + description: Format is the event payload serialization + format. Defaults to json. + enum: + - json + - protobuf + type: string + name: + description: Name is the stable sink identifier. + maxLength: 253 + minLength: 1 + type: string + topic: + description: |- + Topic is the subject prefix. Ledger publishes to + ... + minLength: 1 + type: string + url: + description: |- + URL is the NATS server URL, including the nats:// scheme. Credentials + must not be embedded in this field because Cluster specs are not secret. + minLength: 1 + pattern: ^nats://[^@]+$ + type: string + required: + - name + - topic + - url + type: object + type: array + x-kubernetes-list-map-keys: + - name + x-kubernetes-list-type: map + type: object snapshot: description: Snapshot sync configuration for Raft snapshot transfers. properties: diff --git a/docs/09-Configuration reference/settings.catalog.json b/docs/09-Configuration reference/settings.catalog.json index f8c6e3f2..79788911 100644 --- a/docs/09-Configuration reference/settings.catalog.json +++ b/docs/09-Configuration reference/settings.catalog.json @@ -33,7 +33,7 @@ "key": "broker.dsn", "valueType": "uri", "sources": [ - "internal/resources/brokers/reconcile.go:24", + "internal/resources/brokers/reconcile.go:26", "internal/resources/webhooks/deployment.go:30" ] }, diff --git a/go.mod b/go.mod index ea730008..b9d3c139 100644 --- a/go.mod +++ b/go.mod @@ -48,7 +48,7 @@ require ( github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/emicklei/go-restful/v3 v3.13.0 // indirect github.com/evanphx/json-patch/v5 v5.9.11 // indirect - github.com/formancehq/ledger/misc/operator v0.0.0-20260715094310-76862ea0b681 + github.com/formancehq/ledger/misc/operator v0.0.0-20260831193659-af94b64e0f35 github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-logr/zapr v1.3.0 // indirect diff --git a/go.sum b/go.sum index 44ff5040..53541468 100644 --- a/go.sum +++ b/go.sum @@ -18,8 +18,8 @@ github.com/evanphx/json-patch/v5 v5.9.11 h1:/8HVnzMq13/3x9TPvjG08wUGqBTmZBsCWzjT github.com/evanphx/json-patch/v5 v5.9.11/go.mod h1:3j+LviiESTElxA4p3EMKAB9HXj3/XEtnUf6OZxqIQTM= github.com/formancehq/go-libs/v5 v5.2.0 h1:TpS47F8X5g5cHhnecfD20TrcdBqUVGy/ezZv0oFaQjc= github.com/formancehq/go-libs/v5 v5.2.0/go.mod h1:ms6tCGw1yqB4qtEbAuqPOQegWo4rU48vDobNkK7Ak6U= -github.com/formancehq/ledger/misc/operator v0.0.0-20260715094310-76862ea0b681 h1:PzhfbpKZqMJd4opnKmEtFq0Hx/6Pxgq9GT7XzWcr+Ss= -github.com/formancehq/ledger/misc/operator v0.0.0-20260715094310-76862ea0b681/go.mod h1:tZa1TFBcXJxc1R0NqnsN6gOGoskd82a7wjmGqCBhOBU= +github.com/formancehq/ledger/misc/operator v0.0.0-20260831193659-af94b64e0f35 h1:tkiLCD/yg9wO2pS9o5yuhKV0JDHMVMCVx1I+sum3YY0= +github.com/formancehq/ledger/misc/operator v0.0.0-20260831193659-af94b64e0f35/go.mod h1:tZa1TFBcXJxc1R0NqnsN6gOGoskd82a7wjmGqCBhOBU= github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k= github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0= github.com/fxamacker/cbor/v2 v2.9.0 h1:NpKPmjDBgUfBms6tr6JZkTHtfFGcMKsw3eGcmD/sapM= diff --git a/helm/crds/templates/crds/apiextensions.k8s.io_v1_customresourcedefinition_ledgerconfigurations.formance.com.yaml b/helm/crds/templates/crds/apiextensions.k8s.io_v1_customresourcedefinition_ledgerconfigurations.formance.com.yaml index e9c22a7d..8a264a90 100644 --- a/helm/crds/templates/crds/apiextensions.k8s.io_v1_customresourcedefinition_ledgerconfigurations.formance.com.yaml +++ b/helm/crds/templates/crds/apiextensions.k8s.io_v1_customresourcedefinition_ledgerconfigurations.formance.com.yaml @@ -1237,67 +1237,90 @@ spec: LogLevel takes precedence when both are set; prefer LogLevel for new manifests since it also unlocks the trace level. type: boolean - dnsEndpoint: - description: DNSEndpoint configuration for ExternalDNS. - properties: - annotations: - additionalProperties: - type: string - description: Annotations to add to the DNSEndpoint resource. - type: object - enabled: - description: Enabled enables the DNSEndpoint resource. - type: boolean - endpoints: - description: Endpoints is the list of DNS endpoint entries. - items: - description: DNSEndpointEntry defines a single DNS endpoint. - properties: - dnsName: - description: DNSName is the hostname for the DNS record. - type: string - providerSpecific: - description: ProviderSpecific holds provider-specific - properties. - items: - description: ProviderSpecificProperty defines a provider-specific - key-value pair. - properties: - name: - description: Name is the property name. - type: string - value: - description: Value is the property value. - type: string - required: - - name - - value - type: object - type: array - recordTTL: - description: RecordTTL is the TTL in seconds for the - DNS record. - format: int64 - type: integer - recordType: - description: RecordType is the DNS record type (e.g., - CNAME, A). Defaults to CNAME. - type: string - targets: - description: Targets is the list of target hostnames - or IPs. - items: - type: string - type: array - required: - - dnsName - - targets + dnsEndpoints: + description: |- + DNSEndpoints configures one or more ExternalDNS DNSEndpoint resources. + Each entry is reconciled into its own DNSEndpoint object, so distinct + endpoints (e.g. a public and a private one) can carry different + annotations. Entries are keyed by their unique name. + items: + description: DNSEndpointSpec defines a single ExternalDNS DNSEndpoint + configuration. + properties: + annotations: + additionalProperties: + type: string + description: Annotations to add to the DNSEndpoint resource. type: object - type: array - type: object - x-kubernetes-validations: - - message: endpoints are required when dnsEndpoint is enabled - rule: '!self.enabled || size(self.endpoints) > 0' + enabled: + description: Enabled enables the DNSEndpoint resource. + type: boolean + endpoints: + description: Endpoints is the list of DNS endpoint entries. + items: + description: DNSEndpointEntry defines a single DNS endpoint. + properties: + dnsName: + description: DNSName is the hostname for the DNS record. + type: string + providerSpecific: + description: ProviderSpecific holds provider-specific + properties. + items: + description: ProviderSpecificProperty defines a + provider-specific key-value pair. + properties: + name: + description: Name is the property name. + type: string + value: + description: Value is the property value. + type: string + required: + - name + - value + type: object + type: array + recordTTL: + description: RecordTTL is the TTL in seconds for the + DNS record. + format: int64 + type: integer + recordType: + description: RecordType is the DNS record type (e.g., + CNAME, A). Defaults to CNAME. + type: string + targets: + description: Targets is the list of target hostnames + or IPs. + items: + type: string + type: array + required: + - dnsName + - targets + type: object + type: array + name: + description: |- + Name is a unique, stable identifier for this DNSEndpoint within the + Cluster. It is used as the suffix of the generated DNSEndpoint object's + name (e.g. "public", "private"), so it must be unique across entries and + a valid DNS-1123 label (it is concatenated into the object's metadata.name). + maxLength: 63 + minLength: 1 + pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?$ + type: string + required: + - name + type: object + x-kubernetes-validations: + - message: endpoints are required when the DNSEndpoint is enabled + rule: '!self.enabled || size(self.endpoints) > 0' + type: array + x-kubernetes-list-map-keys: + - name + x-kubernetes-list-type: map extraEnv: description: ExtraEnv is a list of additional environment variables to inject into ledger containers. @@ -2882,8 +2905,10 @@ spec: type: object type: object queryProfileThreshold: - description: QueryProfileThreshold logs and emits OTel attributes - for queries exceeding this duration (0 to disable). + description: |- + QueryProfileThreshold logs and emits OTel attributes for reads whose total + server-side handling (excluding the caller-requested read barrier) exceeds + this duration. 0 disables the slow-read log. type: string raft: description: Raft consensus configuration. @@ -3612,6 +3637,98 @@ spec: description: Name overrides the service account name. type: string type: object + sinks: + description: |- + Sinks declares the NATS event sinks the operator maintains in the + Ledger cluster's Raft-replicated runtime configuration. + + Ownership semantics: + - Field ABSENT (nil) => unmanaged. The operator never lists, creates, + updates, or removes event sinks. This preserves sinks configured + directly with ledgerctl before the CRD field existed. + - Field PRESENT (even empty {}) => managed. The operator creates the + declared sinks and removes only sinks it previously created (tracked + in status.appliedSinks). Externally-created sinks are never adopted or + removed. An empty object therefore means "remove the sinks previously + managed by this Cluster", not "remove every sink". + + Sink changes are applied at runtime through Ledger's Raft API and do not + roll the StatefulSet. + properties: + nats: + description: |- + NATS is the set of NATS JetStream sinks to maintain. Names are unique and + stable: Ledger uses them to persist each sink's delivery cursor and status. + items: + description: NATSEventSinkSpec configures one NATS JetStream + event sink. + properties: + batchDelayMs: + description: |- + BatchDelayMs is the maximum delay before a partial batch is published. + Zero lets Ledger use its default (10ms). + format: int64 + minimum: 0 + type: integer + batchSize: + description: |- + BatchSize is the maximum number of events published per batch. Zero lets + Ledger use its default (64). + format: int32 + maximum: 100000 + minimum: 0 + type: integer + eventTypes: + description: EventTypes filters the emitted events. + Empty means all event types. + items: + enum: + - COMMITTED_TRANSACTION + - REVERTED_TRANSACTION + - SAVED_METADATA + - DELETED_METADATA + - CREATED_LEDGER + - DELETED_LEDGER + - SKIPPED_ORDER + type: string + type: array + x-kubernetes-list-type: set + format: + default: json + description: Format is the event payload serialization + format. Defaults to json. + enum: + - json + - protobuf + type: string + name: + description: Name is the stable sink identifier. + maxLength: 253 + minLength: 1 + type: string + topic: + description: |- + Topic is the subject prefix. Ledger publishes to + ... + minLength: 1 + type: string + url: + description: |- + URL is the NATS server URL, including the nats:// scheme. Credentials + must not be embedded in this field because Cluster specs are not secret. + minLength: 1 + pattern: ^nats://[^@]+$ + type: string + required: + - name + - topic + - url + type: object + type: array + x-kubernetes-list-map-keys: + - name + x-kubernetes-list-type: map + type: object snapshot: description: Snapshot sync configuration for Raft snapshot transfers. properties: diff --git a/internal/resources/brokerconsumers/controller.go b/internal/resources/brokerconsumers/controller.go index 10b48ea4..c812cbe4 100644 --- a/internal/resources/brokerconsumers/controller.go +++ b/internal/resources/brokerconsumers/controller.go @@ -28,10 +28,9 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" - collectionutils "github.com/formancehq/go-libs/v5/pkg/types/collections" - v1beta1 "github.com/formancehq/operator/v3/api/formance.com/v1beta1" "github.com/formancehq/operator/v3/internal/core" + "github.com/formancehq/operator/v3/internal/resources/brokers" "github.com/formancehq/operator/v3/internal/resources/jobs" "github.com/formancehq/operator/v3/internal/resources/registries" ) @@ -41,6 +40,8 @@ const ( ConditionTypeBrokerTopicCreated = "BrokerTopicCreated" ConditionTypeNatsStackConsumerCreated = "NatsStackConsumerCreated" ConditionTypeNatsServiceConsumerCreated = "NatsServiceConsumerCreated" + natsConsumerConfigurationRevision = "NestedSubjectsV1" + natsConsumerJobRevision = "ns1" ) //+kubebuilder:rbac:groups=formance.com,resources=brokerconsumers,verbs=get;list;watch;create;update;patch;delete @@ -134,6 +135,7 @@ func Reconcile(ctx core.Context, stack *v1beta1.Stack, consumer *v1beta1.BrokerC v1beta1.AndConditions( v1beta1.ConditionTypeMatch(ConditionTypeNatsStackConsumerCreated), v1beta1.ConditionGenerationMatch(consumer.Generation), + v1beta1.ConditionReasonMatch(natsConsumerConfigurationRevision), ), ) { if err := createStackNatsConsumer(ctx, stack, consumer, broker); err != nil { @@ -146,7 +148,7 @@ func Reconcile(ctx core.Context, stack *v1beta1.Stack, consumer *v1beta1.BrokerC v1beta1.AndConditions( v1beta1.ConditionTypeMatch(ConditionTypeNatsServiceConsumerCreated), v1beta1.ConditionGenerationMatch(consumer.Generation), - v1beta1.ConditionReasonMatch(service), + v1beta1.ConditionReasonMatch(natsServiceConditionReason(service)), ), ) { if err := createServiceNatsConsumer(ctx, stack, consumer, broker, service); err != nil { @@ -170,16 +172,21 @@ func Reconcile(ctx core.Context, stack *v1beta1.Stack, consumer *v1beta1.BrokerC func createServiceNatsConsumer(ctx core.Context, stack *v1beta1.Stack, consumer *v1beta1.BrokerConsumer, broker *v1beta1.Broker, service string) error { const script = ` - if ! nats --server "$NATS_URI" consumer info "$STACK-$SERVICE" "$NAME" --no-select >/dev/null 2>&1; then - nats --server "$NATS_URI" consumer add "$STACK-$SERVICE" "$NAME" \ + filters="" + for f in $SUBJECTS; do + filters="$filters --filter $f" + done + if nats --server "$NATS_URI" consumer info "$STREAM" "$NAME" --no-select >/dev/null 2>&1; then + nats --server "$NATS_URI" consumer edit "$STREAM" "$NAME" --force $filters + else + nats --server "$NATS_URI" consumer add "$STREAM" "$NAME" \ --deliver-group "$NAME" \ --deliver all \ --max-pending 1024 \ --ack explicit \ --target "$STACK-$NAME" \ --replay instant \ - --filter "$STACK-$SERVICE" \ - --defaults + --defaults $filters fi` natsBoxImage, err := registries.GetNatsBoxImage(ctx, stack, "0.19.2") @@ -187,7 +194,8 @@ func createServiceNatsConsumer(ctx core.Context, stack *v1beta1.Stack, consumer return err } - err = jobs.Handle(ctx, consumer, "cc-"+service, corev1.Container{ + topic := brokers.GetPublisherTopic(stack, broker, service) + err = jobs.Handle(ctx, consumer, "cc-"+natsConsumerJobRevision+"-"+service, corev1.Container{ Image: natsBoxImage.GetFullImageName(), Name: "create-consumer", Args: core.ShellScript(script), @@ -195,18 +203,19 @@ func createServiceNatsConsumer(ctx core.Context, stack *v1beta1.Stack, consumer core.Env("NATS_URI", fmt.Sprintf("nats://%s", broker.Status.URI.Host)), core.Env("STACK", stack.Name), core.Env("NAME", consumer.Spec.QueriedBy), - core.Env("SERVICE", service), + core.Env("STREAM", topic), + core.Env("SUBJECTS", strings.Join(natsConsumerSubjects(stack, broker, service), " ")), }, }, jobs.WithImagePullSecrets(natsBoxImage.PullSecrets), ) condition := v1beta1.NewCondition(ConditionTypeNatsServiceConsumerCreated, consumer.Generation). - SetReason(service) + SetReason(natsServiceConditionReason(service)) defer func() { consumer.Status.Conditions.AppendOrReplace(*condition, v1beta1.AndConditions( v1beta1.ConditionTypeMatch(ConditionTypeNatsServiceConsumerCreated), - v1beta1.ConditionReasonMatch(service), + v1beta1.ConditionReasonMatch(natsServiceConditionReason(service)), )) }() @@ -225,14 +234,18 @@ func createStackNatsConsumer(ctx core.Context, stack *v1beta1.Stack, consumer *v for f in $SUBJECTS; do filters="$filters --filter $f" done - nats --server $NATS_URI consumer add $STREAM $NAME \ - --deliver-group $DELIVER \ - --deliver all \ - --max-pending 1024 \ - --ack explicit \ - --target $STREAM-$NAME \ - --replay instant \ - --defaults $filters + if nats --server "$NATS_URI" consumer info "$STREAM" "$NAME" --no-select >/dev/null 2>&1; then + nats --server "$NATS_URI" consumer edit "$STREAM" "$NAME" --force $filters + else + nats --server "$NATS_URI" consumer add "$STREAM" "$NAME" \ + --deliver-group "$DELIVER" \ + --deliver all \ + --max-pending 1024 \ + --ack explicit \ + --target "$STREAM-$NAME" \ + --replay instant \ + --defaults $filters + fi ` natsBoxImage, err := registries.GetNatsBoxImage(ctx, stack, "0.19.2") @@ -245,7 +258,7 @@ func createStackNatsConsumer(ctx core.Context, stack *v1beta1.Stack, consumer *v consumerName += "_" + consumer.Spec.Name } - err = jobs.Handle(ctx, consumer, "create-consumer", corev1.Container{ + err = jobs.Handle(ctx, consumer, "cc-"+natsConsumerJobRevision, corev1.Container{ Image: natsBoxImage.GetFullImageName(), Name: "create-consumer", Args: core.ShellScript(script), @@ -254,11 +267,7 @@ func createStackNatsConsumer(ctx core.Context, stack *v1beta1.Stack, consumer *v core.Env("STREAM", stack.Name), core.Env("NAME", consumerName), core.Env("DELIVER", consumer.Spec.QueriedBy), - core.Env("SUBJECTS", strings.Join( - collectionutils.Map(consumer.Spec.Services, func(from string) string { - return fmt.Sprintf("%s.%s", stack.Name, from) - }), " ", - )), + core.Env("SUBJECTS", strings.Join(natsConsumerSubjects(stack, broker, consumer.Spec.Services...), " ")), }, }, jobs.WithImagePullSecrets(natsBoxImage.PullSecrets), @@ -267,6 +276,7 @@ func createStackNatsConsumer(ctx core.Context, stack *v1beta1.Stack, consumer *v consumer.GetConditions().AppendOrReplace(v1beta1.Condition{ Type: ConditionTypeNatsStackConsumerCreated, Status: metav1.ConditionFalse, + Reason: natsConsumerConfigurationRevision, ObservedGeneration: consumer.Generation, LastTransitionTime: metav1.Now(), Message: fmt.Sprintf("Error creating consumer on nats: %s", err), @@ -276,6 +286,7 @@ func createStackNatsConsumer(ctx core.Context, stack *v1beta1.Stack, consumer *v consumer.GetConditions().AppendOrReplace(v1beta1.Condition{ Type: ConditionTypeNatsStackConsumerCreated, Status: metav1.ConditionTrue, + Reason: natsConsumerConfigurationRevision, ObservedGeneration: consumer.Generation, LastTransitionTime: metav1.Now(), Message: "Nats consumer created", @@ -284,3 +295,16 @@ func createStackNatsConsumer(ctx core.Context, stack *v1beta1.Stack, consumer *v return nil } + +func natsServiceConditionReason(service string) string { + return service + ":" + natsConsumerConfigurationRevision +} + +func natsConsumerSubjects(stack *v1beta1.Stack, broker *v1beta1.Broker, services ...string) []string { + subjects := make([]string, 0, len(services)*2) + for _, service := range services { + topic := brokers.GetPublisherTopic(stack, broker, service) + subjects = append(subjects, brokers.NatsSubjects(topic)...) + } + return subjects +} diff --git a/internal/resources/brokerconsumers/controller_test.go b/internal/resources/brokerconsumers/controller_test.go new file mode 100644 index 00000000..9e5684a7 --- /dev/null +++ b/internal/resources/brokerconsumers/controller_test.go @@ -0,0 +1,52 @@ +package brokerconsumers + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/formancehq/operator/v3/api/formance.com/v1beta1" +) + +func TestNatsConsumerSubjects(t *testing.T) { + t.Parallel() + + stack := &v1beta1.Stack{} + stack.Name = "stack0" + + tests := []struct { + name string + mode v1beta1.Mode + expected []string + }{ + { + name: "one stream by stack", + mode: v1beta1.ModeOneStreamByStack, + expected: []string{ + "stack0.ledger", "stack0.ledger.>", + "stack0.payments", "stack0.payments.>", + }, + }, + { + name: "one stream by service", + mode: v1beta1.ModeOneStreamByService, + expected: []string{ + "stack0-ledger", "stack0-ledger.>", + "stack0-payments", "stack0-payments.>", + }, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + broker := &v1beta1.Broker{Status: v1beta1.BrokerStatus{Mode: test.mode}} + require.Equal(t, test.expected, natsConsumerSubjects(stack, broker, "ledger", "payments")) + }) + } +} + +func TestNatsServiceConditionReasonIncludesRevision(t *testing.T) { + t.Parallel() + + require.Equal(t, "ledger:NestedSubjectsV1", natsServiceConditionReason("ledger")) +} diff --git a/internal/resources/brokers/reconcile.go b/internal/resources/brokers/reconcile.go index 007bd105..227230d4 100644 --- a/internal/resources/brokers/reconcile.go +++ b/internal/resources/brokers/reconcile.go @@ -19,6 +19,8 @@ import ( "github.com/formancehq/operator/v3/internal/resources/settings" ) +const natsNestedSubjectsRevision = "ns1" + func Reconcile(ctx core.Context, stack *v1beta1.Stack, broker *v1beta1.Broker) error { brokerURI, err := settings.RequireURL(ctx, stack.Name, "broker", "dsn") @@ -173,17 +175,16 @@ func deleteBroker(ctx core.Context, broker *v1beta1.Broker) error { } func createOneStreamByStack(ctx core.Context, stack *v1beta1.Stack, broker *v1beta1.Broker, uri *v1beta1.URI) error { - - if broker.Status.Ready { - return nil - } - const script = ` - if ! nats --server "$NATS_URI" stream info "$STREAM" --no-select >/dev/null 2>&1; then + if nats --server "$NATS_URI" stream info "$STREAM" --no-select >/dev/null 2>&1; then + nats --server "$NATS_URI" stream edit "$STREAM" \ + --subjects "$STREAM.>" \ + --force + else nats stream add \ --server "$NATS_URI" \ --retention interest \ - --subjects "$STREAM.*" \ + --subjects "$STREAM.>" \ --defaults \ --replicas "$REPLICAS" \ --no-allow-direct \ @@ -195,7 +196,7 @@ func createOneStreamByStack(ctx core.Context, stack *v1beta1.Stack, broker *v1be return err } - return jobs.Handle(ctx, broker, "create-stream", corev1.Container{ + return jobs.Handle(ctx, broker, "cs-"+natsNestedSubjectsRevision, corev1.Container{ Image: natsBoxImage.GetFullImageName(), Name: "create-topic", Args: core.ShellScript(script), @@ -224,10 +225,10 @@ func createOneStreamByTopic(ctx core.Context, stack *v1beta1.Stack, broker *v1be for _, item := range l.Items { item := item + if err := createNatsTopic(ctx, stack, broker, &item, brokerURI); err != nil { + return err + } if !collectionutils.Contains(broker.Status.Streams, item.Spec.Service) { - if err := createNatsTopic(ctx, stack, broker, &item, brokerURI); err != nil { - return err - } broker.Status.Streams = append(broker.Status.Streams, item.Spec.Service) } } @@ -239,16 +240,21 @@ func createOneStreamByTopic(ctx core.Context, stack *v1beta1.Stack, broker *v1be func createNatsTopic(ctx core.Context, stack *v1beta1.Stack, broker *v1beta1.Broker, topic *v1beta1.BrokerTopic, brokerURI *v1beta1.URI) error { const script = ` - index=$(nats --server $NATS_URI stream ls -j | jq "index(\"$SUBJECT\")") - if [ "$index" = "null" ]; then + if nats --server "$NATS_URI" stream info "$STREAM" --no-select >/dev/null 2>&1; then + nats --server "$NATS_URI" stream edit "$STREAM" \ + --subjects "$SUBJECT" \ + --subjects "$SUBJECT.>" \ + --force + else nats stream add \ - --server $NATS_URI \ + --server "$NATS_URI" \ --retention interest \ - --subjects $SUBJECT \ + --subjects "$SUBJECT" \ + --subjects "$SUBJECT.>" \ --defaults \ - --replicas $REPLICAS \ + --replicas "$REPLICAS" \ --no-allow-direct \ - $STREAM + "$STREAM" fi` natsBoxImage, err := registries.GetNatsBoxImage(ctx, stack, "0.19.2") @@ -256,7 +262,7 @@ func createNatsTopic(ctx core.Context, stack *v1beta1.Stack, broker *v1beta1.Bro return err } - return jobs.Handle(ctx, broker, "create-topic-"+topic.Spec.Service, corev1.Container{ + return jobs.Handle(ctx, broker, "ct-"+natsNestedSubjectsRevision+"-"+topic.Spec.Service, corev1.Container{ Image: natsBoxImage.GetFullImageName(), Name: "create-topic", Args: core.ShellScript(script), diff --git a/internal/resources/brokers/utils.go b/internal/resources/brokers/utils.go index eded4d49..3bc9f472 100644 --- a/internal/resources/brokers/utils.go +++ b/internal/resources/brokers/utils.go @@ -69,16 +69,14 @@ func GetBrokerEnvVars(ctx core.Context, brokerURI *v1beta1.URI, stackName, servi } func GetPublisherEnvVars(stack *v1beta1.Stack, broker *v1beta1.Broker, service string) []v1.EnvVar { + ret := []v1.EnvVar{ + core.Env("PUBLISHER_TOPIC_MAPPING", "*:"+GetPublisherTopic(stack, broker, service)), + } + switch broker.Status.Mode { case v1beta1.ModeOneStreamByService: - return []v1.EnvVar{ - core.Env("PUBLISHER_TOPIC_MAPPING", "*:"+core.GetObjectName(stack.Name, service)), - } + return ret case v1beta1.ModeOneStreamByStack: - ret := []v1.EnvVar{ - core.Env("PUBLISHER_TOPIC_MAPPING", fmt.Sprintf("*:%s.%s", stack.Name, service)), - } - if broker.Status.URI.Scheme == "nats" { ret = append(ret, core.Env("PUBLISHER_NATS_AUTO_PROVISION", "false")) } @@ -88,6 +86,21 @@ func GetPublisherEnvVars(stack *v1beta1.Stack, broker *v1beta1.Broker, service s } } +func GetPublisherTopic(stack *v1beta1.Stack, broker *v1beta1.Broker, service string) string { + switch broker.Status.Mode { + case v1beta1.ModeOneStreamByService: + return core.GetObjectName(stack.Name, service) + case v1beta1.ModeOneStreamByStack: + return fmt.Sprintf("%s.%s", stack.Name, service) + default: + panic(fmt.Sprintf("mode '%s' not handled", broker.Status.Mode)) + } +} + +func NatsSubjects(topic string) []string { + return []string{topic, topic + ".>"} +} + func GetTopicsEnvVars(ctx core.Context, stack *v1beta1.Stack, key string, services ...string) ([]v1.EnvVar, error) { broker := &v1beta1.Broker{} diff --git a/internal/resources/brokers/utils_test.go b/internal/resources/brokers/utils_test.go index e1228bb3..26d77338 100644 --- a/internal/resources/brokers/utils_test.go +++ b/internal/resources/brokers/utils_test.go @@ -48,3 +48,32 @@ func TestGetBrokerEnvVarsCircuitBreakerIsOptIn(t *testing.T) { }) } } + +func TestGetPublisherTopic(t *testing.T) { + t.Parallel() + + stack := &v1beta1.Stack{} + stack.Name = "stack0" + + tests := []struct { + name string + mode v1beta1.Mode + expected string + }{ + {name: "one stream by service", mode: v1beta1.ModeOneStreamByService, expected: "stack0-ledger"}, + {name: "one stream by stack", mode: v1beta1.ModeOneStreamByStack, expected: "stack0.ledger"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + broker := &v1beta1.Broker{Status: v1beta1.BrokerStatus{Mode: test.mode}} + require.Equal(t, test.expected, GetPublisherTopic(stack, broker, "ledger")) + }) + } +} + +func TestNatsSubjects(t *testing.T) { + t.Parallel() + + require.Equal(t, []string{"stack0.ledger", "stack0.ledger.>"}, NatsSubjects("stack0.ledger")) +} diff --git a/internal/resources/ledgers/init.go b/internal/resources/ledgers/init.go index 8432d5c7..66c8c68e 100644 --- a/internal/resources/ledgers/init.go +++ b/internal/resources/ledgers/init.go @@ -27,6 +27,7 @@ import ( "github.com/formancehq/operator/v3/api/formance.com/v1beta1" . "github.com/formancehq/operator/v3/internal/core" + "github.com/formancehq/operator/v3/internal/resources/brokers" "github.com/formancehq/operator/v3/internal/resources/brokertopics" "github.com/formancehq/operator/v3/internal/resources/databases" "github.com/formancehq/operator/v3/internal/resources/gatewayhttpapis" @@ -164,6 +165,7 @@ func init() { WithWatchSettings[*v1beta1.Ledger](), WithWatchDependency[*v1beta1.Ledger](&v1beta1.Auth{}), WithWatchDependency[*v1beta1.Ledger](&v1beta1.Search{}), + brokers.Watch[*v1beta1.Ledger](), brokertopics.Watch[*v1beta1.Ledger]("ledger"), databases.Watch[*v1beta1.Ledger](), ), diff --git a/internal/resources/ledgers/v3.go b/internal/resources/ledgers/v3.go index 4ba227d9..e4f7927d 100644 --- a/internal/resources/ledgers/v3.go +++ b/internal/resources/ledgers/v3.go @@ -385,6 +385,10 @@ func createOrUpdateV3Cluster(ctx core.Context, stack *v1beta1.Stack, ledger *v1b if err != nil { return nil, nil, err } + eventSinks, err := ledgerV3EventSinks(ctx, stack, baseSpec.Sinks, preview) + if err != nil { + return nil, nil, err + } desiredSpec, err := composeLedgerV3ClusterSpec(baseSpec, ledgerV3SpecOverrides{ ImageRepository: imageRepository(image), ImageTag: image.Version, @@ -401,6 +405,7 @@ func createOrUpdateV3Cluster(ctx core.Context, stack *v1beta1.Stack, ledger *v1b Auth: authConfiguration, ServiceAccountName: serviceAccountName, TopologySpreadConstraints: topologySpreadConstraints, + Sinks: eventSinks, }) if err != nil { return nil, nil, fmt.Errorf("composing Ledger v3 Cluster spec: %w", err) diff --git a/internal/resources/ledgers/v3_sinks.go b/internal/resources/ledgers/v3_sinks.go new file mode 100644 index 00000000..ee13f043 --- /dev/null +++ b/internal/resources/ledgers/v3_sinks.go @@ -0,0 +1,82 @@ +package ledgers + +import ( + "fmt" + "slices" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/types" + + ledgerv1alpha1 "github.com/formancehq/ledger/misc/operator/api/v1alpha1" + + "github.com/formancehq/operator/v3/api/formance.com/v1beta1" + "github.com/formancehq/operator/v3/internal/core" + "github.com/formancehq/operator/v3/internal/resources/brokers" +) + +const ledgerV3BrokerSinkName = "formance-broker" + +// ledgerV3EventSinks merges the sinks declared in LedgerConfiguration with the +// Formance broker sink. Returning a non-nil spec deliberately makes the +// configuration authoritative, so disabling the ledger BrokerTopic removes +// the previously managed sink as well. +func ledgerV3EventSinks( + ctx core.Context, + stack *v1beta1.Stack, + configured *ledgerv1alpha1.EventSinksSpec, + preview bool, +) (*ledgerv1alpha1.EventSinksSpec, error) { + // Preview clusters must not duplicate events emitted by the active Ledger. + if preview { + return mergeLedgerV3EventSinks(configured, nil), nil + } + + topic := &v1beta1.BrokerTopic{} + if err := ctx.GetClient().Get(ctx, types.NamespacedName{ + Name: core.GetObjectName(stack.Name, "ledger"), + }, topic); err != nil { + if apierrors.IsNotFound(err) { + return mergeLedgerV3EventSinks(configured, nil), nil + } + return nil, fmt.Errorf("getting Ledger BrokerTopic: %w", err) + } + if !topic.Status.Ready { + return nil, core.NewPendingError().WithMessage("Ledger BrokerTopic is not ready") + } + + broker := &v1beta1.Broker{} + if err := ctx.GetClient().Get(ctx, types.NamespacedName{Name: stack.Name}, broker); err != nil { + return nil, fmt.Errorf("getting Broker for Ledger v3 sink: %w", err) + } + if !broker.Status.Ready || broker.Status.URI == nil { + return nil, core.NewPendingError().WithMessage("broker is not ready for Ledger v3 sink") + } + if broker.Status.URI.Scheme != "nats" { + return mergeLedgerV3EventSinks(configured, nil), nil + } + + sink := &ledgerv1alpha1.NATSEventSinkSpec{ + Name: ledgerV3BrokerSinkName, + URL: fmt.Sprintf("nats://%s", broker.Status.URI.Host), + Topic: brokers.GetPublisherTopic(stack, broker, "ledger"), + Format: "json", + } + return mergeLedgerV3EventSinks(configured, sink), nil +} + +func mergeLedgerV3EventSinks( + configured *ledgerv1alpha1.EventSinksSpec, + brokerSink *ledgerv1alpha1.NATSEventSinkSpec, +) *ledgerv1alpha1.EventSinksSpec { + sinks := &ledgerv1alpha1.EventSinksSpec{} + if configured != nil { + sinks = configured.DeepCopy() + } + sinks.NATS = slices.DeleteFunc(sinks.NATS, func(sink ledgerv1alpha1.NATSEventSinkSpec) bool { + return sink.Name == ledgerV3BrokerSinkName + }) + if brokerSink != nil { + sinks.NATS = append(sinks.NATS, *brokerSink) + } + return sinks +} diff --git a/internal/resources/ledgers/v3_sinks_test.go b/internal/resources/ledgers/v3_sinks_test.go new file mode 100644 index 00000000..d7a9feb5 --- /dev/null +++ b/internal/resources/ledgers/v3_sinks_test.go @@ -0,0 +1,42 @@ +package ledgers + +import ( + "testing" + + "github.com/stretchr/testify/require" + + ledgerv1alpha1 "github.com/formancehq/ledger/misc/operator/api/v1alpha1" +) + +func TestMergeLedgerV3EventSinks(t *testing.T) { + t.Parallel() + + configured := &ledgerv1alpha1.EventSinksSpec{NATS: []ledgerv1alpha1.NATSEventSinkSpec{ + {Name: "audit", URL: "nats://audit:4222", Topic: "audit"}, + {Name: ledgerV3BrokerSinkName, URL: "nats://stale:4222", Topic: "stale"}, + }} + brokerSink := &ledgerv1alpha1.NATSEventSinkSpec{ + Name: ledgerV3BrokerSinkName, URL: "nats://broker:4222", Topic: "stack0.ledger", Format: "json", + } + + actual := mergeLedgerV3EventSinks(configured, brokerSink) + require.Equal(t, []ledgerv1alpha1.NATSEventSinkSpec{ + {Name: "audit", URL: "nats://audit:4222", Topic: "audit"}, + *brokerSink, + }, actual.NATS) + require.Equal(t, "stale", configured.NATS[1].Topic) +} + +func TestMergeLedgerV3EventSinksRemovesManagedSinkWhenDisabled(t *testing.T) { + t.Parallel() + + configured := &ledgerv1alpha1.EventSinksSpec{NATS: []ledgerv1alpha1.NATSEventSinkSpec{ + {Name: "audit", URL: "nats://audit:4222", Topic: "audit"}, + {Name: ledgerV3BrokerSinkName, URL: "nats://stale:4222", Topic: "stale"}, + }} + + actual := mergeLedgerV3EventSinks(configured, nil) + require.Equal(t, []ledgerv1alpha1.NATSEventSinkSpec{ + {Name: "audit", URL: "nats://audit:4222", Topic: "audit"}, + }, actual.NATS) +} diff --git a/internal/resources/ledgers/v3_spec.go b/internal/resources/ledgers/v3_spec.go index 95912c8e..5b886b0b 100644 --- a/internal/resources/ledgers/v3_spec.go +++ b/internal/resources/ledgers/v3_spec.go @@ -32,6 +32,7 @@ type ledgerV3SpecOverrides struct { Auth *auths.ProtectedAuthConfiguration ServiceAccountName string TopologySpreadConstraints *bool + Sinks *ledgerv1alpha1.EventSinksSpec } // composeLedgerV3ClusterSpec applies the values owned by the Formance Operator @@ -108,6 +109,9 @@ func composeLedgerV3ClusterSpec(base *ledgerv1alpha1.ClusterSpec, overrides ledg spec.TopologySpreadConstraints = nil } } + if overrides.Sinks != nil { + spec.Sinks = overrides.Sinks.DeepCopy() + } return spec, nil } diff --git a/internal/resources/ledgers/v3_spec_test.go b/internal/resources/ledgers/v3_spec_test.go index 02cc4057..c2e8c4b2 100644 --- a/internal/resources/ledgers/v3_spec_test.go +++ b/internal/resources/ledgers/v3_spec_test.go @@ -214,3 +214,22 @@ func TestComposeLedgerV3ClusterSpecRejectsOversizedAuthRetries(t *testing.T) { }) require.ErrorContains(t, err, "must fit in int32") } + +func TestComposeLedgerV3ClusterSpecAppliesSinkOverrideWithoutMutation(t *testing.T) { + t.Parallel() + + base := &ledgerv1alpha1.ClusterSpec{Sinks: &ledgerv1alpha1.EventSinksSpec{ + NATS: []ledgerv1alpha1.NATSEventSinkSpec{{Name: "configured", URL: "nats://configured:4222", Topic: "configured"}}, + }} + override := &ledgerv1alpha1.EventSinksSpec{ + NATS: []ledgerv1alpha1.NATSEventSinkSpec{{Name: ledgerV3BrokerSinkName, URL: "nats://broker:4222", Topic: "stack0.ledger"}}, + } + + actual, err := composeLedgerV3ClusterSpec(base, ledgerV3SpecOverrides{Sinks: override}) + require.NoError(t, err) + require.Equal(t, override, actual.Sinks) + + actual.Sinks.NATS[0].Topic = "mutated" + require.Equal(t, "stack0.ledger", override.NATS[0].Topic) + require.Equal(t, "configured", base.Sinks.NATS[0].Topic) +} diff --git a/tools/kubectl-stacks/go.mod b/tools/kubectl-stacks/go.mod index 135a99c6..0d938551 100644 --- a/tools/kubectl-stacks/go.mod +++ b/tools/kubectl-stacks/go.mod @@ -25,7 +25,7 @@ require ( github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/emicklei/go-restful/v3 v3.13.0 // indirect github.com/evanphx/json-patch/v5 v5.9.11 // indirect - github.com/formancehq/ledger/misc/operator v0.0.0-20260715094310-76862ea0b681 // indirect + github.com/formancehq/ledger/misc/operator v0.0.0-20260831193659-af94b64e0f35 // indirect github.com/fxamacker/cbor/v2 v2.9.0 // indirect github.com/go-errors/errors v1.4.2 // indirect github.com/go-logr/logr v1.4.3 // indirect diff --git a/tools/kubectl-stacks/go.sum b/tools/kubectl-stacks/go.sum index 154ddf91..07dafc18 100644 --- a/tools/kubectl-stacks/go.sum +++ b/tools/kubectl-stacks/go.sum @@ -42,8 +42,8 @@ github.com/evanphx/json-patch/v5 v5.9.11 h1:/8HVnzMq13/3x9TPvjG08wUGqBTmZBsCWzjT github.com/evanphx/json-patch/v5 v5.9.11/go.mod h1:3j+LviiESTElxA4p3EMKAB9HXj3/XEtnUf6OZxqIQTM= github.com/formancehq/go-libs/v5 v5.2.0 h1:TpS47F8X5g5cHhnecfD20TrcdBqUVGy/ezZv0oFaQjc= github.com/formancehq/go-libs/v5 v5.2.0/go.mod h1:ms6tCGw1yqB4qtEbAuqPOQegWo4rU48vDobNkK7Ak6U= -github.com/formancehq/ledger/misc/operator v0.0.0-20260715094310-76862ea0b681 h1:PzhfbpKZqMJd4opnKmEtFq0Hx/6Pxgq9GT7XzWcr+Ss= -github.com/formancehq/ledger/misc/operator v0.0.0-20260715094310-76862ea0b681/go.mod h1:tZa1TFBcXJxc1R0NqnsN6gOGoskd82a7wjmGqCBhOBU= +github.com/formancehq/ledger/misc/operator v0.0.0-20260831193659-af94b64e0f35 h1:tkiLCD/yg9wO2pS9o5yuhKV0JDHMVMCVx1I+sum3YY0= +github.com/formancehq/ledger/misc/operator v0.0.0-20260831193659-af94b64e0f35/go.mod h1:tZa1TFBcXJxc1R0NqnsN6gOGoskd82a7wjmGqCBhOBU= github.com/fxamacker/cbor/v2 v2.9.0 h1:NpKPmjDBgUfBms6tr6JZkTHtfFGcMKsw3eGcmD/sapM= github.com/fxamacker/cbor/v2 v2.9.0/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ= github.com/go-errors/errors v1.4.2 h1:J6MZopCL4uSllY1OfXM374weqZFFItUbrImctkmUxIA= From cf1629df199b128bfc0acf50b7168f1a2cdc8793 Mon Sep 17 00:00:00 2001 From: Maxence Maireaux Date: Mon, 31 Aug 2026 22:07:00 +0200 Subject: [PATCH 2/4] fix(operator): preserve Ledger v3 compatibility --- .../formance.com_ledgerconfigurations.yaml | 65 +++++++++++++++++++ go.mod | 2 +- go.sum | 4 +- ...ion_ledgerconfigurations.formance.com.yaml | 65 +++++++++++++++++++ internal/resources/ledgers/v3_sinks.go | 2 +- internal/resources/ledgers/v3_sinks_test.go | 17 +++++ tools/kubectl-stacks/go.mod | 2 +- tools/kubectl-stacks/go.sum | 4 +- 8 files changed, 154 insertions(+), 7 deletions(-) diff --git a/config/crd/bases/formance.com_ledgerconfigurations.yaml b/config/crd/bases/formance.com_ledgerconfigurations.yaml index 76ef2015..69aa7f0c 100644 --- a/config/crd/bases/formance.com_ledgerconfigurations.yaml +++ b/config/crd/bases/formance.com_ledgerconfigurations.yaml @@ -1234,6 +1234,71 @@ spec: LogLevel takes precedence when both are set; prefer LogLevel for new manifests since it also unlocks the trace level. type: boolean + dnsEndpoint: + description: |- + DNSEndpoint is the legacy single ExternalDNS DNSEndpoint configuration. + + Deprecated: use DNSEndpoints for new configurations. When DNSEndpoints is + non-empty, it takes precedence over this field. + properties: + annotations: + additionalProperties: + type: string + description: Annotations to add to the DNSEndpoint resource. + type: object + enabled: + description: Enabled enables the DNSEndpoint resource. + type: boolean + endpoints: + description: Endpoints is the list of DNS endpoint entries. + items: + description: DNSEndpointEntry defines a single DNS endpoint. + properties: + dnsName: + description: DNSName is the hostname for the DNS record. + type: string + providerSpecific: + description: ProviderSpecific holds provider-specific + properties. + items: + description: ProviderSpecificProperty defines a provider-specific + key-value pair. + properties: + name: + description: Name is the property name. + type: string + value: + description: Value is the property value. + type: string + required: + - name + - value + type: object + type: array + recordTTL: + description: RecordTTL is the TTL in seconds for the + DNS record. + format: int64 + type: integer + recordType: + description: RecordType is the DNS record type (e.g., + CNAME, A). Defaults to CNAME. + type: string + targets: + description: Targets is the list of target hostnames + or IPs. + items: + type: string + type: array + required: + - dnsName + - targets + type: object + type: array + type: object + x-kubernetes-validations: + - message: endpoints are required when the DNSEndpoint is enabled + rule: '!self.enabled || size(self.endpoints) > 0' dnsEndpoints: description: |- DNSEndpoints configures one or more ExternalDNS DNSEndpoint resources. diff --git a/go.mod b/go.mod index b9d3c139..c2303dd2 100644 --- a/go.mod +++ b/go.mod @@ -48,7 +48,7 @@ require ( github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/emicklei/go-restful/v3 v3.13.0 // indirect github.com/evanphx/json-patch/v5 v5.9.11 // indirect - github.com/formancehq/ledger/misc/operator v0.0.0-20260831193659-af94b64e0f35 + github.com/formancehq/ledger/misc/operator v0.0.0-20260831200208-704e27419161 github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-logr/zapr v1.3.0 // indirect diff --git a/go.sum b/go.sum index 53541468..c847fc51 100644 --- a/go.sum +++ b/go.sum @@ -18,8 +18,8 @@ github.com/evanphx/json-patch/v5 v5.9.11 h1:/8HVnzMq13/3x9TPvjG08wUGqBTmZBsCWzjT github.com/evanphx/json-patch/v5 v5.9.11/go.mod h1:3j+LviiESTElxA4p3EMKAB9HXj3/XEtnUf6OZxqIQTM= github.com/formancehq/go-libs/v5 v5.2.0 h1:TpS47F8X5g5cHhnecfD20TrcdBqUVGy/ezZv0oFaQjc= github.com/formancehq/go-libs/v5 v5.2.0/go.mod h1:ms6tCGw1yqB4qtEbAuqPOQegWo4rU48vDobNkK7Ak6U= -github.com/formancehq/ledger/misc/operator v0.0.0-20260831193659-af94b64e0f35 h1:tkiLCD/yg9wO2pS9o5yuhKV0JDHMVMCVx1I+sum3YY0= -github.com/formancehq/ledger/misc/operator v0.0.0-20260831193659-af94b64e0f35/go.mod h1:tZa1TFBcXJxc1R0NqnsN6gOGoskd82a7wjmGqCBhOBU= +github.com/formancehq/ledger/misc/operator v0.0.0-20260831200208-704e27419161 h1:V+FKQETM8FqlFIExOosQo63u0BlLog1mszootnbxRW8= +github.com/formancehq/ledger/misc/operator v0.0.0-20260831200208-704e27419161/go.mod h1:tZa1TFBcXJxc1R0NqnsN6gOGoskd82a7wjmGqCBhOBU= github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k= github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0= github.com/fxamacker/cbor/v2 v2.9.0 h1:NpKPmjDBgUfBms6tr6JZkTHtfFGcMKsw3eGcmD/sapM= diff --git a/helm/crds/templates/crds/apiextensions.k8s.io_v1_customresourcedefinition_ledgerconfigurations.formance.com.yaml b/helm/crds/templates/crds/apiextensions.k8s.io_v1_customresourcedefinition_ledgerconfigurations.formance.com.yaml index 8a264a90..6ff4ec53 100644 --- a/helm/crds/templates/crds/apiextensions.k8s.io_v1_customresourcedefinition_ledgerconfigurations.formance.com.yaml +++ b/helm/crds/templates/crds/apiextensions.k8s.io_v1_customresourcedefinition_ledgerconfigurations.formance.com.yaml @@ -1237,6 +1237,71 @@ spec: LogLevel takes precedence when both are set; prefer LogLevel for new manifests since it also unlocks the trace level. type: boolean + dnsEndpoint: + description: |- + DNSEndpoint is the legacy single ExternalDNS DNSEndpoint configuration. + + Deprecated: use DNSEndpoints for new configurations. When DNSEndpoints is + non-empty, it takes precedence over this field. + properties: + annotations: + additionalProperties: + type: string + description: Annotations to add to the DNSEndpoint resource. + type: object + enabled: + description: Enabled enables the DNSEndpoint resource. + type: boolean + endpoints: + description: Endpoints is the list of DNS endpoint entries. + items: + description: DNSEndpointEntry defines a single DNS endpoint. + properties: + dnsName: + description: DNSName is the hostname for the DNS record. + type: string + providerSpecific: + description: ProviderSpecific holds provider-specific + properties. + items: + description: ProviderSpecificProperty defines a provider-specific + key-value pair. + properties: + name: + description: Name is the property name. + type: string + value: + description: Value is the property value. + type: string + required: + - name + - value + type: object + type: array + recordTTL: + description: RecordTTL is the TTL in seconds for the + DNS record. + format: int64 + type: integer + recordType: + description: RecordType is the DNS record type (e.g., + CNAME, A). Defaults to CNAME. + type: string + targets: + description: Targets is the list of target hostnames + or IPs. + items: + type: string + type: array + required: + - dnsName + - targets + type: object + type: array + type: object + x-kubernetes-validations: + - message: endpoints are required when the DNSEndpoint is enabled + rule: '!self.enabled || size(self.endpoints) > 0' dnsEndpoints: description: |- DNSEndpoints configures one or more ExternalDNS DNSEndpoint resources. diff --git a/internal/resources/ledgers/v3_sinks.go b/internal/resources/ledgers/v3_sinks.go index ee13f043..63dbd3ed 100644 --- a/internal/resources/ledgers/v3_sinks.go +++ b/internal/resources/ledgers/v3_sinks.go @@ -28,7 +28,7 @@ func ledgerV3EventSinks( ) (*ledgerv1alpha1.EventSinksSpec, error) { // Preview clusters must not duplicate events emitted by the active Ledger. if preview { - return mergeLedgerV3EventSinks(configured, nil), nil + return &ledgerv1alpha1.EventSinksSpec{}, nil } topic := &v1beta1.BrokerTopic{} diff --git a/internal/resources/ledgers/v3_sinks_test.go b/internal/resources/ledgers/v3_sinks_test.go index d7a9feb5..cdfd1848 100644 --- a/internal/resources/ledgers/v3_sinks_test.go +++ b/internal/resources/ledgers/v3_sinks_test.go @@ -4,10 +4,27 @@ import ( "testing" "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ledgerv1alpha1 "github.com/formancehq/ledger/misc/operator/api/v1alpha1" + + "github.com/formancehq/operator/v3/api/formance.com/v1beta1" ) +func TestLedgerV3EventSinksDisablesAllPreviewSinks(t *testing.T) { + t.Parallel() + + configured := &ledgerv1alpha1.EventSinksSpec{NATS: []ledgerv1alpha1.NATSEventSinkSpec{{ + Name: "audit", URL: "nats://audit:4222", Topic: "audit", + }}} + stack := &v1beta1.Stack{ObjectMeta: metav1.ObjectMeta{Name: "stack0"}} + + actual, err := ledgerV3EventSinks(newExportsContext(t), stack, configured, true) + require.NoError(t, err) + require.Empty(t, actual.NATS) + require.Len(t, configured.NATS, 1) +} + func TestMergeLedgerV3EventSinks(t *testing.T) { t.Parallel() diff --git a/tools/kubectl-stacks/go.mod b/tools/kubectl-stacks/go.mod index 0d938551..bbab10d4 100644 --- a/tools/kubectl-stacks/go.mod +++ b/tools/kubectl-stacks/go.mod @@ -25,7 +25,7 @@ require ( github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/emicklei/go-restful/v3 v3.13.0 // indirect github.com/evanphx/json-patch/v5 v5.9.11 // indirect - github.com/formancehq/ledger/misc/operator v0.0.0-20260831193659-af94b64e0f35 // indirect + github.com/formancehq/ledger/misc/operator v0.0.0-20260831200208-704e27419161 // indirect github.com/fxamacker/cbor/v2 v2.9.0 // indirect github.com/go-errors/errors v1.4.2 // indirect github.com/go-logr/logr v1.4.3 // indirect diff --git a/tools/kubectl-stacks/go.sum b/tools/kubectl-stacks/go.sum index 07dafc18..0af9f1ef 100644 --- a/tools/kubectl-stacks/go.sum +++ b/tools/kubectl-stacks/go.sum @@ -42,8 +42,8 @@ github.com/evanphx/json-patch/v5 v5.9.11 h1:/8HVnzMq13/3x9TPvjG08wUGqBTmZBsCWzjT github.com/evanphx/json-patch/v5 v5.9.11/go.mod h1:3j+LviiESTElxA4p3EMKAB9HXj3/XEtnUf6OZxqIQTM= github.com/formancehq/go-libs/v5 v5.2.0 h1:TpS47F8X5g5cHhnecfD20TrcdBqUVGy/ezZv0oFaQjc= github.com/formancehq/go-libs/v5 v5.2.0/go.mod h1:ms6tCGw1yqB4qtEbAuqPOQegWo4rU48vDobNkK7Ak6U= -github.com/formancehq/ledger/misc/operator v0.0.0-20260831193659-af94b64e0f35 h1:tkiLCD/yg9wO2pS9o5yuhKV0JDHMVMCVx1I+sum3YY0= -github.com/formancehq/ledger/misc/operator v0.0.0-20260831193659-af94b64e0f35/go.mod h1:tZa1TFBcXJxc1R0NqnsN6gOGoskd82a7wjmGqCBhOBU= +github.com/formancehq/ledger/misc/operator v0.0.0-20260831200208-704e27419161 h1:V+FKQETM8FqlFIExOosQo63u0BlLog1mszootnbxRW8= +github.com/formancehq/ledger/misc/operator v0.0.0-20260831200208-704e27419161/go.mod h1:tZa1TFBcXJxc1R0NqnsN6gOGoskd82a7wjmGqCBhOBU= github.com/fxamacker/cbor/v2 v2.9.0 h1:NpKPmjDBgUfBms6tr6JZkTHtfFGcMKsw3eGcmD/sapM= github.com/fxamacker/cbor/v2 v2.9.0/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ= github.com/go-errors/errors v1.4.2 h1:J6MZopCL4uSllY1OfXM374weqZFFItUbrImctkmUxIA= From ce3a703204879dd9869a7a6abc340d915cb1c159 Mon Sep 17 00:00:00 2001 From: Maxence Maireaux Date: Mon, 31 Aug 2026 22:19:13 +0200 Subject: [PATCH 3/4] fix(broker): persist NATS subject migration --- api/formance.com/v1beta1/broker_types.go | 5 ++ config/crd/bases/formance.com_brokers.yaml | 6 +++ .../02-Custom Resource Definitions.md | 1 + ...sourcedefinition_brokers.formance.com.yaml | 6 +++ internal/resources/brokers/reconcile.go | 26 +++++++-- internal/resources/brokers/reconcile_test.go | 54 +++++++++++++++++++ 6 files changed, 94 insertions(+), 4 deletions(-) create mode 100644 internal/resources/brokers/reconcile_test.go diff --git a/api/formance.com/v1beta1/broker_types.go b/api/formance.com/v1beta1/broker_types.go index 79ef4969..c81e15c3 100644 --- a/api/formance.com/v1beta1/broker_types.go +++ b/api/formance.com/v1beta1/broker_types.go @@ -46,6 +46,11 @@ type BrokerStatus struct { // Streams list streams created when Mode == ModeOneStreamByService //+optional Streams []string `json:"streams,omitempty"` + // StreamSubjectsRevision records the subject-layout migration applied to the + // NATS streams. It prevents completed migration Jobs from being recreated + // after their TTL expires. + //+optional + StreamSubjectsRevision string `json:"streamSubjectsRevision,omitempty"` } //+kubebuilder:object:root=true diff --git a/config/crd/bases/formance.com_brokers.yaml b/config/crd/bases/formance.com_brokers.yaml index 0a4fb99e..7c5aecc2 100644 --- a/config/crd/bases/formance.com_brokers.yaml +++ b/config/crd/bases/formance.com_brokers.yaml @@ -128,6 +128,12 @@ spec: description: Ready indicates if the resource is seen as completely reconciled type: boolean + streamSubjectsRevision: + description: |- + StreamSubjectsRevision records the subject-layout migration applied to the + NATS streams. It prevents completed migration Jobs from being recreated + after their TTL expires. + type: string streams: description: Streams list streams created when Mode == ModeOneStreamByService items: diff --git a/docs/09-Configuration reference/02-Custom Resource Definitions.md b/docs/09-Configuration reference/02-Custom Resource Definitions.md index 8e3f03ec..1c85bcab 100644 --- a/docs/09-Configuration reference/02-Custom Resource Definitions.md +++ b/docs/09-Configuration reference/02-Custom Resource Definitions.md @@ -2035,6 +2035,7 @@ Broker is the Schema for the brokers API | `uri` _string_ | | | Type: string
| | `mode` _[Mode](#mode)_ | Mode indicating the configuration of the nats streams
Two modes are defined :
* ModeOneStreamByService: In this case, each service will have a dedicated stream created
* ModeOneStreamByStack: In this case, a stream will be created for the stack and each service will use a specific subject inside this stream | | Enum: [OneStreamByService OneStreamByStack]
| | `streams` _string array_ | Streams list streams created when Mode == ModeOneStreamByService | | | +| `streamSubjectsRevision` _string_ | StreamSubjectsRevision records the subject-layout migration applied to the
NATS streams. It prevents completed migration Jobs from being recreated
after their TTL expires. | | | ###### Mode diff --git a/helm/crds/templates/crds/apiextensions.k8s.io_v1_customresourcedefinition_brokers.formance.com.yaml b/helm/crds/templates/crds/apiextensions.k8s.io_v1_customresourcedefinition_brokers.formance.com.yaml index edda5aa9..20254c29 100644 --- a/helm/crds/templates/crds/apiextensions.k8s.io_v1_customresourcedefinition_brokers.formance.com.yaml +++ b/helm/crds/templates/crds/apiextensions.k8s.io_v1_customresourcedefinition_brokers.formance.com.yaml @@ -131,6 +131,12 @@ spec: description: Ready indicates if the resource is seen as completely reconciled type: boolean + streamSubjectsRevision: + description: |- + StreamSubjectsRevision records the subject-layout migration applied to the + NATS streams. It prevents completed migration Jobs from being recreated + after their TTL expires. + type: string streams: description: Streams list streams created when Mode == ModeOneStreamByService items: diff --git a/internal/resources/brokers/reconcile.go b/internal/resources/brokers/reconcile.go index 227230d4..e4f55009 100644 --- a/internal/resources/brokers/reconcile.go +++ b/internal/resources/brokers/reconcile.go @@ -175,6 +175,10 @@ func deleteBroker(ctx core.Context, broker *v1beta1.Broker) error { } func createOneStreamByStack(ctx core.Context, stack *v1beta1.Stack, broker *v1beta1.Broker, uri *v1beta1.URI) error { + if broker.Status.StreamSubjectsRevision == natsNestedSubjectsRevision { + return nil + } + const script = ` if nats --server "$NATS_URI" stream info "$STREAM" --no-select >/dev/null 2>&1; then nats --server "$NATS_URI" stream edit "$STREAM" \ @@ -196,7 +200,7 @@ func createOneStreamByStack(ctx core.Context, stack *v1beta1.Stack, broker *v1be return err } - return jobs.Handle(ctx, broker, "cs-"+natsNestedSubjectsRevision, corev1.Container{ + if err := jobs.Handle(ctx, broker, "cs-"+natsNestedSubjectsRevision, corev1.Container{ Image: natsBoxImage.GetFullImageName(), Name: "create-topic", Args: core.ShellScript(script), @@ -212,7 +216,13 @@ func createOneStreamByStack(ctx core.Context, stack *v1beta1.Stack, broker *v1be }, }, jobs.WithImagePullSecrets(natsBoxImage.PullSecrets), - ) + ); err != nil { + return err + } + + broker.Status.StreamSubjectsRevision = natsNestedSubjectsRevision + + return nil } func createOneStreamByTopic(ctx core.Context, stack *v1beta1.Stack, broker *v1beta1.Broker, brokerURI *v1beta1.URI) error { @@ -225,8 +235,10 @@ func createOneStreamByTopic(ctx core.Context, stack *v1beta1.Stack, broker *v1be for _, item := range l.Items { item := item - if err := createNatsTopic(ctx, stack, broker, &item, brokerURI); err != nil { - return err + if brokerNeedsStreamSubjectsMigration(broker, item.Spec.Service) { + if err := createNatsTopic(ctx, stack, broker, &item, brokerURI); err != nil { + return err + } } if !collectionutils.Contains(broker.Status.Streams, item.Spec.Service) { broker.Status.Streams = append(broker.Status.Streams, item.Spec.Service) @@ -234,10 +246,16 @@ func createOneStreamByTopic(ctx core.Context, stack *v1beta1.Stack, broker *v1be } sort.Strings(broker.Status.Streams) + broker.Status.StreamSubjectsRevision = natsNestedSubjectsRevision return nil } +func brokerNeedsStreamSubjectsMigration(broker *v1beta1.Broker, service string) bool { + return broker.Status.StreamSubjectsRevision != natsNestedSubjectsRevision || + !collectionutils.Contains(broker.Status.Streams, service) +} + func createNatsTopic(ctx core.Context, stack *v1beta1.Stack, broker *v1beta1.Broker, topic *v1beta1.BrokerTopic, brokerURI *v1beta1.URI) error { const script = ` if nats --server "$NATS_URI" stream info "$STREAM" --no-select >/dev/null 2>&1; then diff --git a/internal/resources/brokers/reconcile_test.go b/internal/resources/brokers/reconcile_test.go new file mode 100644 index 00000000..a52cff70 --- /dev/null +++ b/internal/resources/brokers/reconcile_test.go @@ -0,0 +1,54 @@ +package brokers + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/formancehq/operator/v3/api/formance.com/v1beta1" +) + +func TestBrokerNeedsStreamSubjectsMigration(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + status v1beta1.BrokerStatus + service string + want bool + }{ + { + name: "migrates every existing stream when the revision is absent", + status: v1beta1.BrokerStatus{Streams: []string{"ledger"}}, + service: "ledger", + want: true, + }, + { + name: "does not recreate a completed migration job", + status: v1beta1.BrokerStatus{ + Streams: []string{"ledger"}, + StreamSubjectsRevision: natsNestedSubjectsRevision, + }, + service: "ledger", + want: false, + }, + { + name: "creates the stream for a service added after migration", + status: v1beta1.BrokerStatus{ + Streams: []string{"payments"}, + StreamSubjectsRevision: natsNestedSubjectsRevision, + }, + service: "ledger", + want: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + broker := &v1beta1.Broker{Status: tt.status} + require.Equal(t, tt.want, brokerNeedsStreamSubjectsMigration(broker, tt.service)) + }) + } +} From 966caf5b62f65aab905f271ec78e25733b692e62 Mon Sep 17 00:00:00 2001 From: Maxence Maireaux Date: Tue, 1 Sep 2026 13:07:49 +0200 Subject: [PATCH 4/4] fix(operator): gate Ledger sinks on CRD readiness --- .../settings.catalog.json | 12 +- internal/resources/ledgers/v3.go | 148 +++++++++- internal/resources/ledgers/v3_preview.go | 12 +- internal/resources/ledgers/v3_test.go | 263 ++++++++++++++++++ .../crds/ledger.formance.com_clusters.yaml | 23 ++ internal/tests/ledger_v3_controller_test.go | 8 + 6 files changed, 451 insertions(+), 15 deletions(-) diff --git a/docs/09-Configuration reference/settings.catalog.json b/docs/09-Configuration reference/settings.catalog.json index 350ff459..abe4d071 100644 --- a/docs/09-Configuration reference/settings.catalog.json +++ b/docs/09-Configuration reference/settings.catalog.json @@ -201,21 +201,21 @@ "key": "deployments.ledger.containers.ledger.resource-requirements.claims", "valueType": "string[]", "sources": [ - "internal/resources/ledgers/v3.go:365" + "internal/resources/ledgers/v3.go:438" ] }, { "key": "deployments.ledger.containers.ledger.resource-requirements.limits", "valueType": "map[string]string", "sources": [ - "internal/resources/ledgers/v3.go:365" + "internal/resources/ledgers/v3.go:438" ] }, { "key": "deployments.ledger.containers.ledger.resource-requirements.requests", "valueType": "map[string]string", "sources": [ - "internal/resources/ledgers/v3.go:365" + "internal/resources/ledgers/v3.go:438" ] }, { @@ -223,14 +223,14 @@ "valueType": "int32", "default": "3", "sources": [ - "internal/resources/ledgers/v3.go:351" + "internal/resources/ledgers/v3.go:424" ] }, { "key": "deployments.ledger.topology-spread-constraints", "valueType": "bool", "sources": [ - "internal/resources/ledgers/v3.go:383" + "internal/resources/ledgers/v3.go:456" ] }, { @@ -473,7 +473,7 @@ "key": "ledger.v3.preview-version", "valueType": "string", "sources": [ - "internal/resources/ledgers/v3_preview.go:44" + "internal/resources/ledgers/v3_preview.go:43" ] }, { diff --git a/internal/resources/ledgers/v3.go b/internal/resources/ledgers/v3.go index e4f7927d..2baa9144 100644 --- a/internal/resources/ledgers/v3.go +++ b/internal/resources/ledgers/v3.go @@ -4,6 +4,7 @@ import ( "fmt" "slices" "strings" + "time" "golang.org/x/mod/semver" appsv1 "k8s.io/api/apps/v1" @@ -35,13 +36,16 @@ import ( ) const ( - ledgerV3Threshold = "v3.0.0-alpha" - ledgerV3ClusterReadyCondition = "LedgerV3ClusterReady" - ledgerV3PreviewReadyCondition = "LedgerV3PreviewReady" - ledgerV3PreviewLabel = "formance.com/ledger-v3-preview" - ledgerV3GRPCPort = int32(8888) - ledgerV3HTTPPort = int32(9000) - ledgerV3PublicGRPCService = "ledger.BucketService" + ledgerV3Threshold = "v3.0.0-alpha" + ledgerV3ClusterReadyCondition = "LedgerV3ClusterReady" + ledgerV3PreviewReadyCondition = "LedgerV3PreviewReady" + ledgerV3SinksSyncedCondition = "SinksSynced" + ledgerV3PreviewLabel = "formance.com/ledger-v3-preview" + ledgerV3ClusterCRDName = "clusters.ledger.formance.com" + ledgerV3CRDDiscoveryRetryDelay = time.Minute + ledgerV3GRPCPort = int32(8888) + ledgerV3HTTPPort = int32(9000) + ledgerV3PublicGRPCService = "ledger.BucketService" ) var ( @@ -87,6 +91,9 @@ func withLedgerV3ClusterWatch() core.ReconcilerOption[*v1beta1.Ledger] { } ledgerV3ClusterAvailable = watchLedgerV3Resource(ctx, b, options, crds, ledgerV3ClusterGVK) + if ledgerV3ClusterAvailable && !ledgerV3ClusterSupportsSinks(crds) { + log.FromContext(ctx).Info("Ledger v3 Cluster sink contract is not available") + } issuerAvailable := watchLedgerV3Resource(ctx, b, options, crds, ledgerV3IssuerGVK) certificateAvailable := watchLedgerV3Resource(ctx, b, options, crds, ledgerV3CertificateGVK) ledgerV3CertManagerAvailable = issuerAvailable && certificateAvailable @@ -95,6 +102,61 @@ func withLedgerV3ClusterWatch() core.ReconcilerOption[*v1beta1.Ledger] { } } +func ledgerV3ClusterSupportsSinks(crds *apiextensionsv1.CustomResourceDefinitionList) bool { + for _, crd := range crds.Items { + if crd.Spec.Group != ledgerV3ClusterGVK.Group || crd.Spec.Names.Kind != ledgerV3ClusterGVK.Kind { + continue + } + for _, version := range crd.Spec.Versions { + if version.Name != ledgerV3ClusterGVK.Version || !version.Served || version.Schema == nil || version.Schema.OpenAPIV3Schema == nil { + continue + } + + specSchema, hasSpec := version.Schema.OpenAPIV3Schema.Properties["spec"] + statusSchema, hasStatus := version.Schema.OpenAPIV3Schema.Properties["status"] + if !hasSpec || specSchema.Type != "object" || !hasStatus || statusSchema.Type != "object" { + return false + } + sinksSchema, hasSinks := specSchema.Properties["sinks"] + appliedSinksSchema, hasAppliedSinks := statusSchema.Properties["appliedSinks"] + if !hasSinks || sinksSchema.Type != "object" || !hasAppliedSinks || appliedSinksSchema.Type != "array" || + appliedSinksSchema.Items == nil || appliedSinksSchema.Items.Schema == nil || appliedSinksSchema.Items.Schema.Type != "string" { + return false + } + + natsSchema, hasNATS := sinksSchema.Properties["nats"] + if !hasNATS || natsSchema.Type != "array" || natsSchema.Items == nil || natsSchema.Items.Schema == nil || natsSchema.Items.Schema.Type != "object" { + return false + } + natsItemSchema := natsSchema.Items.Schema + for _, field := range []string{"name", "url", "topic"} { + fieldSchema, found := natsItemSchema.Properties[field] + if !found || fieldSchema.Type != "string" || !slices.Contains(natsItemSchema.Required, field) { + return false + } + } + + return true + } + } + + return false +} + +func ledgerV3ClusterSupportsSinksAtRuntime(ctx core.Context) (bool, error) { + crd := &apiextensionsv1.CustomResourceDefinition{} + if err := ctx.GetAPIReader().Get(ctx, types.NamespacedName{Name: ledgerV3ClusterCRDName}, crd); err != nil { + if apierrors.IsNotFound(err) { + return false, nil + } + return false, fmt.Errorf("reading Ledger v3 Cluster CRD: %w", err) + } + + return ledgerV3ClusterSupportsSinks(&apiextensionsv1.CustomResourceDefinitionList{ + Items: []apiextensionsv1.CustomResourceDefinition{*crd}, + }), nil +} + func withLedgerConfigurationWatch() core.ReconcilerOption[*v1beta1.Ledger] { return core.WithWatch[*v1beta1.Ledger, *v1beta1.LedgerConfiguration]( func(ctx core.Context, configuration *v1beta1.LedgerConfiguration) []reconcile.Request { @@ -195,6 +257,17 @@ func reconcileV3(ctx core.Context, stack *v1beta1.Stack, ledger *v1beta1.Ledger, setLedgerV3Condition(ledger, metav1.ConditionFalse, "OperatorUnavailable", "Ledger v3 Cluster CRD is not installed") return core.NewPendingError().WithMessage("Ledger v3 operator unavailable: Cluster CRD is not installed") } + sinksSupported, err := ledgerV3ClusterSupportsSinksAtRuntime(ctx) + if err != nil { + setLedgerV3Condition(ledger, metav1.ConditionFalse, "OperatorDiscoveryFailed", err.Error()) + return err + } + if !sinksSupported { + setLedgerV3Condition(ledger, metav1.ConditionFalse, "OperatorIncompatible", "Ledger v3 Cluster CRD does not support managed event sinks") + return core.NewPendingError(). + WithMessage("Ledger v3 operator incompatible: Cluster CRD does not expose the managed spec.sinks.nats and status.appliedSinks contract"). + WithRequeueAfter(ledgerV3CRDDiscoveryRetryDelay) + } clearLegacyLedgerConditions(ledger) @@ -516,6 +589,65 @@ func isV3ClusterReady(cluster *unstructured.Unstructured) (bool, string, error) replicas = 3 } + ready := phase == "Running" && readyReplicas == replicas && observedGeneration == cluster.GetGeneration() message := fmt.Sprintf("phase=%s readyReplicas=%d/%d observedGeneration=%d/%d", phase, readyReplicas, replicas, observedGeneration, cluster.GetGeneration()) - return phase == "Running" && readyReplicas == replicas && observedGeneration == cluster.GetGeneration(), message, nil + + sinks, sinksManaged, err := unstructured.NestedFieldNoCopy(cluster.Object, "spec", "sinks") + if err != nil { + return false, "", err + } + if !sinksManaged || sinks == nil { + return ready, message, nil + } + + conditions, conditionsFound, err := unstructured.NestedSlice(cluster.Object, "status", "conditions") + if err != nil { + return false, "", err + } + if !conditionsFound { + return false, message + " sinksSynced=Unknown reason=ConditionMissing", nil + } + + for _, item := range conditions { + condition, ok := item.(map[string]interface{}) + if !ok { + return false, "", fmt.Errorf("invalid Ledger v3 Cluster status condition type %T", item) + } + conditionType, _, err := unstructured.NestedString(condition, "type") + if err != nil { + return false, "", err + } + if conditionType != ledgerV3SinksSyncedCondition { + continue + } + + status, _, err := unstructured.NestedString(condition, "status") + if err != nil { + return false, "", err + } + sinksObservedGeneration, _, err := unstructured.NestedInt64(condition, "observedGeneration") + if err != nil { + return false, "", err + } + reason, _, err := unstructured.NestedString(condition, "reason") + if err != nil { + return false, "", err + } + conditionMessage, _, err := unstructured.NestedString(condition, "message") + if err != nil { + return false, "", err + } + + message += fmt.Sprintf(" sinksSynced=%s sinksObservedGeneration=%d/%d", status, sinksObservedGeneration, cluster.GetGeneration()) + if reason != "" { + message += " sinksReason=" + reason + } + if conditionMessage != "" { + message += " sinksMessage=" + conditionMessage + } + + return ready && status == string(metav1.ConditionTrue) && sinksObservedGeneration == cluster.GetGeneration(), message, nil + } + + return false, message + " sinksSynced=Unknown reason=ConditionMissing", nil } diff --git a/internal/resources/ledgers/v3_preview.go b/internal/resources/ledgers/v3_preview.go index 12577315..9833ef64 100644 --- a/internal/resources/ledgers/v3_preview.go +++ b/internal/resources/ledgers/v3_preview.go @@ -40,7 +40,6 @@ func ledgerV3PreviewVersion(ctx core.Context, stack *v1beta1.Stack) (string, err if !ledgerV3ClusterAvailable { return "", nil } - version, err := settings.GetStringOrEmpty(ctx, stack.Name, "ledger", "v3", "preview-version") if err != nil { return "", err @@ -56,6 +55,17 @@ func reconcileV3Preview(ctx core.Context, stack *v1beta1.Stack, ledger *v1beta1. setLedgerV3PreviewCondition(ledger, metav1.ConditionFalse, "OperatorUnavailable", "Ledger v3 Cluster CRD is not installed") return core.NewPendingError().WithMessage("Ledger v3 preview unavailable: Cluster CRD is not installed") } + sinksSupported, err := ledgerV3ClusterSupportsSinksAtRuntime(ctx) + if err != nil { + setLedgerV3PreviewCondition(ledger, metav1.ConditionFalse, "OperatorDiscoveryFailed", err.Error()) + return err + } + if !sinksSupported { + setLedgerV3PreviewCondition(ledger, metav1.ConditionFalse, "OperatorIncompatible", "Ledger v3 Cluster CRD does not support managed event sinks") + return core.NewPendingError(). + WithMessage("Ledger v3 preview incompatible: Cluster CRD does not expose the managed spec.sinks.nats and status.appliedSinks contract"). + WithRequeueAfter(ledgerV3CRDDiscoveryRetryDelay) + } tlsReady, tlsMessage, tlsCAHash, err := createOrUpdateV3TLSResources(ctx, stack, ledger, true) if err != nil { diff --git a/internal/resources/ledgers/v3_test.go b/internal/resources/ledgers/v3_test.go index 409cd17a..9eb1d83e 100644 --- a/internal/resources/ledgers/v3_test.go +++ b/internal/resources/ledgers/v3_test.go @@ -3,6 +3,7 @@ package ledgers import ( "context" "errors" + "strings" "testing" appsv1 "k8s.io/api/apps/v1" @@ -141,6 +142,197 @@ func TestLedgerV3DiscoveryFailureDisablesCapabilityWithoutFailing(t *testing.T) } } +func validLedgerV3ClusterCRD() apiextensionsv1.CustomResourceDefinition { + stringSchema := apiextensionsv1.JSONSchemaProps{Type: "string"} + return apiextensionsv1.CustomResourceDefinition{ + ObjectMeta: metav1.ObjectMeta{Name: ledgerV3ClusterCRDName}, + Spec: apiextensionsv1.CustomResourceDefinitionSpec{ + Group: ledgerV3ClusterGVK.Group, + Names: apiextensionsv1.CustomResourceDefinitionNames{Kind: ledgerV3ClusterGVK.Kind, Plural: "clusters"}, + Scope: apiextensionsv1.NamespaceScoped, + Versions: []apiextensionsv1.CustomResourceDefinitionVersion{{ + Name: ledgerV3ClusterGVK.Version, + Served: true, + Storage: true, + Schema: &apiextensionsv1.CustomResourceValidation{OpenAPIV3Schema: &apiextensionsv1.JSONSchemaProps{ + Type: "object", + Properties: map[string]apiextensionsv1.JSONSchemaProps{ + "spec": {Type: "object", Properties: map[string]apiextensionsv1.JSONSchemaProps{ + "sinks": {Type: "object", Properties: map[string]apiextensionsv1.JSONSchemaProps{ + "nats": { + Type: "array", + Items: &apiextensionsv1.JSONSchemaPropsOrArray{Schema: &apiextensionsv1.JSONSchemaProps{ + Type: "object", + Required: []string{"name", "topic", "url"}, + Properties: map[string]apiextensionsv1.JSONSchemaProps{ + "name": stringSchema, "topic": stringSchema, "url": stringSchema, + }, + }}, + }, + }}, + }}, + "status": {Type: "object", Properties: map[string]apiextensionsv1.JSONSchemaProps{ + "appliedSinks": { + Type: "array", + Items: &apiextensionsv1.JSONSchemaPropsOrArray{Schema: &stringSchema}, + }, + }}, + }, + }}, + }}, + }, + } +} + +func TestLedgerV3ClusterSupportsSinks(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + mutate func(*apiextensionsv1.CustomResourceDefinition) + want bool + }{ + {name: "complete sink contract", want: true}, + {name: "version not served", mutate: func(crd *apiextensionsv1.CustomResourceDefinition) { crd.Spec.Versions[0].Served = false }}, + {name: "missing OpenAPI schema", mutate: func(crd *apiextensionsv1.CustomResourceDefinition) { crd.Spec.Versions[0].Schema.OpenAPIV3Schema = nil }}, + {name: "missing desired sink schema", mutate: func(crd *apiextensionsv1.CustomResourceDefinition) { + delete(crd.Spec.Versions[0].Schema.OpenAPIV3Schema.Properties["spec"].Properties, "sinks") + }}, + {name: "wrong desired sink type", mutate: func(crd *apiextensionsv1.CustomResourceDefinition) { + spec := crd.Spec.Versions[0].Schema.OpenAPIV3Schema.Properties["spec"] + sinks := spec.Properties["sinks"] + sinks.Type = "array" + spec.Properties["sinks"] = sinks + }}, + {name: "missing NATS sink schema", mutate: func(crd *apiextensionsv1.CustomResourceDefinition) { + spec := crd.Spec.Versions[0].Schema.OpenAPIV3Schema.Properties["spec"] + sinks := spec.Properties["sinks"] + delete(sinks.Properties, "nats") + spec.Properties["sinks"] = sinks + }}, + {name: "missing required NATS field", mutate: func(crd *apiextensionsv1.CustomResourceDefinition) { + spec := crd.Spec.Versions[0].Schema.OpenAPIV3Schema.Properties["spec"] + sinks := spec.Properties["sinks"] + nats := sinks.Properties["nats"] + nats.Items.Schema.Required = []string{"name", "url"} + sinks.Properties["nats"] = nats + spec.Properties["sinks"] = sinks + }}, + {name: "missing ownership status schema", mutate: func(crd *apiextensionsv1.CustomResourceDefinition) { + delete(crd.Spec.Versions[0].Schema.OpenAPIV3Schema.Properties["status"].Properties, "appliedSinks") + }}, + {name: "wrong ownership item type", mutate: func(crd *apiextensionsv1.CustomResourceDefinition) { + status := crd.Spec.Versions[0].Schema.OpenAPIV3Schema.Properties["status"] + applied := status.Properties["appliedSinks"] + applied.Items.Schema.Type = "integer" + status.Properties["appliedSinks"] = applied + }}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + crd := validLedgerV3ClusterCRD() + if test.mutate != nil { + test.mutate(&crd) + } + crds := &apiextensionsv1.CustomResourceDefinitionList{Items: []apiextensionsv1.CustomResourceDefinition{crd}} + if got := ledgerV3ClusterSupportsSinks(crds); got != test.want { + t.Fatalf("ledgerV3ClusterSupportsSinks() = %t, want %t", got, test.want) + } + }) + } +} + +func TestLedgerV3ClusterSinkContractRefreshesAfterCRDUpgrade(t *testing.T) { + scheme := runtime.NewScheme() + if err := apiextensionsv1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + crd := validLedgerV3ClusterCRD() + spec := crd.Spec.Versions[0].Schema.OpenAPIV3Schema.Properties["spec"] + delete(spec.Properties, "sinks") + kubernetesClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(&crd).Build() + ctx := ledgerV3DiscoveryContext{Context: context.Background(), reader: kubernetesClient} + + supported, err := ledgerV3ClusterSupportsSinksAtRuntime(ctx) + if err != nil { + t.Fatal(err) + } + if supported { + t.Fatal("legacy CRD unexpectedly supports managed sinks") + } + + updated := validLedgerV3ClusterCRD() + updated.ResourceVersion = crd.ResourceVersion + if err := kubernetesClient.Update(context.Background(), &updated); err != nil { + t.Fatal(err) + } + supported, err = ledgerV3ClusterSupportsSinksAtRuntime(ctx) + if err != nil { + t.Fatal(err) + } + if !supported { + t.Fatal("upgraded CRD sink contract was not discovered at reconciliation time") + } +} + +func TestLedgerV3ReconciliationRejectsIncompatibleSinkContract(t *testing.T) { + previous := ledgerV3ClusterAvailable + ledgerV3ClusterAvailable = true + t.Cleanup(func() { ledgerV3ClusterAvailable = previous }) + + scheme := runtime.NewScheme() + if err := apiextensionsv1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + crd := validLedgerV3ClusterCRD() + spec := crd.Spec.Versions[0].Schema.OpenAPIV3Schema.Properties["spec"] + delete(spec.Properties, "sinks") + kubernetesClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(&crd).Build() + ctx := ledgerV3DiscoveryContext{Context: context.Background(), reader: kubernetesClient} + stack := &v1beta1.Stack{ObjectMeta: metav1.ObjectMeta{Name: "stack0"}} + + tests := []struct { + name string + conditionType string + reconcile func(*v1beta1.Ledger) error + }{ + { + name: "primary Ledger v3", + conditionType: ledgerV3ClusterReadyCondition, + reconcile: func(ledger *v1beta1.Ledger) error { + return reconcileV3(ctx, stack, ledger, "v3.0.0") + }, + }, + { + name: "Ledger v3 preview", + conditionType: ledgerV3PreviewReadyCondition, + reconcile: func(ledger *v1beta1.Ledger) error { + return reconcileV3Preview(ctx, stack, ledger, "v3.0.0") + }, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + ledger := &v1beta1.Ledger{ObjectMeta: metav1.ObjectMeta{Name: "ledger0", Generation: 3}} + err := test.reconcile(ledger) + if err == nil { + t.Fatal("reconciliation unexpectedly accepted an incompatible Cluster CRD") + } + if got := core.ApplicationErrorRequeueAfter(err); got != ledgerV3CRDDiscoveryRetryDelay { + t.Fatalf("requeue delay = %s, want %s", got, ledgerV3CRDDiscoveryRetryDelay) + } + condition := ledger.GetConditions().Get(test.conditionType) + if condition == nil { + t.Fatalf("missing %s condition", test.conditionType) + } + if condition.Status != metav1.ConditionFalse || condition.Reason != "OperatorIncompatible" { + t.Fatalf("condition = status %s reason %q, want False/OperatorIncompatible", condition.Status, condition.Reason) + } + }) + } +} + func TestLedgerV3PreviewVersionIgnoredWhenClusterUnavailable(t *testing.T) { previous := ledgerV3ClusterAvailable ledgerV3ClusterAvailable = false @@ -346,3 +538,74 @@ func TestNormalizeLedgerV3Replicas(t *testing.T) { }) } } + +func TestIsV3ClusterReadyIncludesManagedSinks(t *testing.T) { + t.Parallel() + + condition := func(status metav1.ConditionStatus, generation int64, reason, message string) *metav1.Condition { + return &metav1.Condition{ + Type: ledgerV3SinksSyncedCondition, + Status: status, + ObservedGeneration: generation, + Reason: reason, + Message: message, + } + } + tests := []struct { + name string + managedSinks bool + condition *metav1.Condition + baseNotReady bool + wantReady bool + messageChecks []string + }{ + {name: "unmanaged sinks preserve legacy readiness", wantReady: true}, + {name: "managed sinks wait for condition", managedSinks: true, messageChecks: []string{"sinksSynced=Unknown", "ConditionMissing"}}, + {name: "managed sinks propagate failure", managedSinks: true, condition: condition(metav1.ConditionFalse, 7, "Error", "name conflict"), messageChecks: []string{"sinksSynced=False", "sinksReason=Error", "name conflict"}}, + {name: "managed sinks reject stale success", managedSinks: true, condition: condition(metav1.ConditionTrue, 6, "Synced", "configured"), messageChecks: []string{"sinksSynced=True", "6/7"}}, + {name: "managed sinks preserve base readiness", managedSinks: true, condition: condition(metav1.ConditionTrue, 7, "Synced", "configured"), baseNotReady: true, messageChecks: []string{"phase=Pending", "sinksSynced=True"}}, + {name: "managed sinks accept current success", managedSinks: true, condition: condition(metav1.ConditionTrue, 7, "Synced", "configured"), wantReady: true, messageChecks: []string{"sinksSynced=True", "7/7"}}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + cluster := newV3Cluster() + cluster.SetGeneration(7) + spec := map[string]interface{}{"replicas": int64(3)} + if test.managedSinks { + spec["sinks"] = map[string]interface{}{} + } + status := map[string]interface{}{ + "phase": "Running", + "readyReplicas": int64(3), + "observedGeneration": int64(7), + } + if test.baseNotReady { + status["phase"] = "Pending" + } + if test.condition != nil { + conditionMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(test.condition) + if err != nil { + t.Fatal(err) + } + status["conditions"] = []interface{}{conditionMap} + } + cluster.Object["spec"] = spec + cluster.Object["status"] = status + + ready, message, err := isV3ClusterReady(cluster) + if err != nil { + t.Fatalf("isV3ClusterReady() returned error: %v", err) + } + if ready != test.wantReady { + t.Fatalf("isV3ClusterReady() ready = %t, want %t; message=%q", ready, test.wantReady, message) + } + for _, check := range test.messageChecks { + if !strings.Contains(message, check) { + t.Fatalf("isV3ClusterReady() message %q does not contain %q", message, check) + } + } + }) + } +} diff --git a/internal/tests/crds/ledger.formance.com_clusters.yaml b/internal/tests/crds/ledger.formance.com_clusters.yaml index 4c5ee908..ce95c160 100644 --- a/internal/tests/crds/ledger.formance.com_clusters.yaml +++ b/internal/tests/crds/ledger.formance.com_clusters.yaml @@ -34,8 +34,31 @@ spec: x-kubernetes-validations: - rule: oldSelf == '' || self == oldSelf message: bindAddr is immutable once set + sinks: + type: object + properties: + nats: + type: array + items: + type: object + required: + - name + - topic + - url + properties: + name: + type: string + topic: + type: string + url: + type: string status: type: object x-kubernetes-preserve-unknown-fields: true + properties: + appliedSinks: + type: array + items: + type: string subresources: status: {} diff --git a/internal/tests/ledger_v3_controller_test.go b/internal/tests/ledger_v3_controller_test.go index 3f043689..403c86fe 100644 --- a/internal/tests/ledger_v3_controller_test.go +++ b/internal/tests/ledger_v3_controller_test.go @@ -610,6 +610,14 @@ var _ = Describe("Ledger v3 controller", func() { "phase": "Running", "readyReplicas": int64(3), "observedGeneration": cluster.GetGeneration(), + "conditions": []any{ + map[string]any{ + "type": "SinksSynced", + "status": "True", + "reason": "Synced", + "observedGeneration": cluster.GetGeneration(), + }, + }, } if err := TestContext().GetClient().Status().Update(TestContext(), cluster); err != nil { return false