feat: split tpa requests (#429)

Signed-off-by: Ruben Romero Montes <[email protected]>

Author: ruromero

src/main/java/com/redhat/exhort/integration/Constants.java

@@ -58,6 +58,7 @@ private Constants() {}
   public static final MediaType MULTIPART_MIXED_TYPE = new MediaType("multipart", "mixed");
   public static final String MULTIPART_MIXED = MULTIPART_MIXED_TYPE.toString();
   public static final String SPDX_MEDIATYPE_JSON = "application/vnd.spdx+json";
+  public static final String CYCLONEDX_MEDIATYPE_JSON = "application/vnd.cyclonedx+json";
 
   public static final String SNYK_PROVIDER = "snyk";
   public static final String OSS_INDEX_PROVIDER = "oss-index";

src/main/java/com/redhat/exhort/integration/backend/sbom/SbomParserFactory.java

@@ -18,8 +18,6 @@
 
 package com.redhat.exhort.integration.backend.sbom;
 
-import org.cyclonedx.CycloneDxMediaType;
-
 import com.redhat.exhort.integration.Constants;
 import com.redhat.exhort.integration.backend.sbom.cyclonedx.CycloneDxParser;
 import com.redhat.exhort.integration.backend.sbom.spdx.SpdxParser;
@@ -31,7 +29,7 @@ public class SbomParserFactory {
 
   public static final SbomParser newInstance(String mediaType) {
     return switch (mediaType) {
-      case CycloneDxMediaType.APPLICATION_CYCLONEDX_JSON -> new CycloneDxParser();
+      case Constants.CYCLONEDX_MEDIATYPE_JSON -> new CycloneDxParser();
       case Constants.SPDX_MEDIATYPE_JSON -> new SpdxParser();
       default -> throw new ClientErrorException(
           "Unsupported Content-Type header: " + mediaType, Response.Status.UNSUPPORTED_MEDIA_TYPE);

src/main/java/com/redhat/exhort/integration/providers/ossindex/OssIndexIntegration.java

@@ -29,7 +29,6 @@
 import com.redhat.exhort.integration.Constants;
 import com.redhat.exhort.integration.providers.VulnerabilityProvider;
 import com.redhat.exhort.model.DependencyTree;
-import com.redhat.exhort.monitoring.MonitoringProcessor;
 
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
@@ -47,8 +46,6 @@ public class OssIndexIntegration extends EndpointRouteBuilder {
 
   @Inject OssIndexResponseHandler responseHandler;
 
-  @Inject MonitoringProcessor monitoringProcessor;
-
   @Override
   public void configure() {
 

src/main/java/com/redhat/exhort/integration/providers/ossindex/OssIndexRequestBuilder.java

@@ -40,7 +40,7 @@ public class OssIndexRequestBuilder {
 
   private static final int BULK_SIZE = 128;
 
-  private ObjectMapper mapper = ObjectMapperProducer.newInstance();
+  private final ObjectMapper mapper = ObjectMapperProducer.newInstance();
 
   public List<List<PackageRef>> split(DependencyTree tree) {
     List<List<PackageRef>> bulks = new ArrayList<>();

src/main/java/com/redhat/exhort/integration/providers/ossindex/OssIndexResponseHandler.java

@@ -56,6 +56,7 @@ public class OssIndexResponseHandler extends ProviderResponseHandler {
 
   @Inject ObjectMapper mapper;
 
+  @Override
   public ProviderResponse responseToIssues(
       @Body byte[] response,
       @ExchangeProperty(Constants.PROVIDER_PRIVATE_DATA_PROPERTY) String privateProviders,

src/main/java/com/redhat/exhort/integration/providers/tpa/TpaIntegration.java

@@ -22,6 +22,7 @@
 
 import org.apache.camel.Exchange;
 import org.apache.camel.Message;
+import org.apache.camel.builder.AggregationStrategies;
 import org.apache.camel.builder.endpoint.EndpointRouteBuilder;
 import org.eclipse.microprofile.config.inject.ConfigProperty;
 
@@ -65,26 +66,27 @@ public void configure() throws Exception {
         .transform().method(responseHandler, "buildReport")
       .endChoice()
       .otherwise()
-        .to(direct("tpaRequest"))
-        .transform().method(responseHandler, "buildReport")
-      .endChoice();
-
-    from(direct("tpaRequest"))
-      .routeId("tpaRequest")
-      .circuitBreaker()
-        .faultToleranceConfiguration()
-          .timeoutEnabled(true)
-          .timeoutDuration(timeout)
-        .end()
-        .transform(method(requestBuilder, "buildRequest"))
-        .process(this::processRequest)
-        .process(requestBuilder::addAuthentication)
-        .to(http("{{api.tpa.host}}"))
-        .transform().method(responseHandler, "responseToIssues")
-      .endCircuitBreaker()
-      .onFallback()
-        .process(responseHandler::processResponseError)
-      .end();
+        .to(direct("tpaSplitRequest"))
+        .transform().method(responseHandler, "buildReport");
+
+    from(direct("tpaSplitRequest"))
+      .routeId("tpaSplitRequest")
+      .transform(method(TpaRequestBuilder.class, "split"))
+      .split(body(), AggregationStrategies.beanAllowNull(responseHandler, "aggregateSplit"))
+        .parallelProcessing()
+          .transform().method(requestBuilder, "buildRequest")
+          .process(this::processRequest)
+          .process(requestBuilder::addAuthentication)
+        .circuitBreaker()
+          .faultToleranceConfiguration()
+            .timeoutEnabled(true)
+            .timeoutDuration(timeout)
+          .end()
+          .to(http("{{api.tpa.host}}"))
+          .transform(method(responseHandler, "responseToIssues"))
+        .onFallback()
+          .process(responseHandler::processResponseError);
+  
 
     from(direct("tpaHealthCheck"))
       .routeId("tpaHealthCheck")

src/main/java/com/redhat/exhort/integration/providers/tpa/TpaRequestBuilder.java

@@ -18,6 +18,8 @@
 
 package com.redhat.exhort.integration.providers.tpa;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Optional;
 
 import org.apache.camel.Exchange;
@@ -37,19 +39,37 @@
 @RegisterForReflection
 public class TpaRequestBuilder {
 
+  private static final int BULK_SIZE = 128;
+
   @ConfigProperty(name = "api.tpa.token")
   Optional<String> defaultToken;
 
-  private ObjectMapper mapper = ObjectMapperProducer.newInstance();
+  private final ObjectMapper mapper = ObjectMapperProducer.newInstance();
 
-  public String buildRequest(DependencyTree tree) throws JsonProcessingException {
+  public String buildRequest(List<String> refs) throws JsonProcessingException {
     var request = mapper.createObjectNode();
     var purls = mapper.createArrayNode();
-    tree.getAll().forEach(dep -> purls.add(dep.ref()));
+    refs.forEach(dep -> purls.add(dep));
     request.set("purls", purls);
     return mapper.writeValueAsString(request);
   }
 
+  public List<List<String>> split(DependencyTree tree) {
+    List<List<String>> bulks = new ArrayList<>();
+    List<String> bulk = new ArrayList<>();
+    for (var pkg : tree.getAll()) {
+      if (bulk.size() == BULK_SIZE) {
+        bulks.add(bulk);
+        bulk = new ArrayList<>();
+      }
+      bulk.add(pkg.ref());
+    }
+    if (!bulk.isEmpty()) {
+      bulks.add(bulk);
+    }
+    return bulks;
+  }
+
   public void addAuthentication(Exchange exchange) {
     var message = exchange.getMessage();
     var userToken = message.getHeader(Constants.TPA_TOKEN_HEADER, String.class);

src/main/java/com/redhat/exhort/integration/trustedcontent/TcResponseAggregation.java

@@ -18,6 +18,7 @@
 
 package com.redhat.exhort.integration.trustedcontent;
 
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
@@ -27,6 +28,7 @@
 import org.apache.camel.ExchangeProperty;
 
 import com.redhat.exhort.api.PackageRef;
+import com.redhat.exhort.api.v4.ProviderStatus;
 import com.redhat.exhort.integration.Constants;
 import com.redhat.exhort.integration.cache.CacheService;
 import com.redhat.exhort.model.trustedcontent.IndexedRecommendation;
@@ -37,6 +39,7 @@
 
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
+import jakarta.ws.rs.core.Response.Status;
 
 @Singleton
 @RegisterForReflection
@@ -57,6 +60,16 @@ public TrustedContentResponse aggregateCachedResponse(
       Exchange exchange)
       throws ExecutionException {
     var externalResponse = exchange.getIn().getBody(TrustedContentResponse.class);
+    if (externalResponse == null) {
+      externalResponse =
+          new TrustedContentResponse(
+              Collections.emptyMap(),
+              new ProviderStatus()
+                  .name(Constants.TRUSTED_CONTENT_PROVIDER)
+                  .code(Status.OK.getStatusCode())
+                  .message(Status.OK.getReasonPhrase())
+                  .ok(Boolean.TRUE));
+    }
     cacheService.cacheRecommendations(externalResponse, cached.miss());
     Map<PackageRef, IndexedRecommendation> recommendations =
         new HashMap<>(externalResponse.recommendations());

src/main/java/com/redhat/exhort/integration/trustedcontent/TcResponseHandler.java

@@ -62,11 +62,22 @@ public class TcResponseHandler extends ProviderResponseHandler {
 
   @Inject UBIRecommendation ubiRecommendation;
 
-  public TrustedContentResponse parseResponse(
+  public TrustedContentResponse mergeSplitRecommendations(
+      TrustedContentResponse oldResponse, TrustedContentResponse newResponse) {
+    if (oldResponse == null) {
+      return newResponse;
+    }
+    if (!oldResponse.status().getOk()) {
+      return oldResponse;
+    }
+    oldResponse.recommendations().putAll(newResponse.recommendations());
+    return oldResponse;
+  }
+
+  public TrustedContentResponse processRecommendations(
       @Body byte[] tcResponse, @ExchangeProperty(Constants.SBOM_ID_PROPERTY) String sbomId)
       throws IOException {
     var recommendations = mapper.readValue(tcResponse, Recommendations.class);
-
     var mergedRecommendations = mergeRecommendations(recommendations);
     mergedRecommendations.putAll(getUBIRecommendation(sbomId));
 
@@ -82,6 +93,9 @@ public TrustedContentResponse parseResponse(
   private Map<PackageRef, IndexedRecommendation> mergeRecommendations(
       Recommendations recommendations) {
     Map<PackageRef, IndexedRecommendation> result = new HashMap<>();
+    if (recommendations == null) {
+      return result;
+    }
     recommendations.getMatchings().entrySet().stream()
         .forEach(
             e -> result.put(new PackageRef(e.getKey()), aggregateRecommendations(e.getValue())));

src/main/java/com/redhat/exhort/integration/trustedcontent/TrustedContentIntegration.java

@@ -19,6 +19,7 @@
 package com.redhat.exhort.integration.trustedcontent;
 
 import org.apache.camel.Exchange;
+import org.apache.camel.builder.AggregationStrategies;
 import org.apache.camel.builder.endpoint.EndpointRouteBuilder;
 import org.eclipse.microprofile.config.inject.ConfigProperty;
 
@@ -55,19 +56,20 @@ public void configure() {
 
     from(direct("getRemoteTrustedContent"))
       .routeId("getRemoteTrustedContent")
-      .circuitBreaker()
-        .faultToleranceConfiguration()
-          .timeoutEnabled(true)
-          .timeoutDuration(timeout)
-        .end()
-        .transform().method(requestBuilder, "buildRequest")
-        .process(this::handleHeaders)
-        .to(vertxHttp("{{api.trustedcontent.host}}"))
-        .transform(method(TcResponseHandler.class, "parseResponse"))
-      .endCircuitBreaker()
-      .onFallback()
-        .process(responseHandler::processResponseError);
-
+      .transform(method(requestBuilder, "split"))
+      .split(body(), AggregationStrategies.bean(TcResponseHandler.class, "mergeSplitRecommendations"))
+        .parallelProcessing()
+          .transform().method(requestBuilder, "buildRequest")
+          .process(this::handleHeaders)
+        .circuitBreaker()
+          .faultToleranceConfiguration()
+            .timeoutEnabled(true)
+            .timeoutDuration(timeout)
+          .end()
+          .to(vertxHttp("{{api.trustedcontent.host}}"))
+          .transform(method(TcResponseHandler.class, "processRecommendations"))  
+        .onFallback()
+          .process(responseHandler::processResponseError);
     // fmt:on
   }