Add async observability to QuartzJob

service/build.gradle

@@ -113,6 +113,7 @@ dependencies {
 
     annotationProcessor 'org.springframework.boot:spring-boot-configuration-processor'
     runtimeOnly 'org.webjars.npm:swagger-ui-dist:5.9.0'
+    testImplementation 'io.micrometer:micrometer-observation-test'
     testImplementation 'org.springframework.boot:spring-boot-starter-test'
     testImplementation 'org.junit.jupiter:junit-jupiter'
     testImplementation project(':client')

service/src/main/java/org/databiosphere/workspacedataservice/dataimport/pfb/PfbQuartzJob.java

@@ -7,6 +7,7 @@
 import static org.databiosphere.workspacedataservice.shared.model.Schedulable.ARG_URL;
 
 import bio.terra.pfb.PfbReader;
+import io.micrometer.observation.ObservationRegistry;
 import java.net.URL;
 import java.util.Objects;
 import java.util.Optional;
@@ -58,7 +59,9 @@ public PfbQuartzJob(
       RestClientRetry restClientRetry,
       BatchWriteService batchWriteService,
       ActivityLogger activityLogger,
+      ObservationRegistry observationRegistry,
       @Value("${twds.instance.workspace-id}") UUID workspaceId) {
+    super(observationRegistry);
     this.jobDao = jobDao;
     this.wsmDao = wsmDao;
     this.restClientRetry = restClientRetry;

service/src/main/java/org/databiosphere/workspacedataservice/dataimport/tdr/TdrManifestQuartzJob.java

@@ -10,6 +10,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Multimap;
+import io.micrometer.observation.ObservationRegistry;
 import java.io.File;
 import java.io.IOException;
 import java.net.MalformedURLException;
@@ -69,7 +70,9 @@ public TdrManifestQuartzJob(
       BatchWriteService batchWriteService,
       ActivityLogger activityLogger,
       @Value("${twds.instance.workspace-id}") UUID workspaceId,
-      ObjectMapper mapper) {
+      ObjectMapper mapper,
+      ObservationRegistry observationRegistry) {
+    super(observationRegistry);
     this.jobDao = jobDao;
     this.wsmDao = wsmDao;
     this.restClientRetry = restClientRetry;

service/src/main/java/org/databiosphere/workspacedataservice/jobexec/QuartzJob.java

@@ -3,6 +3,8 @@
 import static org.databiosphere.workspacedataservice.sam.BearerTokenFilter.ATTRIBUTE_NAME_TOKEN;
 import static org.databiosphere.workspacedataservice.shared.model.Schedulable.ARG_TOKEN;
 
+import io.micrometer.observation.Observation;
+import io.micrometer.observation.ObservationRegistry;
 import java.net.URL;
 import java.util.UUID;
 import org.databiosphere.workspacedataservice.dao.JobDao;
@@ -29,16 +31,29 @@
 // note this implements Quartz's `Job`, not WDS's own `Job`
 public abstract class QuartzJob implements Job {
 
+  private final ObservationRegistry observationRegistry;
+
+  protected QuartzJob(ObservationRegistry observationRegistry) {
+    this.observationRegistry = observationRegistry;
+  }
+
   /** implementing classes are expected to be beans that inject a JobDao */
   protected abstract JobDao getJobDao();
 
   @Override
   public void execute(JobExecutionContext context) throws org.quartz.JobExecutionException {
     // retrieve jobId
     UUID jobId = UUID.fromString(context.getJobDetail().getKey().getName());
+
+    Observation observation =
+        Observation.start("wds.job.execute", observationRegistry)
+            .contextualName("job-execution")
+            .lowCardinalityKeyValue("jobType", getClass().getSimpleName())
+            .highCardinalityKeyValue("jobId", jobId.toString());
     try {
       // mark this job as running
       getJobDao().running(jobId);
+      observation.event(Observation.Event.of("job.running"));
       // look for an auth token in the Quartz JobDataMap
       String authToken = getJobDataString(context.getMergedJobDataMap(), ARG_TOKEN);
       // and stash the auth token into job context
@@ -50,11 +65,15 @@ public void execute(JobExecutionContext context) throws org.quartz.JobExecutionE
       executeInternal(jobId, context);
       // if we reached here, mark this job as successful
       getJobDao().succeeded(jobId);
+      observation.lowCardinalityKeyValue("outcome", "success");
     } catch (Exception e) {
       // on any otherwise-unhandled exception, mark the job as failed
       getJobDao().fail(jobId, e);
+      observation.error(e);
+      observation.lowCardinalityKeyValue("outcome", "failure");
     } finally {
       JobContextHolder.destroy();
+      observation.stop();
     }
   }
 

service/src/main/java/org/databiosphere/workspacedataservice/metrics/ObservationConfig.java

@@ -0,0 +1,25 @@
+package org.databiosphere.workspacedataservice.metrics;
+
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.observation.DefaultMeterObservationHandler;
+import io.micrometer.observation.ObservationRegistry;
+import org.springframework.boot.actuate.autoconfigure.observation.ObservationRegistryCustomizer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class ObservationConfig {
+  private final MeterRegistry meterRegistry;
+
+  public ObservationConfig(MeterRegistry meterRegistry) {
+    this.meterRegistry = meterRegistry;
+  }
+
+  @Bean
+  ObservationRegistryCustomizer<ObservationRegistry> useAutowiredMeterRegistry() {
+    return observationRegistry ->
+        observationRegistry
+            .observationConfig()
+            .observationHandler(new DefaultMeterObservationHandler(meterRegistry));
+  }
+}

service/src/main/java/org/databiosphere/workspacedataservice/service/RecordService.java

@@ -1,9 +1,5 @@
 package org.databiosphere.workspacedataservice.service;
 
-import static org.databiosphere.workspacedataservice.metrics.MetricsDefinitions.COUNTER_COL_CHANGE;
-import static org.databiosphere.workspacedataservice.metrics.MetricsDefinitions.TAG_ATTRIBUTE_NAME;
-import static org.databiosphere.workspacedataservice.metrics.MetricsDefinitions.TAG_INSTANCE;
-import static org.databiosphere.workspacedataservice.metrics.MetricsDefinitions.TAG_RECORD_TYPE;
 import static org.databiosphere.workspacedataservice.service.model.ReservedNames.RESERVED_NAME_PREFIX;
 
 import bio.terra.common.db.WriteTransaction;
@@ -220,10 +216,10 @@ public Map<String, DataTypeMapping> addOrUpdateColumnIfNeeded(
 
       // update a metrics counter with this schema change
       Counter counter =
-          Counter.builder(COUNTER_COL_CHANGE)
-              .tag(TAG_RECORD_TYPE, recordType.getName())
-              .tag(TAG_ATTRIBUTE_NAME, column)
-              .tag(TAG_INSTANCE, instanceId.toString())
+          Counter.builder("column.change.datatype")
+              .tag("RecordType", recordType.getName())
+              .tag("AttributeName", column)
+              .tag("Instance", instanceId.toString())
               .tag("OldDataType", valueDifference.leftValue().toString())
               .tag("NewDataType", updatedColType.toString())
               .description("Column schema changes")

service/src/test/java/org/databiosphere/workspacedataservice/dataimport/pfb/PfbTestSupport.java

@@ -1,6 +1,7 @@
 package org.databiosphere.workspacedataservice.dataimport.pfb;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import io.micrometer.observation.ObservationRegistry;
 import java.util.UUID;
 import org.databiosphere.workspacedataservice.activitylog.ActivityLogger;
 import org.databiosphere.workspacedataservice.dao.JobDao;
@@ -18,10 +19,17 @@ class PfbTestSupport {
   @Autowired private BatchWriteService batchWriteService;
   @Autowired private ActivityLogger activityLogger;
   @Autowired private ObjectMapper objectMapper;
+  @Autowired private ObservationRegistry observationRegistry;
 
   PfbQuartzJob buildPfbQuartzJob(UUID workspaceId) {
     return new PfbQuartzJob(
-        jobDao, wsmDao, restClientRetry, batchWriteService, activityLogger, workspaceId);
+        jobDao,
+        wsmDao,
+        restClientRetry,
+        batchWriteService,
+        activityLogger,
+        observationRegistry,
+        workspaceId);
   }
 
   PfbQuartzJob buildPfbQuartzJob() {

service/src/test/java/org/databiosphere/workspacedataservice/dataimport/tdr/TdrTestSupport.java

@@ -1,6 +1,7 @@
 package org.databiosphere.workspacedataservice.dataimport.tdr;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import io.micrometer.observation.ObservationRegistry;
 import java.net.URL;
 import java.util.UUID;
 import org.databiosphere.workspacedataservice.activitylog.ActivityLogger;
@@ -19,6 +20,7 @@ class TdrTestSupport {
   @Autowired private BatchWriteService batchWriteService;
   @Autowired private ActivityLogger activityLogger;
   @Autowired private ObjectMapper objectMapper;
+  @Autowired private ObservationRegistry observationRegistry;
 
   /** Returns a TdrManifestQuartzJob that is capable of pulling parquet files from the classpath. */
   TdrManifestQuartzJob buildTdrManifestQuartzJob(UUID workspaceId) {
@@ -29,7 +31,8 @@ TdrManifestQuartzJob buildTdrManifestQuartzJob(UUID workspaceId) {
         batchWriteService,
         activityLogger,
         workspaceId,
-        objectMapper) {
+        objectMapper,
+        observationRegistry) {
       @Override
       protected URL parseUrl(String path) {
         if (path.startsWith("classpath:")) {

service/src/test/java/org/databiosphere/workspacedataservice/jobexec/QuartzJobTest.java

@@ -2,20 +2,30 @@
 
 import static org.databiosphere.workspacedataservice.shared.model.Schedulable.ARG_TOKEN;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import io.micrometer.core.instrument.LongTaskTimer;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.Timer;
+import io.micrometer.observation.ObservationRegistry;
+import io.micrometer.observation.tck.TestObservationRegistry;
+import io.micrometer.observation.tck.TestObservationRegistryAssert;
 import java.time.OffsetDateTime;
 import java.time.ZoneId;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.databiosphere.workspacedataservice.dao.JobDao;
 import org.databiosphere.workspacedataservice.generated.GenericJobServerModel;
 import org.databiosphere.workspacedataservice.sam.TokenContextUtil;
+import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInstance;
@@ -24,6 +34,7 @@
 import org.quartz.JobExecutionException;
 import org.quartz.JobKey;
 import org.quartz.impl.JobDetailImpl;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.mock.mockito.MockBean;
 
@@ -32,6 +43,9 @@
 class QuartzJobTest {
 
   @MockBean JobDao jobDao;
+  @Autowired ObservationRegistry observationRegistry;
+  @Autowired MeterRegistry meterRegistry;
+  private TestObservationRegistry testObservationRegistry;
 
   @BeforeAll
   void beforeAll() {
@@ -50,6 +64,16 @@ void beforeAll() {
         .thenThrow(new RuntimeException("test failed via jobDao.fail()"));
     when(jobDao.fail(any(), anyString()))
         .thenThrow(new RuntimeException("test failed via jobDao.fail()"));
+    // set up metrics registries
+    testObservationRegistry = TestObservationRegistry.create();
+    Metrics.globalRegistry.add(meterRegistry);
+  }
+
+  @AfterAll
+  void afterAll() {
+    testObservationRegistry.clear();
+    meterRegistry.clear();
+    Metrics.globalRegistry.clear();
   }
 
   /**
@@ -61,7 +85,8 @@ class TestableQuartzJob extends QuartzJob {
 
     private final String expectedToken;
 
-    public TestableQuartzJob(String expectedToken) {
+    public TestableQuartzJob(String expectedToken, ObservationRegistry registry) {
+      super(registry);
       this.expectedToken = expectedToken;
     }
 
@@ -80,21 +105,62 @@ protected void executeInternal(UUID jobId, JobExecutionContext context) {
   void tokenIsStashedAndCleaned() throws JobExecutionException {
     // set an example token, via mock, into the Quartz JobDataMap
     String expectedToken = RandomStringUtils.randomAlphanumeric(10);
-    JobExecutionContext mockContext = mock(JobExecutionContext.class);
-    when(mockContext.getMergedJobDataMap())
-        .thenReturn(new JobDataMap(Map.of(ARG_TOKEN, expectedToken)));
-    JobDetailImpl jobDetail = new JobDetailImpl();
-    jobDetail.setKey(new JobKey(UUID.randomUUID().toString(), "bar"));
-    when(mockContext.getJobDetail()).thenReturn(jobDetail);
-
+    JobExecutionContext mockContext = setUpTestJob(expectedToken, UUID.randomUUID().toString());
     // assert that no token exists via TokenContextUtil
     assertTrue(
         TokenContextUtil.getToken().isEmpty(), "no token should exist before running the job");
     // execute the TestableQuartzJob, which asserts that the token passed properly from the
     // JobDataMap into job context and is therefore retrievable via TokenContextUtil
-    new TestableQuartzJob(expectedToken).execute(mockContext);
+    new TestableQuartzJob(expectedToken, observationRegistry).execute(mockContext);
     // assert that the token is cleaned up and no longer reachable via TokenContextUtil
     assertTrue(
         TokenContextUtil.getToken().isEmpty(), "no token should exist after running the job");
   }
+
+  @Test
+  void correctObservation() throws JobExecutionException {
+    String randomToken = RandomStringUtils.randomAlphanumeric(10);
+    String jobUuid = UUID.randomUUID().toString();
+    JobExecutionContext mockContext = setUpTestJob(randomToken, jobUuid);
+
+    // execute the TestableQuartzJob, then use testObservationRegistry to confirm observation
+    new TestableQuartzJob(randomToken, testObservationRegistry).execute(mockContext);
+
+    TestObservationRegistryAssert.assertThat(testObservationRegistry)
+        .doesNotHaveAnyRemainingCurrentObservation()
+        .hasObservationWithNameEqualTo("wds.job.execute")
+        .that()
+        .hasHighCardinalityKeyValue("jobId", jobUuid)
+        .hasLowCardinalityKeyValue("jobType", "TestableQuartzJob")
+        .hasLowCardinalityKeyValue("outcome", "success")
+        .hasBeenStarted()
+        .hasBeenStopped();
+  }
+
+  @Test
+  void correctMetrics() throws JobExecutionException {
+    String randomToken = RandomStringUtils.randomAlphanumeric(10);
+    JobExecutionContext mockContext = setUpTestJob(randomToken, UUID.randomUUID().toString());
+
+    // execute the TestableQuartzJob, then confirm metrics provisioned correctly
+    new TestableQuartzJob(randomToken, observationRegistry).execute(mockContext);
+
+    // metrics provisioned should be longTaskTimer and a Timer, confirm both ran
+    LongTaskTimer longTaskTimer = meterRegistry.find("wds.job.execute.active").longTaskTimer();
+    assertNotEquals(0, longTaskTimer.duration(TimeUnit.SECONDS));
+    Timer timer = meterRegistry.find("wds.job.execute").timer();
+    assertNotEquals(0, timer.count());
+  }
+
+  // sets up a job and returns the job context
+  private JobExecutionContext setUpTestJob(String randomToken, String jobUuid) {
+    // String jobUuid = UUID.randomUUID().toString();
+    JobExecutionContext mockContext = mock(JobExecutionContext.class);
+    when(mockContext.getMergedJobDataMap())
+        .thenReturn(new JobDataMap(Map.of(ARG_TOKEN, randomToken)));
+    JobDetailImpl jobDetail = new JobDetailImpl();
+    jobDetail.setKey(new JobKey(jobUuid, "bar"));
+    when(mockContext.getJobDetail()).thenReturn(jobDetail);
+    return mockContext;
+  }
 }

service/src/test/java/org/databiosphere/workspacedataservice/service/RecordServiceTest.java

@@ -1,8 +1,5 @@
 package org.databiosphere.workspacedataservice.service;
 
-import static org.databiosphere.workspacedataservice.metrics.MetricsDefinitions.TAG_ATTRIBUTE_NAME;
-import static org.databiosphere.workspacedataservice.metrics.MetricsDefinitions.TAG_INSTANCE;
-import static org.databiosphere.workspacedataservice.metrics.MetricsDefinitions.TAG_RECORD_TYPE;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 
@@ -93,9 +90,9 @@ void schemaChangesIncrementMetricsCounter() {
     // assert the counter has been incremented once
     assertEquals(1, counter.count());
     // validate the tags for that counter
-    assertEquals(recordType.getName(), counter.getId().getTag(TAG_RECORD_TYPE));
-    assertEquals("myAttr", counter.getId().getTag(TAG_ATTRIBUTE_NAME));
-    assertEquals(instanceId.toString(), counter.getId().getTag(TAG_INSTANCE));
+    assertEquals(recordType.getName(), counter.getId().getTag("RecordType"));
+    assertEquals("myAttr", counter.getId().getTag("AttributeName"));
+    assertEquals(instanceId.toString(), counter.getId().getTag("Instance"));
     assertEquals(DataTypeMapping.NUMBER.toString(), counter.getId().getTag("OldDataType"));
     assertEquals(DataTypeMapping.STRING.toString(), counter.getId().getTag("NewDataType"));
   }