Offload policy evaluation to internal task queue

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

@@ -0,0 +1,26 @@
+package org.dependencytrack.event;
+
+import alpine.event.framework.AbstractChainableEvent;
+import alpine.event.framework.Event;
+import org.dependencytrack.model.Component;
+
+import java.util.UUID;
+
+/**
+ * Defines an {@link Event} used to trigger policy evaluations for {@link Component}s.
+ *
+ * @since 5.0.0
+ */
+public class ComponentPolicyEvaluationEvent extends AbstractChainableEvent {
+
+    private final UUID uuid;
+
+    public ComponentPolicyEvaluationEvent(final UUID uuid) {
+        this.uuid = uuid;
+    }
+
+    public UUID getUuid() {
+        return uuid;
+    }
+
+}

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

@@ -37,6 +37,7 @@
 import org.dependencytrack.tasks.NewVulnerableDependencyAnalysisTask;
 import org.dependencytrack.tasks.NistMirrorTask;
 import org.dependencytrack.tasks.OsvDownloadTask;
+import org.dependencytrack.tasks.PolicyEvaluationTask;
 import org.dependencytrack.tasks.RepositoryMetaAnalyzerTask;
 import org.dependencytrack.tasks.TaskScheduler;
 import org.dependencytrack.tasks.VexUploadProcessingTask;
