From 0f67945a77948fdf830c8f4a6370c5ec0dcb2083 Mon Sep 17 00:00:00 2001 From: alisher Date: Thu, 30 Jul 2026 22:00:53 +0200 Subject: [PATCH 1/4] =?UTF-8?q?#620=20Monitoring=20=E2=80=93=20OTEL,=20Gra?= =?UTF-8?q?fana,=20graphs=20and=20alerting=20-=20Introduced=20metrics=20fo?= =?UTF-8?q?r=20tracking=20cluster=20leader=20presence,=20node=20leadership?= =?UTF-8?q?=20status,=20and=20partition=20counts.=20-=20Added=20metrics=20?= =?UTF-8?q?for=20job=20lifetime,=20process=20instance=20duration,=20and=20?= =?UTF-8?q?message=20correlation=20failures.=20-=20Enhanced=20the=20monito?= =?UTF-8?q?ring=20stack=20with=20Prometheus=20configuration=20for=20alerti?= =?UTF-8?q?ng.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Makefile | 4 +- docs/reference/observability.md | 135 +++++++++ go.mod | 1 + go.sum | 2 + internal/cluster/jobmanager/otel.go | 9 +- internal/cluster/jobmanager/server.go | 17 +- internal/cluster/node.go | 43 +++ internal/cluster/partition/partition.go | 78 +++++ .../partition/partition_persistence.go | 45 ++- internal/cluster/store/metrics.go | 136 +++++++++ internal/cluster/store/store.go | 5 + internal/cluster/store/store_helper.go | 6 + internal/config/config.go | 3 + internal/otel/metrics.go | 22 +- internal/otel/tracer.go | 21 +- internal/otel/tracer_test.go | 50 ++++ internal/rest/health_test.go | 50 ++++ internal/rest/server.go | 57 +++- pkg/bpmn/engine.go | 3 + pkg/bpmn/engine_api.go | 6 + pkg/bpmn/engine_batch.go | 65 ++++- pkg/bpmn/engine_metrics_test.go | 95 +++++++ pkg/bpmn/events_api.go | 29 ++ pkg/bpmn/jobs_api.go | 14 + .../start_event_instance_creation_handler.go | 4 + pkg/dmn/dmn_engine.go | 58 +++- pkg/dmn/dmn_multiline_test.go | 8 +- pkg/otel/metrics.go | 91 +++++- pkg/otel/traces.go | 31 +- scripts/alertmanager.yml | 26 ++ .../dashboards/zenbpm/cluster.json | 116 ++++++++ .../dashboards/zenbpm/host.json | 92 ++++++ .../dashboards/zenbpm/incidents.json | 122 ++++++++ .../dashboards/zenbpm/latency.json | 99 +++++++ .../dashboards/zenbpm/storage.json | 95 +++++++ scripts/prometheus-rules.yml | 266 ++++++++++++++++++ scripts/prometheus.yml | 8 + 37 files changed, 1872 insertions(+), 40 deletions(-) create mode 100644 docs/reference/observability.md create mode 100644 internal/cluster/store/metrics.go create mode 100644 internal/otel/tracer_test.go create mode 100644 internal/rest/health_test.go create mode 100644 pkg/bpmn/engine_metrics_test.go create mode 100644 scripts/alertmanager.yml create mode 100644 scripts/grafana_provisioning/dashboards/zenbpm/cluster.json create mode 100644 scripts/grafana_provisioning/dashboards/zenbpm/host.json create mode 100644 scripts/grafana_provisioning/dashboards/zenbpm/incidents.json create mode 100644 scripts/grafana_provisioning/dashboards/zenbpm/latency.json create mode 100644 scripts/grafana_provisioning/dashboards/zenbpm/storage.json create mode 100644 scripts/prometheus-rules.yml diff --git a/Makefile b/Makefile index da3a562ff..cb161fad5 100644 --- a/Makefile +++ b/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 diff --git a/docs/reference/observability.md b/docs/reference/observability.md new file mode 100644 index 000000000..054839f8d --- /dev/null +++ b/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, 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 diff --git a/go.mod b/go.mod index 05291ecae..fb1b6a71e 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index c5e19d748..471fa1eb2 100644 --- a/go.sum +++ b/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= diff --git a/internal/cluster/jobmanager/otel.go b/internal/cluster/jobmanager/otel.go index ba2f00111..6f584c0d1 100644 --- a/internal/cluster/jobmanager/otel.go +++ b/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,6 +16,8 @@ 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 { @@ -22,9 +25,13 @@ func registerMetrics() 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 } diff --git a/internal/cluster/jobmanager/server.go b/internal/cluster/jobmanager/server.go index 9d3628cc2..7f631b47f 100644 --- a/internal/cluster/jobmanager/server.go +++ b/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), + )) + } } } } diff --git a/internal/cluster/node.go b/internal/cluster/node.go index b32eb917a..9788671dc 100644 --- a/internal/cluster/node.go +++ b/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; uint32(len(cs.Partitions)) < 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() diff --git a/internal/cluster/partition/partition.go b/internal/cluster/partition/partition.go index 17ee69052..bd34d9b90 100644 --- a/internal/cluster/partition/partition.go +++ b/internal/cluster/partition/partition.go @@ -5,6 +5,8 @@ import ( "errors" "fmt" "net" + "os" + "path/filepath" "strings" "sync" "time" @@ -82,6 +84,22 @@ func (zpn *ZenPartitionNode) createMetrics() { if err != nil { zpn.logger.Error("Failed to register meter for jobsWaiting", "err", err) } + zpn.metrics.hasLeader, err = otel.Meter(partitionMeter).Int64Gauge("partition_raft_has_leader", metric.WithDescription("1 when the partition raft group has an elected leader (local raft view; see partition_has_leader for the replicated cluster-state view), 0 otherwise")) + if err != nil { + zpn.logger.Error("Failed to register meter for hasLeader", "err", err) + } + zpn.metrics.isLeader, err = otel.Meter(partitionMeter).Int64Gauge("partition_node_is_leader", metric.WithDescription("1 when this node is the partition raft leader, 0 otherwise")) + if err != nil { + zpn.logger.Error("Failed to register meter for isLeader", "err", err) + } + zpn.metrics.dbSize, err = otel.Meter(partitionMeter).Int64Gauge("rqlite_db_size", metric.WithUnit("By"), metric.WithDescription("Size of the partition SQLite database files on disk, bytes")) + if err != nil { + zpn.logger.Error("Failed to register meter for dbSize", "err", err) + } + zpn.metrics.leaderChanges, err = otel.Meter(partitionMeter).Int64Counter("partition_leader_changes", metric.WithDescription("Number of partition raft leader changes observed by this node")) + if err != nil { + zpn.logger.Error("Failed to register meter for leaderChanges", "err", err) + } } type PartitionChangesCallbacks struct { @@ -95,6 +113,14 @@ type PartitionChangesCallbacks struct { type partitionMetrics struct { jobsWaiting metric.Int64Gauge processInstancesActive metric.Int64Gauge + // hasLeader reports whether the partition raft group currently has a leader (0/1) + hasLeader metric.Int64Gauge + // isLeader reports whether this node is the partition raft leader (0/1) + isLeader metric.Int64Gauge + // dbSize reports the size of the partition SQLite database files on disk, in bytes + dbSize metric.Int64Gauge + // leaderChanges counts partition raft leader changes observed by this node + leaderChanges metric.Int64Counter } const ( @@ -485,6 +511,11 @@ func (zpn *ZenPartitionNode) observe() (closeCh, doneCh chan struct{}) { } } case raft.LeaderObservation: + if zpn.metrics.leaderChanges != nil && signal.LeaderID != "" { + zpn.metrics.leaderChanges.Add(context.Background(), 1, metric.WithAttributes( + attribute.Int64("partition", int64(zpn.PartitionId)), + )) + } if zpn.stateChangeCallbacks.LeaderChange == nil { break } @@ -514,6 +545,31 @@ func (zpn *ZenPartitionNode) observe() (closeCh, doneCh chan struct{}) { func (zpn *ZenPartitionNode) updatePartitionMetrics() { ctx := context.Background() + + partitionAttr := metric.WithAttributes(attribute.Int64("partition", int64(zpn.PartitionId))) + if zpn.metrics.hasLeader != nil { + leaderAddr, _ := zpn.store.LeaderAddr() + hasLeader := int64(0) + if leaderAddr != "" { + hasLeader = 1 + } + zpn.metrics.hasLeader.Record(ctx, hasLeader, partitionAttr) + } + if zpn.metrics.isLeader != nil { + isLeader := int64(0) + if zpn.store.IsLeader() { + isLeader = 1 + } + zpn.metrics.isLeader.Record(ctx, isLeader, partitionAttr) + } + if zpn.metrics.dbSize != nil { + if size, err := zpn.dbSizeBytes(); err != nil { + zpn.logger.Debug("Failed to compute rqlite db size", "partition", zpn.PartitionId, "err", err) + } else { + zpn.metrics.dbSize.Record(ctx, size, partitionAttr) + } + } + g, gCtx := errgroup.WithContext(ctx) var waitingJobs int64 @@ -546,6 +602,28 @@ func (zpn *ZenPartitionNode) updatePartitionMetrics() { )) } +// dbSizeBytes returns the combined on-disk size of the partition SQLite +// database files (main db + WAL/SHM) in the rqlite data directory. +func (zpn *ZenPartitionNode) dbSizeBytes() (int64, error) { + if zpn.config == nil || zpn.config.DataPath == "" { + return 0, fmt.Errorf("no data path configured") + } + matches, err := filepath.Glob(filepath.Join(zpn.config.DataPath, "db.sqlite*")) + if err != nil { + return 0, fmt.Errorf("failed to glob sqlite files: %w", err) + } + var total int64 + for _, match := range matches { + info, err := os.Stat(match) + if err != nil { + zpn.logger.Debug("Failed to stat sqlite file for db size metric", "file", match, "err", err) + continue + } + total += info.Size() + } + return total, nil +} + func (zpn *ZenPartitionNode) createCredentialStore(cfg *config.RqLite) (*auth.CredentialsStore, error) { if cfg.AuthFile == "" { return nil, nil diff --git a/internal/cluster/partition/partition_persistence.go b/internal/cluster/partition/partition_persistence.go index 5b9133d12..71f379308 100644 --- a/internal/cluster/partition/partition_persistence.go +++ b/internal/cluster/partition/partition_persistence.go @@ -40,6 +40,7 @@ import ( "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/metric" "go.opentelemetry.io/otel/trace" ) @@ -59,6 +60,10 @@ type DB struct { cleanupCancel context.CancelFunc cleanupDone chan struct{} cleanupStopOnce sync.Once + // execDuration measures the duration of rqlite write statements, in ms + execDuration metric.Float64Histogram + // queryDuration measures the duration of rqlite read queries, in ms + queryDuration metric.Float64Histogram } const ( @@ -142,6 +147,20 @@ func newDB(store *store.Store, partition uint32, logger hclog.Logger, cfg config queries := sql.New(db) db.Queries = queries + meter := otel.GetMeterProvider().Meter("partition-rqlite") + db.execDuration, err = meter.Float64Histogram("rqlite_exec_duration", + metric.WithUnit("ms"), + metric.WithDescription("Duration of rqlite write statements, milliseconds")) + if err != nil { + logger.Error("Failed to create rqlite_exec_duration instrument", "err", err) + } + db.queryDuration, err = meter.Float64Histogram("rqlite_query_duration", + metric.WithUnit("ms"), + metric.WithDescription("Duration of rqlite read queries, milliseconds")) + if err != nil { + logger.Error("Failed to create rqlite_query_duration instrument", "err", err) + } + if opts.startDataCleanup { cleanupCtx, cancel := context.WithCancel(context.Background()) db.cleanupCancel = cancel @@ -403,12 +422,23 @@ func (r rqliteResult) RowsAffected() (int64, error) { return r.rowsAffected, nil } -func (rq *DB) ExecContext(ctx context.Context, sql string, args ...interface{}) (ssql.Result, error) { +func (rq *DB) ExecContext(ctx context.Context, sql string, args ...interface{}) (_ ssql.Result, retErr error) { + start := time.Now() ctx, execSpan := rq.tracer.Start(ctx, "rqlite-exec", trace.WithAttributes( attribute.String(otelPkg.AttributeExec, sql), attribute.String(otelPkg.AttributeArgs, fmt.Sprintf("%v", args)), )) defer func() { + if rq.execDuration != nil { + outcome := "success" + if retErr != nil { + outcome = "error" + } + rq.execDuration.Record(ctx, float64(time.Since(start))/float64(time.Millisecond), metric.WithAttributes( + attribute.Int64("partition", int64(rq.Partition)), + attribute.String("outcome", outcome), + )) + } execSpan.End() }() stmt, err := rq.generateStatement(sql, args...) @@ -444,12 +474,23 @@ func (rq *DB) PrepareContext(ctx context.Context, sql string) (*ssql.Stmt, error return nil, errors.New("PrepareContext not supported by rqlite") } -func (rq *DB) QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error) { +func (rq *DB) QueryContext(ctx context.Context, query string, args ...interface{}) (_ *sql.Rows, retErr error) { + start := time.Now() ctx, querySpan := rq.tracer.Start(ctx, "rqlite-query", trace.WithAttributes( attribute.String(otelPkg.AttributeQuery, query), attribute.String(otelPkg.AttributeArgs, fmt.Sprintf("%v", args)), )) defer func() { + if rq.queryDuration != nil { + outcome := "success" + if retErr != nil { + outcome = "error" + } + rq.queryDuration.Record(ctx, float64(time.Since(start))/float64(time.Millisecond), metric.WithAttributes( + attribute.Int64("partition", int64(rq.Partition)), + attribute.String("outcome", outcome), + )) + } querySpan.End() }() results, err := rq.queryDatabase(ctx, query, args...) diff --git a/internal/cluster/store/metrics.go b/internal/cluster/store/metrics.go new file mode 100644 index 000000000..d9a390c74 --- /dev/null +++ b/internal/cluster/store/metrics.go @@ -0,0 +1,136 @@ +package store + +import ( + "context" + "strconv" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +const storeMeter = "zen-store" + +// RegisterMetrics registers observable gauges that report main (Zen) cluster +// raft health: leader presence, this node's leadership status, per-partition +// leader presence (from the replicated cluster state), partition counts and +// raw raft statistics (term, log indexes, FSM backlog). +// +// These gauges are the data source for the NoClusterLeader / NoPartitionLeader / +// PartitionDeficit Prometheus alerts, so they are exported by every node: +// alerting does not depend on a single node being scrapeable. +// +// The callback registration is retained on the store and released in Close so +// the global meter provider does not keep closed stores alive. +func (s *Store) RegisterMetrics() error { + meter := otel.Meter(storeMeter) + + clusterHasLeader, err := meter.Int64ObservableGauge("cluster_has_leader", + metric.WithDescription("1 when the main Zen cluster has an elected raft leader, 0 otherwise")) + if err != nil { + return err + } + nodeIsLeader, err := meter.Int64ObservableGauge("node_is_leader", + metric.WithDescription("1 when this node is the main Zen cluster raft leader, 0 otherwise")) + if err != nil { + return err + } + partitionHasLeader, err := meter.Int64ObservableGauge("partition_has_leader", + metric.WithDescription("1 when the partition has a leader registered in the cluster state (replicated view; see partition_raft_has_leader for the local raft view), 0 otherwise")) + if err != nil { + return err + } + // partition existence gauges: partition_has_leader only emits series for + // partitions that already exist in the cluster state, so a bootstrap + // failure that leaves partitions missing would otherwise be invisible to + // alerting (absent series never fire NoPartitionLeader) + clusterPartitions, err := meter.Int64ObservableGauge("cluster_partitions", + metric.WithDescription("Number of partitions currently present in the cluster state")) + if err != nil { + return err + } + clusterDesiredPartitions, err := meter.Int64ObservableGauge("cluster_desired_partitions", + metric.WithDescription("Number of partitions the cluster is configured to have")) + if err != nil { + return err + } + raftTerm, err := meter.Int64ObservableGauge("raft_term", + metric.WithDescription("Current raft term of the main Zen cluster")) + if err != nil { + return err + } + raftLastLogIndex, err := meter.Int64ObservableGauge("raft_last_log_index", + metric.WithDescription("Last raft log index of the main Zen cluster")) + if err != nil { + return err + } + raftAppliedIndex, err := meter.Int64ObservableGauge("raft_applied_index", + metric.WithDescription("Last applied raft log index of the main Zen cluster")) + if err != nil { + return err + } + raftFsmPending, err := meter.Int64ObservableGauge("raft_fsm_pending", + metric.WithDescription("Number of raft log entries pending application to the FSM")) + if err != nil { + return err + } + + reg, err := meter.RegisterCallback(func(ctx context.Context, o metric.Observer) error { + if !s.open.Load() { + return nil + } + o.ObserveInt64(clusterHasLeader, boolToInt64(s.HasLeader())) + o.ObserveInt64(nodeIsLeader, boolToInt64(s.IsLeader())) + + cs := s.ClusterState() + o.ObserveInt64(clusterPartitions, int64(len(cs.Partitions))) + o.ObserveInt64(clusterDesiredPartitions, int64(cs.Config.DesiredPartitions)) + for id, partition := range cs.Partitions { + o.ObserveInt64(partitionHasLeader, boolToInt64(partition.LeaderId != ""), + metric.WithAttributes(attribute.Int64("partition", int64(id)))) + } + + stats := s.raft.Stats() + observeRaftStat(o, raftTerm, stats, "term") + observeRaftStat(o, raftLastLogIndex, stats, "last_log_index") + observeRaftStat(o, raftAppliedIndex, stats, "applied_index") + observeRaftStat(o, raftFsmPending, stats, "fsm_pending") + return nil + }, clusterHasLeader, nodeIsLeader, partitionHasLeader, clusterPartitions, clusterDesiredPartitions, raftTerm, raftLastLogIndex, raftAppliedIndex, raftFsmPending) + if err != nil { + return err + } + s.metricsRegistration = reg + return nil +} + +// unregisterMetrics releases the observable callback registered by +// RegisterMetrics so a closed store is no longer referenced by the global +// meter provider. Safe to call when metrics were never registered. +func (s *Store) unregisterMetrics() error { + if s.metricsRegistration == nil { + return nil + } + err := s.metricsRegistration.Unregister() + s.metricsRegistration = nil + return err +} + +func observeRaftStat(o metric.Observer, gauge metric.Int64ObservableGauge, stats map[string]string, key string) { + raw, ok := stats[key] + if !ok { + return + } + value, err := strconv.ParseInt(raw, 10, 64) + if err != nil { + return + } + o.ObserveInt64(gauge, value) +} + +func boolToInt64(b bool) int64 { + if b { + return 1 + } + return 0 +} diff --git a/internal/cluster/store/store.go b/internal/cluster/store/store.go index b2893263f..b9643bfc8 100644 --- a/internal/cluster/store/store.go +++ b/internal/cluster/store/store.go @@ -20,6 +20,7 @@ import ( "github.com/pbinitiative/zenbpm/internal/safego" "github.com/pbinitiative/zenbpm/pkg/ptr" "github.com/rqlite/rqlite/v10/tcp" + "go.opentelemetry.io/otel/metric" pb "google.golang.org/protobuf/proto" ) @@ -63,6 +64,10 @@ type Store struct { state state.Cluster clusterStateChangeObserver ClusterStateObserverFunc + + // metricsRegistration is the otel observable-callback registration created + // by RegisterMetrics; released in Close via unregisterMetrics + metricsRegistration metric.Registration } type Config struct { diff --git a/internal/cluster/store/store_helper.go b/internal/cluster/store/store_helper.go index 312865ffb..81ab42708 100644 --- a/internal/cluster/store/store_helper.go +++ b/internal/cluster/store/store_helper.go @@ -416,6 +416,12 @@ func (s *Store) Close(wait bool) (retErr error) { return nil } + // release the otel callback registration so the global meter provider + // does not keep this closed store alive + if err := s.unregisterMetrics(); err != nil { + s.logger.Warn(fmt.Sprintf("failed to unregister store metrics: %s", err)) + } + close(s.observerClose) <-s.observerDone diff --git a/internal/config/config.go b/internal/config/config.go index 8fc274bac..a43a45cc3 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -94,6 +94,9 @@ type Tracing struct { Name string `yaml:"name" json:"name" env:"TRACING_APP_NAME" env-default:"ZenBPM"` // application identifier TransferHeaders []string `yaml:"transferHeaders" json:"transferHeaders" env:"TRACING_TRANSFER_HEADERS"` Endpoint string `yaml:"endpoint" env:"OTEL_EXPORTER_OTLP_ENDPOINT"` + // SamplerRatio controls the fraction of new traces that get sampled (0.0 - 1.0). + // Child spans follow the sampling decision of their parent (ParentBased sampler). + SamplerRatio float64 `yaml:"samplerRatio" json:"samplerRatio" env:"TRACING_SAMPLER_RATIO" env-default:"1.0"` } type Persistence struct { diff --git a/internal/otel/metrics.go b/internal/otel/metrics.go index 5cea28bcf..bf0f9d099 100644 --- a/internal/otel/metrics.go +++ b/internal/otel/metrics.go @@ -6,10 +6,13 @@ import ( "fmt" "github.com/pbinitiative/zenbpm/internal/config" + "github.com/pbinitiative/zenbpm/internal/log" + "go.opentelemetry.io/contrib/instrumentation/runtime" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/exporters/prometheus" metrics "go.opentelemetry.io/otel/metric" + "go.opentelemetry.io/otel/propagation" "go.opentelemetry.io/otel/sdk/metric" "go.opentelemetry.io/otel/sdk/resource" "go.opentelemetry.io/otel/sdk/trace" @@ -35,17 +38,34 @@ func SetupOtel(conf config.Tracing) (*Otel, error) { o := Otel{} var err error + // fail fast on invalid sampler configuration even when tracing is disabled, + // so a latent misconfiguration does not explode on the day tracing is enabled + if err := validateSamplerRatio(conf.SamplerRatio); err != nil { + return nil, err + } + + // register W3C trace context + baggage propagators globally so that any + // instrumentation relying on the global propagator (HTTP, gRPC) joins traces + otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator( + propagation.TraceContext{}, + propagation.Baggage{}, + )) + o.meterProvider, err = setupMeterProvider(conf.Name) if err != nil { return nil, err } otel.SetMeterProvider(o.meterProvider) + if err := runtime.Start(runtime.WithMeterProvider(o.meterProvider)); err != nil { + log.Error("failed to start runtime metrics instrumentation: %s", err) + } if conf.Enabled { o.tracerprovider, err = setupTraceProvider(conf) - otel.SetTracerProvider(o.tracerprovider) if err != nil { + o.Stop(context.Background()) return nil, fmt.Errorf("failed to set up tracer: %w", err) } + otel.SetTracerProvider(o.tracerprovider) } return &o, nil diff --git a/internal/otel/tracer.go b/internal/otel/tracer.go index 7b63aec74..fa27710c0 100644 --- a/internal/otel/tracer.go +++ b/internal/otel/tracer.go @@ -3,6 +3,7 @@ package otel import ( "context" "fmt" + "math" "strings" "time" @@ -14,7 +15,24 @@ import ( semconv "go.opentelemetry.io/otel/semconv/v1.41.0" ) +// validateSamplerRatio checks that the configured trace sampler ratio is a +// finite number within [0, 1]. It is validated regardless of whether tracing +// is enabled so that a latent misconfiguration fails fast at startup instead +// of on the day tracing gets switched on. +func validateSamplerRatio(ratio float64) error { + if math.IsNaN(ratio) || math.IsInf(ratio, 0) || ratio < 0 || ratio > 1 { + return fmt.Errorf("tracing sampler ratio must be between 0 and 1, got %v", ratio) + } + return nil +} + func setupTraceProvider(conf config.Tracing) (*trace.TracerProvider, error) { + // sample new traces at the configured ratio; child spans follow their parent decision. + // An explicit 0 disables root sampling; out-of-range values are a configuration error. + // Validated before any exporter/resource setup to avoid avoidable work on bad config. + if err := validateSamplerRatio(conf.SamplerRatio); err != nil { + return nil, err + } endpoint := conf.Endpoint endpoint = strings.TrimPrefix(endpoint, "https://") endpoint = strings.TrimPrefix(endpoint, "http://") @@ -46,10 +64,11 @@ func setupTraceProvider(conf config.Tracing) (*trace.TracerProvider, error) { } tracerprovider := trace.NewTracerProvider( + trace.WithSampler(trace.ParentBased(trace.TraceIDRatioBased(conf.SamplerRatio))), trace.WithBatcher( exporter, trace.WithMaxExportBatchSize(trace.DefaultMaxExportBatchSize), - trace.WithBatchTimeout(trace.DefaultScheduleDelay*time.Millisecond), + trace.WithBatchTimeout(5*time.Second), ), trace.WithResource(res), ) diff --git a/internal/otel/tracer_test.go b/internal/otel/tracer_test.go new file mode 100644 index 000000000..21505a6a2 --- /dev/null +++ b/internal/otel/tracer_test.go @@ -0,0 +1,50 @@ +package otel + +import ( + "math" + "testing" + + "github.com/pbinitiative/zenbpm/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestValidateSamplerRatio(t *testing.T) { + tests := []struct { + name string + ratio float64 + wantErr bool + }{ + {name: "zero disables root sampling", ratio: 0, wantErr: false}, + {name: "half", ratio: 0.5, wantErr: false}, + {name: "one samples everything", ratio: 1, wantErr: false}, + {name: "negative", ratio: -0.1, wantErr: true}, + {name: "above one", ratio: 1.1, wantErr: true}, + {name: "NaN", ratio: math.NaN(), wantErr: true}, + {name: "positive infinity", ratio: math.Inf(1), wantErr: true}, + {name: "negative infinity", ratio: math.Inf(-1), wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := validateSamplerRatio(tt.ratio) + if tt.wantErr { + assert.Error(t, err) + } else { + assert.NoError(t, err) + } + }) + } +} + +func TestSetupOtelRejectsInvalidSamplerRatioEvenWhenTracingDisabled(t *testing.T) { + // an invalid sampler ratio must fail fast at startup even with tracing + // disabled, so the misconfiguration does not surface on the day tracing + // gets switched on + o, err := SetupOtel(config.Tracing{ + Enabled: false, + Name: "test", + SamplerRatio: 2.0, + }) + require.Error(t, err) + assert.Nil(t, o) +} diff --git a/internal/rest/health_test.go b/internal/rest/health_test.go new file mode 100644 index 000000000..380411961 --- /dev/null +++ b/internal/rest/health_test.go @@ -0,0 +1,50 @@ +package rest + +import ( + "encoding/json" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestWriteHealthResponseHealthy(t *testing.T) { + rec := httptest.NewRecorder() + + writeHealthResponse(rec, slog.Default(), true, nil) + + assert.Equal(t, http.StatusOK, rec.Code) + assert.Equal(t, "application/json", rec.Header().Get("Content-Type")) + body := decodeHealthResponse(t, rec) + assert.Equal(t, "UP", body.Status) + assert.Empty(t, body.Reasons) + // nil reasons must serialize as an empty array, not null + assert.Contains(t, rec.Body.String(), `"reasons":[]`) +} + +func TestWriteHealthResponseUnhealthy(t *testing.T) { + rec := httptest.NewRecorder() + reasons := []string{"no cluster leader elected", "partition 2 has no leader"} + + writeHealthResponse(rec, slog.Default(), false, reasons) + + assert.Equal(t, http.StatusServiceUnavailable, rec.Code) + body := decodeHealthResponse(t, rec) + assert.Equal(t, "DOWN", body.Status) + assert.Equal(t, reasons, body.Reasons) +} + +type healthResponse struct { + Status string `json:"status"` + Reasons []string `json:"reasons"` +} + +func decodeHealthResponse(t *testing.T, rec *httptest.ResponseRecorder) healthResponse { + t.Helper() + var body healthResponse + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &body)) + return body +} diff --git a/internal/rest/server.go b/internal/rest/server.go index 03982ad72..0e7ada102 100644 --- a/internal/rest/server.go +++ b/internal/rest/server.go @@ -98,20 +98,69 @@ func NewServer(node *cluster.ZenNode, conf config.Config) *Server { // register system endpoints r.Route("/system", func(r chi.Router) { r.Get("/metrics", promhttp.Handler().ServeHTTP) + // verbose diagnostic endpoint. Deliberately keeps the legacy contract + // (raw cluster state, always 200) for existing consumers; readiness + // semantics live exclusively on /system/health/ready below. r.Get("/status", func(w http.ResponseWriter, r *http.Request) { - state, _ := json.MarshalIndent(node.GetStatus(), "", " ") - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(200) - _, err := w.Write(state) + body, err := json.MarshalIndent(node.GetStatus(), "", " ") if err != nil { + restLogger.Error("failed to marshal status", "error", err) + w.WriteHeader(http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + if _, err := w.Write(body); err != nil { restLogger.Error("failed to write status", "error", err) return } }) + // liveness probe: reports only that the process is up. It deliberately does not check raft state + // so that a leaderless node is not restarted in a loop by an orchestrator. + r.Get("/health/live", func(w http.ResponseWriter, r *http.Request) { + writeHealthResponse(w, restLogger, true, nil) + }) + // readiness probe: 503 until the cluster has a leader, every partition + // has a leader and all partitions owned by this node are initialized. + r.Get("/health/ready", func(w http.ResponseWriter, r *http.Request) { + healthy, reasons := node.Health() + writeHealthResponse(w, restLogger, healthy, reasons) + }) }) return &s } +// writeHealthResponse writes a minimal health payload with 200/503 semantics +// usable by kubernetes probes and load balancers. +func writeHealthResponse(w http.ResponseWriter, logger *slog.Logger, healthy bool, reasons []string) { + if reasons == nil { + reasons = []string{} + } + resp := struct { + Status string `json:"status"` + Reasons []string `json:"reasons"` + }{ + Status: "UP", + Reasons: reasons, + } + code := http.StatusOK + if !healthy { + resp.Status = "DOWN" + code = http.StatusServiceUnavailable + } + body, err := json.Marshal(resp) + if err != nil { + logger.Error("failed to marshal health response", "error", err) + w.WriteHeader(http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + if _, err := w.Write(body); err != nil { + logger.Error("failed to write health response", "error", err) + } +} + func (s *Server) Start() net.Listener { listener, err := net.Listen("tcp", s.addr) if err != nil { diff --git a/pkg/bpmn/engine.go b/pkg/bpmn/engine.go index a56484a7f..5ca4a4a2f 100644 --- a/pkg/bpmn/engine.go +++ b/pkg/bpmn/engine.go @@ -685,6 +685,9 @@ func (engine *Engine) createTimerStartEventTimers( saved := *timer batch.AddPostFlushAction(ctx, func() { engine.timerManager.registerTimer(saved) + // definition-level timers are saved through a raw storage.Batch and + // bypass EngineBatch.SaveTimer, so record the lifecycle metric here + engine.recordTimerMetric(ctx, saved) }) return nil } diff --git a/pkg/bpmn/engine_api.go b/pkg/bpmn/engine_api.go index 6932a70b2..cdcbe4c42 100644 --- a/pkg/bpmn/engine_api.go +++ b/pkg/bpmn/engine_api.go @@ -322,6 +322,12 @@ mainLoop: engine.metrics.ProcessesEnded.Add(ctx, 1, metric.WithAttributes( attribute.String("bpmn_process_id", instance.ProcessInstance().Definition.BpmnProcessId), )) + if createdAt := instance.ProcessInstance().CreatedAt; !createdAt.IsZero() { + engine.metrics.ProcessInstanceDuration.Record(ctx, float64(time.Since(createdAt))/float64(time.Millisecond), metric.WithAttributes( + attribute.String("bpmn_process_id", instance.ProcessInstance().Definition.BpmnProcessId), + attribute.String("state", instance.ProcessInstance().State.String()), + )) + } } if runErr != nil { diff --git a/pkg/bpmn/engine_batch.go b/pkg/bpmn/engine_batch.go index 66801e916..489eb0406 100644 --- a/pkg/bpmn/engine_batch.go +++ b/pkg/bpmn/engine_batch.go @@ -9,6 +9,8 @@ import ( bpmnruntime "github.com/pbinitiative/zenbpm/pkg/bpmn/runtime" "github.com/pbinitiative/zenbpm/pkg/storage" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" ) type EngineBatch struct { @@ -151,7 +153,14 @@ func (b *EngineBatch) WriteTokenIncident(ctx context.Context, token bpmnruntime. instance.ProcessInstance().State = bpmnruntime.ActivityStateFailed b.b.SaveToken(ctx, token) b.b.SaveProcessInstance(ctx, instance) - b.b.SaveIncident(ctx, createNewIncidentFromToken(err, token, b.engine)) + incident := createNewIncidentFromToken(err, token, b.engine) + if saveErr := b.b.SaveIncident(ctx, incident); saveErr != nil { + b.engine.logger.Error("failed to queue incident for token", "token", token.Key, "err", saveErr) + } else { + b.postFlushActions = append(b.postFlushActions, func() { + b.engine.recordIncidentMetric(ctx, incident) + }) + } } func (b *EngineBatch) WriteMessageIncident(ctx context.Context, message bpmnruntime.MessageSubscription, instance bpmnruntime.ProcessInstance, err error) error { @@ -185,6 +194,9 @@ func (b *EngineBatch) WriteMessageIncident(ctx context.Context, message bpmnrunt if saveErr := b.b.SaveIncident(ctx, incident); saveErr != nil { return fmt.Errorf("failed to save message incident: %w", saveErr) } + b.postFlushActions = append(b.postFlushActions, func() { + b.engine.recordIncidentMetric(ctx, incident) + }) return nil } @@ -205,7 +217,13 @@ func (b *EngineBatch) SaveProcessInstance(ctx context.Context, processInstance b } func (b *EngineBatch) SaveTimer(ctx context.Context, timer bpmnruntime.Timer) error { - return b.b.SaveTimer(ctx, timer) + if err := b.b.SaveTimer(ctx, timer); err != nil { + return err + } + b.postFlushActions = append(b.postFlushActions, func() { + b.engine.recordTimerMetric(ctx, timer) + }) + return nil } func (b *EngineBatch) DeleteProcessDefinitionsTimers(ctx context.Context, processDefinitionKeys []int64) error { @@ -237,7 +255,48 @@ func (b *EngineBatch) UpdateOutputFlowElementInstance(ctx context.Context, histo } func (b *EngineBatch) SaveIncident(ctx context.Context, incident bpmnruntime.Incident) error { - return b.b.SaveIncident(ctx, incident) + if err := b.b.SaveIncident(ctx, incident); err != nil { + return err + } + b.postFlushActions = append(b.postFlushActions, func() { + b.engine.recordIncidentMetric(ctx, incident) + }) + return nil +} + +// recordIncidentMetric increments the incident engine metrics. Incidents with +// ResolvedAt set are counted as resolved, others as created. +func (engine *Engine) recordIncidentMetric(ctx context.Context, incident bpmnruntime.Incident) { + if engine == nil || engine.metrics == nil { + return + } + if incident.ResolvedAt != nil { + engine.metrics.IncidentsResolved.Add(ctx, 1, metric.WithAttributes( + attribute.String("element_id", incident.ElementId), + )) + return + } + engine.metrics.IncidentsCreated.Add(ctx, 1, metric.WithAttributes( + attribute.String("element_id", incident.ElementId), + )) +} + +// recordTimerMetric increments timer engine metrics based on the state the +// timer is being persisted with. It is shared by EngineBatch and the raw +// storage.Batch code paths (definition-level timer start events) so every +// durable timer state transition is counted exactly once. +func (engine *Engine) recordTimerMetric(ctx context.Context, timer bpmnruntime.Timer) { + if engine == nil || engine.metrics == nil { + return + } + switch timer.TimerState { + case bpmnruntime.TimerStateCreated: + engine.metrics.TimersScheduled.Add(ctx, 1) + case bpmnruntime.TimerStateTriggered: + engine.metrics.TimersFired.Add(ctx, 1) + case bpmnruntime.TimerStateCancelled: + engine.metrics.TimersCancelled.Add(ctx, 1) + } } func (b *EngineBatch) SaveErrorSubscription(ctx context.Context, subscription bpmnruntime.ErrorSubscription) error { diff --git a/pkg/bpmn/engine_metrics_test.go b/pkg/bpmn/engine_metrics_test.go new file mode 100644 index 000000000..3b275f249 --- /dev/null +++ b/pkg/bpmn/engine_metrics_test.go @@ -0,0 +1,95 @@ +package bpmn + +import ( + "testing" + "time" + + bpmnruntime "github.com/pbinitiative/zenbpm/pkg/bpmn/runtime" + otelPkg "github.com/pbinitiative/zenbpm/pkg/otel" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" +) + +func TestRecordTimerMetric(t *testing.T) { + engine, reader := newMetricsTestEngine(t) + ctx := t.Context() + + engine.recordTimerMetric(ctx, bpmnruntime.Timer{TimerState: bpmnruntime.TimerStateCreated}) + engine.recordTimerMetric(ctx, bpmnruntime.Timer{TimerState: bpmnruntime.TimerStateCreated}) + engine.recordTimerMetric(ctx, bpmnruntime.Timer{TimerState: bpmnruntime.TimerStateTriggered}) + engine.recordTimerMetric(ctx, bpmnruntime.Timer{TimerState: bpmnruntime.TimerStateCancelled}) + + assert.Equal(t, int64(2), counterValue(t, reader, "timers_scheduled")) + assert.Equal(t, int64(1), counterValue(t, reader, "timers_fired")) + assert.Equal(t, int64(1), counterValue(t, reader, "timers_cancelled")) +} + +func TestRecordIncidentMetric(t *testing.T) { + engine, reader := newMetricsTestEngine(t) + ctx := t.Context() + + engine.recordIncidentMetric(ctx, bpmnruntime.Incident{ElementId: "task-1"}) + engine.recordIncidentMetric(ctx, bpmnruntime.Incident{ElementId: "task-1", ResolvedAt: new(time.Now())}) + + assert.Equal(t, int64(1), counterValue(t, reader, "incidents_created")) + assert.Equal(t, int64(1), counterValue(t, reader, "incidents_resolved")) +} + +func TestRecordMessageCorrelationFailure(t *testing.T) { + engine, reader := newMetricsTestEngine(t) + ctx := t.Context() + + engine.recordMessageCorrelationFailure(ctx, "unknown", "subscription_not_found") + engine.recordMessageCorrelationFailure(ctx, "order-received", "publish_failed") + + assert.Equal(t, int64(2), counterValue(t, reader, "message_correlation_failed")) +} + +func TestMetricRecordersAreNilSafe(t *testing.T) { + // engines without configured metrics (or nil receivers) must not panic + engine := &Engine{} + ctx := t.Context() + assert.NotPanics(t, func() { + engine.recordTimerMetric(ctx, bpmnruntime.Timer{TimerState: bpmnruntime.TimerStateCreated}) + engine.recordIncidentMetric(ctx, bpmnruntime.Incident{}) + engine.recordMessageCorrelationFailure(ctx, "unknown", "subscription_not_found") + }) +} + +// newMetricsTestEngine returns an engine wired to an isolated manual-reader +// meter provider so counter values can be asserted deterministically. +func newMetricsTestEngine(t *testing.T) (*Engine, *sdkmetric.ManualReader) { + t.Helper() + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + t.Cleanup(func() { + _ = provider.Shutdown(t.Context()) + }) + metrics, err := otelPkg.NewMetrics(provider.Meter("bpmn-engine-test")) + require.NoError(t, err) + return &Engine{metrics: metrics}, reader +} + +// counterValue collects the current value of an int64 counter, summing over +// all attribute sets. Returns 0 when the instrument has no data points. +func counterValue(t *testing.T, reader *sdkmetric.ManualReader, name string) int64 { + t.Helper() + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(t.Context(), &rm)) + var total int64 + for _, scope := range rm.ScopeMetrics { + for _, m := range scope.Metrics { + if m.Name != name { + continue + } + sum, ok := m.Data.(metricdata.Sum[int64]) + require.True(t, ok, "metric %s is not an int64 sum", name) + for _, dp := range sum.DataPoints { + total += dp.Value + } + } + } + return total +} diff --git a/pkg/bpmn/events_api.go b/pkg/bpmn/events_api.go index a3b7951e4..dc1f2a2f5 100644 --- a/pkg/bpmn/events_api.go +++ b/pkg/bpmn/events_api.go @@ -8,11 +8,14 @@ import ( "github.com/pbinitiative/zenbpm/pkg/bpmn/model/bpmn20" "github.com/pbinitiative/zenbpm/pkg/bpmn/runtime" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" ) func (engine *Engine) PublishMessageByName(ctx context.Context, name string, correlationKey *string, variables map[string]any) error { message, err := engine.persistence.FindMessageSubscriptionByName(ctx, name, correlationKey, runtime.ActivityStateActive) if err != nil { + engine.recordMessageCorrelationFailure(ctx, "unknown", "subscription_not_found") return errors.Join(newEngineErrorf("failed to find active message subscription with name: %s", name), err) } return engine.PublishMessage(ctx, message, variables) @@ -21,13 +24,39 @@ func (engine *Engine) PublishMessageByName(ctx context.Context, name string, cor func (engine *Engine) PublishMessageByKey(ctx context.Context, subscriptionKey int64, variables map[string]any) error { message, err := engine.persistence.FindMessageSubscriptionByKey(ctx, subscriptionKey, runtime.ActivityStateActive) if err != nil { + engine.recordMessageCorrelationFailure(ctx, "unknown", "subscription_not_found") return errors.Join(newEngineErrorf("failed to find active message subscription: %d", subscriptionKey), err) } return engine.PublishMessage(ctx, message, variables) } +// recordMessageCorrelationFailure increments the failed message correlation +// counter. Only bounded values (names of deployed subscriptions or "unknown") +// may be passed as name to keep metric cardinality under control. +func (engine *Engine) recordMessageCorrelationFailure(ctx context.Context, name string, reason string) { + if engine.metrics == nil { + return + } + engine.metrics.MessageCorrelationFailed.Add(ctx, 1, metric.WithAttributes( + attribute.String("message_name", name), + attribute.String("reason", reason), + )) +} + // PublishMessage publishes a message given by subscription key and also adds variables to the process instance, which fetches this event func (engine *Engine) PublishMessage(ctx context.Context, message runtime.MessageSubscription, variables map[string]interface{}) (retErr error) { + defer func() { + if engine.metrics == nil { + return + } + // bounded cardinality: the name comes from a persisted subscription, i.e. from a deployed process definition + name := message.MessageSubscription().Name + if retErr != nil { + engine.recordMessageCorrelationFailure(ctx, name, "publish_failed") + return + } + engine.metrics.MessagesCorrelated.Add(ctx, 1, metric.WithAttributes(attribute.String("message_name", name))) + }() switch message := message.(type) { case *runtime.DefinitionMessageSubscription: err := engine.publishMessageOnInstanceCreation(ctx, message, variables) diff --git a/pkg/bpmn/jobs_api.go b/pkg/bpmn/jobs_api.go index c630d9364..536a215f5 100644 --- a/pkg/bpmn/jobs_api.go +++ b/pkg/bpmn/jobs_api.go @@ -92,6 +92,7 @@ func (engine *Engine) JobFailByKey(ctx context.Context, jobKey int64, message st attribute.String("type", job.Type), attribute.Bool("internal", false), )) + engine.recordJobLifetime(ctx, job, "failed") } }() @@ -339,6 +340,7 @@ func (engine *Engine) JobCompleteByKey(ctx context.Context, jobKey int64, variab } engine.metrics.JobsCompleted.Add(ctx, 1, metric.WithAttributes(attribute.String("type", job.Type), attribute.Bool("internal", false))) + engine.recordJobLifetime(ctx, job, "completed") return nil } else { err = batch.Flush(ctx) @@ -347,6 +349,7 @@ func (engine *Engine) JobCompleteByKey(ctx context.Context, jobKey int64, variab } engine.metrics.JobsCompleted.Add(ctx, 1, metric.WithAttributes(attribute.String("type", job.Type), attribute.Bool("internal", false))) + engine.recordJobLifetime(ctx, job, "completed") if !messageEndEventHandled { err := engine.RunProcessInstance(ctx, instance, tokens) @@ -375,3 +378,14 @@ func (engine *Engine) refreshAndValidateJob(ctx context.Context, jobKey int64) ( } return job, nil } + +// recordJobLifetime records the time between job creation and its terminal state, in milliseconds. +func (engine *Engine) recordJobLifetime(ctx context.Context, job runtime.Job, outcome string) { + if engine.metrics == nil || job.CreatedAt.IsZero() { + return + } + engine.metrics.JobLifetime.Record(ctx, float64(time.Since(job.CreatedAt))/float64(time.Millisecond), metric.WithAttributes( + attribute.String("type", job.Type), + attribute.String("outcome", outcome), + )) +} diff --git a/pkg/bpmn/start_event_instance_creation_handler.go b/pkg/bpmn/start_event_instance_creation_handler.go index f41ce6d6e..2171d4879 100644 --- a/pkg/bpmn/start_event_instance_creation_handler.go +++ b/pkg/bpmn/start_event_instance_creation_handler.go @@ -272,6 +272,10 @@ func (engine *Engine) processTimerTriggerOnInstanceCreation(ctx context.Context, if err := batch.SaveTimer(ctx, current); err != nil { return fmt.Errorf("failed to update timer state for timer %d: %w", current.Key, err) } + consumed := current + batch.AddPostFlushAction(ctx, func() { + engine.recordTimerMetric(ctx, consumed) + }) return nil }, }) diff --git a/pkg/dmn/dmn_engine.go b/pkg/dmn/dmn_engine.go index 7c1544ed1..538983ef5 100644 --- a/pkg/dmn/dmn_engine.go +++ b/pkg/dmn/dmn_engine.go @@ -13,18 +13,31 @@ import ( "time" "github.com/dop251/goja" + "github.com/hashicorp/go-hclog" "github.com/pbinitiative/zenbpm/pkg/dmn/model/dmn" "github.com/pbinitiative/zenbpm/pkg/dmn/runtime" + otelPkg "github.com/pbinitiative/zenbpm/pkg/otel" "github.com/pbinitiative/zenbpm/pkg/script" "github.com/pbinitiative/zenbpm/pkg/script/feel" "github.com/pbinitiative/zenbpm/pkg/storage" "github.com/pbinitiative/zenbpm/pkg/storage/inmemory" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/metric" + "go.opentelemetry.io/otel/trace" ) +const dmnEngineName = "dmn-engine" + type ZenDmnEngine struct { persistence storage.DecisionStorage feelRuntime script.FeelRuntime ownsFeelRuntime bool + + tracer trace.Tracer + evaluationsTotal metric.Int64Counter + evaluationDuration metric.Float64Histogram } type EngineOption = func(*ZenDmnEngine) @@ -33,6 +46,20 @@ type EngineOption = func(*ZenDmnEngine) func NewEngine(options ...EngineOption) *ZenDmnEngine { engine := ZenDmnEngine{ persistence: inmemory.NewStorage(), + tracer: otel.GetTracerProvider().Tracer(dmnEngineName), + } + meter := otel.GetMeterProvider().Meter(dmnEngineName) + var err error + engine.evaluationsTotal, err = meter.Int64Counter("dmn_evaluations", metric.WithDescription("Number of DMN decision evaluations")) + if err != nil { + hclog.Default().Named(dmnEngineName).Error("Failed to create dmn_evaluations_total instrument", "err", err) + } + engine.evaluationDuration, err = meter.Float64Histogram("dmn_evaluation_duration", + metric.WithUnit("ms"), + metric.WithDescription("Duration of DMN decision evaluations, milliseconds"), + ) + if err != nil { + hclog.Default().Named(dmnEngineName).Error("Failed to create dmn_evaluation_duration instrument", "err", err) } for _, option := range options { @@ -279,7 +306,36 @@ func (engine *ZenDmnEngine) evaluateDRD( dmnResourceDefinition *runtime.DmnResourceDefinition, decisionDefinition *runtime.DecisionDefinition, inputVariableContext map[string]interface{}, -) (*EvaluatedDRDResult, error) { +) (_ *EvaluatedDRDResult, retErr error) { + start := time.Now() + // stable span name keeps the tracing backend operation index bounded; + // the decision id is carried by the zenbpm.decision.id attribute + ctx, evalSpan := engine.tracer.Start(ctx, "dmn.evaluate-decision", trace.WithAttributes( + attribute.String(otelPkg.AttributeDecisionId, decisionDefinition.Id), + attribute.Int64(otelPkg.AttributeDecisionKey, decisionDefinition.Key), + attribute.String(otelPkg.AttributeDrdId, dmnResourceDefinition.Id), + )) + defer func() { + outcome := "success" + if retErr != nil { + outcome = "error" + evalSpan.RecordError(retErr) + evalSpan.SetStatus(codes.Error, retErr.Error()) + } + if engine.evaluationsTotal != nil { + engine.evaluationsTotal.Add(ctx, 1, metric.WithAttributes( + attribute.String("decision_id", decisionDefinition.Id), + attribute.String("outcome", outcome), + )) + } + if engine.evaluationDuration != nil { + engine.evaluationDuration.Record(ctx, float64(time.Since(start))/float64(time.Millisecond), metric.WithAttributes( + attribute.String("decision_id", decisionDefinition.Id), + )) + } + evalSpan.End() + }() + result, dependencies, err := engine.evaluateDecision(ctx, dmnResourceDefinition, decisionDefinition.Id, inputVariableContext) if err != nil { return nil, fmt.Errorf("failed to evaluate DecisionDefinition %s in DmnResourceDefinition %s:%d: %w", diff --git a/pkg/dmn/dmn_multiline_test.go b/pkg/dmn/dmn_multiline_test.go index 91de91345..b45621181 100644 --- a/pkg/dmn/dmn_multiline_test.go +++ b/pkg/dmn/dmn_multiline_test.go @@ -125,12 +125,12 @@ func TestLiteralExpressionMultilineQuotedString(t *testing.T) { definition, xmldata, err := dmnEngine.ParseDmnFromFile(filepath.Join(".", "test-data", "multiline", "literal-expression.dmn")) assert.NoError(t, err) - metadata, decisions, err := dmnEngine.SaveDmnResourceDefinition(nil, definition, xmldata, dmnEngine.generateKey()) + metadata, decisions, err := dmnEngine.SaveDmnResourceDefinition(t.Context(), definition, xmldata, dmnEngine.generateKey()) assert.NoError(t, err) assert.NotNil(t, metadata) assert.Len(t, decisions, 1) - result, err := dmnEngine.FindAndEvaluateDRD(nil, "latest", metadata.Id+"."+decisions[0].Id, "", nil) + result, err := dmnEngine.FindAndEvaluateDRD(t.Context(), "latest", metadata.Id+"."+decisions[0].Id, "", nil) assert.NoError(t, err) assert.Equal(t, "Hello,\nWorld!", result.DecisionOutput) } @@ -141,13 +141,13 @@ func TestDecisionTableMultilineOutput(t *testing.T) { definition, xmldata, err := dmnEngine.ParseDmnFromFile(filepath.Join(".", "test-data", "multiline", "decision-table-output.dmn")) assert.NoError(t, err) - metadata, decisions, err := dmnEngine.SaveDmnResourceDefinition(nil, definition, xmldata, dmnEngine.generateKey()) + metadata, decisions, err := dmnEngine.SaveDmnResourceDefinition(t.Context(), definition, xmldata, dmnEngine.generateKey()) assert.NoError(t, err) assert.NotNil(t, metadata) assert.Len(t, decisions, 1) input := map[string]interface{}{"code": "A1"} - result, err := dmnEngine.FindAndEvaluateDRD(nil, "latest", metadata.Id+"."+decisions[0].Id, "", input) + result, err := dmnEngine.FindAndEvaluateDRD(t.Context(), "latest", metadata.Id+"."+decisions[0].Id, "", input) assert.NoError(t, err) output, ok := result.DecisionOutput.(map[string]interface{}) assert.True(t, ok) diff --git a/pkg/otel/metrics.go b/pkg/otel/metrics.go index a0b87d31c..3ce573f38 100644 --- a/pkg/otel/metrics.go +++ b/pkg/otel/metrics.go @@ -6,6 +6,24 @@ import ( "go.opentelemetry.io/otel/metric" ) +// latencyBucketsMs are histogram bucket boundaries (in milliseconds) suited to +// BPMN workloads: from sub-millisecond engine operations up to hour-long +// process instances. Kept package-private so callers cannot mutate the shared +// definition; use LatencyBucketsMs to obtain a copy. +var latencyBucketsMs = []float64{ + 1, 5, 10, 25, 50, 100, 250, 500, + 1_000, 2_500, 5_000, 10_000, 30_000, 60_000, + 300_000, 900_000, 3_600_000, +} + +// LatencyBucketsMs returns a fresh copy of the shared latency histogram bucket +// boundaries (in milliseconds) so that callers cannot mutate the canonical set. +func LatencyBucketsMs() []float64 { + buckets := make([]float64, len(latencyBucketsMs)) + copy(buckets, latencyBucketsMs) + return buckets +} + type EngineMetrics struct { ProcessesStarted metric.Int64Counter ProcessesEnded metric.Int64Counter @@ -13,6 +31,23 @@ type EngineMetrics struct { JobsCreated metric.Int64Counter JobsCompleted metric.Int64Counter JobsFailed metric.Int64Counter + + // IncidentsCreated counts incidents raised by the engine. + IncidentsCreated metric.Int64Counter + // IncidentsResolved counts incidents that were resolved. + IncidentsResolved metric.Int64Counter + + // ProcessInstanceDuration measures time from instance creation to completion/failure, in ms. + ProcessInstanceDuration metric.Float64Histogram + // JobLifetime measures time from job creation to completion/failure, in ms. + JobLifetime metric.Float64Histogram + + TimersScheduled metric.Int64Counter + TimersFired metric.Int64Counter + TimersCancelled metric.Int64Counter + + MessagesCorrelated metric.Int64Counter + MessageCorrelationFailed metric.Int64Counter } func NewMetrics(meter metric.Meter) (*EngineMetrics, error) { @@ -36,13 +71,57 @@ func NewMetrics(meter metric.Meter) (*EngineMetrics, error) { jobsFailed, err := meter.Int64Counter("jobs_failed", metric.WithDescription("Number of jobs failed")) errJoin = errors.Join(errJoin, err) + incidentsCreated, err := meter.Int64Counter("incidents_created", metric.WithDescription("Number of incidents created")) + errJoin = errors.Join(errJoin, err) + + incidentsResolved, err := meter.Int64Counter("incidents_resolved", metric.WithDescription("Number of incidents resolved")) + errJoin = errors.Join(errJoin, err) + + processInstanceDuration, err := meter.Float64Histogram("process_instance_duration", + metric.WithUnit("ms"), + metric.WithDescription("Time from process instance creation to completion or failure, milliseconds"), + metric.WithExplicitBucketBoundaries(latencyBucketsMs...), + ) + errJoin = errors.Join(errJoin, err) + + jobLifetime, err := meter.Float64Histogram("job_lifetime", + metric.WithUnit("ms"), + metric.WithDescription("Time from job creation to completion or failure, milliseconds"), + metric.WithExplicitBucketBoundaries(latencyBucketsMs...), + ) + errJoin = errors.Join(errJoin, err) + + timersScheduled, err := meter.Int64Counter("timers_scheduled", metric.WithDescription("Number of timers scheduled")) + errJoin = errors.Join(errJoin, err) + + timersFired, err := meter.Int64Counter("timers_fired", metric.WithDescription("Number of timers fired")) + errJoin = errors.Join(errJoin, err) + + timersCancelled, err := meter.Int64Counter("timers_cancelled", metric.WithDescription("Number of timers cancelled")) + errJoin = errors.Join(errJoin, err) + + messagesCorrelated, err := meter.Int64Counter("messages_correlated", metric.WithDescription("Number of messages successfully correlated to subscriptions")) + errJoin = errors.Join(errJoin, err) + + messageCorrelationFailed, err := meter.Int64Counter("message_correlation_failed", metric.WithDescription("Number of failed message correlations")) + errJoin = errors.Join(errJoin, err) + metrics := EngineMetrics{ - ProcessesStarted: processesStartedTotal, - ProcessesEnded: processesCompletedTotal, - ProcessesRunning: processesRunning, - JobsCreated: jobsCreated, - JobsCompleted: jobsCompleted, - JobsFailed: jobsFailed, + ProcessesStarted: processesStartedTotal, + ProcessesEnded: processesCompletedTotal, + ProcessesRunning: processesRunning, + JobsCreated: jobsCreated, + JobsCompleted: jobsCompleted, + JobsFailed: jobsFailed, + IncidentsCreated: incidentsCreated, + IncidentsResolved: incidentsResolved, + ProcessInstanceDuration: processInstanceDuration, + JobLifetime: jobLifetime, + TimersScheduled: timersScheduled, + TimersFired: timersFired, + TimersCancelled: timersCancelled, + MessagesCorrelated: messagesCorrelated, + MessageCorrelationFailed: messageCorrelationFailed, } return &metrics, errJoin } diff --git a/pkg/otel/traces.go b/pkg/otel/traces.go index f48cbcc10..c501a5547 100644 --- a/pkg/otel/traces.go +++ b/pkg/otel/traces.go @@ -1,21 +1,28 @@ package otel const ( - Prefix = "bpmn-" - AttributeProcessInstanceKey = Prefix + "instance-key" - AttributeProcessId = Prefix + "process-id" + // Prefix namespaces all ZenBPM specific span attributes following the + // OpenTelemetry attribute naming conventions (dot separated namespaces). + Prefix = "zenbpm." + AttributeProcessInstanceKey = Prefix + "process.instance_key" + AttributeProcessId = Prefix + "process.id" - AttributeProcessDefinitionKey = Prefix + "definition-key" + AttributeProcessDefinitionKey = Prefix + "process.definition_key" - AttributeToken = Prefix + "token-key" + AttributeToken = Prefix + "token.key" - AttributeElementId = Prefix + "element-id" - AttributeElementKey = Prefix + "element-key" - AttributeElementName = Prefix + "element-name" - AttributeElementType = Prefix + "element-type" + AttributeElementId = Prefix + "element.id" + AttributeElementKey = Prefix + "element.key" + AttributeElementName = Prefix + "element.name" + AttributeElementType = Prefix + "element.type" - AttributeJobKey = Prefix + "job-key" - AttributeIncidentKey = Prefix + "incident-key" + AttributeJobKey = Prefix + "job.key" + AttributeIncidentKey = Prefix + "incident.key" - SpanStatusToken = Prefix + "token-status" + AttributeDecisionId = Prefix + "decision.id" + AttributeDecisionKey = Prefix + "decision.key" + AttributeDecisionInstanceKey = Prefix + "decision.instance_key" + AttributeDrdId = Prefix + "decision.drd_id" + + SpanStatusToken = Prefix + "token.status" ) diff --git a/scripts/alertmanager.yml b/scripts/alertmanager.yml new file mode 100644 index 000000000..177be0e12 --- /dev/null +++ b/scripts/alertmanager.yml @@ -0,0 +1,26 @@ +# Alertmanager configuration for the local ZenBPM monitoring stack. +# +# The default receiver is a "blackhole" (no notifiers), so the local stack +# does not continuously log delivery failures. Alerts remain visible in the +# Alertmanager UI (port 9093). To actually deliver alerts, uncomment the +# webhook below (and start a listener on that port) or replace it with your +# real notification channel (Slack, PagerDuty, e-mail, ...) in production. + +route: + receiver: default + group_by: ['alertname', 'severity'] + group_wait: 30s + group_interval: 5m + repeat_interval: 4h + routes: + - receiver: default + matchers: + - severity = "critical" + repeat_interval: 1h + +receivers: + - name: default + # webhook_configs: + # # Point this at your alert gateway and uncomment to enable delivery. + # - url: http://host.docker.internal:9099/alerts + # send_resolved: true diff --git a/scripts/grafana_provisioning/dashboards/zenbpm/cluster.json b/scripts/grafana_provisioning/dashboards/zenbpm/cluster.json new file mode 100644 index 000000000..0b4658810 --- /dev/null +++ b/scripts/grafana_provisioning/dashboards/zenbpm/cluster.json @@ -0,0 +1,116 @@ +{ + "annotations": { "list": [] }, + "editable": true, + "graphTooltip": 1, + "id": null, + "links": [], + "panels": [ + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { + "defaults": { + "mappings": [ + { "options": { "0": { "color": "red", "index": 1, "text": "NO LEADER" }, "1": { "color": "green", "index": 0, "text": "LEADER OK" } }, "type": "value" } + ], + "thresholds": { "mode": "absolute", "steps": [ { "color": "red", "value": null }, { "color": "green", "value": 1 } ] } + }, + "overrides": [] + }, + "gridPos": { "h": 6, "w": 6, "x": 0, "y": 0 }, + "id": 1, + "options": { "colorMode": "background", "graphMode": "none", "reduceOptions": { "calcs": ["lastNotNull"] } }, + "targets": [ { "expr": "max(cluster_has_leader)", "legendFormat": "cluster leader", "refId": "A" } ], + "title": "Main cluster leader", + "type": "stat" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { + "defaults": { + "mappings": [ + { "options": { "0": { "color": "red", "index": 1, "text": "NO LEADER" }, "1": { "color": "green", "index": 0, "text": "OK" } }, "type": "value" } + ], + "thresholds": { "mode": "absolute", "steps": [ { "color": "red", "value": null }, { "color": "green", "value": 1 } ] } + }, + "overrides": [] + }, + "gridPos": { "h": 6, "w": 6, "x": 6, "y": 0 }, + "id": 2, + "options": { "colorMode": "background", "graphMode": "none", "reduceOptions": { "calcs": ["lastNotNull"] } }, + "targets": [ { "expr": "min by(partition) (partition_has_leader)", "legendFormat": "partition {{partition}}", "refId": "A" } ], + "title": "Partition leaders", + "type": "stat" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 } }, "overrides": [] }, + "gridPos": { "h": 6, "w": 12, "x": 12, "y": 0 }, + "id": 3, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "sum by(partition) (increase(partition_leader_changes_total[5m]))", "legendFormat": "partition {{partition}}", "refId": "A" } + ], + "title": "Leader changes (5m)", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 } }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 6 }, + "id": 4, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "raft_last_log_index - raft_applied_index", "legendFormat": "{{instance}} apply lag", "refId": "A" }, + { "expr": "raft_fsm_pending", "legendFormat": "{{instance}} fsm pending", "refId": "B" } + ], + "title": "Raft log apply lag / FSM backlog", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 0 } }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 6 }, + "id": 5, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "raft_term", "legendFormat": "{{instance}} term", "refId": "A" } + ], + "title": "Raft term", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 0 } }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 14 }, + "id": 6, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "node_is_leader", "legendFormat": "{{instance}} main leader", "refId": "A" }, + { "expr": "partition_node_is_leader", "legendFormat": "{{instance}} partition {{partition}} leader", "refId": "B" } + ], + "title": "Leadership per node", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 } }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 14 }, + "id": 7, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "up{job=\"zenbpm\"}", "legendFormat": "{{instance}} up", "refId": "A" } + ], + "title": "Scrape targets up", + "type": "timeseries" + } + ], + "refresh": "10s", + "schemaVersion": 39, + "tags": ["zenbpm", "cluster"], + "templating": { "list": [] }, + "time": { "from": "now-1h", "to": "now" }, + "timezone": "browser", + "title": "ZenBPM Cluster Health", + "uid": "zenbpm-cluster", + "version": 1 +} diff --git a/scripts/grafana_provisioning/dashboards/zenbpm/host.json b/scripts/grafana_provisioning/dashboards/zenbpm/host.json new file mode 100644 index 000000000..559b7ac65 --- /dev/null +++ b/scripts/grafana_provisioning/dashboards/zenbpm/host.json @@ -0,0 +1,92 @@ +{ + "annotations": { "list": [] }, + "editable": true, + "graphTooltip": 1, + "id": null, + "links": [], + "panels": [ + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "percentunit", "max": 1 }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 0 }, + "id": 1, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "1 - avg by(instance) (rate(node_cpu_seconds_total{mode=\"idle\"}[$__rate_interval]))", "legendFormat": "{{instance}} cpu", "refId": "A" } + ], + "title": "CPU usage", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "percentunit", "max": 1 }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 0 }, + "id": 2, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "1 - node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes", "legendFormat": "{{instance}} memory", "refId": "A" } + ], + "title": "Memory usage", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "percentunit", "max": 1 }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 8 }, + "id": 3, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "1 - node_filesystem_avail_bytes{fstype!~\"tmpfs|overlay\"} / node_filesystem_size_bytes{fstype!~\"tmpfs|overlay\"}", "legendFormat": "{{instance}} {{mountpoint}}", "refId": "A" } + ], + "title": "Disk usage", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "binBps" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 8 }, + "id": 4, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "rate(node_disk_read_bytes_total[$__rate_interval])", "legendFormat": "{{instance}} {{device}} read", "refId": "A" }, + { "expr": "rate(node_disk_written_bytes_total[$__rate_interval])", "legendFormat": "{{instance}} {{device}} write", "refId": "B" } + ], + "title": "Disk I/O", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "percentunit" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 16 }, + "id": 5, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "rate(node_disk_io_time_seconds_total[$__rate_interval])", "legendFormat": "{{instance}} {{device}} io busy", "refId": "A" } + ], + "title": "Disk saturation (io time)", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "binBps" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 16 }, + "id": 6, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "rate(node_network_receive_bytes_total{device!=\"lo\"}[$__rate_interval])", "legendFormat": "{{instance}} {{device}} rx", "refId": "A" }, + { "expr": "rate(node_network_transmit_bytes_total{device!=\"lo\"}[$__rate_interval])", "legendFormat": "{{instance}} {{device}} tx", "refId": "B" } + ], + "title": "Network traffic", + "type": "timeseries" + } + ], + "refresh": "30s", + "schemaVersion": 39, + "tags": ["zenbpm", "host"], + "templating": { "list": [] }, + "time": { "from": "now-3h", "to": "now" }, + "timezone": "browser", + "title": "ZenBPM Host (node_exporter)", + "uid": "zenbpm-host", + "version": 1 +} diff --git a/scripts/grafana_provisioning/dashboards/zenbpm/incidents.json b/scripts/grafana_provisioning/dashboards/zenbpm/incidents.json new file mode 100644 index 000000000..db0ad1a27 --- /dev/null +++ b/scripts/grafana_provisioning/dashboards/zenbpm/incidents.json @@ -0,0 +1,122 @@ +{ + "annotations": { "list": [] }, + "editable": true, + "graphTooltip": 1, + "id": null, + "links": [], + "panels": [ + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 } }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 0 }, + "id": 1, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "sum by(element_id) (rate(incidents_created_total[$__rate_interval]))", "legendFormat": "created {{element_id}}", "refId": "A" }, + { "expr": "sum(rate(incidents_resolved_total[$__rate_interval]))", "legendFormat": "resolved", "refId": "B" } + ], + "title": "Incidents created / resolved", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { + "defaults": { "thresholds": { "mode": "absolute", "steps": [ { "color": "green", "value": null }, { "color": "red", "value": 1 } ] } }, + "overrides": [] + }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 0 }, + "id": 2, + "options": { "colorMode": "background", "graphMode": "area", "reduceOptions": { "calcs": ["lastNotNull"] } }, + "targets": [ + { "expr": "sum(increase(incidents_created_total[24h])) - sum(increase(incidents_resolved_total[24h]))", "legendFormat": "net change (24h window)", "refId": "A" } + ], + "title": "Net incident change (24h)", + "type": "stat" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "percentunit" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 8 }, + "id": 3, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "sum(rate(jobs_failed_total[$__rate_interval])) / sum(rate(jobs_created_total[$__rate_interval]))", "legendFormat": "failure ratio", "refId": "A" } + ], + "title": "Job failure ratio", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 } }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 8 }, + "id": 4, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "topk(10, sum by(type) (rate(jobs_failed_total[$__rate_interval])))", "legendFormat": "{{type}}", "refId": "A" } + ], + "title": "Top failing job types", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 } }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 16 }, + "id": 5, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "sum by(message_name) (rate(message_correlation_failed_total[$__rate_interval]))", "legendFormat": "failed {{message_name}}", "refId": "A" }, + { "expr": "sum(rate(messages_correlated_total[$__rate_interval]))", "legendFormat": "correlated", "refId": "B" } + ], + "title": "Message correlation", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 } }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 16 }, + "id": 6, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "sum(rate(timers_scheduled_total[$__rate_interval]))", "legendFormat": "scheduled", "refId": "A" }, + { "expr": "sum(rate(timers_fired_total[$__rate_interval]))", "legendFormat": "fired", "refId": "B" }, + { "expr": "sum(rate(timers_cancelled_total[$__rate_interval]))", "legendFormat": "cancelled", "refId": "C" } + ], + "title": "Timers", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 } }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 24 }, + "id": 7, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "sum by(decision_id, outcome) (rate(dmn_evaluations_total[$__rate_interval]))", "legendFormat": "{{decision_id}} {{outcome}}", "refId": "A" } + ], + "title": "DMN evaluations", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "ms" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 24 }, + "id": 8, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "histogram_quantile(0.95, sum by(le) (rate(dmn_evaluation_duration_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p95", "refId": "A" }, + { "expr": "histogram_quantile(0.99, sum by(le) (rate(dmn_evaluation_duration_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p99", "refId": "B" } + ], + "title": "DMN evaluation duration", + "type": "timeseries" + } + ], + "refresh": "10s", + "schemaVersion": 39, + "tags": ["zenbpm", "incidents"], + "templating": { "list": [] }, + "time": { "from": "now-3h", "to": "now" }, + "timezone": "browser", + "title": "ZenBPM Incidents & Errors", + "uid": "zenbpm-incidents", + "version": 1 +} diff --git a/scripts/grafana_provisioning/dashboards/zenbpm/latency.json b/scripts/grafana_provisioning/dashboards/zenbpm/latency.json new file mode 100644 index 000000000..534283b3d --- /dev/null +++ b/scripts/grafana_provisioning/dashboards/zenbpm/latency.json @@ -0,0 +1,99 @@ +{ + "annotations": { "list": [] }, + "editable": true, + "graphTooltip": 1, + "id": null, + "links": [], + "panels": [ + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "ms" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 0 }, + "id": 1, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "histogram_quantile(0.50, sum by(le) (rate(process_instance_duration_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p50", "refId": "A" }, + { "expr": "histogram_quantile(0.95, sum by(le) (rate(process_instance_duration_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p95", "refId": "B" }, + { "expr": "histogram_quantile(0.99, sum by(le) (rate(process_instance_duration_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p99", "refId": "C" } + ], + "title": "Process instance duration percentiles", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "ms" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 0 }, + "id": 2, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "histogram_quantile(0.50, sum by(le) (rate(job_lifetime_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p50", "refId": "A" }, + { "expr": "histogram_quantile(0.95, sum by(le) (rate(job_lifetime_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p95", "refId": "B" }, + { "expr": "histogram_quantile(0.99, sum by(le) (rate(job_lifetime_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p99", "refId": "C" } + ], + "title": "Job lifetime percentiles", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "ms" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 8 }, + "id": 3, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "histogram_quantile(0.95, sum by(le, type) (rate(job_activation_latency_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p95 {{type}}", "refId": "A" } + ], + "title": "Job activation latency (created → distributed)", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "ms" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 8 }, + "id": 4, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "histogram_quantile(0.50, sum by(le) (rate(request_duration_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p50", "refId": "A" }, + { "expr": "histogram_quantile(0.95, sum by(le) (rate(request_duration_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p95", "refId": "B" }, + { "expr": "histogram_quantile(0.99, sum by(le) (rate(request_duration_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p99", "refId": "C" } + ], + "title": "REST request duration percentiles", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 } }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 16 }, + "id": 5, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "sum(rate(processes_started_total[$__rate_interval]))", "legendFormat": "started/s", "refId": "A" }, + { "expr": "sum(rate(processes_completed_total[$__rate_interval]))", "legendFormat": "completed/s", "refId": "B" } + ], + "title": "Process throughput", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 } }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 16 }, + "id": 6, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "sum(rate(jobs_created_total[$__rate_interval]))", "legendFormat": "created/s", "refId": "A" }, + { "expr": "sum(rate(jobs_completed_total[$__rate_interval]))", "legendFormat": "completed/s", "refId": "B" }, + { "expr": "sum(rate(jobs_distributed_total[$__rate_interval]))", "legendFormat": "distributed/s", "refId": "C" } + ], + "title": "Job throughput", + "type": "timeseries" + } + ], + "refresh": "10s", + "schemaVersion": 39, + "tags": ["zenbpm", "latency"], + "templating": { "list": [] }, + "time": { "from": "now-3h", "to": "now" }, + "timezone": "browser", + "title": "ZenBPM Latency & Throughput", + "uid": "zenbpm-latency", + "version": 1 +} diff --git a/scripts/grafana_provisioning/dashboards/zenbpm/storage.json b/scripts/grafana_provisioning/dashboards/zenbpm/storage.json new file mode 100644 index 000000000..ba1daee4e --- /dev/null +++ b/scripts/grafana_provisioning/dashboards/zenbpm/storage.json @@ -0,0 +1,95 @@ +{ + "annotations": { "list": [] }, + "editable": true, + "graphTooltip": 1, + "id": null, + "links": [], + "panels": [ + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "bytes" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 0 }, + "id": 1, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "rqlite_db_size_bytes", "legendFormat": "partition {{partition}} {{instance}}", "refId": "A" } + ], + "title": "rqlite DB size", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "binBps" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 0 }, + "id": 2, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "deriv(rqlite_db_size_bytes[30m])", "legendFormat": "partition {{partition}} growth", "refId": "A" } + ], + "title": "rqlite DB growth rate (30m)", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "ms" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 8 }, + "id": 3, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "histogram_quantile(0.50, sum by(le, partition) (rate(rqlite_exec_duration_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p50 partition {{partition}}", "refId": "A" }, + { "expr": "histogram_quantile(0.95, sum by(le, partition) (rate(rqlite_exec_duration_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p95 partition {{partition}}", "refId": "B" }, + { "expr": "histogram_quantile(0.99, sum by(le, partition) (rate(rqlite_exec_duration_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p99 partition {{partition}}", "refId": "C" } + ], + "title": "rqlite write latency percentiles", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "ms" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 8 }, + "id": 4, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "histogram_quantile(0.50, sum by(le, partition) (rate(rqlite_query_duration_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p50 partition {{partition}}", "refId": "A" }, + { "expr": "histogram_quantile(0.95, sum by(le, partition) (rate(rqlite_query_duration_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p95 partition {{partition}}", "refId": "B" }, + { "expr": "histogram_quantile(0.99, sum by(le, partition) (rate(rqlite_query_duration_milliseconds_bucket[$__rate_interval])))", "legendFormat": "p99 partition {{partition}}", "refId": "C" } + ], + "title": "rqlite read latency percentiles", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 } }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 16 }, + "id": 5, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "sum by(partition) (rate(rqlite_exec_duration_milliseconds_count[$__rate_interval]))", "legendFormat": "writes/s partition {{partition}}", "refId": "A" }, + { "expr": "sum by(partition) (rate(rqlite_query_duration_milliseconds_count[$__rate_interval]))", "legendFormat": "reads/s partition {{partition}}", "refId": "B" } + ], + "title": "rqlite operation throughput", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 }, "unit": "bytes" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 16 }, + "id": 6, + "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, + "targets": [ + { "expr": "node_filesystem_avail_bytes{fstype!~\"tmpfs|overlay\"}", "legendFormat": "{{instance}} {{mountpoint}} free", "refId": "A" } + ], + "title": "Disk free (node_exporter)", + "type": "timeseries" + } + ], + "refresh": "30s", + "schemaVersion": 39, + "tags": ["zenbpm", "storage"], + "templating": { "list": [] }, + "time": { "from": "now-6h", "to": "now" }, + "timezone": "browser", + "title": "ZenBPM Storage / rqlite", + "uid": "zenbpm-storage", + "version": 1 +} diff --git a/scripts/prometheus-rules.yml b/scripts/prometheus-rules.yml new file mode 100644 index 000000000..81c98c294 --- /dev/null +++ b/scripts/prometheus-rules.yml @@ -0,0 +1,266 @@ +# Prometheus alerting rules for ZenBPM. +# Mounted into the Prometheus container by `make start-monitoring` +# (see Makefile) and referenced from scripts/prometheus.yml. +# +# Metric name notes (OpenTelemetry -> Prometheus exporter): +# counters get a `_total` suffix, `ms` histograms get `_milliseconds_bucket`. + +groups: + # ────────────────────────────────────────────────────────────────────── + # Technical alerts + # ────────────────────────────────────────────────────────────────────── + - name: zenbpm-technical + rules: + # A-T1: raft leadership + - alert: NoClusterLeader + expr: max(cluster_has_leader) == 0 + for: 30s + labels: + severity: critical + annotations: + summary: "ZenBPM main cluster has no elected raft leader" + description: "No node reports an elected leader for the main Zen cluster for more than 30s. Writes and cluster coordination are unavailable." + + - alert: NoPartitionLeader + expr: max by(partition) (partition_has_leader) == 0 + for: 30s + labels: + severity: critical + annotations: + summary: "ZenBPM partition {{ $labels.partition }} has no leader" + description: "Partition {{ $labels.partition }} has no leader registered in the cluster state for more than 30s. Process execution on this partition is stalled." + + # partition_has_leader only emits series for partitions that already exist + # in the cluster state, so NoPartitionLeader alone is blind to partitions + # that were never created (bootstrap/controller failure). This alert covers + # that gap using the explicit count gauges exported by every node. + - alert: PartitionDeficit + expr: max(cluster_partitions) < max(cluster_desired_partitions) + for: 1m + labels: + severity: critical + annotations: + summary: "ZenBPM cluster has fewer partitions than desired" + description: "The cluster state contains {{ with query \"max(cluster_partitions)\" }}{{ . | first | value }}{{ end }} of the desired {{ with query \"max(cluster_desired_partitions)\" }}{{ . | first | value }}{{ end }} partitions. Work routed to missing partitions cannot execute." + + # A-T2: scrape target down + - alert: TargetDown + expr: up{job="zenbpm"} == 0 + for: 1m + labels: + severity: critical + annotations: + summary: "ZenBPM instance {{ $labels.instance }} is down" + description: "Prometheus has been unable to scrape {{ $labels.instance }} for more than 1 minute." + + # A-T3: REST error rate + # NOTE: the REST middleware exports the HTTP status under the `status` + # label (internal/rest/middleware/otel.go). clamp_min protects the ratio + # from a zero denominator when there is no traffic. + - alert: HighErrorRate + expr: | + ( + sum(rate(request_uri_total{status=~"5.."}[5m])) + / + clamp_min(sum(rate(request_uri_total[5m])), 1e-9) + ) > 0.05 + for: 5m + labels: + severity: critical + annotations: + summary: "ZenBPM REST 5xx error ratio above 5%" + description: "More than 5% of REST requests returned a 5xx status over the last 5 minutes." + + # A-T4: latency degradation + - alert: RestLatencyDegradation + expr: | + histogram_quantile(0.99, + sum by(le) (rate(request_duration_milliseconds_bucket[5m]))) > 1000 + for: 10m + labels: + severity: warning + annotations: + summary: "ZenBPM REST p99 latency above 1s" + description: "The p99 REST request duration has been above 1s for 10 minutes." + + - alert: RqliteExecLatencyDegradation + expr: | + histogram_quantile(0.99, + sum by(le, partition) (rate(rqlite_exec_duration_milliseconds_bucket[5m]))) > 500 + for: 10m + labels: + severity: warning + annotations: + summary: "rqlite write p99 latency above 500ms on partition {{ $labels.partition }}" + description: "Raft-replicated writes are slow; check disk I/O, network between partition nodes and write batching." + + # A-T5: rqlite db size + - alert: RqliteDbSizeLarge + expr: rqlite_db_size_bytes > 5e9 + for: 15m + labels: + severity: warning + annotations: + summary: "rqlite database of partition {{ $labels.partition }} exceeds 5GB" + description: "Check instance history TTL / data cleanup configuration before the disk fills up." + + - alert: RqliteDbGrowthPrediction + expr: predict_linear(rqlite_db_size_bytes[6h], 24 * 3600) > 10e9 + for: 30m + labels: + severity: warning + annotations: + summary: "rqlite database of partition {{ $labels.partition }} predicted to exceed 10GB within 24h" + description: "Current growth rate would exceed 10GB within a day. Investigate runaway process instances or missing history cleanup." + + # A-T6: disk space + - alert: DiskSpaceLow + expr: | + node_filesystem_avail_bytes{fstype!~"tmpfs|overlay"} + / + node_filesystem_size_bytes{fstype!~"tmpfs|overlay"} < 0.10 + for: 5m + labels: + severity: critical + annotations: + summary: "Less than 10% disk space left on {{ $labels.instance }} ({{ $labels.mountpoint }})" + description: "Raft log, snapshots and the rqlite database will fail once the disk is full." + + # A-T7: CPU / memory + - alert: HighCPU + expr: | + 1 - avg by(instance) (rate(node_cpu_seconds_total{mode="idle"}[5m])) > 0.90 + for: 10m + labels: + severity: warning + annotations: + summary: "CPU usage above 90% on {{ $labels.instance }}" + description: "Sustained high CPU can cause raft heartbeat misses and leader flapping." + + - alert: HighMemory + expr: | + 1 - node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes > 0.90 + for: 10m + labels: + severity: warning + annotations: + summary: "Memory usage above 90% on {{ $labels.instance }}" + description: "The host is close to memory exhaustion; the ZenBPM process may be OOM-killed." + + # A-T8: throughput drop + - alert: ThroughputDrop + expr: | + sum(rate(processes_started_total[10m])) + < + 0.2 * sum(rate(processes_started_total[10m] offset 1h)) + for: 10m + labels: + severity: warning + annotations: + summary: "Process start throughput dropped by more than 80%" + description: "Instances started per second dropped more than 80% compared to one hour ago. Check upstream callers and cluster health." + + # A-T9: raft instability + - alert: RaftLeaderFlapping + expr: increase(partition_leader_changes_total[10m]) > 3 + labels: + severity: warning + annotations: + summary: "More than 3 leader changes in 10m on partition {{ $labels.partition }}" + description: "Frequent leader elections indicate network instability or resource starvation." + + # A-T10: goroutine leak + # Uses go_goroutines from the prometheus client_golang collector, which + # is always exported by the /system/metrics promhttp handler. + - alert: GoroutineLeak + expr: | + go_goroutines > 1000 + and + delta(go_goroutines[1h]) > 500 + for: 30m + labels: + severity: warning + annotations: + summary: "Goroutine count is growing on {{ $labels.instance }}" + description: "Goroutines have grown by more than 500 in the last hour; possible goroutine leak." + + # ────────────────────────────────────────────────────────────────────── + # Business alerts + # ────────────────────────────────────────────────────────────────────── + - name: zenbpm-business + rules: + # A-B1: incidents + # NOTE: deliberately non-paging (warning): incidents are a + # business-as-usual signal in a busy BPMN system. Route "warning" to a + # low-urgency receiver in alertmanager.yml; use HighJobFailureRate for + # paging-worthy systemic failures. + - alert: IncidentCreated + expr: increase(incidents_created_total[5m]) > 0 + labels: + severity: warning + annotations: + summary: "BPMN incident(s) created (element {{ $labels.element_id }})" + description: "{{ $value }} incident(s) were raised in the last 5 minutes. Inspect /v1 incidents API or the incidents dashboard." + + # A-B2: job failures + - alert: HighJobFailureRate + expr: | + ( + sum(rate(jobs_failed_total[10m])) + / + clamp_min(sum(rate(jobs_created_total[10m])), 1e-9) + ) > 0.1 + for: 10m + labels: + severity: critical + annotations: + summary: "More than 10% of jobs are failing" + description: "Job failure ratio exceeded 10% over the last 10 minutes. Check worker health and incident list." + + # A-B3: job backlog + # NOTE: every replica of a partition exports jobs_waiting for its local + # copy of the data, so the series must be deduplicated with max by() + # instead of summed — sum would multiply the backlog by the replication + # factor and fire false alerts. + - alert: JobBacklogGrowing + expr: max by(partition) (jobs_waiting) > 1000 + for: 15m + labels: + severity: warning + annotations: + summary: "Job backlog above 1000 on partition {{ $labels.partition }}" + description: "Jobs are waiting longer than expected; workers may be down, too slow or not subscribed." + + # A-B4: stuck process instances + # NOTE: processes_completed_total counts *ended* instances (completed or + # failed), so this alert only fires when instances neither complete nor + # fail while the active count keeps growing. + # process_instances_active is exported by every partition replica, so + # replicas are deduplicated with max by(partition) before summing. + - alert: StuckProcessInstances + expr: | + sum(max by(partition) (process_instances_active)) > 0 + and + sum(rate(processes_completed_total[15m])) == 0 + and + delta(sum(max by(partition) (process_instances_active))[15m:1m]) > 0 + for: 15m + labels: + severity: warning + annotations: + summary: "Active process instances are accumulating while none complete" + description: "The number of active instances keeps growing while completion rate is zero. Engine or downstream systems may be stuck." + + # A-B5: job distribution broken + # jobs_waiting is exported per replica; max by(partition) deduplicates + - alert: NoJobDistribution + expr: | + sum(rate(jobs_distributed_total[10m])) == 0 + and + sum(max by(partition) (jobs_waiting)) > 0 + for: 10m + labels: + severity: critical + annotations: + summary: "Jobs are waiting but none are being distributed to workers" + description: "The job streaming pipeline is broken: no partition leader for the job server, no connected workers, or a gRPC issue." diff --git a/scripts/prometheus.yml b/scripts/prometheus.yml index d5503e920..b43e0d4b1 100644 --- a/scripts/prometheus.yml +++ b/scripts/prometheus.yml @@ -1,6 +1,14 @@ global: scrape_interval: 5s +rule_files: + - /etc/prometheus/prometheus-rules.yml + +alerting: + alertmanagers: + - static_configs: + - targets: ['host.docker.internal:9093'] + scrape_configs: - job_name: zenbpm metrics_path: /system/metrics From 36f7f5e437fa29e6ebfb4fdc977495c312f7f6b3 Mon Sep 17 00:00:00 2001 From: alisher Date: Fri, 31 Jul 2026 12:42:52 +0200 Subject: [PATCH 2/4] =?UTF-8?q?#620=20Monitoring=20=E2=80=93=20OTEL,=20Gra?= =?UTF-8?q?fana,=20graphs=20and=20alerting=20=20-=20fixes=20for=20CI=20aut?= =?UTF-8?q?o=20AI=20findings:=20update=20OpenTelemetry=20attribute=20names?= =?UTF-8?q?=20for=20consistency=20and=20improve=20Grafana=20query=20for=20?= =?UTF-8?q?partition=20leaders?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/cluster/node.go | 2 +- internal/cluster/partition/partition_persistence.go | 6 ++++++ internal/cluster/store/metrics.go | 2 +- internal/rest/server.go | 6 ++++-- pkg/bpmn/engine.go | 4 ++-- pkg/bpmn/engine_api.go | 8 ++++---- pkg/bpmn/multi_instance.go | 2 +- pkg/bpmn/sub_process.go | 4 ++-- pkg/dmn/dmn_engine.go | 4 ++-- pkg/otel/traces.go | 9 +++++---- .../grafana_provisioning/dashboards/zenbpm/cluster.json | 2 +- 11 files changed, 29 insertions(+), 20 deletions(-) diff --git a/internal/cluster/node.go b/internal/cluster/node.go index 9788671dc..921ecdad9 100644 --- a/internal/cluster/node.go +++ b/internal/cluster/node.go @@ -1888,7 +1888,7 @@ func (node *ZenNode) Health() (bool, []string) { reasons = append(reasons, "no cluster leader elected") } cs := node.store.ClusterState() - if desired := cs.Config.DesiredPartitions; uint32(len(cs.Partitions)) < desired { + 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) diff --git a/internal/cluster/partition/partition_persistence.go b/internal/cluster/partition/partition_persistence.go index 71f379308..ef1029803 100644 --- a/internal/cluster/partition/partition_persistence.go +++ b/internal/cluster/partition/partition_persistence.go @@ -422,6 +422,9 @@ func (r rqliteResult) RowsAffected() (int64, error) { return r.rowsAffected, nil } +// ExecContext executes a single write SQL statement against the partition +// rqlite store. It records an "rqlite-exec" trace span and the execution +// duration metric labeled with the partition and outcome. func (rq *DB) ExecContext(ctx context.Context, sql string, args ...interface{}) (_ ssql.Result, retErr error) { start := time.Now() ctx, execSpan := rq.tracer.Start(ctx, "rqlite-exec", trace.WithAttributes( @@ -474,6 +477,9 @@ func (rq *DB) PrepareContext(ctx context.Context, sql string) (*ssql.Stmt, error return nil, errors.New("PrepareContext not supported by rqlite") } +// QueryContext executes a read SQL query against the partition rqlite store. +// It records an "rqlite-query" trace span and the query duration metric +// labeled with the partition and outcome. func (rq *DB) QueryContext(ctx context.Context, query string, args ...interface{}) (_ *sql.Rows, retErr error) { start := time.Now() ctx, querySpan := rq.tracer.Start(ctx, "rqlite-query", trace.WithAttributes( diff --git a/internal/cluster/store/metrics.go b/internal/cluster/store/metrics.go index d9a390c74..50e11e9fe 100644 --- a/internal/cluster/store/metrics.go +++ b/internal/cluster/store/metrics.go @@ -75,7 +75,7 @@ func (s *Store) RegisterMetrics() error { return err } - reg, err := meter.RegisterCallback(func(ctx context.Context, o metric.Observer) error { + reg, err := meter.RegisterCallback(func(_ context.Context, o metric.Observer) error { if !s.open.Load() { return nil } diff --git a/internal/rest/server.go b/internal/rest/server.go index 0e7ada102..c07cc36c7 100644 --- a/internal/rest/server.go +++ b/internal/rest/server.go @@ -1,3 +1,5 @@ +// Package rest exposes the public REST API of a ZenBPM node, including the +// OpenAPI-generated endpoints, Prometheus metrics and health probes. package rest import ( @@ -117,12 +119,12 @@ func NewServer(node *cluster.ZenNode, conf config.Config) *Server { }) // liveness probe: reports only that the process is up. It deliberately does not check raft state // so that a leaderless node is not restarted in a loop by an orchestrator. - r.Get("/health/live", func(w http.ResponseWriter, r *http.Request) { + r.Get("/health/live", func(w http.ResponseWriter, _ *http.Request) { writeHealthResponse(w, restLogger, true, nil) }) // readiness probe: 503 until the cluster has a leader, every partition // has a leader and all partitions owned by this node are initialized. - r.Get("/health/ready", func(w http.ResponseWriter, r *http.Request) { + r.Get("/health/ready", func(w http.ResponseWriter, _ *http.Request) { healthy, reasons := node.Health() writeHealthResponse(w, restLogger, healthy, reasons) }) diff --git a/pkg/bpmn/engine.go b/pkg/bpmn/engine.go index 5ca4a4a2f..de7f9cbc3 100644 --- a/pkg/bpmn/engine.go +++ b/pkg/bpmn/engine.go @@ -392,7 +392,7 @@ func (engine *Engine) createInstance( ctx, createSpan := engine.tracer.Start(ctx, fmt.Sprintf("create-instance:%s", instance.ProcessInstance().Definition.BpmnProcessId), trace.WithAttributes( attribute.Int64(otelPkg.AttributeProcessInstanceKey, instance.ProcessInstance().Key), - attribute.String(otelPkg.AttributeProcessId, instance.ProcessInstance().Definition.BpmnProcessId), + attribute.String(otelPkg.AttributeProcessID, instance.ProcessInstance().Definition.BpmnProcessId), attribute.Int64(otelPkg.AttributeProcessDefinitionKey, instance.ProcessInstance().Definition.Key), )) defer func() { @@ -482,7 +482,7 @@ func (engine *Engine) createInstanceWithStartingElements( } ctx, createSpan := engine.tracer.Start(ctx, fmt.Sprintf("start-instance-on-elements: %s %s", instance.ProcessInstance().Definition.BpmnProcessId, startNodeIds), trace.WithAttributes( attribute.Int64(otelPkg.AttributeProcessInstanceKey, instance.ProcessInstance().Key), - attribute.String(otelPkg.AttributeProcessId, instance.ProcessInstance().Definition.BpmnProcessId), + attribute.String(otelPkg.AttributeProcessID, instance.ProcessInstance().Definition.BpmnProcessId), attribute.Int64(otelPkg.AttributeProcessDefinitionKey, instance.ProcessInstance().Definition.Key), )) defer func() { diff --git a/pkg/bpmn/engine_api.go b/pkg/bpmn/engine_api.go index cdcbe4c42..7147d8d9a 100644 --- a/pkg/bpmn/engine_api.go +++ b/pkg/bpmn/engine_api.go @@ -177,7 +177,7 @@ func (engine *Engine) RunProcessInstance(ctx context.Context, instance runtime.P ctx, instanceSpan := engine.tracer.Start(ctx, fmt.Sprintf("run-instance:%s", instance.ProcessInstance().Definition.BpmnProcessId), trace.WithAttributes( attribute.Int64(otelPkg.AttributeProcessInstanceKey, instance.ProcessInstance().Key), - attribute.String(otelPkg.AttributeProcessId, instance.ProcessInstance().Definition.BpmnProcessId), + attribute.String(otelPkg.AttributeProcessID, instance.ProcessInstance().Definition.BpmnProcessId), attribute.Int64(otelPkg.AttributeProcessDefinitionKey, instance.ProcessInstance().Definition.Key), )) @@ -207,7 +207,7 @@ mainLoop: continue } ctx, tokenSpan := engine.tracer.Start(ctx, fmt.Sprintf("token:%s", currentToken.ElementId), trace.WithAttributes( - attribute.String(otelPkg.AttributeElementId, currentToken.ElementId), + attribute.String(otelPkg.AttributeElementID, currentToken.ElementId), attribute.Int64(otelPkg.AttributeElementKey, currentToken.ElementInstanceKey), attribute.Int64(otelPkg.AttributeToken, currentToken.Key), )) @@ -507,7 +507,7 @@ func (engine *Engine) ModifyInstance(ctx context.Context, processInstanceKey int ctx, createSpan := engine.tracer.Start(ctx, fmt.Sprintf("modify-instance:%s", processInstance.ProcessInstance().Definition.BpmnProcessId), trace.WithAttributes( attribute.Int64(otelPkg.AttributeProcessInstanceKey, processInstance.ProcessInstance().Key), - attribute.String(otelPkg.AttributeProcessId, processInstance.ProcessInstance().Definition.BpmnProcessId), + attribute.String(otelPkg.AttributeProcessID, processInstance.ProcessInstance().Definition.BpmnProcessId), attribute.Int64(otelPkg.AttributeProcessDefinitionKey, processInstance.ProcessInstance().Definition.Key), )) defer createSpan.End() @@ -581,7 +581,7 @@ func (engine *Engine) DeleteInstanceVariable(ctx context.Context, processInstanc ctx, createSpan := engine.tracer.Start(ctx, fmt.Sprintf("delete-instance-variable:%s", processInstance.ProcessInstance().Definition.BpmnProcessId), trace.WithAttributes( attribute.Int64(otelPkg.AttributeProcessInstanceKey, processInstance.ProcessInstance().Key), - attribute.String(otelPkg.AttributeProcessId, processInstance.ProcessInstance().Definition.BpmnProcessId), + attribute.String(otelPkg.AttributeProcessID, processInstance.ProcessInstance().Definition.BpmnProcessId), attribute.Int64(otelPkg.AttributeProcessDefinitionKey, processInstance.ProcessInstance().Definition.Key), )) defer createSpan.End() diff --git a/pkg/bpmn/multi_instance.go b/pkg/bpmn/multi_instance.go index ac180645e..359d3f2ff 100644 --- a/pkg/bpmn/multi_instance.go +++ b/pkg/bpmn/multi_instance.go @@ -343,7 +343,7 @@ func (engine *Engine) handleParentProcessContinuationForMultiInstance(ctx contex return nil } ctx, tokenSpan := engine.tracer.Start(ctx, fmt.Sprintf("token:%s", updatedParentToken.ElementId), trace.WithAttributes( - attribute.String(otelPkg.AttributeElementId, updatedParentToken.ElementId), + attribute.String(otelPkg.AttributeElementID, updatedParentToken.ElementId), attribute.Int64(otelPkg.AttributeElementKey, updatedParentToken.ElementInstanceKey), attribute.Int64(otelPkg.AttributeToken, updatedParentToken.Key), )) diff --git a/pkg/bpmn/sub_process.go b/pkg/bpmn/sub_process.go index 2d65dcdef..ca9a59d54 100644 --- a/pkg/bpmn/sub_process.go +++ b/pkg/bpmn/sub_process.go @@ -207,7 +207,7 @@ func (engine *Engine) handleParentProcessContinuationForSubProcess(ctx context.C return nil } ctx, tokenSpan := engine.tracer.Start(ctx, fmt.Sprintf("token:%s", updatedParentToken.ElementId), trace.WithAttributes( - attribute.String(otelPkg.AttributeElementId, updatedParentToken.ElementId), + attribute.String(otelPkg.AttributeElementID, updatedParentToken.ElementId), attribute.Int64(otelPkg.AttributeElementKey, updatedParentToken.ElementInstanceKey), attribute.Int64(otelPkg.AttributeToken, updatedParentToken.Key), )) @@ -338,7 +338,7 @@ func (engine *Engine) handleParentProcessContinuationForCallActivity(ctx context return nil } ctx, tokenSpan := engine.tracer.Start(ctx, fmt.Sprintf("token:%s", updatedParentToken.ElementId), trace.WithAttributes( - attribute.String(otelPkg.AttributeElementId, updatedParentToken.ElementId), + attribute.String(otelPkg.AttributeElementID, updatedParentToken.ElementId), attribute.Int64(otelPkg.AttributeElementKey, updatedParentToken.ElementInstanceKey), attribute.Int64(otelPkg.AttributeToken, updatedParentToken.Key), )) diff --git a/pkg/dmn/dmn_engine.go b/pkg/dmn/dmn_engine.go index 538983ef5..5b3b08cb6 100644 --- a/pkg/dmn/dmn_engine.go +++ b/pkg/dmn/dmn_engine.go @@ -311,9 +311,9 @@ func (engine *ZenDmnEngine) evaluateDRD( // stable span name keeps the tracing backend operation index bounded; // the decision id is carried by the zenbpm.decision.id attribute ctx, evalSpan := engine.tracer.Start(ctx, "dmn.evaluate-decision", trace.WithAttributes( - attribute.String(otelPkg.AttributeDecisionId, decisionDefinition.Id), + attribute.String(otelPkg.AttributeDecisionID, decisionDefinition.Id), attribute.Int64(otelPkg.AttributeDecisionKey, decisionDefinition.Key), - attribute.String(otelPkg.AttributeDrdId, dmnResourceDefinition.Id), + attribute.String(otelPkg.AttributeDrdID, dmnResourceDefinition.Id), )) defer func() { outcome := "success" diff --git a/pkg/otel/traces.go b/pkg/otel/traces.go index c501a5547..02b40751f 100644 --- a/pkg/otel/traces.go +++ b/pkg/otel/traces.go @@ -1,17 +1,18 @@ package otel +// ZenBPM specific span attribute names shared by the BPMN and DMN engines. const ( // Prefix namespaces all ZenBPM specific span attributes following the // OpenTelemetry attribute naming conventions (dot separated namespaces). Prefix = "zenbpm." AttributeProcessInstanceKey = Prefix + "process.instance_key" - AttributeProcessId = Prefix + "process.id" + AttributeProcessID = Prefix + "process.id" AttributeProcessDefinitionKey = Prefix + "process.definition_key" AttributeToken = Prefix + "token.key" - AttributeElementId = Prefix + "element.id" + AttributeElementID = Prefix + "element.id" AttributeElementKey = Prefix + "element.key" AttributeElementName = Prefix + "element.name" AttributeElementType = Prefix + "element.type" @@ -19,10 +20,10 @@ const ( AttributeJobKey = Prefix + "job.key" AttributeIncidentKey = Prefix + "incident.key" - AttributeDecisionId = Prefix + "decision.id" + AttributeDecisionID = Prefix + "decision.id" AttributeDecisionKey = Prefix + "decision.key" AttributeDecisionInstanceKey = Prefix + "decision.instance_key" - AttributeDrdId = Prefix + "decision.drd_id" + AttributeDrdID = Prefix + "decision.drd_id" SpanStatusToken = Prefix + "token.status" ) diff --git a/scripts/grafana_provisioning/dashboards/zenbpm/cluster.json b/scripts/grafana_provisioning/dashboards/zenbpm/cluster.json index 0b4658810..b2b6561e5 100644 --- a/scripts/grafana_provisioning/dashboards/zenbpm/cluster.json +++ b/scripts/grafana_provisioning/dashboards/zenbpm/cluster.json @@ -37,7 +37,7 @@ "gridPos": { "h": 6, "w": 6, "x": 6, "y": 0 }, "id": 2, "options": { "colorMode": "background", "graphMode": "none", "reduceOptions": { "calcs": ["lastNotNull"] } }, - "targets": [ { "expr": "min by(partition) (partition_has_leader)", "legendFormat": "partition {{partition}}", "refId": "A" } ], + "targets": [ { "expr": "max by(partition) (partition_has_leader)", "legendFormat": "partition {{partition}}", "refId": "A" } ], "title": "Partition leaders", "type": "stat" }, From 0194352968010840eb6b390449415a75975a199a Mon Sep 17 00:00:00 2001 From: alisher Date: Fri, 31 Jul 2026 12:58:11 +0200 Subject: [PATCH 3/4] =?UTF-8?q?#620=20Monitoring=20=E2=80=93=20OTEL,=20Gra?= =?UTF-8?q?fana,=20graphs=20and=20alerting=20=20-=20feat(monitoring):=20ad?= =?UTF-8?q?d=20partition=20deficit=20metric=20to=20Grafana=20dashboard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/reference/observability.md | 2 +- .../dashboards/zenbpm/cluster.json | 21 ++++++++++++++++++- 2 files changed, 21 insertions(+), 2 deletions(-) diff --git a/docs/reference/observability.md b/docs/reference/observability.md index 054839f8d..66e5fbe45 100644 --- a/docs/reference/observability.md +++ b/docs/reference/observability.md @@ -127,7 +127,7 @@ placeholder webhook receiver). Key alerts: Provisioned automatically from `scripts/grafana_provisioning/dashboards/zenbpm/`: - `main.json` — processes, jobs, distribution, request duration -- `cluster.json` — leadership, raft health, leader changes +- `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 diff --git a/scripts/grafana_provisioning/dashboards/zenbpm/cluster.json b/scripts/grafana_provisioning/dashboards/zenbpm/cluster.json index b2b6561e5..df61ac687 100644 --- a/scripts/grafana_provisioning/dashboards/zenbpm/cluster.json +++ b/scripts/grafana_provisioning/dashboards/zenbpm/cluster.json @@ -41,10 +41,29 @@ "title": "Partition leaders", "type": "stat" }, + { + "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, + "description": "Desired minus existing partitions. partition_has_leader only emits series for partitions present in the cluster state, so a partition that was never created would silently disappear from the Partition leaders panel. Mirrors the PartitionDeficit alert.", + "fieldConfig": { + "defaults": { + "mappings": [ + { "options": { "0": { "color": "green", "index": 0, "text": "OK" } }, "type": "value" } + ], + "thresholds": { "mode": "absolute", "steps": [ { "color": "green", "value": null }, { "color": "red", "value": 1 } ] } + }, + "overrides": [] + }, + "gridPos": { "h": 6, "w": 6, "x": 12, "y": 0 }, + "id": 8, + "options": { "colorMode": "background", "graphMode": "none", "reduceOptions": { "calcs": ["lastNotNull"] } }, + "targets": [ { "expr": "max(cluster_desired_partitions) - max(cluster_partitions)", "legendFormat": "missing partitions", "refId": "A" } ], + "title": "Partition deficit", + "type": "stat" + }, { "datasource": { "type": "prometheus", "uid": "aew9mifcmw0e8d" }, "fieldConfig": { "defaults": { "custom": { "fillOpacity": 10 } }, "overrides": [] }, - "gridPos": { "h": 6, "w": 12, "x": 12, "y": 0 }, + "gridPos": { "h": 6, "w": 6, "x": 18, "y": 0 }, "id": 3, "options": { "legend": { "displayMode": "list", "placement": "bottom" } }, "targets": [ From e737d5ec3fdbb0204f3b6a90ff96e2be8e45e7ff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ali=C5=A1er?= Date: Fri, 31 Jul 2026 13:19:00 +0200 Subject: [PATCH 4/4] Apply suggestions from code review Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- pkg/bpmn/engine_api.go | 12 +++++++----- pkg/bpmn/engine_batch.go | 11 ++++++++--- pkg/bpmn/events_api.go | 2 +- pkg/bpmn/jobs_api.go | 2 +- pkg/dmn/dmn_engine.go | 2 +- 5 files changed, 18 insertions(+), 11 deletions(-) diff --git a/pkg/bpmn/engine_api.go b/pkg/bpmn/engine_api.go index 7147d8d9a..aa9a6daa8 100644 --- a/pkg/bpmn/engine_api.go +++ b/pkg/bpmn/engine_api.go @@ -322,11 +322,13 @@ mainLoop: engine.metrics.ProcessesEnded.Add(ctx, 1, metric.WithAttributes( attribute.String("bpmn_process_id", instance.ProcessInstance().Definition.BpmnProcessId), )) - if createdAt := instance.ProcessInstance().CreatedAt; !createdAt.IsZero() { - engine.metrics.ProcessInstanceDuration.Record(ctx, float64(time.Since(createdAt))/float64(time.Millisecond), metric.WithAttributes( - attribute.String("bpmn_process_id", instance.ProcessInstance().Definition.BpmnProcessId), - attribute.String("state", instance.ProcessInstance().State.String()), - )) + if engine.metrics != nil && engine.metrics.ProcessInstanceDuration != nil { + if createdAt := instance.ProcessInstance().CreatedAt; !createdAt.IsZero() { + engine.metrics.ProcessInstanceDuration.Record(ctx, float64(time.Since(createdAt))/float64(time.Millisecond), metric.WithAttributes( + attribute.String("bpmn_process_id", instance.ProcessInstance().Definition.BpmnProcessId), + attribute.String("state", instance.ProcessInstance().State.String()), + )) + } } } diff --git a/pkg/bpmn/engine_batch.go b/pkg/bpmn/engine_batch.go index 489eb0406..aac4bfe61 100644 --- a/pkg/bpmn/engine_batch.go +++ b/pkg/bpmn/engine_batch.go @@ -271,9 +271,14 @@ func (engine *Engine) recordIncidentMetric(ctx context.Context, incident bpmnrun return } if incident.ResolvedAt != nil { - engine.metrics.IncidentsResolved.Add(ctx, 1, metric.WithAttributes( - attribute.String("element_id", incident.ElementId), - )) + if engine.metrics.IncidentsResolved != nil { + engine.metrics.IncidentsResolved.Add(ctx, 1, metric.WithAttributes( + attribute.String("element_id", incident.ElementId), + )) + } + return + } + if engine.metrics.IncidentsCreated == nil { return } engine.metrics.IncidentsCreated.Add(ctx, 1, metric.WithAttributes( diff --git a/pkg/bpmn/events_api.go b/pkg/bpmn/events_api.go index dc1f2a2f5..3b0417049 100644 --- a/pkg/bpmn/events_api.go +++ b/pkg/bpmn/events_api.go @@ -34,7 +34,7 @@ func (engine *Engine) PublishMessageByKey(ctx context.Context, subscriptionKey i // counter. Only bounded values (names of deployed subscriptions or "unknown") // may be passed as name to keep metric cardinality under control. func (engine *Engine) recordMessageCorrelationFailure(ctx context.Context, name string, reason string) { - if engine.metrics == nil { + if engine.metrics == nil || engine.metrics.MessageCorrelationFailed == nil { return } engine.metrics.MessageCorrelationFailed.Add(ctx, 1, metric.WithAttributes( diff --git a/pkg/bpmn/jobs_api.go b/pkg/bpmn/jobs_api.go index 536a215f5..fb3ef3b4e 100644 --- a/pkg/bpmn/jobs_api.go +++ b/pkg/bpmn/jobs_api.go @@ -381,7 +381,7 @@ func (engine *Engine) refreshAndValidateJob(ctx context.Context, jobKey int64) ( // recordJobLifetime records the time between job creation and its terminal state, in milliseconds. func (engine *Engine) recordJobLifetime(ctx context.Context, job runtime.Job, outcome string) { - if engine.metrics == nil || job.CreatedAt.IsZero() { + if engine.metrics == nil || engine.metrics.JobLifetime == nil || job.CreatedAt.IsZero() { return } engine.metrics.JobLifetime.Record(ctx, float64(time.Since(job.CreatedAt))/float64(time.Millisecond), metric.WithAttributes( diff --git a/pkg/dmn/dmn_engine.go b/pkg/dmn/dmn_engine.go index 5b3b08cb6..94687a942 100644 --- a/pkg/dmn/dmn_engine.go +++ b/pkg/dmn/dmn_engine.go @@ -52,7 +52,7 @@ func NewEngine(options ...EngineOption) *ZenDmnEngine { var err error engine.evaluationsTotal, err = meter.Int64Counter("dmn_evaluations", metric.WithDescription("Number of DMN decision evaluations")) if err != nil { - hclog.Default().Named(dmnEngineName).Error("Failed to create dmn_evaluations_total instrument", "err", err) + hclog.Default().Named(dmnEngineName).Error("Failed to create dmn_evaluations instrument", "err", err) } engine.evaluationDuration, err = meter.Float64Histogram("dmn_evaluation_duration", metric.WithUnit("ms"),