add attempt metrics (#21286)

airbyte-commons/src/main/java/io/airbyte/commons/util/MoreLists.java

@@ -4,6 +4,7 @@
 
 package io.airbyte.commons.util;
 
+import com.google.common.base.Preconditions;
 import java.util.List;
 import java.util.stream.Stream;
 
@@ -21,4 +22,13 @@ public static <T> List<T> concat(final List<T>... lists) {
     return Stream.of(lists).flatMap(List::stream).toList();
   }
 
+  public static <T> T getOrNull(final List<T> list, final int index) {
+    Preconditions.checkNotNull(list);
+    if (list.size() > index) {
+      return list.get(index);
+    } else {
+      return null;
+    }
+  }
+
 }

airbyte-commons/src/test/java/io/airbyte/commons/util/MoreListsTest.java

@@ -5,6 +5,8 @@
 package io.airbyte.commons.util;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 import java.util.List;
 import org.junit.jupiter.api.Test;
@@ -19,4 +21,13 @@ void testConcat() {
     assertEquals(expected, actual);
   }
 
+  @Test
+  void testGetOrNull() {
+    assertThrows(NullPointerException.class, () -> MoreLists.getOrNull(null, 0));
+    assertEquals(1, MoreLists.getOrNull(List.of(1, 2, 3), 0));
+    assertEquals(2, MoreLists.getOrNull(List.of(1, 2, 3), 1));
+    assertEquals(3, MoreLists.getOrNull(List.of(1, 2, 3), 2));
+    assertNull(MoreLists.getOrNull(List.of(1, 2, 3), 3));
+  }
+
 }

airbyte-metrics/metrics-lib/src/main/java/io/airbyte/metrics/lib/MetricTags.java

