feat: make max concurrent importers configurable

Signed-off-by: Jim Crossley <[email protected]>

Author: jcrossley3

modules/importer/src/server/mod.rs

@@ -24,12 +24,14 @@ pub async fn importer(
     storage: DispatchBackend,
     working_dir: Option<PathBuf>,
     analysis: Option<AnalysisService>,
+    concurrency: usize,
 ) -> anyhow::Result<()> {
     Server {
         db,
         storage,
         working_dir,
         analysis,
+        concurrency,
     }
     .run()
     .await
@@ -56,6 +58,7 @@ struct Server {
     storage: DispatchBackend,
     working_dir: Option<PathBuf>,
     analysis: Option<AnalysisService>,
+    concurrency: usize,
 }
 
 impl Server {
@@ -70,15 +73,16 @@ impl Server {
 
         loop {
             interval.tick().await;
-            tracing::debug!(jim = "", "checking importers");
-
-            let importers = service
-                .list()
-                .await?
-                .into_iter()
-                .filter(|i| !(i.data.configuration.disabled || can_wait(i)))
-                .map(|i| self.import(i, &service));
-            futures::future::join_all(importers).await;
+            futures::future::join_all(
+                service
+                    .list()
+                    .await?
+                    .into_iter()
+                    .filter(|i| !(i.data.configuration.disabled || can_wait(i)))
+                    .take(self.concurrency)
+                    .map(|i| self.import(i, &service)),
+            )
+            .await;
         }
     }
 

modules/importer/src/service.rs

@@ -110,7 +110,7 @@ impl ImporterService {
             .into_iter()
             .map(Importer::try_from)
             .collect::<Result<_, _>>()?;
-        result.sort_unstable_by_key(|i| (i.data.configuration.disabled, i.name.clone()));
+        result.sort_unstable_by_key(|i| (i.data.configuration.disabled, i.data.last_run));
         Ok(result)
     }
 

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

@@ -45,7 +45,7 @@ impl AdvisoryVulnerabilityContext<'_> {
         )
     }
 
-    #[instrument(skip(self, connection), ret)]
+    #[instrument(skip(self, connection))]
     pub async fn ingest_package_status<C: ConnectionTrait>(
         &self,
         cpe_context: Option<Cpe>,

server/src/profile/importer.rs

@@ -49,9 +49,18 @@ use utoipa_redoc::{Redoc, Servable};
 #[derive(clap::Args, Debug)]
 pub struct Run {
     /// The importer working directory
-    #[arg(long, env)]
+    #[arg(long, id = "working_dir", env = "IMPORTER_WORKING_DIR")]
     pub working_dir: Option<PathBuf>,
 
+    /// The max number of concurrent importer runs
+    #[arg(
+        long,
+        id = "concurrency",
+        env = "IMPORTER_CONCURRENCY",
+        default_value = "1"
+    )]
+    pub concurrency: usize,
+
     // flattened commands must go last
     //
     /// Database configuration
@@ -73,6 +82,7 @@ struct InitData {
     storage: DispatchBackend,
     tracing: Tracing,
     working_dir: Option<PathBuf>,
+    concurrency: usize,
 }
 
 impl Run {
@@ -126,6 +136,7 @@ impl InitData {
             tracing: run.infra.tracing,
             storage,
             working_dir: run.working_dir,
+            concurrency: run.concurrency,
         })
     }
 
@@ -138,8 +149,8 @@ impl InitData {
                 db,
                 storage,
                 self.working_dir,
-                // Running the importer, we don't need an analysis graph update
-                None,
+                None, // Running the importer, we don't need an analysis graph update
+                self.concurrency,
             )
             .await
         }