#620 Monitoring – OTEL, Grafana, graphs and alerting

Makefile

@@ -281,7 +281,8 @@ run2: ## Start 2nd node
 .PHONY: start-monitoring
 start-monitoring: ## Start monitoring stack
 	@docker run -d --rm --add-host=host.docker.internal:host-gateway --name jaeger -p 4318:4318 -p 16686:16686 jaegertracing/jaeger:2.6.0
-	@docker run -d --rm --add-host=host.docker.internal:host-gateway --name prometheus -p 9101:9090 -v ./scripts/prometheus.yml:/etc/prometheus/prometheus.yml prom/prometheus  --config.file=/etc/prometheus/prometheus.yml --web.enable-remote-write-receiver
+	@docker run -d --rm --add-host=host.docker.internal:host-gateway --name alertmanager -p 9093:9093 -v ./scripts/alertmanager.yml:/etc/alertmanager/alertmanager.yml prom/alertmanager --config.file=/etc/alertmanager/alertmanager.yml
+	@docker run -d --rm --add-host=host.docker.internal:host-gateway --name prometheus -p 9101:9090 -v ./scripts/prometheus.yml:/etc/prometheus/prometheus.yml -v ./scripts/prometheus-rules.yml:/etc/prometheus/prometheus-rules.yml prom/prometheus  --config.file=/etc/prometheus/prometheus.yml --web.enable-remote-write-receiver
 	@docker run -d --rm --add-host=host.docker.internal:host-gateway --name=grafana -p 9100:3000 -v ./scripts/grafana_provisioning:/etc/grafana/provisioning grafana/grafana
 	@# Host CPU / memory / disk usage + disk I/O. On Linux this reports the real host;
 	@# on macOS Docker Desktop it reports the Docker VM (see scripts/prometheus.yml).
@@ -291,6 +292,7 @@ start-monitoring: ## Start monitoring stack
 stop-monitoring:
 	@docker stop grafana
 	@docker stop prometheus
+	@docker stop alertmanager
 	@docker stop jaeger
 	@docker stop node-exporter
 

docs/reference/observability.md

