aggregate

Running

Configuration

Configuration is loaded in this order:

If AGGREGATE_CONFIG_PATH is set and points to a readable config file, load that file and apply env overrides.
Otherwise load defaults and apply env overrides.

Supported config formats:

YAML (.yaml, .yml)
TOML (.toml)
JSON (.json)

Environment Variables

Core

AGGREGATE_NATS_URL (default: nats://localhost:4222): NATS server URL.
AGGREGATE_STORAGE_PATH (default: ./data): Path used by the snapshot storage.
AGGREGATE_SNAPSHOT_THRESHOLD (default: 10): Save snapshot when events since last snapshot reach this threshold.
AGGREGATE_MAX_RETRIES (default: 3): Max retries for version conflicts in command handling.
AGGREGATE_HTTP_ADDR (default: 0.0.0.0:8080): HTTP bind address.
AGGREGATE_GRPC_ADDR (default: 0.0.0.0:50051): gRPC bind address for command submission.

Multi-tenant

AGGREGATE_MULTI_TENANT (default: true): Enables multi-tenant behavior when parsing/validating tenant ids.
AGGREGATE_DEFAULT_TENANT_ID (default: unset): Default tenant id when the incoming request doesn't specify one.
AGGREGATE_SHARD_ID (default: local): Shard id used when applying placement maps.

Logging

AGGREGATE_LOGGER_SOCKET (default: unset): Socket path for edge-logger-client integration (if enabled).

Server

AGGREGATE_CONFIG_PATH (default: unset): Path to a YAML/TOML/JSON config file.

Placement

AGGREGATE_PLACEMENT_BUCKET (default: AGGREGATE_PLACEMENT): NATS KV bucket to watch.
AGGREGATE_PLACEMENT_KEY (default: aggregate_placement): NATS KV key to watch. Value is a JSON object mapping tenant_id -> shard_id.

Runtime Programs

AGGREGATE_DECIDE_PROGRAM / AGGREGATE_APPLY_PROGRAM: Inline program source strings.
AGGREGATE_DECIDE_PROGRAM_PATH / AGGREGATE_APPLY_PROGRAM_PATH: File paths to program source strings.

HTTP Endpoints

GET /health → JSON health report
GET /ready → JSON boolean readiness
GET /metrics → Prometheus text format
GET /admin/tenants → JSON list of hosted tenants
POST /admin/drain → marks tenant draining and waits for in-flight commands to finish ({"tenant_id":"..."})
POST /admin/reload → updates hosted tenant allowlist ({"hosted_tenants":[...]}) or applies a placement map ({"placement":{...}})
GET /admin/tenant/{tenant_id}/status → JSON tenant status (hosted, accepting, draining, in_flight)
GET /admin/tenant/{tenant_id}/ready → JSON boolean (node ready AND accepting tenant)
POST /admin/tenant/{tenant_id}/drain → drains tenant with optional timeout ({"timeout_ms":10000})

gRPC

Aggregate exposes a command submission API for the Gateway:

Service: aggregate.gateway.v1.CommandService
Method: SubmitCommand
Metadata: x-tenant-id (tenant routing hint)

Proto definition: aggregate.proto

Container

Build and run locally:

docker build -t cloudlysis/aggregate:local -f docker/Dockerfile.rust --build-arg PACKAGE=aggregate --build-arg BIN=aggregate .
docker compose up -d --build

Container smoke test (requires Docker installed):

sh docker/scripts/verify_aggregate_container.sh

3.2 KiB Raw Permalink Blame History