Implement submitting in-progress pipelines data

src/main/java/org/datadog/jenkins/plugins/datadog/DatadogUtilities.java

@@ -99,7 +99,6 @@ public class DatadogUtilities {
     /**
      * @return - The descriptor for the Datadog plugin. In this case the global configuration.
      */
-    @Nullable
     public static DatadogGlobalConfiguration getDatadogGlobalDescriptor() {
         try {
             return ExtensionList.lookupSingleton(DatadogGlobalConfiguration.class);

src/main/java/org/datadog/jenkins/plugins/datadog/listeners/DatadogBuildListener.java

@@ -105,7 +105,11 @@ public void onInitialize(Run run) {
             BuildData buildData;
             try {
                 buildData = new BuildData(run, null);
-            } catch (IOException | InterruptedException e) {
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                DatadogUtilities.severe(logger, e, "Interrupted while trying to parse initialized build data");
+                return;
+            } catch (IOException e) {
                 DatadogUtilities.severe(logger, e, "Failed to parse initialized build data");
                 return;
             }
@@ -189,7 +193,11 @@ public void onStarted(Run run, TaskListener listener) {
             BuildData buildData;
             try {
                 buildData = new BuildData(run, listener);
-            } catch (IOException | InterruptedException e) {
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                DatadogUtilities.severe(logger, e, "Interrupted while trying to parse started build data");
+                return;
+            } catch (IOException e) {
                 DatadogUtilities.severe(logger, e, "Failed to parse started build data");
                 return;
             }
@@ -224,7 +232,18 @@ public void onStarted(Run run, TaskListener listener) {
             // Submit counter
             client.incrementCounter("jenkins.job.started", hostname, tags);
 
+            // APM Traces
+            if (DatadogUtilities.getDatadogGlobalDescriptor().getEnableCiVisibility()) {
+                TraceWriter traceWriter = TraceWriterFactory.getTraceWriter();
+                if (traceWriter != null) {
+                    traceWriter.submitBuild(buildData, run);
+                }
+            }
+
             logger.fine("End DatadogBuildListener#onStarted");
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            DatadogUtilities.severe(logger, e, "Interrupted while trying to process build start");
         } catch (Exception e) {
             DatadogUtilities.severe(logger, e, "Failed to process build start");
         }
@@ -372,7 +391,11 @@ public void onFinalized(Run run) {
             BuildData buildData;
             try {
                 buildData = new BuildData(run, null);
-            } catch (IOException | InterruptedException e) {
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                DatadogUtilities.severe(logger, e, "Interrupted while trying to parse finalized build data");
+                return;
+            } catch (IOException e) {
                 DatadogUtilities.severe(logger, e, "Failed to parse finalized build data");
                 return;
             }
@@ -413,7 +436,11 @@ public void onDeleted(Run run) {
             BuildData buildData;
             try {
                 buildData = new BuildData(run, null);
-            } catch (IOException | InterruptedException | NullPointerException e) {
+            } catch (InterruptedException e){
+                Thread.currentThread().interrupt();
+                DatadogUtilities.severe(logger, e, "Interrupted while trying to parse deleted build data");
+                return;
+            } catch (IOException | NullPointerException e) {
                 DatadogUtilities.severe(logger, e, "Failed to parse deleted build data");
                 return;
             }

src/main/java/org/datadog/jenkins/plugins/datadog/listeners/DatadogGraphListener.java

@@ -145,6 +145,9 @@ public void onNewHead(FlowNode flowNode) {
             metrics.gauge("jenkins.job.stage_duration", getTime(startNode, endNode), hostname, tags);
             metrics.gauge("jenkins.job.stage_pause_duration", pauseDuration, hostname, tags);
             client.incrementCounter("jenkins.job.stage_completed", hostname, tags);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            DatadogUtilities.severe(logger, e, "Interrupted while trying to submit the stage duration metric for " + getStageName(startNode));
         } catch (Exception e) {
             DatadogUtilities.severe(logger, e, "Unable to submit the stage duration metric for " + getStageName(startNode));
         }

src/main/java/org/datadog/jenkins/plugins/datadog/listeners/DatadogSCMListener.java

@@ -54,6 +54,8 @@ of this software and associated documentation files (the "Software"), to deal
 import org.datadog.jenkins.plugins.datadog.model.GitCommitAction;
 import org.datadog.jenkins.plugins.datadog.model.GitRepositoryAction;
 import org.datadog.jenkins.plugins.datadog.traces.GitInfoUtils;
+import org.datadog.jenkins.plugins.datadog.traces.write.TraceWriter;
+import org.datadog.jenkins.plugins.datadog.traces.write.TraceWriterFactory;
 import org.datadog.jenkins.plugins.datadog.util.git.GitUtils;
 import org.datadog.jenkins.plugins.datadog.util.git.RepositoryInfo;
 import org.eclipse.jgit.lib.PersonIdent;
@@ -102,6 +104,27 @@ public void onCheckout(Run<?, ?> build, SCM scm, FilePath workspace, TaskListene
                         + (scm != null ? scm.getType() : null));
             }
 
+            // Collect Build Data
+            BuildData buildData;
+            try {
+                buildData = new BuildData(build, listener);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                DatadogUtilities.severe(logger, e, "Interrupted while trying to parse checked out build data");
+                return;
+            } catch (IOException e) {
+                DatadogUtilities.severe(logger, e, "Failed to parse checked out build data");
+                return;
+            }
+
+            // We have Git info available now - submit a pipeline event so the backend could update its data
+            if (DatadogUtilities.getDatadogGlobalDescriptor().getEnableCiVisibility()) {
+                TraceWriter traceWriter = TraceWriterFactory.getTraceWriter();
+                if (traceWriter != null) {
+                    traceWriter.submitBuild(buildData, build);
+                }
+            }
+
             DatadogJobProperty prop = DatadogUtilities.getDatadogJobProperties(build);
             if (prop == null || !prop.isEmitSCMEvents()) {
                 return;
@@ -113,15 +136,6 @@ public void onCheckout(Run<?, ?> build, SCM scm, FilePath workspace, TaskListene
                 return;
             }
 
-            // Collect Build Data
-            BuildData buildData;
-            try {
-                buildData = new BuildData(build, listener);
-            } catch (IOException | InterruptedException e) {
-                DatadogUtilities.severe(logger, e, "Failed to parse checked out build data");
-                return;
-            }
-
             // Send event
             boolean shouldSendEvent = DatadogUtilities.shouldSendEvent(SCM_CHECKOUT_COMPLETED_EVENT_NAME);
             if (shouldSendEvent) {
@@ -135,6 +149,9 @@ public void onCheckout(Run<?, ?> build, SCM scm, FilePath workspace, TaskListene
             client.incrementCounter("jenkins.scm.checkout", hostname, tags);
 
             logger.fine("End DatadogSCMListener#onCheckout");
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            DatadogUtilities.severe(logger, e, "Interrupted while trying to process build checkout event");
         } catch (Exception e) {
             DatadogUtilities.severe(logger, e, "Failed to process build checkout event");
         }

src/main/java/org/datadog/jenkins/plugins/datadog/listeners/DatadogStepListener.java

@@ -25,12 +25,15 @@
 import org.datadog.jenkins.plugins.datadog.DatadogGlobalConfiguration;
 import org.datadog.jenkins.plugins.datadog.DatadogUtilities;
 import org.datadog.jenkins.plugins.datadog.audit.DatadogAudit;
+import org.datadog.jenkins.plugins.datadog.model.BuildData;
 import org.datadog.jenkins.plugins.datadog.model.GitCommitAction;
 import org.datadog.jenkins.plugins.datadog.model.GitRepositoryAction;
 import org.datadog.jenkins.plugins.datadog.model.PipelineNodeInfoAction;
 import org.datadog.jenkins.plugins.datadog.model.node.NodeInfoAction;
 import org.datadog.jenkins.plugins.datadog.traces.BuildSpanAction;
 import org.datadog.jenkins.plugins.datadog.traces.GitInfoUtils;
+import org.datadog.jenkins.plugins.datadog.traces.write.TraceWriter;
+import org.datadog.jenkins.plugins.datadog.traces.write.TraceWriterFactory;
 import org.datadog.jenkins.plugins.datadog.util.SuppressFBWarnings;
 import org.datadog.jenkins.plugins.datadog.util.git.GitUtils;
 import org.jenkinsci.plugins.workflow.cps.nodes.StepAtomNode;
@@ -330,6 +333,11 @@ private void findStartOfPipeline(final Run<?,?> run, final NodeInfoAction nodeIn
                 final FlowNode candidate = blockStartNodes.next();
                 if("Start of Pipeline".equals(candidate.getDisplayName())) {
                     run.addOrReplaceAction(new PipelineNodeInfoAction(nodeInfoAction.getNodeName() != null ? nodeInfoAction.getNodeName() : "master", nodeInfoAction.getNodeLabels(), nodeInfoAction.getNodeHostname(), nodeInfoAction.getNodeWorkspace()));
+
+                    if (DatadogUtilities.getDatadogGlobalDescriptor().getEnableCiVisibility()) {
+                        // we have node info available now - submit a pipeline event so the backend could update its data
+                        submitPipelineData(run);
+                    }
                 }
             }
         } finally {
@@ -338,6 +346,24 @@ private void findStartOfPipeline(final Run<?,?> run, final NodeInfoAction nodeIn
         }
     }
 
+    private static void submitPipelineData(Run<?, ?> run) {
+        TraceWriter traceWriter = TraceWriterFactory.getTraceWriter();
+        if (traceWriter == null) {
+            return;
+        }
+        try {
+            BuildData buildData = new BuildData(run, null);
+            traceWriter.submitBuild(buildData, run);
+
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            DatadogUtilities.severe(logger, e, "Interrupted while trying to submit pipeline data update");
+
+        } catch (Exception e) {
+            DatadogUtilities.severe(logger, e, "Failed to submit pipeline data update");
+        }
+    }
+
     private FlowNode findFirstAllocateNodeStart(FlowNode current) {
         long start = System.currentTimeMillis();
         try {

src/main/java/org/datadog/jenkins/plugins/datadog/logs/DatadogConsoleLogFilter.java

@@ -52,7 +52,7 @@ public DatadogConsoleLogFilter(Run<?, ?> run) {
         this.run = run;
     }
 
-    public OutputStream decorateLogger(Run build, OutputStream outputStream) throws IOException, InterruptedException {
+    public OutputStream decorateLogger(Run build, OutputStream outputStream) {
         try {
             if (!DatadogUtilities.getDatadogGlobalDescriptor().isCollectBuildLogs()) {
                 logger.fine("Log Collection disabled");
@@ -68,6 +68,9 @@ public OutputStream decorateLogger(Run build, OutputStream outputStream) throws
             } else {
                 return outputStream;
             }
+        } catch (InterruptedException e){
+            Thread.currentThread().interrupt();
+            DatadogUtilities.severe(logger, e, "Interrupted while trying to wrap logger, logs will not be collected");
         } catch (Exception e){
             DatadogUtilities.severe(logger, e, "Failed to wrap logger, logs will not be collected");
         }

src/main/java/org/datadog/jenkins/plugins/datadog/logs/DatadogTaskListenerDecorator.java

@@ -47,6 +47,9 @@ public class DatadogTaskListenerDecorator extends TaskListenerDecorator {
     public DatadogTaskListenerDecorator(WorkflowRun run) {
         try {
             this.buildData = new BuildData(run, null);
+        } catch (InterruptedException e){
+            Thread.currentThread().interrupt();
+            DatadogUtilities.severe(LOGGER, e, null);
         } catch (Exception e) {
             DatadogUtilities.severe(LOGGER, e, null);
         }

src/main/java/org/datadog/jenkins/plugins/datadog/model/BuildData.java

@@ -74,6 +74,7 @@ of this software and associated documentation files (the "Software"), to deal
 import org.datadog.jenkins.plugins.datadog.traces.BuildConfigurationParser;
 import org.datadog.jenkins.plugins.datadog.traces.BuildSpanAction;
 import org.datadog.jenkins.plugins.datadog.traces.BuildSpanManager;
+import org.datadog.jenkins.plugins.datadog.traces.CITags;
 import org.datadog.jenkins.plugins.datadog.traces.message.TraceSpan;
 import org.datadog.jenkins.plugins.datadog.util.TagsUtil;
 import org.datadog.jenkins.plugins.datadog.util.git.GitUtils;
@@ -137,6 +138,17 @@ public class BuildData implements Serializable {
     private Long millisInQueue;
     private Long propagatedMillisInQueue;
 
+    /**
+     * Monotonically increasing "version" of the build data.
+     * As the pipeline progresses, it can be reported to the backend more than once:
+     * <ul>
+     * <li>when it starts executing</li>
+     * <li>when git info or node info become available</li>
+     * <li>when it finishes.</li>
+     * </ul>
+     * The backend needs version to determine the relative order of these multiple events.
+     */
+    private Integer version;
     private String traceId;
     private String spanId;
 
@@ -149,11 +161,13 @@ public BuildData(Run<?, ?> run, @Nullable TaskListener listener) throws IOExcept
         this.tags = DatadogUtilities.getBuildTags(run, envVars);
 
         this.buildUrl = envVars.get("BUILD_URL");
-        if (buildUrl == null) {
-            BuildSpanAction buildSpanAction = run.getAction(BuildSpanAction.class);
-            if (buildSpanAction != null) {
+
+        BuildSpanAction buildSpanAction = run.getAction(BuildSpanAction.class);
+        if (buildSpanAction != null) {
+            if (buildUrl == null) {
                 buildUrl = buildSpanAction.getBuildUrl();
             }
+            version = buildSpanAction.getAndIncrementVersion();
         }
 
         // Populate instance using environment variables.
@@ -163,18 +177,6 @@ public BuildData(Run<?, ?> run, @Nullable TaskListener listener) throws IOExcept
         // Set all Git commit related variables.
         populateGitVariables(run);
 
-        // Populate instance using run instance
-        // Set StartTime, EndTime and Duration
-        this.startTime = run.getStartTimeInMillis();
-        long durationInMs = run.getDuration();
-        if (durationInMs == 0 && run.getStartTimeInMillis() != 0) {
-            durationInMs = System.currentTimeMillis() - run.getStartTimeInMillis();
-        }
-        this.duration = durationInMs;
-        if (durationInMs != 0 && run.getStartTimeInMillis() != 0) {
-            this.endTime = run.getStartTimeInMillis() + durationInMs;
-        }
-
         // Set Jenkins Url
         this.jenkinsUrl = DatadogUtilities.getJenkinsUrl();
         // Set UserId
@@ -189,11 +191,26 @@ public BuildData(Run<?, ?> run, @Nullable TaskListener listener) throws IOExcept
         if (runResult != null) {
             this.result = runResult.toString();
             this.isCompleted = runResult.completeBuild;
+        } else if (run.isBuilding() && !run.hasntStartedYet()) {
+            // #isBuilding() includes queued runs, so we check #hasntStartedYet() as well
+            this.result = CITags.STATUS_RUNNING;
+            this.isCompleted = false;
         } else {
             this.result = null;
             this.isCompleted = false;
         }
 
+        // Set StartTime, EndTime and Duration
+        this.startTime = run.getStartTimeInMillis();
+        long durationInMs = run.getDuration();
+        if (durationInMs == 0 && startTime != 0) {
+            durationInMs = System.currentTimeMillis() - startTime;
+        }
+        this.duration = durationInMs;
+        if (duration != 0 && startTime != 0 && isCompleted) {
+            this.endTime = startTime + duration;
+        }
+
         // Set Build Number
         this.buildNumber = String.valueOf(run.getNumber());
 
@@ -647,6 +664,10 @@ public Long getPropagatedMillisInQueue(Long value) {
         return defaultIfNull(propagatedMillisInQueue, value);
     }
 
+    public Integer getVersion() {
+        return version;
+    }
+
     public String getBuildTag(String value) {
         return defaultIfNull(buildTag, value);
     }