@@ -0,0 +1,135 @@
+# Observability
+
+ZenBPM exposes metrics via a Prometheus endpoint, distributed traces via OTLP
+(e.g. to Jaeger) and health endpoints for orchestrators and load balancers.
+
+## Endpoints
+
+| Endpoint | Purpose |
+| -------- | ------- |
+| `GET /system/metrics` | Prometheus metrics scrape endpoint |
+| `GET /system/status` | Verbose diagnostic status: full cluster state. Always returns **200** (legacy contract, kept stable for existing consumers) |
+| `GET /system/health/live` | Liveness probe. Returns 200 whenever the process is up. Does **not** check raft state, so a leaderless node is not restarted in a loop |
+| `GET /system/health/ready` | Readiness probe. Returns **503** until: the main cluster has a raft leader, the desired number of partitions exists, every partition has a leader, this node is registered in the cluster state, and all partitions owned by this node are initialized |
+
+Health responses have the shape:
+
+```json
+{"status": "DOWN", "reasons": ["no cluster leader elected", "partition 2 has no leader"]}
+```
+
+`/system/status` returns the raw cluster state (unchanged from previous releases).
+
+## Tracing
+
+- Configured via `tracing` section (`internal/config`): `enabled`, `endpoint`
+  (OTLP HTTP), `name`, `samplerRatio` (env `TRACING_SAMPLER_RATIO`, default
+  `1.0`; `0` disables root sampling; values outside `[0, 1]` fail startup).
+  Sampling is parent-based: child spans follow the parent decision.
+- Spans cover: REST requests, BPMN engine operations (instances, tokens,
+  flow nodes, jobs, timers, incidents), DMN decision evaluations, and rqlite
+  exec/query statements.
+- The W3C `TraceContext` + `Baggage` propagators are registered globally.
+  gRPC (public and node-to-node) is **not yet instrumented** — cross-partition
+  proxied requests currently start separate traces (tracked as tasks T1/T2 in
+  the observability task list).
+- Span attributes use the `zenbpm.` namespace (e.g. `zenbpm.process.instance_key`,
+  `zenbpm.job.key`, `zenbpm.decision.id`). Span (operation) names are stable —
+  e.g. DMN evaluations use `dmn.evaluate-decision` with the decision id carried
+  as an attribute — to keep tracing-backend operation indexes bounded.
+
+> ⚠️ **BREAKING CHANGE:** span attributes were renamed from the legacy `bpmn-*`
+> keys (dash separated) to the `zenbpm.*` namespace in this release, with no
+> dual-write overlap period. Saved Jaeger searches, collector `attributes`
+> processors and tail-sampling policies that reference the old keys **must** be
+> updated before upgrading. This must be called out in the release notes.
+
+## Metrics catalog
+
+Prometheus names shown (OpenTelemetry counters get a `_total` suffix, `ms`
+histograms a `_milliseconds` suffix).
+
+### Engine (per node)
+
+| Metric | Type | Attributes | Description |
+| ------ | ---- | ---------- | ----------- |
+| `processes_started_total` | counter | `bpmn_process_id` | Process instances started |
+| `processes_completed_total` | counter | `bpmn_process_id` | Process instances ended (completed or failed) |
+| `processes_running` | up/down counter | `bpmn_process_id` | Instances currently being executed |
+| `process_instance_duration_milliseconds` | histogram | `bpmn_process_id`, `state` | Creation → completion/failure duration |
+| `jobs_created_total` / `jobs_completed_total` / `jobs_failed_total` | counter | `type`, `internal` | Job lifecycle counters |
+| `job_lifetime_milliseconds` | histogram | `type`, `outcome` | Job creation → terminal state duration |
+| `incidents_created_total` / `incidents_resolved_total` | counter | `element_id` | Incident lifecycle counters (recorded only after the write batch is successfully flushed) |
+| `timers_scheduled_total` / `timers_fired_total` / `timers_cancelled_total` | counter | — | Timer lifecycle counters (recorded only after the write batch is successfully flushed; covers instance-level timers as well as definition-level timer start events and their cycle renewals) |
+| `messages_correlated_total` / `message_correlation_failed_total` | counter | `message_name` (+ `reason` on failures) | Message correlation outcomes. Failed lookups (no active subscription) use `message_name="unknown"`, `reason="subscription_not_found"` — the caller-provided name is never used as a label to keep cardinality bounded. Failures after a subscription was found use the (definition-bounded) real name with `reason="publish_failed"` |
+
+### DMN
+
+| Metric | Type | Attributes | Description |
+| ------ | ---- | ---------- | ----------- |
+| `dmn_evaluations_total` | counter | `decision_id`, `outcome` | Decision evaluations |
+| `dmn_evaluation_duration_milliseconds` | histogram | `decision_id` | Evaluation duration |
+
+### Cluster / raft (main Zen cluster, exported by every node)
+
+| Metric | Type | Attributes | Description |
+| ------ | ---- | ---------- | ----------- |
+| `cluster_has_leader` | gauge 0/1 | — | Main cluster has an elected leader |
+| `node_is_leader` | gauge 0/1 | — | This node is the main cluster leader |
+| `partition_has_leader` | gauge 0/1 | `partition` | Partition leader registered in cluster state (replicated view; see `partition_raft_has_leader` for the local raft view) |
+| `cluster_partitions` / `cluster_desired_partitions` | gauge | — | Actual vs. configured partition count; feeds the `PartitionDeficit` alert (missing partitions emit no `partition_has_leader` series, so a count comparison is required) |
+| `raft_term`, `raft_last_log_index`, `raft_applied_index`, `raft_fsm_pending` | gauge | — | Raft internals of the main cluster |
+
+### Partition / rqlite
+
+| Metric | Type | Attributes | Description |
+| ------ | ---- | ---------- | ----------- |
+| `jobs_waiting` | gauge | `partition` | Jobs waiting to be worked on. Exported by **every replica** of a partition — deduplicate with `max by(partition)` before aggregating |
+| `process_instances_active` | gauge | `partition` | Active process instances. Exported by every replica — deduplicate with `max by(partition)` |
+| `partition_raft_has_leader` | gauge 0/1 | `partition` | Partition raft group has a leader (local raft view; see `partition_has_leader` for the replicated cluster-state view) |
+| `partition_node_is_leader` | gauge 0/1 | `partition` | This node leads the partition raft group |
+| `partition_leader_changes_total` | counter | `partition` | New-leader elections observed by this node (leadership-loss observations are not counted) |
+| `rqlite_db_size_bytes` | gauge | `partition` | SQLite files size on disk (db + WAL/SHM) |
+| `rqlite_exec_duration_milliseconds` | histogram | `partition`, `outcome` | Raft-replicated write duration |
+| `rqlite_query_duration_milliseconds` | histogram | `partition`, `outcome` | Read query duration |
+
+### Job manager
+
+| Metric | Type | Attributes | Description |
+| ------ | ---- | ---------- | ----------- |
+| `jobs_distributed_total` | counter | `type`, `client` | Jobs successfully handed to worker streams |
+| `job_activation_latency_milliseconds` | histogram | `type` | Job creation → distribution latency (successful sends only; clamped at 0 to guard against cross-node clock skew) |
+
+### REST / runtime
+
+- `request_total`, `request_uri_total`, `request_body_size`, `response_body_size`,
+  `request_duration_milliseconds` — REST server. `request_uri_total` carries
+  `path`, `method` and `status` labels.
+- Go runtime metrics (`go_*`) — via `go.opentelemetry.io/contrib/instrumentation/runtime`.
+
+## Alerting
+
+`make start-monitoring` starts Prometheus with `scripts/prometheus-rules.yml`
+(technical + business alerts) and an Alertmanager (`scripts/alertmanager.yml`,
+placeholder webhook receiver). Key alerts:
+
+- **NoClusterLeader / NoPartitionLeader / PartitionDeficit** (critical) — driven
+  by the leadership and partition-count gauges exported by every node, not by
+  probing a single node's health URL. `PartitionDeficit` covers partitions that
+  were never created (absent series cannot fire `NoPartitionLeader`).
+- **TargetDown, HighErrorRate, RestLatencyDegradation, RqliteExecLatencyDegradation**
+- **RqliteDbSizeLarge, RqliteDbGrowthPrediction, DiskSpaceLow, HighCPU, HighMemory**
+- **ThroughputDrop, RaftLeaderFlapping, GoroutineLeak**
+- **IncidentCreated, HighJobFailureRate, JobBacklogGrowing, StuckProcessInstances, NoJobDistribution**
+
+## Dashboards
+
+Provisioned automatically from `scripts/grafana_provisioning/dashboards/zenbpm/`:
+
+- `main.json` — processes, jobs, distribution, request duration
+- `cluster.json` — leadership, partition deficit, raft health, leader changes
+- `incidents.json` — incidents, job failures, message correlation, timers, DMN
+- `storage.json` — rqlite DB size/growth, read/write latency percentiles, disk
+- `latency.json` — business latency percentiles and throughput
+- `host.json` — node_exporter CPU/memory/disk/network
+- `go.json` — Go runtime

