diff --git a/charts/promanomaly/templates/NOTES.txt b/charts/promanomaly/templates/NOTES.txt index 578b4fa..41764e6 100644 --- a/charts/promanomaly/templates/NOTES.txt +++ b/charts/promanomaly/templates/NOTES.txt @@ -10,6 +10,19 @@ hot-reloads on change (validation failure rolls back automatically). Datasource: URL: {{ .Values.datasource.url }} +{{- if .Values.scaleOut.enabled }} + +Scale-out (active/active sharding) is ENABLED: + - A StatefulSet of {{ .Values.scaleOut.shards }} shard(s) is running; each pod + scores a disjoint slice of the series (shard index from its ordinal). + - safety.max_total_series is a PER-SHARD cap, so the estate ceiling is + max_total_series x shards. Size it per shard, not for the whole estate. + - To resize, change scaleOut.shards (the StatefulSet replicas follow it); + do NOT ``kubectl scale`` the StatefulSet directly. + - Set max_series_per_query above your largest single query's cardinality: + a query truncated at that cap shards non-deterministically. + See docs/operations.md (Horizontal Scale-Out). +{{- end }} Config schema: this chart renders ``apiVersion: promanomaly.io/v1`` (stable since the v1.0 release). The legacy ``promanomaly.io/v1alpha1`` diff --git a/charts/promanomaly/templates/configmap.yaml b/charts/promanomaly/templates/configmap.yaml index 8073dd9..d400809 100644 --- a/charts/promanomaly/templates/configmap.yaml +++ b/charts/promanomaly/templates/configmap.yaml @@ -105,6 +105,9 @@ data: {{- with .Values.safety.fail_ready_after }} fail_ready_after: {{ . }} {{- end }} + {{- with .Values.safety.maxConcurrentGroups }} + max_concurrent_groups: {{ . }} + {{- end }} {{- with .Values.safety.query_cache }} query_cache: enabled: {{ .enabled }} @@ -134,6 +137,19 @@ data: retry_period: {{ .Values.highAvailability.retry_period | quote }} snapshot_ttl: {{ .Values.highAvailability.snapshot_ttl | quote }} {{- end }} + {{- if .Values.scaleOut.enabled }} + scaleOut: + enabled: true + shards: {{ .Values.scaleOut.shards }} + {{- /* + shard_index is normally left unset: each replica self-assigns from + its StatefulSet pod ordinal (POD_NAME, injected via the Downward + API). Only rendered when an operator pins it explicitly. + */}} + {{- with .Values.scaleOut.shard_index }} + shard_index: {{ . }} + {{- end }} + {{- end }} defaults: window: {{ .Values.defaults.window | quote }} step: {{ .Values.defaults.step | quote }} diff --git a/charts/promanomaly/templates/deployment.yaml b/charts/promanomaly/templates/deployment.yaml index 57798dd..62391e4 100644 --- a/charts/promanomaly/templates/deployment.yaml +++ b/charts/promanomaly/templates/deployment.yaml @@ -1,3 +1,9 @@ +{{- /* + Single-replica / active-passive HA topology renders a Deployment. + Active/active scale-out (``scaleOut.enabled``) renders a StatefulSet + instead (statefulset.yaml) so each replica gets a stable shard ordinal. +*/ -}} +{{- if not .Values.scaleOut.enabled -}} {{- /* Bearer-token Secret mounts for the active push sink(s). The scalar ``sink:`` form mounts a single Secret at /etc/promanomaly-sink-auth; the @@ -151,3 +157,4 @@ spec: affinity: {{- toYaml . | nindent 8 }} {{- end }} +{{- end }} diff --git a/charts/promanomaly/templates/prometheusrule.yaml b/charts/promanomaly/templates/prometheusrule.yaml index b02619a..58b5e17 100644 --- a/charts/promanomaly/templates/prometheusrule.yaml +++ b/charts/promanomaly/templates/prometheusrule.yaml @@ -261,4 +261,22 @@ spec: See anomaly_selftest_failures_total. Requires server.selftest.enabled. {{- end }} {{- end }} + {{- with $rules.anomalyGroupStarved }} + {{- if .enabled }} + - alert: AnomalyGroupStarved + expr: rate(anomaly_group_skipped_total[15m]) > 0 + for: {{ .for }} + labels: + severity: {{ .severity }} + annotations: + summary: "promanomaly is shedding runs for {{ "{{" }} $labels.group {{ "}}" }} ({{ "{{" }} $labels.reason {{ "}}" }})" + description: | + Group {{ "{{" }} $labels.group {{ "}}" }} has had runs skipped + (reason={{ "{{" }} $labels.reason {{ "}}" }}) over the last 15 minutes. + The detector cannot keep every group on its cadence under the current + load. Scores for this group are delayed, not suppressed. Raise + safety.max_concurrent_groups, lengthen the group's refresh_interval, + or split its heavy queries. + {{- end }} + {{- end }} {{- end }} diff --git a/charts/promanomaly/templates/rbac-ha.yaml b/charts/promanomaly/templates/rbac-ha.yaml index 5c707ff..e8f0f3a 100644 --- a/charts/promanomaly/templates/rbac-ha.yaml +++ b/charts/promanomaly/templates/rbac-ha.yaml @@ -1,5 +1,6 @@ -{{- if .Values.highAvailability.enabled -}} -# Lease-API permissions required by the HA leader-election loop. +{{- if and .Values.highAvailability.enabled (not .Values.scaleOut.enabled) -}} +# Lease-API permissions required by the HA leader-election loop. Never +# emitted in scale-out mode (active/active sharding needs no Lease). # Only the get/list/watch/create/update verbs on coordination.k8s.io # Leases are required — no cluster-scoped permissions, scoped to a # single Lease resource by name. Renders only when HA is enabled so diff --git a/charts/promanomaly/templates/statefulset.yaml b/charts/promanomaly/templates/statefulset.yaml new file mode 100644 index 0000000..fc412f6 --- /dev/null +++ b/charts/promanomaly/templates/statefulset.yaml @@ -0,0 +1,177 @@ +{{- /* + Active/active scale-out (``scaleOut.enabled``) renders a StatefulSet so + every replica gets a stable ordinal (pod-0..pod-N). The detector reads + that ordinal from POD_NAME and self-assigns its shard index; each shard + scores a disjoint slice of the series via rendezvous hashing. The + Deployment (deployment.yaml) renders in every other topology. The pod + spec mirrors the Deployment's; the only differences are the StatefulSet + wrapper (serviceName, ordinal-derived replicas, parallel pod management) + and the absence of a rollout strategy. +*/ -}} +{{- if .Values.scaleOut.enabled -}} +{{- /* Fail fast on the mutually-exclusive combination rather than letting + the pods CrashLoop at boot on the pydantic validation (and granting a + scale-out workload Lease RBAC it can never use). Mirrors the sink/sinks + fail guard in configmap.yaml. */ -}} +{{- if .Values.highAvailability.enabled -}} +{{- fail "scaleOut.enabled and highAvailability.enabled are mutually exclusive: scale-out is active/active sharding, HA is active/passive leader election. Enable exactly one." -}} +{{- end -}} +{{- /* Bearer-token Secret mounts for the active push sink(s) — identical + to deployment.yaml so the two topologies mount sinks the same way. */ -}} +{{- $sinkMounts := list -}} +{{- if .Values.sinks -}} + {{- range $sink := .Values.sinks -}} + {{- $secret := dig $sink.type "auth" "existingSecret" "" $sink -}} + {{- if $secret -}} + {{- $name := printf "sink-auth-%s" (replace "_" "-" $sink.type) -}} + {{- $sinkMounts = append $sinkMounts (dict "name" $name "secret" $secret "path" (include "promanomaly.sinkAuthDir" $sink.type)) -}} + {{- end -}} + {{- end -}} +{{- else if .Values.sink.type -}} + {{- $scalarSecret := dig .Values.sink.type "auth" "existingSecret" "" .Values.sink -}} + {{- if $scalarSecret -}} + {{- $sinkMounts = append $sinkMounts (dict "name" "sink-auth" "secret" $scalarSecret "path" "/etc/promanomaly-sink-auth") -}} + {{- end -}} +{{- end -}} +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: {{ include "promanomaly.fullname" . }} + labels: {{- include "promanomaly.labels" . | nindent 4 }} +spec: + # One replica per shard; replicaCount is ignored in scale-out mode. + replicas: {{ .Values.scaleOut.shards }} + # Stable headless Service backing the pods' DNS identities. + serviceName: {{ include "promanomaly.fullname" . }}-headless + # Parallel: every shard owns a disjoint slice, so there is no ordering + # dependency between pods — start and reschedule them together. + podManagementPolicy: Parallel + selector: + matchLabels: {{- include "promanomaly.selectorLabels" . | nindent 6 }} + template: + metadata: + annotations: + {{- with .Values.podAnnotations }} + {{- toYaml . | nindent 8 }} + {{- end }} + checksum/config: {{ toYaml .Values | sha256sum }} + labels: + {{- include "promanomaly.selectorLabels" . | nindent 8 }} + {{- with .Values.podLabels }} + {{- toYaml . | nindent 8 }} + {{- end }} + spec: + serviceAccountName: {{ include "promanomaly.serviceAccountName" . }} + {{- with .Values.imagePullSecrets }} + imagePullSecrets: + {{- toYaml . | nindent 8 }} + {{- end }} + securityContext: + {{- toYaml .Values.securityContext | nindent 8 }} + containers: + - name: promanomaly + image: "{{ .Values.image.repository }}:{{ default .Chart.AppVersion .Values.image.tag }}" + imagePullPolicy: {{ .Values.image.pullPolicy }} + args: + - "--config" + - "/etc/promanomaly/config.yaml" + env: + # Downward API: POD_NAME carries the StatefulSet ordinal, which + # the detector parses into its shard index at boot. No per-pod + # config — the ordinal self-assigns shards 0..N-1. + - name: POD_NAME + valueFrom: + fieldRef: + fieldPath: metadata.name + - name: POD_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace + ports: + - name: http + containerPort: 9092 + protocol: TCP + livenessProbe: + httpGet: + path: /healthz + port: http + initialDelaySeconds: 10 + periodSeconds: 30 + readinessProbe: + httpGet: + path: /ready + port: http + initialDelaySeconds: 5 + periodSeconds: 10 + resources: + {{- toYaml .Values.resources | nindent 12 }} + securityContext: + {{- toYaml .Values.containerSecurityContext | nindent 12 }} + volumeMounts: + - name: config + mountPath: /etc/promanomaly + readOnly: true + {{- if .Values.datasource.auth.existingSecret }} + - name: datasource-auth + mountPath: /etc/promanomaly-datasource-auth + readOnly: true + {{- end }} + {{- range $sinkMounts }} + - name: {{ .name }} + mountPath: {{ .path }} + readOnly: true + {{- end }} + - name: tmp + mountPath: /tmp + volumes: + - name: config + configMap: + name: {{ include "promanomaly.configMapName" . }} + {{- if .Values.datasource.auth.existingSecret }} + - name: datasource-auth + secret: + secretName: {{ .Values.datasource.auth.existingSecret | quote }} + {{- end }} + {{- range $sinkMounts }} + - name: {{ .name }} + secret: + secretName: {{ .secret | quote }} + {{- end }} + - name: tmp + emptyDir: {} + {{- with .Values.nodeSelector }} + nodeSelector: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.tolerations }} + tolerations: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.affinity }} + affinity: + {{- toYaml . | nindent 8 }} + {{- end }} +--- +{{- /* + Headless Service required by the StatefulSet's ``serviceName``. The + detector derives its shard index from the POD_NAME ordinal alone (no + peer discovery), so this Service exists only to satisfy the StatefulSet + contract and give the pods stable DNS should anything ever need it. + Scraping goes through the regular Service (service.yaml), which selects + all shards' pods, so the ServiceMonitor collects every shard's + /metrics — the union covers the whole estate. +*/}} +apiVersion: v1 +kind: Service +metadata: + name: {{ include "promanomaly.fullname" . }}-headless + labels: {{- include "promanomaly.labels" . | nindent 4 }} +spec: + clusterIP: None + ports: + - name: http + port: {{ .Values.service.port }} + targetPort: http + protocol: TCP + selector: {{- include "promanomaly.selectorLabels" . | nindent 4 }} +{{- end }} diff --git a/charts/promanomaly/values.schema.json b/charts/promanomaly/values.schema.json index 263dfc7..a8a7d19 100644 --- a/charts/promanomaly/values.schema.json +++ b/charts/promanomaly/values.schema.json @@ -20,7 +20,7 @@ "replicaCount": { "type": "integer", "minimum": 1, - "description": "Replica count. Single-replica mode requires 1; bump only when highAvailability.enabled=true." + "description": "Replica count. Single-replica mode requires 1; bump only when highAvailability.enabled=true. Ignored when scaleOut.enabled=true (the StatefulSet runs scaleOut.shards replicas)." }, "highAvailability": { "type": "object", @@ -34,6 +34,15 @@ "snapshot_ttl": {"type": "string"} } }, + "scaleOut": { + "type": "object", + "description": "Active/active horizontal sharding. Mutually exclusive with highAvailability; renders a StatefulSet of `shards` replicas, each scoring a disjoint slice of the series.", + "properties": { + "enabled": {"type": "boolean"}, + "shards": {"type": "integer", "minimum": 1}, + "shard_index": {"type": ["integer", "null"], "minimum": 0} + } + }, "datasource": { "type": "object", "required": ["url"], @@ -128,6 +137,7 @@ "query_timeout": {"type": "string"}, "on_source_failure": {"type": "string", "enum": ["serve_stale", "drop_scores", "fail_ready"]}, "fail_ready_after": {"type": "integer", "minimum": 1}, + "maxConcurrentGroups": {"type": ["integer", "null"], "minimum": 1}, "max_stratified_cache_entries": {"type": "integer", "minimum": 1}, "query_cache": { "type": "object", diff --git a/charts/promanomaly/values.yaml b/charts/promanomaly/values.yaml index 38e1046..7df0f67 100644 --- a/charts/promanomaly/values.yaml +++ b/charts/promanomaly/values.yaml @@ -17,6 +17,8 @@ replicaCount: 1 # in single-replica mode this MUST stay at 1. # to bump (typically to 2 or 3) and the chart # automatically flips to a RollingUpdate strategy # via the override below. + # Ignored when scaleOut.enabled=true — the + # StatefulSet runs exactly scaleOut.shards replicas. strategy: # Recreate is correct for the single-replica MVP. The chart's helper @@ -105,6 +107,15 @@ safety: # Only honoured when on_source_failure: fail_ready. A single # transient TSDB blip must not take the pod out of the service. fail_ready_after: 3 + # Priority-aware load shedding. Unset (null) leaves group scheduling + # unbounded — every group runs on its own cadence with no admission + # control. Set to an integer to cap how many groups run at once; under + # TSDB backpressure the scheduler then prefers higher-priority groups + # and sheds (skips) low-priority runs that can't get a slot within + # their refresh interval, surfacing them on + # anomaly_group_skipped_total{reason="backpressure"}. See + # docs/operations/degraded-modes.md. + maxConcurrentGroups: null # In-process LRU+TTL cache for overlapping PromQL responses. The # Redis-backed HA variant is selected with ``backend: redis`` and # requires the ``redis:`` block below. See docs/operations.md for @@ -150,6 +161,24 @@ highAvailability: retry_period: 2s snapshot_ttl: 5m +# Active/active horizontal scale-out. Orthogonal to (and mutually +# exclusive with) highAvailability: HA is active/passive (one leader does +# all detection), scale-out is active/active (``shards`` replicas each +# score a disjoint slice of the series via rendezvous hashing and run +# concurrently). When enabled the chart renders a StatefulSet instead of +# the Deployment so each pod gets a stable ordinal — the shard index, +# self-assigned from POD_NAME with no per-pod config — plus a headless +# Service for stable pod DNS. ``replicaCount`` is ignored in this mode; +# the StatefulSet runs exactly ``shards`` replicas. Remember that +# ``safety.max_total_series`` becomes a PER-SHARD cap, so the estate +# ceiling is ``max_total_series x shards`` (see docs/operations.md). +scaleOut: + enabled: false + shards: 1 + # Leave unset so each pod derives its index from its StatefulSet + # ordinal. Pin only for non-StatefulSet topologies. + shard_index: null + defaults: window: 1h step: 15s @@ -437,3 +466,11 @@ prometheusRule: enabled: true for: 5m severity: critical + # AnomalyGroupStarved fires when the priority-aware scheduler sheds + # group runs under sustained backpressure (safety.max_concurrent_groups) + # or a group overruns its refresh interval. A skip delays a score, it + # never suppresses a detected anomaly. Dormant unless shedding occurs. + anomalyGroupStarved: + enabled: true + for: 15m + severity: warning diff --git a/detector/src/promanomaly/cli/_cost.py b/detector/src/promanomaly/cli/_cost.py index df014d4..ca985e7 100644 --- a/detector/src/promanomaly/cli/_cost.py +++ b/detector/src/promanomaly/cli/_cost.py @@ -86,6 +86,13 @@ class GroupCostEstimate: recurrence_queries_per_day: float = 0.0 memory_bytes: float = 0.0 cpu_cores: float = 0.0 + # Effective refresh cadence (per-group override or server default) and + # the derived short-query rate. Per-group cadences make a raw + # per-refresh count misleading — a 30s group and a 5m group issuing + # one query each are 10x apart in TSDB load — so the rate is what an + # operator sizes the backend against. + refresh_interval_seconds: float = 0.0 + short_queries_per_minute: float = 0.0 queries: list[QueryCostEstimate] = field(default_factory=list) @@ -94,6 +101,7 @@ def estimate_query_cost( cfg: Config, group_name: str, query: Any, + refresh_seconds: float | None = None, ) -> QueryCostEstimate: """Project one query's worst-case cost without touching the TSDB. @@ -157,10 +165,16 @@ def estimate_query_cost( # more (or less) work this query's longest window implies. anchor_points = max(default_window / step_seconds, 1.0) work_factor = _nlogn(window_points) / _nlogn(int(anchor_points)) - refresh_seconds = max(cfg.server.refresh_interval_seconds, 1e-9) + # A faster per-group cadence packs more passes per minute, so the + # steady-state CPU scales inversely with the *group's* refresh, not + # the server default. Falls back to the server default when unset. + effective_refresh = ( + refresh_seconds if refresh_seconds is not None else (cfg.server.refresh_interval_seconds) + ) + effective_refresh = max(effective_refresh, 1e-9) cpu_cores = ( (projected_series * detector_count / _ANCHOR_SERIES) - * (_ANCHOR_REFRESH_SECONDS / refresh_seconds) + * (_ANCHOR_REFRESH_SECONDS / effective_refresh) * work_factor ) @@ -183,9 +197,15 @@ def estimate_cost(cfg: Config) -> list[GroupCostEstimate]: """Project per-group cost across the whole config.""" groups: list[GroupCostEstimate] = [] for group in cfg.groups: - gest = GroupCostEstimate(group=group.name) + refresh_seconds = cfg.group_refresh_seconds(group) + gest = GroupCostEstimate(group=group.name, refresh_interval_seconds=refresh_seconds) for query in group.queries: - qest = estimate_query_cost(cfg=cfg, group_name=group.name, query=query) + qest = estimate_query_cost( + cfg=cfg, + group_name=group.name, + query=query, + refresh_seconds=refresh_seconds, + ) gest.queries.append(qest) gest.projected_series += qest.projected_series gest.short_queries_per_refresh += qest.short_queries_per_refresh @@ -193,6 +213,12 @@ def estimate_cost(cfg: Config) -> list[GroupCostEstimate]: gest.stratified_baseline_queries_per_day += qest.stratified_baseline_queries_per_day gest.memory_bytes += qest.memory_bytes gest.cpu_cores += qest.cpu_cores + # Short-query rate folds the per-group cadence in so a too-aggressive + # per-group interval shows up as outsized TSDB load pre-deploy. + if refresh_seconds > 0: + gest.short_queries_per_minute = round( + gest.short_queries_per_refresh * 60.0 / refresh_seconds, 2 + ) # Recurrence analysis: one TSDB range query per refresh_interval # (not per group refresh), amortised to queries-per-day. if group.recurrence.enabled: @@ -215,6 +241,7 @@ def run_estimate_cost(cfg: Config, *, strict: bool) -> int: total_memory = 0.0 total_cpu = 0.0 short_total = 0 + short_per_minute_total = 0.0 probe_total = 0 stratified_total = 0.0 recurrence_total = 0.0 @@ -224,6 +251,7 @@ def run_estimate_cost(cfg: Config, *, strict: bool) -> int: total_memory += gest.memory_bytes total_cpu += gest.cpu_cores short_total += gest.short_queries_per_refresh + short_per_minute_total += gest.short_queries_per_minute probe_total += gest.discovery_probe_queries_per_refresh stratified_total += gest.stratified_baseline_queries_per_day recurrence_total += gest.recurrence_queries_per_day @@ -233,7 +261,9 @@ def run_estimate_cost(cfg: Config, *, strict: bool) -> int: "group": gest.group, "status": "estimate", "projected_series": gest.projected_series, + "refresh_interval_seconds": gest.refresh_interval_seconds, "short_queries_per_refresh": gest.short_queries_per_refresh, + "short_queries_per_minute": gest.short_queries_per_minute, "discovery_probe_queries_per_refresh": (gest.discovery_probe_queries_per_refresh), "stratified_baseline_queries_per_day": round( gest.stratified_baseline_queries_per_day, 2 @@ -257,6 +287,7 @@ def run_estimate_cost(cfg: Config, *, strict: bool) -> int: "max_total_series": cfg.safety.max_total_series, "over_budget": over_budget, "short_queries_per_refresh": short_total, + "short_queries_per_minute": round(short_per_minute_total, 2), "discovery_probe_queries_per_refresh": probe_total, "stratified_baseline_queries_per_day": round(stratified_total, 2), "recurrence_queries_per_day": round(recurrence_total, 2), diff --git a/detector/src/promanomaly/config.py b/detector/src/promanomaly/config.py index c6a6d88..ae7c779 100644 --- a/detector/src/promanomaly/config.py +++ b/detector/src/promanomaly/config.py @@ -446,9 +446,62 @@ def snapshot_ttl_seconds(self) -> float: return parse_duration(self.snapshot_ttl) +class ScaleOutConfig(_ModelBase): + """Active/active horizontal sharding via consistent hashing. + + Orthogonal to :class:`HighAvailabilityConfig`. HA mode is + active/passive — one leader runs the scheduler, followers serve a + cached ``/metrics`` — which caps total detection throughput at a + single process. Scale-out mode is active/active: ``shards`` replicas + each own a disjoint slice of the series (assigned by rendezvous / + highest-random-weight hashing on the series key) and run + concurrently, so throughput scales horizontally. The union of the + replicas' ``/metrics`` covers the whole estate. + + Off by default (``shards: 1``, ``enabled: false``) — the MVP topology + is a single replica. The two modes are mutually exclusive: enabling + both is rejected at load time (leader-per-shard layering is a + documented future direction, not this release). + + Each replica's ``shard_index`` is resolved at boot from the + StatefulSet pod ordinal (the trailing ``-N`` of ``POD_NAME``); set + ``shard_index`` explicitly only for non-StatefulSet topologies or + tests. ``max_total_series`` becomes a *per-shard* cap, so the whole + estate's ceiling is ``max_total_series x shards`` — see + docs/operations.md for sizing. + """ + + enabled: bool = False + # Total number of shards the estate is partitioned across. With a + # StatefulSet this equals the replica count. Rendezvous hashing keeps + # resharding on a count change to ~1/shards of the keys. + shards: int = Field(default=1, ge=1) + # Explicit shard index override. Normally unset — the runtime derives + # it from the pod ordinal so each replica self-assigns. Must be in + # ``[0, shards)`` when set; validated at load time. + shard_index: int | None = Field(default=None, ge=0) + + @model_validator(mode="after") + def _validate_index_in_range(self) -> ScaleOutConfig: + if self.shard_index is not None and self.shard_index >= self.shards: + raise ValueError( + f"scaleOut.shard_index={self.shard_index} must be < scaleOut.shards={self.shards}" + ) + return self + + class SafetyConfig(_ModelBase): max_series_per_query: int = Field(default=1000, ge=1) max_total_series: int = Field(default=20000, ge=1) + # Upper bound on how many groups run concurrently. ``None`` (default) + # leaves group scheduling unbounded — every group runs on its own + # cadence with no admission control, the historical behaviour. When + # set, a priority-aware gate admits at most this many group runs at + # once; under saturation lower-``priority`` groups are shed (skipped, + # never queued unboundedly) and the skip is surfaced on + # ``anomaly_group_skipped_total{reason="backpressure"}``. See + # docs/operations/degraded-modes.md. + max_concurrent_groups: int | None = Field(default=None, ge=1) detect_timeout: Duration = "5s" on_source_failure: Literal["serve_stale", "drop_scores", "fail_ready"] = "serve_stale" # Only consulted in fail_ready mode: after this many consecutive @@ -913,6 +966,13 @@ class GroupConfig(_ModelBase): ensemble: EnsembleConfig | None = None auto_select: AutoSelect = False auto_select_interval: Duration = "24h" + # Per-group refresh cadence override. ``None`` (default) falls back to + # ``server.refresh_interval``. Lets a cheap fast-moving group (error + # rates every 30s) and an expensive stratified/cohort group (every 5m) + # coexist in one detector without a global compromise. The per-group + # lock already prevents overlap; the ``refresh_interval >= window/6`` + # quality guard is enforced per group at boot. + refresh_interval: Duration | None = None recurrence: RecurrenceConfig = Field(default_factory=RecurrenceConfig) # Per-group tenant override. When set, this tenant header is injected # on every PromQL request for this group instead of the datasource-level @@ -925,6 +985,18 @@ class GroupConfig(_ModelBase): def auto_select_interval_seconds(self) -> float: return parse_duration(self.auto_select_interval) + @property + def refresh_interval_override_seconds(self) -> float | None: + """This group's explicit refresh cadence in seconds, or ``None``. + + ``None`` signals the caller to fall back to + ``server.refresh_interval``; see + :meth:`Config.group_refresh_seconds`. + """ + if self.refresh_interval is None: + return None + return parse_duration(self.refresh_interval) + @field_validator("name") @classmethod def _validate_name(cls, value: str) -> str: @@ -969,6 +1041,7 @@ class Config(_ModelBase): defaults: DefaultsConfig = Field(default_factory=DefaultsConfig) exporter: ExporterConfig = Field(default_factory=ExporterConfig) highAvailability: HighAvailabilityConfig = Field(default_factory=HighAvailabilityConfig) + scaleOut: ScaleOutConfig = Field(default_factory=ScaleOutConfig) telemetry: TelemetryConfig = Field(default_factory=TelemetryConfig) # Optional push sink(s) active alongside the always-on ``/metrics`` # exporter. Both default to pull-only. @@ -1021,6 +1094,21 @@ def effective_sinks(self) -> list[SinkConfig]: return [self.sink] return [] + @model_validator(mode="after") + def _validate_scaleout_exclusive_with_ha(self) -> Config: + # Active/active sharding and active/passive leader election are + # two answers to "how many replicas do detection work" and don't + # compose in this release (leader-per-shard is a documented + # future direction). Reject the combination at boot rather than + # let an operator deploy a topology whose behaviour is undefined. + if self.scaleOut.enabled and self.highAvailability.enabled: + raise ValueError( + "scaleOut.enabled and highAvailability.enabled are mutually exclusive; " + "scale-out is active/active sharding, HA is active/passive leader election. " + "Pick one (leader-per-shard layering is not supported in this release)." + ) + return self + @model_validator(mode="after") def _validate_ha_requires_redis(self) -> Config: # HA mode needs the Redis snapshot cache; without it followers @@ -1091,6 +1179,19 @@ def _validate_groups(cls, groups: list[GroupConfig]) -> list[GroupConfig]: raise ValueError(f"duplicate group name: {names}") return groups + def group_refresh_seconds(self, group: GroupConfig) -> float: + """Effective refresh cadence for ``group`` in seconds. + + Returns the group's own ``refresh_interval`` when set, else falls + back to ``server.refresh_interval``. Single source of truth for + the scheduler, the cost estimator, and the aggressive-refresh + guard so per-group cadence is computed identically everywhere. + """ + override = group.refresh_interval_override_seconds + if override is not None: + return override + return self.server.refresh_interval_seconds + @property def aggressive_refresh(self) -> bool: # Operational guard: ``refresh_interval`` should be at least @@ -1098,9 +1199,29 @@ def aggressive_refresh(self) -> bool: # six or more passes per window's worth of data and the rolling # baseline lags. The roadmap is explicit that this is a warning, # not a hard fail — main.py logs at boot when this returns True. + # Reflects the *server* default; per-group overrides are surfaced + # separately via :meth:`aggressive_refresh_groups`. floor = self.defaults.window_seconds / 6.0 return self.server.refresh_interval_seconds < floor + def aggressive_refresh_groups(self) -> list[tuple[str, float]]: + """Groups whose effective refresh is below the ``window/6`` floor. + + Returns ``(group_name, effective_refresh_seconds)`` for every + group running too aggressively for its rolling window — including + groups that inherit an aggressive server default. The boot path + logs one ``aggressive_refresh_interval`` line per entry so a + too-tight per-group override is as visible as a too-tight global + one. + """ + floor = self.defaults.window_seconds / 6.0 + offending: list[tuple[str, float]] = [] + for group in self.groups: + effective = self.group_refresh_seconds(group) + if effective < floor: + offending.append((group.name, effective)) + return offending + def load_config(path: str | Path) -> Config: """Parse a YAML config file or directory into a validated :class:`Config`. diff --git a/detector/src/promanomaly/exporter.py b/detector/src/promanomaly/exporter.py index 8a820b7..58bc761 100644 --- a/detector/src/promanomaly/exporter.py +++ b/detector/src/promanomaly/exporter.py @@ -277,6 +277,7 @@ def __init__(self) -> None: self._declare_runner_metrics() self._declare_cache_metrics() self._declare_leader_metrics() + self._declare_scaleout_metrics() self._declare_discovery_metrics() self._declare_self_observability_metrics() self._declare_selftest_metrics() @@ -363,6 +364,18 @@ def _declare_runner_metrics(self) -> None: ["group", "detector"], registry=self.registry, ) + self.group_skipped_total = Counter( + "anomaly_group_skipped_total", + ( + "Group runs skipped by the priority-aware scheduler, bucketed by " + "reason: 'backpressure' (no concurrency slot within the refresh " + "interval under safety.max_concurrent_groups) or 'overrun' (the " + "previous run was still in flight at the next tick). A skip delays " + "a score; it never suppresses a detected anomaly." + ), + ["group", "reason"], + registry=self.registry, + ) # ------------------------------------------------------------------ # Cache-owned metrics. Cardinality is bounded — one set of series per @@ -414,6 +427,25 @@ def _declare_leader_metrics(self) -> None: registry=self.registry, ) + # ------------------------------------------------------------------ + # Scale-out metrics (active/active sharding only). Cardinality is + # bounded — one ``anomaly_shard`` series per replica and one + # ``anomaly_shard_series_count`` series per shard index. + # ------------------------------------------------------------------ + def _declare_scaleout_metrics(self) -> None: + self.shard = Gauge( + "anomaly_shard", + "Shard index owned by this replica in scaleOut mode (0 in single-shard mode).", + ["instance"], + registry=self.registry, + ) + self.shard_series_count = Gauge( + "anomaly_shard_series_count", + "Series scored by this shard across all groups in the latest runs.", + ["shard"], + registry=self.registry, + ) + # ------------------------------------------------------------------ # Discovery metrics. Per-run fan-out gauges and the per- # variable failure counter; cardinality is bounded by the number of diff --git a/detector/src/promanomaly/main.py b/detector/src/promanomaly/main.py index 921fbce..a8554bd 100644 --- a/detector/src/promanomaly/main.py +++ b/detector/src/promanomaly/main.py @@ -28,6 +28,7 @@ from typing import Any import uvicorn +from apscheduler.events import EVENT_JOB_MAX_INSTANCES from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.interval import IntervalTrigger from fastapi import FastAPI @@ -39,8 +40,10 @@ from .leader import resolve_identity from .logging import configure_logging, get_logger from .runner import Runner +from .scheduling import LoadShedder from .selftest import run_selftest from .server import build_app as _build_app +from .sharding import ShardAssignment, resolve_shard_index from .sinks import Sink, build_sinks from .source import PromQLSource, QueryResult from .state import SnapshotStore @@ -75,13 +78,23 @@ def __init__(self, config: Config, config_path: Path) -> None: # touches telemetry config requires a restart, which is documented # in docs/telemetry.md alongside the zero-overhead caveat. self._telemetry = Telemetry.from_config(config.telemetry) + # Active/active scale-out: resolve this replica's shard from the + # StatefulSet pod ordinal (or an explicit override) once at boot. + # ``None`` when scale-out is off, so the runner's per-series shard + # filter is a no-op and behaviour matches the single-replica MVP. + self._shard = _resolve_shard(config) self._runner = Runner( config=config, source=self._source, store=self._store, operational=self._ops, telemetry=self._telemetry, + shard=self._shard, ) + # Priority-aware admission gate for group scheduling. A pass- + # through (no shedding) unless ``safety.max_concurrent_groups`` + # is set, so default scheduling is byte-identical to before. + self._load_shedder = LoadShedder(config.safety.max_concurrent_groups) self._exporter = Exporter(self._store, self._ops, self._source) # Optional push sinks (remote_write / grafana_annotations) running # alongside the always-on /metrics exporter. Empty when pull-only. @@ -110,6 +123,10 @@ def __init__(self, config: Config, config_path: Path) -> None: # it. Single-replica leaves this dict empty. self._running_group_tasks: dict[str, set[asyncio.Task[None]]] = {} self._identity = resolve_identity(config.highAvailability.identity) + if self._shard is not None: + # Publish which shard this replica owns so an operator can see + # the partition (and confirm every shard index is covered). + self._ops.shard.labels(instance=self._identity).set(float(self._shard.index)) self._stamp_config_hash(config) if self._ha_enabled: self._ops.leader_elected.labels(instance=self._identity).set(0.0) @@ -133,6 +150,12 @@ async def startup(self) -> None: for sink in self._sinks: await sink.start() self._scheduler = AsyncIOScheduler() + # Register the overrun listener exactly once for the scheduler's + # lifetime. APScheduler's add_listener appends without de-duping, + # so installing it inside _reschedule_jobs (which reruns on every + # reload) would stack duplicate listeners and multiply the + # anomaly_group_skipped_total{reason="overrun"} counter per reload. + self._scheduler.add_listener(self._on_job_max_instances, EVENT_JOB_MAX_INSTANCES) self._reschedule_jobs(self._scheduler) self._scheduler.start() # SIGHUP must be wired here, not in serve(), because only inside @@ -202,6 +225,24 @@ async def reload(self) -> tuple[bool, str]: logger.error("reload_validation_failed", error=str(exc)) return False, f"validation failed: {exc}" + # Scale-out sharding is resolved once at boot from the pod + # ordinal; the shard assignment is baked into the runner and + # the anomaly_shard gauge. A hot reload can't safely change it + # (the replica count / ordinals only change via a StatefulSet + # rollout, which restarts pods anyway). Reject the reload + # rather than silently keep scoring the old slice under the new + # shard count — that would leave the estate partially scored. + if new_config.scaleOut != self._config.scaleOut: + logger.error( + "reload_scaleout_change_rejected", + old_shards=self._config.scaleOut.shards, + new_shards=new_config.scaleOut.shards, + ) + return False, ( + "scaleOut changes require a pod restart (roll the StatefulSet), " + "not a hot reload; the previous config stays active" + ) + self._runner.replace_config(new_config) # Close stale tenant sources — the runner will lazily create # new ones on the next group run with the new config's tenants. @@ -231,6 +272,10 @@ async def reload(self) -> tuple[bool, str]: await self._swap_sinks(new_config) self._config = new_config self._stamp_config_hash(new_config) + # Pick up a changed concurrency cap without dropping the live + # running count (lowering it just admits fewer new runs until + # the count drains; raising it wakes any queued waiters). + self._load_shedder.set_max_concurrent(new_config.safety.max_concurrent_groups) if self._scheduler is not None: self._reschedule_jobs(self._scheduler) # Trigger an immediate run for any groups that may have @@ -259,34 +304,46 @@ def _reschedule_jobs(self, scheduler: AsyncIOScheduler) -> None: # but a full rebuild keeps reload semantics simple and avoids # subtle drift between the live job set and the new config. scheduler.remove_all_jobs() - interval = self._config.server.refresh_interval_seconds - # APScheduler's IntervalTrigger defers the first run by one - # full interval; the immediate first run is fired from startup() - # separately so /metrics has content before that delay. + # Per-group refresh cadence: a group's own ``refresh_interval`` + # wins, else ``server.refresh_interval``. The per-group lock + # already prevents overlap; APScheduler's IntervalTrigger defers + # the first run by one full interval, so the immediate first run + # is fired from startup() separately for /metrics warmth. for group in self._config.groups: scheduler.add_job( self._tick, - trigger=IntervalTrigger(seconds=interval), + trigger=IntervalTrigger(seconds=self._config.group_refresh_seconds(group)), args=[group.name], id=f"group:{group.name}", replace_existing=True, max_instances=1, coalesce=True, ) - # End-to-end detection self-test, on the same cadence as the - # detector runs. Gated inside ``_selftest_tick`` by the leader flag - # like ``_tick``, so it fires on every replica but only the leader - # runs it in HA mode. + # End-to-end detection self-test, on the server cadence. Gated + # inside ``_selftest_tick`` by the leader flag like ``_tick``, so + # it fires on every replica but only the leader runs it in HA mode. if self._config.server.selftest.enabled: scheduler.add_job( self._selftest_tick, - trigger=IntervalTrigger(seconds=interval), + trigger=IntervalTrigger(seconds=self._config.server.refresh_interval_seconds), id="selftest", replace_existing=True, max_instances=1, coalesce=True, ) + def _on_job_max_instances(self, event: Any) -> None: + # APScheduler emits EVENT_JOB_MAX_INSTANCES when a group's previous + # run is still in flight at the next fire time (we run each group + # with max_instances=1). That overrun is a real degraded signal — + # the group can't keep up with its cadence — so it is counted with + # reason="overrun" rather than letting the coalesce vanish silently. + job_id = getattr(event, "job_id", "") or "" + if not job_id.startswith("group:"): + return + group_name = job_id[len("group:") :] + self._ops.group_skipped_total.labels(group=group_name, reason="overrun").inc() + async def _tick(self, group_name: str) -> None: # In HA mode followers must never run the detector pipeline — # only the elected leader writes snapshots. Polling the leader @@ -296,7 +353,48 @@ async def _tick(self, group_name: str) -> None: # behaviour identical when the leader changes mid-flight. if self._ha_enabled and not self._is_leader: return - await self._safe_group_run(group_name) + await self._admit_and_run(group_name) + + async def _admit_and_run(self, group_name: str) -> None: + """Run a group under the priority-aware admission gate. + + Uncapped (``safety.max_concurrent_groups`` unset) this is a + straight pass-through to ``_safe_group_run``. Capped, the gate + admits at most ``max_concurrent_groups`` runs at once, prefers + higher-``priority`` groups when saturated, and sheds (skips) a run + that can't get a slot within its own refresh interval — counted on + ``anomaly_group_skipped_total{reason="backpressure"}``. + """ + if self._load_shedder.max_concurrent is None: + await self._safe_group_run(group_name) + return + group = self._group_by_name(group_name) + priority = group.priority if group is not None else 1 + # Admission deadline is half the refresh interval, not a full one, + # so a shed resolves *before* the group's next tick fires. A + # full-interval wait would still be blocking when APScheduler + # tries to fire again (max_instances=1), tripping a spurious + # EVENT_JOB_MAX_INSTANCES (reason="overrun") on top of the + # backpressure shed and double-reporting one root cause. Halving + # keeps overrun ("the run is too slow") and backpressure + # ("admission is saturated") as distinct, non-overlapping signals. + refresh = self._config.group_refresh_seconds(group) if group is not None else 0.0 + deadline = refresh / 2.0 + admitted = await self._load_shedder.acquire(priority=priority, deadline_seconds=deadline) + if not admitted: + self._ops.group_skipped_total.labels(group=group_name, reason="backpressure").inc() + logger.info("group_run_shed", group=group_name, reason="backpressure") + return + try: + await self._safe_group_run(group_name) + finally: + self._load_shedder.release() + + def _group_by_name(self, name: str) -> Any | None: + for group in self._config.groups: + if group.name == name: + return group + return None async def _safe_group_run(self, group_name: str) -> None: # Register the task so a leadership loss can cancel it. We do @@ -600,6 +698,21 @@ def _build_query_cache( ) +def _resolve_shard(config: Config) -> ShardAssignment | None: + """Build this replica's shard assignment, or ``None`` when scale-out is off. + + Returns ``None`` unless ``scaleOut.enabled`` (so the runner's shard + filter is skipped entirely in the common single-replica case). When + enabled, the index comes from ``scaleOut.shard_index`` or the + StatefulSet pod ordinal; an out-of-range resolution raises at boot. + """ + scale_out = config.scaleOut + if not scale_out.enabled: + return None + index = resolve_shard_index(scale_out.shard_index, shards=scale_out.shards) + return ShardAssignment(index=index, count=scale_out.shards) + + def _maybe_build_redis(config: Config) -> Any | None: """Construct a Redis client iff any consumer in the config needs one. @@ -652,10 +765,11 @@ def serve(config_path: str | Path) -> None: api_version=config.apiVersion, replacement=CURRENT_API_VERSION, ) - if config.aggressive_refresh: + for group_name, effective in config.aggressive_refresh_groups(): logger.warning( "aggressive_refresh_interval", - refresh_interval_seconds=config.server.refresh_interval_seconds, + group=group_name, + refresh_interval_seconds=effective, window_seconds=config.defaults.window_seconds, recommended_minimum_seconds=config.defaults.window_seconds / 6.0, note="refresh_interval below window/6 — rolling baseline lags", diff --git a/detector/src/promanomaly/runner.py b/detector/src/promanomaly/runner.py index 37a672a..83f0401 100644 --- a/detector/src/promanomaly/runner.py +++ b/detector/src/promanomaly/runner.py @@ -68,6 +68,7 @@ from .selector import select_winner from .self_observability import estimate_samples_bytes from .severity import compute_severity +from .sharding import ShardAssignment from .source import PromQLSource, QueryResult, QuerySeries, SourceQueryError from .state import GroupSnapshot, Sample, SnapshotStore from .stratified import ( @@ -222,11 +223,16 @@ def __init__( *, executor: concurrent.futures.ThreadPoolExecutor | None = None, telemetry: Telemetry | None = None, + shard: ShardAssignment | None = None, ) -> None: self._config = config self._source = source self._store = store self._ops = operational + # Active/active scale-out: when set, only series whose key hashes + # to this shard are scored. ``None`` (single-shard / scale-out + # off) owns everything, so the per-series filter is a no-op. + self._shard = shard # OpenTelemetry tracing is optional and zero-overhead when disabled. # ``Telemetry.disabled()`` returns a singleton with no-op spans so # callers can ``with self._telemetry.span(...)`` unconditionally. @@ -367,6 +373,11 @@ def replace_config(self, new_config: Config) -> None: self._group_locks.pop(removed, None) self._recurrence_cache.pop(removed, None) self._recurrence_last_run.pop(removed, None) + # Drop the sticky series count too: leaving it behind would + # permanently subtract a phantom count from the cross-group + # max_total_series budget (shrinking it for surviving groups) + # and inflate anomaly_shard_series_count until restart. + self._last_series_count.pop(removed, None) for added in new_names - old_names: self._group_locks[added] = asyncio.Lock() # Bound the rolling detector-success ledger to live groups so a @@ -497,12 +508,12 @@ async def run_group(self, group_name: str) -> GroupRunResult: elif result.succeeded: self._store.reset_source_failure_streak(group_name) + # Updates the sticky per-group series count (drives the + # cross-group cap) and the operational gauges. Only the + # success path writes the sticky count, so a failed run never + # silently shrinks the budget for other groups. self._update_operational_metrics(result) if result.succeeded: - # Sticky series count drives the cross-group cap on the - # next pass. Only update on success — failed runs must - # not silently shrink the budget for other groups. - self._last_series_count[group_name] = result.series_count self._store.write( GroupSnapshot( group=group_name, @@ -545,7 +556,16 @@ async def _run_group_inner(self, group: GroupConfig) -> _GroupInnerResult: # to the snapshot so followers serve it and serve_stale keeps the # last good value during source outages, consistent with every # other snapshot metric. - out.samples.extend(self._density_samples(group, out.samples)) + # + # Suppressed in sharded mode: each shard sees only its slice of + # the group's entities, so an in-process density would be a + # per-shard fraction — partial *and* emitted once per shard, + # violating the "sharding never multiplies series" contract. + # Operators compute true fleet density via a recording rule over + # the per-series anomaly_outside_threshold (the shard union is + # complete). See docs/operations.md. + if not self._is_sharded: + out.samples.extend(self._density_samples(group, out.samples)) # Recurrence pattern analysis. Opt-in per group; queries the # TSDB for historical anomaly_outside_threshold and emits a 7x24 @@ -554,7 +574,13 @@ async def _run_group_inner(self, group: GroupConfig) -> _GroupInnerResult: # 1-minute group refresh. Between refreshes, the last-good # recurrence samples are carried forward in the snapshot via # _recurrence_cache. - if group.recurrence.enabled: + # + # In sharded mode only the group's owner shard emits it: the + # analysis re-reads the *complete* history from the TSDB (every + # shard's scores land there), so it is correct on any shard, but + # emitting from all shards would multiply the heatmap series. + # Gating to one shard keeps a single, correct copy. + if group.recurrence.enabled and self._shard_owns_group(group.name): now = time.time() last = self._recurrence_last_run.get(group.name, 0.0) interval = group.recurrence.refresh_interval_seconds @@ -663,7 +689,12 @@ async def _process_discovered_query( seen_keys: list[tuple[tuple[str, str], ...]] = [] for variant in variants: - seen_keys.append(variant.label_key) + # Track absence only for variants this shard owns, so the + # per-shard discovery tracker scopes itself to its slice and + # ``anomaly_signal_missing`` is emitted by exactly one shard. + # Scoring is still sharded per series inside the variant pass. + if self._shard_owns_key(variant.label_key): + seen_keys.append(variant.label_key) await self._process_query_variant( group=group, query=query, @@ -765,8 +796,47 @@ def surface_misconfigurations(self) -> None: use that detector — we don't want to double-log here). """ self._check_cohort_with_auto_select() + self._check_cohort_with_scaleout() self._check_stratified_findings() + def _check_cohort_with_scaleout(self) -> None: + """Cohort detector under active/active sharding -> split cohort. + + Sharding partitions by series key, so a cohort's members hash to + different shards and each shard's Cohort detector sees only its + own slice of the population — a wrong baseline. Warn (don't fail) + so the operator moves cohort groups to an unsharded deployment or + accepts the partitioned comparison knowingly. Only fires when + this replica is genuinely sharded (``count > 1``). + """ + if not self._is_sharded: + return + assert self._shard is not None # narrowed by _is_sharded + for group in self._config.groups: + for query in group.queries: + for entry in query.detectors: + try: + cls = get(entry.name) + except UnknownDetectorError: + continue + if is_cohort_aware(cls): + logger.warning( + "cohort_with_scaleout", + group=group.name, + query=query.id, + detector=entry.name, + instance=entry.instance, + shards=self._shard.count, + note=( + "Cohort-aware detectors need cross-member " + "visibility, but scaleOut sharding splits a " + "cohort's members across shards so each shard " + "scores against a partial population. Run " + "cohort groups in a separate unsharded " + "deployment, or accept the per-shard comparison." + ), + ) + def _check_cohort_with_auto_select(self) -> None: """Cohort + ``auto_select: true`` -> silently-suppressed output. @@ -941,7 +1011,6 @@ async def _process_query_variant( qresult, plans = await self._fetch_and_plan( group, effective_query, - inflight_series_count=out.series_count, template_query=query, ) except SourceQueryError as exc: @@ -991,7 +1060,12 @@ async def _process_query_variant( out.dropped[REASON_CARDINALITY] = out.dropped.get(REASON_CARDINALITY, 0) + 1 self._ops.series_dropped_total.labels(group=group.name, reason=REASON_CARDINALITY).inc() - if effective_query.expect: + # Whole-query ``expect`` absence is evaluated against the full + # fetch (so every shard agrees the query did/didn't return data), + # but emitted by only the group's owner shard. ``anomaly_signal_missing`` + # carries the (id, group) join labels, so a per-shard duplicate + # would break the "sharding never multiplies series" contract. + if effective_query.expect and self._shard_owns_group(group.name): misses = self._store.record_expect_observation( group.name, effective_query.id, has_results=bool(qresult.series) ) @@ -1008,13 +1082,26 @@ async def _process_query_variant( ) ) + # Active/active scale-out: keep only the series this shard owns, + # then enforce max_total_series against that owned slice so the + # cap is per-shard (estate ceiling = max_total_series x shards). + # Sharding before the cap — not after — is what makes the cap + # per-shard rather than an estate-wide cap split N ways. + owned_series = self._shard_owned_series(qresult.series) + scored_series, capped = self._cap_to_total_series_budget( + group.name, owned_series, inflight=out.series_count + ) + if capped: + out.dropped[REASON_CARDINALITY] = out.dropped.get(REASON_CARDINALITY, 0) + 1 + self._ops.series_dropped_total.labels(group=group.name, reason=REASON_CARDINALITY).inc() + # Cohort detectors need cross-member visibility. Build the # per-cohort baseline once here (rather than re-deriving it per # member) so every member sees the same numbers. Cheap no-op # when no plan in the query opts into ``cohort_aware``. - cohort_context = build_cohort_context(plans, qresult.series) + cohort_context = build_cohort_context(plans, scored_series) - for series in qresult.series: + for series in scored_series: out.series_count += 1 produced = await self._score_series( group=group, @@ -1026,12 +1113,75 @@ async def _process_query_variant( ) out.samples.extend(produced) + @property + def _is_sharded(self) -> bool: + """True when this replica scores a strict subset of the estate.""" + return self._shard is not None and self._shard.count > 1 + + def _shard_owned_series(self, series: list[QuerySeries]) -> list[QuerySeries]: + """Filter ``series`` to those this shard owns under scale-out. + + Returns the input unchanged when sharding is off or single-shard. + Ownership is decided by rendezvous hashing on the series label + key, so every replica partitions the same fetched series into + disjoint, non-overlapping slices that union to the whole estate. + """ + if not self._is_sharded: + return series + assert self._shard is not None # narrowed by _is_sharded + return [s for s in series if self._shard.owns(s.label_key)] + + def _shard_owns_group(self, group_name: str) -> bool: + """Whether this shard owns a group's whole-group emitted series. + + Group-level signals that are not per-input-series — recurrence + heatmaps, and whole-query ``expect`` absence — must be emitted by + exactly one shard so the partition never multiplies a series. The + owner is chosen by hashing a synthetic per-group key, which + spreads ownership across shards rather than piling every group's + aggregates on shard 0. Always True when not sharded. + """ + if not self._is_sharded: + return True + assert self._shard is not None # narrowed by _is_sharded + return self._shard.owns((("__group__", group_name),)) + + def _shard_owns_key(self, key: tuple[tuple[str, str], ...]) -> bool: + """Whether this shard owns an arbitrary series/variant label key.""" + if not self._is_sharded: + return True + assert self._shard is not None # narrowed by _is_sharded + return self._shard.owns(key) + + def _cap_to_total_series_budget( + self, group_name: str, series: list[QuerySeries], *, inflight: int + ) -> tuple[list[QuerySeries], bool]: + """Trim ``series`` to the remaining ``max_total_series`` budget. + + Applied to the shard-owned series so the cap is enforced **per + shard**: each replica scores up to ``max_total_series`` of its own + slice, making the estate ceiling ``max_total_series x shards`` (as + documented for scale-out). With no sharding the owned set is the + whole fetch, so behaviour is identical to the single-replica cap. + The cross-group budget reads other groups' sticky owned counts + plus this group's in-flight owned count. + """ + cap = self._config.safety.max_total_series + if not cap: + return series, False + other_groups_total = sum( + count for name, count in self._last_series_count.items() if name != group_name + ) + running_budget = max(0, cap - other_groups_total - inflight) + if len(series) > running_budget: + return series[:running_budget], True + return series, False + async def _fetch_and_plan( self, group: GroupConfig, query: QueryConfig, *, - inflight_series_count: int = 0, template_query: QueryConfig | None = None, ) -> tuple[QueryResult, list[_DetectorPlan]]: defaults = self._config.defaults @@ -1086,20 +1236,12 @@ async def _fetch_and_plan( max_series=self._config.safety.max_series_per_query, ) - # Enforce max_total_series across all groups + queries in the - # current pass. Each group runs in its own scheduler tick, so - # "total so far" reads from: - # - the latest successful series counts of *other* groups - # (sticky in self._last_series_count), plus - # - whatever this group has already accumulated in this pass. - cap = self._config.safety.max_total_series - if cap: - other_groups_total = sum( - count for name, count in self._last_series_count.items() if name != group.name - ) - running_budget = max(0, cap - other_groups_total - inflight_series_count) - if len(result.series) > running_budget: - result = QueryResult(series=result.series[:running_budget], truncated=True) + # The cross-group max_total_series cap is NOT applied here: it is + # enforced in _process_query_variant *after* the shard filter, so + # the cap counts only the series this shard actually scores (a + # per-shard cap). _fetch_and_plan returns the full fetch so the + # caller can still evaluate whole-query signals (expect: absence) + # against the complete result before sharding. return result, plans def _merge_params( @@ -2028,6 +2170,20 @@ def _update_operational_metrics(self, result: GroupRunResult) -> None: self._ops.group_memory_bytes.labels(group=result.group).set( estimate_samples_bytes(result.samples) ) + # Canonical write of the sticky per-group series count (drives + # the cross-group max_total_series cap and the shard gauge + # below). Must precede the shard-series-count sum so the gauge + # reflects this run. + self._last_series_count[result.group] = result.series_count + # Scale-out: expose how many series this shard scored across all + # groups so an operator can confirm the partition is even. Set + # only in sharded mode — a single-shard replica owns everything, + # so the gauge would just mirror the global series total. + if self._is_sharded: + assert self._shard is not None # narrowed by _is_sharded + self._ops.shard_series_count.labels(shard=str(self._shard.index)).set( + float(sum(self._last_series_count.values())) + ) # Per-kind log emitters for stratified misconfigurations. Each diff --git a/detector/src/promanomaly/scheduling.py b/detector/src/promanomaly/scheduling.py new file mode 100644 index 0000000..b26c464 --- /dev/null +++ b/detector/src/promanomaly/scheduling.py @@ -0,0 +1,174 @@ +"""Priority-aware admission control for group scheduling. + +By default the scheduler runs every group on its own cadence with no +admission control — the historical behaviour, preserved exactly when +``safety.max_concurrent_groups`` is unset. When the operator sets that +cap, this gate bounds how many group runs execute at once and protects +the most important groups under TSDB backpressure: + +* **Priority ordering.** When the gate is saturated, pending runs wait + on a priority-ordered queue; a freed slot wakes the highest-``priority`` + waiter first, so a slow source delays low-priority signals before the + ones the operator cares most about. +* **Load shedding.** A run that cannot acquire a slot within its own + refresh interval is *skipped*, not queued unboundedly — the next tick + will try again with fresh data. The skip is surfaced on + ``anomaly_group_skipped_total{group, reason="backpressure"}`` and never + hidden: shedding delays a score, it never suppresses a detected + anomaly (the last good snapshot stays in place per serve-stale). + +This is strictly a *scheduling* concern, not a *silencing* one. Overrun +(a group's previous run still in flight when the next tick fires) is +surfaced separately by the APScheduler ``max_instances`` listener in +:mod:`promanomaly.main` as ``reason="overrun"``. +""" + +from __future__ import annotations + +import asyncio +import heapq +import itertools +from dataclasses import dataclass, field + + +@dataclass(order=True) +class _Waiter: + # Heap ordering: higher priority first (negated), then FIFO by + # insertion sequence so equal-priority groups keep arrival order. + sort_key: tuple[int, int] + future: asyncio.Future[None] = field(compare=False) + + +class LoadShedder: + """Bounded, priority-ordered admission gate for group runs. + + Thread-confined to the event loop (all methods are called from + scheduler tasks on the same loop), so the running count and waiter + heap need no locking. ``max_concurrent=None`` makes every method a + pass-through with zero overhead and no behavioural change. + """ + + def __init__(self, max_concurrent: int | None) -> None: + self._max_concurrent = max_concurrent + self._running = 0 + self._waiters: list[_Waiter] = [] + self._seq = itertools.count() + + @property + def max_concurrent(self) -> int | None: + return self._max_concurrent + + def set_max_concurrent(self, value: int | None) -> None: + """Update the cap on reload, preserving the live running count. + + Lowering the cap doesn't preempt in-flight runs; it just admits + fewer new ones until the count drains below the new bound. + Raising it wakes any waiters that now fit. Removing it (``None``) + is "admit everyone" — every queued waiter is released immediately + rather than left to time out and shed against a cap that no + longer exists. + """ + self._max_concurrent = value + if value is None: + # Uncapped: drain the whole queue. Each woken waiter takes a + # slot in the running count so the eventual release() (which + # decrements directly in the uncapped branch) stays balanced. + while self._wake_one(): + self._running += 1 + else: + self._drain_waiters() + + @property + def running(self) -> int: + return self._running + + async def acquire(self, *, priority: int, deadline_seconds: float) -> bool: + """Acquire a concurrency slot, or return ``False`` if shed. + + Returns ``True`` immediately when uncapped or a slot is free. + When saturated, waits up to ``deadline_seconds`` for a slot, + woken in priority order; returns ``False`` (shed) if the deadline + passes first. The caller MUST call :meth:`release` iff this + returned ``True``. + """ + if self._max_concurrent is None: + self._running += 1 + return True + if self._running < self._max_concurrent: + self._running += 1 + return True + # Saturated: queue this run, highest priority served first. + loop = asyncio.get_running_loop() + future: asyncio.Future[None] = loop.create_future() + waiter = _Waiter(sort_key=(-priority, next(self._seq)), future=future) + heapq.heappush(self._waiters, waiter) + try: + await asyncio.wait_for(future, timeout=max(0.0, deadline_seconds)) + except TimeoutError: + # Shed: no slot freed within the deadline. The waiter may have + # been popped-and-resolved by a concurrent release in the race + # window; if so, hand the slot straight back. + self._abandon_waiter(waiter, future) + return False + except asyncio.CancelledError: + # Leadership loss / shutdown cancelled us — release any slot + # that raced in and let the cancellation propagate so the task + # unwinds cleanly rather than being treated as a shed. + self._abandon_waiter(waiter, future) + raise + # Woken by release(): the slot was already accounted to us there. + return True + + def _abandon_waiter(self, waiter: _Waiter, future: asyncio.Future[None]) -> None: + """Give back a slot won in the timeout/cancel race, else drop the waiter.""" + if future.done() and not future.cancelled(): + self.release() + else: + self._discard_waiter(waiter) + + def release(self) -> None: + """Release a held slot and wake the next-highest-priority waiter.""" + if self._max_concurrent is None: + self._running = max(0, self._running - 1) + return + # Hand the slot directly to the best waiter (count stays put) so + # there is no admission gap a lower-priority newcomer could slip + # through. Only decrement when nobody is waiting. + woke = self._wake_one() + if not woke: + self._running = max(0, self._running - 1) + + def _wake_one(self) -> bool: + while self._waiters: + waiter = heapq.heappop(self._waiters) + if waiter.future.done(): + continue # already timed out / cancelled + waiter.future.set_result(None) + return True + return False + + def _drain_waiters(self) -> None: + # Admit as many queued waiters as the (possibly raised) cap allows. + while ( + self._max_concurrent is not None + and self._running < self._max_concurrent + and self._wake_one_into_slot() + ): + pass + + def _wake_one_into_slot(self) -> bool: + if self._wake_one(): + self._running += 1 + return True + return False + + def _discard_waiter(self, waiter: _Waiter) -> None: + waiter.future.cancel() + try: + self._waiters.remove(waiter) + heapq.heapify(self._waiters) + except ValueError: + pass + + +__all__ = ["LoadShedder"] diff --git a/detector/src/promanomaly/sharding.py b/detector/src/promanomaly/sharding.py new file mode 100644 index 0000000..55cc759 --- /dev/null +++ b/detector/src/promanomaly/sharding.py @@ -0,0 +1,137 @@ +"""Consistent-hash series sharding for active/active scale-out. + +In scale-out mode (``scaleOut.enabled``) ``N`` replicas each own a +disjoint slice of the series and run concurrently. Ownership is decided +by **rendezvous hashing** (a.k.a. highest-random-weight, HRW): for a +given series key every shard computes a deterministic weight and the +highest-weighted shard owns the series. Compared with plain +``hash(key) % shards`` this keeps resharding on a shard-count change to +roughly ``1/shards`` of the keys moving — the "minimal movement" +property the roadmap asks for — without a ring or virtual nodes. + +The hash must be **stable across processes**: Python's built-in ``hash`` +is salted per interpreter (``PYTHONHASHSEED``), so two replicas would +disagree on ownership. We use BLAKE2b over the canonical series-key +bytes instead, which every replica computes identically. + +Sharding is a *partition* of the existing work, never a multiplier: +each series is owned by exactly one shard, so the union of the shards' +output is the whole estate with no duplication and no new cardinality. +""" + +from __future__ import annotations + +import hashlib +import os +import re +from dataclasses import dataclass + +# A StatefulSet pod is named ``-``; the trailing +# integer is the stable shard index. Matches the longest trailing run of +# digits so a release name containing digits (``promanomaly2-0``) still +# resolves the ordinal correctly. +_ORDINAL_RE = re.compile(r"-(\d+)$") + +# Series labels travel as a sorted tuple of (name, value) pairs +# (``QuerySeries.label_key``). The canonical byte encoding joins them +# with control characters that cannot appear in a Prometheus label name +# or value, so two distinct label sets can never collide on the same +# bytes. +_PAIR_SEP = "\x1f" # unit separator, between name and value +_LABEL_SEP = "\x1e" # record separator, between pairs + + +def _key_bytes(series_key: tuple[tuple[str, str], ...]) -> bytes: + """Canonical, process-stable byte encoding of a series label key.""" + return _LABEL_SEP.join(f"{name}{_PAIR_SEP}{value}" for name, value in series_key).encode( + "utf-8" + ) + + +def _weight(key_bytes: bytes, shard_index: int) -> int: + """Rendezvous weight of ``shard_index`` for a series, as a 64-bit int. + + BLAKE2b is keyed by the shard index so the per-(series, shard) + weights are independent draws; the highest wins. Deterministic and + identical on every replica. + """ + digest = hashlib.blake2b(key_bytes, digest_size=8, salt=shard_index.to_bytes(8, "big")).digest() + return int.from_bytes(digest, "big") + + +@dataclass(frozen=True) +class ShardAssignment: + """This replica's place in the shard ring. + + ``index`` is the shard this replica owns (``0 <= index < count``); + ``count`` is the total number of shards. A single-shard assignment + (``count == 1``) owns everything, so the hashing fast-path short + circuits to ``True``. + """ + + index: int + count: int + + def __post_init__(self) -> None: + if self.count < 1: + raise ValueError(f"shard count must be >= 1, got {self.count}") + if not 0 <= self.index < self.count: + raise ValueError(f"shard index {self.index} out of range for count {self.count}") + + def owns(self, series_key: tuple[tuple[str, str], ...]) -> bool: + """Return whether this shard owns ``series_key`` under HRW hashing. + + Ties (identical weights — astronomically unlikely with a 64-bit + digest) break toward the lower shard index so every replica + resolves them identically. + """ + if self.count == 1: + return True + key_bytes = _key_bytes(series_key) + best_index = 0 + best_weight = _weight(key_bytes, 0) + for candidate in range(1, self.count): + weight = _weight(key_bytes, candidate) + if weight > best_weight: + best_weight = weight + best_index = candidate + return best_index == self.index + + +def resolve_shard_index(configured: int | None, *, shards: int) -> int: + """Pick this replica's shard index. + + Order: explicit ``scaleOut.shard_index`` > the StatefulSet pod + ordinal parsed from ``POD_NAME`` (the trailing ``-N``) > ``0``. The + chart injects ``POD_NAME`` via the Downward API for every replica, so + a StatefulSet's pods self-assign ``0..shards-1`` with no operator + action. Raises ``ValueError`` when the resolved index falls outside + ``[0, shards)`` so a misconfiguration fails loudly at boot rather + than silently scoring the wrong slice. + """ + index = configured + if index is None: + index = _ordinal_from_env() + if index is None: + index = 0 + if not 0 <= index < shards: + raise ValueError( + f"resolved shard index {index} out of range for scaleOut.shards={shards}; " + "set scaleOut.shard_index explicitly or run under a StatefulSet so POD_NAME " + "carries a valid ordinal" + ) + return index + + +def _ordinal_from_env() -> int | None: + """Parse the StatefulSet ordinal from ``POD_NAME``, or ``None``.""" + pod_name = os.environ.get("POD_NAME") + if not pod_name: + return None + match = _ORDINAL_RE.search(pod_name) + if match is None: + return None + return int(match.group(1)) + + +__all__ = ["ShardAssignment", "resolve_shard_index"] diff --git a/detector/tests/test_horizontal_scaleout.py b/detector/tests/test_horizontal_scaleout.py new file mode 100644 index 0000000..824bd93 --- /dev/null +++ b/detector/tests/test_horizontal_scaleout.py @@ -0,0 +1,661 @@ +"""Horizontal scale-out and resilience. + +Covers three orthogonal capabilities: + +* Active/active sharding via consistent (rendezvous) hashing. +* Per-group refresh intervals. +* Priority-aware scheduling and load shedding. +""" + +from __future__ import annotations + +import asyncio +import types +from pathlib import Path +from typing import Any + +import pytest +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from pydantic import ValidationError + +from promanomaly.cli._cost import estimate_cost +from promanomaly.config import CURRENT_API_VERSION, Config +from promanomaly.exporter import OperationalMetrics +from promanomaly.main import Application +from promanomaly.runner import Runner +from promanomaly.scheduling import LoadShedder +from promanomaly.sharding import ShardAssignment, resolve_shard_index +from promanomaly.state import SnapshotStore +from tests.conftest import StubResponse, StubSource, make_series +from tests.fixtures import clean_baseline + + +def _metric_value(ops: OperationalMetrics, name: str, labels: dict[str, str]) -> float: + value = ops.registry.get_sample_value(name, labels) + return float(value) if value is not None else 0.0 + + +# ── Config: scale-out, per-group refresh, max_concurrent_groups ────── + + +def _base_config(**overrides: Any) -> dict[str, Any]: + raw: dict[str, Any] = { + "apiVersion": CURRENT_API_VERSION, + "datasource": {"url": "http://vm:8428/"}, + "groups": [ + { + "name": "g1", + "queries": [{"id": "metric_a", "promql": "up", "detectors": [{"name": "MAD"}]}], + } + ], + } + raw.update(overrides) + return raw + + +class TestScaleOutConfig: + def test_defaults_off(self) -> None: + cfg = Config.model_validate(_base_config()) + assert cfg.scaleOut.enabled is False + assert cfg.scaleOut.shards == 1 + assert cfg.scaleOut.shard_index is None + + def test_shard_index_must_be_below_shards(self) -> None: + with pytest.raises(Exception, match=r"shard_index=2 must be < scaleOut.shards=2"): + Config.model_validate( + _base_config(scaleOut={"enabled": True, "shards": 2, "shard_index": 2}) + ) + + def test_scaleout_and_ha_mutually_exclusive(self) -> None: + with pytest.raises(Exception, match=r"mutually exclusive"): + Config.model_validate( + _base_config( + scaleOut={"enabled": True, "shards": 2}, + highAvailability={"enabled": True}, + safety={"redis": {"url": "redis://localhost:6379/0"}}, + ) + ) + + def test_scaleout_enabled_validates(self) -> None: + cfg = Config.model_validate( + _base_config(scaleOut={"enabled": True, "shards": 4, "shard_index": 1}) + ) + assert cfg.scaleOut.enabled + assert cfg.scaleOut.shards == 4 + assert cfg.scaleOut.shard_index == 1 + + +class TestPerGroupRefresh: + def test_group_override_takes_precedence(self) -> None: + raw = _base_config(server={"refresh_interval": "1m"}) + raw["groups"][0]["refresh_interval"] = "30s" + cfg = Config.model_validate(raw) + assert cfg.group_refresh_seconds(cfg.groups[0]) == 30.0 + + def test_falls_back_to_server_default(self) -> None: + cfg = Config.model_validate(_base_config(server={"refresh_interval": "90s"})) + assert cfg.groups[0].refresh_interval_override_seconds is None + assert cfg.group_refresh_seconds(cfg.groups[0]) == 90.0 + + def test_aggressive_refresh_groups_flags_per_group_override(self) -> None: + # window=1h → floor = 600s. A 30s per-group override is aggressive. + raw = _base_config( + server={"refresh_interval": "10m"}, + defaults={"window": "1h"}, + ) + raw["groups"][0]["refresh_interval"] = "30s" + cfg = Config.model_validate(raw) + offending = cfg.aggressive_refresh_groups() + assert offending == [("g1", 30.0)] + + def test_aggressive_refresh_groups_empty_when_within_floor(self) -> None: + cfg = Config.model_validate( + _base_config(server={"refresh_interval": "10m"}, defaults={"window": "1h"}) + ) + assert cfg.aggressive_refresh_groups() == [] + + +class TestMaxConcurrentConfig: + def test_defaults_to_none(self) -> None: + cfg = Config.model_validate(_base_config()) + assert cfg.safety.max_concurrent_groups is None + + def test_accepts_positive_int(self) -> None: + cfg = Config.model_validate(_base_config(safety={"max_concurrent_groups": 2})) + assert cfg.safety.max_concurrent_groups == 2 + + def test_rejects_zero(self) -> None: + with pytest.raises(ValidationError): + Config.model_validate(_base_config(safety={"max_concurrent_groups": 0})) + + +# ── Sharding: rendezvous hashing ───────────────────────────────────── + + +def _key(instance: str) -> tuple[tuple[str, str], ...]: + return (("instance", instance),) + + +class TestShardAssignment: + def test_single_shard_owns_everything(self) -> None: + shard = ShardAssignment(index=0, count=1) + assert all(shard.owns(_key(f"host-{i}")) for i in range(20)) + + def test_partition_is_disjoint_and_covers_all(self) -> None: + keys = [_key(f"host-{i}") for i in range(50)] + shards = [ShardAssignment(index=i, count=3) for i in range(3)] + owned: list[set[int]] = [] + for shard in shards: + owned.append({i for i, k in enumerate(keys) if shard.owns(k)}) + # Every key owned by exactly one shard. + union: set[int] = set().union(*owned) + assert union == set(range(50)) + assert sum(len(o) for o in owned) == 50 # disjoint + + def test_ownership_is_deterministic(self) -> None: + shard_a = ShardAssignment(index=2, count=5) + shard_b = ShardAssignment(index=2, count=5) + key = _key("host-xyz") + assert shard_a.owns(key) == shard_b.owns(key) + + def test_minimal_movement_on_resize(self) -> None: + # Rendezvous hashing: growing the ring should move far fewer keys + # than a modulo scheme (which reshuffles almost everything). + keys = [_key(f"host-{i}") for i in range(300)] + + def owner(count: int, key: tuple[tuple[str, str], ...]) -> int: + for i in range(count): + if ShardAssignment(index=i, count=count).owns(key): + return i + raise AssertionError("no owner") + + before = {i: owner(3, k) for i, k in enumerate(keys)} + after = {i: owner(4, k) for i, k in enumerate(keys)} + moved = sum(1 for i in before if before[i] != after[i]) + # Ideal HRW reassigns ~1/4 of keys when going 3→4; allow slack. + assert moved < len(keys) * 0.5 + + def test_rejects_out_of_range_index(self) -> None: + with pytest.raises(ValueError): + ShardAssignment(index=3, count=3) + + +class TestResolveShardIndex: + def test_explicit_index_wins(self) -> None: + assert resolve_shard_index(2, shards=4) == 2 + + def test_reads_statefulset_ordinal_from_pod_name(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("POD_NAME", "promanomaly-2") + assert resolve_shard_index(None, shards=4) == 2 + + def test_defaults_to_zero_without_pod_name(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("POD_NAME", raising=False) + assert resolve_shard_index(None, shards=1) == 0 + + def test_out_of_range_ordinal_raises(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("POD_NAME", "promanomaly-9") + with pytest.raises(ValueError, match=r"out of range"): + resolve_shard_index(None, shards=4) + + +# ── Runner: shard filter scores only owned series ──────────────────── + + +def _runner_config(shards: int = 2, *, max_total_series: int | None = None) -> Config: + raw: dict[str, Any] = { + "apiVersion": CURRENT_API_VERSION, + "datasource": {"url": "http://stub"}, + "defaults": {"window": "15m", "step": "15s", "min_points": 30}, + "groups": [ + {"name": "g1", "queries": [{"id": "m", "promql": "up", "detectors": [{"name": "MAD"}]}]} + ], + } + if shards: + raw["scaleOut"] = {"enabled": True, "shards": shards} + if max_total_series is not None: + raw["safety"] = {"max_total_series": max_total_series} + return Config.model_validate(raw) + + +def _series_set(n: int) -> list[Any]: + df = clean_baseline(n=120) + return [make_series({"instance": f"host-{i}"}, df) for i in range(n)] + + +@pytest.mark.asyncio +async def test_shards_partition_series(stub_source: StubSource) -> None: + n = 40 + stub_source.respond(lambda _: StubResponse(series=_series_set(n))) + cfg = _runner_config(shards=2) + + counts = [] + owned_ids: set[str] = set() + for index in range(2): + store = SnapshotStore() + ops = OperationalMetrics() + runner = Runner(cfg, stub_source, store, ops, shard=ShardAssignment(index, 2)) + result = await runner.run_group("g1") + assert result.succeeded + counts.append(result.series_count) + ids = { + dict(s.labels).get("instance") for s in result.samples if s.metric == "anomaly_score" + } + owned_ids |= {i for i in ids if i is not None} + + # Disjoint partition that together covers every input series. + assert sum(counts) == n + assert len(owned_ids) == n + + +@pytest.mark.asyncio +async def test_no_shard_scores_everything(stub_source: StubSource) -> None: + stub_source.respond(lambda _: StubResponse(series=_series_set(10))) + cfg = _runner_config(shards=0) + store = SnapshotStore() + ops = OperationalMetrics() + runner = Runner(cfg, stub_source, store, ops, shard=None) + result = await runner.run_group("g1") + assert result.series_count == 10 + + +@pytest.mark.asyncio +async def test_shard_series_count_metric_emitted(stub_source: StubSource) -> None: + stub_source.respond(lambda _: StubResponse(series=_series_set(30))) + cfg = _runner_config(shards=2) + store = SnapshotStore() + ops = OperationalMetrics() + runner = Runner(cfg, stub_source, store, ops, shard=ShardAssignment(0, 2)) + result = await runner.run_group("g1") + gauge = _metric_value(ops, "anomaly_shard_series_count", {"shard": "0"}) + assert gauge == float(result.series_count) + + +@pytest.mark.asyncio +async def test_max_total_series_is_a_per_shard_cap(stub_source: StubSource) -> None: + # 60 series across 2 shards (each owns ~30), cap=5 per shard. With a + # per-shard cap each shard scores exactly 5 and the estate covers 10 + # (cap x shards) — not 5 total, which an estate-wide cap would give. + stub_source.respond(lambda _: StubResponse(series=_series_set(60))) + cfg = _runner_config(shards=2, max_total_series=5) + estate = 0 + for index in range(2): + runner = Runner( + cfg, stub_source, SnapshotStore(), OperationalMetrics(), shard=ShardAssignment(index, 2) + ) + result = await runner.run_group("g1") + assert result.series_count == 5 # each shard capped at its own budget + estate += result.series_count + assert estate == 10 # cap (5) x shards (2) + + +@pytest.mark.asyncio +async def test_density_suppressed_under_sharding(stub_source: StubSource) -> None: + stub_source.respond(lambda _: StubResponse(series=_series_set(20))) + raw: dict[str, Any] = { + "apiVersion": CURRENT_API_VERSION, + "datasource": {"url": "http://stub"}, + "defaults": {"window": "15m", "step": "15s", "min_points": 30, "density_by": ["instance"]}, + "scaleOut": {"enabled": True, "shards": 2}, + "groups": [ + {"name": "g1", "queries": [{"id": "m", "promql": "up", "detectors": [{"name": "MAD"}]}]} + ], + } + cfg = Config.model_validate(raw) + runner = Runner( + cfg, stub_source, SnapshotStore(), OperationalMetrics(), shard=ShardAssignment(0, 2) + ) + result = await runner.run_group("g1") + assert not [s for s in result.samples if s.metric == "anomaly_density"] + # …but the unsharded runner still emits it. + cfg_plain = Config.model_validate({**raw, "scaleOut": {"enabled": False}}) + plain = Runner(cfg_plain, stub_source, SnapshotStore(), OperationalMetrics(), shard=None) + plain_result = await plain.run_group("g1") + assert [s for s in plain_result.samples if s.metric == "anomaly_density"] + + +@pytest.mark.asyncio +async def test_expect_absence_emitted_by_exactly_one_shard(stub_source: StubSource) -> None: + # An expected query that returns nothing should emit a single + # anomaly_signal_missing across the shard set, not one per shard. + stub_source.respond(lambda _: StubResponse(series=[])) + raw: dict[str, Any] = { + "apiVersion": CURRENT_API_VERSION, + "datasource": {"url": "http://stub"}, + "defaults": {"window": "15m", "step": "15s", "min_points": 30}, + "scaleOut": {"enabled": True, "shards": 3}, + "groups": [ + { + "name": "g1", + "queries": [ + { + "id": "heartbeat", + "promql": "up", + "expect": True, + "expect_grace_runs": 1, + "detectors": [{"name": "MAD"}], + } + ], + } + ], + } + cfg = Config.model_validate(raw) + emitting = 0 + for index in range(3): + runner = Runner( + cfg, stub_source, SnapshotStore(), OperationalMetrics(), shard=ShardAssignment(index, 3) + ) + result = await runner.run_group("g1") + if [s for s in result.samples if s.metric == "anomaly_signal_missing"]: + emitting += 1 + assert emitting == 1 # exactly one shard owns the group's whole-query absence + + +def test_shard_owns_group_has_exactly_one_owner() -> None: + cfg = _runner_config(shards=4) + owners = [ + Runner( + cfg, StubSource(), SnapshotStore(), OperationalMetrics(), shard=ShardAssignment(i, 4) + )._shard_owns_group("g1") + for i in range(4) + ] + assert owners.count(True) == 1 + + +# ── Scheduling: priority-aware load shedding ───────────────────────── + + +class TestLoadShedder: + @pytest.mark.asyncio + async def test_uncapped_is_passthrough(self) -> None: + shedder = LoadShedder(None) + assert await shedder.acquire(priority=1, deadline_seconds=0.0) + assert shedder.running == 1 + shedder.release() + assert shedder.running == 0 + + @pytest.mark.asyncio + async def test_admits_up_to_cap(self) -> None: + shedder = LoadShedder(2) + assert await shedder.acquire(priority=1, deadline_seconds=1.0) + assert await shedder.acquire(priority=1, deadline_seconds=1.0) + assert shedder.running == 2 + + @pytest.mark.asyncio + async def test_sheds_when_saturated(self) -> None: + shedder = LoadShedder(1) + assert await shedder.acquire(priority=1, deadline_seconds=1.0) + shed = await shedder.acquire(priority=1, deadline_seconds=0.05) + assert shed is False + assert shedder.running == 1 + + @pytest.mark.asyncio + async def test_cancellation_propagates_and_frees_no_slot(self) -> None: + # A waiter cancelled (leadership loss / shutdown) must raise + # CancelledError, not silently resolve as a shed, and must not + # leak a concurrency slot. + shedder = LoadShedder(1) + assert await shedder.acquire(priority=1, deadline_seconds=5.0) + + task = asyncio.create_task(shedder.acquire(priority=1, deadline_seconds=5.0)) + while len(shedder._waiters) < 1: + await asyncio.sleep(0) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + # The original holder still holds exactly one slot; the cancelled + # waiter left nothing behind. + assert shedder.running == 1 + assert shedder._waiters == [] + + @pytest.mark.asyncio + async def test_higher_priority_wakes_first(self) -> None: + shedder = LoadShedder(1) + assert await shedder.acquire(priority=1, deadline_seconds=5.0) + order: list[int] = [] + + async def waiter(priority: int) -> None: + admitted = await shedder.acquire(priority=priority, deadline_seconds=5.0) + order.append(priority) + if admitted: + shedder.release() + + low = asyncio.create_task(waiter(1)) + high = asyncio.create_task(waiter(10)) + while len(shedder._waiters) < 2: + await asyncio.sleep(0) + shedder.release() # frees one slot; highest priority served first + await asyncio.gather(low, high) + assert order[0] == 10 + + @pytest.mark.asyncio + async def test_raising_cap_admits_waiter(self) -> None: + shedder = LoadShedder(1) + assert await shedder.acquire(priority=1, deadline_seconds=5.0) + admitted: list[bool] = [] + + async def waiter() -> None: + admitted.append(await shedder.acquire(priority=1, deadline_seconds=5.0)) + + task = asyncio.create_task(waiter()) + while len(shedder._waiters) < 1: + await asyncio.sleep(0) + shedder.set_max_concurrent(2) # now the queued waiter fits + await task + assert admitted == [True] + + +# ── estimate-cost accounts for per-group cadence ───────────────────── + + +def test_estimate_cost_uses_per_group_refresh() -> None: + raw = { + "apiVersion": CURRENT_API_VERSION, + "datasource": {"url": "http://vm:8428/"}, + "server": {"refresh_interval": "60s"}, + "groups": [ + { + "name": "fast", + "refresh_interval": "30s", + "queries": [{"id": "a", "promql": "up", "detectors": [{"name": "MAD"}]}], + }, + { + "name": "slow", + "refresh_interval": "300s", + "queries": [{"id": "b", "promql": "up", "detectors": [{"name": "MAD"}]}], + }, + ], + } + cfg = Config.model_validate(raw) + estimates = {g.group: g for g in estimate_cost(cfg)} + # One short query per refresh each, but the 30s group issues 10x the + # per-minute load of the 5m group. + assert estimates["fast"].refresh_interval_seconds == 30.0 + assert estimates["slow"].refresh_interval_seconds == 300.0 + assert estimates["fast"].short_queries_per_minute == 2.0 + assert estimates["slow"].short_queries_per_minute == 0.2 + # Faster cadence ⇒ more steady-state CPU for the same series count. + assert estimates["fast"].cpu_cores > estimates["slow"].cpu_cores + + +# ── Application scheduler wiring ───────────────────────────────────── + + +def _app(cfg: Config) -> Application: + return Application(cfg, Path("dummy.yaml")) + + +def _job_intervals(app: Application) -> dict[str, float]: + scheduler = AsyncIOScheduler() + app._reschedule_jobs(scheduler) + intervals: dict[str, float] = {} + for job in scheduler.get_jobs(): + intervals[job.id] = job.trigger.interval.total_seconds() + return intervals + + +def test_scheduler_uses_per_group_refresh_interval() -> None: + raw = { + "apiVersion": CURRENT_API_VERSION, + "datasource": {"url": "http://stub"}, + "server": {"refresh_interval": "60s"}, + "groups": [ + { + "name": "fast", + "refresh_interval": "30s", + "queries": [{"id": "a", "promql": "up", "detectors": [{"name": "MAD"}]}], + }, + { + "name": "slow", + "queries": [{"id": "b", "promql": "up", "detectors": [{"name": "MAD"}]}], + }, + ], + } + intervals = _job_intervals(_app(Config.model_validate(raw))) + assert intervals["group:fast"] == 30.0 + assert intervals["group:slow"] == 60.0 # inherits server default + + +def test_overrun_listener_increments_counter() -> None: + app = _app(Config.model_validate(_base_config())) + event = types.SimpleNamespace(job_id="group:g1") + app._on_job_max_instances(event) + assert ( + _metric_value(app._ops, "anomaly_group_skipped_total", {"group": "g1", "reason": "overrun"}) + == 1.0 + ) + + +def test_overrun_listener_ignores_non_group_jobs() -> None: + app = _app(Config.model_validate(_base_config())) + app._on_job_max_instances(types.SimpleNamespace(job_id="selftest")) + samples = list(app._ops.group_skipped_total.collect()[0].samples) + assert samples == [] # nothing counted for a non-group job + + +@pytest.mark.asyncio +async def test_admit_and_run_sheds_under_backpressure() -> None: + raw = { + "apiVersion": CURRENT_API_VERSION, + "datasource": {"url": "http://stub"}, + "safety": {"max_concurrent_groups": 1}, + "groups": [ + { + "name": "low", + "priority": 1, + "refresh_interval": "0.1s", + "queries": [{"id": "a", "promql": "up", "detectors": [{"name": "MAD"}]}], + } + ], + } + app = _app(Config.model_validate(raw)) + # Saturate the single slot so the next admission must shed. + assert await app._load_shedder.acquire(priority=5, deadline_seconds=10.0) + await app._admit_and_run("low") + assert ( + _metric_value( + app._ops, "anomaly_group_skipped_total", {"group": "low", "reason": "backpressure"} + ) + == 1.0 + ) + + +def test_reschedule_jobs_does_not_register_listeners() -> None: + # Regression: the overrun listener must be registered once (in + # startup), never inside _reschedule_jobs — otherwise every reload + # stacks a duplicate listener and multiplies the overrun counter. + app = _app(Config.model_validate(_base_config())) + scheduler = AsyncIOScheduler() + before = len(scheduler._listeners) + app._reschedule_jobs(scheduler) + app._reschedule_jobs(scheduler) + assert len(scheduler._listeners) == before # reschedule is listener-neutral + + +# ── Regression: scaleOut reload + group-removal cleanup ────────────── + + +def _yaml_config(tmp_path: Path, shards: int) -> Path: + import yaml + + raw = { + "apiVersion": CURRENT_API_VERSION, + "datasource": {"url": "http://stub"}, + "scaleOut": {"enabled": True, "shards": shards}, + "groups": [ + {"name": "g1", "queries": [{"id": "m", "promql": "up", "detectors": [{"name": "MAD"}]}]} + ], + } + path = tmp_path / "config.yaml" + path.write_text(yaml.safe_dump(raw)) + return path + + +@pytest.mark.asyncio +async def test_reload_rejects_scaleout_change( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + from promanomaly.config import load_config + + monkeypatch.delenv("POD_NAME", raising=False) # shard index resolves to 0 + monkeypatch.setattr("promanomaly.main.PromQLSource", lambda **_k: StubSource()) + path = _yaml_config(tmp_path, shards=2) + app = Application(load_config(path), path) + + # Operator edits shards on disk, then triggers a hot reload. + path.write_text(path.read_text().replace("shards: 2", "shards: 3")) + ok, msg = await app.reload() + + assert ok is False + assert "scaleOut" in msg and "restart" in msg + assert app._config.scaleOut.shards == 2 # previous config stays active + + +def test_replace_config_drops_removed_group_series_count(stub_source: StubSource) -> None: + def _two_group_cfg(names: list[str]) -> Config: + return Config.model_validate( + { + "apiVersion": CURRENT_API_VERSION, + "datasource": {"url": "http://stub"}, + "defaults": {"window": "15m", "step": "15s", "min_points": 30}, + "groups": [ + { + "name": n, + "queries": [{"id": "m", "promql": "up", "detectors": [{"name": "MAD"}]}], + } + for n in names + ], + } + ) + + runner = Runner( + _two_group_cfg(["g1", "gone"]), stub_source, SnapshotStore(), OperationalMetrics() + ) + runner._last_series_count["g1"] = 100 + runner._last_series_count["gone"] = 50 + # Reload drops 'gone'; its sticky count must go too so it can't keep + # shrinking the cross-group budget for surviving groups after reload. + runner.replace_config(_two_group_cfg(["g1"])) + assert "gone" not in runner._last_series_count + assert runner._last_series_count.get("g1") == 100 + + +class TestLoadShedderReload: + @pytest.mark.asyncio + async def test_uncapping_releases_queued_waiters(self) -> None: + # set_max_concurrent(None) means "admit everyone" — a queued + # waiter must be released immediately, not left to shed against a + # cap that no longer exists. + shedder = LoadShedder(1) + assert await shedder.acquire(priority=1, deadline_seconds=5.0) + admitted: list[bool] = [] + + async def waiter() -> None: + admitted.append(await shedder.acquire(priority=1, deadline_seconds=5.0)) + + task = asyncio.create_task(waiter()) + while len(shedder._waiters) < 1: + await asyncio.sleep(0) + shedder.set_max_concurrent(None) # uncap → drain the queue now + await task + assert admitted == [True] + assert shedder._waiters == [] diff --git a/docs/operations.md b/docs/operations.md index bab9e6e..cb86a67 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -280,6 +280,96 @@ Two new operational metrics: **Sizing.** Three replicas is the typical HA setup. Anything past that is leader-election overhead without benefit — only one replica runs detectors at any moment. +## Horizontal Scale-Out (active/active sharding) + +HA mode is **active/passive**: leader election guarantees exactly one replica runs detection, which is correct but caps total throughput at a single process. When one process can no longer score the whole estate within a refresh interval, switch to **active/active scale-out**: `N` replicas each own a disjoint shard of the series and run concurrently. + +```yaml +scaleOut: + enabled: true + shards: 3 # run 3 replicas, each scoring ~1/3 of the series +``` + +The two modes are **mutually exclusive** — `scaleOut.enabled` and `highAvailability.enabled` both `true` is rejected at boot. Pick by question: + +| Question | Mode | +|----------|------| +| "I want redundancy; one process can keep up." | **HA** (active/passive) | +| "One process can't score everything within a refresh." | **Scale-out** (active/active) | + +How it works: + +- **Stable shard index.** The chart renders a **StatefulSet** of `shards` pods (`pod-0`…`pod-N-1`); each derives its shard index from the pod ordinal in `POD_NAME` — no per-pod config. (Pin `scaleOut.shard_index` only for non-StatefulSet topologies.) +- **Series partition.** Each replica computes a **rendezvous hash** (highest-random-weight) of every candidate series key and scores only the series that hash to its shard. The partition is disjoint and the union covers the whole estate. The hash is process-stable (BLAKE2b, not Python's salted `hash`), so all replicas agree on ownership. +- **Resharding is minimal.** Scaling `shards` up or down moves only ~`1/shards` of the series (the consistent-hashing property), and a newly-started pod runs its first detection pass *immediately* on boot, so a moved series is re-scored on acquisition rather than after a tick — mirroring the HA failover rule. +- **Scraping.** The regular `Service` selects every shard's pod, so the existing `ServiceMonitor` collects all shards; their `/metrics` union is the whole estate. A headless Service (`-headless`) backs the StatefulSet's stable DNS. + +Two operational metrics expose the partition: + +- `anomaly_shard{instance=""}` — the shard index this replica owns. +- `anomaly_shard_series_count{shard=""}` — series scored by that shard, so you can confirm the partition is even. + +**`max_total_series` becomes a per-shard cap.** Each replica enforces `safety.max_total_series` against *its own* shard's series (the cap is applied after the shard filter), so the whole-estate ceiling is `max_total_series × shards`. Size it as `(expected total series / shards) × headroom`, not the whole estate — see the sizing table in the [production checklist](production-checklist.md). + +**Set `max_series_per_query` above your largest single query's cardinality.** `max_series_per_query` is a *per-query fetch* ceiling applied at the TSDB, **before** sharding — every shard fetches the same (TSDB-ordered) first `max_series_per_query` series and then keeps its slice. So sharding distributes scoring *within* that ceiling; it cannot scale a single query beyond it. Worse, if a query *truncates* (you'll see `anomaly_series_dropped_total{reason="cardinality_overflow"}`), different shards may fetch different truncated subsets and the partition is no longer guaranteed complete. Keep `max_series_per_query` comfortably above any single query's true cardinality so no query truncates; scale the *estate* by adding more queries/series, not by relying on sharding to subset one oversized query. + +**No per-shard redundancy.** Scale-out is throughput, not availability — there is no leader/follower within a shard. If a shard's pod is down, that shard's slice of the estate goes unscored until Kubernetes reschedules it (Prometheus staleness handles the gap), and that shard's owner-only signals (recurrence, `expect:`/discovery absence) pause with it. Layer a `PodDisruptionBudget` and sane resource requests on the StatefulSet; if you need both redundancy *and* horizontal scale, that's a future "leader-per-shard" direction, not this release. + +**Scaling — change `scaleOut.shards`, not the replica count.** The chart ties the StatefulSet's `replicas` to `scaleOut.shards`, so resizing is a single Helm value change that moves both together. Do **not** `kubectl scale` the StatefulSet directly: dropping replicas below `shards` orphans a shard's series (no pod owns that index and those series go unscored), and adding replicas past `shards` makes the extra pod fail at boot (its ordinal is out of range). Keep replicas == shards. + +**TSDB query load is not reduced by sharding.** PromQL returns all of a metric's series, so every shard still runs the *full* query and then discards the series it doesn't own — sharding scales detection *compute*, not query throughput. If the TSDB read volume is the constraint, enable a shared query cache (`safety.query_cache.backend: redis` + `safety.redis.url`) so the shards' identical fetches collapse onto one TSDB hit per cache bucket. + +**Group-level rollups under sharding.** A few signals are whole-group aggregates rather than per-series, and are handled so the partition never multiplies a series: + +- **`anomaly_density`** (fleet density rollup) is **suppressed** in sharded mode — each shard sees only its slice, so an in-process value would be a partial fraction. Compute true fleet density with a recording rule over the per-series `anomaly_outside_threshold` (the shard union is complete), per [patterns](patterns.md). +- **Recurrence** heatmaps and **`expect:` / discovery `anomaly_signal_missing`** are emitted by a single *owner shard* (chosen by hashing the group / series identity), so each appears exactly once across the fleet. Recurrence re-reads the complete history from the TSDB, so the single copy is correct. + +**Limitation — cohort detectors.** Sharding partitions by series key, so a cohort's members hash to different shards and each shard's Cohort detector would see only a partial population. Run cohort groups in a separate unsharded deployment (or accept the per-shard comparison). The detector logs `cohort_with_scaleout` at boot when it detects this combination. + +## Per-Group Refresh Intervals + +`server.refresh_interval` sets the default cadence for every group. A group can override it so cheap fast-moving signals and expensive stratified/cohort groups coexist in one detector without a global compromise: + +```yaml +server: + refresh_interval: 1m # default for groups without an override +groups: + - name: error_rates + refresh_interval: 30s # score hot signals more often + queries: [ ... ] + - name: stratified_latency + refresh_interval: 5m # expensive multi-week baseline — refresh rarely + queries: [ ... ] +``` + +- An unset group `refresh_interval` falls back to `server.refresh_interval` — default behaviour is unchanged. +- The per-group lock already prevents a group's runs from overlapping; each group runs on its own scheduler cadence. +- The `refresh_interval >= window/6` quality guard is enforced **per group** at boot — a too-tight override logs `aggressive_refresh_interval{group}` just like a too-tight server default. +- Stratified detectors' separate baseline-refresh cadence is independent and composes with the per-group short-window cadence. +- `validate --estimate-cost` accounts for per-group cadences — each group reports its `refresh_interval_seconds` and a `short_queries_per_minute` rate, so a too-aggressive interval shows up as outsized TSDB load before deploy. + +## Priority-Aware Scheduling and Load Shedding + +By default group scheduling is unbounded — every group runs on its cadence with no admission control. When the TSDB is occasionally slow or throttled, cap concurrency so the detector protects its most important groups first: + +```yaml +safety: + max_concurrent_groups: 4 # at most 4 groups run at once +groups: + - name: critical_slos + priority: 10 # admitted ahead of lower-priority groups under load + queries: [ ... ] +``` + +Under contention the scheduler orders pending runs by `priority` (the same field that drives cardinality overflow) and, when a group can't get a slot within its refresh interval, **sheds** the run rather than queuing unboundedly. Shedding is visible, never silent: + +- `anomaly_group_skipped_total{group, reason="backpressure"}` — no slot within the interval under `max_concurrent_groups`. +- `anomaly_group_skipped_total{group, reason="overrun"}` — the previous run was still in flight at the next tick. + +A skip **delays** a score; it never **suppresses** a detected anomaly — the last good snapshot stays in place per `serve_stale`, and the `AnomalyGroupStarved` reference alert fires on sustained shedding. This is a *scheduling* concern, not a *silencing* one; see [degraded-modes](operations/degraded-modes.md). + +The cap governs *scheduled ticks*. The immediate first run fired on boot, on HA leader-acquisition, and after a config reload runs every group once regardless, so `/metrics` is populated promptly even with a low cap; admission control kicks in from the first scheduled tick onward. (Reload is operator-triggered, so a reload of a many-group config briefly fans out all groups at once — size the TSDB for that burst, or stagger heavy groups across deployments.) + ## Dynamic Series Discovery Hand-writing one query per cohort member doesn't scale past a handful of services. `discover:` lets one templated query fan out across the distinct values of a label at runtime: diff --git a/docs/operations/degraded-modes.md b/docs/operations/degraded-modes.md index cefbdf4..e45db0f 100644 --- a/docs/operations/degraded-modes.md +++ b/docs/operations/degraded-modes.md @@ -39,6 +39,7 @@ All failures increment `anomaly_failures_total{group, detector, reason}` (bounde | Detector degrading | `anomaly_detect_success_ratio{window="1h"} < 0.9` | Inspect failures by detector | | Dead detection pipeline | `anomaly_selftest_ok == 0` (opt-in self-test) | Check threshold / exporter regression / global misconfig | | Leader churn (HA) | `rate(anomaly_leader_transitions_total[15m]) > 2` | Check Lease contention / pod restarts | +| Group starvation (shedding) | `rate(anomaly_group_skipped_total[15m]) > 0` | Raise max_concurrent_groups / lengthen refresh / scale out | ## TSDB Read Failure @@ -113,10 +114,34 @@ The detect/threshold/export path is failing to surface a known anomaly: check fo **Recovery** Check Lease contention, pod restarts, and `lease_duration` / `retry_period` tuning. Brief churn is safe; sustained churn wastes runs. +## Group Starvation (load shedding) + +When `safety.max_concurrent_groups` is set, the scheduler is a priority-aware gate: under sustained TSDB backpressure it admits at most that many group runs at once, prefers higher-`priority` groups, and **sheds** (skips) a run rather than queuing it unboundedly. Like every other degraded mode here, shedding is **visible, never silent** — a skip delays a score, it never suppresses a detected anomaly (the last good snapshot stays in place per `serve_stale`). + +| Reason | Meaning | +|--------|---------| +| `backpressure` | No concurrency slot freed within the group's refresh interval under `max_concurrent_groups`. | +| `overrun` | The group's previous run was still in flight when the next tick fired (it can't keep up with its own cadence). | + +**Detection** +`rate(anomaly_group_skipped_total[15m]) > 0` + +**Recovery** +Sustained shedding means the detector is starved of capacity for the load. Options, cheapest first: + +- Raise `safety.max_concurrent_groups` (if CPU/TSDB headroom exists). +- Lengthen the starved group's `refresh_interval` (per-group override) so it demands fewer passes. +- Split the group's heavy queries, or move expensive stratified/cohort groups to their own deployment. +- Scale out (`scaleOut.enabled`) so multiple replicas share the work. + +A `backpressure` skip on a **low**-priority group while high-priority groups stay fresh is the gate working as designed — raise the starved group's `priority` only if it genuinely matters more. + +**Shipped alert:** `AnomalyGroupStarved` + ## Putting It Together - **Page** on `AnomalySourceFailing`, `AnomalyStale`, `AnomalySnapshotStale`, and (if the self-test is enabled) `AnomalyPipelineDead`. -- **Ticket** on `AnomalyDetectorDegraded`. +- **Ticket** on `AnomalyDetectorDegraded` and `AnomalyGroupStarved`. All other silencing, deploy damping, and flap suppression belongs in Alertmanager (silences + `for:` clauses). diff --git a/docs/production-checklist.md b/docs/production-checklist.md index 53f0f63..f1711f4 100644 --- a/docs/production-checklist.md +++ b/docs/production-checklist.md @@ -16,12 +16,24 @@ A single 1-CPU replica comfortably handles **~10 000 series at 1-minute refresh* | 1 000 – 10 000 | 1 m | 1 | 512 Mi | 1 or 2 (HA) | | 10 000 – 25 000 | 1 m | 2 | 1 Gi | 2 (HA) | | 25 000 – 50 000 | 2 m | 4 | 2 Gi | 2–3 (HA) | -| > 50 000 | varies | — | — | Shard by group | +| > 50 000 | varies | — | — | Scale out (shard) | **Always run** `promanomaly validate --config --estimate-cost` for a config-specific projection (projected series, TSDB queries per refresh, and coarse CPU/memory estimate). **Refresh-interval impact** -CPU scales linearly with `1 / refresh_interval`. Keep `refresh_interval` no smaller than `defaults.window / 6` (the detector warns on aggressive settings). +CPU scales linearly with `1 / refresh_interval`. Keep `refresh_interval` no smaller than `defaults.window / 6` (the detector warns on aggressive settings). With per-group `refresh_interval` overrides, the cost estimator reports each group's `short_queries_per_minute` so a fast group's load is visible at PR time. + +**Scaling past one process (`scaleOut`)** +A single process — HA leader included — caps detection at what one CPU can score within a refresh. Past that ceiling, switch to active/active scale-out: `shards` replicas each score `total / shards` series. Sizing extends the single-replica model by dividing the estate across shards: + +| Total series | Mode | `scaleOut.shards` | Per-shard series | Per-shard CPU | `max_total_series` | +|--------------|------|-------------------|------------------|---------------|--------------------| +| ≤ 50 000 | HA (active/passive) | — | (whole estate) | as above | whole-estate cap | +| 50 000 – 150 000 | Scale-out | 3 | ~`total/3` | 2 | per-shard (`total/3 × headroom`) | +| 150 000 – 300 000 | Scale-out | 6 | ~`total/6` | 2 | per-shard | +| > 300 000 | Scale-out | `ceil(total / 50 000)` | ~50 000 | 2 | per-shard | + +`safety.max_total_series` is enforced **per shard** in scale-out mode, so the whole-estate ceiling is `max_total_series × shards`. Size each shard's cap as `(expected total / shards) × headroom`, not the whole estate. Contrast with HA, where one replica enforces the cap against the whole estate. See [Horizontal Scale-Out](operations.md#horizontal-scale-out-activeactive-sharding). **Stratified detectors** add TSDB reads (not CPU). The cost estimator reports `stratified_baseline_queries_per_day` separately. @@ -31,7 +43,8 @@ CPU scales linearly with `1 / refresh_interval`. Keep `refresh_interval` no smal |----------------------------|--------------------------------------------------|-------------| | **A. Single-replica Recreate** | < 10 k series, brief `/metrics` gap OK during deploys | `replicaCount: 1`
`strategy: Recreate`
`highAvailability.enabled: false` | | **B. HA leader-elected** | Need continuous `/metrics` availability | `replicaCount: 2`
`highAvailability.enabled: true`
`safety.redis.url: …` | -| **C. Multi-cluster** | Many clusters (per-cluster or federated) | See [multi-cluster reference architectures](architecture/multi-cluster.md) | +| **C. Active/active scale-out** | One process can't score the estate within a refresh | `scaleOut.enabled: true`
`scaleOut.shards: N` (renders a StatefulSet) | +| **D. Multi-cluster** | Many clusters (per-cluster or federated) | See [multi-cluster reference architectures](architecture/multi-cluster.md) | ## Production-Readiness Checklist diff --git a/examples/alerts/promanomaly-rules.yaml b/examples/alerts/promanomaly-rules.yaml index 2135753..9231c8d 100644 --- a/examples/alerts/promanomaly-rules.yaml +++ b/examples/alerts/promanomaly-rules.yaml @@ -277,3 +277,27 @@ spec: → export pipeline is not surfacing anomalies even though the process is up — check for a config/threshold mistake or an exporter regression. See anomaly_selftest_failures_total. + + - alert: AnomalyGroupStarved + # Priority-aware load shedding (safety.max_concurrent_groups): + # under sustained TSDB backpressure the scheduler sheds + # low-priority group runs (reason="backpressure") or a group + # cannot finish within its refresh interval (reason="overrun"). + # A skip delays a score; it never suppresses a detected anomaly + # (the last good snapshot stays per serve_stale). Sustained + # shedding means the detector is starved of capacity — raise the + # concurrency cap, lengthen the group's refresh_interval, or + # lighten its config. + expr: rate(anomaly_group_skipped_total[15m]) > 0 + for: 15m + labels: + severity: warning + annotations: + summary: "promanomaly is shedding runs for {{ $labels.group }} ({{ $labels.reason }})" + description: | + Group {{ $labels.group }} has had runs skipped + (reason={{ $labels.reason }}) over the last 15 minutes. The + detector cannot keep every group on its cadence under the + current load. Scores for this group are delayed, not + suppressed. Raise safety.max_concurrent_groups, lengthen the + group's refresh_interval, or split its heavy queries. diff --git a/examples/configs/scale-out.yaml b/examples/configs/scale-out.yaml new file mode 100644 index 0000000..ccd82ba --- /dev/null +++ b/examples/configs/scale-out.yaml @@ -0,0 +1,90 @@ +# Active/active horizontal scale-out + per-group cadence + load shedding. +# +# The resilience topology for an estate too large for one process to +# score within a refresh interval. Three replicas each own a +# disjoint shard of the series (rendezvous hashing on the series key) and +# run concurrently; the union of their /metrics covers the whole estate. +# +# Deployment notes: +# - The Helm chart renders a StatefulSet of ``scaleOut.shards`` pods when +# ``scaleOut.enabled=true``; each pod self-assigns its shard index from +# its StatefulSet ordinal (POD_NAME) — no per-pod config. +# - ``scaleOut`` and ``highAvailability`` are mutually exclusive: HA is +# active/passive (one leader), scale-out is active/active (N shards). +# - ``safety.max_total_series`` is enforced PER SHARD here, so the whole +# estate ceiling is ``max_total_series x shards`` (15000 x 3 = 45000). +# - Demonstrates per-group ``refresh_interval`` overrides and +# priority-aware load shedding (``safety.max_concurrent_groups``). +# See docs/operations.md and docs/production-checklist.md. + +apiVersion: promanomaly.io/v1 + +datasource: + url: http://victoriametrics:8428/ + timeout: 10s + +server: + listen: ":9092" + # Default cadence for groups without their own refresh_interval. + refresh_interval: 1m + reload: + enabled: true + watch_configmap: true + ready_endpoint: true + +safety: + # Per-shard cap: the whole-estate ceiling is this x scaleOut.shards. + max_series_per_query: 2000 + max_total_series: 15000 + on_source_failure: serve_stale + # Priority-aware load shedding: at most 2 groups run at once. Under + # TSDB backpressure the scheduler prefers higher-priority groups and + # sheds (skips) low-priority runs that can't get a slot within their + # refresh interval, surfacing them on anomaly_group_skipped_total. + max_concurrent_groups: 2 + +# Active/active sharding. Mutually exclusive with highAvailability. +scaleOut: + enabled: true + shards: 3 + +defaults: + window: 1h + step: 15s + min_points: 60 + warmup_policy: emit_warming_up + emit_duration: true + emit_severity: true + alert_thresholds: + score: 3.0 + +groups: + # Hot, cheap signal — scored more often than the server default. + - name: error_rates + priority: 10 # admitted ahead of lower-priority groups + refresh_interval: 30s + queries: + - id: api_error_rate + promql: | + sum by (service) (rate(http_requests_total{code=~"5.."}[5m])) + / + sum by (service) (rate(http_requests_total[5m])) + min_relative_delta: 0.05 + detectors: + - name: MAD + - name: ZScoreEWMA + params: + alpha: 0.05 + + # Expensive stratified baseline — refreshed rarely so it doesn't + # compete with the hot group on every tick. + - name: weekly_latency + priority: 1 + refresh_interval: 5m + queries: + - id: api_p95_latency + promql: histogram_quantile(0.95, sum by (service, le) (rate(http_request_duration_seconds_bucket[5m]))) + detectors: + - name: HourOfDayMAD + params: + lookback: 2w