Improve performance of CSAF ingestion

integration-tests/Cargo.toml

@@ -12,6 +12,7 @@ trustify-module-storage = { workspace = true }
 
 anyhow = { workspace = true }
 bytes = { workspace = true }
+csaf = { workspace = true }
 cyclonedx-bom = { workspace = true }
 humantime = { workspace = true }
 log = { workspace = true }

integration-tests/src/csaf/mod.rs

@@ -0,0 +1 @@
+mod perf;

integration-tests/src/csaf/perf.rs

@@ -0,0 +1,32 @@
+#![cfg(test)]
+
+use std::time::Instant;
+use test_context::test_context;
+use test_log::test;
+use tracing::instrument;
+use trustify_common::{db::test::TrustifyContext, hashing::Digests};
+use trustify_module_ingestor::{graph::Graph, service::advisory::csaf::loader::CsafLoader};
+
+#[test_context(TrustifyContext, skip_teardown)]
+#[test(tokio::test)]
+#[instrument]
+async fn ingest(ctx: TrustifyContext) -> anyhow::Result<()> {
+    let db = ctx.db;
+    let graph = Graph::new(db.clone());
+
+    let start = Instant::now();
+
+    // let data = include_bytes!("../../../etc/test-data/csaf/CVE-2023-20862.json");
+    let data = include_bytes!("../../../etc/test-data/csaf/cve-2023-33201.json");
+
+    let digests = Digests::digest(data);
+    CsafLoader::new(&graph)
+        .load((), &data[..], &digests)
+        .await?;
+
+    let ingest_time = start.elapsed();
+
+    log::info!("ingest: {}", humantime::Duration::from(ingest_time));
+
+    Ok(())
+}

integration-tests/src/lib.rs

@@ -1 +1,3 @@
+mod csaf;
 mod sbom;
+mod stream;

integration-tests/src/sbom/mod.rs

@@ -1,4 +1,3 @@
 mod graph;
 mod reingest;
-mod stream;
 mod test;

integration-tests/src/sbom/reingest/mod.rs

@@ -1,10 +1,9 @@
 //! Testing to re-ingest a document, ensuring there is not stale data
 #![cfg(test)]
 
-use crate::sbom::stream::xz_stream;
+use crate::stream::{stream, xz_stream};
 use bytes::Bytes;
 use serde_json::Value;
-use std::convert::Infallible;
 use std::str::FromStr;
 use test_context::futures::stream;
 use test_context::test_context;
@@ -39,11 +38,9 @@ async fn quarkus(ctx: TrustifyContext) -> Result<(), anyhow::Error> {
             ("source", "test"),
             None,
             Format::SPDX,
-            stream::once(async {
-                Ok::<_, Infallible>(Bytes::from_static(include_bytes!(
-                    "data/quarkus/v1/quarkus-bom-2.13.8.Final-redhat-00004.json"
-                )))
-            }),
+            stream(include_bytes!(
+                "data/quarkus/v1/quarkus-bom-2.13.8.Final-redhat-00004.json"
+            )),
         )
         .await?;
 
@@ -56,11 +53,9 @@ async fn quarkus(ctx: TrustifyContext) -> Result<(), anyhow::Error> {
             ("source", "test"),
             None,
             Format::SPDX,
-            stream::once(async {
-                Ok::<_, Infallible>(Bytes::from_static(include_bytes!(
-                    "data/quarkus/v2/quarkus-bom-2.13.8.Final-redhat-00004.json"
-                )))
-            }),
+            stream(include_bytes!(
+                "data/quarkus/v2/quarkus-bom-2.13.8.Final-redhat-00004.json"
+            )),
         )
         .await?;
 

integration-tests/src/sbom/test/mod.rs

@@ -166,7 +166,6 @@ where
 
     let start = Instant::now();
     i(&ctx, sbom, &tx).await?;
-    // sbom.ingest_spdx(sbom.clone(), &tx).await?;
     let ingest_time_2 = start.elapsed();
 
     // commit

integration-tests/src/sbom/stream.rs → integration-tests/src/stream.rs

@@ -2,6 +2,7 @@
 
 use bytes::Bytes;
 use lzma::LzmaError;
+use std::convert::Infallible;
 
 use test_context::futures::{stream, Stream};
 
@@ -12,3 +13,8 @@ pub fn xz_stream(data: &[u8]) -> impl Stream<Item = Result<Bytes, LzmaError>> {
     let result = lzma::decompress(data).map(|data| data.into());
     stream::once(async { result })
 }
+
+/// Create a stream from a static BLOB.
+pub fn stream(data: &'static [u8]) -> impl Stream<Item = Result<Bytes, Infallible>> {
+    stream::once(async move { Ok(Bytes::from_static(data)) })
+}

migration/src/lib.rs

@@ -43,6 +43,7 @@ mod m0000350_remove_old_assertion_tables;
 mod m0000355_labels;
 mod m0000360_add_sbom_file;
 mod m0000370_add_cwe;
+mod m0000380_create_package_status_index;
 
 pub struct Migrator;
 
@@ -92,6 +93,7 @@ impl MigratorTrait for Migrator {
             Box::new(m0000355_labels::Migration),
             Box::new(m0000360_add_sbom_file::Migration),
             Box::new(m0000370_add_cwe::Migration),
+            Box::new(m0000380_create_package_status_index::Migration),
         ]
     }
 }