go.mod

@@ -25,6 +25,7 @@ require (
 	github.com/senseyeio/duration v0.0.0-20180430131211-7c2a214ada46
 	github.com/stretchr/testify v1.11.1
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0
+	go.opentelemetry.io/contrib/instrumentation/runtime v0.69.0
 	go.opentelemetry.io/otel v1.44.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.44.0

go.sum

@@ -337,6 +337,8 @@ go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ
 go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 h1:8tvICD4vSTOOsNrsI4Ljf6C+6UKvpTEH5XY3JMoyPoo=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0/go.mod h1:z9+yiacE0IHRqM4qFfkbt/JYlmYXgss8GY/jXoNuPJI=
+go.opentelemetry.io/contrib/instrumentation/runtime v0.69.0 h1:MtkMsuRo3zEXTTMALfyrszwCDZTkB6wolyPjbwFAdq0=
+go.opentelemetry.io/contrib/instrumentation/runtime v0.69.0/go.mod h1:FYTxnpsm+UPD0erZNq20GvnM8T2YQHiHtT2vokdpoac=
 go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU=
 go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 h1:4YsVu3B8+3qtWYYrsUYgn0OG78pN0rnNPRGX4SbokQI=

internal/cluster/jobmanager/otel.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 
+	otelPkg "github.com/pbinitiative/zenbpm/pkg/otel"
 	"go.opentelemetry.io/otel"
 
 	"go.opentelemetry.io/otel/metric"
@@ -15,16 +16,22 @@ const (
 
 var (
 	JobsDistributed metric.Int64Counter
+	// JobActivationLatency measures time between job creation and its distribution to a worker, in ms.
+	JobActivationLatency metric.Float64Histogram
 )
 
 func registerMetrics() error {
 	var err error
 	var errJoin error
 	JobsDistributed, err = otel.Meter(jobManagerMeter).Int64Counter("jobs_distributed", metric.WithDescription("Number of jobs sent to the clients"))
 	errJoin = errors.Join(errJoin, err)
+	JobActivationLatency, err = otel.Meter(jobManagerMeter).Float64Histogram("job_activation_latency",
+		metric.WithUnit("ms"),
+		metric.WithDescription("Time between job creation and distribution to a worker, milliseconds"),
+		metric.WithExplicitBucketBoundaries(otelPkg.LatencyBucketsMs()...))
 	errJoin = errors.Join(errJoin, err)
 	if errJoin != nil {
-		return fmt.Errorf("failed to create otel instruments: %w", err)
+		return fmt.Errorf("failed to create otel instruments: %w", errJoin)
 	}
 	return nil
 }

internal/cluster/jobmanager/server.go

@@ -233,14 +233,23 @@ func (s *jobServer) distributeJobs() {
 					CreatedAt:      &job.CreatedAt,
 				},
 			})
