Skip to content

feat: Add individual task telemetry to dumper system #949

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
kaxuna wants to merge 5 commits into from
Closed

feat: Add individual task telemetry to dumper system #949

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6e8bdac
feat: Add individual task telemetry to dumper system
kaxuna Aug 27, 2025
8316a20
added taskType
kaxuna Aug 27, 2025
5136f05
fix: Add trailing comma to EventType enum for better compatibility
kaxuna Aug 27, 2025
541f5cb
removed unneccesery test
kaxuna Aug 27, 2025
17b84ab
moved logic to telemetryStrategy impls
kaxuna Aug 28, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion ...r/app/src/main/java/com/google/edwmigration/dumper/application/dumper/MetadataDumper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
5 changes: 5 additions & 0 deletions ...rc/main/java/com/google/edwmigration/dumper/application/dumper/NoOpTelemetryStrategy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,4 +41,9 @@ public void processDumperRunMetrics(
public void writeTelemetry(FileSystem fileSystem, ClientTelemetry clientTelemetry) {
// Do nothing - this is the no-op strategy
}

@Override
public void addToPayload(ClientTelemetry clientTelemetry, TaskRunMetrics taskMetrics) {
// do nothing
}
}
39 changes: 38 additions & 1 deletion dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/TasksRunner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -35,8 +36,12 @@
import java.nio.charset.StandardCharsets;
import java.sql.SQLException;
import java.time.Duration;
import java.time.Instant;
import java.time.ZonedDateTime;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import javax.annotation.CheckForNull;
import javax.annotation.Nonnull;
Expand All @@ -56,17 +61,23 @@ 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) {
Preconditions.checkNotNull(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();
Expand All @@ -93,7 +104,15 @@ public <T> T runChildTask(@Nonnull Task<T> task) throws MetadataDumperUsageExcep

public void run() throws MetadataDumperUsageException {
for (Task<?> task : tasks) {
ZonedDateTime now = ZonedDateTime.now();
Stopwatch stopwatch = Stopwatch.createStarted();


handleTask(task);

Instant taskEndTime = Instant.now();
TaskState finalState = getTaskState(task);
addTaskTelemetry(task.getName(), stopwatch, now, finalState);
}
}

Expand Down Expand Up @@ -168,6 +187,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)
Expand All @@ -178,6 +198,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;
Expand All @@ -188,4 +209,20 @@ private int countTasks(List<Task<?>> tasks) {
.mapToInt(task -> task instanceof TaskGroup ? countTasks(((TaskGroup) task).getTasks()) : 1)
.sum();
}

private void addTaskTelemetry(
String taskName, Stopwatch stopwatch, ZonedDateTime startTime, TaskState state) {
Preconditions.checkNotNull(telemetryProcessor);

TaskRunMetrics taskMetrics =
new TaskRunMetrics(
taskName,
state.name(),
stopwatch.elapsed(TimeUnit.SECONDS),
startTime,
metricToErrorMap.getOrDefault(taskName, null));

// Add to the telemetry payload
telemetryProcessor.addToPayload(taskMetrics);
}
}
7 changes: 7 additions & 0 deletions ...p/src/main/java/com/google/edwmigration/dumper/application/dumper/TelemetryProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,13 @@ public void addDumperRunMetricsToPayload(
clientTelemetry, arguments, state, stopwatch, success);
}

/**
* addss evenets to the paylaod
*/
public void addToPayload(TaskRunMetrics taskMetrics) {
telemetryStrategy.addToPayload(clientTelemetry, taskMetrics);
}

public void processTelemetry(FileSystem fileSystem) {
telemetryStrategy.writeTelemetry(fileSystem, clientTelemetry);
}
Expand Down
2 changes: 2 additions & 0 deletions ...pp/src/main/java/com/google/edwmigration/dumper/application/dumper/TelemetryStrategy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,4 +50,6 @@ void processDumperRunMetrics(
* @param clientTelemetry the telemetry data to write
*/
void writeTelemetry(FileSystem fileSystem, ClientTelemetry clientTelemetry);

void addToPayload(ClientTelemetry clientTelemetry, TaskRunMetrics taskMetrics);
}
5 changes: 5 additions & 0 deletions ...c/main/java/com/google/edwmigration/dumper/application/dumper/WriteTelemetryStrategy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,4 +92,9 @@ public void writeTelemetry(FileSystem fileSystem, ClientTelemetry clientTelemetr
logger.warn("Failed to write telemetry", e);
}
}

@Override
public void addToPayload(ClientTelemetry clientTelemetry, TaskRunMetrics taskMetrics) {
clientTelemetry.addToPayload(taskMetrics);
}
}
2 changes: 2 additions & 0 deletions ...main/java/com/google/edwmigration/dumper/application/dumper/metrics/DumperRunMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@ public class DumperRunMetrics implements TelemetryPayload {

@JsonProperty private ZonedDateTime measureStartTime;

@JsonProperty private EventType eventType = EventType.DUMPER_RUN_METRICS;

@JsonProperty private Long runDurationInMinutes;

@JsonProperty private String overallStatus;
Expand Down
3 changes: 2 additions & 1 deletion ...pp/src/main/java/com/google/edwmigration/dumper/application/dumper/metrics/EventType.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,5 +17,6 @@
package com.google.edwmigration.dumper.application.dumper.metrics;

public enum EventType {
DUMPER_RUN_METRICS
DUMPER_RUN_METRICS,
TASK_RUN_METRICS,
}
82 changes: 82 additions & 0 deletions ...c/main/java/com/google/edwmigration/dumper/application/dumper/metrics/TaskRunMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
/*
* 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;
import java.time.ZonedDateTime;

/**
* 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;
Copy link
Member

@vladislav-sidorovich vladislav-sidorovich Aug 28, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I thinks we need both, taskName and taskClass. Another option is to redesign somehow our hierarchy of connectors and tasks.


@JsonProperty private EventType eventType = EventType.TASK_RUN_METRICS;
Copy link
Member

@vladislav-sidorovich vladislav-sidorovich Aug 28, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It should be on level above, before payload.
The logic on the service to parse event should be something like this:

EventType type = event.getType();
if (type is A) {
//payload is audi, do audi specific
} else if (type is B) {
//payload is bwm, do bwm specific
}


@JsonProperty private String overallStatus;

@JsonProperty private long runDurationInSeconds;

@JsonProperty private ZonedDateTime measureStartTime;

@JsonProperty private String error;

public TaskRunMetrics() {
// Default constructor for Jackson deserialization
}

public TaskRunMetrics(
String name,
String overallStatus,
long runDurationInSeconds,
ZonedDateTime measureStartTime,
String error) {
this.name = name;
this.overallStatus = overallStatus;
this.runDurationInSeconds = runDurationInSeconds;
this.measureStartTime = measureStartTime;
this.error = error;
}

public EventType getEventType() {
return eventType;
}

public String getName() {
return name;
}

public String getOverallStatus() {
return overallStatus;
}

public long getRunDurationInSeconds() {
return runDurationInSeconds;
}

public ZonedDateTime getMeasureStartTime() {
return measureStartTime;
}

public String getError() {
return error;
}
}
Loading