query to find out jobs running unusally long

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

@@ -103,6 +103,12 @@ public enum OssMetricsRegistry implements MetricsRegistry {
       MetricEmittingApps.METRICS_REPORTER,
       "num_total_scheduled_syncs_last_day",
       "number of total syncs runs in last day."),
+
+  NUM_UNUSUALLY_LONG_SYNCS(
+      MetricEmittingApps.METRICS_REPORTER,
+      "num_unusually_long_syncs",
+      "number of unusual long syncs compared to their historic performance."),
+
   OLDEST_PENDING_JOB_AGE_SECS(MetricEmittingApps.METRICS_REPORTER,
       "oldest_pending_job_age_secs",
       "oldest pending job in seconds"),

airbyte-metrics/reporter/src/main/java/io/airbyte/metrics/reporter/Emitter.java

@@ -113,6 +113,24 @@ public Duration getDuration() {
 
 }
 
+@Singleton
+final class NumUnusuallyLongSyncs extends Emitter {
+
+  NumUnusuallyLongSyncs(final MetricClient client, final MetricRepository db) {
+    super(client, () -> {
+      final var count = db.numberOfJobsRunningUnusuallyLong();
+      client.gauge(OssMetricsRegistry.NUM_UNUSUALLY_LONG_SYNCS, count);
+      return null;
+    });
+  }
+
+  @Override
+  public Duration getDuration() {
+    return Duration.ofHours(1);
+  }
+
+}
+
 @Singleton
 final class TotalScheduledSyncs extends Emitter {
 

airbyte-metrics/reporter/src/main/java/io/airbyte/metrics/reporter/MetricRepository.java

@@ -140,6 +140,30 @@ having count(*) < 1440 / cast(c.schedule::jsonb->'units' as integer)
         + ctx.fetchOne(queryForAbnormalSyncInMinutesInLastDay).get("cnt", long.class);
   }
 
