Rbroughan/add logpath and input payload blob (#9832)

Author: tryangul

airbyte-api/src/main/openapi/workload-openapi.yaml

@@ -203,6 +203,8 @@ components:
     Workload:
       required:
       - id
+      - inputPayload
+      - logPath
       type: object
       properties:
         id:
@@ -223,6 +225,10 @@ components:
           nullable: true
           items:
             $ref: '#/components/schemas/WorkloadLabel'
+        inputPayload:
+          type: string
+        logPath:
+          type: string
     WorkloadCancelRequest:
       required:
       - reason
@@ -248,6 +254,7 @@ components:
           type: string
     WorkloadCreateRequest:
       required:
+      - logPath
       - workloadId
       - workloadInput
       type: object
@@ -261,6 +268,8 @@ components:
             $ref: '#/components/schemas/WorkloadLabel'
         workloadInput:
           type: string
+        logPath:
+          type: string
     WorkloadHeartbeatRequest:
       required:
       - workloadId

airbyte-bootloader/src/test/java/io/airbyte/bootloader/BootloaderTest.java

@@ -92,7 +92,7 @@ class BootloaderTest {
 
   // ⚠️ This line should change with every new migration to show that you meant to make a new
   // migration to the prod database
-  private static final String CURRENT_CONFIGS_MIGRATION_VERSION = "0.50.33.005";
+  private static final String CURRENT_CONFIGS_MIGRATION_VERSION = "0.50.33.006";
   private static final String CURRENT_JOBS_MIGRATION_VERSION = "0.50.4.001";
   private static final String CDK_VERSION = "1.2.3";
 

airbyte-commons-worker/src/main/java/io/airbyte/workers/sync/WorkloadApiWorker.java

@@ -56,7 +56,7 @@ public WorkloadApiWorker(final DocumentStoreClient documentStoreClient,
   }
 
   @Override
-  public ReplicationOutput run(final ReplicationInput replicationInput, Path jobRoot) throws WorkerException {
+  public ReplicationOutput run(final ReplicationInput replicationInput, final Path jobRoot) throws WorkerException {
     final String serializedInput = Jsons.serialize(input);
     final String workloadId = workloadIdGenerator.generate(replicationInput.getConnectionId(),
         Long.parseLong(replicationInput.getJobRunConfig().getJobId()),
@@ -69,6 +69,7 @@ public ReplicationOutput run(final ReplicationInput replicationInput, Path jobRo
     // Create the workload
     createWorkload(new WorkloadCreateRequest(workloadId,
         serializedInput,
+        jobRoot.toString(),
         List.of(
             new WorkloadLabel("connectionId", replicationInput.getConnectionId().toString()),
             new WorkloadLabel("jobId", replicationInput.getJobRunConfig().getJobId()),

airbyte-db/db-lib/src/main/java/io/airbyte/db/instance/configs/migrations/V0_50_33_006__AddInputPayloadAndLogPathColumnsToWorkload.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.db.instance.configs.migrations;
+
+import org.flywaydb.core.api.migration.BaseJavaMigration;
+import org.flywaydb.core.api.migration.Context;
+import org.jooq.DSLContext;
+import org.jooq.Field;
+import org.jooq.impl.DSL;
+import org.jooq.impl.SQLDataType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class V0_50_33_006__AddInputPayloadAndLogPathColumnsToWorkload extends BaseJavaMigration {
+
+  private static final String TABLE = "workload";
+  private static final String PAYLOAD_COLUMN_NAME = "input_payload";
+  private static final String LOG_PATH_COLUMN_NAME = "log_path";
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(V0_50_33_006__AddInputPayloadAndLogPathColumnsToWorkload.class);
+
+  @Override
+  public void migrate(final Context context) throws Exception {
+    LOGGER.info("Running migration: {}", this.getClass().getSimpleName());
+
+    // Warning: please do not use any jOOQ generated code to write a migration.
+    // As database schema changes, the generated jOOQ code can be deprecated. So
+    // old migration may not compile if there is any generated code.
+    final DSLContext ctx = DSL.using(context.getConnection());
+
+    final Field<String> payloadColumn = DSL.field(PAYLOAD_COLUMN_NAME, SQLDataType.CLOB.nullable(false));
+    ctx.alterTable(TABLE)
+        .addColumnIfNotExists(payloadColumn)
+        .execute();
+
+    final Field<String> logPathColumn = DSL.field(LOG_PATH_COLUMN_NAME, SQLDataType.CLOB.nullable(false));
+    ctx.alterTable(TABLE)
+        .addColumnIfNotExists(logPathColumn)
+        .execute();
+  }
+
+}

airbyte-db/db-lib/src/main/resources/configs_database/schema_dump.txt

@@ -357,6 +357,8 @@ create table "public"."workload" (
   "created_at" timestamp(6) with time zone not null default current_timestamp,
   "updated_at" timestamp(6) with time zone not null default current_timestamp,
   "last_heartbeat_at" timestamp(6) with time zone,
+  "input_payload" text not null,
+  "log_path" text not null,
   constraint "workload_pkey"
     primary key ("id")
 );

airbyte-workload-api-server/src/main/kotlin/io/airbyte/workload/api/WorkloadApi.kt

@@ -73,6 +73,8 @@ open class WorkloadApi(
     workloadHandler.createWorkload(
       workloadCreateRequest.workloadId,
       workloadCreateRequest.labels,
+      workloadCreateRequest.workloadInput,
+      workloadCreateRequest.logPath,
     )
     workloadService.create(workloadId = workloadCreateRequest.workloadId, workloadInput = workloadCreateRequest.workloadInput)
   }

airbyte-workload-api-server/src/main/kotlin/io/airbyte/workload/api/domain/Workload.kt

@@ -10,4 +10,6 @@ data class Workload(
   var status: WorkloadStatus? = null,
   var lastHeartbeatAt: OffsetDateTime? = null,
   var labels: MutableList<WorkloadLabel>? = null,
+  var inputPayload: String = "",
+  var logPath: String = "",
 )

airbyte-workload-api-server/src/main/kotlin/io/airbyte/workload/api/domain/WorkloadCreateRequest.kt

@@ -8,4 +8,5 @@ data class WorkloadCreateRequest(
   var labels: List<WorkloadLabel>? = null,
   @Schema(required = true)
   var workloadInput: String = "",
+  var logPath: String = "",
 )

airbyte-workload-api-server/src/main/kotlin/io/airbyte/workload/handler/WorkloadHandler.kt

@@ -7,7 +7,7 @@ import java.time.OffsetDateTime
 
 /**
  * In order to mock a class it needs to be open. We have added this interface to avoid making [WorkloadHandler] an open class.
- * There is a chance that once we move to micronaut 4 this issue will be resolved and we can remove this interface.
+ * There is a chance that once we move to micronaut 4 this issue will be resolved, and we can remove this interface.
  */
 interface WorkloadHandler {
   fun getWorkload(workloadId: String): ApiWorkload
@@ -22,6 +22,8 @@ interface WorkloadHandler {
   fun createWorkload(
     workloadId: String,
     labels: List<WorkloadLabel>?,
+    input: String,
+    logPath: String,
   )
 
   @Transactional

airbyte-workload-api-server/src/main/kotlin/io/airbyte/workload/handler/WorkloadHandlerImpl.kt

@@ -51,6 +51,8 @@ class WorkloadHandlerImpl(
   override fun createWorkload(
     workloadId: String,
     labels: List<WorkloadLabel>?,
+    input: String,
+    logPath: String,
   ) {
     val workloadAlreadyExists = workloadRepository.existsById(workloadId)
     if (workloadAlreadyExists) {
@@ -63,6 +65,8 @@ class WorkloadHandlerImpl(
         status = WorkloadStatus.pending,
         lastHeartbeatAt = null,
         workloadLabels = labels?.map { it.toDomain() },
+        inputPayload = input,
+        logPath = logPath,
       )
 
     workloadRepository.save(domainWorkload).toApi()