feat: expose worker Prometheus metrics for matching and eval ingest

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 = {

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
           '';

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.

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

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(

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),