water-go

water-go is a small Go data pipeline that ingests Norwegian hydrological time series from the NVE HydAPI, stores them in TimescaleDB, and serves them over a read-only HTTP API. It began as a personal exploration of Go and Terraform against a real data source, so the repository covers the whole path from polling an external API to deployable AWS infrastructure with monitoring and alerting.

The ingester runs four stages connected by an in-process Watermill bus. A poller fetches observations from NVE on a schedule, a normalizer cleans units, timezones, and quality flags before upserting into TimescaleDB hypertables, an anomaly detector flags readings whose rolling z-score crosses a threshold, and an alerter logs those anomalies and optionally forwards them to a webhook. A separate chi binary serves the stored data as JSON.

Features

• The ingester polls one or more NVE stations on a configurable interval, with retries and client-side rate limiting toward the NVE API.
• Pipeline stages communicate through Watermill's in-process gochannel pub/sub, so each stage stays independent and separately testable.
• Upserts into TimescaleDB hypertables are idempotent, which makes re-polling overlapping windows safe.
• Database access is generated by sqlc on top of pgx/v5, so queries are type-checked at build time.
• A read-only JSON API serves stations, latest readings, time-ranged observations, and recent anomalies, with pagination.
• Anomaly detection computes a rolling z-score per station and parameter, records anomalies, and can POST alerts to a webhook.
• Data endpoints support optional API-key auth and per-client rate limiting.
• Both processes expose Prometheus metrics, and the local stack includes Prometheus plus a provisioned Grafana dashboard.
• make up starts everything locally: TimescaleDB, ingester, API, Prometheus, and Grafana.
• The Terraform stack deploys to AWS: VPC, ALB, ECS/Fargate services, ECR, RDS, Secrets Manager, CloudWatch alarms, and SNS alert delivery.

Architecture

NVE HydAPI ──▶ Poller ──▶ Normalizer ──▶ Anomaly detector ──▶ Alerter ──▶ alert webhook
                              │                  │
                        observations         anomalies
                              ▼                  ▼
                           TimescaleDB (hypertables)
                                       │
                                       ▼
                      client ◀── chi HTTP API (read-only)

The four ingester stages run in one process, connected by an in-process Watermill gochannel bus. The api is a separate binary that reads the same database, so the two deploy and scale independently.

Tech stack

Concern	Choice
Language	Go
Ingest pub/sub	Watermill (in-process gochannel)
Database	TimescaleDB (Postgres + hypertables)
DB driver	pgx/v5
Type-safe queries	sqlc
HTTP API	chi
Observability	Prometheus + Grafana locally, CloudWatch in AWS
Tests	Go testing + testcontainers integration suite
Containers	Docker + Docker Compose
Infrastructure	Terraform (AWS ECS/Fargate + RDS + ALB + CloudWatch)

Project layout

cmd/ingester          poller → normalizer → anomaly detector → alerter
cmd/api               chi HTTP API over stored data
internal/nve          NVE HydAPI client + response types
internal/pipeline     Watermill stages: poller, normalizer, anomaly detector, alerter
internal/analytics    pure statistics helpers (rolling z-score)
internal/api          chi router + auth and rate-limit middleware
internal/config       environment-based configuration
internal/metrics      Prometheus metrics instrumentation
internal/store        pgx pool + migration runner + sqlc queries
internal/db           sqlc-generated code (run `make sqlc`)
internal/integration  testcontainers integration tests (build tag `integration`)
db/migrations         TimescaleDB schema
db/queries            sqlc query definitions
ops/                  Prometheus config + Grafana provisioning and dashboards
terraform/            AWS deployment (VPC, ALB, ECS/Fargate, RDS, alarms)

Quick start (Docker Compose)

cp .env.example .env        # then set NVE_API_KEY
make up                     # build + start db, ingester, api, prometheus, grafana

Then query the API (the ingester needs a poll cycle or two before data appears):

