diff --git a/contrib/grafana-dashboard.json b/contrib/grafana-dashboard.json index 868b8b82..13ad03a1 100644 --- a/contrib/grafana-dashboard.json +++ b/contrib/grafana-dashboard.json @@ -956,7 +956,6 @@ }, "editorMode": "code", "expr": "sectracker_cache_regeneration_duration_seconds{job=\"node\", instance=\"$Instance\"}", - "legendFormat": "{{mode}}", "range": true, "refId": "A" } @@ -1027,6 +1026,522 @@ ], "title": "Cache regeneration suggestions", "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "Duration of the last garbage collection run (oneshot batch job)", + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + }, + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { + "h": 5, + "w": 6, + "x": 12, + "y": 18 + }, + "id": 23, + "options": { + "colorMode": "none", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.1.1", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "sectracker_garbage_collect_duration_seconds{job=\"node\", instance=\"$Instance\", step=\"total\"}", + "range": true, + "refId": "A" + } + ], + "title": "Garbage collection duration", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "Duration of the last CVE delta ingest run (oneshot batch job)", + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + }, + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { + "h": 5, + "w": 6, + "x": 18, + "y": 18 + }, + "id": 24, + "options": { + "colorMode": "none", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.1.1", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "sectracker_cve_delta_ingest_duration_seconds{job=\"node\", instance=\"$Instance\"}", + "range": true, + "refId": "A" + } + ], + "title": "CVE delta ingest duration", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "Total rows deleted in the last garbage collection run", + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 5, + "w": 6, + "x": 0, + "y": 23 + }, + "id": 25, + "options": { + "colorMode": "none", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.1.1", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "sum(sectracker_garbage_collect_deleted{job=\"node\", instance=\"$Instance\"})", + "range": true, + "refId": "A" + } + ], + "title": "Garbage collection deleted", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "CVEs ingested in the last CVE delta ingest run", + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 5, + "w": 6, + "x": 6, + "y": 23 + }, + "id": 26, + "options": { + "colorMode": "none", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.1.1", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "sectracker_cve_delta_ingest_cves{job=\"node\", instance=\"$Instance\"}", + "range": true, + "refId": "A" + } + ], + "title": "CVE delta ingest CVEs", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "Average matching duration over the last 5 minutes (listen worker)", + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + }, + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { + "h": 5, + "w": 6, + "x": 0, + "y": 28 + }, + "id": 27, + "options": { + "colorMode": "none", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.1.1", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "rate(sectracker_matching_duration_seconds_sum{job=\"sectracker-worker\", instance=\"$Instance\"}[5m]) / rate(sectracker_matching_duration_seconds_count{job=\"sectracker-worker\", instance=\"$Instance\"}[5m])", + "range": true, + "refId": "A" + } + ], + "title": "Matching duration (avg)", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "Average matching candidate count over the last 5 minutes", + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 5, + "w": 6, + "x": 6, + "y": 28 + }, + "id": 28, + "options": { + "colorMode": "none", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.1.1", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "rate(sectracker_matching_candidates_sum{job=\"sectracker-worker\", instance=\"$Instance\"}[5m]) / rate(sectracker_matching_candidates_count{job=\"sectracker-worker\", instance=\"$Instance\"}[5m])", + "range": true, + "refId": "A" + } + ], + "title": "Matching candidates (avg)", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "Average eval batch ingest duration over the last 5 minutes", + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + }, + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { + "h": 5, + "w": 6, + "x": 12, + "y": 28 + }, + "id": 29, + "options": { + "colorMode": "none", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.1.1", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "rate(sectracker_nix_evaluation_batch_ingest_duration_seconds_sum{job=\"sectracker-evaluator\", instance=\"$Instance\"}[5m]) / rate(sectracker_nix_evaluation_batch_ingest_duration_seconds_count{job=\"sectracker-evaluator\", instance=\"$Instance\"}[5m])", + "range": true, + "refId": "A" + } + ], + "title": "Eval batch ingest duration (avg)", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "Average attributes per eval ingest batch over the last 5 minutes", + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 5, + "w": 6, + "x": 18, + "y": 28 + }, + "id": 30, + "options": { + "colorMode": "none", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "12.1.1", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "rate(sectracker_nix_evaluation_batch_ingest_attributes_sum{job=\"sectracker-evaluator\", instance=\"$Instance\"}[5m]) / rate(sectracker_nix_evaluation_batch_ingest_attributes_count{job=\"sectracker-evaluator\", instance=\"$Instance\"}[5m])", + "range": true, + "refId": "A" + } + ], + "title": "Eval batch attributes (avg)", + "type": "stat" } ], "schemaVersion": 41, @@ -1035,13 +1550,13 @@ "list": [ { "current": {}, - "definition": "label_values(sql_delta,instance)", + "definition": "label_values(up, instance)", "label": "Instance", "name": "Instance", "options": [], "query": { "qryType": 1, - "query": "label_values(sql_delta,instance)", + "query": "label_values(up, instance)", "refId": "PrometheusVariableQueryEditor-VariableQuery" }, "refresh": 1, diff --git a/infra/common.nix b/infra/common.nix index 584fbd18..79ee7465 100644 --- a/infra/common.nix +++ b/infra/common.nix @@ -98,6 +98,7 @@ in systemd.tmpfiles.rules = [ "d ${config.services.nix-security-tracker.settings.METRICS_TEXTFILE_DIR} 2750 nix-security-tracker ${config.services.prometheus.exporters.node.user} -" + "d /var/lib/nix-security-tracker/prometheus-multiproc 0750 nix-security-tracker nix-security-tracker -" ]; services.prometheus.exporters.postgres = { diff --git a/nix/configuration.nix b/nix/configuration.nix index 169fe937..03536edc 100644 --- a/nix/configuration.nix +++ b/nix/configuration.nix @@ -181,9 +181,27 @@ in type = types.int; default = 2; }; + metricsWorkerPort = mkOption { + description = '' + TCP port for the main listen worker Prometheus /metrics endpoint (matching). + ''; + type = types.port; + default = 9252; + }; + metricsEvaluatorPort = mkOption { + description = '' + TCP port for the evaluator multiprocess Prometheus /metrics sidecar. + ''; + type = types.port; + default = 9253; + }; }; config = mkIf cfg.enable { + networking.firewall.allowedTCPPorts = [ + cfg.metricsWorkerPort + cfg.metricsEvaluatorPort + ]; environment.systemPackages = [ wstExternalManageScript ]; services = { # TODO(@fricklerhandwerk): move all configuration over to pydantic-settings @@ -332,7 +350,14 @@ in ]; wantedBy = [ "multi-user.target" ]; + environment = { + PROMETHEUS_MULTIPROC_DIR = "/var/lib/nix-security-tracker/prometheus-multiproc"; + }; + script = '' + mkdir -p "$PROMETHEUS_MULTIPROC_DIR" + # prometheus_client multiprocess mode requires an empty directory at start. + find "$PROMETHEUS_MULTIPROC_DIR" -mindepth 1 -delete # Before starting, crash all the in-progress evaluations. # This will prevent them from being stalled forever, since workers would not pick up evaluations marked as in-progress. wst-manage crash_all_evaluations @@ -343,6 +368,31 @@ in ''; }; + nix-security-tracker-evaluator-metrics = { + description = "Web security tracker - evaluator Prometheus metrics sidecar"; + after = [ + "network.target" + "nix-security-tracker-evaluator.service" + ]; + requires = [ "nix-security-tracker-evaluator.service" ]; + wantedBy = [ "multi-user.target" ]; + + environment = { + PROMETHEUS_MULTIPROC_DIR = "/var/lib/nix-security-tracker/prometheus-multiproc"; + DJANGO_SETTINGS = builtins.toJSON ( + cfg.settings + // { + METRICS_HTTP_PORT = cfg.metricsEvaluatorPort; + } + ); + }; + + script = '' + mkdir -p "$PROMETHEUS_MULTIPROC_DIR" + wst-manage serve_worker_metrics + ''; + }; + nix-security-tracker-caching = { description = "Web security tracker - cache regeneration"; after = [ @@ -380,6 +430,13 @@ in ]; wantedBy = [ "multi-user.target" ]; + environment.DJANGO_SETTINGS = builtins.toJSON ( + cfg.settings + // { + METRICS_HTTP_PORT = cfg.metricsWorkerPort; + } + ); + script = '' wst-manage listen --recover \ --channels \ @@ -425,7 +482,11 @@ in "postgresql.service" "nix-security-tracker-worker.service" ]; - serviceConfig.Type = "oneshot"; + serviceConfig = { + Type = "oneshot"; + # Make performance metrics file, produced as a side effect, readable by Prometheus node exporter + UMask = "0027"; + }; script = '' wst-manage ingest_delta_cve "$(date --date='yesterday' --iso)" ${ @@ -450,7 +511,11 @@ in ]; wantedBy = [ "multi-user.target" ]; - serviceConfig.Type = "oneshot"; + serviceConfig = { + Type = "oneshot"; + # Make performance metrics file, produced as a side effect, readable by Prometheus node exporter + UMask = "0027"; + }; script = '' wst-manage garbage_collect ''; diff --git a/src/project/settings.py b/src/project/settings.py index 7031a950..0326c79e 100644 --- a/src/project/settings.py +++ b/src/project/settings.py @@ -106,6 +106,12 @@ class DjangoSettings(BaseModel): """, default=None, ) + METRICS_HTTP_PORT: int | None = Field( + description=""" + If set, expose prometheus_client /metrics on this TCP port (listen workers). + """, + default=None, + ) CHANNEL_MONITORING_URL: HttpUrl = Field( description=""" URL from which to fetch the current channel structure. diff --git a/src/shared/apps.py b/src/shared/apps.py index 4dd9e134..3d7606b6 100644 --- a/src/shared/apps.py +++ b/src/shared/apps.py @@ -7,8 +7,16 @@ class SharedConfig(AppConfig): name = "shared" def ready(self) -> None: + import sys + import shared.listeners # noqa + # Expose /metrics only in pgpubsub listen --worker processes (not web/oneshots). + if settings.METRICS_HTTP_PORT is not None and "--worker" in sys.argv: + from shared.metrics import start_worker_metrics_server + + start_worker_metrics_server() + # TODO: run this as a separate service, as this is almost exclusively a deployment concern if settings.SYNC_GITHUB_STATE_AT_STARTUP: from shared.auth.github_state import GithubState diff --git a/src/shared/listeners/automatic_linkage.py b/src/shared/listeners/automatic_linkage.py index b1e5f829..547662c1 100644 --- a/src/shared/listeners/automatic_linkage.py +++ b/src/shared/listeners/automatic_linkage.py @@ -8,6 +8,7 @@ # ============================================================================ import logging +import time from dataclasses import dataclass, field import pgpubsub @@ -24,6 +25,7 @@ ) from shared.channels import ContainerChannel +from shared.metrics import observe_matching from shared.models.cve import Container, Cpe from shared.models.linkage import ( CVEDerivationClusterProposal, @@ -252,52 +254,61 @@ def produce_linkage_candidates( def build_new_links(container: Container) -> bool: - if container.cve.triaged: - logger.info( - "Container received for '%s', but already triaged, skipping linkage.", - container.cve, - ) - return False - - if CVEDerivationClusterProposal.objects.filter( - cve=container.cve, - algorithm_version=CVEDerivationClusterProposal.CURRENT_ALGORITHM_VERSION, - ).exists(): - logger.info("Suggestion already exists for '%s', skipping", container.cve) - return False - - outcome = resolve_linkage_candidates(container) - - proposal = CVEDerivationClusterProposal.objects.create( - cve=container.cve, - status=( - CVEDerivationClusterProposal.Status.REJECTED + start = time.time() + candidates = 0 + try: + if container.cve.triaged: + logger.info( + "Container received for '%s', but already triaged, skipping linkage.", + container.cve, + ) + return False + + if CVEDerivationClusterProposal.objects.filter( + cve=container.cve, + algorithm_version=CVEDerivationClusterProposal.CURRENT_ALGORITHM_VERSION, + ).exists(): + logger.info("Suggestion already exists for '%s', skipping", container.cve) + return False + + outcome = resolve_linkage_candidates(container) + if outcome.rejection is not None and outcome.rejection.match_count: + candidates = outcome.rejection.match_count + elif outcome.derivations is not None: + candidates = outcome.derivations.count() + + proposal = CVEDerivationClusterProposal.objects.create( + cve=container.cve, + status=( + CVEDerivationClusterProposal.Status.REJECTED + if outcome.rejection + else CVEDerivationClusterProposal.Status.PENDING + ), + rejection_reason=outcome.rejection.reason if outcome.rejection else None, + rejection_match_count=outcome.rejection.match_count or None if outcome.rejection - else CVEDerivationClusterProposal.Status.PENDING - ), - rejection_reason=outcome.rejection.reason if outcome.rejection else None, - rejection_match_count=outcome.rejection.match_count or None - if outcome.rejection - else None, - rejection_max_matches_limit=outcome.rejection.max_matches_limit or None - if outcome.rejection - else None, - algorithm_version=CVEDerivationClusterProposal.CURRENT_ALGORITHM_VERSION, - ) - - if outcome.derivations: - links = build_derivation_links(proposal, outcome.derivations) - DerivationClusterProposalLink.objects.bulk_create(links) - pkg_links = build_package_links(proposal, outcome.derivations) - PackageClusterProposalLink.objects.bulk_create(pkg_links) - logger.info( - "Matching suggestion for '%s': %d derivations, %d packages found.", - container.cve, - len(links), - len(pkg_links), + else None, + rejection_max_matches_limit=outcome.rejection.max_matches_limit or None + if outcome.rejection + else None, + algorithm_version=CVEDerivationClusterProposal.CURRENT_ALGORITHM_VERSION, ) - return True + if outcome.derivations: + links = build_derivation_links(proposal, outcome.derivations) + DerivationClusterProposalLink.objects.bulk_create(links) + pkg_links = build_package_links(proposal, outcome.derivations) + PackageClusterProposalLink.objects.bulk_create(pkg_links) + logger.info( + "Matching suggestion for '%s': %d derivations, %d packages found.", + container.cve, + len(links), + len(pkg_links), + ) + + return True + finally: + observe_matching(time.time() - start, candidates) def refresh_suggestion_derivation_links( diff --git a/src/shared/listeners/nix_evaluation.py b/src/shared/listeners/nix_evaluation.py index b2abbbe8..8e51d89c 100644 --- a/src/shared/listeners/nix_evaluation.py +++ b/src/shared/listeners/nix_evaluation.py @@ -22,6 +22,7 @@ parse_evaluation_result, ) from shared.git import GitRepo +from shared.metrics import observe_eval_batch_ingest from shared.models import NixDerivation, NixEvaluation logger = logging.getLogger(__name__) @@ -117,6 +118,7 @@ async def realtime_batch_process_attributes( drvs = await sync_to_async(ingester.ingest)() elapsed = time.time() - start + observe_eval_batch_ingest(elapsed, len(drvs)) logger.info( "%d attributes were ingested in %f seconds (%s, %s)", len(drvs), diff --git a/src/shared/management/commands/garbage_collect.py b/src/shared/management/commands/garbage_collect.py index 3d65ab80..364b9fc5 100644 --- a/src/shared/management/commands/garbage_collect.py +++ b/src/shared/management/commands/garbage_collect.py @@ -1,3 +1,4 @@ +import time from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser from datetime import timedelta from typing import Any @@ -6,7 +7,9 @@ from django.db import connection, models from django.db.models import Q, QuerySet from django.utils import timezone +from prometheus_client import CollectorRegistry, Gauge +from shared.metrics import write_metrics_textfile from shared.models import ( # type: ignore CVEDerivationClusterProposalStatusEvent, DerivationClusterProposalLinkEvent, @@ -25,6 +28,17 @@ DEFAULT_CUTOFF_DAYS = 365 +DELETED_KINDS = ( + "proposals", + "proposal_status_events", + "derivation_link_events", + "derivations", + "derivation_metas", + "evaluations", + "channels", + "stale_package_attrpaths", +) + class Command(BaseCommand): help = "Garbage collect stale proposals, derivations, evaluations and channels" @@ -82,22 +96,84 @@ def handle(self, *args: Any, **options: Any) -> None: # Each step satisfies the cascading constraints that gate the next step. # `pghistory` events are never auto-deleted — each step explicitly clears relevant events first. + deleted: dict[str, int] = {kind: 0 for kind in DELETED_KINDS} + step_durations: dict[str, float] = {} + start_total = time.time() + self.stdout.write("\n[1/6] Deleting stale matches") - self._delete_stale_matches(cutoff, batch_size, dry_run) + step_start = time.time() + for kind, count in self._delete_stale_matches( + cutoff, batch_size, dry_run + ).items(): + deleted[kind] += count + step_durations["stale_matches"] = time.time() - step_start + self.stdout.write("\n[2/6] Deleting unmatched derivations") - self._delete_unmatched_derivations(cutoff, batch_size, dry_run) + step_start = time.time() + for kind, count in self._delete_unmatched_derivations( + cutoff, batch_size, dry_run + ).items(): + deleted[kind] += count + step_durations["unmatched_derivations"] = time.time() - step_start + self.stdout.write("\n[3/5] Deleting empty evaluations") - self._delete_empty_evaluations(cutoff, batch_size, dry_run) + step_start = time.time() + for kind, count in self._delete_empty_evaluations( + cutoff, batch_size, dry_run + ).items(): + deleted[kind] += count + step_durations["empty_evaluations"] = time.time() - step_start + self.stdout.write("\n[4/5] Deleting inactive channels") - self._delete_inactive_channels(batch_size, dry_run) + step_start = time.time() + for kind, count in self._delete_inactive_channels(batch_size, dry_run).items(): + deleted[kind] += count + step_durations["inactive_channels"] = time.time() - step_start + self.stdout.write("\n[5/5] Pruning stale package attrpaths") - self._prune_stale_package_attrpaths(batch_size, dry_run) + step_start = time.time() + for kind, count in self._prune_stale_package_attrpaths( + batch_size, dry_run + ).items(): + deleted[kind] += count + step_durations["stale_attrpaths"] = time.time() - step_start + + total_duration = time.time() - start_total + self._write_metrics(total_duration, step_durations, deleted) self.stdout.write(self.style.SUCCESS("\nGarbage collection complete.")) + def _write_metrics( + self, + total_duration: float, + step_durations: dict[str, float], + deleted: dict[str, int], + ) -> None: + registry = CollectorRegistry() + duration_gauge = Gauge( + "sectracker_garbage_collect_duration_seconds", + "Duration of last garbage collection run by step", + ["step"], + registry=registry, + ) + duration_gauge.labels(step="total").set(total_duration) + for step, duration in step_durations.items(): + duration_gauge.labels(step=step).set(duration) + + deleted_gauge = Gauge( + "sectracker_garbage_collect_deleted", + "Rows deleted in last garbage collection run by kind", + ["kind"], + registry=registry, + ) + for kind, count in deleted.items(): + deleted_gauge.labels(kind=kind).set(float(count)) + + write_metrics_textfile("garbage_collection", registry) + def _delete_stale_matches( self, cutoff: Any, batch_size: int, dry_run: bool - ) -> None: + ) -> dict[str, int]: candidates = ( CVEDerivationClusterProposal.objects.filter( Q(created_at__lt=cutoff) @@ -121,7 +197,7 @@ def _delete_stale_matches( self.stdout.write(f"Found {total} eligible proposals.") if total == 0 or dry_run: - return + return {} proposal_ids = list(candidates.values_list("id", flat=True)) link_ids = list( @@ -130,28 +206,29 @@ def _delete_stale_matches( ).values_list("id", flat=True) ) - self._purge_events( - CVEDerivationClusterProposalStatusEvent, - pgh_obj_id__in=proposal_ids, - label="proposal status events", - ) - self._purge_events( - DerivationClusterProposalLinkEvent, - pgh_obj_id__in=link_ids, - label="derivation link events", - ) - - self._delete_in_batches( - qs=candidates, - model=CVEDerivationClusterProposal, - pk_field="id", - label="proposals", - batch_size=batch_size, - ) + return { + "proposal_status_events": self._purge_events( + CVEDerivationClusterProposalStatusEvent, + pgh_obj_id__in=proposal_ids, + label="proposal status events", + ), + "derivation_link_events": self._purge_events( + DerivationClusterProposalLinkEvent, + pgh_obj_id__in=link_ids, + label="derivation link events", + ), + "proposals": self._delete_in_batches( + qs=candidates, + model=CVEDerivationClusterProposal, + pk_field="id", + label="proposals", + batch_size=batch_size, + ), + } def _delete_unmatched_derivations( self, cutoff: Any, batch_size: int, dry_run: bool - ) -> None: + ) -> dict[str, int]: failed_crashed = NixDerivation.objects.filter( parent_evaluation__state__in=[ NixEvaluation.EvaluationState.FAILED, @@ -173,28 +250,28 @@ def _delete_unmatched_derivations( ) ) - self._delete_in_batches( - qs=candidates, - model=NixDerivation, - pk_field="id", - label="derivations", - batch_size=batch_size, - dry_run=dry_run, - ) - - meta_candidates = NixDerivationMeta.objects.filter(pk__in=meta_ids) - self._delete_in_batches( - qs=meta_candidates, - model=NixDerivationMeta, - pk_field="id", - label="derivation metas", - batch_size=batch_size, - dry_run=dry_run, - ) + return { + "derivations": self._delete_in_batches( + qs=candidates, + model=NixDerivation, + pk_field="id", + label="derivations", + batch_size=batch_size, + dry_run=dry_run, + ), + "derivation_metas": self._delete_in_batches( + qs=NixDerivationMeta.objects.filter(pk__in=meta_ids), + model=NixDerivationMeta, + pk_field="id", + label="derivation metas", + batch_size=batch_size, + dry_run=dry_run, + ), + } def _delete_empty_evaluations( self, cutoff: Any, batch_size: int, dry_run: bool - ) -> None: + ) -> dict[str, int]: candidates = NixEvaluation.objects.filter( state__in=[ NixEvaluation.EvaluationState.FAILED, @@ -203,16 +280,20 @@ def _delete_empty_evaluations( derivations__isnull=True, ) - self._delete_in_batches( - qs=candidates, - model=NixEvaluation, - pk_field="id", - label="evaluations", - batch_size=batch_size, - dry_run=dry_run, - ) + return { + "evaluations": self._delete_in_batches( + qs=candidates, + model=NixEvaluation, + pk_field="id", + label="evaluations", + batch_size=batch_size, + dry_run=dry_run, + ) + } - def _delete_inactive_channels(self, batch_size: int, dry_run: bool) -> None: + def _delete_inactive_channels( + self, batch_size: int, dry_run: bool + ) -> dict[str, int]: candidates = ( NixChannel.objects.filter( state__in=[ @@ -235,32 +316,38 @@ def _delete_inactive_channels(self, batch_size: int, dry_run: bool) -> None: self.stdout.write(f"Found {total} eligible inactive channels.") if total == 0 or dry_run: - return - - self._delete_in_batches( - qs=candidates, - model=NixChannel, - pk_field="channel_branch", - label="channels", - batch_size=batch_size, - ) - - def _prune_stale_package_attrpaths(self, batch_size: int, dry_run: bool) -> None: - self._delete_in_batches( - qs=PackageAttrpath.objects.stale(), - model=PackageAttrpath, - pk_field="attrpath", - label="stale package attrpaths", - batch_size=batch_size, - dry_run=dry_run, - ) + return {} + + return { + "channels": self._delete_in_batches( + qs=candidates, + model=NixChannel, + pk_field="channel_branch", + label="channels", + batch_size=batch_size, + ) + } + + def _prune_stale_package_attrpaths( + self, batch_size: int, dry_run: bool + ) -> dict[str, int]: + return { + "stale_package_attrpaths": self._delete_in_batches( + qs=PackageAttrpath.objects.stale(), + model=PackageAttrpath, + pk_field="attrpath", + label="stale package attrpaths", + batch_size=batch_size, + dry_run=dry_run, + ) + } def _purge_events( self, event_model: type[models.Model], label: str, **filter_kwargs: Any, - ) -> None: + ) -> int: table_name = event_model._meta.db_table try: @@ -272,6 +359,7 @@ def _purge_events( deleted, _ = event_model.objects.filter(**filter_kwargs).delete() self.stdout.write(f"Purged {deleted} {label}.") + return deleted finally: with connection.cursor() as cursor: cursor.execute(f"ALTER TABLE {table_name} ENABLE TRIGGER ALL") @@ -284,12 +372,12 @@ def _delete_in_batches( label: str, batch_size: int, dry_run: bool = False, - ) -> None: + ) -> int: total = qs.count() self.stdout.write(f"Found {total} eligible {label}.") if total == 0 or dry_run: - return + return 0 deleted_total = 0 batch_num = 1 @@ -307,3 +395,4 @@ def _delete_in_batches( batch_num += 1 self.stdout.write(self.style.SUCCESS(f"Done. Deleted {deleted_total} {label}.")) + return deleted_total diff --git a/src/shared/management/commands/ingest_delta_cve.py b/src/shared/management/commands/ingest_delta_cve.py index 40228898..597e2463 100644 --- a/src/shared/management/commands/ingest_delta_cve.py +++ b/src/shared/management/commands/ingest_delta_cve.py @@ -3,6 +3,7 @@ import json import logging import tempfile +import time import zipfile from concurrent.futures import ProcessPoolExecutor, as_completed from glob import glob @@ -13,10 +14,12 @@ from django.db import transaction from github import UnknownObjectException from github.Repository import Repository +from prometheus_client import CollectorRegistry, Gauge from shared import models from shared.fetchers import make_cve from shared.github import get_gh +from shared.metrics import write_metrics_textfile from shared.models import CveIngestion logger = logging.getLogger(__name__) @@ -34,7 +37,7 @@ def __init__(self, day: datetime.datetime) -> None: super().__init__(f"No release for day {day}") -def ingest_day(repo: Repository, day: datetime.datetime) -> CveIngestion: +def ingest_day(repo: Repository, day: datetime.datetime) -> int: # Fetch the latest daily release try: release = repo.get_release(f"cve_{day}_at_end_of_day") @@ -91,17 +94,19 @@ def ingest_day(repo: Repository, day: datetime.datetime) -> CveIngestion: # Record the ingestion logger.info(f"Saving the ingestion valid up to {day}") - return CveIngestion.objects.create(valid_to=day, delta=True) + CveIngestion.objects.create(valid_to=day, delta=True) + return len(cve_list) -def process_day(repo: Repository, day: datetime.datetime) -> None: - """Wrapper function to process a single day.""" +def process_day(repo: Repository, day: datetime.datetime) -> tuple[int, bool]: + """Process a single day. Returns (cves_ingested, skipped).""" try: - ingest_day(repo, day) + return ingest_day(repo, day), False except NoReleaseError: logger.exception( f"CVE ingestion on day {day} is impossible as there's no release, continuing for the next days" ) + return 0, True def parallel_ingestion( @@ -109,7 +114,7 @@ def parallel_ingestion( next_ingestion: datetime.datetime, until_date: datetime.datetime, num_processes: int, -) -> None: +) -> tuple[int, int]: """ Parallel ingestion of CVE data. @@ -117,12 +122,16 @@ def parallel_ingestion( :param next_ingestion: The starting date for ingestion. :param date: The end date for ingestion. :param num_processes: Number of parallel processes. + :returns: (cves_ingested, days_succeeded) """ days = [ next_ingestion + datetime.timedelta(days=x) for x in range((until_date - next_ingestion).days + 1) ] + cves_ingested = 0 + days_succeeded = 0 + with ProcessPoolExecutor(max_workers=num_processes) as executor: # Submit tasks to the executor futures = {executor.submit(process_day, repo, day): day for day in days} @@ -130,10 +139,15 @@ def parallel_ingestion( for future in as_completed(futures): day = futures[future] try: - future.result() + day_cves, skipped = future.result() + if not skipped: + cves_ingested += day_cves + days_succeeded += 1 except Exception as e: logger.exception(f"An error occurred while processing day {day}: {e}") + return cves_ingested, days_succeeded + class Command(BaseCommand): help = "Ingest CVEs day per day using the MITRE CVE repo" @@ -164,6 +178,9 @@ def handle(self, *args: Any, **kwargs: Any) -> None: until_date = kwargs["date"] default_start_ingestion = kwargs["default_start_ingestion"] num_processes = kwargs["num_parallel_processes"] + start = time.time() + cves_ingested = 0 + days_succeeded = 0 if num_processes > 1: logger.warning( @@ -174,7 +191,7 @@ def handle(self, *args: Any, **kwargs: Any) -> None: logger.warning( f"The database already contains the delta contents from {until_date}." ) - + self._write_metrics(time.time() - start, cves_ingested, days_succeeded) return last_ingestion = ( @@ -199,4 +216,28 @@ def handle(self, *args: Any, **kwargs: Any) -> None: # Select the CVEList repository repo = g.get_repo("CVEProject/cvelistV5") - parallel_ingestion(repo, next_ingestion, until_date, num_processes) + cves_ingested, days_succeeded = parallel_ingestion( + repo, next_ingestion, until_date, num_processes + ) + self._write_metrics(time.time() - start, cves_ingested, days_succeeded) + + def _write_metrics( + self, duration: float, cves_ingested: int, days_succeeded: int + ) -> None: + registry = CollectorRegistry() + Gauge( + "sectracker_cve_delta_ingest_duration_seconds", + "Duration of last CVE delta ingest run", + registry=registry, + ).set(duration) + Gauge( + "sectracker_cve_delta_ingest_cves", + "CVEs ingested in last CVE delta ingest run", + registry=registry, + ).set(float(cves_ingested)) + Gauge( + "sectracker_cve_delta_ingest_days", + "Days successfully ingested in last CVE delta ingest run", + registry=registry, + ).set(float(days_succeeded)) + write_metrics_textfile("cve_delta_ingest", registry) diff --git a/src/shared/management/commands/serve_worker_metrics.py b/src/shared/management/commands/serve_worker_metrics.py new file mode 100644 index 00000000..5755f139 --- /dev/null +++ b/src/shared/management/commands/serve_worker_metrics.py @@ -0,0 +1,30 @@ +"""Serve aggregated prometheus_client multiprocess metrics for the evaluator.""" + +from __future__ import annotations + +import time +from typing import Any + +from django.conf import settings +from django.core.management.base import BaseCommand, CommandError + +from shared.metrics import serve_multiprocess_metrics + + +class Command(BaseCommand): + help = ( + "Serve Prometheus /metrics from PROMETHEUS_MULTIPROC_DIR " + "(evaluator metrics sidecar)." + ) + + def handle(self, *args: Any, **options: Any) -> None: + port = settings.METRICS_HTTP_PORT + if port is None: + raise CommandError("METRICS_HTTP_PORT must be set to serve worker metrics") + + serve_multiprocess_metrics(port) + self.stdout.write( + self.style.SUCCESS(f"Serving multiprocess metrics on port {port}") + ) + while True: + time.sleep(3600) diff --git a/src/shared/metrics.py b/src/shared/metrics.py index a003d26b..f20dae0e 100644 --- a/src/shared/metrics.py +++ b/src/shared/metrics.py @@ -2,8 +2,42 @@ from __future__ import annotations +import logging +import os + from django.conf import settings -from prometheus_client import CollectorRegistry, write_to_textfile +from prometheus_client import ( + CollectorRegistry, + Histogram, + start_http_server, + write_to_textfile, +) +from prometheus_client.multiprocess import MultiProcessCollector + +logger = logging.getLogger(__name__) + +_matching_duration = Histogram( + "sectracker_matching_duration_seconds", + "Duration of CVE-to-derivation matching (build_new_links)", + buckets=(0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, float("inf")), +) +_matching_candidates = Histogram( + "sectracker_matching_candidates", + "Candidate derivations considered during matching", + buckets=(0, 1, 5, 10, 25, 50, 100, 250, 500, 1000, 5000, float("inf")), +) +_eval_batch_duration = Histogram( + "sectracker_nix_evaluation_batch_ingest_duration_seconds", + "Duration of realtime Nixpkgs evaluation attribute batch ingest", + buckets=(0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, float("inf")), +) +_eval_batch_attributes = Histogram( + "sectracker_nix_evaluation_batch_ingest_attributes", + "Attributes ingested per realtime evaluation batch", + buckets=(1, 5, 10, 25, 50, 100, 250, 500, 1000, 5000, 25000, float("inf")), +) + +_metrics_server_started = False def write_metrics_textfile(name: str, registry: CollectorRegistry) -> None: @@ -14,3 +48,43 @@ def write_metrics_textfile(name: str, registry: CollectorRegistry) -> None: return prom_path = (settings.METRICS_TEXTFILE_DIR / name).with_suffix(".prom") write_to_textfile(str(prom_path), registry) + + +def observe_matching(duration_seconds: float, candidates: int) -> None: + """Record matching wall time and candidate cardinality.""" + _matching_duration.observe(duration_seconds) + _matching_candidates.observe(float(candidates)) + + +def observe_eval_batch_ingest(duration_seconds: float, attributes: int) -> None: + """Record eval batch ingest wall time and batch size.""" + _eval_batch_duration.observe(duration_seconds) + _eval_batch_attributes.observe(float(attributes)) + + +def start_worker_metrics_server() -> None: + """ + Expose the default registry on METRICS_HTTP_PORT. + + No-op when the port is unset, when already started, or when multiprocess + mode is active (the evaluator sidecar serves MultiProcessCollector instead). + """ + global _metrics_server_started + if _metrics_server_started: + return + if os.environ.get("PROMETHEUS_MULTIPROC_DIR"): + return + port = settings.METRICS_HTTP_PORT + if port is None: + return + start_http_server(port) + _metrics_server_started = True + logger.info("Prometheus worker metrics listening on port %s", port) + + +def serve_multiprocess_metrics(port: int) -> None: + """Serve aggregated multiprocess metrics (evaluator sidecar).""" + registry = CollectorRegistry() + MultiProcessCollector(registry) + start_http_server(port, registry=registry) + logger.info("Prometheus multiprocess metrics listening on port %s", port) diff --git a/src/shared/tests/test_metrics.py b/src/shared/tests/test_metrics.py index bf01275a..691724e2 100644 --- a/src/shared/tests/test_metrics.py +++ b/src/shared/tests/test_metrics.py @@ -3,7 +3,12 @@ from django.test import override_settings from prometheus_client import CollectorRegistry, Gauge -from shared.metrics import write_metrics_textfile +from shared.metrics import ( + observe_eval_batch_ingest, + observe_matching, + start_worker_metrics_server, + write_metrics_textfile, +) def test_write_metrics_textfile_produces_prometheus_format(tmp_path: Path) -> None: @@ -52,3 +57,128 @@ def test_write_metrics_textfile_writes_cache_regeneration_metrics( ) in content assert "sectracker_cache_regeneration_duration_seconds 42.0" in content assert "sectracker_cache_regeneration_suggestions 7.0" in content + + +def test_write_metrics_textfile_writes_garbage_collection_metrics( + tmp_path: Path, +) -> None: + with override_settings(METRICS_TEXTFILE_DIR=tmp_path): + registry = CollectorRegistry() + duration = Gauge( + "sectracker_garbage_collect_duration_seconds", + "Duration of last garbage collection run by step", + ["step"], + registry=registry, + ) + duration.labels(step="total").set(12.0) + duration.labels(step="stale_matches").set(3.5) + deleted = Gauge( + "sectracker_garbage_collect_deleted", + "Rows deleted in last garbage collection run by kind", + ["kind"], + registry=registry, + ) + deleted.labels(kind="proposals").set(4.0) + deleted.labels(kind="derivations").set(9.0) + + write_metrics_textfile("garbage_collection", registry) + + content = (tmp_path / "garbage_collection.prom").read_text() + assert ( + "# HELP sectracker_garbage_collect_duration_seconds " + "Duration of last garbage collection run by step" + ) in content + assert ( + "# HELP sectracker_garbage_collect_deleted " + "Rows deleted in last garbage collection run by kind" + ) in content + assert 'sectracker_garbage_collect_duration_seconds{step="total"} 12.0' in content + assert ( + 'sectracker_garbage_collect_duration_seconds{step="stale_matches"} 3.5' + in content + ) + assert 'sectracker_garbage_collect_deleted{kind="proposals"} 4.0' in content + assert 'sectracker_garbage_collect_deleted{kind="derivations"} 9.0' in content + + +def test_write_metrics_textfile_writes_cve_delta_ingest_metrics( + tmp_path: Path, +) -> None: + with override_settings(METRICS_TEXTFILE_DIR=tmp_path): + registry = CollectorRegistry() + Gauge( + "sectracker_cve_delta_ingest_duration_seconds", + "Duration of last CVE delta ingest run", + registry=registry, + ).set(8.25) + Gauge( + "sectracker_cve_delta_ingest_cves", + "CVEs ingested in last CVE delta ingest run", + registry=registry, + ).set(15.0) + Gauge( + "sectracker_cve_delta_ingest_days", + "Days successfully ingested in last CVE delta ingest run", + registry=registry, + ).set(2.0) + + write_metrics_textfile("cve_delta_ingest", registry) + + content = (tmp_path / "cve_delta_ingest.prom").read_text() + assert ( + "# HELP sectracker_cve_delta_ingest_duration_seconds " + "Duration of last CVE delta ingest run" + ) in content + assert ( + "# HELP sectracker_cve_delta_ingest_cves " + "CVEs ingested in last CVE delta ingest run" + ) in content + assert ( + "# HELP sectracker_cve_delta_ingest_days " + "Days successfully ingested in last CVE delta ingest run" + ) in content + assert "sectracker_cve_delta_ingest_duration_seconds 8.25" in content + assert "sectracker_cve_delta_ingest_cves 15.0" in content + assert "sectracker_cve_delta_ingest_days 2.0" in content + + +def test_observe_matching_records_histogram_samples() -> None: + from prometheus_client import REGISTRY + + before = ( + REGISTRY.get_sample_value("sectracker_matching_duration_seconds_count") or 0.0 + ) + observe_matching(0.12, 7) + after = ( + REGISTRY.get_sample_value("sectracker_matching_duration_seconds_count") or 0.0 + ) + assert after == before + 1.0 + candidates = ( + REGISTRY.get_sample_value("sectracker_matching_candidates_count") or 0.0 + ) + assert candidates >= 1.0 + + +def test_observe_eval_batch_ingest_records_histogram_samples() -> None: + from prometheus_client import REGISTRY + + before = ( + REGISTRY.get_sample_value( + "sectracker_nix_evaluation_batch_ingest_duration_seconds_count" + ) + or 0.0 + ) + observe_eval_batch_ingest(0.45, 12) + after = ( + REGISTRY.get_sample_value( + "sectracker_nix_evaluation_batch_ingest_duration_seconds_count" + ) + or 0.0 + ) + assert after == before + 1.0 + + +def test_start_worker_metrics_server_noop_without_port() -> None: + with override_settings(METRICS_HTTP_PORT=None): + # Must not raise when port is unset. + start_worker_metrics_server()