Improve efficiency of repository and integrity meta analysis

* Removes "preparation" logic of `IntegrityMetaComponent` records in `BomUploadProcessingTask`. Preparing records one-by-one is too resource-intensive, but doing it in batches has a too high potential for deadlocks since the table is written to by many threads in parallel.
* Refactors `RepositoryMetaResultProcessor` to consume records in batches. Since incoming records are keyed by PURL coordinates, we can safely perform batch operations in the database without the risk of running into deadlocks.

Signed-off-by: nscuro <[email protected]>

src/main/java/org/dependencytrack/common/ConfigKey.java

@@ -79,9 +79,6 @@ public enum ConfigKey implements Config.Key {
     WORKFLOW_RETENTION_DURATION("workflow.retention.duration", "P3D"),
     WORKFLOW_STEP_TIMEOUT_DURATION("workflow.step.timeout.duration", "PT1H"),
     TMP_DELAY_BOM_PROCESSED_NOTIFICATION("tmp.delay.bom.processed.notification", "false"),
-    INTEGRITY_META_INITIALIZER_LOCK_AT_MOST_FOR("integrityMetaInitializer.lockAtMostForInMillis", String.valueOf(Duration.ofMinutes(15).toMillis())),
-    INTEGRITY_META_INITIALIZER_LOCK_AT_LEAST_FOR("integrityMetaInitializer.lockAtLeastForInMillis", String.valueOf(Duration.ofMinutes(5).toMillis())),
-    INTEGRITY_INITIALIZER_ENABLED("integrity.initializer.enabled", "false"),
     INTEGRITY_CHECK_ENABLED("integrity.check.enabled", "false"),
     VULNERABILITY_POLICY_ANALYSIS_ENABLED("vulnerability.policy.analysis.enabled", false),
     VULNERABILITY_POLICY_BUNDLE_URL("vulnerability.policy.bundle.url", null),

src/main/java/org/dependencytrack/event/ComponentRepositoryMetaAnalysisEvent.java

@@ -19,19 +19,15 @@
 package org.dependencytrack.event;
 
 import alpine.event.framework.Event;
+import com.github.packageurl.PackageURL;
 import org.dependencytrack.model.Component;
-import org.dependencytrack.proto.repometaanalysis.v1.FetchMeta;
-
-import java.util.UUID;
 
 /**
  * Defines an {@link Event} triggered when requesting a component to be analyzed for meta information.
  *
- * @param purlCoordinates The package URL coordinates of the {@link Component} to analyze
- * @param internal        Whether the {@link Component} is internal
- * @param fetchMeta       Whether component hash data or component meta data needs to be fetched from external api
+ * @param purl     The {@link PackageURL} of the {@link Component} to analyze
+ * @param internal Whether the {@link Component} is internal
  */
-public record ComponentRepositoryMetaAnalysisEvent(UUID componentUuid, String purlCoordinates, Boolean internal,
-                                                   FetchMeta fetchMeta) implements Event {
+public record ComponentRepositoryMetaAnalysisEvent(PackageURL purl, Boolean internal) implements Event {
 
 }

src/main/java/org/dependencytrack/event/EventSubsystemInitializer.java

@@ -30,8 +30,6 @@
 import org.dependencytrack.tasks.EpssMirrorTask;
 import org.dependencytrack.tasks.FortifySscUploadTask;
 import org.dependencytrack.tasks.GitHubAdvisoryMirrorTask;
-import org.dependencytrack.tasks.IntegrityAnalysisTask;
-import org.dependencytrack.tasks.IntegrityMetaInitializerTask;
 import org.dependencytrack.tasks.InternalComponentIdentificationTask;
 import org.dependencytrack.tasks.KennaSecurityUploadTask;
 import org.dependencytrack.tasks.LdapSyncTaskWrapper;
@@ -42,12 +40,12 @@
 import org.dependencytrack.tasks.TaskScheduler;
 import org.dependencytrack.tasks.VexUploadProcessingTask;
 import org.dependencytrack.tasks.VulnerabilityAnalysisTask;
-import org.dependencytrack.tasks.vulnerabilitypolicy.VulnerabilityPolicyFetchTask;
 import org.dependencytrack.tasks.VulnerabilityScanCleanupTask;
 import org.dependencytrack.tasks.WorkflowStateCleanupTask;
 import org.dependencytrack.tasks.metrics.PortfolioMetricsUpdateTask;
 import org.dependencytrack.tasks.metrics.ProjectMetricsUpdateTask;
 import org.dependencytrack.tasks.metrics.VulnerabilityMetricsUpdateTask;
+import org.dependencytrack.tasks.vulnerabilitypolicy.VulnerabilityPolicyFetchTask;
 
 import jakarta.servlet.ServletContextEvent;
 import jakarta.servlet.ServletContextListener;
@@ -100,8 +98,6 @@ public void contextInitialized(final ServletContextEvent event) {
         EVENT_SERVICE.subscribe(ComponentPolicyEvaluationEvent.class, PolicyEvaluationTask.class);
         EVENT_SERVICE.subscribe(ProjectPolicyEvaluationEvent.class, PolicyEvaluationTask.class);
         EVENT_SERVICE.subscribe(WorkflowStateCleanupEvent.class, WorkflowStateCleanupTask.class);
-        EVENT_SERVICE.subscribe(IntegrityMetaInitializerEvent.class, IntegrityMetaInitializerTask.class);
-        EVENT_SERVICE.subscribe(IntegrityAnalysisEvent.class, IntegrityAnalysisTask.class);
 
         TaskScheduler.getInstance();
     }
@@ -135,8 +131,6 @@ public void contextDestroyed(final ServletContextEvent event) {
         EVENT_SERVICE.unsubscribe(EpssMirrorTask.class);
         EVENT_SERVICE.unsubscribe(PolicyEvaluationTask.class);
         EVENT_SERVICE.unsubscribe(WorkflowStateCleanupTask.class);
-        EVENT_SERVICE.unsubscribe(IntegrityMetaInitializerTask.class);
-        EVENT_SERVICE.unsubscribe(IntegrityAnalysisTask.class);
         EVENT_SERVICE.unsubscribe(VulnerabilityPolicyFetchTask.class);
         EVENT_SERVICE.shutdown(DRAIN_TIMEOUT_DURATION);
     }

src/main/java/org/dependencytrack/event/kafka/KafkaEventConverter.java

@@ -19,6 +19,7 @@
 package org.dependencytrack.event.kafka;
 
 import alpine.event.framework.Event;
+import com.github.packageurl.MalformedPackageURLException;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.Message;
 import org.dependencytrack.event.ComponentRepositoryMetaAnalysisEvent;
@@ -52,8 +53,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.UUID;
 
+import static com.github.packageurl.PackageURLBuilder.aPackageURL;
 import static org.apache.commons.lang3.ObjectUtils.requireNonEmpty;
 
 /**
@@ -130,21 +131,31 @@ static KafkaEvent<ScanKey, ScanCommand> convert(final ComponentVulnerabilityAnal
     }
 
     static KafkaEvent<String, AnalysisCommand> convert(final ComponentRepositoryMetaAnalysisEvent event) {
-        if (event == null || event.purlCoordinates() == null) {
+        if (event == null || event.purl() == null) {
             return null;
         }
+
+        final String key;
+        try {
+            key = aPackageURL()
+                    .withType(event.purl().getType())
+                    .withNamespace(event.purl().getNamespace())
+                    .withName(event.purl().getName())
+                    .build()
+                    .toString();
+        } catch (MalformedPackageURLException e) {
+            throw new IllegalStateException("Failed to build PURL coordinates without version", e);
+        }
 
         final var componentBuilder = org.dependencytrack.proto.repometaanalysis.v1.Component.newBuilder()
-                .setPurl(event.purlCoordinates());
+                .setPurl(event.purl().toString());
         Optional.ofNullable(event.internal()).ifPresent(componentBuilder::setInternal);
-        Optional.ofNullable(event.componentUuid()).map(UUID::toString).ifPresent(componentBuilder::setUuid);
 
         final var analysisCommand = AnalysisCommand.newBuilder()
                 .setComponent(componentBuilder)
-                .setFetchMeta(event.fetchMeta())
                 .build();
 
-        return new KafkaEvent<>(KafkaTopics.REPO_META_ANALYSIS_COMMAND, event.purlCoordinates(), analysisCommand, null);
+        return new KafkaEvent<>(KafkaTopics.REPO_META_ANALYSIS_COMMAND, key, analysisCommand, null);
     }
 
     static KafkaEvent<String, String> convert(final GitHubAdvisoryMirrorEvent ignored) {