curl localhost:8080/healthz
curl localhost:8080/stations
curl localhost:8080/stations/2.32.0/latest
curl "localhost:8080/stations/2.32.0/observations?parameter=1000&from=2026-06-28T00:00:00Z"
curl localhost:8080/stations/2.32.0/anomalies

If you set API_KEYS, include one on data requests:

curl -H "Authorization: Bearer $API_KEY" localhost:8080/stations

The stack also starts Prometheus at http://localhost:9091 and Grafana at http://localhost:3000 (admin/admin), where a water-go dashboard is provisioned automatically. Stop the stack with make down and tail logs with make logs.

Local development (without Docker)

Run just the database in a container and the binaries on the host:

docker compose up -d db     # start only TimescaleDB
make run-ingester           # in one shell
make run-api                # in another shell

make build compiles both binaries into ./bin. Run make help to list all targets.

Testing

make test runs the unit tests, which need no external services. make test-integration runs the integration suite, which spins up a real TimescaleDB via testcontainers and therefore needs Docker. make cover prints a per-package coverage summary.

Regenerating database code

internal/db is generated from db/migrations (schema) and db/queries and committed, so a fresh clone builds without sqlc installed. Rerun after changing either:

make sqlc

API reference

Base URL: http://localhost:8080. All responses are JSON. The API is read-only. When API_KEYS is set, /stations/* endpoints require either Authorization: Bearer <key> or X-API-Key: <key>. Health, readiness, and metrics endpoints remain unauthenticated for probes and scraping.

Method	Path	Description
GET	/healthz	Liveness check
GET	/readyz	Readiness + DB connectivity check
GET	/metrics	Prometheus metrics
GET	/stations	List known stations
GET	/stations/{id}/latest	Latest observation per parameter for station
GET	/stations/{id}/observations	Observations for a parameter over a window
GET	/stations/{id}/anomalies	Recent anomaly records for a station

/stations/{id}/observations query parameters:

Param	Required	Format	Default	Notes
parameter	yes	integer	—	e.g. 1000 (see parameter codes below)
from	no	RFC3339 timestamp	now − 7 days	Inclusive lower bound
to	no	RFC3339 timestamp	now	Inclusive upper bound
limit	no	integer	API_DEFAULT_PAGE_SIZE	Clamped to API_MAX_PAGE_SIZE
offset	no	integer	0	Must be non-negative

Parameter codes: 1000 water level (Vannstand, m), 1001 discharge (Vannføring, m³/s), 1003 water temperature.

Examples

GET /healthz

{ "status": "ok" }

GET /stations

[
  {
    "station_id": "2.32.0",
    "name": "Atnasjø",
    "river_name": "Atna",
    "latitude": 61.85188,
    "longitude": 10.21976,
    "masl": 701.0,
    "updated_at": "2026-06-29T08:00:03.512Z"
  }
]

GET /stations/2.32.0/latest

[
  {
    "time": "2026-06-29T07:00:00Z",
    "station_id": "2.32.0",
    "parameter": 1000,
    "parameter_name": "Vannstand",
    "unit": "m",
    "resolution_time": 60,
    "value": 4.213,
    "quality": 0,
    "correction": 0,
    "ingested_at": "2026-06-29T08:00:03.514Z"
  },
  {
    "time": "2026-06-29T07:00:00Z",
    "station_id": "2.32.0",
    "parameter": 1001,
    "parameter_name": "Vannføring",
    "unit": "m³/s",
    "resolution_time": 60,
    "value": 21.7,
    "quality": 0,
    "correction": 0,
    "ingested_at": "2026-06-29T08:00:03.516Z"
  }
]

GET /stations/2.32.0/observations?parameter=1000&from=2026-06-28T00:00:00Z

