Skip to content

configstream.md

Latest commit

 

History

History
145 lines (114 loc) · 5.61 KB

configstream.md

File metadata and controls

145 lines (114 loc) · 5.61 KB
Raw

configstream — xDS-style streaming config push (M11)

Server-streaming gRPC service that pushes compiled AuthConfig snapshots to many subscribers with latest-wins conflation — slow consumers can never block Publish. Late subscribers are primed with the current snapshot so a mid-flight pod restart catches up immediately.

Source: pkg/configstream. Service: lightweightauth.v1.ConfigDiscovery / StreamAuthConfig (one server-streaming RPC).

When to use

  • Sidecar / fleet topologies where one lwauth control-plane pushes to many lwauth data-plane replicas.
  • Embedders building their own Compile-and-Swap path (e.g. a custom proxy) that need an authoritative push channel rather than periodic polling of a ConfigMap.

Not needed for the standard CRD/file deployment — the controller swaps engines in-process.

Server side

The reconciler optionally publishes to a Broker after each successful Compile-and-Swap:

import "github.com/mikeappsec/lightweightauth/pkg/configstream"

br := configstream.NewBroker()                // 1 publisher, N subscribers
gs := grpc.NewServer()
authz := func(ctx context.Context) error {
  // Inspect peer TLS state / metadata and return a gRPC status error
  // on unauthenticated or unauthorized callers.
  return nil
}
configstream.NewServer(br, authz).Register(gs) // mounts ConfigDiscovery
go gs.Serve(lis)

// In your reconciler, after a successful Compile():
br.Publish(authConfig)                        // *config.AuthConfig, version-tagged

NewServer requires a non-nil Authorizer. Passing nil panics so embedders cannot accidentally expose full AuthConfig snapshots by registering the service on an unauthenticated listener. If you truly intend an open stream in a trusted test harness, pass an explicit allow-all function in that code.

Broker.Publish supports concurrent multi-writer calls. This used to have a known caveat (an older snapshot could win against a newer one under contention), but that's fixed as of v1.1 (pkg/configstream/broker.go's subscription.deliver rejects any snapshot whose version isn't strictly greater than the subscriber's high-water mark; see DESIGN.md's M12-BROKER-MW entry, "shipped (v1.1)"), fenced by TestBrokerStress_MultiWriter and TestBrokerDeliver_RejectsStaleVersion in pkg/configstream/stress_test.go.

Client side

import "github.com/mikeappsec/lightweightauth/pkg/configstream"

conn, _ := grpc.NewClient("control-plane:9001", grpc.WithTransportCredentials(...))
defer conn.Close()

err := configstream.Stream(ctx, conn, "node-id-here", func(ctx context.Context, version uint64, cfg *config.AuthConfig) error {
  eng, err := config.Compile(cfg)
  if err != nil { return err }
  engineHolder.Swap(eng)                     // your Compile-and-Swap path
  return nil
})

The helper:

  • Sends an initial StreamAuthConfigRequest{nodeId} so the server can scope the snapshot to the caller.
  • Calls handler synchronously per snapshot — return non-nil to abort the stream.
  • Is one-shot; reconnect/backoff logic belongs in the caller so it can share fleet-specific jitter and shutdown policy.

Wire shape

service ConfigDiscovery {
  rpc StreamAuthConfig(StreamAuthConfigRequest)
      returns (stream AuthConfigSnapshot);
}

message AuthConfigSnapshot {
  uint64 version  = 1;   // monotonic per-Broker
  bytes  spec_json = 2;  // JSON-encoded AuthConfig
}

JSON over the wire is deliberate: module-specific free-form config maps round-trip cleanly through it but not through protobuf's struct/Any. The cost is one JSON round-trip per snapshot — negligible versus push frequency.

Conflation contract

Publish(v=1)        →  Publish(v=2)  →  Publish(v=3)
                     ↘                     ↓
   slow subscriber  ←——————————————— sees v=3 only

A subscriber that hasn't drained its channel by the time Publish(v=3) arrives loses v=2 — by design. The contract is "eventually converge to latest", never "every snapshot delivered".

Storm resilience (M12 slice 5)

Validated under reconnect storms in pkg/configstream/grpc_storm_test.go:

  • 16 clients × 3 reconnects + 1 long-lived over single HTTP/2 conn vs 100-publish stream — runs <250 ms under -race.
  • Server-cancel (gs.Stop()) closes all client streams cleanly with zero goroutine leaks.

Operational notes

  • Versioning. version is a monotonic counter per Broker instance. On control-plane restart, the counter resets to 0 — clients must treat version comparison as "snapshot bytes != prior snapshot" rather than "version > prior version".
  • Authorization. NewServer requires an application authorizer for every stream. Operators should also protect the listener with mTLS or an equivalent private control-plane boundary.
  • Multi-tenant. A single Broker pushes the cluster-wide snapshot; tenant filtering is the subscriber's responsibility (the nodeId argument lets the server scope replies in custom builds).

References