diff --git a/gobblin-metrics-libs/gobblin-metrics-base/src/main/avro/GaaSJobObservabilityEvent.avsc b/gobblin-metrics-libs/gobblin-metrics-base/src/main/avro/GaaSJobObservabilityEvent.avsc index 53671f84170..165560f9be9 100644 --- a/gobblin-metrics-libs/gobblin-metrics-base/src/main/avro/GaaSJobObservabilityEvent.avsc +++ b/gobblin-metrics-libs/gobblin-metrics-base/src/main/avro/GaaSJobObservabilityEvent.avsc @@ -264,6 +264,18 @@ "type": ["null","string"], "doc": "Whether the dataset passed the overall data quality check", "default": null + }, + { + "name": "snapshotsCommitted", + "type": ["null","string"], + "doc": "Comma-separated list of the ids of the table snapshots committed for the dataset in this run (e.g. Iceberg snapshot-replication commits), null if unsupported/unreported by the writer", + "default": null + }, + { + "name": "partitionsCommitted", + "type": ["null","string"], + "doc": "Comma-separated list of the partitions committed for the dataset in this run, null if the writer does not commit partitions or does not report them", + "default": null } ] } diff --git a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/pegasus/org/apache/gobblin/service/JobStatistics.pdl b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/pegasus/org/apache/gobblin/service/JobStatistics.pdl index c0c072e3065..9644824387f 100644 --- a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/pegasus/org/apache/gobblin/service/JobStatistics.pdl +++ b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/pegasus/org/apache/gobblin/service/JobStatistics.pdl @@ -29,4 +29,35 @@ record JobStatistics { * estimate of time left until job completion */ estimatedSecondsToCompletion: long + + /** + * Number of bytes written/copied for the dataset in this job execution. + * Optional; populated by writers that report byte counts (e.g. DDM file/blob replication), -1 if unsupported. + */ + bytesWritten: optional long + + /** + * Number of records (or files, for file/blob copies) written for the dataset in this job execution. + * Optional; -1 if unsupported by the writer. + */ + recordsWritten: optional long + + /** + * Number of files committed for the dataset in this job execution. + * Optional; populated by file/blob writers (e.g. DDM Iceberg replication), absent otherwise. + */ + filesCommitted: optional long + + /** + * Comma-separated list of the ids of the table snapshots committed for the dataset in this job + * execution (e.g. Iceberg snapshot-replication commits). + * Optional; absent when the writer does not commit snapshots or does not report them. + */ + snapshotsCommitted: optional string + + /** + * Comma-separated list of the partitions committed for the dataset in this job execution. + * Optional; absent when the writer does not commit partitions or does not report them. + */ + partitionsCommitted: optional string } \ No newline at end of file diff --git a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowexecutions.snapshot.json b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowexecutions.snapshot.json index b2cdddc5e3a..ff29c500693 100644 --- a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowexecutions.snapshot.json +++ b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowexecutions.snapshot.json @@ -186,6 +186,31 @@ "name" : "estimatedSecondsToCompletion", "type" : "long", "doc" : "estimate of time left until job completion" + }, { + "name" : "bytesWritten", + "type" : "long", + "doc" : "Number of bytes written/copied for the dataset in this job execution.\nOptional; populated by writers that report byte counts (e.g. DDM file/blob replication), -1 if unsupported.", + "optional" : true + }, { + "name" : "recordsWritten", + "type" : "long", + "doc" : "Number of records (or files, for file/blob copies) written for the dataset in this job execution.\nOptional; -1 if unsupported by the writer.", + "optional" : true + }, { + "name" : "filesCommitted", + "type" : "long", + "doc" : "Number of files committed for the dataset in this job execution.\nOptional; populated by file/blob writers (e.g. DDM Iceberg replication), absent otherwise.", + "optional" : true + }, { + "name" : "snapshotsCommitted", + "type" : "string", + "doc" : "Comma-separated list of the ids of the table snapshots committed for the dataset in this job\nexecution (e.g. Iceberg snapshot-replication commits).\nOptional; absent when the writer does not commit snapshots or does not report them.", + "optional" : true + }, { + "name" : "partitionsCommitted", + "type" : "string", + "doc" : "Comma-separated list of the partitions committed for the dataset in this job execution.\nOptional; absent when the writer does not commit partitions or does not report them.", + "optional" : true } ] }, "doc" : "Statistics from the job execution. The values may be updated during the run of a job." diff --git a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowstatuses.snapshot.json b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowstatuses.snapshot.json index df4d02067a5..aba90d9dac1 100644 --- a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowstatuses.snapshot.json +++ b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowstatuses.snapshot.json @@ -188,6 +188,31 @@ "name" : "estimatedSecondsToCompletion", "type" : "long", "doc" : "estimate of time left until job completion" + }, { + "name" : "bytesWritten", + "type" : "long", + "doc" : "Number of bytes written/copied for the dataset in this job execution.\nOptional; populated by writers that report byte counts (e.g. DDM file/blob replication), -1 if unsupported.", + "optional" : true + }, { + "name" : "recordsWritten", + "type" : "long", + "doc" : "Number of records (or files, for file/blob copies) written for the dataset in this job execution.\nOptional; -1 if unsupported by the writer.", + "optional" : true + }, { + "name" : "filesCommitted", + "type" : "long", + "doc" : "Number of files committed for the dataset in this job execution.\nOptional; populated by file/blob writers (e.g. DDM Iceberg replication), absent otherwise.", + "optional" : true + }, { + "name" : "snapshotsCommitted", + "type" : "string", + "doc" : "Comma-separated list of the ids of the table snapshots committed for the dataset in this job\nexecution (e.g. Iceberg snapshot-replication commits).\nOptional; absent when the writer does not commit snapshots or does not report them.", + "optional" : true + }, { + "name" : "partitionsCommitted", + "type" : "string", + "doc" : "Comma-separated list of the partitions committed for the dataset in this job execution.\nOptional; absent when the writer does not commit partitions or does not report them.", + "optional" : true } ] }, "doc" : "Statistics from the job execution. The values may be updated during the run of a job." diff --git a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowExecutionResource.java b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowExecutionResource.java index 0bcdf569430..23dcbd8fd51 100644 --- a/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowExecutionResource.java +++ b/gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowExecutionResource.java @@ -184,17 +184,35 @@ public static FlowExecution convertFlowStatus(FlowStatus monitoringFlowStatus, Long timeLeft = estimateCopyTimeLeft(queriedJobStatus.getLastProgressEventTime(), queriedJobStatus.getStartTime(), queriedJobStatus.getProgressPercentage()); + JobStatistics jobStatistics = new JobStatistics() + .setExecutionStartTime(queriedJobStatus.getStartTime()) + .setExecutionEndTime(queriedJobStatus.getEndTime()) + .setProcessedCount(queriedJobStatus.getProcessedCount()) + .setJobProgress(queriedJobStatus.getProgressPercentage()) + .setEstimatedSecondsToCompletion(timeLeft); + // Optional per-dataset copy metrics (e.g. DDM file/blob replication); set only when reported. + if (queriedJobStatus.getBytesWritten() >= 0) { + jobStatistics.setBytesWritten(queriedJobStatus.getBytesWritten()); + } + if (queriedJobStatus.getRecordsWritten() >= 0) { + jobStatistics.setRecordsWritten(queriedJobStatus.getRecordsWritten()); + } + if (queriedJobStatus.getFilesCommitted() >= 0) { + jobStatistics.setFilesCommitted(queriedJobStatus.getFilesCommitted()); + } + if (queriedJobStatus.getSnapshotsCommitted() != null) { + jobStatistics.setSnapshotsCommitted(queriedJobStatus.getSnapshotsCommitted()); + } + if (queriedJobStatus.getPartitionsCommitted() != null) { + jobStatistics.setPartitionsCommitted(queriedJobStatus.getPartitionsCommitted()); + } + jobStatus.setFlowId(flowId) .setJobId(new JobId() .setJobName(queriedJobStatus.getJobName()) .setJobGroup(queriedJobStatus.getJobGroup())) .setJobTag(queriedJobStatus.getJobTag(), SetMode.IGNORE_NULL) - .setExecutionStatistics(new JobStatistics() - .setExecutionStartTime(queriedJobStatus.getStartTime()) - .setExecutionEndTime(queriedJobStatus.getEndTime()) - .setProcessedCount(queriedJobStatus.getProcessedCount()) - .setJobProgress(queriedJobStatus.getProgressPercentage()) - .setEstimatedSecondsToCompletion(timeLeft)) + .setExecutionStatistics(jobStatistics) .setExecutionStatus(ExecutionStatus.valueOf(queriedJobStatus.getEventName())) .setMessage(queriedJobStatus.getMessage()) .setJobState(new JobState() diff --git a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/DatasetTaskSummary.java b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/DatasetTaskSummary.java index f8c1d2a1e08..adc8a15159e 100644 --- a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/DatasetTaskSummary.java +++ b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/DatasetTaskSummary.java @@ -18,6 +18,7 @@ package org.apache.gobblin.runtime; import lombok.AccessLevel; +import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; import lombok.NonNull; @@ -34,6 +35,7 @@ @Data @Setter(AccessLevel.NONE) // NOTE: non-`final` members solely to enable deserialization @RequiredArgsConstructor +@AllArgsConstructor @NoArgsConstructor @ToString public class DatasetTaskSummary { @@ -42,11 +44,18 @@ public class DatasetTaskSummary { @NonNull private long bytesWritten; @NonNull private boolean successfullyCommitted; @NonNull private String dataQualityStatus; + // NOTE: intentionally NOT @NonNull so the 5-arg @RequiredArgsConstructor stays intact for native + // Gobblin call sites (AbstractJobLauncher). Usually populated via reflection during JSON + // deserialization of events that carry them (e.g. DDM Iceberg snapshot replication); the + // @AllArgsConstructor gives native producers a direct path. Both are comma-separated lists (a + // run can commit more than one snapshot/partition); null = unsupported/unreported. + private String snapshotsCommitted; + private String partitionsCommitted; /** * Convert a {@link DatasetTaskSummary} to a {@link DatasetMetric}. */ public static DatasetMetric toDatasetMetric(DatasetTaskSummary datasetTaskSummary) { - return new DatasetMetric(datasetTaskSummary.getDatasetUrn(), datasetTaskSummary.getBytesWritten(), datasetTaskSummary.getRecordsWritten(), datasetTaskSummary.isSuccessfullyCommitted(), datasetTaskSummary.getDataQualityStatus()); + return new DatasetMetric(datasetTaskSummary.getDatasetUrn(), datasetTaskSummary.getBytesWritten(), datasetTaskSummary.getRecordsWritten(), datasetTaskSummary.isSuccessfullyCommitted(), datasetTaskSummary.getDataQualityStatus(), datasetTaskSummary.getSnapshotsCommitted(), datasetTaskSummary.getPartitionsCommitted()); } } diff --git a/gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatus.java b/gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatus.java index 8dd91139914..a52cf0e874b 100644 --- a/gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatus.java +++ b/gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatus.java @@ -60,4 +60,12 @@ public class JobStatus { private final Supplier> issues; private final int progressPercentage; private final long lastProgressEventTime; + // Optional per-dataset copy metrics surfaced on the job-status REST API (JobStatistics). + // Populated for writers that report them (e.g. DDM file/blob replication); -1 = unset/unsupported + // for the counts. The snapshot ids and partitions are comma-separated lists; null = unreported. + @Builder.Default private final long bytesWritten = -1L; + @Builder.Default private final long recordsWritten = -1L; + @Builder.Default private final long filesCommitted = -1L; + @Builder.Default private final String snapshotsCommitted = null; + @Builder.Default private final String partitionsCommitted = null; } diff --git a/gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatusRetriever.java b/gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatusRetriever.java index fa56a577967..c82e806c641 100644 --- a/gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatusRetriever.java +++ b/gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatusRetriever.java @@ -60,6 +60,16 @@ public abstract class JobStatusRetriever implements LatestFlowExecutionIdTracker public static final String EVENT_NAME_FIELD = "eventName"; public static final String NA_KEY = "NA"; + // File/blob movement writers report per-dataset copy metrics on the JobSummary event via these + // flat metadata keys; the job-status monitor persists them into the job state, and they are + // surfaced here on the JobStatistics REST surface. -1 = absent/unsupported for the counts; + // the snapshot ids and partitions are comma-separated lists, null = absent/unreported. + public static final String BYTES_WRITTEN_FIELD = "bytesCopied"; + public static final String RECORDS_WRITTEN_FIELD = "rowsCopied"; + public static final String FILES_COMMITTED_FIELD = "filesCommitted"; + public static final String SNAPSHOTS_COMMITTED_FIELD = "snapshotsCommitted"; + public static final String PARTITIONS_COMMITTED_FIELD = "partitionsCommitted"; + @Getter protected final MetricContext metricContext; @@ -156,13 +166,20 @@ public static JobStatus.JobStatusBuilder createJobStatusBuilderFromState(State j boolean shouldRetry = Boolean.parseBoolean(jobState.getProp(TimingEvent.FlowEventConstants.SHOULD_RETRY_FIELD, "false")); int progressPercentage = jobState.getPropAsInt(TimingEvent.JOB_COMPLETION_PERCENTAGE, 0); long lastProgressEventTime = jobState.getPropAsLong(TimingEvent.JOB_LAST_PROGRESS_EVENT_TIME, 0); + long bytesWritten = jobState.getPropAsLong(BYTES_WRITTEN_FIELD, -1L); + long recordsWritten = jobState.getPropAsLong(RECORDS_WRITTEN_FIELD, -1L); + long filesCommitted = jobState.getPropAsLong(FILES_COMMITTED_FIELD, -1L); + String snapshotsCommitted = jobState.getProp(SNAPSHOTS_COMMITTED_FIELD, null); + String partitionsCommitted = jobState.getProp(PARTITIONS_COMMITTED_FIELD, null); return JobStatus.builder().flowName(flowName).flowGroup(flowGroup).flowExecutionId(flowExecutionId).jobName(jobName) .jobGroup(jobGroup).jobTag(jobTag).jobExecutionId(jobExecutionId).eventName(eventName).lowWatermark(lowWatermark) .highWatermark(highWatermark).orchestratedTime(orchestratedTime).startTime(startTime).endTime(endTime) .message(message).processedCount(processedCount).maxAttempts(maxAttempts).currentAttempts(currentAttempts) .currentGeneration(currentGeneration).shouldRetry(shouldRetry).progressPercentage(progressPercentage) - .lastProgressEventTime(lastProgressEventTime); + .lastProgressEventTime(lastProgressEventTime) + .bytesWritten(bytesWritten).recordsWritten(recordsWritten).filesCommitted(filesCommitted) + .snapshotsCommitted(snapshotsCommitted).partitionsCommitted(partitionsCommitted); } protected static final String getFlowGroup(State jobState) {