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]>

# Conflicts:
#	src/main/resources/migration/changelog-v5.6.0.xml

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

@@ -39,8 +39,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;
@@ -113,8 +111,6 @@ public void contextInitialized(final ServletContextEvent event) {
         EVENT_SERVICE.subscribe(EpssMirrorEvent.class, EpssMirrorTask.class);
         EVENT_SERVICE.subscribe(ComponentPolicyEvaluationEvent.class, PolicyEvaluationTask.class);
         EVENT_SERVICE.subscribe(ProjectPolicyEvaluationEvent.class, PolicyEvaluationTask.class);
-        EVENT_SERVICE.subscribe(IntegrityMetaInitializerEvent.class, IntegrityMetaInitializerTask.class);
-        EVENT_SERVICE.subscribe(IntegrityAnalysisEvent.class, IntegrityAnalysisTask.class);
 
         // Execute maintenance tasks on the single-threaded event service.
         // This way, they are not blocked by, and don't block, actual processing tasks on the main event service.
@@ -155,8 +151,6 @@ public void contextDestroyed(final ServletContextEvent event) {
         EVENT_SERVICE.unsubscribe(NistMirrorTask.class);
         EVENT_SERVICE.unsubscribe(EpssMirrorTask.class);
         EVENT_SERVICE.unsubscribe(PolicyEvaluationTask.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;
@@ -53,8 +54,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;
 
 /**
@@ -131,21 +132,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) {