Add attempt status by release stage metrics. (#10659)

Add,

- attempt_created_by_release_stage
- attempt_failed_by_release_stage
- attempt_succeeded_by_release_stage

Author: davinchia

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

@@ -30,6 +30,18 @@
  */
 public enum MetricsRegistry {
 
+  ATTEMPT_CREATED_BY_RELEASE_STAGE(
+      MetricEmittingApps.WORKER,
+      "attempt_created_by_release_stage",
+      "increments when a new attempt is created. attempts are double counted as this is tagged by release stage."),
+  ATTEMPT_FAILED_BY_RELEASE_STAGE(
+      MetricEmittingApps.WORKER,
+      "attempt_failed_by_release_stage",
+      "increments when an attempt fails. attempts are double counted as this is tagged by release stage."),
+  ATTEMPT_SUCCEEDED_BY_RELEASE_STAGE(
+      MetricEmittingApps.WORKER,
+      "attempt_succeeded_by_release_stage",
+      "increments when an attempts succeeds. attempts are double counted as this is tagged by release stage."),
   JOB_CANCELLED_BY_RELEASE_STAGE(
       MetricEmittingApps.WORKER,
       "job_cancelled_by_release_stage",

airbyte-workers/src/main/java/io/airbyte/workers/temporal/scheduling/activities/JobCreationAndStatusUpdateActivityImpl.java

@@ -109,14 +109,15 @@ private void emitSrcIdDstIdToReleaseStagesMetric(final UUID srcId, final UUID ds
   @Override
   public AttemptCreationOutput createNewAttempt(final AttemptCreationInput input) throws RetryableException {
     try {
-      final Job createdJob = jobPersistence.getJob(input.getJobId());
+      final long jobId = input.getJobId();
+      final Job createdJob = jobPersistence.getJob(jobId);
 
       final WorkerRun workerRun = temporalWorkerRunFactory.create(createdJob);
       final Path logFilePath = workerRun.getJobRoot().resolve(LogClientSingleton.LOG_FILENAME);
-      final int persistedAttemptId = jobPersistence.createAttempt(input.getJobId(), logFilePath);
+      final int persistedAttemptId = jobPersistence.createAttempt(jobId, logFilePath);
+      emitJobIdToReleaseStagesMetric(MetricsRegistry.ATTEMPT_CREATED_BY_RELEASE_STAGE, jobId);
 
       LogClientSingleton.getInstance().setJobMdc(workerEnvironment, logConfigs, workerRun.getJobRoot());
-
       return new AttemptCreationOutput(persistedAttemptId);
     } catch (final IOException e) {
       throw new RetryableException(e);
@@ -136,6 +137,7 @@ public void jobSuccess(final JobSuccessInput input) {
         log.warn("The job {} doesn't have any output for the attempt {}", jobId, attemptId);
       }
       jobPersistence.succeedAttempt(jobId, attemptId);
+      emitJobIdToReleaseStagesMetric(MetricsRegistry.ATTEMPT_SUCCEEDED_BY_RELEASE_STAGE, jobId);
       final Job job = jobPersistence.getJob(jobId);
 
       jobNotifier.successJob(job);
@@ -164,14 +166,18 @@ public void jobFailure(final JobFailureInput input) {
   @Override
   public void attemptFailure(final AttemptFailureInput input) {
     try {
-      jobPersistence.failAttempt(input.getJobId(), input.getAttemptId());
-      jobPersistence.writeAttemptFailureSummary(input.getJobId(), input.getAttemptId(), input.getAttemptFailureSummary());
+      final int attemptId = input.getAttemptId();
+      final long jobId = input.getJobId();
+
+      jobPersistence.failAttempt(jobId, attemptId);
+      jobPersistence.writeAttemptFailureSummary(jobId, attemptId, input.getAttemptFailureSummary());
 
       if (input.getStandardSyncOutput() != null) {
         final JobOutput jobOutput = new JobOutput().withSync(input.getStandardSyncOutput());
-        jobPersistence.writeOutput(input.getJobId(), input.getAttemptId(), jobOutput);
+        jobPersistence.writeOutput(jobId, attemptId, jobOutput);
       }
 
+      emitJobIdToReleaseStagesMetric(MetricsRegistry.ATTEMPT_FAILED_BY_RELEASE_STAGE, jobId);
     } catch (final IOException e) {
       throw new RetryableException(e);
     }