Externalize BOM ingestion pipeline

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

src/main/java/org/dependencytrack/dev/DevServicesInitializer.java

@@ -150,6 +150,7 @@ public void contextInitialized(final ServletContextEvent event) {
         }
 
         final var topicsToCreate = new ArrayList<>(List.of(
+                new NewTopic(KafkaTopics.EVENT_BOM_UPLOADED.name(), 1, (short) 1),
                 new NewTopic(KafkaTopics.NEW_EPSS.name(), 1, (short) 1).configs(Map.of(CLEANUP_POLICY_CONFIG, CLEANUP_POLICY_COMPACT)),
                 new NewTopic(KafkaTopics.NEW_VULNERABILITY.name(), 1, (short) 1).configs(Map.of(CLEANUP_POLICY_CONFIG, CLEANUP_POLICY_COMPACT)),
                 new NewTopic(KafkaTopics.NOTIFICATION_ANALYZER.name(), 1, (short) 1),

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

@@ -21,8 +21,6 @@
 import alpine.event.framework.AbstractChainableEvent;
 import org.dependencytrack.model.Project;
 
-import java.io.File;
-
 /**
  * Defines an event triggered when a bill-of-material (bom) document is submitted.
  *
@@ -32,18 +30,13 @@
 public class BomUploadEvent extends AbstractChainableEvent {
 
     private final Project project;
-    private final File file;
 
-    public BomUploadEvent(final Project project, final File file) {
+    public BomUploadEvent(final Project project) {
         this.project = project;
-        this.file = file;
     }
 
     public Project getProject() {
         return project;
     }
 
-    public File getFile() {
-        return file;
-    }
 }

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

@@ -23,7 +23,6 @@
 import alpine.event.LdapSyncEvent;
 import alpine.event.framework.EventService;
 import org.dependencytrack.common.ConfigKey;
-import org.dependencytrack.tasks.BomUploadProcessingTask;
 import org.dependencytrack.tasks.CallbackTask;
 import org.dependencytrack.tasks.CloneProjectTask;
 import org.dependencytrack.tasks.DefectDojoUploadTask;
@@ -42,12 +41,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 javax.servlet.ServletContextEvent;
 import javax.servlet.ServletContextListener;
@@ -75,7 +74,6 @@ public class EventSubsystemInitializer implements ServletContextListener {
     public void contextInitialized(final ServletContextEvent event) {
         LOGGER.info("Initializing asynchronous event subsystem");
 
-        EVENT_SERVICE.subscribe(BomUploadEvent.class, BomUploadProcessingTask.class);
         EVENT_SERVICE.subscribe(VexUploadEvent.class, VexUploadProcessingTask.class);
         EVENT_SERVICE.subscribe(LdapSyncEvent.class, LdapSyncTaskWrapper.class);
         EVENT_SERVICE.subscribe(GitHubAdvisoryMirrorEvent.class, GitHubAdvisoryMirrorTask.class);
@@ -114,7 +112,6 @@ public void contextDestroyed(final ServletContextEvent event) {
         LOGGER.info("Shutting down asynchronous event subsystem");
         TaskScheduler.getInstance().shutdown();
 
-        EVENT_SERVICE.unsubscribe(BomUploadProcessingTask.class);
         EVENT_SERVICE.unsubscribe(VexUploadProcessingTask.class);
         EVENT_SERVICE.unsubscribe(LdapSyncTaskWrapper.class);
         EVENT_SERVICE.unsubscribe(GitHubAdvisoryMirrorTask.class);

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

@@ -21,6 +21,7 @@
 import alpine.event.framework.Event;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.Message;
+import org.dependencytrack.event.BomUploadEvent;
 import org.dependencytrack.event.ComponentRepositoryMetaAnalysisEvent;
 import org.dependencytrack.event.ComponentVulnerabilityAnalysisEvent;
 import org.dependencytrack.event.EpssMirrorEvent;
@@ -30,6 +31,7 @@
 import org.dependencytrack.event.kafka.KafkaTopics.Topic;
 import org.dependencytrack.model.Vulnerability;
 import org.dependencytrack.parser.dependencytrack.NotificationModelConverter;
+import org.dependencytrack.proto.event.v1alpha1.BomUploadedEvent;
 import org.dependencytrack.proto.notification.v1.BomConsumedOrProcessedSubject;
 import org.dependencytrack.proto.notification.v1.BomProcessingFailedSubject;
 import org.dependencytrack.proto.notification.v1.NewVulnerabilitySubject;
@@ -67,6 +69,7 @@ private KafkaEventConverter() {
 
     static KafkaEvent<?, ?> convert(final Event event) {
         return switch (event) {
+            case BomUploadEvent e -> convert(e);
             case ComponentRepositoryMetaAnalysisEvent e -> convert(e);
             case ComponentVulnerabilityAnalysisEvent e -> convert(e);
             case GitHubAdvisoryMirrorEvent e -> convert(e);
@@ -104,6 +107,22 @@ private KafkaEventConverter() {
         return kafkaEvents;
     }
 
+    static KafkaEvent<UUID, BomUploadedEvent> convert(final BomUploadEvent event) {
+        final BomUploadedEvent.Project.Builder projectBuilder = BomUploadedEvent.Project.newBuilder()
+                .setUuid(event.getProject().getUuid().toString())
+                .setName(event.getProject().getName());
+        Optional.ofNullable(event.getProject().getVersion()).ifPresent(projectBuilder::setVersion);
+
+        return new KafkaEvent<>(
+                KafkaTopics.EVENT_BOM_UPLOADED,
+                event.getProject().getUuid(),
+                BomUploadedEvent.newBuilder()
+                        .setToken(event.getChainIdentifier().toString())
+                        .setProject(projectBuilder)
+                        .build()
+        );
+    }
+
     static KafkaEvent<ScanKey, ScanCommand> convert(final ComponentVulnerabilityAnalysisEvent event) {
         final var componentBuilder = Component.newBuilder()
                 .setUuid(event.uuid().toString());

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

@@ -24,6 +24,7 @@
 import org.cyclonedx.proto.v1_6.Bom;
 import org.dependencytrack.common.ConfigKey;
 import org.dependencytrack.event.kafka.serialization.KafkaProtobufSerde;
+import org.dependencytrack.proto.event.v1alpha1.BomUploadedEvent;
 import org.dependencytrack.proto.mirror.v1.EpssItem;
 import org.dependencytrack.proto.notification.v1.Notification;
 import org.dependencytrack.proto.repometaanalysis.v1.AnalysisCommand;
@@ -32,8 +33,11 @@
 import org.dependencytrack.proto.vulnanalysis.v1.ScanKey;
 import org.dependencytrack.proto.vulnanalysis.v1.ScanResult;
 
+import java.util.UUID;
+
 public final class KafkaTopics {
 
+    public static final Topic<UUID, BomUploadedEvent> EVENT_BOM_UPLOADED;
     public static final Topic<String, Notification> NOTIFICATION_ANALYZER;
     public static final Topic<String, Notification> NOTIFICATION_BOM;
     public static final Topic<String, Notification> NOTIFICATION_CONFIGURATION;
@@ -61,6 +65,7 @@ public final class KafkaTopics {
     private static final Serde<Notification> NOTIFICATION_SERDE = new KafkaProtobufSerde<>(Notification.parser());
 
     static {
+        EVENT_BOM_UPLOADED = new Topic<>("dtrack.event.bom-uploaded", Serdes.UUID(), new KafkaProtobufSerde<>(BomUploadedEvent.parser()));
         NOTIFICATION_ANALYZER = new Topic<>("dtrack.notification.analyzer", Serdes.String(), NOTIFICATION_SERDE);
         NOTIFICATION_BOM = new Topic<>("dtrack.notification.bom", Serdes.String(), NOTIFICATION_SERDE);
         NOTIFICATION_CONFIGURATION = new Topic<>("dtrack.notification.configuration", Serdes.String(), NOTIFICATION_SERDE);