diff --git a/persys-meter/Dockerfile b/persys-meter/Dockerfile new file mode 100644 index 0000000..3976ab9 --- /dev/null +++ b/persys-meter/Dockerfile @@ -0,0 +1,32 @@ +# syntax=docker/dockerfile:1.7 + +# --- build stage --------------------------------------------------------- +FROM golang:1.25-alpine AS builder + +RUN apk add --no-cache git ca-certificates + +WORKDIR /src + +# Cache module downloads separately from source changes. +COPY go.mod go.sum* ./ +RUN go mod download + +COPY . . +# CGO disabled: clickhouse-go's native protocol driver is pure Go, so a +# static binary is possible and preferable for a distroless runtime image. +RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 \ + go build -trimpath -ldflags="-s -w" -o /out/persys-meter ./cmd/meter + +# --- runtime stage -------------------------------------------------------- +# distroless static + nonroot: no shell, no package manager, runs as an +# unprivileged user by default - minimal attack surface for a service that +# handles usage/billing data. +FROM alpine:latest + +COPY --from=builder /out/persys-meter /persys-meter + +# 9091: health/readiness/Prometheus metrics (METER_HEALTH_ADDR) +# 9092: query API (METER_API_ADDR) +EXPOSE 9091 9092 + +ENTRYPOINT ["/persys-meter"] diff --git a/persys-meter/Makefile b/persys-meter/Makefile new file mode 100644 index 0000000..de40986 --- /dev/null +++ b/persys-meter/Makefile @@ -0,0 +1,54 @@ +BINARY := persys-meter +CMD := ./cmd/meter +BIN_DIR := bin +IMAGE ?= persys-dev/persys-meter +TAG ?= latest +GO ?= go + +.PHONY: all +all: build + +.PHONY: build +build: ## Build the persys-meter binary into bin/ + CGO_ENABLED=0 $(GO) build -trimpath -ldflags="-s -w" -o $(BIN_DIR)/$(BINARY) $(CMD) + +.PHONY: run +run: ## Run persys-meter directly with `go run` (uses env vars / config defaults) + $(GO) run $(CMD) + +.PHONY: test +test: ## Run the test suite with race detection and coverage + $(GO) test ./... -race -cover + +.PHONY: tidy +tidy: ## Resolve/verify module dependencies and generate go.sum + $(GO) mod tidy + +.PHONY: fmt +fmt: ## Format all Go source + gofmt -l -w . + +.PHONY: vet +vet: ## Run go vet + $(GO) vet ./... + +.PHONY: lint +lint: fmt vet ## Format + vet (add golangci-lint here if/when it's adopted repo-wide) + +.PHONY: docker +docker: ## Build the container image + docker build -t $(IMAGE):$(TAG) . + +.PHONY: docker-push +docker-push: docker ## Build and push the container image + docker push $(IMAGE):$(TAG) + +.PHONY: clean +clean: ## Remove build artifacts + rm -rf $(BIN_DIR) + +.PHONY: help +help: ## Show this help + @grep -E '^[a-zA-Z_-]+:.*## ' $(MAKEFILE_LIST) | awk 'BEGIN {FS = ":.*## "}; {printf " \033[36m%-15s\033[0m %s\n", $$1, $$2}' + +.DEFAULT_GOAL := build diff --git a/persys-meter/README.md b/persys-meter/README.md new file mode 100644 index 0000000..1ff7627 --- /dev/null +++ b/persys-meter/README.md @@ -0,0 +1,227 @@ +# persys-meter + +Consumes per-workload resource usage events published by `persys-scheduler` +onto a Redis Stream, stores them durably in ClickHouse, and exposes them +three ways: live Prometheus metrics, a JSON query API, and raw historical +rows in ClickHouse. It's the piece that makes historical/aggregated +per-workload usage queryable at all - `compute-agent` and `persys-scheduler` +only ever expose the *latest* sample. + +## Where this fits + +``` +compute-agent --(heartbeat, real per-workload usage)--> persys-scheduler + | + XADD persys:usage:stream + | + v + persys-meter + (this service) + / | \ + Prometheus /metrics | JSON API + (live, per-workload) | (:9092) + v + ClickHouse + (durable history) +``` + +`persys-meter` is a Redis Streams **consumer group**: run as many replicas +as you want with the same `METER_CONSUMER_GROUP`, and Redis hands each one a +different slice of the stream automatically. Nothing here talks directly to +`compute-agent` or `persys-scheduler` beyond reading the stream they already +publish to. + +## What this service is - and isn't + +**Is:** the accurate, queryable record of what every workload has actually +consumed, live and historical. This is the necessary input to billing and +quota enforcement. + +**Isn't:** a billing engine or a quota enforcer. It doesn't define price +tiers, quota limits, or what happens when a limit is exceeded (deny new +workloads? throttle? alert?) - those are product decisions that belong in +whatever service makes admission/billing decisions (`persys-scheduler`, or a +dedicated billing service), consuming this service's API. Baking a specific +quota policy in here would mean guessing at a business model this service +has no way to know. + +## Running it + +Needs a reachable Redis (the same one `persys-scheduler` publishes to) and a +ClickHouse server/cluster. + +```bash +export REDIS_ADDR=localhost:6379 +export CLICKHOUSE_ADDR=localhost:9000 +make run +``` + +The `usage_events` table is created automatically on startup if it doesn't +exist (see `internal/store/clickhouse.go`). + +### Docker + +```bash +make docker # builds persys-dev/persys-meter:latest +docker run --rm \ + -e REDIS_ADDR=redis:6379 \ + -e CLICKHOUSE_ADDR=clickhouse:9000 \ + -p 9091:9091 -p 9092:9092 \ + persys-dev/persys-meter:latest +``` + +## Configuration + +Everything is an env var; every one has a default suitable for local dev +against `localhost` services. + +| Variable | Default | Notes | +|---|---|---| +| `REDIS_ADDR` | `127.0.0.1:6379` | Must be the same Redis `persys-scheduler` publishes to | +| `REDIS_PASSWORD` | _(empty)_ | | +| `REDIS_DB` | `0` | | +| `METER_REDIS_STREAM` | `persys:usage:stream` | **Must match** persys-scheduler's `METER_REDIS_STREAM` | +| `METER_CONSUMER_GROUP` | `persys-meter` | Shared across all replicas | +| `METER_CONSUMER_NAME` | `-` | Should be unique per running instance | +| `METER_READ_COUNT` | `200` | XREADGROUP COUNT (messages per read) | +| `METER_READ_BLOCK` | `5s` | XREADGROUP BLOCK duration | +| `METER_CLAIM_MIN_IDLE` | `30s` | XAUTOCLAIM: how long a message can sit pending before being reclaimed | +| `METER_CLAIM_INTERVAL` | `15s` | How often the reclaim sweep runs | +| `METER_WORKERS` | `4` | Concurrent XREADGROUP goroutines | +| `METER_DEDUPE_TTL` | `24h` | How long an event_id is remembered to catch redeliveries | +| `METER_BATCH_SIZE` | `500` | Max records per ClickHouse insert | +| `METER_BATCH_FLUSH_INTERVAL` | `2s` | Max time a partial batch waits before flushing | +| `CLICKHOUSE_ADDR` | `127.0.0.1:9000` | Native protocol port | +| `CLICKHOUSE_DATABASE` | `persys` | | +| `CLICKHOUSE_USERNAME` | `default` | | +| `CLICKHOUSE_PASSWORD` | _(empty)_ | | +| `CLICKHOUSE_TLS` | `false` | | +| `METER_RETENTION_DAYS` | `90` | TTL on `usage_events` - **tune this to your actual billing/audit retention needs**, 90 is a placeholder | +| `METER_HEALTH_ADDR` | `:9091` | Serves `/healthz`, `/readyz`, `/metrics` | +| `METER_API_ADDR` | `:9092` | Serves the JSON query API (see below) | +| `METER_API_TOKEN` | _(empty)_ | If set, required as `Authorization: Bearer ` on every API request. Empty = no auth | +| `METER_CACHE_MAX_AGE` | `5m` | How stale a workload's cached sample can be and still be considered "live" (affects both the API and Prometheus metrics) | +| `METER_CACHE_PRUNE_INTERVAL` | `2m` | How often stale cache entries are actually freed from memory | + +## Metrics (`GET :9091/metrics`) + +Two distinct sets, both under the `persys_meter_*` namespace: + +**Actual workload metrics** - what you almost certainly want for dashboards +and alerting, one series per currently-live workload: + +| Metric | Type | Meaning | +|---|---|---| +| `persys_meter_workload_cpu_percent{workload_id,node_id,workload_type}` | gauge | Latest CPU utilization % | +| `persys_meter_workload_memory_bytes{...}` | gauge | Latest resident memory usage | +| `persys_meter_workload_disk_read_bytes_total{...}` | counter | Cumulative disk bytes read (resets if the workload restarts) | +| `persys_meter_workload_disk_write_bytes_total{...}` | counter | Cumulative disk bytes written | +| `persys_meter_workload_net_rx_bytes_total{...}` | counter | Cumulative network bytes received | +| `persys_meter_workload_net_tx_bytes_total{...}` | counter | Cumulative network bytes transmitted | +| `persys_meter_workload_sample_age_seconds{...}` | gauge | Seconds since the last sample - alert on this being persistently high to catch an agent that's stopped reporting | + +These are emitted by a custom Prometheus collector reading the shared +in-memory cache (`internal/cache`), so a workload that stops reporting +simply disappears from scrapes after `METER_CACHE_MAX_AGE` rather than +leaving a stale series behind forever. + +**Pipeline health metrics** - about persys-meter itself, not workloads: + +| Metric | Meaning | +|---|---| +| `persys_meter_events_consumed_total{stream}` | Messages read via XREADGROUP | +| `persys_meter_events_deduped_total{stream}` | Skipped as a redelivery | +| `persys_meter_events_parse_failed_total{stream}` | Malformed payloads (logged at Error level too) | +| `persys_meter_events_written_total{stream}` | Successfully written to ClickHouse | +| `persys_meter_batch_write_failed_total{stream}` | Failed batch writes (left un-acked for retry) | +| `persys_meter_messages_reclaimed_total{stream}` | Recovered via XAUTOCLAIM from a stalled consumer | +| `persys_meter_batch_flush_duration_seconds{stream,result}` | ClickHouse write latency | +| `persys_meter_batch_size{stream}` | Records per flushed batch | +| `persys_meter_consumer_lag_seconds{stream}` | Time between an event being reported and processed | + +## Query API (`:9092`) + +JSON over HTTP. Set `METER_API_TOKEN` and send `Authorization: Bearer +` in production - unauthenticated by default for local dev only. + +| Endpoint | Description | +|---|---| +| `GET /v1/workloads` | Every live workload's latest sample. `?workload_type=container\|vm` to filter | +| `GET /v1/workloads/{id}` | Latest sample for one workload. `404` if none within `METER_CACHE_MAX_AGE` | +| `GET /v1/workloads/{id}/history?from=&to=&limit=` | Raw historical samples from ClickHouse, newest first. `from`/`to` are RFC3339, default to the last 7 days | +| `GET /v1/workloads/{id}/summary?from=&to=` | Aggregated usage over a window - see below | +| `GET /v1/nodes/{id}/workloads` | Live workloads on a given node | + +### Example: usage summary + +``` +GET /v1/workloads/wl-abc123/summary?from=2026-07-01T00:00:00Z&to=2026-07-08T00:00:00Z +``` + +```json +{ + "workload_id": "wl-abc123", + "from": "2026-07-01T00:00:00Z", + "to": "2026-07-08T00:00:00Z", + "sample_count": 40320, + "avg_cpu_percent": 12.4, + "max_cpu_percent": 87.1, + "avg_memory_bytes": 268435456, + "max_memory_bytes": 402653184, + "disk_read_bytes_delta": 1073741824, + "disk_write_bytes_delta": 536870912, + "net_rx_bytes_delta": 209715200, + "net_tx_bytes_delta": 104857600, + "first_sample": "2026-07-01T00:00:03Z", + "last_sample": "2026-07-07T23:59:58Z", + "estimated_cpu_core_seconds": 75277.44 +} +``` + +**Read the caveats before wiring this into anything that charges money:** + +- `*_bytes_delta` fields are `max(counter) - min(counter)` over the window. + Since the underlying counters are cumulative *since the workload's runtime + started* (see `compute-agent`), a restart during the window resets them to + zero, which understates (or even negatives) the delta. This isn't + silently patched over - it's a known limitation documented on + `store.UsageSummary`. A more correct version would track restarts + explicitly and sum deltas between them. +- `estimated_cpu_core_seconds` is `avg(cpu_percent)/100 * window_seconds` - + a simple approximation, not a rigorous integral over irregularly spaced + samples. Fine as a first cut; revisit if billing needs tighter accuracy. +- `sample_count: 0` means no data was reported in the window at all - it + does **not** mean usage was zero. Every other field will also be + zero-valued in that case; check `sample_count` first. + +## Known limitations / things to revisit + +- **Cross-repo schema coupling.** `internal/ingest/event.go` hand-mirrors + the JSON shape `persys-scheduler` publishes (`internal/scheduler/usage_stream.go` + and its `models.WorkloadUsage`). The two aren't shared via a common + package (separate Go modules) - if the scheduler's shape changes, this + file needs a matching manual update, and nothing will catch a drift at + compile time. +- **Dedup is TTL-bounded** (`METER_DEDUPE_TTL`, default 24h). A duplicate + delivery arriving after the TTL expires would be double-counted. In + practice redeliveries happen within seconds-to-minutes of + `METER_CLAIM_MIN_IDLE`, not hours, so this is a deliberate, bounded + tradeoff, not an oversight. +- **Retention default (90 days) is a placeholder** - set + `METER_RETENTION_DAYS` to match your actual billing/audit cycle. +- **Not compiled in the environment that generated it.** Every non-trivial + third-party API call here (`go-redis` streams commands, `clickhouse-go`'s + `Query`/`QueryRow`/`PrepareBatch`, `client_golang`'s custom Collector) was + checked against the real tagged source rather than written from memory, + but this has not been run through `go build`/`go vet`. Run `make tidy && + make build && make vet` before deploying. + +## Development + +```bash +make help # list targets +make tidy # resolve dependencies, generate go.sum (needed once, and after any go.mod edit) +make build # compile to bin/persys-meter +make test # go test ./... -race -cover +make lint # gofmt + go vet +``` diff --git a/persys-meter/cmd/meter/main.go b/persys-meter/cmd/meter/main.go new file mode 100644 index 0000000..d627797 --- /dev/null +++ b/persys-meter/cmd/meter/main.go @@ -0,0 +1,167 @@ +// Command persys-meter consumes per-workload usage events published by +// persys-scheduler onto a Redis Stream and durably stores them in +// ClickHouse, so downstream billing/analytics can query historical +// per-workload resource usage - something neither compute-agent nor +// persys-scheduler retain on their own (they only ever expose the latest +// sample). +package main + +import ( + "context" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/persys-dev/persys-cloud/persys-meter/internal/api" + "github.com/persys-dev/persys-cloud/persys-meter/internal/cache" + "github.com/persys-dev/persys-cloud/persys-meter/internal/config" + "github.com/persys-dev/persys-cloud/persys-meter/internal/consumer" + "github.com/persys-dev/persys-cloud/persys-meter/internal/health" + "github.com/persys-dev/persys-cloud/persys-meter/internal/metrics" + "github.com/persys-dev/persys-cloud/persys-meter/internal/store" + "github.com/redis/go-redis/v9" + "github.com/sirupsen/logrus" +) + +// compositeChecker reports readiness across every dependency this service +// actually needs: Redis (to consume) and ClickHouse (to write). Either one +// being unreachable means the service can't do useful work right now. +type compositeChecker struct { + redisClient *redis.Client + chStore *store.ClickHouseStore +} + +func (c *compositeChecker) Ready(ctx context.Context) error { + if err := c.redisClient.Ping(ctx).Err(); err != nil { + return err + } + return c.chStore.Ping(ctx) +} + +func main() { + logger := logrus.New() + logger.SetFormatter(&logrus.JSONFormatter{}) + log := logger.WithField("component", "cmd.meter") + + cfg, err := config.Load() + if err != nil { + log.WithError(err).Fatal("failed to load configuration") + } + + metrics.Register() + + redisClient := redis.NewClient(&redis.Options{ + Addr: cfg.RedisAddr, + Password: cfg.RedisPassword, + DB: cfg.RedisDB, + }) + defer redisClient.Close() + + { + pingCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := redisClient.Ping(pingCtx).Err(); err != nil { + log.WithError(err).Fatal("failed to connect to redis") + } + } + log.WithField("addr", cfg.RedisAddr).Info("connected to redis") + + chStore, err := store.NewClickHouseStore(store.ClickHouseConfig{ + Addr: cfg.ClickHouseAddr, + Database: cfg.ClickHouseDatabase, + Username: cfg.ClickHouseUsername, + Password: cfg.ClickHousePassword, + TLS: cfg.ClickHouseTLS, + RetentionDays: cfg.RetentionDays, + }, log) + if err != nil { + log.WithError(err).Fatal("failed to connect to clickhouse") + } + defer chStore.Close() + log.WithField("addr", cfg.ClickHouseAddr).Info("connected to clickhouse") + + initCtx, initCancel := context.WithTimeout(context.Background(), 15*time.Second) + if err := chStore.Init(initCtx); err != nil { + initCancel() + log.WithError(err).Fatal("failed to initialize clickhouse schema") + } + initCancel() + + // latestCache backs the per-workload Prometheus metrics AND the API's + // live endpoints - one shared, in-memory view of "what is every + // workload doing right now", kept current by the consumer. + latestCache := cache.New() + + workloadCollector := metrics.NewWorkloadCollector(latestCache, cfg.CacheMaxAge) + metrics.RegisterWorkloadCollector(workloadCollector) + + c := consumer.New(cfg, redisClient, chStore, latestCache, log) + + healthSrv := health.Serve(cfg.HealthAddr, &compositeChecker{redisClient: redisClient, chStore: chStore}, log) + defer func() { + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer shutdownCancel() + if err := healthSrv.Shutdown(shutdownCtx); err != nil && err != http.ErrServerClosed { + log.WithError(err).Warn("health server shutdown error") + } + }() + + // chStore satisfies store.QueryStore (History/Summary) as well as + // store.Store - the API only needs the read side. + queryAPI := api.New(latestCache, chStore, log, cfg.APIToken, cfg.CacheMaxAge) + apiSrv := &http.Server{Addr: cfg.APIAddr, Handler: queryAPI.Handler()} + go func() { + if err := apiSrv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.WithError(err).Error("API server stopped unexpectedly") + } + }() + if cfg.APIToken == "" { + log.WithField("addr", cfg.APIAddr).Warn("query API listening with no auth token configured (METER_API_TOKEN unset) - fine for local dev, not for anything reachable outside a trusted network") + } else { + log.WithField("addr", cfg.APIAddr).Info("query API listening") + } + defer func() { + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer shutdownCancel() + if err := apiSrv.Shutdown(shutdownCtx); err != nil && err != http.ErrServerClosed { + log.WithError(err).Warn("API server shutdown error") + } + }() + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + // Periodically evict cache entries for workloads that have stopped + // reporting (deleted workloads, decommissioned nodes, etc.) so the + // in-memory cache doesn't grow forever. Independent of CacheMaxAge, + // which only controls what All()/ByNode() *return*, not what's *kept*. + go func() { + ticker := time.NewTicker(cfg.CachePruneInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if removed := latestCache.Prune(cfg.CacheMaxAge); removed > 0 { + log.WithField("removed", removed).Debug("pruned stale entries from usage cache") + } + } + } + }() + + log.WithFields(logrus.Fields{ + "stream": cfg.StreamName, + "consumer_group": cfg.ConsumerGroup, + "consumer_name": cfg.ConsumerName, + "workers": cfg.Workers, + }).Info("persys-meter starting") + + if err := c.Run(ctx); err != nil { + log.WithError(err).Fatal("consumer exited with error") + } + + log.Info("persys-meter stopped cleanly") +} diff --git a/persys-meter/docker-compose.yml b/persys-meter/docker-compose.yml new file mode 100644 index 0000000..e57bb8e --- /dev/null +++ b/persys-meter/docker-compose.yml @@ -0,0 +1,12 @@ +services: + clickhouse: + image: clickhouse/clickhouse-server:latest + container_name: clickhouse + restart: unless-stopped + ports: + - "8123:8123" # HTTP interface (for queries, UI, etc.) + - "9000:9000" # Native TCP interface (for clients like clickhouse-client) + environment: + - CLICKHOUSE_DB=persys + - CLICKHOUSE_USER=default + - CLICKHOUSE_PASSWORD=persys # Change this! diff --git a/persys-meter/go.mod b/persys-meter/go.mod new file mode 100644 index 0000000..c8064fb --- /dev/null +++ b/persys-meter/go.mod @@ -0,0 +1,37 @@ +module github.com/persys-dev/persys-cloud/persys-meter + +go 1.24 + +require ( + github.com/ClickHouse/clickhouse-go/v2 v2.30.0 + github.com/prometheus/client_golang v1.20.5 + github.com/redis/go-redis/v9 v9.19.0 + github.com/sirupsen/logrus v1.9.3 +) + +require ( + github.com/ClickHouse/ch-go v0.61.5 // indirect + github.com/andybalholm/brotli v1.1.1 // indirect + github.com/beorn7/perks v1.0.1 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/go-faster/city v1.0.1 // indirect + github.com/go-faster/errors v0.7.1 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/klauspost/compress v1.17.9 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/paulmach/orb v0.11.1 // indirect + github.com/pierrec/lz4/v4 v4.1.21 // indirect + github.com/pkg/errors v0.9.1 // indirect + github.com/prometheus/client_model v0.6.1 // indirect + github.com/prometheus/common v0.55.0 // indirect + github.com/prometheus/procfs v0.15.1 // indirect + github.com/segmentio/asm v1.2.0 // indirect + github.com/shopspring/decimal v1.4.0 // indirect + go.opentelemetry.io/otel v1.26.0 // indirect + go.opentelemetry.io/otel/trace v1.26.0 // indirect + go.uber.org/atomic v1.11.0 // indirect + golang.org/x/sys v0.30.0 // indirect + google.golang.org/protobuf v1.34.2 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) + diff --git a/persys-meter/go.sum b/persys-meter/go.sum new file mode 100644 index 0000000..5f5f88b --- /dev/null +++ b/persys-meter/go.sum @@ -0,0 +1,146 @@ +github.com/ClickHouse/ch-go v0.61.5 h1:zwR8QbYI0tsMiEcze/uIMK+Tz1D3XZXLdNrlaOpeEI4= +github.com/ClickHouse/ch-go v0.61.5/go.mod h1:s1LJW/F/LcFs5HJnuogFMta50kKDO0lf9zzfrbl0RQg= +github.com/ClickHouse/clickhouse-go/v2 v2.30.0 h1:AG4D/hW39qa58+JHQIFOSnxyL46H6h2lrmGGk17dhFo= +github.com/ClickHouse/clickhouse-go/v2 v2.30.0/go.mod h1:i9ZQAojcayW3RsdCb3YR+n+wC2h65eJsZCscZ1Z1wyo= +github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA= +github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA= +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/go-faster/city v1.0.1 h1:4WAxSZ3V2Ws4QRDrscLEDcibJY8uf41H6AhXDrNDcGw= +github.com/go-faster/city v1.0.1/go.mod h1:jKcUJId49qdW3L1qKHH/3wPeUstCVpVSXTM6vO3VcTw= +github.com/go-faster/errors v0.7.1 h1:MkJTnDoEdi9pDabt1dpWf7AA8/BaSYZqibYyhZ20AYg= +github.com/go-faster/errors v0.7.1/go.mod h1:5ySTjWFiphBs07IKuiL69nxdfd5+fzh1u7FPGZP2quo= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= +github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= +github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= +github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= +github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= +github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk= +github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA= +github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw= +github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= +github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= +github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= +github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/paulmach/orb v0.11.1 h1:3koVegMC4X/WeiXYz9iswopaTwMem53NzTJuTF20JzU= +github.com/paulmach/orb v0.11.1/go.mod h1:5mULz1xQfs3bmQm63QEJA6lNGujuRafwA5S/EnuLaLU= +github.com/paulmach/protoscan v0.2.1/go.mod h1:SpcSwydNLrxUGSDvXvO0P7g7AuhJ7lcKfDlhJCDw2gY= +github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ= +github.com/pierrec/lz4/v4 v4.1.21/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= +github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v1.20.5 h1:cxppBPuYhUnsO6yo/aoRol4L7q7UFfdm+bR9r+8l63Y= +github.com/prometheus/client_golang v1.20.5/go.mod h1:PIEt8X02hGcP8JWbeHyeZ53Y/jReSnHgO035n//V5WE= +github.com/prometheus/client_model v0.6.1 h1:ZKSh/rekM+n3CeS952MLRAdFwIKqeY8b62p8ais2e9E= +github.com/prometheus/client_model v0.6.1/go.mod h1:OrxVMOVHjw3lKMa8+x6HeMGkHMQyHDk9E3jmP2AmGiY= +github.com/prometheus/common v0.55.0 h1:KEi6DK7lXW/m7Ig5i47x0vRzuBsHuvJdi5ee6Y3G1dc= +github.com/prometheus/common v0.55.0/go.mod h1:2SECS4xJG1kd8XF9IcM1gMX6510RAEL65zxzNImwdc8= +github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc= +github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk= +github.com/redis/go-redis/v9 v9.19.0 h1:XPVaaPSnG6RhYf7p+rmSa9zZfeVAnWsH5h3lxthOm/k= +github.com/redis/go-redis/v9 v9.19.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= +github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= +github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= +github.com/segmentio/asm v1.2.0 h1:9BQrFxC+YOHJlTlHGkTrFWf59nbL3XnCoFLTwDCI7ys= +github.com/segmentio/asm v1.2.0/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs= +github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k= +github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= +github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= +github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= +github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk= +github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI= +github.com/xdg-go/scram v1.1.1/go.mod h1:RaEWvsqvNKKvBPvcKeFjrG2cJqOkHTiyTpzz23ni57g= +github.com/xdg-go/stringprep v1.0.3/go.mod h1:W3f5j4i+9rC0kuIEJL0ky1VpHXQU3ocBgklLGvcBnW8= +github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= +github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= +github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7JulP+udvsHwJoVG1YGAP6VLg4y9I5dyZdqmA= +github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= +go.mongodb.org/mongo-driver v1.11.4/go.mod h1:PTSz5yu21bkT/wXpkS7WR5f0ddqw5quethTUn9WM+2g= +go.opentelemetry.io/otel v1.26.0 h1:LQwgL5s/1W7YiiRwxf03QGnWLb2HW4pLiAhaA5cZXBs= +go.opentelemetry.io/otel v1.26.0/go.mod h1:UmLkJHUAidDval2EICqBMbnAd0/m2vmpf/dAM+fvFs4= +go.opentelemetry.io/otel/trace v1.26.0 h1:1ieeAUb4y0TE26jUFrCIXKpTuVK7uJGN9/Z/2LP5sQA= +go.opentelemetry.io/otel/trace v1.26.0/go.mod h1:4iDxvGDQuUkHve82hJJ8UqrwswHYsZuWCBllGV2U2y0= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= +golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= +golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= +golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= +golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc= +golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= +golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= +google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= +google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg= +google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/persys-meter/internal/api/api.go b/persys-meter/internal/api/api.go new file mode 100644 index 0000000..167df51 --- /dev/null +++ b/persys-meter/internal/api/api.go @@ -0,0 +1,268 @@ +// Package api exposes an HTTP/JSON query surface over the usage data +// persys-meter ingests - both the live in-memory cache (for "what's this +// workload doing right now") and ClickHouse (for history and the +// aggregated summaries billing/quota enforcement is expected to build on). +// +// This is intentionally a read-only query API. It does not define or +// enforce quotas: quota *policy* (what the limits are, what happens when +// one is exceeded - deny new workloads? throttle? alert?) is a product +// decision that belongs to whichever service makes admission/enforcement +// decisions (persys-scheduler, or a dedicated billing service), not this +// one. What this API gives that caller is an accurate, queryable answer to +// "how much has this workload actually used" - the necessary input to any +// such decision, not the decision itself. +package api + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "strconv" + "time" + + "github.com/persys-dev/persys-cloud/persys-meter/internal/cache" + "github.com/persys-dev/persys-cloud/persys-meter/internal/store" + "github.com/sirupsen/logrus" +) + +// defaultHistoryWindow is used when a /history or /summary request omits +// ?from - a week is a reasonable default for "recent history" without +// forcing every caller to compute a timestamp for the common case. +const defaultHistoryWindow = 7 * 24 * time.Hour + +// API holds the dependencies every handler needs. +type API struct { + cache *cache.LatestCache + query store.QueryStore + logger *logrus.Entry + token string // if non-empty, required as "Bearer " on every request + cacheMaxAge time.Duration // how stale a cache entry can be and still be considered "live" +} + +// New constructs an API. token may be empty to disable auth entirely (fine +// for local dev; for anything else, put this behind the same +// network/mTLS boundary the rest of Persys uses internally, or set a +// token - see README). cacheMaxAge should match whatever value is passed +// to metrics.NewWorkloadCollector, so the API and the Prometheus metrics +// agree on what counts as a "live" workload. +func New(c *cache.LatestCache, q store.QueryStore, logger *logrus.Entry, token string, cacheMaxAge time.Duration) *API { + return &API{cache: c, query: q, logger: logger, token: token, cacheMaxAge: cacheMaxAge} +} + +// Handler builds the HTTP handler for the API, including auth middleware. +// Uses Go 1.22's method+pattern ServeMux (e.g. "GET /v1/workloads/{id}") +// rather than pulling in a router dependency for a handful of routes. +func (a *API) Handler() http.Handler { + mux := http.NewServeMux() + + mux.HandleFunc("GET /v1/workloads", a.listWorkloads) + mux.HandleFunc("GET /v1/workloads/{workloadID}", a.getWorkload) + mux.HandleFunc("GET /v1/workloads/{workloadID}/history", a.getHistory) + mux.HandleFunc("GET /v1/workloads/{workloadID}/summary", a.getSummary) + mux.HandleFunc("GET /v1/nodes/{nodeID}/workloads", a.listWorkloadsByNode) + + return a.withAuth(mux) +} + +// withAuth requires "Authorization: Bearer " on every request when a +// token is configured. A no-op passthrough otherwise. +func (a *API) withAuth(next http.Handler) http.Handler { + if a.token == "" { + return next + } + want := "Bearer " + a.token + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != want { + writeError(w, http.StatusUnauthorized, "unauthorized") + return + } + next.ServeHTTP(w, r) + }) +} + +// --- handlers --- + +// workloadView is the JSON shape for a single workload's live state - +// cache.Entry reshaped so the API's response format isn't just whatever +// internal struct layout the cache happens to use today. +type workloadView struct { + WorkloadID string `json:"workload_id"` + NodeID string `json:"node_id"` + RevisionID string `json:"revision_id,omitempty"` + WorkloadType string `json:"workload_type"` + CPUPercent float64 `json:"cpu_percent"` + MemoryBytes int64 `json:"memory_bytes"` + DiskReadBytes int64 `json:"disk_read_bytes"` + DiskWriteBytes int64 `json:"disk_write_bytes"` + NetRXBytes int64 `json:"net_rx_bytes"` + NetTXBytes int64 `json:"net_tx_bytes"` + ReportedAt time.Time `json:"reported_at"` + LastSeenAt time.Time `json:"last_seen_at"` + SampleAgeSec float64 `json:"sample_age_seconds"` +} + +func toView(e cache.Entry) workloadView { + return workloadView{ + WorkloadID: e.Record.WorkloadID, + NodeID: e.Record.NodeID, + RevisionID: e.Record.RevisionID, + WorkloadType: e.Record.WorkloadType, + CPUPercent: e.Record.CPUPercent, + MemoryBytes: e.Record.MemoryBytes, + DiskReadBytes: e.Record.DiskReadBytes, + DiskWriteBytes: e.Record.DiskWriteBytes, + NetRXBytes: e.Record.NetRXBytes, + NetTXBytes: e.Record.NetTXBytes, + ReportedAt: e.Record.ReportedAt, + LastSeenAt: e.UpdatedAt, + SampleAgeSec: time.Since(e.UpdatedAt).Seconds(), + } +} + +// GET /v1/workloads +// Lists every workload with a non-stale sample. Optional ?workload_type= +// filters by type (e.g. "container", "vm"). +func (a *API) listWorkloads(w http.ResponseWriter, r *http.Request) { + entries := a.cache.All(a.cacheMaxAge) + workloadType := r.URL.Query().Get("workload_type") + + views := make([]workloadView, 0, len(entries)) + for _, e := range entries { + if workloadType != "" && e.Record.WorkloadType != workloadType { + continue + } + views = append(views, toView(e)) + } + writeJSON(w, http.StatusOK, map[string]any{"workloads": views, "count": len(views)}) +} + +// GET /v1/nodes/{nodeID}/workloads +func (a *API) listWorkloadsByNode(w http.ResponseWriter, r *http.Request) { + nodeID := r.PathValue("nodeID") + entries := a.cache.ByNode(nodeID, a.cacheMaxAge) + + views := make([]workloadView, 0, len(entries)) + for _, e := range entries { + views = append(views, toView(e)) + } + writeJSON(w, http.StatusOK, map[string]any{"node_id": nodeID, "workloads": views, "count": len(views)}) +} + +// GET /v1/workloads/{workloadID} +// Live latest sample for one workload. 404 if we have no (non-stale) +// sample for it - which is also what you'd see for a workload ID that +// never existed, since this cache has no concept of "known but idle". +func (a *API) getWorkload(w http.ResponseWriter, r *http.Request) { + workloadID := r.PathValue("workloadID") + entry, ok := a.cache.Get(workloadID) + if !ok || time.Since(entry.UpdatedAt) > a.cacheMaxAge { + writeError(w, http.StatusNotFound, "no current usage sample for workload "+workloadID) + return + } + writeJSON(w, http.StatusOK, toView(entry)) +} + +// GET /v1/workloads/{workloadID}/history?from=RFC3339&to=RFC3339&limit=N +// Raw historical samples from ClickHouse, newest first. +func (a *API) getHistory(w http.ResponseWriter, r *http.Request) { + workloadID := r.PathValue("workloadID") + + from, to, err := parseWindow(r, defaultHistoryWindow) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + + limit := 1000 + if raw := r.URL.Query().Get("limit"); raw != "" { + parsed, err := strconv.Atoi(raw) + if err != nil || parsed <= 0 { + writeError(w, http.StatusBadRequest, "limit must be a positive integer") + return + } + limit = parsed + } + + ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second) + defer cancel() + + records, err := a.query.History(ctx, workloadID, from, to, limit) + if err != nil { + a.logger.WithError(err).WithField("workload_id", workloadID).Error("history query failed") + writeError(w, http.StatusInternalServerError, "failed to query usage history") + return + } + + writeJSON(w, http.StatusOK, map[string]any{ + "workload_id": workloadID, + "from": from, + "to": to, + "count": len(records), + "samples": records, + }) +} + +// GET /v1/workloads/{workloadID}/summary?from=RFC3339&to=RFC3339 +// Aggregated usage over a window - the primitive for billing/quota +// enforcement. See store.UsageSummary's doc comments for accuracy caveats +// (counter resets across restarts, CPU-seconds approximation) before +// wiring this into anything that charges money or denies workloads. +func (a *API) getSummary(w http.ResponseWriter, r *http.Request) { + workloadID := r.PathValue("workloadID") + + from, to, err := parseWindow(r, defaultHistoryWindow) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + + ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second) + defer cancel() + + summary, err := a.query.Summary(ctx, workloadID, from, to) + if err != nil { + a.logger.WithError(err).WithField("workload_id", workloadID).Error("summary query failed") + writeError(w, http.StatusInternalServerError, "failed to compute usage summary") + return + } + + writeJSON(w, http.StatusOK, summary) +} + +// --- helpers --- + +// parseWindow reads ?from and ?to (RFC3339) from the request, defaulting to +// [now-defaultWindow, now] when omitted, and validates from < to. +func parseWindow(r *http.Request, defaultWindow time.Duration) (from, to time.Time, err error) { + now := time.Now().UTC() + to = now + from = now.Add(-defaultWindow) + + if raw := r.URL.Query().Get("to"); raw != "" { + to, err = time.Parse(time.RFC3339, raw) + if err != nil { + return from, to, errors.New("to must be an RFC3339 timestamp") + } + } + if raw := r.URL.Query().Get("from"); raw != "" { + from, err = time.Parse(time.RFC3339, raw) + if err != nil { + return from, to, errors.New("from must be an RFC3339 timestamp") + } + } + if !from.Before(to) { + return from, to, errors.New("from must be before to") + } + return from, to, nil +} + +func writeJSON(w http.ResponseWriter, status int, body any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(body) +} + +func writeError(w http.ResponseWriter, status int, message string) { + writeJSON(w, status, map[string]string{"error": message}) +} diff --git a/persys-meter/internal/cache/cache.go b/persys-meter/internal/cache/cache.go new file mode 100644 index 0000000..370b848 --- /dev/null +++ b/persys-meter/internal/cache/cache.go @@ -0,0 +1,114 @@ +// Package cache holds the most recent usage sample per workload in memory. +// +// This exists alongside ClickHouse, not instead of it: ClickHouse is the +// durable, queryable history; this cache is a fast, always-current view of +// "what is every workload doing right now" - used to back the per-workload +// Prometheus gauges and the low-latency parts of the query API without +// hitting ClickHouse on every scrape/request. +package cache + +import ( + "sync" + "time" + + "github.com/persys-dev/persys-cloud/persys-meter/internal/ingest" +) + +// Entry is one workload's latest known usage sample, plus when this cache +// last heard about it (which may lag the sample's own ReportedAt slightly, +// but is what staleness/eviction decisions are based on). +type Entry struct { + Record ingest.UsageRecord + UpdatedAt time.Time +} + +// LatestCache is a concurrency-safe map of workload ID -> latest Entry. +type LatestCache struct { + mu sync.RWMutex + entries map[string]Entry +} + +// New creates an empty LatestCache. +func New() *LatestCache { + return &LatestCache{entries: make(map[string]Entry)} +} + +// Set records r as the latest known sample for its workload, overwriting +// any previous entry. Safe to call for a duplicate/redelivered event - it's +// just a redundant write of the same data. +func (c *LatestCache) Set(r ingest.UsageRecord) { + c.mu.Lock() + defer c.mu.Unlock() + c.entries[r.WorkloadID] = Entry{Record: r, UpdatedAt: time.Now()} +} + +// Get returns the latest entry for a workload, if known. +func (c *LatestCache) Get(workloadID string) (Entry, bool) { + c.mu.RLock() + defer c.mu.RUnlock() + e, ok := c.entries[workloadID] + return e, ok +} + +// All returns every entry last updated within maxAge. Pass 0 to disable +// staleness filtering and return everything regardless of age. +// +// Filtering matters here because nothing currently tells this cache when a +// workload is deleted - without a staleness cutoff, a Prometheus scrape (or +// a `/v1/workloads` API call) would keep showing a workload's last-ever +// values forever, long after it stopped existing. +func (c *LatestCache) All(maxAge time.Duration) []Entry { + c.mu.RLock() + defer c.mu.RUnlock() + + out := make([]Entry, 0, len(c.entries)) + now := time.Now() + for _, e := range c.entries { + if maxAge > 0 && now.Sub(e.UpdatedAt) > maxAge { + continue + } + out = append(out, e) + } + return out +} + +// ByNode returns every non-stale entry (per the same maxAge rule as All) +// whose NodeID matches nodeID. +func (c *LatestCache) ByNode(nodeID string, maxAge time.Duration) []Entry { + all := c.All(maxAge) + out := all[:0] // safe in-place filter: write index never exceeds read index + for _, e := range all { + if e.Record.NodeID == nodeID { + out = append(out, e) + } + } + return out +} + +// Prune deletes entries not updated within maxAge and returns how many were +// removed. Call this periodically (see cmd/meter/main.go) to bound memory +// use for workloads that have been deleted and will never report again - +// All()'s filtering alone only hides stale entries from callers, it doesn't +// reclaim their memory. +func (c *LatestCache) Prune(maxAge time.Duration) int { + c.mu.Lock() + defer c.mu.Unlock() + + now := time.Now() + removed := 0 + for id, e := range c.entries { + if now.Sub(e.UpdatedAt) > maxAge { + delete(c.entries, id) + removed++ + } + } + return removed +} + +// Len reports the current number of entries, stale or not - useful as a +// cheap gauge of cache size for diagnostics. +func (c *LatestCache) Len() int { + c.mu.RLock() + defer c.mu.RUnlock() + return len(c.entries) +} diff --git a/persys-meter/internal/config/config.go b/persys-meter/internal/config/config.go new file mode 100644 index 0000000..e25c386 --- /dev/null +++ b/persys-meter/internal/config/config.go @@ -0,0 +1,186 @@ +package config + +import ( + "fmt" + "os" + "strconv" + "strings" + "time" +) + +// Config holds all runtime configuration for persys-meter, loaded from +// environment variables. Every field has a sane default so the service can +// run against a local Redis + ClickHouse with zero configuration. +type Config struct { + // Redis - must point at the same Redis instance persys-scheduler + // publishes usage events to. + RedisAddr string + RedisPassword string + RedisDB int + + // Stream / consumer group. StreamName MUST match persys-scheduler's + // METER_REDIS_STREAM (default on both sides: "persys:usage:stream"). + StreamName string + ConsumerGroup string + // ConsumerName should be unique per running instance/pod. Defaults to + // "-" if not set, which is normally enough for k8s + // deployments (pod hostname is already unique). + ConsumerName string + + // XREADGROUP tuning. + ReadCount int64 // COUNT: max messages to pull per read + ReadBlock time.Duration // BLOCK: how long to wait for new messages + + // XAUTOCLAIM tuning - recovers messages left pending by a consumer that + // crashed or hung before ack'ing (e.g. a killed pod). + ClaimMinIdle time.Duration + ClaimInterval time.Duration + + // Number of concurrent XREADGROUP worker goroutines. + Workers int + + // Dedup guard TTL. Each event_id gets a Redis SETNX key with this TTL + // before being handed to the batch writer; a redelivered/duplicate event + // is recognized and acked without being written twice. + DedupeTTL time.Duration + + // Batch writer tuning - ClickHouse strongly prefers batched inserts over + // one-row-at-a-time. A batch flushes when either threshold is hit first. + BatchSize int + BatchFlushInterval time.Duration + + // ClickHouse connection. + ClickHouseAddr string + ClickHouseDatabase string + ClickHouseUsername string + ClickHousePassword string + ClickHouseTLS bool + // RetentionDays controls the TTL clause on the usage_events table. + // Metering/billing use cases typically need longer retention than raw + // ops telemetry - tune this to your billing cycle / audit requirements. + RetentionDays int + + // HTTP server exposing /healthz, /readyz, and /metrics (Prometheus). + HealthAddr string + + // HTTP query API - GET /v1/workloads, /v1/workloads/{id}/history, etc. + // Deliberately a separate address/port from HealthAddr: health/metrics + // are typically scraped by cluster infra, while this API is meant to be + // called by application logic (a billing service, or persys-scheduler + // making a quota decision) - keeping them separate lets you apply + // different network policy/auth to each. + APIAddr string + // APIToken, if set, is required as "Authorization: Bearer " on + // every API request. Empty disables auth entirely - fine for local + // dev, but put this behind your existing internal network/mTLS + // boundary (or set a token) for anything else. See README. + APIToken string + + // CacheMaxAge bounds how stale the in-memory latest-usage cache entry + // for a workload can be and still be considered "live" - used by both + // the API's live endpoints and the per-workload Prometheus metrics, so + // the two agree on what counts as live. + CacheMaxAge time.Duration + // CachePruneInterval controls how often stale cache entries are + // actually evicted (freeing memory for workloads that will never + // report again, e.g. deleted ones) - independent of CacheMaxAge, which + // only controls what's *returned*, not what's *retained*. + CachePruneInterval time.Duration +} + +// Load reads configuration from the environment, applying defaults for +// anything unset. +func Load() (*Config, error) { + consumerName := envOr("METER_CONSUMER_NAME", "") + if consumerName == "" { + host, err := os.Hostname() + if err != nil || host == "" { + host = "persys-meter" + } + consumerName = fmt.Sprintf("%s-%d", host, os.Getpid()) + } + + cfg := &Config{ + RedisAddr: envOr("REDIS_ADDR", "redis:6379"), + RedisPassword: envOr("REDIS_PASSWORD", ""), + RedisDB: envIntOr("REDIS_DB", 1), + + StreamName: envOr("METER_REDIS_STREAM", "persys:usage:stream"), + ConsumerGroup: envOr("METER_CONSUMER_GROUP", "persys-meter"), + ConsumerName: consumerName, + + ReadCount: int64(envIntOr("METER_READ_COUNT", 200)), + ReadBlock: envDurationOr("METER_READ_BLOCK", 5*time.Second), + + ClaimMinIdle: envDurationOr("METER_CLAIM_MIN_IDLE", 30*time.Second), + ClaimInterval: envDurationOr("METER_CLAIM_INTERVAL", 15*time.Second), + + Workers: envIntOr("METER_WORKERS", 4), + + DedupeTTL: envDurationOr("METER_DEDUPE_TTL", 24*time.Hour), + + BatchSize: envIntOr("METER_BATCH_SIZE", 500), + BatchFlushInterval: envDurationOr("METER_BATCH_FLUSH_INTERVAL", 2*time.Second), + + ClickHouseAddr: envOr("CLICKHOUSE_ADDR", "clickhouse:9000"), + ClickHouseDatabase: envOr("CLICKHOUSE_DATABASE", "persys"), + ClickHouseUsername: envOr("CLICKHOUSE_USERNAME", "default"), + ClickHousePassword: envOr("CLICKHOUSE_PASSWORD", "persys"), + ClickHouseTLS: envBoolOr("CLICKHOUSE_TLS", false), + RetentionDays: envIntOr("METER_RETENTION_DAYS", 90), + + HealthAddr: envOr("METER_HEALTH_ADDR", ":9091"), + + APIAddr: envOr("METER_API_ADDR", ":9092"), + APIToken: envOr("METER_API_TOKEN", ""), + + CacheMaxAge: envDurationOr("METER_CACHE_MAX_AGE", 5*time.Minute), + CachePruneInterval: envDurationOr("METER_CACHE_PRUNE_INTERVAL", 2*time.Minute), + } + + if cfg.Workers < 1 { + return nil, fmt.Errorf("METER_WORKERS must be >= 1, got %d", cfg.Workers) + } + if cfg.BatchSize < 1 { + return nil, fmt.Errorf("METER_BATCH_SIZE must be >= 1, got %d", cfg.BatchSize) + } + if strings.TrimSpace(cfg.StreamName) == "" { + return nil, fmt.Errorf("METER_REDIS_STREAM must not be empty") + } + + return cfg, nil +} + +func envOr(key, fallback string) string { + if v, ok := os.LookupEnv(key); ok && strings.TrimSpace(v) != "" { + return v + } + return fallback +} + +func envBoolOr(key string, fallback bool) bool { + if v, ok := os.LookupEnv(key); ok { + if b, err := strconv.ParseBool(v); err == nil { + return b + } + } + return fallback +} + +func envIntOr(key string, fallback int) int { + if v, ok := os.LookupEnv(key); ok { + if i, err := strconv.Atoi(v); err == nil { + return i + } + } + return fallback +} + +func envDurationOr(key string, fallback time.Duration) time.Duration { + if v, ok := os.LookupEnv(key); ok { + if d, err := time.ParseDuration(v); err == nil { + return d + } + } + return fallback +} diff --git a/persys-meter/internal/consumer/consumer.go b/persys-meter/internal/consumer/consumer.go new file mode 100644 index 0000000..b1c320f --- /dev/null +++ b/persys-meter/internal/consumer/consumer.go @@ -0,0 +1,427 @@ +// Package consumer implements the Redis Streams consumer-group side of the +// meter ingestion pipeline: multiple worker goroutines read new messages via +// XREADGROUP, a background sweep reclaims messages abandoned by dead +// consumers via XAUTOCLAIM, a Redis-backed guard deduplicates redelivered +// events, and a single batching writer flushes accumulated records to the +// store before acking the corresponding stream messages. +package consumer + +import ( + "context" + "errors" + "strconv" + "strings" + "sync" + "time" + + "github.com/persys-dev/persys-cloud/persys-meter/internal/cache" + "github.com/persys-dev/persys-cloud/persys-meter/internal/config" + "github.com/persys-dev/persys-cloud/persys-meter/internal/ingest" + "github.com/persys-dev/persys-cloud/persys-meter/internal/metrics" + "github.com/persys-dev/persys-cloud/persys-meter/internal/store" + "github.com/redis/go-redis/v9" + "github.com/sirupsen/logrus" +) + +// pendingRecord couples a parsed usage record with the stream message ID it +// came from, so the batch writer can ack the right IDs after a successful +// flush - and can just as easily leave them un-acked after a failed one. +type pendingRecord struct { + messageID string + record ingest.UsageRecord +} + +// Consumer wires together the Redis stream reader, dedup guard, and batch +// writer. +type Consumer struct { + cfg *config.Config + redis *redis.Client + store store.Store + cache *cache.LatestCache + logger *logrus.Entry + + pending chan pendingRecord + + wg sync.WaitGroup + cancel context.CancelFunc +} + +// New constructs a Consumer. Call Run to start it. c is the shared +// in-memory latest-usage cache (also read by the API and the Prometheus +// WorkloadCollector) - the consumer is what keeps it up to date. +func New(cfg *config.Config, redisClient *redis.Client, st store.Store, c *cache.LatestCache, logger *logrus.Entry) *Consumer { + return &Consumer{ + cfg: cfg, + redis: redisClient, + store: st, + cache: c, + logger: logger, + // Buffered so a slow flush cycle doesn't immediately stall workers; + // sized generously relative to one batch so a full batch can queue + // up while the previous one is still being sent. + pending: make(chan pendingRecord, cfg.BatchSize*2), + } +} + +// Ready implements health.Checker: reports whether Redis is reachable. +// (ClickHouse readiness is reported by the store's own health check if the +// caller wants to compose one; keeping this narrow to Redis since that's +// what this package owns.) +func (c *Consumer) Ready(ctx context.Context) error { + return c.redis.Ping(ctx).Err() +} + +// Run ensures the consumer group exists, then starts all worker goroutines, +// the reclaim sweep, and the batch flusher. It blocks until ctx is +// cancelled, then drains in-flight work before returning. +func (c *Consumer) Run(ctx context.Context) error { + runCtx, cancel := context.WithCancel(ctx) + c.cancel = cancel + + if err := c.ensureGroup(ctx); err != nil { + cancel() + return err + } + + c.wg.Add(1) + go func() { + defer c.wg.Done() + c.flushLoop(runCtx) + }() + + c.wg.Add(1) + go func() { + defer c.wg.Done() + c.reclaimLoop(runCtx) + }() + + for i := 0; i < c.cfg.Workers; i++ { + c.wg.Add(1) + go func(workerNum int) { + defer c.wg.Done() + c.readLoop(runCtx, workerNum) + }(i) + } + + <-runCtx.Done() + c.logger.Info("consumer shutting down, waiting for in-flight work to drain") + c.wg.Wait() + return nil +} + +// Stop signals all goroutines to exit and blocks until they do (including a +// final batch flush of anything still buffered). +func (c *Consumer) Stop() { + if c.cancel != nil { + c.cancel() + } +} + +// ensureGroup creates the consumer group (and the stream itself, via +// MKSTREAM, if it doesn't exist yet - e.g. persys-meter started before +// persys-scheduler ever published anything). Starting at "0" means a +// brand-new group sees the whole stream history; "$" would mean "only new +// messages from now on". "0" is the safer default for a metering pipeline - +// better to reprocess something already ingested (harmless: downstream +// dedup by event_id in your own consumer already guards this) than to +// silently skip the backlog. +func (c *Consumer) ensureGroup(ctx context.Context) error { + err := c.redis.XGroupCreateMkStream(ctx, c.cfg.StreamName, c.cfg.ConsumerGroup, "0").Err() + if err != nil && !isBusyGroupErr(err) { + return err + } + c.logger.WithFields(logrus.Fields{ + "stream": c.cfg.StreamName, + "group": c.cfg.ConsumerGroup, + }).Info("consumer group ready") + return nil +} + +func isBusyGroupErr(err error) bool { + return err != nil && strings.Contains(err.Error(), "BUSYGROUP") +} + +// readLoop is the main per-worker XREADGROUP loop, reading only new ('>') +// messages. Reclaimed (previously-pending) messages are handled separately +// by reclaimLoop so a stalled worker's backlog doesn't monopolize every +// other worker's read calls. +func (c *Consumer) readLoop(ctx context.Context, workerNum int) { + consumerName := c.cfg.ConsumerName + if c.cfg.Workers > 1 { + consumerName = c.cfg.ConsumerName + "-w" + strconv.Itoa(workerNum) + } + log := c.logger.WithField("worker", consumerName) + log.Info("read worker starting") + + for { + select { + case <-ctx.Done(): + log.Info("read worker stopping") + return + default: + } + + streams, err := c.redis.XReadGroup(ctx, &redis.XReadGroupArgs{ + Group: c.cfg.ConsumerGroup, + Consumer: consumerName, + Streams: []string{c.cfg.StreamName, ">"}, + Count: c.cfg.ReadCount, + Block: c.cfg.ReadBlock, + }).Result() + if err != nil { + if errors.Is(err, context.Canceled) { + return + } + if errors.Is(err, redis.Nil) { + // Block duration elapsed with nothing new - expected, loop again. + continue + } + log.WithError(err).Warn("XREADGROUP failed, backing off") + select { + case <-time.After(time.Second): + case <-ctx.Done(): + return + } + continue + } + + for _, s := range streams { + metrics.EventsConsumedTotal.WithLabelValues(s.Stream).Add(float64(len(s.Messages))) + c.handleMessages(ctx, log, s.Stream, s.Messages) + } + } +} + +// reclaimLoop periodically claims pending messages that have sat un-acked +// longer than ClaimMinIdle - i.e. messages delivered to a consumer that then +// crashed, was killed, or hung before ack'ing. Without this, a dead pod's +// in-flight messages would never be retried by anyone. +func (c *Consumer) reclaimLoop(ctx context.Context) { + log := c.logger.WithField("component", "reclaim") + ticker := time.NewTicker(c.cfg.ClaimInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + + start := "0-0" + for { + messages, next, err := c.redis.XAutoClaim(ctx, &redis.XAutoClaimArgs{ + Stream: c.cfg.StreamName, + Group: c.cfg.ConsumerGroup, + Consumer: c.cfg.ConsumerName, + MinIdle: c.cfg.ClaimMinIdle, + Start: start, + Count: int64(c.cfg.BatchSize), + }).Result() + if err != nil { + log.WithError(err).Warn("XAUTOCLAIM failed") + break + } + if len(messages) > 0 { + metrics.MessagesReclaimedTotal.WithLabelValues(c.cfg.StreamName).Add(float64(len(messages))) + log.WithField("count", len(messages)).Info("reclaimed pending messages from a stalled consumer") + c.handleMessages(ctx, log, c.cfg.StreamName, messages) + } + if next == "0-0" || len(messages) == 0 { + break + } + start = next + } + } +} + +// handleMessages parses, dedups, and enqueues a batch of messages for +// writing. Messages that fail to parse are logged and acked immediately +// (see the comment inline) rather than retried forever. +func (c *Consumer) handleMessages(ctx context.Context, log *logrus.Entry, streamName string, messages []redis.XMessage) { + for _, msg := range messages { + record, err := ingest.ParseStreamMessage(msg.Values) + if err != nil { + metrics.EventsParseFailedTotal.WithLabelValues(streamName).Inc() + log.WithError(err).WithField("message_id", msg.ID).Error( + "failed to parse stream message; acking anyway to avoid a poison-pill redelivery loop") + // A message that can never parse would otherwise be reclaimed + // and retried forever. We've already logged it at Error level + // for investigation - that's the appropriate place to catch + // this, not an infinite XAUTOCLAIM loop. + c.ackAndLog(ctx, log, streamName, msg.ID) + continue + } + + if !record.ReportedAt.IsZero() { + metrics.ConsumerLagSeconds.WithLabelValues(streamName).Observe(time.Since(record.ReportedAt).Seconds()) + } + + // Update the live cache unconditionally, even for what turns out to + // be a duplicate below - it's the same data either way, and this is + // what backs the per-workload Prometheus metrics and the API's live + // endpoints, which don't care about ClickHouse-side dedup at all. + c.cache.Set(*record) + + dup, err := c.isDuplicate(ctx, record.EventID) + if err != nil { + // Redis hiccup on the dedupe check - fail open (treat as not a + // duplicate) rather than dropping a legitimate, un-deduped + // event. Worst case is a rare double-write, which the ORDER BY + // event_id in ClickHouse at least makes filterable later. + log.WithError(err).WithField("event_id", record.EventID).Warn( + "dedupe check failed, proceeding as if not a duplicate") + } + if dup { + metrics.EventsDedupedTotal.WithLabelValues(streamName).Inc() + c.ackAndLog(ctx, log, streamName, msg.ID) + continue + } + + select { + case c.pending <- pendingRecord{messageID: msg.ID, record: *record}: + case <-ctx.Done(): + return + } + } +} + +// isDuplicate uses SETNX with a TTL as a lightweight, explicit dedup guard: +// the first consumer to see a given event_id "claims" it; any redelivery of +// the same event_id (at-least-once delivery guarantees this can happen) +// finds the key already set and is treated as a duplicate. +// +// The TTL bound (config.DedupeTTL) means a duplicate arriving after the TTL +// expires would slip through - that's an accepted tradeoff, not a bug: it +// bounds how much memory the dedupe keyspace uses, and redeliveries in +// practice happen within seconds to minutes (XAUTOCLAIM's MinIdle), not +// hours or days. +func (c *Consumer) isDuplicate(ctx context.Context, eventID string) (bool, error) { + if eventID == "" { + // No event_id to dedupe on - treat as unique rather than blocking + // ingestion entirely on a missing field. + return false, nil + } + key := "persys:meter:dedupe:" + eventID + set, err := c.redis.SetNX(ctx, key, 1, c.cfg.DedupeTTL).Result() + if err != nil { + return false, err + } + return !set, nil +} + +// releaseDedupeGuard removes the dedupe key for a record whose batch write +// failed, so a future redelivery (or the next reclaim sweep) isn't +// incorrectly treated as an already-processed duplicate. +func (c *Consumer) releaseDedupeGuard(ctx context.Context, eventID string) { + if eventID == "" { + return + } + if err := c.redis.Del(ctx, "persys:meter:dedupe:"+eventID).Err(); err != nil { + c.logger.WithError(err).WithField("event_id", eventID).Warn( + "failed to release dedupe guard after a failed write; it will simply expire via TTL instead") + } +} + +func (c *Consumer) ackAndLog(ctx context.Context, log *logrus.Entry, streamName, messageID string) { + if err := c.redis.XAck(ctx, streamName, c.cfg.ConsumerGroup, messageID).Err(); err != nil { + log.WithError(err).WithField("message_id", messageID).Warn("failed to ack message") + } +} + +// flushLoop is the single writer goroutine: it accumulates pendingRecords +// off the shared channel and flushes them to the store whenever either the +// configured batch size or flush interval is reached, whichever comes +// first. Centralizing writes in one goroutine (rather than having each +// worker write its own small batches) means bigger, more efficient +// ClickHouse inserts regardless of how many read workers are configured. +func (c *Consumer) flushLoop(ctx context.Context) { + log := c.logger.WithField("component", "flusher") + batch := make([]pendingRecord, 0, c.cfg.BatchSize) + ticker := time.NewTicker(c.cfg.BatchFlushInterval) + defer ticker.Stop() + + flush := func() { + if len(batch) == 0 { + return + } + c.writeBatch(context.Background(), log, batch) + batch = batch[:0] + } + + for { + select { + case <-ctx.Done(): + // Drain whatever's left in the channel (non-blocking) before + // the final flush, so a shutdown doesn't drop already-read + // records that were waiting on the channel. + for { + select { + case pr := <-c.pending: + batch = append(batch, pr) + if len(batch) >= c.cfg.BatchSize { + flush() + } + default: + flush() + return + } + } + + case pr := <-c.pending: + batch = append(batch, pr) + if len(batch) >= c.cfg.BatchSize { + flush() + } + + case <-ticker.C: + flush() + } + } +} + +// writeBatch sends one batch to the store, then acks (on success) or +// releases the dedupe guard and leaves messages pending for retry (on +// failure). +func (c *Consumer) writeBatch(ctx context.Context, log *logrus.Entry, batch []pendingRecord) { + records := make([]ingest.UsageRecord, len(batch)) + ids := make([]string, len(batch)) + for i, pr := range batch { + records[i] = pr.record + ids[i] = pr.messageID + } + + start := time.Now() + err := c.store.WriteBatch(ctx, records) + elapsed := time.Since(start) + + if err != nil { + metrics.BatchWriteFailedTotal.WithLabelValues(c.cfg.StreamName).Inc() + metrics.BatchFlushDuration.WithLabelValues(c.cfg.StreamName, "failure").Observe(elapsed.Seconds()) + log.WithError(err).WithFields(logrus.Fields{ + "batch_size": len(batch), + "elapsed_ms": elapsed.Milliseconds(), + }).Error("batch write failed; leaving messages pending for retry") + + // Let a future redelivery try again rather than silently losing + // these events to the dedupe guard. + for _, pr := range batch { + c.releaseDedupeGuard(ctx, pr.record.EventID) + } + return + } + + metrics.EventsWrittenTotal.WithLabelValues(c.cfg.StreamName).Add(float64(len(batch))) + metrics.BatchSize.WithLabelValues(c.cfg.StreamName).Observe(float64(len(batch))) + metrics.BatchFlushDuration.WithLabelValues(c.cfg.StreamName, "success").Observe(elapsed.Seconds()) + + if err := c.redis.XAck(ctx, c.cfg.StreamName, c.cfg.ConsumerGroup, ids...).Err(); err != nil { + log.WithError(err).WithField("batch_size", len(batch)).Warn( + "batch written successfully but XACK failed; messages may be redelivered and will be deduped") + } + + log.WithFields(logrus.Fields{ + "batch_size": len(batch), + "elapsed_ms": elapsed.Milliseconds(), + }).Debug("batch flushed") +} diff --git a/persys-meter/internal/health/health.go b/persys-meter/internal/health/health.go new file mode 100644 index 0000000..589fc72 --- /dev/null +++ b/persys-meter/internal/health/health.go @@ -0,0 +1,65 @@ +// Package health serves /healthz, /readyz, and /metrics over plain HTTP - +// intentionally simple (no mTLS/auth) since this is meant for k8s liveness/ +// readiness probes and Prometheus scraping from inside the cluster network, +// not for external clients. +package health + +import ( + "context" + "net/http" + "time" + + "github.com/prometheus/client_golang/prometheus/promhttp" + "github.com/sirupsen/logrus" +) + +// Checker reports whether the service is ready to do work (e.g. Redis and +// ClickHouse are both reachable). +type Checker interface { + Ready(ctx context.Context) error +} + +// Serve starts the health/metrics HTTP server in a background goroutine and +// returns the *http.Server so the caller can Shutdown() it gracefully. +func Serve(addr string, checker Checker, logger *logrus.Entry) *http.Server { + mux := http.NewServeMux() + + mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { + // Liveness: the process is up and serving HTTP. Doesn't check + // dependencies - that's what /readyz is for. A live-but-not-ready + // process shouldn't be killed by a liveness probe, just pulled out + // of rotation by a readiness probe. + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok")) + }) + + mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) { + ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second) + defer cancel() + + if err := checker.Ready(ctx); err != nil { + logger.WithError(err).Warn("readiness check failed") + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = w.Write([]byte("not ready: " + err.Error())) + return + } + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ready")) + }) + + mux.Handle("/metrics", promhttp.Handler()) + + srv := &http.Server{ + Addr: addr, + Handler: mux, + } + + go func() { + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + logger.WithError(err).Error("health server stopped unexpectedly") + } + }() + + logger.WithField("addr", addr).Info("health/metrics server listening") + return srv +} diff --git a/persys-meter/internal/ingest/event.go b/persys-meter/internal/ingest/event.go new file mode 100644 index 0000000..39fb655 --- /dev/null +++ b/persys-meter/internal/ingest/event.go @@ -0,0 +1,140 @@ +// Package ingest defines the wire format published by persys-scheduler onto +// the Redis usage stream, and turns raw stream messages into the flat +// UsageRecord shape the store package writes to ClickHouse. +package ingest + +import ( + "encoding/json" + "fmt" + "time" +) + +// Event mirrors persys-scheduler's internal usageStreamEvent struct +// (internal/scheduler/usage_stream.go) field-for-field, including its +// snake_case JSON tags for the envelope. This is a cross-repo contract that +// isn't shared via a common package (the two services are separate Go +// modules) - if persys-scheduler's envelope shape changes, this struct must +// be updated to match by hand. Same caveat applies to UsageSnapshot below. +type Event struct { + EventID string `json:"event_id"` + NodeID string `json:"node_id"` + WorkloadID string `json:"workload_id"` + RevisionID string `json:"revision_id,omitempty"` + ReportedAt string `json:"reported_at"` + Usage *UsageSnapshot `json:"usage"` +} + +// UsageSnapshot mirrors persys-scheduler's internal/models.WorkloadUsage +// JSON shape - NOT compute-agent's. This matters: compute-agent's +// models.WorkloadUsage uses snake_case tags (e.g. "cpu_percent"), but the +// scheduler re-serializes usage data through its *own* models.WorkloadUsage +// type before publishing, which uses camelCase tags (e.g. "cpuPercent"). +// Since the scheduler is what actually writes the stream payload, its tags +// are what persys-meter must parse against. +type UsageSnapshot struct { + WorkloadID string `json:"workloadId,omitempty"` + Type string `json:"type,omitempty"` + CPUPercent float64 `json:"cpuPercent,omitempty"` + MemoryBytes int64 `json:"memoryBytes,omitempty"` + DiskReadBytes int64 `json:"diskReadBytes,omitempty"` + DiskWriteBytes int64 `json:"diskWriteBytes,omitempty"` + NetRXBytes int64 `json:"netRxBytes,omitempty"` + NetTXBytes int64 `json:"netTxBytes,omitempty"` + CollectedAt time.Time `json:"collectedAt,omitempty"` + Source string `json:"source,omitempty"` +} + +// UsageRecord is the flat row shape written to ClickHouse - one row per +// usage sample, with the envelope and usage snapshot merged together and +// ReportedAt parsed into a real time.Time. +type UsageRecord struct { + EventID string + NodeID string + WorkloadID string + RevisionID string + WorkloadType string + ReportedAt time.Time + CPUPercent float64 + MemoryBytes int64 + DiskReadBytes int64 + DiskWriteBytes int64 + NetRXBytes int64 + NetTXBytes int64 + Source string +} + +// ParseStreamMessage turns a Redis XMessage's Values map into a UsageRecord. +// +// persys-scheduler publishes both a "payload" field (the full event as a +// JSON string) and individual flat fields (event_id, node_id, workload_id, +// revision_id, reported_at) on every stream entry. This function prefers +// "payload" since it's the only place the actual usage numbers live; the +// flat fields exist so the stream is at least partially inspectable with +// plain `redis-cli XRANGE` without decoding JSON. +func ParseStreamMessage(values map[string]interface{}) (*UsageRecord, error) { + payload, ok := stringValue(values, "payload") + if !ok || payload == "" { + return nil, fmt.Errorf("stream message has no payload field") + } + + var event Event + if err := json.Unmarshal([]byte(payload), &event); err != nil { + return nil, fmt.Errorf("failed to unmarshal payload: %w", err) + } + + if event.WorkloadID == "" { + return nil, fmt.Errorf("event payload missing workload_id") + } + if event.Usage == nil { + return nil, fmt.Errorf("event payload for workload %s has no usage data", event.WorkloadID) + } + + reportedAt, err := time.Parse(time.RFC3339Nano, event.ReportedAt) + if err != nil { + // Fall back to the usage snapshot's own CollectedAt, and failing + // that, now - a malformed timestamp shouldn't drop an otherwise + // valid usage sample. + if !event.Usage.CollectedAt.IsZero() { + reportedAt = event.Usage.CollectedAt + } else { + reportedAt = time.Now().UTC() + } + } + + return &UsageRecord{ + EventID: event.EventID, + NodeID: event.NodeID, + WorkloadID: event.WorkloadID, + RevisionID: event.RevisionID, + WorkloadType: event.Usage.Type, + ReportedAt: reportedAt.UTC(), + CPUPercent: event.Usage.CPUPercent, + MemoryBytes: event.Usage.MemoryBytes, + DiskReadBytes: event.Usage.DiskReadBytes, + DiskWriteBytes: event.Usage.DiskWriteBytes, + NetRXBytes: event.Usage.NetRXBytes, + NetTXBytes: event.Usage.NetTXBytes, + Source: event.Usage.Source, + }, nil +} + +// stringValue safely extracts a string field from a Redis XMessage.Values +// map. go-redis returns stream field values as `interface{}` holding a +// `string` in the common case, but is defensive here in case a future +// go-redis version or a non-Go producer writes a different underlying type. +func stringValue(values map[string]interface{}, key string) (string, bool) { + raw, ok := values[key] + if !ok || raw == nil { + return "", false + } + switch v := raw.(type) { + case string: + return v, true + case []byte: + return string(v), true + case fmt.Stringer: + return v.String(), true + default: + return fmt.Sprintf("%v", v), true + } +} diff --git a/persys-meter/internal/metrics/metrics.go b/persys-meter/internal/metrics/metrics.go new file mode 100644 index 0000000..bf220e4 --- /dev/null +++ b/persys-meter/internal/metrics/metrics.go @@ -0,0 +1,227 @@ +// Package metrics exposes Prometheus metrics for persys-meter itself, in the +// same "persys" namespace / manual CounterVec style already used by +// persys-scheduler's internal/metrics package, for consistency across +// services' dashboards. +package metrics + +import ( + "sync" + "time" + + "github.com/persys-dev/persys-cloud/persys-meter/internal/cache" + "github.com/prometheus/client_golang/prometheus" +) + +var registerOnce sync.Once + +var ( + EventsConsumedTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "persys", + Subsystem: "meter", + Name: "events_consumed_total", + Help: "Total number of usage events read from the Redis stream via XREADGROUP.", + }, + []string{"stream"}, + ) + + EventsDedupedTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "persys", + Subsystem: "meter", + Name: "events_deduped_total", + Help: "Total number of events skipped because their event_id was already seen (redelivery).", + }, + []string{"stream"}, + ) + + EventsParseFailedTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "persys", + Subsystem: "meter", + Name: "events_parse_failed_total", + Help: "Total number of stream messages that could not be parsed into a usage record.", + }, + []string{"stream"}, + ) + + EventsWrittenTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "persys", + Subsystem: "meter", + Name: "events_written_total", + Help: "Total number of usage records successfully written to the store.", + }, + []string{"stream"}, + ) + + BatchWriteFailedTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "persys", + Subsystem: "meter", + Name: "batch_write_failed_total", + Help: "Total number of store batch writes that failed (records are left un-acked for retry).", + }, + []string{"stream"}, + ) + + MessagesReclaimedTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "persys", + Subsystem: "meter", + Name: "messages_reclaimed_total", + Help: "Total number of pending messages reclaimed from stalled consumers via XAUTOCLAIM.", + }, + []string{"stream"}, + ) + + BatchFlushDuration = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: "persys", + Subsystem: "meter", + Name: "batch_flush_duration_seconds", + Help: "Time taken to write a batch of usage records to the store.", + Buckets: prometheus.DefBuckets, + }, + []string{"stream", "result"}, + ) + + BatchSize = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: "persys", + Subsystem: "meter", + Name: "batch_size", + Help: "Number of records in each flushed batch.", + Buckets: []float64{1, 10, 50, 100, 250, 500, 1000, 2500, 5000}, + }, + []string{"stream"}, + ) + + ConsumerLagSeconds = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: "persys", + Subsystem: "meter", + Name: "consumer_lag_seconds", + Help: "Time between a usage event being reported and persys-meter processing it.", + Buckets: []float64{.1, .5, 1, 2, 5, 10, 30, 60, 300}, + }, + []string{"stream"}, + ) +) + +// Register registers all meter metrics with the default Prometheus +// registry. Safe to call more than once. +func Register() { + registerOnce.Do(func() { + prometheus.MustRegister( + EventsConsumedTotal, + EventsDedupedTotal, + EventsParseFailedTotal, + EventsWrittenTotal, + BatchWriteFailedTotal, + MessagesReclaimedTotal, + BatchFlushDuration, + BatchSize, + ConsumerLagSeconds, + ) + }) +} + +// WorkloadCollector emits live, per-workload usage as Prometheus metrics - +// this is what makes actual workload metrics (not just meter-pipeline +// health metrics) available for scraping/dashboards/alerting. Implemented +// as a custom prometheus.Collector (rather than a persistent GaugeVec) so +// that a workload which stops reporting simply stops appearing in scrapes, +// instead of leaving a stale, never-cleaned-up series behind forever. +type WorkloadCollector struct { + cache *cache.LatestCache + maxAge time.Duration + + cpuPercent *prometheus.Desc + memoryBytes *prometheus.Desc + diskReadBytes *prometheus.Desc + diskWriteBytes *prometheus.Desc + netRXBytes *prometheus.Desc + netTXBytes *prometheus.Desc + sampleAge *prometheus.Desc +} + +// NewWorkloadCollector builds a collector backed directly by the shared +// in-memory cache. maxAge bounds how long a workload keeps appearing in +// scrapes after its last reported sample - see cache.LatestCache.All. +func NewWorkloadCollector(c *cache.LatestCache, maxAge time.Duration) *WorkloadCollector { + labels := []string{"workload_id", "node_id", "workload_type"} + return &WorkloadCollector{ + cache: c, + maxAge: maxAge, + cpuPercent: prometheus.NewDesc( + "persys_meter_workload_cpu_percent", + "Latest observed CPU utilization percent for a workload.", + labels, nil, + ), + memoryBytes: prometheus.NewDesc( + "persys_meter_workload_memory_bytes", + "Latest observed resident memory usage in bytes for a workload.", + labels, nil, + ), + diskReadBytes: prometheus.NewDesc( + "persys_meter_workload_disk_read_bytes_total", + "Cumulative disk bytes read, as last reported by the workload's runtime (resets if the workload restarts).", + labels, nil, + ), + diskWriteBytes: prometheus.NewDesc( + "persys_meter_workload_disk_write_bytes_total", + "Cumulative disk bytes written, as last reported by the workload's runtime (resets if the workload restarts).", + labels, nil, + ), + netRXBytes: prometheus.NewDesc( + "persys_meter_workload_net_rx_bytes_total", + "Cumulative network bytes received, as last reported by the workload's runtime (resets if the workload restarts).", + labels, nil, + ), + netTXBytes: prometheus.NewDesc( + "persys_meter_workload_net_tx_bytes_total", + "Cumulative network bytes transmitted, as last reported by the workload's runtime (resets if the workload restarts).", + labels, nil, + ), + sampleAge: prometheus.NewDesc( + "persys_meter_workload_sample_age_seconds", + "Seconds since the last usage sample for a workload was received. A consistently high value means the workload's agent may have stopped reporting.", + labels, nil, + ), + } +} + +// Describe implements prometheus.Collector. +func (c *WorkloadCollector) Describe(ch chan<- *prometheus.Desc) { + ch <- c.cpuPercent + ch <- c.memoryBytes + ch <- c.diskReadBytes + ch <- c.diskWriteBytes + ch <- c.netRXBytes + ch <- c.netTXBytes + ch <- c.sampleAge +} + +// Collect implements prometheus.Collector, emitting one set of metrics per +// currently-live workload on every scrape. +func (c *WorkloadCollector) Collect(ch chan<- prometheus.Metric) { + now := time.Now() + for _, e := range c.cache.All(c.maxAge) { + labels := []string{e.Record.WorkloadID, e.Record.NodeID, e.Record.WorkloadType} + ch <- prometheus.MustNewConstMetric(c.cpuPercent, prometheus.GaugeValue, e.Record.CPUPercent, labels...) + ch <- prometheus.MustNewConstMetric(c.memoryBytes, prometheus.GaugeValue, float64(e.Record.MemoryBytes), labels...) + ch <- prometheus.MustNewConstMetric(c.diskReadBytes, prometheus.CounterValue, float64(e.Record.DiskReadBytes), labels...) + ch <- prometheus.MustNewConstMetric(c.diskWriteBytes, prometheus.CounterValue, float64(e.Record.DiskWriteBytes), labels...) + ch <- prometheus.MustNewConstMetric(c.netRXBytes, prometheus.CounterValue, float64(e.Record.NetRXBytes), labels...) + ch <- prometheus.MustNewConstMetric(c.netTXBytes, prometheus.CounterValue, float64(e.Record.NetTXBytes), labels...) + ch <- prometheus.MustNewConstMetric(c.sampleAge, prometheus.GaugeValue, now.Sub(e.UpdatedAt).Seconds(), labels...) + } +} + +// RegisterWorkloadCollector registers a WorkloadCollector with the default +// registry. Separate from Register() since it needs the cache instance, +// which isn't available at package-init time. +func RegisterWorkloadCollector(c *WorkloadCollector) { + prometheus.MustRegister(c) +} diff --git a/persys-meter/internal/store/clickhouse.go b/persys-meter/internal/store/clickhouse.go new file mode 100644 index 0000000..a589411 --- /dev/null +++ b/persys-meter/internal/store/clickhouse.go @@ -0,0 +1,288 @@ +package store + +import ( + "context" + "crypto/tls" + "fmt" + "time" + + clickhouse "github.com/ClickHouse/clickhouse-go/v2" + "github.com/persys-dev/persys-cloud/persys-meter/internal/ingest" + "github.com/sirupsen/logrus" +) + +// ClickHouseConfig is the subset of config needed to construct a +// ClickHouseStore - kept narrow so this package doesn't depend on the +// top-level config package. +type ClickHouseConfig struct { + Addr string + Database string + Username string + Password string + TLS bool + RetentionDays int +} + +// ClickHouseStore writes usage records to a ClickHouse MergeTree table. +// +// Dedup note: this table is a plain MergeTree, not ReplacingMergeTree. We +// deliberately do NOT rely on ClickHouse-side dedup because ReplacingMergeTree +// only collapses duplicate rows during background merges, which are +// asynchronous and not guaranteed to have run before a query executes - +// queries can and do see duplicate rows before a merge. Real dedup happens +// upstream, in the consumer, via a Redis SETNX guard on event_id before a +// record ever reaches this store (see internal/consumer). This table's +// ORDER BY still includes event_id so that if a duplicate ever does slip +// through, it's at least easy to filter with `LIMIT 1 BY event_id` in +// queries, or backfilled into a ReplacingMergeTree later if needed. +type ClickHouseStore struct { + conn clickhouse.Conn + database string + retentionDays int + logger *logrus.Entry +} + +// NewClickHouseStore opens a connection to ClickHouse. It does not create +// the table yet - call Init for that. +func NewClickHouseStore(cfg ClickHouseConfig, logger *logrus.Entry) (*ClickHouseStore, error) { + if cfg.RetentionDays <= 0 { + cfg.RetentionDays = 90 + } + + opts := &clickhouse.Options{ + Addr: []string{cfg.Addr}, + Auth: clickhouse.Auth{ + Database: cfg.Database, + Username: cfg.Username, + Password: cfg.Password, + }, + DialTimeout: 5 * time.Second, + } + if cfg.TLS { + opts.TLS = &tls.Config{} + } + + conn, err := clickhouse.Open(opts) + if err != nil { + return nil, fmt.Errorf("failed to open clickhouse connection: %w", err) + } + + pingCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := conn.Ping(pingCtx); err != nil { + return nil, fmt.Errorf("failed to ping clickhouse at %s: %w", cfg.Addr, err) + } + + return &ClickHouseStore{ + conn: conn, + database: cfg.Database, + retentionDays: cfg.RetentionDays, + logger: logger, + }, nil +} + +// Init creates the usage_events table if it doesn't already exist. +// Idempotent - safe to call on every startup. +func (s *ClickHouseStore) Init(ctx context.Context) error { + ddl := fmt.Sprintf(` +CREATE TABLE IF NOT EXISTS %s.usage_events +( + event_id String, + node_id String, + workload_id String, + revision_id String, + workload_type LowCardinality(String), + reported_at DateTime64(9, 'UTC'), + ingested_at DateTime64(3, 'UTC') DEFAULT now64(3), + cpu_percent Float64, + memory_bytes Int64, + disk_read_bytes Int64, + disk_write_bytes Int64, + net_rx_bytes Int64, + net_tx_bytes Int64, + source LowCardinality(String) +) +ENGINE = MergeTree +PARTITION BY toYYYYMMDD(reported_at) +ORDER BY (workload_id, reported_at, event_id) +TTL toDateTime(reported_at) + INTERVAL %d DAY +`, s.database, s.retentionDays) + + if err := s.conn.Exec(ctx, ddl); err != nil { + return fmt.Errorf("failed to create usage_events table: %w", err) + } + + s.logger.WithFields(logrus.Fields{ + "database": s.database, + "retention_days": s.retentionDays, + }).Info("usage_events table ready") + return nil +} + +// WriteBatch inserts every record in a single ClickHouse batch insert. +// All-or-nothing: if Send() fails partway, ClickHouse discards the whole +// batch, and the caller (consumer) is expected to leave the corresponding +// stream messages un-acked so they're retried. +func (s *ClickHouseStore) WriteBatch(ctx context.Context, records []ingest.UsageRecord) error { + if len(records) == 0 { + return nil + } + + // Columns are listed explicitly (rather than a bare "INSERT INTO + // usage_events") specifically to exclude ingested_at, which has a + // DEFAULT now64(3) clause and is meant to be filled in by the server on + // every insert - it isn't one of the values Append() provides below. + insertQuery := fmt.Sprintf(`INSERT INTO %s.usage_events ( + event_id, node_id, workload_id, revision_id, workload_type, + reported_at, cpu_percent, memory_bytes, disk_read_bytes, + disk_write_bytes, net_rx_bytes, net_tx_bytes, source + )`, s.database) + + batch, err := s.conn.PrepareBatch(ctx, insertQuery) + if err != nil { + return fmt.Errorf("failed to prepare clickhouse batch: %w", err) + } + // Safe to call even after a successful Send(): Close() on an + // already-sent batch is a no-op per the driver's contract. + defer batch.Abort() + + for _, r := range records { + err := batch.Append( + r.EventID, + r.NodeID, + r.WorkloadID, + r.RevisionID, + r.WorkloadType, + r.ReportedAt, + r.CPUPercent, + r.MemoryBytes, + r.DiskReadBytes, + r.DiskWriteBytes, + r.NetRXBytes, + r.NetTXBytes, + r.Source, + ) + if err != nil { + return fmt.Errorf("failed to append record (workload=%s, event=%s) to batch: %w", r.WorkloadID, r.EventID, err) + } + } + + if err := batch.Send(); err != nil { + return fmt.Errorf("failed to send clickhouse batch of %d records: %w", len(records), err) + } + + return nil +} + +// Ping reports whether ClickHouse is currently reachable. Cheap by design - +// used for readiness probes, so it must not run DDL or touch usage_events. +func (s *ClickHouseStore) Ping(ctx context.Context) error { + return s.conn.Ping(ctx) +} + +const maxHistoryLimit = 10000 + +// History returns raw usage samples for a workload within [from, to], +// newest first, capped at limit rows (limit <= 0 defaults to 1000, and +// anything above maxHistoryLimit is capped there - this is meant for +// "show me recent samples", not bulk export). +func (s *ClickHouseStore) History(ctx context.Context, workloadID string, from, to time.Time, limit int) ([]ingest.UsageRecord, error) { + if limit <= 0 { + limit = 1000 + } + if limit > maxHistoryLimit { + limit = maxHistoryLimit + } + + query := fmt.Sprintf(` + SELECT event_id, node_id, workload_id, revision_id, workload_type, reported_at, + cpu_percent, memory_bytes, disk_read_bytes, disk_write_bytes, + net_rx_bytes, net_tx_bytes, source + FROM %s.usage_events + WHERE workload_id = ? AND reported_at >= ? AND reported_at <= ? + ORDER BY reported_at DESC + LIMIT ? + `, s.database) + + rows, err := s.conn.Query(ctx, query, workloadID, from, to, limit) + if err != nil { + return nil, fmt.Errorf("failed to query usage history for workload %s: %w", workloadID, err) + } + defer rows.Close() + + records := make([]ingest.UsageRecord, 0, limit) + for rows.Next() { + var r ingest.UsageRecord + if err := rows.Scan( + &r.EventID, &r.NodeID, &r.WorkloadID, &r.RevisionID, &r.WorkloadType, &r.ReportedAt, + &r.CPUPercent, &r.MemoryBytes, &r.DiskReadBytes, &r.DiskWriteBytes, + &r.NetRXBytes, &r.NetTXBytes, &r.Source, + ); err != nil { + return nil, fmt.Errorf("failed to scan usage history row for workload %s: %w", workloadID, err) + } + records = append(records, r) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("error iterating usage history rows for workload %s: %w", workloadID, err) + } + + return records, nil +} + +// Summary aggregates a workload's usage over [from, to]. See the +// EstimatedCPUCoreSeconds and *BytesDelta doc comments on UsageSummary for +// important accuracy caveats before wiring this into real billing. +func (s *ClickHouseStore) Summary(ctx context.Context, workloadID string, from, to time.Time) (*UsageSummary, error) { + summary := &UsageSummary{WorkloadID: workloadID, From: from, To: to} + + // Check for zero rows first and return early: ClickHouse's avg()/min()/ + // max() all return NULL over an empty set, and scanning NULL into a + // non-pointer Go float64/int64 would error. Easier to special-case + // "no data" explicitly than to scan everything as nullable types. + countQuery := fmt.Sprintf(` + SELECT count() FROM %s.usage_events + WHERE workload_id = ? AND reported_at >= ? AND reported_at <= ? + `, s.database) + if err := s.conn.QueryRow(ctx, countQuery, workloadID, from, to).Scan(&summary.SampleCount); err != nil { + return nil, fmt.Errorf("failed to count usage samples for workload %s: %w", workloadID, err) + } + if summary.SampleCount == 0 { + return summary, nil + } + + aggQuery := fmt.Sprintf(` + SELECT + avg(cpu_percent), + max(cpu_percent), + avg(memory_bytes), + max(memory_bytes), + max(disk_read_bytes) - min(disk_read_bytes), + max(disk_write_bytes) - min(disk_write_bytes), + max(net_rx_bytes) - min(net_rx_bytes), + max(net_tx_bytes) - min(net_tx_bytes), + min(reported_at), + max(reported_at) + FROM %s.usage_events + WHERE workload_id = ? AND reported_at >= ? AND reported_at <= ? + `, s.database) + + row := s.conn.QueryRow(ctx, aggQuery, workloadID, from, to) + if err := row.Scan( + &summary.AvgCPUPercent, &summary.MaxCPUPercent, + &summary.AvgMemoryBytes, &summary.MaxMemoryBytes, + &summary.DiskReadBytesDelta, &summary.DiskWriteBytesDelta, + &summary.NetRXBytesDelta, &summary.NetTXBytesDelta, + &summary.FirstSample, &summary.LastSample, + ); err != nil { + return nil, fmt.Errorf("failed to aggregate usage summary for workload %s: %w", workloadID, err) + } + + summary.EstimatedCPUCoreSeconds = (summary.AvgCPUPercent / 100.0) * to.Sub(from).Seconds() + + return summary, nil +} + +// Close closes the underlying ClickHouse connection. +func (s *ClickHouseStore) Close() error { + return s.conn.Close() +} diff --git a/persys-meter/internal/store/store.go b/persys-meter/internal/store/store.go new file mode 100644 index 0000000..53674c1 --- /dev/null +++ b/persys-meter/internal/store/store.go @@ -0,0 +1,94 @@ +// Package store defines where ingested usage records end up. ClickHouse is +// the only implementation today (per the architecture decision to use it for +// this workload), but consumer code depends only on this interface so a +// second backend (e.g. Postgres for smaller deployments) can be added later +// without touching the consumer/ingest packages. +package store + +import ( + "context" + "time" + + "github.com/persys-dev/persys-cloud/persys-meter/internal/ingest" +) + +// Store persists batches of usage records. +type Store interface { + // Init prepares the backend for writes (e.g. creating tables). Called + // once at startup; must be safe to call repeatedly (idempotent). + Init(ctx context.Context) error + + // Ping reports whether the backend is currently reachable, for use in + // readiness checks. Should be cheap - implementations should NOT run + // DDL or anything beyond a lightweight connectivity check. + Ping(ctx context.Context) error + + // WriteBatch durably writes every record in the batch. It must either + // fully succeed or return an error - callers rely on all-or-nothing + // semantics to decide whether to ack the corresponding stream messages. + WriteBatch(ctx context.Context, records []ingest.UsageRecord) error + + // Close releases any underlying connections. + Close() error +} + +// QueryStore is the read side: historical lookups against durably stored +// usage data. Kept as a separate interface from Store (rather than adding +// these methods to it) since not every future write backend necessarily +// wants to implement rich queries - a single concrete type is free to +// implement both, as ClickHouseStore does. +type QueryStore interface { + // History returns raw samples for a workload within [from, to], + // newest first, capped at limit rows. + History(ctx context.Context, workloadID string, from, to time.Time, limit int) ([]ingest.UsageRecord, error) + + // Summary aggregates a workload's usage over [from, to] - this is the + // primitive a billing/quota-enforcement caller is expected to build on: + // "how much did this workload consume in this window", not a policy + // decision about limits or what to do when they're exceeded. + Summary(ctx context.Context, workloadID string, from, to time.Time) (*UsageSummary, error) +} + +// UsageSummary aggregates a workload's usage samples over a time window. +type UsageSummary struct { + WorkloadID string `json:"workload_id"` + From time.Time `json:"from"` + To time.Time `json:"to"` + // SampleCount is how many raw usage_events rows fell in the window. A + // summary with SampleCount 0 means no data was reported in this window + // at all (not that usage was zero) - callers should treat these + // differently, and this struct's zero-valued fields alongside + // SampleCount == 0 make that distinguishable. + SampleCount uint64 `json:"sample_count"` + + AvgCPUPercent float64 `json:"avg_cpu_percent"` + MaxCPUPercent float64 `json:"max_cpu_percent"` + + AvgMemoryBytes float64 `json:"avg_memory_bytes"` + MaxMemoryBytes int64 `json:"max_memory_bytes"` + + // *BytesDelta are computed as max(counter) - min(counter) within the + // window. Since these source counters are cumulative-since-runtime-start + // (see ingest.UsageRecord), this is only correct if the workload didn't + // restart during the window - a restart resets the runtime's counter to + // zero, which would make a delta look artificially small or even + // negative. This is a known limitation, not silently "handled" - + // something a real billing pipeline should account for (e.g. by + // tracking restarts explicitly and summing deltas between them). + DiskReadBytesDelta int64 `json:"disk_read_bytes_delta"` + DiskWriteBytesDelta int64 `json:"disk_write_bytes_delta"` + NetRXBytesDelta int64 `json:"net_rx_bytes_delta"` + NetTXBytesDelta int64 `json:"net_tx_bytes_delta"` + + FirstSample time.Time `json:"first_sample"` + LastSample time.Time `json:"last_sample"` + + // EstimatedCPUCoreSeconds approximates CPU-core-seconds consumed over + // the window as avg(cpu_percent)/100 * window_seconds. This is a + // deliberately simple approximation, not a rigorous integral over + // irregularly spaced samples (that would need trapezoidal integration + // between consecutive points) - it assumes usage sampling is roughly + // uniform across the window. Good enough for a first cut; revisit if + // billing accuracy requirements need more precision than this gives. + EstimatedCPUCoreSeconds float64 `json:"estimated_cpu_core_seconds"` +}