feat: Add individual task telemetry to dumper system

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/MetadataDumper.java

@@ -148,7 +148,8 @@ protected boolean run(@Nonnull Connector connector) throws Exception {
                 connectorArguments.getThreadPoolSize(),
                 state,
                 tasks,
-                connectorArguments)
+                connectorArguments,
+                telemetryProcessor)
             .run();
 
         requiredTaskSucceeded = checkRequiredTaskSuccess(summaryPrinter, state, outputFileLocation);

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/TasksRunner.java

@@ -25,6 +25,7 @@
 import com.google.edwmigration.dumper.application.dumper.io.OutputHandle;
 import com.google.edwmigration.dumper.application.dumper.io.OutputHandle.WriteMode;
 import com.google.edwmigration.dumper.application.dumper.io.OutputHandleFactory;
+import com.google.edwmigration.dumper.application.dumper.metrics.TaskRunMetrics;
 import com.google.edwmigration.dumper.application.dumper.task.Task;
 import com.google.edwmigration.dumper.application.dumper.task.TaskGroup;
 import com.google.edwmigration.dumper.application.dumper.task.TaskRunContext;
@@ -35,7 +36,9 @@
 import java.nio.charset.StandardCharsets;
 import java.sql.SQLException;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 import javax.annotation.CheckForNull;
@@ -56,17 +59,21 @@ public class TasksRunner implements TaskRunContextOps {
   private final TaskRunContext context;
   private final TaskSetState.Impl state;
   private final List<Task<?>> tasks;
+  private final TelemetryProcessor telemetryProcessor;
+  private final HashMap<String, String> MetricToErrorMap = new HashMap<>();
 
   public TasksRunner(
       OutputHandleFactory sinkFactory,
       Handle handle,
       int threadPoolSize,
       @Nonnull TaskSetState.Impl state,
       List<Task<?>> tasks,
-      ConnectorArguments arguments) {
+      ConnectorArguments arguments,
+      TelemetryProcessor telemetryProcessor) {
     context = createContext(sinkFactory, handle, threadPoolSize, arguments);
     this.state = state;
     this.tasks = tasks;
+    this.telemetryProcessor = telemetryProcessor;
     totalNumberOfTasks = countTasks(tasks);
     stopwatch = Stopwatch.createStarted();
     numberOfCompletedTasks = new AtomicInteger();
@@ -93,7 +100,13 @@ public <T> T runChildTask(@Nonnull Task<T> task) throws MetadataDumperUsageExcep
 
   public void run() throws MetadataDumperUsageException {
     for (Task<?> task : tasks) {
+      Instant taskStartTime = Instant.now();
+
       handleTask(task);
+
+      Instant taskEndTime = Instant.now();
+      TaskState finalState = getTaskState(task);
+      addTaskTelemetry(task.getName(), taskStartTime, taskEndTime, finalState);
     }
   }
 
@@ -168,6 +181,7 @@ private <T> T runTask(Task<T> task) throws MetadataDumperUsageException {
       else if (!task.handleException(e))
         logger.warn("Task failed: {}: {}", task, e.getMessage(), e);
       state.setTaskException(task, TaskState.FAILED, e);
+      MetricToErrorMap.put(task.getName(), e.getMessage());
       try {
         OutputHandle sink = context.newOutputFileHandle(task.getTargetPath() + ".exception.txt");
         sink.asCharSink(StandardCharsets.UTF_8, WriteMode.CREATE_TRUNCATE)
@@ -178,6 +192,7 @@ else if (!task.handleException(e))
                     String.valueOf(new DumperDiagnosticQuery(e).call())));
       } catch (Exception f) {
         logger.warn("Exception-recorder failed:  {}", f.getMessage(), f);
+        MetricToErrorMap.put(task.getName(), f.getMessage());
       }
     }
     return null;
@@ -188,4 +203,24 @@ private int countTasks(List<Task<?>> tasks) {
         .mapToInt(task -> task instanceof TaskGroup ? countTasks(((TaskGroup) task).getTasks()) : 1)
         .sum();
   }
+
+  private void addTaskTelemetry(
+      String taskName, Instant startTime, Instant endTime, TaskState state) {
+    if (telemetryProcessor != null) {
+      try {
+        TaskRunMetrics taskMetrics =
+            new TaskRunMetrics(
+                taskName,
+                state.name(),
+                startTime,
+                endTime,
+                MetricToErrorMap.getOrDefault(taskName, null));
+
+        // Add to the telemetry payload
+        telemetryProcessor.addTaskTelemetry(taskMetrics);
+      } catch (Exception e) {
+        logger.warn("Failed to add task telemetry for task: {}", taskName, e);
+      }
+    }
+  }
 }

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/TelemetryProcessor.java

@@ -50,6 +50,10 @@ public void addDumperRunMetricsToPayload(
         clientTelemetry, arguments, state, stopwatch, success);
   }
 
+  public void addTaskTelemetry(TaskRunMetrics taskMetrics) {
+    clientTelemetry.addToPayload(taskMetrics);
+  }
+
   public void processTelemetry(FileSystem fileSystem) {
     telemetryStrategy.writeTelemetry(fileSystem, clientTelemetry);
   }

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/metrics/DumperRunMetrics.java

@@ -27,6 +27,8 @@ public class DumperRunMetrics implements TelemetryPayload {
   @JsonProperty private String id;
 
   @JsonProperty private ZonedDateTime measureStartTime;
+
+  @JsonProperty private EventType eventType = EventType.DUMPER_RUN_METRICS;
 
   @JsonProperty private Long runDurationInMinutes;
 

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/metrics/EventType.java

@@ -17,5 +17,6 @@
 package com.google.edwmigration.dumper.application.dumper.metrics;
 
 public enum EventType {
-  DUMPER_RUN_METRICS
+  DUMPER_RUN_METRICS,
+  TASK_RUN_METRICS,
 }

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/metrics/TaskRunMetrics.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2022-2025 Google LLC
+ * Copyright 2013-2021 CompilerWorks
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.edwmigration.dumper.application.dumper.metrics;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.time.Instant;
+
+/**
+ * keep immutable. TaskRunner is multi-threaded, so we need to make it thread-safe.
+ *
+ * @author kakha
+ */
+public class TaskRunMetrics implements TelemetryPayload {
+
+  @JsonProperty private String name;
+
+  @JsonProperty private EventType eventType = EventType.TASK_RUN_METRICS;
+
+  @JsonProperty private String overallStatus;
+
+  @JsonProperty private Instant measureStartInstance;
+
+  @JsonProperty private Instant measureEndInstance;
+
+  @JsonProperty private String error;
+
+  public TaskRunMetrics() {
+    // Default constructor for Jackson deserialization
+  }
+
+  public TaskRunMetrics(
+      String name,
+      String overallStatus,
+      Instant measureStartInstance,
+      Instant measureEndInstance,
+      String error) {
+    this.name = name;
+    this.overallStatus = overallStatus;
+    this.measureStartInstance = measureStartInstance;
+    this.measureEndInstance = measureEndInstance;
+    this.error = error;
+  }
+
+  public EventType getEventType() {
+    return eventType;
+  }
+
+  public String getName() {
+    return name;
+  }
+
+  public String getOverallStatus() {
+    return overallStatus;
+  }
+
+  public Instant getMeasureStartInstance() {
+    return measureStartInstance;
+  }
+
+  public Instant getMeasureEndInstance() {
+    return measureEndInstance;
+  }
+
+  public String getError() {
+    return error;
+  }
+}