Add async observability to QuartzJob

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,10 @@
 import static org.databiosphere.workspacedataservice.sam.BearerTokenFilter.ATTRIBUTE_NAME_TOKEN;
 import static org.databiosphere.workspacedataservice.shared.model.Schedulable.ARG_TOKEN;
 
+import io.micrometer.core.instrument.observation.DefaultMeterObservationHandler;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+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 +33,32 @@
 // 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());
-    try {
+    // configure observation to collect counter, timer, and longtasktimer metrics
+    observationRegistry
+        .observationConfig()
+        .observationHandler(new DefaultMeterObservationHandler(new SimpleMeterRegistry()));
+    Observation observation =
+        Observation.start("wds.job.execute", observationRegistry)
+            .contextualName("job-execution")
+            .lowCardinalityKeyValue("jobType", getClass().getSimpleName())
+            .highCardinalityKeyValue("jobId", jobId.toString());
+    try (Observation.Scope scope = observation.openScope()) {
       // 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 +70,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.event(Observation.Event.of("job.succeeded"));
     } catch (Exception e) {
       // on any otherwise-unhandled exception, mark the job as failed
       getJobDao().fail(jobId, e);
+      observation.error(e);
+      observation.event(Observation.Event.of("job.failed"));
     } finally {
       JobContextHolder.destroy();
+      observation.stop();
     }
   }
 

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

@@ -8,6 +8,7 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import io.micrometer.observation.ObservationRegistry;
 import java.time.OffsetDateTime;
 import java.time.ZoneId;
 import java.util.Map;
@@ -24,6 +25,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 +34,7 @@
 class QuartzJobTest {
 
   @MockBean JobDao jobDao;
+  @Autowired ObservationRegistry observationRegistry;
 
   @BeforeAll
   void beforeAll() {
@@ -62,6 +65,7 @@ class TestableQuartzJob extends QuartzJob {
     private final String expectedToken;
 
     public TestableQuartzJob(String expectedToken) {
+      super(observationRegistry);
       this.expectedToken = expectedToken;
     }