Add malware advisory ingestion pipeline

Bala Kumarrequested to merge
malware-advisories-ingestion-602431 into master

What does this MR do and why?

Wave-1 ingestion pipeline for the malware advisory sync (epic gitlab-org&20876). It parses GLAM v3 NDJSON records into value objects and upserts them into the malware advisory tables (pm_malware_advisories, pm_malware_affected_packages).

Changes

  • MalwareAdvisoryDataObject + MalwareAdvisoryDataObjectFabricator — wrap a parsed GLAM v3 record (one advisory plus its affected package(s)).
  • MalwareAdvisoryIngestionService — orchestrates advisory then affected-package ingestion in a single transaction. The malware tables live on the sec database (SecApplicationRecord), unlike the other pm_ tables (gitlab_pm), so the wrapping transaction runs on that connection to avoid a cross-database write.
  • Ingestion::MalwareAdvisory::MalwareAdvisoryIngestionTask and MalwareAffectedPackageIngestionTask — use validation-aware bulk_upsert!, mirroring the public PackageMetadata::Advisory ingestion: each record is validated individually, invalid records are logged via Gitlab::ErrorTracking and skipped, so a single malformed record never fails the whole batch. Both tasks also de-duplicate on their upsert conflict key so a batch that repeats a record can't hit ON CONFLICT DO UPDATE ... cannot affect row a second time.
  • Feature-flag gated — MalwareAdvisoryIngestionService skips the entire upsert unless the ingest_malware_advisories flag is enabled (type: wip, default off, #604584). With it off, the sync still fetches and parses from PDS but writes nothing to pm_malware_*, so fetch/parse can be validated before enabling DB writes (pairs with sync_malware_advisories, #604583).
  • Specs for both tasks and the orchestrating service (including the flag-off "no writes" path).

Why bulk_upsert! instead of upsert_all

The POC used upsert_all, which bypasses ActiveRecord validations and sends the chunk in one statement — so a single bad record (e.g. an over-limit affected_range) raised PG::CheckViolation and dropped the entire batch. This MR adopts the public-advisory pattern: per-record validate → log → skip, then bulk_upsert! the valid records.

Event publishing deferred

The POC published an IngestedMalwareAdvisoryEvent per recent advisory. Event publishing is only needed for the downstream SBOM / CVS scanning integration, not for the sync itself, so it is removed here and deferred to #606613, under epic gitlab-org&21156.

Local testing

Ran the ingestion service directly against the sec database on a GDK, feeding two GLAM v3 records through PackageMetadata::MalwareAdvisoryIngestionService.execute:

run 1: advisories +2, affected +2
run 2 (idempotent): advisories +2, affected +2      # re-run creates no duplicates
advisory[0] urls=["https://example.test/GLAM-90001"] withdrawn_date=nil
advisory[1] withdrawn_date=2025-05-01                 # withdrawn captured
advisory[0] affected=[["pkg-GLAM-2099-01-90001", ">=0"]]
  • Advisories and their affected packages are upserted together in one sec-DB transaction.
  • Re-running is idempotent (upsert, no duplicate rows).
  • urls and withdrawn_date are persisted.

The full pipeline (PDS / offline sync → this ingestion) was also validated end-to-end against the GLAM v3 npm example dataset via the sync MRs — 212,988 (full) / 211,316 (delta) advisories, 0 records dropped.

Specs

bundle exec rspec \
  ee/spec/services/package_metadata/ingestion/malware_advisory/ \
  ee/spec/services/package_metadata/malware_advisory_ingestion_service_spec.rb
# 10 examples, 0 failures

MR acceptance checklist

  • Ingestion is resilient: invalid records are logged and skipped, and the batch continues.
  • Unit specs for both tasks and the service (validate / skip / log, advisory id map).
  • Idempotent upsert verified locally; urls / withdrawn_date persisted.
  • Malware schema (#589555 (closed)) is merged to master; the specs are green in CI.

Related to #602431.

Database review — upsert query plans

This MR introduces the malware advisory upserts (bulk_upsert! in the two ingestion tasks). bulk_upsert! batches at BulkInsertSafe::DEFAULT_BATCH_SIZE = 500, so the database sees 500-row INSERT … ON CONFLICT … DO UPDATE upserts, one per malware table (both on the sec database).

1. pm_malware_advisories

INSERT INTO "pm_malware_advisories"
  ("advisory_xid", "source_xid", "title", "description", "published_date",
   "withdrawn_date", "identifiers", "urls", "created_at", "updated_at")
VALUES (...), (...)   -- 500 value rows
ON CONFLICT ("advisory_xid", "source_xid") DO UPDATE SET
  "title" = excluded."title", "description" = excluded."description",
  "published_date" = excluded."published_date", "withdrawn_date" = excluded."withdrawn_date",
  "identifiers" = excluded."identifiers", "urls" = excluded."urls",
  "updated_at" = excluded."updated_at"
RETURNING "advisory_xid", "id";

Database Lab (gitlab-production-sec clone, 500-row batch):

Conflict arbiter index i_pm_malware_advisories_xid_source — no sequential scan.

2. pm_malware_affected_packages

INSERT INTO "pm_malware_affected_packages"
  ("pm_malware_advisory_id", "purl_type", "package_name", "affected_range", "created_at", "updated_at")
VALUES (...), (...)   -- 500 value rows
ON CONFLICT ("pm_malware_advisory_id", "purl_type", "package_name") DO UPDATE SET
  "affected_range" = excluded."affected_range", "updated_at" = excluded."updated_at";

Database Lab (gitlab-production-sec clone, 500-row batch):

Conflict arbiter index i_pm_malware_affected_packages_unique_for_upsert — no sequential scan. Because pm_malware_advisory_id is an FK, the Database Lab batch sourced its ids from existing advisories (ORDER BY random() LIMIT 500), so the linked plans include a small sample node ahead of the upsert; the upsert/conflict cost is the relevant part.

Edited by Bala Kumar

Merge request reports