Skip to content

feat(operator): configure Ledger v3 NATS sinks #530

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
flemzord wants to merge 5 commits into
base: main
Choose a base branch
from
Open

feat(operator): configure Ledger v3 NATS sinks #530

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
d5194ce
feat(operator): configure Ledger v3 NATS sinks
flemzord Aug 31, 2026
cf1629d
fix(operator): preserve Ledger v3 compatibility
flemzord Aug 31, 2026
ce3a703
fix(broker): persist NATS subject migration
flemzord Aug 31, 2026
5bacb70
chore(operator): sync Ledger sinks branch with main
flemzord Sep 1, 2026
966caf5
fix(operator): gate Ledger sinks on CRD readiness
flemzord Sep 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions api/formance.com/v1beta1/broker_types.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
6 changes: 6 additions & 0 deletions config/crd/bases/formance.com_brokers.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,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:
Expand Down
190 changes: 186 additions & 4 deletions config/crd/bases/formance.com_ledgerconfigurations.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1235,7 +1235,11 @@ spec:
manifests since it also unlocks the trace level.
type: boolean
dnsEndpoint:
description: DNSEndpoint configuration for ExternalDNS.
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:
Expand Down Expand Up @@ -1293,8 +1297,92 @@ spec:
type: array
type: object
x-kubernetes-validations:
- message: endpoints are required when dnsEndpoint is enabled
- message: endpoints are required when the DNSEndpoint is enabled
rule: '!self.enabled || size(self.endpoints) > 0'
dnsEndpoints:
Comment thread
NumaryBot marked this conversation as resolved.
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
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.
Expand Down Expand Up @@ -2879,8 +2967,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.
Expand Down Expand Up @@ -3609,6 +3699,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
<topic>.<ledger>.<event-type-lowercase>.
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:
Expand Down
1 change: 1 addition & 0 deletions docs/09-Configuration reference/02-Custom Resource Definitions.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2035,6 +2035,7 @@ Broker is the Schema for the brokers API
| `uri` _string_ | | | Type: string <br /> |
| `mode` _[Mode](#mode)_ | Mode indicating the configuration of the NATS streams<br />Two modes are defined:<br />- ModeOneStreamByService: In this case, each service will have a dedicated stream created<br />- 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] <br /> |
| `streams` _string array_ | Streams list streams created when Mode == ModeOneStreamByService | | |
| `streamSubjectsRevision` _string_ | StreamSubjectsRevision records the subject-layout migration applied to the<br />NATS streams. It prevents completed migration Jobs from being recreated<br />after their TTL expires. | | |

###### Mode

Expand Down
14 changes: 7 additions & 7 deletions docs/09-Configuration reference/settings.catalog.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
]
},
Expand Down Expand Up @@ -201,36 +201,36 @@
"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"
]
},
{
"key": "deployments.ledger.replicas",
"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"
]
},
{
Expand Down Expand Up @@ -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"
]
},
{
Expand Down
2 changes: 1 addition & 1 deletion go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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-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
Expand Down
4 changes: 2 additions & 2 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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-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=
Expand Down
6 changes: 6 additions & 0 deletions ...templates/crds/apiextensions.k8s.io_v1_customresourcedefinition_brokers.formance.com.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,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:
Expand Down
Loading