@@ -99,6 +100,8 @@ public void contextInitialized(final ServletContextEvent event) {
         EVENT_SERVICE.subscribe(VulnerabilityScanCleanupEvent.class, VulnerabilityScanCleanupTask.class);
         EVENT_SERVICE.subscribe(NistMirrorEvent.class, NistMirrorTask.class);
         EVENT_SERVICE.subscribe(EpssMirrorEvent.class, EpssMirrorTask.class);
+        EVENT_SERVICE.subscribe(ComponentPolicyEvaluationEvent.class, PolicyEvaluationTask.class);
+        EVENT_SERVICE.subscribe(ProjectPolicyEvaluationEvent.class, PolicyEvaluationTask.class);
 
         EVENT_SERVICE_ST.subscribe(IndexEvent.class, IndexTask.class);
 
@@ -134,6 +137,7 @@ public void contextDestroyed(final ServletContextEvent event) {
         EVENT_SERVICE.unsubscribe(VulnerabilityScanCleanupTask.class);
         EVENT_SERVICE.unsubscribe(NistMirrorTask.class);
         EVENT_SERVICE.unsubscribe(EpssMirrorTask.class);
+        EVENT_SERVICE.unsubscribe(PolicyEvaluationTask.class);
         EVENT_SERVICE.shutdown();
 
         EVENT_SERVICE_ST.unsubscribe(IndexTask.class);

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

@@ -0,0 +1,26 @@
+package org.dependencytrack.event;
+
+import alpine.event.framework.AbstractChainableEvent;
+import alpine.event.framework.Event;
+import org.dependencytrack.model.Project;
+
+import java.util.UUID;
+
+/**
+ * Defines an {@link Event} used to trigger policy evaluations for {@link Project}s.
+ *
+ * @since 5.0.0
+ */
+public class ProjectPolicyEvaluationEvent extends AbstractChainableEvent {
+
+    private final UUID uuid;
+
+    public ProjectPolicyEvaluationEvent(final UUID uuid) {
+        this.uuid = uuid;
+    }
+
+    public UUID getUuid() {
+        return uuid;
+    }
+
+}

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

@@ -1,5 +1,6 @@
 package org.dependencytrack.event.kafka;
 
+import alpine.event.framework.ChainableEvent;
 import alpine.event.framework.Event;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.StreamsBuilder;
@@ -11,13 +12,14 @@
 import org.apache.kafka.streams.kstream.Repartitioned;
 import org.datanucleus.PropertyNames;
 import org.dependencytrack.event.ComponentMetricsUpdateEvent;
+import org.dependencytrack.event.ComponentPolicyEvaluationEvent;
+import org.dependencytrack.event.ProjectPolicyEvaluationEvent;
 import org.dependencytrack.event.ProjectMetricsUpdateEvent;
 import org.dependencytrack.event.kafka.processor.MirrorVulnerabilityProcessor;
 import org.dependencytrack.event.kafka.processor.RepositoryMetaResultProcessor;
 import org.dependencytrack.event.kafka.processor.VulnerabilityScanResultProcessor;
 import org.dependencytrack.model.VulnerabilityScan;
 import org.dependencytrack.persistence.QueryManager;
-import org.dependencytrack.policy.PolicyEngine;
 import org.hyades.proto.vulnanalysis.v1.ScanKey;
 import org.hyades.proto.vulnanalysis.v1.ScanResult;
 
@@ -79,21 +81,23 @@ Topology createTopology() {
                         Named.as("filter_completed_vuln_scans"));
 
         completedVulnScanStream
-                .peek((scanToken, vulnScan) -> {
-                    final var policyEngine = new PolicyEngine();
-                    switch (vulnScan.getTargetType()) {
-                        case COMPONENT -> policyEngine.evaluate(vulnScan.getTargetIdentifier());
-                        case PROJECT -> policyEngine.evaluateProject(vulnScan.getTargetIdentifier());
-                    }
-                }, Named.as("execute_policy_evaluation"))
                 .foreach((scanToken, vulnScan) -> {
+                    final ChainableEvent policyEvaluationEvent = switch (vulnScan.getTargetType()) {
+                        case COMPONENT -> new ComponentPolicyEvaluationEvent(vulnScan.getTargetIdentifier());
+                        case PROJECT -> new ProjectPolicyEvaluationEvent(vulnScan.getTargetIdentifier());
+                    };
+                    policyEvaluationEvent.setChainIdentifier(UUID.fromString(vulnScan.getToken()));
+
+                    // Trigger a metrics update no matter if the policy evaluation succeeded or not.
                     final Event metricsUpdateEvent = switch (vulnScan.getTargetType()) {
                         case COMPONENT -> new ComponentMetricsUpdateEvent(vulnScan.getTargetIdentifier());
                         case PROJECT -> new ProjectMetricsUpdateEvent(vulnScan.getTargetIdentifier());
                     };
+                    policyEvaluationEvent.onFailure(metricsUpdateEvent);
+                    policyEvaluationEvent.onSuccess(metricsUpdateEvent);
 
-                    Event.dispatch(metricsUpdateEvent);
-                }, Named.as("trigger_metrics_update"));
+                    Event.dispatch(policyEvaluationEvent);
+                }, Named.as("trigger_policy_evaluation"));
 
         streamsBuilder
                 .stream(KafkaTopics.REPO_META_ANALYSIS_RESULT.name(),

src/main/java/org/dependencytrack/tasks/PolicyEvaluationTask.java

@@ -0,0 +1,38 @@
+package org.dependencytrack.tasks;
+
+import alpine.common.logging.Logger;
+import alpine.event.framework.Event;
+import alpine.event.framework.Subscriber;
+import org.dependencytrack.event.ComponentPolicyEvaluationEvent;
+import org.dependencytrack.event.ProjectPolicyEvaluationEvent;
+import org.dependencytrack.model.Component;
+import org.dependencytrack.model.Project;
+import org.dependencytrack.policy.PolicyEngine;
+
+/**
+ * A {@link Subscriber} task that executes policy evaluations for {@link Project}s or {@link Component}s.
+ *
+ * @since 5.0.0
+ */
+public class PolicyEvaluationTask implements Subscriber {
+
+    private static final Logger LOGGER = Logger.getLogger(PolicyEvaluationTask.class);
+
+    @Override
+    public void inform(final Event e) {
+        if (e instanceof final ProjectPolicyEvaluationEvent event) {
+            try {
+                new PolicyEngine().evaluateProject(event.getUuid());
+            } catch (Exception ex) {
+                LOGGER.error("An unexpected error occurred while evaluating policies for project " + event.getUuid(), ex);
+            }
+        } else if (e instanceof final ComponentPolicyEvaluationEvent event) {
+            try {
+                new PolicyEngine().evaluate(event.getUuid());
+            } catch (Exception ex) {
+                LOGGER.error("An unexpected error occurred while evaluating policies for component " + event.getUuid(), ex);
+            }
+        }
+    }
+
+}

src/test/java/org/dependencytrack/event/kafka/KafkaStreamsTopologyTest.java

@@ -8,6 +8,7 @@
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.dependencytrack.event.ProjectMetricsUpdateEvent;
+import org.dependencytrack.event.ProjectPolicyEvaluationEvent;
 import org.dependencytrack.event.kafka.serialization.KafkaProtobufSerializer;
 import org.dependencytrack.model.Policy;
 import org.dependencytrack.model.PolicyCondition;
@@ -16,6 +17,7 @@
 import org.dependencytrack.model.RepositoryType;
 import org.dependencytrack.model.VulnerabilityScan;
 import org.dependencytrack.model.VulnerabilityScan.TargetType;
+import org.dependencytrack.tasks.PolicyEvaluationTask;
 import org.hyades.proto.repometaanalysis.v1.AnalysisResult;
 import org.hyades.proto.vulnanalysis.v1.ScanKey;
 import org.hyades.proto.vulnanalysis.v1.ScanResult;
@@ -60,6 +62,7 @@ public void inform(final Event event) {
 
     @BeforeClass
     public static void setUpClass() {
+        EventService.getInstance().subscribe(ProjectPolicyEvaluationEvent.class, PolicyEvaluationTask.class);
         EventService.getInstance().subscribe(ProjectMetricsUpdateEvent.class, EventSubscriber.class);
     }
 
@@ -71,6 +74,7 @@ public void tearDown() {
 
     @AfterClass
     public static void tearDownClass() {
+        EventService.getInstance().unsubscribe(PolicyEvaluationTask.class);
         EventService.getInstance().unsubscribe(EventSubscriber.class);
     }