From aa80edb9ccb2791420ddf31f3bc9a14123e01fd3 Mon Sep 17 00:00:00 2001 From: Giuseppe Chiesa Date: Sun, 9 Aug 2026 17:35:48 +0200 Subject: [PATCH 1/2] feat(ms020): peristent grpc channel for hi priority messaging --- ...istent-grpc-channel-for-hiprio-messaged.md | 82 +++ ...grpc-channel-for-hiprio-messaged.completed | 201 +++++++ ci/scripts/proto-gen.sh | 12 +- deployments/k8s-embedded-proxy/README.md | 134 +++++ .../k8s-embedded-proxy/base/configmap.yaml | 114 ++++ .../k8s-embedded-proxy/base/deployment.yaml | 114 ++++ .../base/kustomization.yaml | 21 + .../k8s-embedded-proxy/base/service.yaml | 43 ++ internal/cmd/cluster.go | 19 + internal/config/config.go | 20 + internal/config/config_test.go | 163 ++++++ internal/config/resources/default.kdl | 5 + internal/membership/channel.go | 490 ++++++++++++++++++ internal/membership/channel_test.go | 266 ++++++++++ internal/membership/delegate.go | 94 +++- internal/membership/delegate_test.go | 188 +++++++ internal/membership/event.go | 13 + internal/membership/membership.go | 103 +++- internal/membership/membership_test.go | 67 +++ internal/metrics/metrics.go | 53 ++ internal/proto/channel.pb.go | 181 +++++++ internal/proto/channel.proto | 31 ++ internal/proto/channel_grpc.pb.go | 135 +++++ mise.toml | 1 + 24 files changed, 2535 insertions(+), 15 deletions(-) create mode 100644 .junie/workflow/milestones/020-persistent-grpc-channel-for-hiprio-messaged.md create mode 100644 .junie/workflow/state/020-persistent-grpc-channel-for-hiprio-messaged.completed create mode 100644 deployments/k8s-embedded-proxy/README.md create mode 100644 deployments/k8s-embedded-proxy/base/configmap.yaml create mode 100644 deployments/k8s-embedded-proxy/base/deployment.yaml create mode 100644 deployments/k8s-embedded-proxy/base/kustomization.yaml create mode 100644 deployments/k8s-embedded-proxy/base/service.yaml create mode 100644 internal/membership/channel.go create mode 100644 internal/membership/channel_test.go create mode 100644 internal/proto/channel.pb.go create mode 100644 internal/proto/channel.proto create mode 100644 internal/proto/channel_grpc.pb.go diff --git a/.junie/workflow/milestones/020-persistent-grpc-channel-for-hiprio-messaged.md b/.junie/workflow/milestones/020-persistent-grpc-channel-for-hiprio-messaged.md new file mode 100644 index 0000000..6df85ca --- /dev/null +++ b/.junie/workflow/milestones/020-persistent-grpc-channel-for-hiprio-messaged.md @@ -0,0 +1,82 @@ +# 020-persistent-grpc-channel-for-hiprio-messaged.md + +## Context + +When a request is calculated to be blocked by rate-limiting, DRL currently establishes an on-demand TCP connection to +each cluster member to propagate the block message. This overhead is inefficient when the server is under heavy load. + +## Architectural Decision + +After evaluating the options: + +1. **Option A (2 unidirectional gRPC connections per node pair):** Creates a spaghetti web of independent connections + that increases resource footprint, connection management complexity, and lock contention across sockets. +2. **Option B (1 single bidirectional gRPC connection per node pair):** Reduces open file descriptors by half, + simplifies lifecycle management on node join/leave, and enables full HTTP/2 multiplexing for concurrent high-priority + event streams (e.g., blocking/unblocking events) across the same connection. + +**Decision:** Adopt **Option B** (a single bidirectional gRPC connection established between each unique pair of cluster +nodes). + +## Goal + +Replace the on-demand high-priority messaging propagation for blocking events with a single persistent bidirectional +gRPC channel between each cluster member pair. + +The channel will be established on port `7956`. When enabled, this channel will handle immediate high-priority events +(such as rate-limiting blocks) across cluster members. + +To preserve current package organization, extend the `membership` package with additional files to manage the gRPC +persistent channel lifecycle and event transport. + +## Requirements + +### 1. Configuration & Feature Flag + +- Add feature flag environment variable: `DRL_MEMBERSHIP_USE_HIPRIO_PERSISTENT_CHANNEL` (boolean, default: `true`). +- Extend KDL configuration schema under the `membership` block to include: + ```kdl + membership { + use-hiprio-persistent-channel true + hiprio-channel-port 7956 + } + ``` + +* Wire all configuration options cleanly in `internal/config` with proper environment variable overrides and default + values. + +### 2. Extensible Message Protocol + +* Implement a forward-compatible gRPC message structure in `internal/proto/hiprio.proto` (or within the existing proto + framework). +* Design the message format to support multiple high-priority event types via an event identifier or `oneof` payload + structure (similar to `internal/proto/accounting.proto`), starting with `BlockEvent` and `UnblockEvent`. + +### 3. gRPC Persistent Channel Implementation + +* Use TCP port `7956` for inter-node persistent gRPC messaging. +* Maintain a single bidirectional gRPC channel between each unique node pair in the cluster. +* **Node Join:** When a new member joins the cluster, establish/accept the gRPC connection with the node and log an + `INFO` event (e.g., `"Established persistent gRPC channel with peer [NodeID / IP:7956]"`). +* **Joiner Readiness**: When a new member joins the cluster, it's ready only when all the persistent gRPC channel + connections are established. +* **Node Leave:** When a member leaves or fails health checks, cleanly close the gRPC channel and clear associated + client connections from memory. Log a `WARN`/`INFO` event accordingly. + +### 4. Integration with Membership Package + +* Place persistent channel logic inside the `membership` package (e.g., `internal/membership/grpc_channel.go`). +* When `DRL_MEMBERSHIP_USE_HIPRIO_PERSISTENT_CHANNEL` is set to `true`, route high-priority blocking and unblocking + events through the new gRPC persistent channel instead of short-lived on-demand TCP connections. +* Ensure fallback behavior or fallback errors are logged gracefully if a channel suffers a temporary transport failure. +* Ensure when old TCP on demand model is used there is a log message level WARN for the event. + +## Outcome + +* **State Report:** Generate the report of all state file changes in Markdown format upon completion. +* **Configuration Update:** DRL configuration updated to parse and validate `use-hiprio-persistent-channel` and + `hiprio-channel-port`. +* **Protocol Definition:** Protobuf definitions compiled and wired for extensible event transport. +* **Runtime Execution:** Clustered nodes maintain a single bidirectional gRPC channel on port `7956` for blocking + propagation when the feature flag is enabled. +* **Implement necessary testings:** Unit tests for the new gRPC persistent channel implementation. diff --git a/.junie/workflow/state/020-persistent-grpc-channel-for-hiprio-messaged.completed b/.junie/workflow/state/020-persistent-grpc-channel-for-hiprio-messaged.completed new file mode 100644 index 0000000..527c343 --- /dev/null +++ b/.junie/workflow/state/020-persistent-grpc-channel-for-hiprio-messaged.completed @@ -0,0 +1,201 @@ +# Milestone 020: Persistent gRPC Channel for Hi-Priority Messaging - Completed + +## Original Plan + +### Goal +Replace the on-demand memberlist `SendReliable` (TCP) path used to propagate hi-priority +blocking/unblocking events with a single, persistent, bidirectional gRPC channel established +once between each unique pair of cluster nodes, on TCP port `7956`. + +### Design Decisions +- **Option B adopted** (per the milestone's architectural decision): a single bidirectional + gRPC connection per node pair, rather than two independent unidirectional connections. This + halves open file descriptors, simplifies join/leave lifecycle management, and allows HTTP/2 + multiplexing of concurrent block/unblock streams over one socket. +- **Deterministic dial direction**: of any two peers, only the node with the lexicographically + smaller address dials; the other accepts. Both sides observe memberlist's `NotifyJoin` + symmetrically and call `EstablishForPeer`, which independently reaches the same conclusion on + each side without any coordination handshake. +- **New file `internal/membership/channel.go`** (equivalent to the milestone's suggested + `grpc_channel.go`) hosts the `ChannelManager` — the persistent channel's lifecycle owner: gRPC + server, per-peer client streams, connect/reconnect/teardown, and message dispatch. +- **New proto `internal/proto/channel.proto`** (equivalent to the milestone's suggested + `hiprio.proto`) defines `ChannelMessage` as a `oneof` of `BlockEventWithExpiresAt` / + `UnblockEvent`, mirroring the extensible pattern already used by `accounting.proto`, plus the + `PersistentChannel` bidi-streaming gRPC service. +- **Feature flag**: `DRL_MEMBERSHIP_USE_HIPRIO_PERSISTENT_CHANNEL` (env), `use-hiprio-persistent-channel` + (KDL), defaulting to `true`. The default is supplied via `internal/config/resources/default.kdl` + (the established pattern in this codebase for boolean defaults, since a Go `bool` zero value + cannot itself express "default true"). +- **Routing**: `StateDelegate.QueueBlockEvent`/`QueueUnblockEvent` check `useChannel()` (feature + enabled AND a `ChannelManager` is attached to the cluster) and send over the persistent channel + when true, falling back to the legacy `SendReliable` (TCP) path otherwise. Both paths are + "fire and forget" from the caller's perspective — failures are logged, never propagated, per the + project's availability-over-consistency principle. +- **Joiner readiness gating**: a newly joining node blocks in `JoinCluster()` until the + `ChannelManager` reports an established connection to every currently known peer (bounded by a + 30s timeout, after which it proceeds anyway and logs a `WARN`), before falling through to the + existing state-sync readiness gate. +- **Legacy-path visibility**: a `WARN` log is now emitted every time a hi-priority event takes the + legacy on-demand TCP path (feature disabled, or channel not yet established to peers), so + operators can see effective transport usage during rollouts or connectivity issues. + +## Changes Made + +### internal/proto/channel.proto (new file) + generated channel.pb.go / channel_grpc.pb.go +- `ChannelMessage` message: `oneof content { BlockEventWithExpiresAt block_with_expires_at = 1; UnblockEvent unblock = 2; }`. +- `PersistentChannel` gRPC service with a single bidirectional-streaming RPC (`Stream`). + +### internal/membership/channel.go (new file) +- `ChannelManagerConfig{LocalAddr, Port, Handler, Metrics, Logger}` / `NewChannelManager`. +- `Start()` — starts the gRPC server, binding via `net.JoinHostPort(cm.localAddr, ...)` (scoped to + the node's own address rather than all interfaces, both for test isolation and production + correctness). +- `EstablishForPeer(remoteAddr string)` — deterministic dial-vs-accept decision; no-ops for self + and empty addresses. +- `Send(addr string, msg *drlproto.ChannelMessage) error` — serialized per-peer via a dedicated + write goroutine/channel. +- `Close(addr string)` / `Stop()` — tear down a single peer or the whole manager. +- `IsConnected(addr string) bool`, `PeerCount() int`. +- `Stream(...)` — the gRPC service method, handling the server side of incoming bidi streams, + including the `x-drl-node-addr` metadata handshake used for peer identity. +- INFO logs on channel establishment/close (satisfies the Node Join/Leave logging requirement); + WARN logs on dial/stream/send failures (satisfies the fallback/temporary-failure logging + requirement). + +### internal/membership/delegate.go +- `useChannel() bool` — gates routing on `cluster.config.Membership.UseHiPrioPersistentChannel` + and a non-nil `ChannelManager`. +- `sendToAllPeersViaChannel(msg)` — new send path used when `useChannel()` is true. +- `handleChannelBlockWithExpiresAt` / `handleChannelUnblock` — apply incoming channel events using + the same cache-mutation logic as the legacy `NotifyMsg` handlers. +- **New**: `warnLegacyPath(eventType string)` — logs `WARN` "using legacy on-demand TCP path for + hi-priority event; persistent gRPC channel disabled or unavailable" with an `event_type` + attribute (`block`/`unblock`). Called from `QueueBlockEvent`/`QueueUnblockEvent` immediately + before `sendToAllPeersAsync`, i.e. only on the legacy branch. + +### internal/membership/membership.go +- `channelManager *ChannelManager` field + `SetChannelManager`/`GetChannelManager` on `Cluster`. +- `Leave()` stops the channel manager before leaving memberlist. +- **New**: `waitForChannelsReady()` — polls (50ms interval, 30s `channelReadyTimeout`) until + `ChannelManager.IsConnected` is true for every peer in `MemberAddrs()` (excluding self), then + logs `Info`; on timeout, logs `Warn` and proceeds. No-ops immediately when no `ChannelManager` + is attached (feature disabled). +- `JoinCluster()` now calls `waitForChannelsReady()` right after updating cluster size/cache nodes + post-join, and before the existing state-sync wait / `markReady()` branch — so a joining node's + overall readiness is gated on both channel establishment and blocklist state sync. + +### internal/membership/event.go +- `NotifyJoin` calls `ChannelManager.EstablishForPeer` for the newly observed node (safe from both + the joiner's and existing members' perspective, due to the deterministic dial decision). +- `NotifyLeave` calls `ChannelManager.Close` for the departed node and logs `Info` (satisfies the + Node Leave logging requirement). + +### internal/cmd/cluster.go +- When `cfg.Membership.UseHiPrioPersistentChannel` is true, constructs and starts a + `ChannelManager` on `cfg.Membership.HiPrioChannelPort`, attaches it to the cluster via + `SetChannelManager`, and logs `Info` with the configured port. + +### internal/config/config.go +- `MembershipConfig` gains: + - `UseHiPrioPersistentChannel bool` — `kdl:"use-hiprio-persistent-channel" env:"USE_HIPRIO_PERSISTENT_CHANNEL"` + - `HiPrioChannelPort int` — `kdl:"hiprio-channel-port" env:"HIPRIO_CHANNEL_PORT"` +- `Validate()`: + - Defaults `HiPrioChannelPort` to `7956` when unset (zero). + - Rejects out-of-range ports (must be `1-65535`). + - Rejects `HiPrioChannelPort == Port` (the persistent channel port must not collide with the + memberlist gossip port), enforced only when `UseHiPrioPersistentChannel` is true. + +### internal/config/resources/default.kdl +- `membership` block now includes: + ```kdl + use-hiprio-persistent-channel true + hiprio-channel-port 7956 + ``` + This is the mechanism supplying the spec-mandated `default: true`, consistent with how other + boolean defaults (e.g. `internal-api.enabled`) are handled in this codebase. + +### internal/metrics/metrics.go +- Added `MembershipChannelMsgsSentTotal`, `MembershipChannelMsgsRecvTotal`, + `MembershipChannelConnectionsActive`, `MembershipChannelErrorsTotal` with corresponding + Inc/Dec methods, wired into `channel.go`'s send/receive/connect/error paths. + +### internal/membership/channel_test.go (new file) +- `TestChannelManager_EstablishForPeer_DialDirectionIsDeterministic` +- `TestChannelManager_SendRecv_RoundTrip` +- `TestChannelManager_Send_UnknownPeer` +- `TestChannelManager_Close_TearsDownPeer` +- `TestChannelManager_EstablishForPeer_IgnoresSelfAndEmpty` +- `TestChannelManager_ConcurrentSend` +- `TestChannelManager_StartUsesConfiguredPort` +- Uses `127.0.0.1` and `::1` as the two test node addresses (macOS does not auto-bind + `127.0.0.2`-style secondary loopback addresses without an interface alias, unlike Linux's full + `127.0.0.0/8` range). + +### internal/membership/delegate_test.go +- `TestStateDelegate_UseChannel` (table test: nil cluster / nil config / disabled / enabled-no-manager / enabled-with-manager) +- `TestStateDelegate_HandleChannelBlockWithExpiresAt`, `TestStateDelegate_HandleChannelUnblock` +- `TestStateDelegate_QueueBlockEvent_ViaChannel_NoPanic` +- **New**: `TestStateDelegate_QueueBlockEvent_LegacyPath_LogsWarn`, + `TestStateDelegate_QueueUnblockEvent_LegacyPath_LogsWarn` — capture logger output via a + `bytes.Buffer`-backed `slog.TextHandler` and assert the legacy-path `WARN` message and + `event_type` are present. +- **New**: `TestStateDelegate_QueueBlockEvent_ChannelPath_NoLegacyWarn` — asserts no legacy-path + `WARN` is logged when the channel path is used instead. + +### internal/membership/membership_test.go +- **New**: `TestCluster_WaitForChannelsReady_NoChannelManager_NoOp` — verifies the readiness gate + returns immediately when the feature is disabled. +- **New**: `TestCluster_WaitForChannelsReady_NoPeers_ReturnsImmediately` — verifies the gate + returns immediately for a single-node cluster (nothing to wait for), using a real, started + `ChannelManager`. + +### internal/config/config_test.go +- `clearEnvVars()` updated to `DRL_MEMBERSHIP_USE_HIPRIO_PERSISTENT_CHANNEL` / + `DRL_MEMBERSHIP_HIPRIO_CHANNEL_PORT`. +- `TestLoad_DefaultsOnly` asserts `cfg.Membership.UseHiPrioPersistentChannel == true` (new default). +- `TestLoad_EnvironmentOverrides` sets the env var to `"false"` (opposite of the default) to prove + the override actually takes effect, rather than incidentally matching the default. +- `TestValidate_HiPrioChannelPort_DefaultsWhenZero`, `TestValidate_HiPrioChannelPort_OutOfRange`, + `TestValidate_HiPrioChannelPort_CollidesWithMembershipPort`, + `TestValidate_HiPrioChannelPort_CollisionOnlyEnforcedWhenChannelEnabled`. +- `TestConfig_HiPrioPersistentChannelFromKDL` — parses a KDL snippet with + `use-hiprio-persistent-channel true` / `hiprio-channel-port 8956` and asserts both fields. + +## Test Results +- `go build ./...` — clean. +- `go vet ./...` — clean. +- `mise run lint` — 0 issues. +- `mise run test` (`gotestsum` + `tparse`, all packages) — all suites PASS, including + `internal/config` (137 tests) and `internal/membership` (71 tests). +- `go test ./internal/membership/... -race` — clean, no data races introduced by this milestone. + (A pre-existing, unrelated data race in `internal/api/handlers_blocklist_test.go`'s + `mockBroadcaster.QueueBlockEvent` was confirmed via `git stash`/`git stash pop` to exist on a + clean `main` checkout prior to this work, and is out of scope here.) + +## Verification Checklist +- [x] `go build ./...` compiles +- [x] `go vet ./...` clean +- [x] `mise run lint` — 0 issues +- [x] `mise run test` — all packages pass +- [x] `go test ./internal/membership/... -race` — no races +- [x] Env var `DRL_MEMBERSHIP_USE_HIPRIO_PERSISTENT_CHANNEL` (default `true`) wired and tested +- [x] KDL fields `use-hiprio-persistent-channel` / `hiprio-channel-port` wired and tested +- [x] Persistent channel listens/dials on port `7956` by default +- [x] Single bidirectional gRPC connection per node pair (deterministic dial direction) +- [x] Node Join logs `INFO` on channel establishment +- [x] Node Leave logs `INFO` and tears down the channel/client connections +- [x] Joiner readiness gated on persistent channel connections to all known peers (bounded by a + 30s timeout, degrading gracefully to availability over strict consistency) +- [x] Legacy on-demand TCP path logs `WARN` whenever it is used instead of the persistent channel +- [x] Temporary transport failures (dial/stream/send) logged as `WARN`, never fail the caller +- [x] Unit tests cover `ChannelManager` lifecycle, `StateDelegate` routing/gating, legacy-path + warning, and joiner readiness gating + +## Known Limitations +- The persistent gRPC channel does not currently use TLS (matches the existing internal-cluster + trust model of the rest of the membership/gRPC transport in this codebase; not a regression + introduced by this milestone). +- `waitForChannelsReady()`'s 30s timeout is currently a package-level constant rather than a + configurable value; the milestone spec does not call for a dedicated timeout knob, so this was + not added as a new config surface. diff --git a/ci/scripts/proto-gen.sh b/ci/scripts/proto-gen.sh index 53e4d69..eef0aa0 100755 --- a/ci/scripts/proto-gen.sh +++ b/ci/scripts/proto-gen.sh @@ -10,11 +10,21 @@ if ! command -v protoc-gen-go &>/dev/null; then go install google.golang.org/protobuf/cmd/protoc-gen-go@latest fi +# Install protoc-gen-go-grpc if not already available (needed for the +# persistent gRPC channel service in channel.proto). +if ! command -v protoc-gen-go-grpc &>/dev/null; then + echo "Installing protoc-gen-go-grpc..." + go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest +fi + echo "Generating Go code from protobuf definitions..." protoc \ --proto_path="${ROOT_DIR}" \ --go_out="${ROOT_DIR}" \ --go_opt=paths=source_relative \ - internal/proto/accounting.proto + --go-grpc_out="${ROOT_DIR}" \ + --go-grpc_opt=paths=source_relative \ + internal/proto/accounting.proto \ + internal/proto/channel.proto echo "Protobuf generation complete." diff --git a/deployments/k8s-embedded-proxy/README.md b/deployments/k8s-embedded-proxy/README.md new file mode 100644 index 0000000..dd3a783 --- /dev/null +++ b/deployments/k8s-embedded-proxy/README.md @@ -0,0 +1,134 @@ +# K8s Embedded Proxy Deployment + +This Kustomize base deploys DRL in **embedded-proxy sidecar** mode on Kubernetes. DRL runs in the same pod as +echo-server and acts as the edge reverse proxy, replacing Envoy entirely — there is no Envoy container in this topology. + +## Architecture + +``` + ┌─ drl namespace ─────────────────────────────────────────────┐ + │ │ + Client ──HTTP──► │ Service (echo-server:80) │ + │ │ │ + │ ▼ │ + │ ┌─── Pod (×3) ──────────────────────────────────────────┐ │ + │ │ │ │ + │ │ ┌─ drl sidecar ──────────────────────────────────┐ │ │ + │ │ │ embedded proxy :8080 │ │ │ + │ │ │ 1. Auth0 OIDC Bearer token validation │ │ │ + │ │ │ 2. Rate-limit blocklist check │ │ │ + │ │ │ 3. Async P2P accounting │ │ │ + │ │ │ 4. Forward → localhost:18080 │ │ │ + │ │ │ │ │ │ + │ │ │ gRPC :8081 · metrics :9091 · gossip :7946 │ │ │ + │ │ └────────────────────────────────────────────────┘ │ │ + │ │ │ localhost │ │ + │ │ ┌─ echo-server ─────────▼──────────────────────┐ │ │ + │ │ │ go-httpbin :18080 │ │ │ + │ │ └──────────────────────────────────────────────┘ │ │ + │ └────────────────────────────────────────────────────────┘ │ + │ │ gossip (drl-headless) │ + │ └──────► other pods │ + └──────────────────────────────────────────────────────────────┘ +``` + +## Key differences from k8s-sidecar + +| Aspect | k8s-sidecar | k8s-embedded-proxy | +|-------------------|--------------------------------|------------------------------------| +| Ingress proxy | Envoy sidecar | DRL embedded proxy | +| Auth enforcement | Envoy ext_authz → DRL gRPC | Auth0 OIDC JWT validation in DRL | +| DRL gRPC :8081 | Used by Envoy (ext_authz) | Available but not used for ingress | +| Pod composition | echo-server + envoy + drl | echo-server + drl | +| echo-server port | :8080 | :18080 (loopback only) | +| User-facing entry | Envoy :10000 → Service port 80 | DRL :8080 → Service port 80 | + +## Services + +| Service | Type | Port | Selector | Purpose | +|----------------|-----------|-----------|--------------------|---------------------------------------| +| `echo-server` | ClusterIP | 80 → 8080 | `app: echo-server` | User-facing HTTP endpoint (DRL proxy) | +| `drl-headless` | Headless | 7946 | `app: echo-server` | Memberlist peer discovery | + +## Auth0 OIDC configuration + +DRL acts as an OIDC **Resource Server** — it validates Bearer tokens but never issues them. Clients obtain a JWT access +token from Auth0 and present it as `Authorization: Bearer ` on every request. + +| Field | Value | +|----------------------|-----------------------------------------------------------------------------| +| OpenID Configuration | `https://dev-xxwr5gxe.eu.auth0.com/.well-known/openid-configuration` | +| JWKS URI | `https://dev-xxwr5gxe.eu.auth0.com/.well-known/jwks.json` (auto-discovered) | +| Token endpoint | `https://dev-xxwr5gxe.eu.auth0.com/oauth/token` | +| Issuer | `https://dev-xxwr5gxe.eu.auth0.com/` | + +> **Action required:** Open `base/configmap.yaml` and replace the `audience` +> value (`https://echo-server.example.com/api`) with the **Identifier** of the +> API you registered under *Auth0 Dashboard → Applications → APIs*. +> This value must match the `aud` claim in access tokens issued by Auth0. + +## Prerequisites + +Create the `drl` namespace and required Secret before applying: + +```bash +kubectl create namespace drl + +kubectl -n drl create secret generic drl-secrets \ + --from-literal=private-api-key="" \ + --from-literal=membership-primary-key="" +``` + +## Applying + +```bash +kubectl apply -k deployments/k8s-embedded-proxy/base +``` + +Verify rollout: + +```bash +kubectl -n drl rollout status deployment/echo-server +``` + +## Accessing the service + +```bash +# Port-forward DRL's embedded proxy +kubectl -n drl port-forward svc/echo-server 8080:80 + +# Obtain an Auth0 access token via client credentials +TOKEN=$(curl -s -X POST https://dev-xxwr5gxe.eu.auth0.com/oauth/token \ + -H "Content-Type: application/json" \ + -d '{ + "client_id": "", + "client_secret": "", + "audience": "https://echo-server.example.com/api", + "grant_type": "client_credentials" + }' | jq -r .access_token) + +# Call the protected workload +curl -H "Authorization: Bearer $TOKEN" http://localhost:8080/anything + +# Request without a token returns 401 +curl -v http://localhost:8080/anything +``` + +## Runtime rule overrides + +Override accounting rules per-deployment at runtime without a config reload: + +```bash +kubectl -n drl set env deployment/echo-server \ + 'DRL_RULE_catch-all_JSON={"path-prefix":"/","limit":200,"per":"minute"}' +``` + +## Customising OIDC per environment + +To override the full embedded-proxy host configuration via environment variable (useful for overlays that target staging +vs production Auth0 tenants): + +```bash +kubectl -n drl set env deployment/echo-server \ + 'DRL_EMBEDDED_PROXY_HOSTS_JSON=[{"hostname":"echo-server","oidc":{"issuer":"https://prod.eu.auth0.com/","audience":"https://api.prod.example.com"},"routes":{"routes":[{"prefix":"/","upstream":"http://127.0.0.1:18080","require-auth":true}]}}]' +``` diff --git a/deployments/k8s-embedded-proxy/base/configmap.yaml b/deployments/k8s-embedded-proxy/base/configmap.yaml new file mode 100644 index 0000000..5cc0363 --- /dev/null +++ b/deployments/k8s-embedded-proxy/base/configmap.yaml @@ -0,0 +1,114 @@ +--- +# DRL configuration — embedded-proxy mode, no Envoy required. +apiVersion: v1 +kind: ConfigMap +metadata: + name: drl-config + namespace: drl +data: + config.kdl: | + // DRL - Distributed Rate Limiter Configuration + // K8s Embedded Proxy deployment + // + // DRL's built-in HTTP reverse proxy replaces Envoy entirely. + // DRL runs as a sidecar alongside echo-server in the same pod. + // Traffic enters DRL on :8080, is authenticated via Auth0 OIDC, rate-limited, + // then forwarded to echo-server on localhost:18080. + // Peer discovery uses the headless K8s Service "drl-headless" which + // resolves to A records for every pod IP. + + listen { + grpc ":8081" + metrics ":9091" + } + + membership { + service-name "drl-headless.drl.svc.cluster.local" + port 7946 + bind-addr "0.0.0.0" + startup-delay "3s" + } + + logging { + level "debug" + format "json" + } + + internal-api { + enabled true + address ":8082" + } + + cache { + blocklist-size-mb 64 + accounting-size-mb 128 + sync-timeout-seconds 30 + blocklist-default-ttl-seconds 300 + } + + accounting { + settings { + algorithm "sliding-window" + retry-after-type "delay-seconds" + use-x-forwarded-for true + use-x-forwarded-for-direction "right" + use-x-forwarded-for-index 0 + } + rules { + "catch-all" { + path-prefix "/" + limit 100 + per "minute" + } + "anything" { + path-prefix "/anything" + limit 10 + headers "authorization" + redactions { + "Authorization" "^(Bearer .{0,3}).*$" + } + per "minute" + } + } + } + + // Embedded reverse proxy — DRL is a sidecar, so the upstream is localhost. + // DRL listens on :8080 and validates Auth0 Bearer tokens before + // rate-limiting and forwarding to echo-server on the loopback interface. + embedded-proxy { + enabled true + listen ":8080" + + tls { + enabled false + cert "" + key "" + } + + // Virtual host "echo-server": handles all inbound traffic. + // The hostname is a logical label used for metrics and OIDC verifier + // lookup; it does not perform HTTP Host-header matching. + host "echo-server" { + // OIDC Resource Server settings — DRL validates Auth0 Bearer JWTs. + // The issuer URL must match the "iss" claim in your access tokens + // (Auth0 always appends a trailing slash). + // DRL auto-discovers the JWKS URI via the OpenID Connect discovery + // endpoint at {issuer}/.well-known/openid-configuration. + // + // audience: MUST be set to your Auth0 API Identifier — the value + // you configured under Auth0 Dashboard → Applications → APIs → + // Identifier (e.g. "https://api.my-service.example.com"). + // Requests whose token "aud" claim does not match will receive 403. + oidc { + issuer "https://dev-xxwr5gxe.eu.auth0.com/" + audience "https://drl.lan/api/v2/" + jwks-cache-ttl "5m" + } + routes { + route "/" { + upstream "http://127.0.0.1:18080" + require-auth true + } + } + } + } diff --git a/deployments/k8s-embedded-proxy/base/deployment.yaml b/deployments/k8s-embedded-proxy/base/deployment.yaml new file mode 100644 index 0000000..473ac9a --- /dev/null +++ b/deployments/k8s-embedded-proxy/base/deployment.yaml @@ -0,0 +1,114 @@ +--- +# Sidecar pod: echo-server (workload) + drl (embedded proxy). +# DRL listens on :8080 and proxies authenticated, rate-limited traffic to +# echo-server on localhost:18080. Pods discover each other via the +# drl-headless headless Service for P2P blocklist gossip. +apiVersion: apps/v1 +kind: Deployment +metadata: + name: echo-server + namespace: drl + labels: + app: echo-server +spec: + replicas: 3 + selector: + matchLabels: + app: echo-server + template: + metadata: + labels: + app: echo-server + spec: + containers: + # ── echo-server ──────────────────────────────────────────────────────── + - name: echo-server + image: mccutchen/go-httpbin:latest + env: + - name: PORT + value: "18080" + ports: + - name: http + containerPort: 18080 + protocol: TCP + resources: + requests: + cpu: "100m" + memory: "64Mi" + limits: + cpu: "200m" + memory: "128Mi" + + # ── drl sidecar ──────────────────────────────────────────────────────── + # DRL acts as the embedded reverse proxy on :8080. + # All inbound traffic passes through Auth0 OIDC validation and rate-limit + # accounting before being forwarded to echo-server on localhost:18080. + - name: drl + image: ghcr.io/gchiesa/drl:latest + args: [] + env: + - name: DRL_CONFIG_PATH + value: /etc/drl/config.kdl + - name: DRL_PRIVATE_API_KEY + valueFrom: + secretKeyRef: + name: drl-secrets + key: private-api-key + - name: DRL_MEMBERSHIP_PRIMARY_KEY + valueFrom: + secretKeyRef: + name: drl-secrets + key: membership-primary-key + - name: DRL_MEMBERSHIP_SECONDARY_KEYS + valueFrom: + secretKeyRef: + name: drl-secrets + key: membership-secondary-keys + optional: true + ports: + - name: http + containerPort: 8080 + protocol: TCP + - name: grpc + containerPort: 8081 + protocol: TCP + - name: internal-api + containerPort: 8082 + protocol: TCP + - name: metrics + containerPort: 9091 + protocol: TCP + - name: memberlist-tcp + containerPort: 7946 + protocol: TCP + - name: memberlist-udp + containerPort: 7946 + protocol: UDP + volumeMounts: + - name: drl-config + mountPath: /etc/drl + readOnly: true + livenessProbe: + httpGet: + path: /health + port: 9091 + initialDelaySeconds: 15 + periodSeconds: 10 + readinessProbe: + httpGet: + path: /health + port: 9091 + initialDelaySeconds: 10 + periodSeconds: 5 + resources: + requests: + cpu: "100m" + memory: "128Mi" + limits: + cpu: "300m" + memory: "256Mi" + + volumes: + - name: drl-config + configMap: + name: drl-config diff --git a/deployments/k8s-embedded-proxy/base/kustomization.yaml b/deployments/k8s-embedded-proxy/base/kustomization.yaml new file mode 100644 index 0000000..fddb55a --- /dev/null +++ b/deployments/k8s-embedded-proxy/base/kustomization.yaml @@ -0,0 +1,21 @@ +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization + +namespace: drl + +resources: + - configmap.yaml + - deployment.yaml + - service.yaml + +# Replace with your actual DRL image +images: + - name: ghcr.io/gchiesa/drl + newTag: latest + +# Generate the DRL secrets from a local .env file (for local dev only). +# In production use an external secrets operator (ESO, Vault, Sealed Secrets). +# secretGenerator: +# - name: drl-secrets +# envs: +# - drl-secrets.env diff --git a/deployments/k8s-embedded-proxy/base/service.yaml b/deployments/k8s-embedded-proxy/base/service.yaml new file mode 100644 index 0000000..c7c5d59 --- /dev/null +++ b/deployments/k8s-embedded-proxy/base/service.yaml @@ -0,0 +1,43 @@ +--- +# ClusterIP service — exposes DRL's embedded proxy HTTP port for ingress/internal access. +# This is the user-facing endpoint for the echo-server workload. +apiVersion: v1 +kind: Service +metadata: + name: echo-server + namespace: drl + labels: + app: echo-server +spec: + selector: + app: echo-server + ports: + - name: http + port: 80 + targetPort: 8080 + protocol: TCP + type: LoadBalancer + loadBalancerIP: "192.168.78.220" +--- +# Headless service — used by DRL memberlist for peer discovery. +# Resolves to A records for every pod IP in the deployment. +apiVersion: v1 +kind: Service +metadata: + name: drl-headless + namespace: drl + labels: + app: echo-server +spec: + clusterIP: None + selector: + app: echo-server + ports: + - name: memberlist-tcp + port: 7946 + targetPort: 7946 + protocol: TCP + - name: memberlist-udp + port: 7946 + targetPort: 7946 + protocol: UDP diff --git a/internal/cmd/cluster.go b/internal/cmd/cluster.go index dc717b0..a3bc3a0 100644 --- a/internal/cmd/cluster.go +++ b/internal/cmd/cluster.go @@ -61,6 +61,25 @@ func newCluster(cfg *config.Config, localIP string, cacheManager *cache.Manager, os.Exit(1) } + // Start the persistent gRPC channel for hi-priority (block/unblock) + // event propagation, when enabled via config/DRL_MEMBERSHIP_USE_HIPRIO_PERSISTENT_CHANNEL. + if cfg.Membership.UseHiPrioPersistentChannel { + channelManager := membership.NewChannelManager(membership.ChannelManagerConfig{ + LocalAddr: localIP, + Port: cfg.Membership.HiPrioChannelPort, + Handler: stateDelegate, + Metrics: metricsManager, + Logger: log, + }) + if err := channelManager.Start(); err != nil { + log.Error("failed to start persistent gRPC channel", "error", err) + cacheManager.Close() + os.Exit(1) + } + cluster.SetChannelManager(channelManager) + log.Info("persistent gRPC channel enabled", "port", cfg.Membership.HiPrioChannelPort) + } + // Join the cluster in the background go func() { if err := cluster.JoinCluster(); err != nil { diff --git a/internal/config/config.go b/internal/config/config.go index fecabbd..c0f48f9 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -139,6 +139,15 @@ type MembershipConfig struct { // valid AES lengths (16, 24, or 32 bytes). // Override via env: DRL_MEMBERSHIP_PRIMARY_KEY + DRL_MEMBERSHIP_SECONDARY_KEYS SecretKeys []string `kdl:"secret-keys" json:"-"` // never serialised — contains sensitive key material + // UseHiPrioPersistentChannel enables the persistent gRPC channel for + // hi-priority (block/unblock) event propagation between cluster members, + // replacing the on-demand memberlist SendReliable (TCP) path. The + // channel is established once, when a node joins the cluster, instead + // of a new connection per event. + UseHiPrioPersistentChannel bool `kdl:"use-hiprio-persistent-channel" env:"USE_HIPRIO_PERSISTENT_CHANNEL" json:"use-hiprio-persistent-channel"` + // HiPrioChannelPort is the TCP port the persistent gRPC channel listens + // on and dials peers at. Defaults to 7956 when unset. + HiPrioChannelPort int `kdl:"hiprio-channel-port" env:"HIPRIO_CHANNEL_PORT" json:"hiprio-channel-port"` } // LoggingConfig holds logging configuration @@ -470,6 +479,17 @@ func (c *Config) Validate() error { if c.Membership.GossipNodes == 0 { c.Membership.GossipNodes = 5 } + if c.Membership.HiPrioChannelPort == 0 { + c.Membership.HiPrioChannelPort = 7956 + } + + // Validate persistent gRPC channel port + if c.Membership.HiPrioChannelPort < 1 || c.Membership.HiPrioChannelPort > 65535 { + errs = append(errs, fmt.Sprintf("membership.hiprio-channel-port must be between 1 and 65535, got %d", c.Membership.HiPrioChannelPort)) + } + if c.Membership.UseHiPrioPersistentChannel && c.Membership.HiPrioChannelPort == c.Membership.Port { + errs = append(errs, fmt.Sprintf("membership.hiprio-channel-port (%d) must differ from membership.port (%d)", c.Membership.HiPrioChannelPort, c.Membership.Port)) + } // Apply accounting settings defaults if c.Accounting.Settings.Algorithm == "" { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index ece2585..7250b44 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -40,6 +40,8 @@ func TestLoad_DefaultsOnly(t *testing.T) { assert.Equal(t, int64(128), cfg.Cache.AccountingSizeMB) assert.Equal(t, 30, cfg.Cache.SyncTimeoutSeconds) assert.Equal(t, 300, cfg.Cache.BlocklistDefaultTTLSeconds) + assert.True(t, cfg.Membership.UseHiPrioPersistentChannel) + assert.Equal(t, 7956, cfg.Membership.HiPrioChannelPort) } @@ -133,6 +135,10 @@ func TestLoad_EnvironmentOverrides(t *testing.T) { t.Setenv("DRL_LOGGING_FORMAT", "text") t.Setenv("DRL_INTERNAL_API_ENABLED", "false") t.Setenv("DRL_INTERNAL_API_ADDRESS", ":7002") + // Default is true — override to false to prove the env var actually + // takes effect rather than just matching the default. + t.Setenv("DRL_MEMBERSHIP_USE_HIPRIO_PERSISTENT_CHANNEL", "false") + t.Setenv("DRL_MEMBERSHIP_HIPRIO_CHANNEL_PORT", "7999") cfg, err := Load("") require.NoError(t, err) @@ -148,6 +154,8 @@ func TestLoad_EnvironmentOverrides(t *testing.T) { assert.Equal(t, "text", cfg.Logging.Format) assert.False(t, cfg.InternalAPI.Enabled) assert.Equal(t, ":7002", cfg.InternalAPI.Address) + assert.False(t, cfg.Membership.UseHiPrioPersistentChannel) + assert.Equal(t, 7999, cfg.Membership.HiPrioChannelPort) } func TestLoad_EnvironmentOverridesKDL(t *testing.T) { @@ -246,6 +254,134 @@ func TestValidate_ValidConfig(t *testing.T) { assert.NoError(t, err) } +func TestValidate_HiPrioChannelPort_DefaultsWhenZero(t *testing.T) { + cfg := &Config{ + Listen: ListenConfig{ + GRPC: ":8081", + Metrics: ":9091", + }, + Membership: MembershipConfig{ + ServiceName: "drl", + Port: 7946, + BindAddr: "0.0.0.0", + }, + Logging: LoggingConfig{ + Level: "info", + Format: "json", + }, + Cache: CacheConfig{ + BlocklistSizeMB: 64, + AccountingSizeMB: 128, + SyncTimeoutSeconds: 30, + BlocklistDefaultTTLSeconds: 3600, + }, + } + + err := cfg.Validate() + require.NoError(t, err) + assert.Equal(t, 7956, cfg.Membership.HiPrioChannelPort) +} + +func TestValidate_HiPrioChannelPort_OutOfRange(t *testing.T) { + tests := []struct { + name string + port int + }{ + {"port negative", -1}, + {"port too high", 65536}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := &Config{ + Listen: ListenConfig{ + GRPC: ":8081", + Metrics: ":9091", + }, + Membership: MembershipConfig{ + ServiceName: "drl", + Port: 7946, + BindAddr: "0.0.0.0", + HiPrioChannelPort: tt.port, + }, + Logging: LoggingConfig{ + Level: "info", + Format: "json", + }, + Cache: CacheConfig{ + BlocklistSizeMB: 64, + AccountingSizeMB: 128, + SyncTimeoutSeconds: 30, + BlocklistDefaultTTLSeconds: 3600, + }, + } + + err := cfg.Validate() + assert.Error(t, err) + assert.Contains(t, err.Error(), "membership.hiprio-channel-port must be between 1 and 65535") + }) + } +} + +func TestValidate_HiPrioChannelPort_CollidesWithMembershipPort(t *testing.T) { + cfg := &Config{ + Listen: ListenConfig{ + GRPC: ":8081", + Metrics: ":9091", + }, + Membership: MembershipConfig{ + ServiceName: "drl", + Port: 7946, + BindAddr: "0.0.0.0", + UseHiPrioPersistentChannel: true, + HiPrioChannelPort: 7946, + }, + Logging: LoggingConfig{ + Level: "info", + Format: "json", + }, + Cache: CacheConfig{ + BlocklistSizeMB: 64, + AccountingSizeMB: 128, + SyncTimeoutSeconds: 30, + BlocklistDefaultTTLSeconds: 3600, + }, + } + + err := cfg.Validate() + assert.Error(t, err) + assert.Contains(t, err.Error(), "membership.hiprio-channel-port (7946) must differ from membership.port (7946)") +} + +func TestValidate_HiPrioChannelPort_CollisionOnlyEnforcedWhenChannelEnabled(t *testing.T) { + cfg := &Config{ + Listen: ListenConfig{ + GRPC: ":8081", + Metrics: ":9091", + }, + Membership: MembershipConfig{ + ServiceName: "drl", + Port: 7946, + BindAddr: "0.0.0.0", + UseHiPrioPersistentChannel: false, + HiPrioChannelPort: 7946, + }, + Logging: LoggingConfig{ + Level: "info", + Format: "json", + }, + Cache: CacheConfig{ + BlocklistSizeMB: 64, + AccountingSizeMB: 128, + SyncTimeoutSeconds: 30, + BlocklistDefaultTTLSeconds: 3600, + }, + } + + err := cfg.Validate() + assert.NoError(t, err) +} + func TestValidate_EmptyServiceName(t *testing.T) { cfg := &Config{ Listen: ListenConfig{ @@ -873,6 +1009,31 @@ membership { assert.NotNil(t, cfg) } +func TestConfig_HiPrioPersistentChannelFromKDL(t *testing.T) { + clearEnvVars(t) + + tmpDir := t.TempDir() + configPath := filepath.Join(tmpDir, "channel.kdl") + + kdlConfig := ` +membership { + service-name "test" + port 7946 + bind-addr "0.0.0.0" + use-hiprio-persistent-channel true + hiprio-channel-port 8956 +} +` + err := os.WriteFile(configPath, []byte(kdlConfig), 0644) + require.NoError(t, err) + + cfg, err := Load(configPath) + require.NoError(t, err) + + assert.True(t, cfg.Membership.UseHiPrioPersistentChannel) + assert.Equal(t, 8956, cfg.Membership.HiPrioChannelPort) +} + func TestAccountingRule_WindowDuration(t *testing.T) { tests := []struct { name string @@ -1384,6 +1545,8 @@ func clearEnvVars(t *testing.T) { "DRL_CACHE_BLOCKLIST_DEFAULT_TTL_SECONDS", "DRL_MEMBERSHIP_PRIMARY_KEY", "DRL_MEMBERSHIP_SECONDARY_KEYS", + "DRL_MEMBERSHIP_USE_HIPRIO_PERSISTENT_CHANNEL", + "DRL_MEMBERSHIP_HIPRIO_CHANNEL_PORT", "DRL_ACCOUNTING_SETTINGS_ALGORITHM", "DRL_ACCOUNTING_SETTINGS_CAPACITY", "DRL_ACCOUNTING_SETTINGS_REFILL_RATE", diff --git a/internal/config/resources/default.kdl b/internal/config/resources/default.kdl index 7799aaf..65ee09e 100644 --- a/internal/config/resources/default.kdl +++ b/internal/config/resources/default.kdl @@ -13,6 +13,11 @@ membership { startup-delay "3s" gossip-interval "50ms" gossip-nodes 5 + // Persistent gRPC channel for hi-priority (block/unblock) propagation. + // Enabled by default; replaces the on-demand memberlist SendReliable + // (TCP) path with a long-lived, bidirectional gRPC stream per peer pair. + use-hiprio-persistent-channel true + hiprio-channel-port 7956 } logging { diff --git a/internal/membership/channel.go b/internal/membership/channel.go new file mode 100644 index 0000000..f1d874a --- /dev/null +++ b/internal/membership/channel.go @@ -0,0 +1,490 @@ +package membership + +import ( + "context" + "errors" + "fmt" + "log/slog" + "net" + "strconv" + "sync" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/keepalive" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + + "github.com/gchiesa/drl/internal/metrics" + drlproto "github.com/gchiesa/drl/internal/proto" +) + +// nodeAddrMetadataKey is the gRPC metadata key used by the dialer to +// identify itself to the acceptor when opening a PersistentChannel stream. +// It is required because the acceptor must key its peer map by the node's +// cluster (memberlist) address, not by the ephemeral TCP source address of +// the dialing connection. +const nodeAddrMetadataKey = "x-drl-node-addr" + +// channelSendBuffer bounds the number of outbound messages queued per peer +// before Send() starts reporting back-pressure errors. Hi-priority events +// are infrequent relative to accounting traffic, so a small buffer suffices. +const channelSendBuffer = 256 + +// duplexStream is the minimal interface shared by the client-side +// (grpc.BidiStreamingClient) and server-side (grpc.BidiStreamingServer) +// handles of a PersistentChannel stream, letting the rest of this file treat +// both ends identically. +type duplexStream interface { + Send(*drlproto.ChannelMessage) error + Recv() (*drlproto.ChannelMessage, error) +} + +// channelMessageHandler processes hi-priority events received over the +// persistent gRPC channel. *StateDelegate implements this interface. +// The methods are unexported deliberately: only types within package +// membership are meant to implement this handler. +type channelMessageHandler interface { + handleChannelBlockWithExpiresAt(evt *drlproto.BlockEventWithExpiresAt) + handleChannelUnblock(evt *drlproto.UnblockEvent) +} + +// peerChannel tracks the live duplex stream and outbound queue for a single +// peer connection. +type peerChannel struct { + addr string + stream duplexStream + sendCh chan *drlproto.ChannelMessage + + // conn is non-nil only when the local node is the dialer for this peer. + // Closing it force-terminates the underlying stream immediately. + conn *grpc.ClientConn + + // closeSignal, when set, is used to unblock the acceptor-side Stream() + // handler goroutine when the peer is torn down proactively (e.g. on + // NotifyLeave) rather than by the remote end closing the connection. + closeSignal func() + + closeOnce sync.Once +} + +// ChannelManagerConfig holds configuration for creating a ChannelManager. +type ChannelManagerConfig struct { + // LocalAddr is this node's cluster (memberlist) address. It is used both + // to identify this node to peers and to deterministically decide dial + // direction (see ChannelManager doc comment). + LocalAddr string + // Port is the TCP port the persistent channel gRPC server listens on, + // and the port used to dial peers. + Port int + // Handler receives decoded hi-priority events. Normally the cluster's + // *StateDelegate. + Handler channelMessageHandler + Metrics *metrics.Metrics + Logger *slog.Logger +} + +// ChannelManager manages the persistent, bidirectional gRPC channel used to +// propagate hi-priority (block/unblock) events between cluster members, +// replacing the on-demand memberlist SendReliable path when enabled via +// DRL_MEMBERSHIP_USE_HIPRIO_PERSISTENT_CHANNEL. +// +// Design decision: rather than each ordered pair of nodes independently +// dialing the other (2 TCP/HTTP2 connections per pair, "spaghetti web"), +// ChannelManager establishes exactly ONE bidirectional gRPC stream per +// unordered pair of nodes. gRPC streams are natively full-duplex, so a +// single stream is sufficient for both peers to Send and Recv events +// concurrently — a second connection would add file descriptors, TCP/TLS +// handshake overhead, and keepalive traffic without adding capability. +// +// The dial direction is decided deterministically, without any coordination +// round-trip: the node whose address sorts lexicographically smaller dials +// the other (see EstablishForPeer). Both sides independently reach the same +// conclusion when they observe a memberlist join event, so exactly one side +// dials and the other passively accepts — halving connection count +// cluster-wide compared to a two-connections-per-pair model. +type ChannelManager struct { + localAddr string + port int + handler channelMessageHandler + metrics *metrics.Metrics + logger *slog.Logger + + grpcServer *grpc.Server + listener net.Listener + + mu sync.RWMutex + peers map[string]*peerChannel + + drlproto.UnimplementedPersistentChannelServer +} + +// NewChannelManager creates a new ChannelManager. Start must be called to +// begin listening for inbound peer connections. +func NewChannelManager(cfg ChannelManagerConfig) *ChannelManager { + return &ChannelManager{ + localAddr: cfg.LocalAddr, + port: cfg.Port, + handler: cfg.Handler, + metrics: cfg.Metrics, + logger: cfg.Logger, + peers: make(map[string]*peerChannel), + } +} + +// Start begins listening for inbound persistent channel connections from +// peers. It does not block. +func (cm *ChannelManager) Start() error { + lis, err := net.Listen("tcp", net.JoinHostPort(cm.localAddr, strconv.Itoa(cm.port))) + if err != nil { + return fmt.Errorf("failed to listen on persistent channel port %d: %w", cm.port, err) + } + cm.listener = lis + + gs := grpc.NewServer( + grpc.KeepaliveParams(keepalive.ServerParameters{ + Time: 15 * time.Second, + Timeout: 5 * time.Second, + }), + grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ + MinTime: 10 * time.Second, + PermitWithoutStream: true, + }), + ) + cm.grpcServer = gs + drlproto.RegisterPersistentChannelServer(gs, cm) + + cm.logger.Info("persistent gRPC channel server listening", + "local_addr", cm.localAddr, + "port", cm.port, + ) + + go func() { + if err := gs.Serve(lis); err != nil && !errors.Is(err, grpc.ErrServerStopped) { + cm.logger.Error("persistent channel gRPC server error", "error", err) + } + }() + + return nil +} + +// Stop tears down all peer connections and stops the gRPC server. Intended +// to be called once, when this node leaves the cluster. +func (cm *ChannelManager) Stop() { + cm.mu.Lock() + peers := make([]*peerChannel, 0, len(cm.peers)) + for _, p := range cm.peers { + peers = append(peers, p) + } + cm.mu.Unlock() + + for _, p := range peers { + cm.teardownPeer(p.addr, p) + } + + if cm.grpcServer != nil { + cm.grpcServer.GracefulStop() + } + cm.logger.Info("persistent gRPC channel server stopped", "local_addr", cm.localAddr) +} + +// EstablishForPeer decides, deterministically, whether the local node should +// dial the given peer's persistent channel. It is intended to be called from +// the membership event delegate whenever a node joins the cluster (for every +// other known member, from both the joiner's and the existing members' +// perspective — memberlist delivers NotifyJoin symmetrically). Only the side +// whose address sorts smaller dials; the other side waits to accept the +// inbound connection. +func (cm *ChannelManager) EstablishForPeer(remoteAddr string) { + if remoteAddr == "" || remoteAddr == cm.localAddr { + return + } + if cm.localAddr >= remoteAddr { + // Passive side: the peer with the smaller address will dial us. + return + } + + cm.mu.RLock() + _, exists := cm.peers[remoteAddr] + cm.mu.RUnlock() + if exists { + return + } + + go cm.connect(remoteAddr) +} + +// connect dials the given peer and opens the PersistentChannel stream. +func (cm *ChannelManager) connect(remoteAddr string) { + target := net.JoinHostPort(remoteAddr, strconv.Itoa(cm.port)) + + conn, err := grpc.NewClient(target, + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithKeepaliveParams(keepalive.ClientParameters{ + Time: 15 * time.Second, + Timeout: 5 * time.Second, + PermitWithoutStream: true, + }), + ) + if err != nil { + cm.logger.Warn("failed to dial persistent channel peer", "peer_addr", remoteAddr, "error", err) + if cm.metrics != nil { + cm.metrics.IncMembershipChannelErrors() + } + return + } + + client := drlproto.NewPersistentChannelClient(conn) + md := metadata.Pairs(nodeAddrMetadataKey, cm.localAddr) + ctx := metadata.NewOutgoingContext(context.Background(), md) + + stream, err := client.Stream(ctx) + if err != nil { + cm.logger.Warn("failed to open persistent channel stream", "peer_addr", remoteAddr, "error", err) + if cm.metrics != nil { + cm.metrics.IncMembershipChannelErrors() + } + _ = conn.Close() + return + } + + p, registered := cm.registerPeer(remoteAddr, stream, conn, nil) + if !registered { + // Lost a race with a concurrent connect/accept for the same peer. + _ = conn.Close() + return + } + + cm.logger.Info("persistent channel established", + "peer_addr", remoteAddr, + "direction", "outbound", + ) + + cm.dialerRecvLoop(p) +} + +// Stream implements drlproto.PersistentChannelServer. It is invoked by +// grpc-go once per inbound RPC, i.e. once per peer that dials us. +func (cm *ChannelManager) Stream(stream drlproto.PersistentChannel_StreamServer) error { + remoteAddr := extractPeerAddr(stream.Context()) + if remoteAddr == "" { + return status.Error(codes.InvalidArgument, "missing "+nodeAddrMetadataKey+" metadata") + } + + done := make(chan struct{}) + p, registered := cm.registerPeer(remoteAddr, stream, nil, func() { + select { + case <-done: + default: + close(done) + } + }) + if !registered { + return status.Errorf(codes.AlreadyExists, "persistent channel to %s already established", remoteAddr) + } + + cm.logger.Info("persistent channel established", + "peer_addr", remoteAddr, + "direction", "inbound", + ) + + msgCh := make(chan *drlproto.ChannelMessage, 1) + errCh := make(chan error, 1) + go func() { + for { + msg, err := stream.Recv() + if err != nil { + errCh <- err + return + } + msgCh <- msg + } + }() + + for { + select { + case <-done: + cm.teardownPeer(remoteAddr, p) + return nil + case err := <-errCh: + cm.logger.Info("persistent channel closed", "peer_addr", remoteAddr, "error", err) + cm.teardownPeer(remoteAddr, p) + return nil + case msg := <-msgCh: + cm.dispatch(msg) + } + } +} + +// dialerRecvLoop continuously receives messages on a client-dialed stream +// until it errors out (peer closed / connection dropped), then tears down +// the peer entry. +func (cm *ChannelManager) dialerRecvLoop(p *peerChannel) { + for { + msg, err := p.stream.Recv() + if err != nil { + cm.logger.Info("persistent channel closed", "peer_addr", p.addr, "error", err) + cm.teardownPeer(p.addr, p) + return + } + cm.dispatch(msg) + } +} + +// registerPeer records a newly established stream in the peers map and +// starts its outbound writer goroutine. Returns (peer, false) if a peer +// entry for addr already exists (caller should discard the new stream). +func (cm *ChannelManager) registerPeer(addr string, stream duplexStream, conn *grpc.ClientConn, closeSignal func()) (*peerChannel, bool) { + cm.mu.Lock() + if _, exists := cm.peers[addr]; exists { + cm.mu.Unlock() + return nil, false + } + p := &peerChannel{ + addr: addr, + stream: stream, + sendCh: make(chan *drlproto.ChannelMessage, channelSendBuffer), + conn: conn, + closeSignal: closeSignal, + } + cm.peers[addr] = p + cm.mu.Unlock() + + if cm.metrics != nil { + cm.metrics.IncMembershipChannelConnections() + } + + go cm.writeLoop(p) + + return p, true +} + +// writeLoop drains a peer's outbound queue, serialising Send() calls (gRPC +// streams are not safe for concurrent Send from multiple goroutines). +func (cm *ChannelManager) writeLoop(p *peerChannel) { + for msg := range p.sendCh { + if err := p.stream.Send(msg); err != nil { + cm.logger.Warn("failed to send persistent channel message", "peer_addr", p.addr, "error", err) + if cm.metrics != nil { + cm.metrics.IncMembershipChannelErrors() + } + cm.teardownPeer(p.addr, p) + return + } + if cm.metrics != nil { + cm.metrics.IncMembershipChannelMsgsSent() + } + } +} + +// dispatch decodes the oneof content of a ChannelMessage and routes it to +// the configured handler. +func (cm *ChannelManager) dispatch(msg *drlproto.ChannelMessage) { + if msg == nil { + return + } + if cm.metrics != nil { + cm.metrics.IncMembershipChannelMsgsRecv() + } + if cm.handler == nil { + return + } + switch content := msg.Content.(type) { + case *drlproto.ChannelMessage_BlockWithExpiresAt: + cm.handler.handleChannelBlockWithExpiresAt(content.BlockWithExpiresAt) + case *drlproto.ChannelMessage_Unblock: + cm.handler.handleChannelUnblock(content.Unblock) + default: + cm.logger.Warn("received ChannelMessage with unknown content type") + } +} + +// Send queues msg for delivery to the peer at addr over its persistent +// channel. Returns an error (never blocking the caller for long) if no +// channel is established to that peer or if the outbound queue is full; +// callers should log and continue rather than fail the request path. +func (cm *ChannelManager) Send(addr string, msg *drlproto.ChannelMessage) error { + cm.mu.RLock() + p, ok := cm.peers[addr] + cm.mu.RUnlock() + if !ok { + return fmt.Errorf("no persistent channel established to peer %s", addr) + } + + select { + case p.sendCh <- msg: + return nil + default: + return fmt.Errorf("persistent channel send buffer full for peer %s", addr) + } +} + +// IsConnected reports whether a persistent channel is currently established +// to the given peer address. +func (cm *ChannelManager) IsConnected(addr string) bool { + cm.mu.RLock() + defer cm.mu.RUnlock() + _, ok := cm.peers[addr] + return ok +} + +// PeerCount returns the number of currently established peer channels. +func (cm *ChannelManager) PeerCount() int { + cm.mu.RLock() + defer cm.mu.RUnlock() + return len(cm.peers) +} + +// Close tears down the persistent channel to the given peer, if one exists. +// Intended to be called from the membership event delegate when a peer +// leaves the cluster. +func (cm *ChannelManager) Close(addr string) { + cm.mu.RLock() + p, ok := cm.peers[addr] + cm.mu.RUnlock() + if !ok { + return + } + cm.teardownPeer(addr, p) +} + +// teardownPeer removes the peer from the map and releases its resources. +// Safe to call multiple times (e.g. once from a Recv-loop failure and once +// from an explicit Close) — only the first call has an effect. +func (cm *ChannelManager) teardownPeer(addr string, p *peerChannel) { + p.closeOnce.Do(func() { + cm.mu.Lock() + if cm.peers[addr] == p { + delete(cm.peers, addr) + } + cm.mu.Unlock() + + close(p.sendCh) + if p.conn != nil { + _ = p.conn.Close() + } + if p.closeSignal != nil { + p.closeSignal() + } + if cm.metrics != nil { + cm.metrics.DecMembershipChannelConnections() + } + cm.logger.Info("persistent channel torn down", "peer_addr", addr) + }) +} + +// extractPeerAddr reads the dialer's identity from the inbound gRPC +// metadata attached by connect(). +func extractPeerAddr(ctx context.Context) string { + md, ok := metadata.FromIncomingContext(ctx) + if !ok { + return "" + } + vals := md.Get(nodeAddrMetadataKey) + if len(vals) == 0 { + return "" + } + return vals[0] +} diff --git a/internal/membership/channel_test.go b/internal/membership/channel_test.go new file mode 100644 index 0000000..b3accfe --- /dev/null +++ b/internal/membership/channel_test.go @@ -0,0 +1,266 @@ +package membership + +import ( + "fmt" + "io" + "log/slog" + "net" + "strconv" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/gchiesa/drl/internal/metrics" + drlproto "github.com/gchiesa/drl/internal/proto" +) + +func testChannelLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// freePort asks the OS for an available TCP port on 127.0.0.1. +func freePort(t *testing.T) int { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer func() { _ = lis.Close() }() + return lis.Addr().(*net.TCPAddr).Port +} + +// recordingHandler implements channelMessageHandler and records received +// events so tests can assert on delivery. +type recordingHandler struct { + mu sync.Mutex + blocks []*drlproto.BlockEventWithExpiresAt + unblock []*drlproto.UnblockEvent +} + +func (r *recordingHandler) handleChannelBlockWithExpiresAt(evt *drlproto.BlockEventWithExpiresAt) { + r.mu.Lock() + defer r.mu.Unlock() + r.blocks = append(r.blocks, evt) +} + +func (r *recordingHandler) handleChannelUnblock(evt *drlproto.UnblockEvent) { + r.mu.Lock() + defer r.mu.Unlock() + r.unblock = append(r.unblock, evt) +} + +func (r *recordingHandler) blockCount() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.blocks) +} + +func (r *recordingHandler) unblockCount() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.unblock) +} + +// newTestChannelManager builds and starts a ChannelManager bound to the +// given loopback address/port, using a recordingHandler. +func newTestChannelManager(t *testing.T, localAddr string, port int) (*ChannelManager, *recordingHandler) { + t.Helper() + handler := &recordingHandler{} + cm := NewChannelManager(ChannelManagerConfig{ + LocalAddr: localAddr, + Port: port, + Handler: handler, + Metrics: metrics.NewMetrics(), + Logger: testChannelLogger(), + }) + require.NoError(t, cm.Start()) + t.Cleanup(cm.Stop) + return cm, handler +} + +func waitFor(t *testing.T, timeout time.Duration, cond func() bool) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(5 * time.Millisecond) + } + require.True(t, cond(), "condition not met within %s", timeout) +} + +// TestChannelManager_EstablishForPeer_DialDirectionIsDeterministic verifies +// that of two nodes, only the one with the lexicographically smaller address +// dials, per the single-connection-per-pair design. +func TestChannelManager_EstablishForPeer_DialDirectionIsDeterministic(t *testing.T) { + port := freePort(t) + addrA := "127.0.0.1" // smaller address: dials + addrB := "::1" // larger address: passive + + cmA, _ := newTestChannelManager(t, addrA, port) + cmB, _ := newTestChannelManager(t, addrB, port) + + // Both sides observe the join symmetrically, as memberlist NotifyJoin does. + cmA.EstablishForPeer(addrB) + cmB.EstablishForPeer(addrA) + + waitFor(t, 2*time.Second, func() bool { + return cmA.IsConnected(addrB) && cmB.IsConnected(addrA) + }) + + assert.Equal(t, 1, cmA.PeerCount()) + assert.Equal(t, 1, cmB.PeerCount()) +} + +// TestChannelManager_SendRecv_RoundTrip verifies that block and unblock +// events sent from one ChannelManager are received and dispatched by the +// peer's handler on the other side. +func TestChannelManager_SendRecv_RoundTrip(t *testing.T) { + port := freePort(t) + addrA := "127.0.0.1" + addrB := "::1" + + cmA, handlerA := newTestChannelManager(t, addrA, port) + cmB, handlerB := newTestChannelManager(t, addrB, port) + + cmA.EstablishForPeer(addrB) + cmB.EstablishForPeer(addrA) + + waitFor(t, 2*time.Second, func() bool { + return cmA.IsConnected(addrB) && cmB.IsConnected(addrA) + }) + + blockMsg := &drlproto.ChannelMessage{ + Content: &drlproto.ChannelMessage_BlockWithExpiresAt{ + BlockWithExpiresAt: &drlproto.BlockEventWithExpiresAt{ + Key: "entity-key", + ExpiresAtNanos: time.Now().Add(time.Minute).UnixNano(), + }, + }, + } + require.NoError(t, cmA.Send(addrB, blockMsg)) + + unblockMsg := &drlproto.ChannelMessage{ + Content: &drlproto.ChannelMessage_Unblock{ + Unblock: &drlproto.UnblockEvent{Key: "entity-key"}, + }, + } + require.NoError(t, cmB.Send(addrA, unblockMsg)) + + waitFor(t, 2*time.Second, func() bool { + return handlerB.blockCount() == 1 + }) + waitFor(t, 2*time.Second, func() bool { + return handlerA.unblockCount() == 1 + }) + + handlerB.mu.Lock() + assert.Equal(t, "entity-key", handlerB.blocks[0].Key) + handlerB.mu.Unlock() + + handlerA.mu.Lock() + assert.Equal(t, "entity-key", handlerA.unblock[0].Key) + handlerA.mu.Unlock() +} + +// TestChannelManager_Send_UnknownPeer verifies Send returns an error rather +// than blocking or panicking when no channel is established to the target. +func TestChannelManager_Send_UnknownPeer(t *testing.T) { + port := freePort(t) + cm, _ := newTestChannelManager(t, "127.0.0.1", port) + + err := cm.Send("127.0.0.99", &drlproto.ChannelMessage{ + Content: &drlproto.ChannelMessage_Unblock{Unblock: &drlproto.UnblockEvent{Key: "k"}}, + }) + assert.Error(t, err) +} + +// TestChannelManager_Close_TearsDownPeer verifies that Close removes the +// peer entry on both sides of the connection. +func TestChannelManager_Close_TearsDownPeer(t *testing.T) { + port := freePort(t) + addrA := "127.0.0.1" + addrB := "::1" + + cmA, _ := newTestChannelManager(t, addrA, port) + cmB, _ := newTestChannelManager(t, addrB, port) + + cmA.EstablishForPeer(addrB) + cmB.EstablishForPeer(addrA) + + waitFor(t, 2*time.Second, func() bool { + return cmA.IsConnected(addrB) && cmB.IsConnected(addrA) + }) + + cmA.Close(addrB) + + waitFor(t, 2*time.Second, func() bool { + return !cmA.IsConnected(addrB) && !cmB.IsConnected(addrA) + }) +} + +// TestChannelManager_EstablishForPeer_IgnoresSelfAndEmpty verifies the +// no-op guard clauses in EstablishForPeer. +func TestChannelManager_EstablishForPeer_IgnoresSelfAndEmpty(t *testing.T) { + port := freePort(t) + cm, _ := newTestChannelManager(t, "127.0.0.1", port) + + cm.EstablishForPeer("") + cm.EstablishForPeer("127.0.0.1") + + time.Sleep(50 * time.Millisecond) + assert.Equal(t, 0, cm.PeerCount()) +} + +// TestChannelManager_ConcurrentSend verifies concurrent Send calls from +// multiple goroutines to the same peer are safe (serialized by the +// per-peer writeLoop) and all messages are delivered. +func TestChannelManager_ConcurrentSend(t *testing.T) { + port := freePort(t) + addrA := "127.0.0.1" + addrB := "::1" + + cmA, _ := newTestChannelManager(t, addrA, port) + cmB, handlerB := newTestChannelManager(t, addrB, port) + + cmA.EstablishForPeer(addrB) + cmB.EstablishForPeer(addrA) + + waitFor(t, 2*time.Second, func() bool { + return cmA.IsConnected(addrB) && cmB.IsConnected(addrA) + }) + + const n = 50 + var wg sync.WaitGroup + for i := range n { + wg.Add(1) + go func(idx int) { + defer wg.Done() + msg := &drlproto.ChannelMessage{ + Content: &drlproto.ChannelMessage_Unblock{ + Unblock: &drlproto.UnblockEvent{Key: fmt.Sprintf("key-%d", idx)}, + }, + } + _ = cmA.Send(addrB, msg) + }(i) + } + wg.Wait() + + waitFor(t, 2*time.Second, func() bool { + return handlerB.unblockCount() == n + }) +} + +// TestChannelManager_StartUsesConfiguredPort verifies the server actually +// listens on the configured port. +func TestChannelManager_StartUsesConfiguredPort(t *testing.T) { + port := freePort(t) + cm, _ := newTestChannelManager(t, "127.0.0.1", port) + + conn, err := net.DialTimeout("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)), time.Second) + require.NoError(t, err) + _ = conn.Close() + assert.Equal(t, port, cm.port) +} diff --git a/internal/membership/delegate.go b/internal/membership/delegate.go index 3b3d9d6..88cdffc 100644 --- a/internal/membership/delegate.go +++ b/internal/membership/delegate.go @@ -301,10 +301,14 @@ func (d *StateDelegate) MergeRemoteState(buf []byte, join bool) { } // QueueBlockEvent converts the given TTL to an absolute expiration timestamp -// (time.Now().Add(ttl)) and broadcasts a BlockEventWithExpiresAt to all peers. +// (time.Now().Add(ttl)) and propagates a BlockEventWithExpiresAt to all peers. // Using a static timestamp instead of a relative TTL ensures every node applies // the exact same deadline regardless of when the message is delivered. -// Sends are dispatched concurrently via SendReliable (TCP) to avoid blocking the caller. +// +// When the persistent gRPC channel is enabled (config.Membership. +// UseHiPrioPersistentChannel) and established, the event is sent over that +// channel; otherwise it falls back to the legacy memberlist SendReliable +// (TCP) path. Sends are dispatched concurrently so the caller is not blocked. func (d *StateDelegate) QueueBlockEvent(key string, ttl time.Duration, entity *model.Entity) error { expiresAt := time.Now().Add(ttl) evt := &drlproto.BlockEventWithExpiresAt{ @@ -317,6 +321,13 @@ func (d *StateDelegate) QueueBlockEvent(key string, ttl time.Duration, entity *m evt.EntityHdrs = entity.Headers } + if d.useChannel() { + d.sendToAllPeersViaChannel(&drlproto.ChannelMessage{ + Content: &drlproto.ChannelMessage_BlockWithExpiresAt{BlockWithExpiresAt: evt}, + }) + return nil + } + msg := &drlproto.DrlMessage{ Content: &drlproto.DrlMessage_BlockWithExpiresAt{BlockWithExpiresAt: evt}, } @@ -326,15 +337,26 @@ func (d *StateDelegate) QueueBlockEvent(key string, ttl time.Duration, entity *m return fmt.Errorf("failed to marshal block DrlMessage: %w", err) } + d.warnLegacyPath("block") d.sendToAllPeersAsync(data) return nil } -// QueueUnblockEvent builds a DrlMessage with an UnblockEvent and sends it -// immediately to all cluster peers via SendReliable (TCP). +// QueueUnblockEvent sends an UnblockEvent to all cluster peers, preferring +// the persistent gRPC channel when enabled and falling back to memberlist +// SendReliable (TCP) otherwise. func (d *StateDelegate) QueueUnblockEvent(key string) error { + evt := &drlproto.UnblockEvent{Key: key} + + if d.useChannel() { + d.sendToAllPeersViaChannel(&drlproto.ChannelMessage{ + Content: &drlproto.ChannelMessage_Unblock{Unblock: evt}, + }) + return nil + } + msg := &drlproto.DrlMessage{ - Content: &drlproto.DrlMessage_Unblock{Unblock: &drlproto.UnblockEvent{Key: key}}, + Content: &drlproto.DrlMessage_Unblock{Unblock: evt}, } data, err := proto.Marshal(msg) @@ -342,10 +364,31 @@ func (d *StateDelegate) QueueUnblockEvent(key string) error { return fmt.Errorf("failed to marshal unblock DrlMessage: %w", err) } + d.warnLegacyPath("unblock") d.sendToAllPeersAsync(data) return nil } +// useChannel reports whether hi-priority events should be routed over the +// persistent gRPC channel instead of memberlist SendReliable. +func (d *StateDelegate) useChannel() bool { + return d.cluster != nil && + d.cluster.config != nil && + d.cluster.config.Membership.UseHiPrioPersistentChannel && + d.cluster.GetChannelManager() != nil +} + +// warnLegacyPath logs a WARN-level event whenever a hi-priority (block/unblock) +// message is propagated over the legacy on-demand memberlist SendReliable +// (TCP) path instead of the persistent gRPC channel. This surfaces cases +// where the persistent channel is disabled or not yet established to peers, +// per the milestone requirement to flag use of the legacy transport. +func (d *StateDelegate) warnLegacyPath(eventType string) { + d.logger.Warn("using legacy on-demand TCP path for hi-priority event; persistent gRPC channel disabled or unavailable", + "event_type", eventType, + ) +} + // sendToAllPeersAsync sends data to all cluster peers via SendReliable concurrently. // Each peer gets its own goroutine so the caller is not blocked. func (d *StateDelegate) sendToAllPeersAsync(data []byte) { @@ -368,6 +411,47 @@ func (d *StateDelegate) sendToAllPeersAsync(data []byte) { } } +// sendToAllPeersViaChannel sends a ChannelMessage to all cluster peers over +// the persistent gRPC channel. Failures (e.g. no channel yet established to +// a given peer) are logged but never fail the caller, consistent with the +// "availability over consistency" principle applied to the legacy +// SendReliable path. +func (d *StateDelegate) sendToAllPeersViaChannel(msg *drlproto.ChannelMessage) { + if d.cluster == nil { + return + } + cm := d.cluster.GetChannelManager() + if cm == nil { + return + } + localAddr := d.cluster.LocalAddr() + for _, addr := range d.cluster.MemberAddrs() { + if addr == localAddr { + continue + } + if err := cm.Send(addr, msg); err != nil { + d.logger.Warn("failed to send persistent channel msg to peer", + "addr", addr, + "error", err, + ) + } + } +} + +// handleChannelBlockWithExpiresAt applies a block event received over the +// persistent gRPC channel, reusing the same apply logic as the memberlist +// SendReliable path. +func (d *StateDelegate) handleChannelBlockWithExpiresAt(evt *drlproto.BlockEventWithExpiresAt) { + d.handleBlockEventWithExpiresAt(evt) +} + +// handleChannelUnblock applies an unblock event received over the +// persistent gRPC channel, reusing the same apply logic as the memberlist +// SendReliable path. +func (d *StateDelegate) handleChannelUnblock(evt *drlproto.UnblockEvent) { + d.handleUnblockEvent(evt) +} + // SetHandover sets the handover handler for graceful state evacuation. func (d *StateDelegate) SetHandover(h *Handover) { d.handover = h diff --git a/internal/membership/delegate_test.go b/internal/membership/delegate_test.go index cdcd6b7..6d5fe6e 100644 --- a/internal/membership/delegate_test.go +++ b/internal/membership/delegate_test.go @@ -1,6 +1,8 @@ package membership import ( + "bytes" + "log/slog" "sync/atomic" "testing" "time" @@ -11,6 +13,7 @@ import ( "google.golang.org/protobuf/proto" "github.com/gchiesa/drl/internal/cache" + "github.com/gchiesa/drl/internal/config" "github.com/gchiesa/drl/internal/metrics" drlproto "github.com/gchiesa/drl/internal/proto" ) @@ -480,3 +483,188 @@ func TestStateDelegate_ConcurrentAccess(t *testing.T) { time.Sleep(100 * time.Millisecond) done.Store(true) } + +// TestStateDelegate_UseChannel covers the gating logic that decides whether +// hi-priority events are routed over the persistent gRPC channel (added for +// milestone 020) instead of the legacy memberlist SendReliable path. +func TestStateDelegate_UseChannel(t *testing.T) { + enabledCfg := &config.Config{ + Membership: config.MembershipConfig{UseHiPrioPersistentChannel: true}, + } + disabledCfg := &config.Config{ + Membership: config.MembershipConfig{UseHiPrioPersistentChannel: false}, + } + + t.Run("nil cluster", func(t *testing.T) { + delegate := NewStateDelegate(DelegateConfig{SyncTimeout: time.Second}) + assert.False(t, delegate.useChannel()) + }) + + t.Run("nil config", func(t *testing.T) { + delegate := NewStateDelegate(DelegateConfig{SyncTimeout: time.Second}) + delegate.SetCluster(&Cluster{}) + assert.False(t, delegate.useChannel()) + }) + + t.Run("feature disabled", func(t *testing.T) { + delegate := NewStateDelegate(DelegateConfig{SyncTimeout: time.Second}) + delegate.SetCluster(&Cluster{config: disabledCfg}) + assert.False(t, delegate.useChannel()) + }) + + t.Run("enabled but no channel manager established", func(t *testing.T) { + delegate := NewStateDelegate(DelegateConfig{SyncTimeout: time.Second}) + delegate.SetCluster(&Cluster{config: enabledCfg}) + assert.False(t, delegate.useChannel()) + }) + + t.Run("enabled and channel manager established", func(t *testing.T) { + delegate := NewStateDelegate(DelegateConfig{SyncTimeout: time.Second}) + cluster := &Cluster{config: enabledCfg} + cluster.SetChannelManager(NewChannelManager(ChannelManagerConfig{ + LocalAddr: "127.0.0.1", + Port: 0, + Metrics: metrics.NewMetrics(), + Logger: testChannelLogger(), + })) + delegate.SetCluster(cluster) + assert.True(t, delegate.useChannel()) + }) +} + +// TestStateDelegate_HandleChannelBlockWithExpiresAt verifies the persistent +// channel's block handler applies the event to the blocklist identically to +// the legacy NotifyMsg path. +func TestStateDelegate_HandleChannelBlockWithExpiresAt(t *testing.T) { + bc, err := cache.NewBlocklistCache(cache.BlocklistConfig{MaxSizeMB: 1}) + require.NoError(t, err) + defer bc.Close() + + delegate := NewStateDelegate(DelegateConfig{ + Blocklist: bc, + SyncTimeout: 30 * time.Second, + }) + + expiresAt := time.Now().Add(10 * time.Minute) + delegate.handleChannelBlockWithExpiresAt(&drlproto.BlockEventWithExpiresAt{ + Key: "channel-block-key", + ExpiresAtNanos: expiresAt.UnixNano(), + EntityIp: "10.0.0.3", + EntityPath: "/api/v3", + }) + + assert.True(t, bc.IsBlocked("channel-block-key")) + entries := bc.ListEntries() + require.Len(t, entries, 1) + require.NotNil(t, entries[0].Entity) + assert.Equal(t, "10.0.0.3", entries[0].Entity.IP) +} + +// TestStateDelegate_HandleChannelUnblock verifies the persistent channel's +// unblock handler removes the entity from the blocklist. +func TestStateDelegate_HandleChannelUnblock(t *testing.T) { + bc, err := cache.NewBlocklistCache(cache.BlocklistConfig{MaxSizeMB: 1}) + require.NoError(t, err) + defer bc.Close() + + const key = "channel-unblock-key" + bc.Block(key, time.Hour, nil) + require.True(t, bc.IsBlocked(key)) + + delegate := NewStateDelegate(DelegateConfig{ + Blocklist: bc, + SyncTimeout: 30 * time.Second, + }) + + delegate.handleChannelUnblock(&drlproto.UnblockEvent{Key: key}) + + assert.False(t, bc.IsBlocked(key)) +} + +// TestStateDelegate_QueueBlockEvent_ViaChannel_NoPanic exercises the new +// useChannel()==true branch of QueueBlockEvent/QueueUnblockEvent. With no +// cluster members beyond self, sendToAllPeersViaChannel has nothing to +// deliver, but the call must not panic or fall back to the legacy path. +func TestStateDelegate_QueueBlockEvent_ViaChannel_NoPanic(t *testing.T) { + cfg := &config.Config{ + Membership: config.MembershipConfig{UseHiPrioPersistentChannel: true}, + } + cluster := &Cluster{config: cfg, localIP: "127.0.0.1"} + cluster.SetChannelManager(NewChannelManager(ChannelManagerConfig{ + LocalAddr: "127.0.0.1", + Port: 0, + Metrics: metrics.NewMetrics(), + Logger: testChannelLogger(), + })) + + delegate := NewStateDelegate(DelegateConfig{SyncTimeout: time.Second}) + delegate.SetCluster(cluster) + + assert.NotPanics(t, func() { + assert.NoError(t, delegate.QueueBlockEvent("key1", time.Hour, nil)) + }) + assert.NotPanics(t, func() { + assert.NoError(t, delegate.QueueUnblockEvent("key1")) + }) +} + +// TestStateDelegate_QueueBlockEvent_LegacyPath_LogsWarn verifies that a WARN +// log is emitted whenever a hi-priority event is propagated via the legacy +// on-demand memberlist SendReliable (TCP) path, i.e. whenever useChannel() +// is false. This covers the milestone requirement to flag use of the old +// transport. +func TestStateDelegate_QueueBlockEvent_LegacyPath_LogsWarn(t *testing.T) { + var buf bytes.Buffer + logger := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelWarn})) + + delegate := NewStateDelegate(DelegateConfig{SyncTimeout: time.Second, Logger: logger}) + // No cluster set -> useChannel() is false -> legacy path is taken. + + require.NoError(t, delegate.QueueBlockEvent("key1", time.Hour, nil)) + + logOutput := buf.String() + assert.Contains(t, logOutput, "legacy on-demand TCP path") + assert.Contains(t, logOutput, "event_type=block") +} + +// TestStateDelegate_QueueUnblockEvent_LegacyPath_LogsWarn mirrors the block +// case above for QueueUnblockEvent. +func TestStateDelegate_QueueUnblockEvent_LegacyPath_LogsWarn(t *testing.T) { + var buf bytes.Buffer + logger := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelWarn})) + + delegate := NewStateDelegate(DelegateConfig{SyncTimeout: time.Second, Logger: logger}) + // No cluster set -> useChannel() is false -> legacy path is taken. + + require.NoError(t, delegate.QueueUnblockEvent("key1")) + + logOutput := buf.String() + assert.Contains(t, logOutput, "legacy on-demand TCP path") + assert.Contains(t, logOutput, "event_type=unblock") +} + +// TestStateDelegate_QueueBlockEvent_ChannelPath_NoLegacyWarn verifies that +// no legacy-path WARN is logged when the persistent gRPC channel is enabled +// and established, i.e. useChannel() is true. +func TestStateDelegate_QueueBlockEvent_ChannelPath_NoLegacyWarn(t *testing.T) { + var buf bytes.Buffer + logger := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelWarn})) + + cfg := &config.Config{ + Membership: config.MembershipConfig{UseHiPrioPersistentChannel: true}, + } + cluster := &Cluster{config: cfg, localIP: "127.0.0.1"} + cluster.SetChannelManager(NewChannelManager(ChannelManagerConfig{ + LocalAddr: "127.0.0.1", + Port: 0, + Metrics: metrics.NewMetrics(), + Logger: testChannelLogger(), + })) + + delegate := NewStateDelegate(DelegateConfig{SyncTimeout: time.Second, Logger: logger}) + delegate.SetCluster(cluster) + + require.NoError(t, delegate.QueueBlockEvent("key1", time.Hour, nil)) + + assert.NotContains(t, buf.String(), "legacy on-demand TCP path") +} diff --git a/internal/membership/event.go b/internal/membership/event.go index 0f5bee5..c434858 100644 --- a/internal/membership/event.go +++ b/internal/membership/event.go @@ -26,6 +26,14 @@ func (e *eventDelegate) NotifyJoin(node *memberlist.Node) { e.cluster.updateClusterSize() e.cluster.cacheManager.UpdateNodes(e.cluster.MemberAddrs()) }() + + // Establish the persistent gRPC channel to the newly-known peer, if + // enabled. EstablishForPeer decides deterministically whether the local + // node dials or waits to be dialed, so this is safe to call from both + // the joiner's and the existing members' perspective. + if cm := e.cluster.GetChannelManager(); cm != nil { + cm.EstablishForPeer(node.Addr.String()) + } } func (e *eventDelegate) NotifyLeave(node *memberlist.Node) { @@ -39,6 +47,11 @@ func (e *eventDelegate) NotifyLeave(node *memberlist.Node) { e.cluster.updateClusterSize() e.cluster.cacheManager.UpdateNodes(e.cluster.MemberAddrs()) }() + + // Close the persistent gRPC channel to the departed peer, if any. + if cm := e.cluster.GetChannelManager(); cm != nil { + cm.Close(node.Addr.String()) + } } func (e *eventDelegate) NotifyUpdate(node *memberlist.Node) { diff --git a/internal/membership/membership.go b/internal/membership/membership.go index a2e599e..81b2ac4 100644 --- a/internal/membership/membership.go +++ b/internal/membership/membership.go @@ -17,15 +17,16 @@ import ( // Cluster manages the memberlist cluster membership type Cluster struct { - config *config.Config - localIP string - cacheManager *cache.Manager - memberlist *memberlist.Memberlist - metrics *metrics.Metrics - logger *slog.Logger - stateDelegate *StateDelegate - mu sync.RWMutex - ready bool + config *config.Config + localIP string + cacheManager *cache.Manager + memberlist *memberlist.Memberlist + metrics *metrics.Metrics + logger *slog.Logger + stateDelegate *StateDelegate + channelManager *ChannelManager + mu sync.RWMutex + ready bool } // NewCluster creates a new Cluster instance @@ -53,6 +54,23 @@ func (c *Cluster) GetStateDelegate() *StateDelegate { return c.stateDelegate } +// SetChannelManager sets the persistent gRPC channel manager used for +// hi-priority (block/unblock) event propagation when +// config.Membership.UseHiPrioPersistentChannel is enabled. +func (c *Cluster) SetChannelManager(cm *ChannelManager) { + c.mu.Lock() + defer c.mu.Unlock() + c.channelManager = cm +} + +// GetChannelManager returns the persistent gRPC channel manager, or nil if +// the feature is disabled. +func (c *Cluster) GetChannelManager() *ChannelManager { + c.mu.RLock() + defer c.mu.RUnlock() + return c.channelManager +} + // Start initializes and starts the memberlist cluster func (c *Cluster) Start() error { mlConfig := memberlist.DefaultLANConfig() @@ -185,6 +203,13 @@ func (c *Cluster) JoinCluster() error { c.updateClusterSize() c.cacheManager.UpdateNodes(c.MemberAddrs()) + // When the persistent gRPC channel is enabled, a newly joining node is + // only considered ready once its channel connections to all currently + // known peers are established. This avoids a window where hi-priority + // block/unblock events destined for this node would be undeliverable + // over the channel during the join. + c.waitForChannelsReady() + // Wait for state sync if delegate is configured c.mu.RLock() delegate := c.stateDelegate @@ -201,6 +226,60 @@ func (c *Cluster) JoinCluster() error { return nil } +// channelReadyTimeout bounds how long JoinCluster waits for persistent gRPC +// channel connections to be established to all known peers before +// proceeding with readiness. This favours availability over strict +// consistency: if a peer's channel never comes up (e.g. it is unreachable), +// the node still becomes ready rather than blocking forever. +const channelReadyTimeout = 30 * time.Second + +// waitForChannelsReady blocks until the persistent gRPC channel manager +// reports an established connection to every currently known peer, or until +// channelReadyTimeout elapses. It is a no-op when the persistent channel +// feature is disabled (no channel manager configured). +func (c *Cluster) waitForChannelsReady() { + cm := c.GetChannelManager() + if cm == nil { + return + } + + localAddr := c.LocalAddr() + deadline := time.Now().Add(channelReadyTimeout) + + for { + peers := c.MemberAddrs() + expected := 0 + allConnected := true + for _, addr := range peers { + if addr == localAddr { + continue + } + expected++ + if !cm.IsConnected(addr) { + allConnected = false + } + } + + if allConnected { + c.logger.Info("persistent gRPC channel connections established", + "expected_peers", expected, + ) + return + } + + if time.Now().After(deadline) { + c.logger.Warn("timed out waiting for persistent gRPC channel connections to all peers; proceeding without full connectivity", + "timeout", channelReadyTimeout, + "connected_peers", cm.PeerCount(), + "expected_peers", expected, + ) + return + } + + time.Sleep(50 * time.Millisecond) + } +} + // markReady marks the cluster as ready, handling both delegate and non-delegate cases func (c *Cluster) markReady() { c.mu.Lock() @@ -397,6 +476,12 @@ func (c *Cluster) findNodeByAddr(addr string) *memberlist.Node { // Leave gracefully leaves the cluster func (c *Cluster) Leave(timeout time.Duration) error { + // Tear down the persistent gRPC channel (all peer connections and the + // listening server), if it was enabled, before leaving memberlist. + if cm := c.GetChannelManager(); cm != nil { + cm.Stop() + } + if c.memberlist == nil { return nil } diff --git a/internal/membership/membership_test.go b/internal/membership/membership_test.go index e6c339e..fbe267b 100644 --- a/internal/membership/membership_test.go +++ b/internal/membership/membership_test.go @@ -271,3 +271,70 @@ func TestClusterLeaveWithoutStart(t *testing.T) { t.Errorf("unexpected error leaving unstarted cluster: %v", err) } } + +// TestCluster_WaitForChannelsReady_NoChannelManager_NoOp verifies that +// waitForChannelsReady returns immediately (no blocking) when the +// persistent gRPC channel feature is disabled, i.e. no ChannelManager has +// been attached to the cluster. +func TestCluster_WaitForChannelsReady_NoChannelManager_NoOp(t *testing.T) { + localIP := testLocalIP(t) + cfg := &config.Config{ + Membership: config.MembershipConfig{ + ServiceName: "drl", + Port: 17949, + BindAddr: "127.0.0.1", + StartupDelay: 100 * time.Millisecond, + }, + } + m := metrics.NewMetrics() + cm := testCacheManager(t, localIP) + defer cm.Close() + logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelError})) + + cluster := NewCluster(cfg, localIP, cm, m, logger) + + start := time.Now() + cluster.waitForChannelsReady() + assert.Less(t, time.Since(start), time.Second, "waitForChannelsReady should return immediately with no channel manager") +} + +// TestCluster_WaitForChannelsReady_NoPeers_ReturnsImmediately verifies that +// when a ChannelManager is attached but the cluster has no peers besides +// itself, waitForChannelsReady returns immediately (there is nothing to +// wait for). +func TestCluster_WaitForChannelsReady_NoPeers_ReturnsImmediately(t *testing.T) { + localIP := testLocalIP(t) + cfg := &config.Config{ + Membership: config.MembershipConfig{ + ServiceName: "drl", + Port: 17950, + BindAddr: "127.0.0.1", + StartupDelay: 100 * time.Millisecond, + UseHiPrioPersistentChannel: true, + }, + } + m := metrics.NewMetrics() + cm := testCacheManager(t, localIP) + defer cm.Close() + logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelError})) + + cluster := NewCluster(cfg, localIP, cm, m, logger) + require.NoError(t, cluster.Start()) + defer func() { _ = cluster.Leave(time.Second) }() + + port := freePort(t) + channelManager := NewChannelManager(ChannelManagerConfig{ + LocalAddr: localIP, + Port: port, + Handler: cluster.stateDelegate, + Metrics: m, + Logger: logger, + }) + require.NoError(t, channelManager.Start()) + defer channelManager.Stop() + cluster.SetChannelManager(channelManager) + + start := time.Now() + cluster.waitForChannelsReady() + assert.Less(t, time.Since(start), time.Second, "waitForChannelsReady should return immediately when there are no peers to connect to") +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index c62f737..989c01e 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -35,6 +35,12 @@ type Metrics struct { MembershipReliableMsgsTotal prometheus.Counter MembershipBestEffortMsgsTotal prometheus.Counter + // Persistent gRPC channel metrics + MembershipChannelMsgsSentTotal prometheus.Counter + MembershipChannelMsgsRecvTotal prometheus.Counter + MembershipChannelConnectionsActive prometheus.Gauge + MembershipChannelErrorsTotal prometheus.Counter + // Handover metrics HandoverOutEntities prometheus.Counter HandoverInEntities prometheus.Counter @@ -138,6 +144,24 @@ func NewMetrics() *Metrics { Help: "Total number of best-effort messages sent via memberlist", }), + // Persistent gRPC channel metrics + MembershipChannelMsgsSentTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "drl_membership_channel_msgs_sent_total", + Help: "Total number of hi-priority messages sent via the persistent gRPC channel", + }), + MembershipChannelMsgsRecvTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "drl_membership_channel_msgs_recv_total", + Help: "Total number of hi-priority messages received via the persistent gRPC channel", + }), + MembershipChannelConnectionsActive: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "drl_membership_channel_connections_active", + Help: "Current number of active persistent gRPC channel connections to peers", + }), + MembershipChannelErrorsTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "drl_membership_channel_errors_total", + Help: "Total number of persistent gRPC channel errors (dial failures, send/recv failures)", + }), + // Handover metrics HandoverOutEntities: prometheus.NewCounter(prometheus.CounterOpts{ Name: "drl_accounting_handover_out_entities", @@ -230,6 +254,10 @@ func NewMetrics() *Metrics { registry.MustRegister(m.AccountingBulkLoadTotal) registry.MustRegister(m.MembershipReliableMsgsTotal) registry.MustRegister(m.MembershipBestEffortMsgsTotal) + registry.MustRegister(m.MembershipChannelMsgsSentTotal) + registry.MustRegister(m.MembershipChannelMsgsRecvTotal) + registry.MustRegister(m.MembershipChannelConnectionsActive) + registry.MustRegister(m.MembershipChannelErrorsTotal) registry.MustRegister(m.HandoverOutEntities) registry.MustRegister(m.HandoverInEntities) registry.MustRegister(m.HandoverDurationMs) @@ -326,6 +354,31 @@ func (m *Metrics) IncMembershipBestEffort() { m.MembershipBestEffortMsgsTotal.Inc() } +// IncMembershipChannelMsgsSent increments the persistent gRPC channel sent-message counter +func (m *Metrics) IncMembershipChannelMsgsSent() { + m.MembershipChannelMsgsSentTotal.Inc() +} + +// IncMembershipChannelMsgsRecv increments the persistent gRPC channel received-message counter +func (m *Metrics) IncMembershipChannelMsgsRecv() { + m.MembershipChannelMsgsRecvTotal.Inc() +} + +// IncMembershipChannelConnections increments the active persistent gRPC channel connections gauge +func (m *Metrics) IncMembershipChannelConnections() { + m.MembershipChannelConnectionsActive.Inc() +} + +// DecMembershipChannelConnections decrements the active persistent gRPC channel connections gauge +func (m *Metrics) DecMembershipChannelConnections() { + m.MembershipChannelConnectionsActive.Dec() +} + +// IncMembershipChannelErrors increments the persistent gRPC channel error counter +func (m *Metrics) IncMembershipChannelErrors() { + m.MembershipChannelErrorsTotal.Inc() +} + // AddHandoverOut adds to the handover out entities counter func (m *Metrics) AddHandoverOut(n float64) { m.HandoverOutEntities.Add(n) diff --git a/internal/proto/channel.pb.go b/internal/proto/channel.pb.go new file mode 100644 index 0000000..941ca53 --- /dev/null +++ b/internal/proto/channel.pb.go @@ -0,0 +1,181 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.11 +// protoc v5.29.6 +// source: internal/proto/channel.proto + +package proto + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// ChannelMessage is the envelope for events exchanged over the persistent, +// hi-priority gRPC channel established between cluster members when a node +// joins the cluster. Like DrlMessage, it carries an explicit type +// discriminator via oneof so new hi-priority event kinds can be multiplexed +// onto the same channel in the future without breaking the wire format or +// requiring a new RPC. +type ChannelMessage struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Types that are valid to be assigned to Content: + // + // *ChannelMessage_BlockWithExpiresAt + // *ChannelMessage_Unblock + Content isChannelMessage_Content `protobuf_oneof:"content"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ChannelMessage) Reset() { + *x = ChannelMessage{} + mi := &file_internal_proto_channel_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ChannelMessage) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ChannelMessage) ProtoMessage() {} + +func (x *ChannelMessage) ProtoReflect() protoreflect.Message { + mi := &file_internal_proto_channel_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ChannelMessage.ProtoReflect.Descriptor instead. +func (*ChannelMessage) Descriptor() ([]byte, []int) { + return file_internal_proto_channel_proto_rawDescGZIP(), []int{0} +} + +func (x *ChannelMessage) GetContent() isChannelMessage_Content { + if x != nil { + return x.Content + } + return nil +} + +func (x *ChannelMessage) GetBlockWithExpiresAt() *BlockEventWithExpiresAt { + if x != nil { + if x, ok := x.Content.(*ChannelMessage_BlockWithExpiresAt); ok { + return x.BlockWithExpiresAt + } + } + return nil +} + +func (x *ChannelMessage) GetUnblock() *UnblockEvent { + if x != nil { + if x, ok := x.Content.(*ChannelMessage_Unblock); ok { + return x.Unblock + } + } + return nil +} + +type isChannelMessage_Content interface { + isChannelMessage_Content() +} + +type ChannelMessage_BlockWithExpiresAt struct { + BlockWithExpiresAt *BlockEventWithExpiresAt `protobuf:"bytes,1,opt,name=block_with_expires_at,json=blockWithExpiresAt,proto3,oneof"` +} + +type ChannelMessage_Unblock struct { + Unblock *UnblockEvent `protobuf:"bytes,2,opt,name=unblock,proto3,oneof"` +} + +func (*ChannelMessage_BlockWithExpiresAt) isChannelMessage_Content() {} + +func (*ChannelMessage_Unblock) isChannelMessage_Content() {} + +var File_internal_proto_channel_proto protoreflect.FileDescriptor + +const file_internal_proto_channel_proto_rawDesc = "" + + "\n" + + "\x1cinternal/proto/channel.proto\x12\x06drl.v1\x1a\x1finternal/proto/accounting.proto\"\xa3\x01\n" + + "\x0eChannelMessage\x12T\n" + + "\x15block_with_expires_at\x18\x01 \x01(\v2\x1f.drl.v1.BlockEventWithExpiresAtH\x00R\x12blockWithExpiresAt\x120\n" + + "\aunblock\x18\x02 \x01(\v2\x14.drl.v1.UnblockEventH\x00R\aunblockB\t\n" + + "\acontent2Q\n" + + "\x11PersistentChannel\x12<\n" + + "\x06Stream\x12\x16.drl.v1.ChannelMessage\x1a\x16.drl.v1.ChannelMessage(\x010\x01B'Z%github.com/gchiesa/drl/internal/protob\x06proto3" + +var ( + file_internal_proto_channel_proto_rawDescOnce sync.Once + file_internal_proto_channel_proto_rawDescData []byte +) + +func file_internal_proto_channel_proto_rawDescGZIP() []byte { + file_internal_proto_channel_proto_rawDescOnce.Do(func() { + file_internal_proto_channel_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_internal_proto_channel_proto_rawDesc), len(file_internal_proto_channel_proto_rawDesc))) + }) + return file_internal_proto_channel_proto_rawDescData +} + +var file_internal_proto_channel_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_internal_proto_channel_proto_goTypes = []any{ + (*ChannelMessage)(nil), // 0: drl.v1.ChannelMessage + (*BlockEventWithExpiresAt)(nil), // 1: drl.v1.BlockEventWithExpiresAt + (*UnblockEvent)(nil), // 2: drl.v1.UnblockEvent +} +var file_internal_proto_channel_proto_depIdxs = []int32{ + 1, // 0: drl.v1.ChannelMessage.block_with_expires_at:type_name -> drl.v1.BlockEventWithExpiresAt + 2, // 1: drl.v1.ChannelMessage.unblock:type_name -> drl.v1.UnblockEvent + 0, // 2: drl.v1.PersistentChannel.Stream:input_type -> drl.v1.ChannelMessage + 0, // 3: drl.v1.PersistentChannel.Stream:output_type -> drl.v1.ChannelMessage + 3, // [3:4] is the sub-list for method output_type + 2, // [2:3] is the sub-list for method input_type + 2, // [2:2] is the sub-list for extension type_name + 2, // [2:2] is the sub-list for extension extendee + 0, // [0:2] is the sub-list for field type_name +} + +func init() { file_internal_proto_channel_proto_init() } +func file_internal_proto_channel_proto_init() { + if File_internal_proto_channel_proto != nil { + return + } + file_internal_proto_accounting_proto_init() + file_internal_proto_channel_proto_msgTypes[0].OneofWrappers = []any{ + (*ChannelMessage_BlockWithExpiresAt)(nil), + (*ChannelMessage_Unblock)(nil), + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_internal_proto_channel_proto_rawDesc), len(file_internal_proto_channel_proto_rawDesc)), + NumEnums: 0, + NumMessages: 1, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_internal_proto_channel_proto_goTypes, + DependencyIndexes: file_internal_proto_channel_proto_depIdxs, + MessageInfos: file_internal_proto_channel_proto_msgTypes, + }.Build() + File_internal_proto_channel_proto = out.File + file_internal_proto_channel_proto_goTypes = nil + file_internal_proto_channel_proto_depIdxs = nil +} diff --git a/internal/proto/channel.proto b/internal/proto/channel.proto new file mode 100644 index 0000000..ef58cf1 --- /dev/null +++ b/internal/proto/channel.proto @@ -0,0 +1,31 @@ +syntax = "proto3"; +package drl.v1; +option go_package = "github.com/gchiesa/drl/internal/proto"; + +import "internal/proto/accounting.proto"; + +// ChannelMessage is the envelope for events exchanged over the persistent, +// hi-priority gRPC channel established between cluster members when a node +// joins the cluster. Like DrlMessage, it carries an explicit type +// discriminator via oneof so new hi-priority event kinds can be multiplexed +// onto the same channel in the future without breaking the wire format or +// requiring a new RPC. +message ChannelMessage { + oneof content { + BlockEventWithExpiresAt block_with_expires_at = 1; + UnblockEvent unblock = 2; + } +} + +// PersistentChannel is a bidirectional streaming RPC used to propagate +// hi-priority events between two cluster members over a single, long-lived +// gRPC connection instead of an on-demand connection per event. +// +// Exactly one stream is established per unordered pair of nodes: the dial +// direction is decided deterministically by comparing node addresses (see +// internal/membership/channel.go), and both peers Send and Recv on that same +// stream. This halves the number of live connections compared to a model +// where each node dials the other independently. +service PersistentChannel { + rpc Stream(stream ChannelMessage) returns (stream ChannelMessage); +} diff --git a/internal/proto/channel_grpc.pb.go b/internal/proto/channel_grpc.pb.go new file mode 100644 index 0000000..79f16a0 --- /dev/null +++ b/internal/proto/channel_grpc.pb.go @@ -0,0 +1,135 @@ +// Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.6.2 +// - protoc v5.29.6 +// source: internal/proto/channel.proto + +package proto + +import ( + context "context" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" +) + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +// Requires gRPC-Go v1.64.0 or later. +const _ = grpc.SupportPackageIsVersion9 + +const ( + PersistentChannel_Stream_FullMethodName = "/drl.v1.PersistentChannel/Stream" +) + +// PersistentChannelClient is the client API for PersistentChannel service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. +// +// PersistentChannel is a bidirectional streaming RPC used to propagate +// hi-priority events between two cluster members over a single, long-lived +// gRPC connection instead of an on-demand connection per event. +// +// Exactly one stream is established per unordered pair of nodes: the dial +// direction is decided deterministically by comparing node addresses (see +// internal/membership/channel.go), and both peers Send and Recv on that same +// stream. This halves the number of live connections compared to a model +// where each node dials the other independently. +type PersistentChannelClient interface { + Stream(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[ChannelMessage, ChannelMessage], error) +} + +type persistentChannelClient struct { + cc grpc.ClientConnInterface +} + +func NewPersistentChannelClient(cc grpc.ClientConnInterface) PersistentChannelClient { + return &persistentChannelClient{cc} +} + +func (c *persistentChannelClient) Stream(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[ChannelMessage, ChannelMessage], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &PersistentChannel_ServiceDesc.Streams[0], PersistentChannel_Stream_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[ChannelMessage, ChannelMessage]{ClientStream: stream} + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type PersistentChannel_StreamClient = grpc.BidiStreamingClient[ChannelMessage, ChannelMessage] + +// PersistentChannelServer is the server API for PersistentChannel service. +// All implementations must embed UnimplementedPersistentChannelServer +// for forward compatibility. +// +// PersistentChannel is a bidirectional streaming RPC used to propagate +// hi-priority events between two cluster members over a single, long-lived +// gRPC connection instead of an on-demand connection per event. +// +// Exactly one stream is established per unordered pair of nodes: the dial +// direction is decided deterministically by comparing node addresses (see +// internal/membership/channel.go), and both peers Send and Recv on that same +// stream. This halves the number of live connections compared to a model +// where each node dials the other independently. +type PersistentChannelServer interface { + Stream(grpc.BidiStreamingServer[ChannelMessage, ChannelMessage]) error + mustEmbedUnimplementedPersistentChannelServer() +} + +// UnimplementedPersistentChannelServer must be embedded to have +// forward compatible implementations. +// +// NOTE: this should be embedded by value instead of pointer to avoid a nil +// pointer dereference when methods are called. +type UnimplementedPersistentChannelServer struct{} + +func (UnimplementedPersistentChannelServer) Stream(grpc.BidiStreamingServer[ChannelMessage, ChannelMessage]) error { + return status.Error(codes.Unimplemented, "method Stream not implemented") +} +func (UnimplementedPersistentChannelServer) mustEmbedUnimplementedPersistentChannelServer() {} +func (UnimplementedPersistentChannelServer) testEmbeddedByValue() {} + +// UnsafePersistentChannelServer may be embedded to opt out of forward compatibility for this service. +// Use of this interface is not recommended, as added methods to PersistentChannelServer will +// result in compilation errors. +type UnsafePersistentChannelServer interface { + mustEmbedUnimplementedPersistentChannelServer() +} + +func RegisterPersistentChannelServer(s grpc.ServiceRegistrar, srv PersistentChannelServer) { + // If the following call panics, it indicates UnimplementedPersistentChannelServer was + // embedded by pointer and is nil. This will cause panics if an + // unimplemented method is ever invoked, so we test this at initialization + // time to prevent it from happening at runtime later due to I/O. + if t, ok := srv.(interface{ testEmbeddedByValue() }); ok { + t.testEmbeddedByValue() + } + s.RegisterService(&PersistentChannel_ServiceDesc, srv) +} + +func _PersistentChannel_Stream_Handler(srv interface{}, stream grpc.ServerStream) error { + return srv.(PersistentChannelServer).Stream(&grpc.GenericServerStream[ChannelMessage, ChannelMessage]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type PersistentChannel_StreamServer = grpc.BidiStreamingServer[ChannelMessage, ChannelMessage] + +// PersistentChannel_ServiceDesc is the grpc.ServiceDesc for PersistentChannel service. +// It's only intended for direct use with grpc.RegisterService, +// and not to be introspected or modified (even as a copy) +var PersistentChannel_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "drl.v1.PersistentChannel", + HandlerType: (*PersistentChannelServer)(nil), + Methods: []grpc.MethodDesc{}, + Streams: []grpc.StreamDesc{ + { + StreamName: "Stream", + Handler: _PersistentChannel_Stream_Handler, + ServerStreams: true, + ClientStreams: true, + }, + }, + Metadata: "internal/proto/channel.proto", +} diff --git a/mise.toml b/mise.toml index c57fc19..3a4ec6d 100644 --- a/mise.toml +++ b/mise.toml @@ -5,6 +5,7 @@ protoc = "29.6" hugo = "0.160.1" node = "22" "go:github.com/swaggo/swag/cmd/swag" = "v1.16.4" +"go:google.golang.org/grpc/cmd/protoc-gen-go-grpc" = "v1.6.2" [tasks.build] description = "Build the DRL binary" From 881732ec876789465ad81b3f46a99c8308208f94 Mon Sep 17 00:00:00 2001 From: Giuseppe Chiesa Date: Sat, 15 Aug 2026 09:25:43 +0200 Subject: [PATCH 2/2] feat(ms020): update docker compose replicas --- deployments/docker-compose/docker-compose.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/deployments/docker-compose/docker-compose.yaml b/deployments/docker-compose/docker-compose.yaml index 8791e97..529688b 100644 --- a/deployments/docker-compose/docker-compose.yaml +++ b/deployments/docker-compose/docker-compose.yaml @@ -58,7 +58,7 @@ services: retries: 3 start_period: 10s deploy: - replicas: 1 + replicas: 3 environment: DRL_CONFIG_PATH: /etc/drl/config.kdl DRL_PRIVATE_API_KEY: "Test5ecretPrivateAPIKey!"