[
  {
    "time": "2026-06-28T02:00:00Z",
    "station_id": "2.32.0",
    "parameter": 1000,
    "parameter_name": "Vannstand",
    "unit": "m",
    "resolution_time": 60,
    "value": 4.205,
    "quality": 0,
    "correction": 0,
    "ingested_at": "2026-06-28T03:00:02.110Z"
  },
  {
    "time": "2026-06-28T01:00:00Z",
    "station_id": "2.32.0",
    "parameter": 1000,
    "parameter_name": "Vannstand",
    "unit": "m",
    "resolution_time": 60,
    "value": 4.204,
    "quality": 0,
    "correction": 0,
    "ingested_at": "2026-06-28T02:00:02.090Z"
  }
]

Observations are returned newest-first. value, quality, and correction may be null when the source reports no reading.

Configuration

Configuration is read from the environment, and a local .env is loaded if present. Copy .env.example to .env and set at least NVE_API_KEY.

Variable	Used by	Default	Description
NVE_API_KEY	ingester	(required)	NVE HydAPI key from https://hydapi.nve.no
NVE_BASE_URL	ingester	https://hydapi.nve.no/api/v1	NVE HydAPI base URL
NVE_MAX_RETRIES	ingester	3	Retry count for retryable NVE failures
NVE_RATE_LIMIT	ingester	5	NVE client requests per second (<=0 disables)
NVE_TIMEOUT	ingester	30s	Per-request NVE HTTP timeout
DATABASE_URL	ingester, api	postgres://water:water@localhost:5432/water?sslmode=disable	Postgres/TimescaleDB connection string
POLL_INTERVAL	ingester	5m	How often to poll NVE (Go duration)
STATION_IDS	ingester	2.32.0	Comma-separated NVE station ids; all enables discovery
MAX_STATIONS	ingester	25	Station discovery cap
PARAMETERS	ingester	1000,1001,1003	Comma-separated parameter codes to fetch
RESOLUTION_TIME	ingester	60	Resolution in minutes: 0 instantaneous, 60 hourly, 1440 daily
LOOKBACK	ingester	24h	How far back each poll requests observations (Go duration)
METRICS_ADDR	ingester	:9090	Ingester metrics/health listen address
ANOMALY_THRESHOLD	ingester	3	Z-score threshold for anomaly detection
ANOMALY_WINDOW	ingester	100	Rolling sample window for anomaly detection
ALERT_WEBHOOK_URL	ingester	empty	Optional anomaly alert webhook
API_ADDR	api	:8080	Listen address for the HTTP API
API_KEYS	api	empty	Comma-separated keys; empty leaves data endpoints public
API_RATE_LIMIT	api	10	Per-client data endpoint requests per second (<=0 disables)
API_RATE_BURST	api	20	Per-client rate limit burst
API_DEFAULT_PAGE_SIZE	api	500	Default observations/anomalies page size
API_MAX_PAGE_SIZE	api	5000	Maximum page size

Observability

Both binaries expose Prometheus metrics. The API serves them on its main listener at /metrics, while the ingester runs a separate listener (METRICS_ADDR, default :9090) that also handles its health checks. The compose stack runs Prometheus preconfigured to scrape both processes and provisions a Grafana dashboard covering pipeline and HTTP behavior. In AWS, CloudWatch metric and log-error alarms watch both services and can notify an email address through SNS.

Infrastructure

The terraform/ directory provisions a deployable AWS stack:

• VPC with public subnets for the ALB and private subnets for ECS/RDS.
• ECR repository for the image that contains both binaries.
• ECS/Fargate services for ingester and api.
• Internet-facing ALB for the API, with /healthz target checks.
• RDS Postgres in private subnets, with generated DB credentials.
• Secrets Manager entries for DATABASE_URL, NVE_API_KEY, optional API_KEYS, and optional ALERT_WEBHOOK_URL.
• CloudWatch log groups, metric alarms, log-error alarms, and an optional SNS email subscription via alert_email.

cd terraform
cp terraform.tfvars.example terraform.tfvars
terraform init
terraform plan

Set nve_api_key before applying if you want the ingester service to run, set api_keys to require API keys on the data endpoints, and set alert_email to receive CloudWatch alarm notifications.

License

MIT — see LICENSE.