Documentation
¶
Index ¶
- type AuthReader
- type CatalogReader
- type Compiler
- func (c *Compiler) Compile(ctx context.Context) (*readmodel.Snapshot, error)
- func (c *Compiler) CompileAll(ctx context.Context) (*readmodel.Snapshot, map[string]*readmodel.Snapshot, *readmodel.Snapshot, ...)
- func (c *Compiler) CompileFor(ctx context.Context, scope string) (*readmodel.Snapshot, error)
- type ConsumerReader
- type Dispatcher
- type DispatcherConfig
- type GatewayReader
- type Holder
- func (h *Holder) Set(raw []byte, version string)
- func (h *Holder) SetPartitioned(raw []byte, version string, scoped map[string]ScopedSnapshot)
- func (h *Holder) Snapshot() (raw []byte, version string, ok bool)
- func (h *Holder) SnapshotFor(scope string) (raw []byte, version string, ok bool)
- func (h *Holder) Version() string
- type PartitionedCompiler
- type PolicyReader
- type RegistryReader
- type RoleReader
- type ScopedSnapshot
- type Signaler
- type SnapshotCompiler
- type SnapshotVersionPublisher
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type AuthReader ¶
type AuthReader interface {
List(ctx context.Context, filter authdomain.ListFilter) ([]*authdomain.Auth, int, error)
}
type CatalogReader ¶
type Compiler ¶
type Compiler struct {
// contains filtered or unexported fields
}
func NewCompiler ¶
func NewCompiler( gateways GatewayReader, consumers ConsumerReader, registries RegistryReader, policies PolicyReader, auths AuthReader, roles RoleReader, catalog CatalogReader, logger *slog.Logger, ) *Compiler
func (*Compiler) CompileAll ¶
func (c *Compiler) CompileAll(ctx context.Context) (*readmodel.Snapshot, map[string]*readmodel.Snapshot, *readmodel.Snapshot, error)
CompileAll compiles the global snapshot, one tenant-only snapshot per gateway scope, and the shared catalog as its own snapshot. The catalog (providers + models, usually the bulk of the encoded bytes) is deliberately NOT merged into the global or scoped snapshots: the dispatcher encodes it once and appends the encoded bytes to every snapshot it publishes, instead of re-encoding the same catalog per gateway on every compile.
type ConsumerReader ¶
type ConsumerReader interface {
ListByGateway(ctx context.Context, gatewayID ids.GatewayID) ([]*consumerdomain.Consumer, error)
// List with a zero GatewayID pages across every gateway; the compiler's
// bulk collect path uses it to load all consumers in one scan.
List(ctx context.Context, filter consumerdomain.ListFilter) ([]*consumerdomain.Consumer, int, error)
}
type Dispatcher ¶
type Dispatcher struct {
// contains filtered or unexported fields
}
Dispatcher compiles the config snapshot, holds it for the gRPC server to serve, broadcasts the new version to connected data planes, and drains the change-marker outbox that survives a control-plane restart. A committed admin write leaves a durable marker; the dispatcher folds every marker visible before the compile into one recompile and drains exactly that observed set.
func NewDispatcher ¶
func NewDispatcher( compiler SnapshotCompiler, codec configsync.SnapshotCodec[*readmodel.Snapshot], holder *Holder, broadcaster configsyncport.VersionBroadcaster, outbox configsyncport.OutboxRepository, logger *slog.Logger, cfg DispatcherConfig, ) *Dispatcher
NewDispatcher builds the control-plane snapshot dispatcher.
func (*Dispatcher) Dispatch ¶
func (d *Dispatcher) Dispatch(ctx context.Context) error
Dispatch runs one dispatch cycle synchronously. It must not run concurrently with Run or another Dispatch: the outbox frontier (MaxSeq/DeleteUpTo) is not mutually exclusive across cycles, so overlapping callers could drain against different frontiers.
func (*Dispatcher) Run ¶
func (d *Dispatcher) Run(ctx context.Context) error
Run drives the dispatch loop until ctx is cancelled: an immediate dispatch on the first Signal after a quiet period so a single admin write (a new consumer, key, …) reaches data planes without sitting out the debounce, a trailing debounced dispatch that folds the rest of a Signal burst, a jittered backstop dispatch, and an initial catch-up when the outbox holds markers left by a write that committed before a prior restart.
func (*Dispatcher) Signal ¶
func (d *Dispatcher) Signal()
Signal requests a dispatch after an admin write: immediate when the loop is quiet, folded into the trailing debounced dispatch during a write burst.
type DispatcherConfig ¶
type DispatcherConfig struct {
Debounce time.Duration
Backstop time.Duration
Retention time.Duration
MaxRows int
}
DispatcherConfig tunes the debounce/backstop cadence and the outbox safety bound.
type GatewayReader ¶
type GatewayReader interface {
List(ctx context.Context, filter gatewaydomain.ListFilter) ([]*gatewaydomain.Gateway, int, error)
}
type Holder ¶
type Holder struct {
// contains filtered or unexported fields
}
func (*Holder) SetPartitioned ¶
func (h *Holder) SetPartitioned(raw []byte, version string, scoped map[string]ScopedSnapshot)
func (*Holder) SnapshotFor ¶
type PartitionedCompiler ¶
type PartitionedCompiler interface {
CompileAll(ctx context.Context) (global *readmodel.Snapshot, scoped map[string]*readmodel.Snapshot, catalog *readmodel.Snapshot, err error)
}
PartitionedCompiler compiles the global snapshot, tenant-only scoped snapshots, and the shared catalog separately, so the dispatcher can encode the catalog once and append the encoded bytes to every published snapshot. Protobuf wire format merges concatenated messages (repeated fields append), so tenant bytes + catalog bytes decode as one full snapshot on the data plane; both halves use deterministic marshaling, keeping version hashes stable across replicas.
type PolicyReader ¶
type PolicyReader interface {
ListByGateway(ctx context.Context, gatewayID ids.GatewayID) ([]*policydomain.Policy, error)
// List with a zero GatewayID pages across every gateway; the compiler's
// bulk collect path uses it to load all policies in one scan.
List(ctx context.Context, filter policydomain.ListFilter) ([]*policydomain.Policy, int, error)
}
type RegistryReader ¶
type RegistryReader interface {
List(ctx context.Context, filter registrydomain.ListFilter) ([]*registrydomain.Registry, int, error)
}
type RoleReader ¶
type RoleReader interface {
ListByGateway(ctx context.Context, gatewayID ids.GatewayID) ([]*roledomain.Role, error)
// List with a zero GatewayID pages across every gateway; the compiler's
// bulk collect path uses it to load all roles in one scan.
List(ctx context.Context, filter roledomain.ListFilter) ([]*roledomain.Role, int, error)
}
type ScopedSnapshot ¶
type SnapshotCompiler ¶
SnapshotCompiler compiles the current State-of-the-World from the live config.
type SnapshotVersionPublisher ¶
type SnapshotVersionPublisher struct {
// contains filtered or unexported fields
}
func NewSnapshotVersionPublisher ¶
func NewSnapshotVersionPublisher(signaler Signaler) *SnapshotVersionPublisher
func (*SnapshotVersionPublisher) Signal ¶
func (p *SnapshotVersionPublisher) Signal(_ context.Context)
Source Files