+  long numberOfJobsRunningUnusuallyLong() {
+    // Definition of unusually long means runtime is more than 2x historic avg run time or 15
+    // minutes more than avg run time, whichever is greater.
+    final var query =
+        """
+        select
+        	scope as connection_id,
+        	avg(extract(epoch from age(updated_at, created_at))) as avg_run_sec,
+        	extract(epoch from age(max(updated_at), max(created_at))) as latest_run_sec
+        from
+        	jobs
+        where
+        	updated_at >= NOW() - interval '168 HOUR'
+        	and jobs.status = 'succeeded'
+        	and jobs.config_type = 'sync'
+        group by 1
+        having
+        	count(1) > 4
+        	and extract(epoch from age(max(updated_at), max(created_at))) > greatest(avg(extract(epoch from age(updated_at, created_at))) * 2, avg(extract(epoch from age(updated_at, created_at))) + 900)
+        """;
+    final var queryResults = ctx.fetch(query);
+    return queryResults.getValues("connection_id").size();
+  }
+
   Map<JobStatus, Double> overallJobRuntimeForTerminalJobsInLastHour() {
     final var query = """
                       SELECT status, extract(epoch from age(updated_at, created_at)) AS sec FROM jobs

airbyte-metrics/reporter/src/test/java/io/airbyte/metrics/reporter/MetricRepositoryTest.java

@@ -503,4 +503,77 @@ void shouldNotCountNormalJobsInAbnormalMetric() {
 
   }
 
+  @Nested
+  class UnusuallyLongJobs {
+
+    @Test
+    void shouldCountInJobsWithUnusuallyLongTime() throws SQLException {
+      final var connectionId = UUID.randomUUID();
+      final var syncConfigType = JobConfigType.sync;
+
+      // Latest job runs 12 hours while the previous 4 jobs runs 2 hours. Avg will be 4 hours.
+      // Thus latest job will be counted as an unusually long-running job.
+      ctx.insertInto(JOBS, JOBS.ID, JOBS.SCOPE, JOBS.STATUS, JOBS.CREATED_AT, JOBS.UPDATED_AT, JOBS.CONFIG_TYPE)
+          .values(100L, connectionId.toString(), JobStatus.succeeded, OffsetDateTime.now().minus(28, ChronoUnit.HOURS),
+              OffsetDateTime.now().minus(26, ChronoUnit.HOURS), syncConfigType)
+          .values(1L, connectionId.toString(), JobStatus.succeeded, OffsetDateTime.now().minus(20, ChronoUnit.HOURS),
+              OffsetDateTime.now().minus(18, ChronoUnit.HOURS), syncConfigType)
+          .values(2L, connectionId.toString(), JobStatus.succeeded, OffsetDateTime.now().minus(18, ChronoUnit.HOURS),
+              OffsetDateTime.now().minus(16, ChronoUnit.HOURS), syncConfigType)
+          .values(3L, connectionId.toString(), JobStatus.succeeded, OffsetDateTime.now().minus(16, ChronoUnit.HOURS),
+              OffsetDateTime.now().minus(14, ChronoUnit.HOURS), syncConfigType)
+          .values(4L, connectionId.toString(), JobStatus.succeeded, OffsetDateTime.now().minus(14, ChronoUnit.HOURS),
+              OffsetDateTime.now().minus(2, ChronoUnit.HOURS), syncConfigType)
+          .execute();
+
+      final var numOfJubsRunningUnusallyLong = db.numberOfJobsRunningUnusuallyLong();
+      assertEquals(1, numOfJubsRunningUnusallyLong);
+    }
+
+    @Test
+    void shouldNotCountInJobsWithinFifteenMinutes() throws SQLException {
+      final var connectionId = UUID.randomUUID();
+      final var syncConfigType = JobConfigType.sync;
+
+      // Latest job runs 12 minutes while the previous 2 jobs runs 3 minutes.
+      // Despite it's more than 2x than avg it's still within 15 minutes threshold, thus this shouldn't be
+      // counted in.
+      ctx.insertInto(JOBS, JOBS.ID, JOBS.SCOPE, JOBS.STATUS, JOBS.CREATED_AT, JOBS.UPDATED_AT, JOBS.CONFIG_TYPE)
+          .values(100L, connectionId.toString(), JobStatus.succeeded, OffsetDateTime.now().minus(28, ChronoUnit.MINUTES),
+              OffsetDateTime.now().minus(26, ChronoUnit.MINUTES), syncConfigType)
+          .values(1L, connectionId.toString(), JobStatus.succeeded, OffsetDateTime.now().minus(20, ChronoUnit.MINUTES),
+              OffsetDateTime.now().minus(18, ChronoUnit.MINUTES), syncConfigType)
+          .values(2L, connectionId.toString(), JobStatus.succeeded, OffsetDateTime.now().minus(18, ChronoUnit.MINUTES),
+              OffsetDateTime.now().minus(16, ChronoUnit.MINUTES), syncConfigType)
+          .values(3L, connectionId.toString(), JobStatus.succeeded, OffsetDateTime.now().minus(16, ChronoUnit.MINUTES),
+              OffsetDateTime.now().minus(14, ChronoUnit.MINUTES), syncConfigType)
+          .values(4L, connectionId.toString(), JobStatus.succeeded, OffsetDateTime.now().minus(14, ChronoUnit.MINUTES),
+              OffsetDateTime.now().minus(2, ChronoUnit.MINUTES), syncConfigType)
+          .execute();
+
+      final var numOfJubsRunningUnusallyLong = db.numberOfJobsRunningUnusuallyLong();
+      assertEquals(0, numOfJubsRunningUnusallyLong);
+    }
+
+    @Test
+    void shouldSkipInsufficientJobRuns() throws SQLException {
+      final var connectionId = UUID.randomUUID();
+      final var syncConfigType = JobConfigType.sync;
+
+      // Require at least 5 runs in last week to get meaningful average runtime.
+      ctx.insertInto(JOBS, JOBS.ID, JOBS.SCOPE, JOBS.STATUS, JOBS.CREATED_AT, JOBS.UPDATED_AT, JOBS.CONFIG_TYPE)
+          .values(100L, connectionId.toString(), JobStatus.succeeded, OffsetDateTime.now().minus(28, ChronoUnit.HOURS),
+              OffsetDateTime.now().minus(26, ChronoUnit.HOURS), syncConfigType)
+          .values(1L, connectionId.toString(), JobStatus.succeeded, OffsetDateTime.now().minus(20, ChronoUnit.HOURS),
+              OffsetDateTime.now().minus(18, ChronoUnit.HOURS), syncConfigType)
+          .values(2L, connectionId.toString(), JobStatus.succeeded, OffsetDateTime.now().minus(18, ChronoUnit.HOURS),
+              OffsetDateTime.now().minus(16, ChronoUnit.HOURS), syncConfigType)
+          .execute();
+
+      final var numOfJubsRunningUnusallyLong = db.numberOfJobsRunningUnusuallyLong();
+      assertEquals(0, numOfJubsRunningUnusallyLong);
+    }
+
+  }
+
 }