@@ -26,6 +26,13 @@ public class MetricTags {
   public static final String GEOGRAPHY = "geography";
   public static final String UNKNOWN = "unknown";
 
+  // the release stage of the highest release connector in the sync (GA > Beta > Alpha)
+  public static final String MAX_CONNECTOR_RELEASE_STATE = "max_connector_release_stage";
+  // the release stage of the lowest release stage connector in the sync (GA > Beta > Alpha)
+  public static final String MIN_CONNECTOR_RELEASE_STATE = "min_connector_release_stage";
+  public static final String ATTEMPT_OUTCOME = "attempt_outcome"; // succeeded|failed
+  public static final String ATTEMPT_NUMBER = "attempt_number"; // 0|1|2|3
+
   public static String getReleaseStage(final ReleaseStage stage) {
     return stage != null ? stage.getLiteral() : UNKNOWN;
   }

airbyte-metrics/metrics-lib/src/main/java/io/airbyte/metrics/lib/OssMetricsRegistry.java

@@ -5,6 +5,8 @@
 package io.airbyte.metrics.lib;
 
 import com.google.api.client.util.Preconditions;
+import java.util.Arrays;
+import java.util.List;
 
 /**
  * Enum source of truth of all Airbyte metrics. Each enum value represent a metric and is linked to
@@ -137,21 +139,48 @@ public enum OssMetricsRegistry implements MetricsRegistry {
       "number of records synced during replication"),
   RESET_REQUEST(MetricEmittingApps.WORKER,
       "reset_request",
-      "number of requested resets");
+      "number of requested resets"),
+
+  ATTEMPTS_CREATED(
+      MetricEmittingApps.WORKER,
+      "attempt_created",
+      "increments when a new attempt is created. one is emitted per attempt",
+      MetricTags.GEOGRAPHY,
+      MetricTags.ATTEMPT_NUMBER,
+      MetricTags.MIN_CONNECTOR_RELEASE_STATE,
+      MetricTags.MAX_CONNECTOR_RELEASE_STATE),
+  ATTEMPTS_COMPLETED(
+      MetricEmittingApps.WORKER,
+      "attempt_completed",
+      "increments when a new attempt is completed. one is emitted per attempt",
+      MetricTags.GEOGRAPHY,
+      MetricTags.ATTEMPT_NUMBER,
+      MetricTags.MIN_CONNECTOR_RELEASE_STATE,
+      MetricTags.MAX_CONNECTOR_RELEASE_STATE,
+      MetricTags.ATTEMPT_QUEUE,
+      MetricTags.ATTEMPT_OUTCOME,
+      MetricTags.FAILURE_ORIGIN, // only includes the first failure origin
+      MetricTags.FAILURE_TYPE); // only includes the first failure type
 
   private final MetricEmittingApp application;
   private final String metricName;
   private final String metricDescription;
 
+  // added this field to declare metric attributes, but we never read them.
+  @SuppressWarnings("FieldCanBeLocal")
+  private final List<String> metricTags;
+
   OssMetricsRegistry(final MetricEmittingApp application,
                      final String metricName,
-                     final String metricDescription) {
+                     final String metricDescription,
+                     final String... metricTags) {
     Preconditions.checkNotNull(metricDescription);
     Preconditions.checkNotNull(application);
 
     this.application = application;
     this.metricName = metricName;
     this.metricDescription = metricDescription;
+    this.metricTags = Arrays.asList(metricTags);
   }
 
   @Override

airbyte-persistence/job-persistence/src/main/java/io/airbyte/persistence/job/DefaultJobPersistence.java

@@ -157,6 +157,7 @@ private static String jobSelectAndJoin(final String jobsSubquery) {
         + "attempts.log_path AS log_path,\n"
         + "attempts.output AS attempt_output,\n"
         + "attempts.status AS attempt_status,\n"
+        + "attempts.processing_task_queue AS processing_task_queue,\n"
         + "attempts.failure_summary AS attempt_failure_summary,\n"
         + "attempts.created_at AS attempt_created_at,\n"
         + "attempts.updated_at AS attempt_updated_at,\n"
@@ -931,6 +932,7 @@ private static Attempt getAttemptFromRecord(final Record record) {
         Path.of(record.get("log_path", String.class)),
         record.get("attempt_output", String.class) == null ? null : Jsons.deserialize(record.get("attempt_output", String.class), JobOutput.class),
         Enums.toEnum(record.get("attempt_status", String.class), AttemptStatus.class).orElseThrow(),
+        record.get("processing_task_queue", String.class),
         record.get("attempt_failure_summary", String.class) == null ? null
             : Jsons.deserialize(record.get("attempt_failure_summary", String.class), AttemptFailureSummary.class),
         getEpoch(record, "attempt_created_at"),

airbyte-persistence/job-persistence/src/main/java/io/airbyte/persistence/job/models/Attempt.java

@@ -17,6 +17,7 @@ public class Attempt {
   private final long jobId;
   private final JobOutput output;
   private final AttemptStatus status;
+  private final String processingTaskQueue;
   private final AttemptFailureSummary failureSummary;
   private final Path logPath;
   private final long updatedAtInSecond;
@@ -28,6 +29,7 @@ public Attempt(final int attemptNumber,
                  final Path logPath,
                  final @Nullable JobOutput output,
                  final AttemptStatus status,
+                 final String processingTaskQueue,
                  final @Nullable AttemptFailureSummary failureSummary,
                  final long createdAtInSecond,
                  final long updatedAtInSecond,
@@ -36,6 +38,7 @@ public Attempt(final int attemptNumber,
     this.jobId = jobId;
     this.output = output;
     this.status = status;
+    this.processingTaskQueue = processingTaskQueue;
     this.failureSummary = failureSummary;
     this.logPath = logPath;
     this.updatedAtInSecond = updatedAtInSecond;
@@ -59,6 +62,10 @@ public AttemptStatus getStatus() {
     return status;
   }
 
+  public String getProcessingTaskQueue() {
+    return processingTaskQueue;
+  }
+
   public Optional<AttemptFailureSummary> getFailureSummary() {
     return Optional.ofNullable(failureSummary);
   }

airbyte-persistence/job-persistence/src/main/java/io/airbyte/persistence/job/models/Job.java

@@ -126,6 +126,12 @@ public Optional<Attempt> getLastAttemptWithOutput() {
         .findFirst();
   }
 
+  public Optional<Attempt> getLastAttempt() {
+    return getAttempts()
+        .stream()
+        .max(Comparator.comparing(Attempt::getCreatedAtInSecond));
+  }
+
   public boolean hasRunningAttempt() {
     return getAttempts().stream().anyMatch(a -> !Attempt.isAttemptInTerminalState(a));
   }

airbyte-persistence/job-persistence/src/test/java/io/airbyte/persistence/job/DefaultJobPersistenceTest.java

@@ -152,6 +152,7 @@ private static Attempt createAttempt(final int id, final long jobId, final Attem
         null,
         status,
         null,
+        null,
         NOW.getEpochSecond(),
         NOW.getEpochSecond(),
         NOW.getEpochSecond());
@@ -165,6 +166,7 @@ private static Attempt createUnfinishedAttempt(final int id, final long jobId, f
         null,
         status,
         null,
+        null,
         NOW.getEpochSecond(),
         NOW.getEpochSecond(),
         null);

airbyte-persistence/job-persistence/src/test/java/io/airbyte/persistence/job/models/AttemptTest.java

@@ -19,7 +19,7 @@ void testIsAttemptInTerminalState() {
   }
 
   private static Attempt attemptWithStatus(final AttemptStatus attemptStatus) {
-    return new Attempt(1, 1L, null, null, attemptStatus, null, 0L, 0L, null);
+    return new Attempt(1, 1L, null, null, attemptStatus, null, null, 0L, 0L, null);
   }
 
 }

airbyte-persistence/job-persistence/src/test/java/io/airbyte/persistence/job/models/JobTest.java

@@ -43,7 +43,7 @@ void testHasRunningAttempt() {
 
   private static Job jobWithAttemptWithStatus(final AttemptStatus... attemptStatuses) {
     final List<Attempt> attempts = IntStream.range(0, attemptStatuses.length)
-        .mapToObj(idx -> new Attempt(idx + 1, 1L, null, null, attemptStatuses[idx], null, idx, 0L, null))
+        .mapToObj(idx -> new Attempt(idx + 1, 1L, null, null, attemptStatuses[idx], null, null, idx, 0L, null))
         .collect(Collectors.toList());
     return new Job(1L, null, null, null, attempts, null, 0L, 0L, 0L);
   }
@@ -71,6 +71,13 @@ void testGetLastFailedAttempt() {
     assertEquals(2, job.getLastFailedAttempt().get().getAttemptNumber());
   }
 
+  @Test
+  void testGetLastAttempt() {
+    final Job job = jobWithAttemptWithStatus(AttemptStatus.FAILED, AttemptStatus.FAILED, AttemptStatus.SUCCEEDED);
+    assertTrue(job.getLastAttempt().isPresent());
+    assertEquals(3, job.getLastAttempt().get().getAttemptNumber());
+  }
+
   @Test
   void testValidateStatusTransitionFromPending() {
     final Job pendingJob = jobWithStatus(JobStatus.PENDING);

airbyte-persistence/job-persistence/src/test/java/io/airbyte/persistence/job/tracker/TrackingMetadataTest.java

@@ -61,7 +61,7 @@ void testgenerateJobAttemptMetadataWithNulls() {
     final StandardSyncSummary standardSyncSummary = new StandardSyncSummary().withTotalStats(syncStats);
     final StandardSyncOutput standardSyncOutput = new StandardSyncOutput().withStandardSyncSummary(standardSyncSummary);
     final JobOutput jobOutput = new JobOutput().withSync(standardSyncOutput);
-    final Attempt attempt = new Attempt(0, 10L, Path.of("test"), jobOutput, AttemptStatus.SUCCEEDED, null, 100L, 100L, 99L);
+    final Attempt attempt = new Attempt(0, 10L, Path.of("test"), jobOutput, AttemptStatus.SUCCEEDED, null, null, 100L, 100L, 99L);
     final Job job = mock(Job.class);
     when(job.getAttempts()).thenReturn(List.of(attempt));
 

airbyte-server/src/test/java/io/airbyte/server/handlers/JobHistoryHandlerTest.java

@@ -157,7 +157,7 @@ private static AttemptRead toAttemptRead(final Attempt a) {
   }
 
   private static Attempt createAttempt(final long jobId, final long timestamps, final AttemptStatus status) {
-    return new Attempt(ATTEMPT_NUMBER, jobId, LOG_PATH, null, status, null, timestamps, timestamps, timestamps);
+    return new Attempt(ATTEMPT_NUMBER, jobId, LOG_PATH, null, status, null, null, timestamps, timestamps, timestamps);
   }
 
   @BeforeEach
@@ -442,11 +442,11 @@ void testEnumConversion() {
   @DisplayName("Should return attempt normalization info for the job")
   void testGetAttemptNormalizationStatuses() throws IOException {
 
-    AttemptNormalizationStatus databaseReadResult = new AttemptNormalizationStatus(1, Optional.of(10L), /* hasNormalizationFailed= */ false);
+    final AttemptNormalizationStatus databaseReadResult = new AttemptNormalizationStatus(1, Optional.of(10L), /* hasNormalizationFailed= */ false);
 
     when(jobPersistence.getAttemptNormalizationStatusesForJob(JOB_ID)).thenReturn(List.of(databaseReadResult));
 
-    AttemptNormalizationStatusReadList expectedStatus = new AttemptNormalizationStatusReadList().attemptNormalizationStatuses(
+    final AttemptNormalizationStatusReadList expectedStatus = new AttemptNormalizationStatusReadList().attemptNormalizationStatuses(
         List.of(new AttemptNormalizationStatusRead().attemptNumber(1).hasRecordsCommitted(true).hasNormalizationFailed(false).recordsCommitted(10L)));
 
     assertEquals(expectedStatus, jobHistoryHandler.getAttemptNormalizationStatuses(new JobIdRequestBody().id(JOB_ID)));