Split workflow executions in half if request body size is too big

dockstore · Jan 19, 2024 · 311671f · 311671f
1 parent 38b05b4
commit 311671f
Show file tree

Hide file tree

Showing 2 changed files with 40 additions and 8 deletions.
diff --git a/metricsaggregator/pom.xml b/metricsaggregator/pom.xml
@@ -156,6 +156,10 @@
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-csv</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-api</artifactId>
@@ -292,6 +296,7 @@
                                 <usedDependency>org.javamoney.moneta:moneta-core</usedDependency>
                                 <usedDependency>ch.qos.logback:logback-classic</usedDependency>
                                 <usedDependency>ch.qos.logback:logback-core</usedDependency>
+                                <usedDependency>com.google.guava:guava</usedDependency>
                             </usedDependencies>
                         </configuration>
                     </execution>

diff --git a/...egator/src/main/java/io/dockstore/metricsaggregator/client/cli/TerraMetricsSubmitter.java b/...egator/src/main/java/io/dockstore/metricsaggregator/client/cli/TerraMetricsSubmitter.java
@@ -5,6 +5,7 @@
 import static io.dockstore.utils.ExceptionHandler.exceptionMessage;
 import static java.util.stream.Collectors.groupingBy;
 
+import com.google.common.collect.Lists;
 import io.dockstore.common.Partner;
 import io.dockstore.metricsaggregator.MetricsAggregatorConfig;
 import io.dockstore.metricsaggregator.client.cli.CommandLineArgs.SubmitTerraMetrics;
@@ -44,6 +45,7 @@
 import org.apache.commons.csv.CSVPrinter;
 import org.apache.commons.csv.CSVRecord;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.http.HttpStatus;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -167,21 +169,46 @@ private void submitWorkflowExecutions(String sourceUrl, List<CSVRecord> workflow
         }
 
         final SourceUrlTrsInfo sourceUrlTrsInfo = sourceUrlToSourceUrlTrsInfo.get(sourceUrl);
-        final List<RunExecution> workflowExecutionsToSubmit = workflowMetricRecords.stream()
+        List<RunExecution> workflowExecutionsToSubmit = workflowMetricRecords.stream()
                 .map(workflowExecution -> getTerraWorkflowExecutionFromCsvRecord(workflowExecution, sourceUrlTrsInfo.sourceUrl(), skippedExecutionsCsvPrinter))
                 .filter(Optional::isPresent)
                 .map(Optional::get)
                 .toList();
-        final ExecutionsRequestBody executionsRequestBody = new ExecutionsRequestBody().runExecutions(workflowExecutionsToSubmit);
+        String description = "Submitted using the metricsaggregator's submit-terra-metrics command";
+        if (StringUtils.isNotBlank(submitTerraMetricsCommand.getDescription())) {
+            description += ". " + submitTerraMetricsCommand.getDescription();
+        }
+
+        executionMetricsPost(workflowExecutionsToSubmit, sourceUrlTrsInfo, description, extendedGa4GhApi, workflowMetricRecords, skippedExecutionsCsvPrinter);
+    }
+
+    /**
+     * Submit Terra workflow executions to Dockstore.
+     * If the request fails with a 413 Request Entity Too Large, the function halves the number of workflow executions to submit then re-attempts submission
+     * until it's successful or a non-413 error occurs.
+     * @param workflowExecutionsToSubmit
+     * @param sourceUrlTrsInfo
+     * @param description
+     * @param extendedGa4GhApi
+     * @param workflowMetricRecords
+     * @param skippedExecutionsCsvPrinter
+     */
+    private void executionMetricsPost(List<RunExecution> workflowExecutionsToSubmit, SourceUrlTrsInfo sourceUrlTrsInfo, String description, ExtendedGa4GhApi extendedGa4GhApi, List<CSVRecord> workflowMetricRecords, CSVPrinter skippedExecutionsCsvPrinter) {
         try {
-            String description = "Submitted using the metricsaggregator's submit-terra-metrics command";
-            if (StringUtils.isNotBlank(submitTerraMetricsCommand.getDescription())) {
-                description += ". " + submitTerraMetricsCommand.getDescription();
-            }
-            extendedGa4GhApi.executionMetricsPost(executionsRequestBody, Partner.TERRA.toString(), sourceUrlTrsInfo.trsId(), sourceUrlTrsInfo.version(), description);
+            extendedGa4GhApi.executionMetricsPost(new ExecutionsRequestBody().runExecutions(workflowExecutionsToSubmit), Partner.TERRA.toString(), sourceUrlTrsInfo.trsId(),
+                    sourceUrlTrsInfo.version(), description);
             numberOfExecutionsSubmitted.addAndGet(workflowMetricRecords.size());
         } catch (ApiException e) {
-            logSkippedExecutions(sourceUrlTrsInfo.sourceUrl(), workflowMetricRecords, String.format("Could not submit execution metrics to Dockstore for workflow %s: %s", sourceUrlTrsInfo, e.getMessage()), skippedExecutionsCsvPrinter, false);
+            if (e.getCode() == HttpStatus.SC_REQUEST_TOO_LONG) {
+                List<List<RunExecution>> workflowExecutionsToSubmitPartitions = Lists.partition(workflowExecutionsToSubmit, 2);
+                for (List<RunExecution> partition: workflowExecutionsToSubmitPartitions) {
+                    executionMetricsPost(partition, sourceUrlTrsInfo, description, extendedGa4GhApi, workflowMetricRecords, skippedExecutionsCsvPrinter);
+                }
+            } else {
+                logSkippedExecutions(sourceUrlTrsInfo.sourceUrl(), workflowMetricRecords,
+                        String.format("Could not submit execution metrics to Dockstore for workflow %s: %s", sourceUrlTrsInfo,
+                                e.getMessage()), skippedExecutionsCsvPrinter, false);
+            }
         }
     }