Externalize BOM ingestion pipeline

pom.xml

@@ -517,6 +517,12 @@
             <version>${lib.json-unit.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>minio</artifactId>
+            <version>${lib.testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.testcontainers</groupId>
             <artifactId>kafka</artifactId>

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

@@ -38,6 +38,8 @@ public enum ConfigKey implements Config.Key {
 
     TASK_SCHEDULER_INITIAL_DELAY("task.scheduler.initial.delay", "180000"),
     TASK_SCHEDULER_POLLING_INTERVAL("task.scheduler.polling.interval", "60000"),
+    BOM_UPLOAD_STORAGE_COMPRESSION_LEVEL("bom.upload.storage.compression.level", "3"),
+    BOM_UPLOAD_STORAGE_RETENTION_DURATION("bom.upload.storage.retention.duration", "PT3H"),
     TMP_DELAY_BOM_PROCESSED_NOTIFICATION("tmp.delay.bom.processed.notification", "false"),
     INTEGRITY_INITIALIZER_ENABLED("integrity.initializer.enabled", "false"),
     INTEGRITY_CHECK_ENABLED("integrity.check.enabled", "false"),

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

@@ -151,6 +151,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

@@ -26,13 +26,13 @@
 import jakarta.servlet.ServletContextEvent;
 import jakarta.servlet.ServletContextListener;
 import org.dependencytrack.common.ConfigKey;
+import org.dependencytrack.event.maintenance.BomUploadStorageMaintenanceEvent;
 import org.dependencytrack.event.maintenance.ComponentMetadataMaintenanceEvent;
 import org.dependencytrack.event.maintenance.MetricsMaintenanceEvent;
 import org.dependencytrack.event.maintenance.TagMaintenanceEvent;
 import org.dependencytrack.event.maintenance.VulnerabilityDatabaseMaintenanceEvent;
 import org.dependencytrack.event.maintenance.VulnerabilityScanMaintenanceEvent;
 import org.dependencytrack.event.maintenance.WorkflowMaintenanceEvent;
-import org.dependencytrack.tasks.BomUploadProcessingTask;
 import org.dependencytrack.tasks.CallbackTask;
 import org.dependencytrack.tasks.CloneProjectTask;
 import org.dependencytrack.tasks.DefectDojoUploadTask;
@@ -51,6 +51,7 @@
 import org.dependencytrack.tasks.TaskScheduler;
 import org.dependencytrack.tasks.VexUploadProcessingTask;
 import org.dependencytrack.tasks.VulnerabilityAnalysisTask;
+import org.dependencytrack.tasks.maintenance.BomUploadStorageMaintenanceTask;
 import org.dependencytrack.tasks.maintenance.ComponentMetadataMaintenanceTask;
 import org.dependencytrack.tasks.maintenance.MetricsMaintenanceTask;
 import org.dependencytrack.tasks.maintenance.TagMaintenanceTask;
@@ -90,7 +91,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);
@@ -118,6 +118,7 @@ public void contextInitialized(final ServletContextEvent event) {
 
         // 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.
+        EVENT_SERVICE_ST.subscribe(BomUploadStorageMaintenanceEvent.class, BomUploadStorageMaintenanceTask.class);
         EVENT_SERVICE_ST.subscribe(ComponentMetadataMaintenanceEvent.class, ComponentMetadataMaintenanceTask.class);
         EVENT_SERVICE_ST.subscribe(MetricsMaintenanceEvent.class, MetricsMaintenanceTask.class);
         EVENT_SERVICE_ST.subscribe(TagMaintenanceEvent.class, TagMaintenanceTask.class);
@@ -136,7 +137,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);
@@ -160,6 +160,7 @@ public void contextDestroyed(final ServletContextEvent event) {
         EVENT_SERVICE.unsubscribe(VulnerabilityPolicyFetchTask.class);
         EVENT_SERVICE.shutdown(DRAIN_TIMEOUT_DURATION);
 
+        EVENT_SERVICE_ST.unsubscribe(BomUploadStorageMaintenanceTask.class);
         EVENT_SERVICE_ST.unsubscribe(ComponentMetadataMaintenanceTask.class);
         EVENT_SERVICE_ST.unsubscribe(MetricsMaintenanceTask.class);
         EVENT_SERVICE_ST.unsubscribe(TagMaintenanceTask.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.BomValidationFailedSubject;
@@ -68,6 +70,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);
@@ -105,6 +108,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);