-			JobsDistributed.Add(context.Background(), 1, metric.WithAttributes(
-				attribute.String("type", job.Type),
-				attribute.String("client", string(clientID)),
-			))
 			if err != nil {
 				s.logger.Error("Failed to send job to node", "jobType", jType, "key", job.Key, "err", err)
 				continue
 			}
+			JobsDistributed.Add(s.ctx, 1, metric.WithAttributes(
+				attribute.String("type", job.Type),
+				attribute.String("client", string(clientID)),
+			))
+			if JobActivationLatency != nil && job.CreatedAt > 0 {
+				latencyMs := float64(time.Now().UnixMilli() - job.CreatedAt)
+				if latencyMs < 0 {
+					latencyMs = 0
+				}
+				JobActivationLatency.Record(s.ctx, latencyMs, metric.WithAttributes(
+					attribute.String("type", job.Type),
+				))
+			}
 		}
 	}
 }

internal/cluster/node.go

@@ -11,6 +11,7 @@ import (
 	"hash/fnv"
 	"net"
 	"slices"
+	"sort"
 	"time"
 
 	"github.com/bwmarrin/snowflake"
@@ -113,6 +114,9 @@ func StartZenNode(mainCtx context.Context, conf config.Config) (*ZenNode, error)
 	if err = node.store.Open(); err != nil {
 		return nil, fmt.Errorf("failed to open store: %w", err)
 	}
+	if err = node.store.RegisterMetrics(); err != nil {
+		node.logger.Error("Failed to register cluster store metrics", "err", err)
+	}
 
 	node.client = client.NewClientManager(node.store)
 	err = node.controller.Start(node.store, node.client)
@@ -1868,6 +1872,45 @@ func (node *ZenNode) GetStatus() state.Cluster {
 	return node.store.ClusterState()
 }
 
+// Health evaluates the readiness of this node in the cluster. It returns
+// false together with the list of reasons when:
+//   - the main Zen cluster has no elected raft leader,
+//   - fewer partitions exist in the cluster state than desired,
+//   - any partition has no leader registered in the cluster state,
+//   - this node is not (yet) present in the cluster state,
+//   - this node owns a partition that is not in the Initialized state.
+func (node *ZenNode) Health() (bool, []string) {
+	if node.store == nil {
+		return false, []string{"cluster store is not initialized"}
+	}
+	reasons := make([]string, 0)
+	if !node.store.HasLeader() {
+		reasons = append(reasons, "no cluster leader elected")
+	}
+	cs := node.store.ClusterState()
+	if desired := cs.Config.DesiredPartitions; int64(len(cs.Partitions)) < int64(desired) {
+		reasons = append(reasons, fmt.Sprintf("cluster has %d of %d desired partitions", len(cs.Partitions), desired))
+	}
+	partitionReasons := make([]string, 0)
+	for id, partition := range cs.Partitions {
+		if partition.LeaderId == "" {
+			partitionReasons = append(partitionReasons, fmt.Sprintf("partition %d has no leader", id))
+		}
+	}
+	if self, err := cs.GetNode(node.store.NodeID()); err != nil {
+		partitionReasons = append(partitionReasons, "node is not registered in the cluster state")
+	} else {
+		for id, partition := range self.Partitions {
+			if partition.State != state.NodePartitionStateInitialized {
+				partitionReasons = append(partitionReasons, fmt.Sprintf("partition %d on this node is in state %s", id, partition.State))
+			}
+		}
+	}
+	sort.Strings(partitionReasons)
+	reasons = append(reasons, partitionReasons...)
+	return len(reasons) == 0, reasons
+}
+
 func (node *ZenNode) StartProcessInstanceOnElements(ctx context.Context, processDefinitionKey int64, startingElementIds []string, variables map[string]any) (*proto.ProcessInstance, error) {
 	state := node.store.ClusterState()
 	candidateNode, err := state.GetLeastStressedPartitionLeader()