migration/src/m0000380_create_package_status_index.rs

@@ -0,0 +1,43 @@
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .create_index(
+                Index::create()
+                    .table(PackageStatus::Table)
+                    .name("package_status_idx")
+                    .col(PackageStatus::PackageId)
+                    .col(PackageStatus::AdvisoryId)
+                    .col(PackageStatus::StatusId)
+                    .to_owned(),
+            )
+            .await?;
+
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .drop_index(
+                Index::drop()
+                    .table(PackageStatus::Table)
+                    .name("package_status_idx")
+                    .if_exists()
+                    .to_owned(),
+            )
+            .await
+    }
+}
+
+#[derive(DeriveIden)]
+enum PackageStatus {
+    Table,
+    AdvisoryId,
+    StatusId,
+    PackageId,
+}

modules/ingestor/src/graph/advisory/advisory_vulnerability.rs

@@ -31,7 +31,7 @@ pub enum Version {
 }
 
 impl VersionInfo {
-    fn into_active_model(self) -> version_range::ActiveModel {
+    pub fn into_active_model(self) -> version_range::ActiveModel {
         version_range::ActiveModel {
             id: Default::default(),
             version_scheme_id: Set(self.scheme),
@@ -400,6 +400,7 @@ impl<'g> AdvisoryVulnerabilityContext<'g> {
             .map(|cvss| cvss.into()))
     }
 
+    #[instrument(skip(self, tx), err)]
     pub async fn ingest_cvss3_score<TX: AsRef<Transactional>>(
         &self,
         cvss3: Cvss3Base,

modules/ingestor/src/graph/advisory/mod.rs

@@ -10,6 +10,7 @@ use sea_orm::{ColumnTrait, QuerySelect, RelationTrait};
 use sea_query::{Condition, JoinType};
 use std::fmt::{Debug, Formatter};
 use time::OffsetDateTime;
+use tracing::instrument;
 use trustify_common::db::Transactional;
 use trustify_common::hashing::Digests;
 use trustify_entity as entity;
@@ -65,6 +66,7 @@ impl Graph {
             .map(|advisory| AdvisoryContext::new(self, advisory)))
     }
 
+    #[instrument(skip(self, tx), err)]
     pub async fn get_advisory_by_digest<TX: AsRef<Transactional>>(
         &self,
         sha256: &str,
@@ -90,9 +92,10 @@ impl Graph {
             .collect())
     }
 
+    #[instrument(skip(self, labels, information, tx), err)]
     pub async fn ingest_advisory<TX: AsRef<Transactional>>(
         &self,
-        identifier: impl Into<String>,
+        identifier: impl Into<String> + Debug,
         labels: impl Into<Labels>,
         digests: &Digests,
         information: impl Into<AdvisoryInformation>,
@@ -200,6 +203,7 @@ impl<'g> AdvisoryContext<'g> {
         self.advisory.withdrawn
     }
 
+    #[instrument(skip(self, tx), err)]
     pub async fn get_vulnerability<TX: AsRef<Transactional>>(
         &self,
         identifier: &str,
@@ -217,6 +221,7 @@ impl<'g> AdvisoryContext<'g> {
             .map(|vuln| (self, vuln).into()))
     }
 
+    #[instrument(skip(self, information, tx), err)]
     pub async fn link_to_vulnerability<TX: AsRef<Transactional>>(
         &self,
         identifier: &str,

modules/ingestor/src/graph/organization.rs

@@ -1,5 +1,6 @@
 use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};
-
+use std::fmt::Debug;
+use tracing::instrument;
 use trustify_common::db::Transactional;
 use trustify_entity::organization;
 
@@ -52,9 +53,10 @@ impl super::Graph {
             .map(|organization| OrganizationContext::new(self, organization)))
     }
 
+    #[instrument(skip(self, information, tx), err)]
     pub async fn ingest_organization<TX: AsRef<Transactional>>(
         &self,
-        name: impl Into<String>,
+        name: impl Into<String> + Debug,
         information: impl Into<OrganizationInformation>,
         tx: TX,
     ) -> Result<OrganizationContext, Error> {

modules/ingestor/src/graph/vulnerability/mod.rs

@@ -10,6 +10,7 @@ use sea_orm::{
 use sea_query::{Condition, JoinType};
 use std::fmt::{Debug, Formatter};
 use time::OffsetDateTime;
+use tracing::instrument;
 use trustify_common::db::Transactional;
 use trustify_entity as entity;
 use trustify_entity::{advisory, advisory_vulnerability, vulnerability, vulnerability_description};
@@ -46,6 +47,7 @@ impl From<()> for VulnerabilityInformation {
 }
 
 impl Graph {
+    #[instrument(skip(self, information, tx), err)]
     pub async fn ingest_vulnerability<
         I: Into<VulnerabilityInformation>,
         TX: AsRef<Transactional>,
@@ -55,6 +57,7 @@ impl Graph {
         information: I,
         tx: TX,
     ) -> Result<VulnerabilityContext, Error> {
+        // TODO: consider transforming into upsert
         let information = information.into();
         if let Some(found) = self.get_vulnerability(identifier, &tx).await? {
             if information.has_data() {
@@ -87,6 +90,7 @@ impl Graph {
         }
     }
 
+    #[instrument(skip(self, tx), err)]
     pub async fn get_vulnerability<TX: AsRef<Transactional>>(
         &self,
         identifier: &str,