Bmoric/extract logs api

airbyte-server/src/main/java/io/airbyte/server/ConfigurationApiFactory.java

@@ -36,7 +36,6 @@ public class ConfigurationApiFactory implements Factory<ConfigurationApi> {
   private static TrackingClient trackingClient;
   private static WorkerEnvironment workerEnvironment;
   private static LogConfigs logConfigs;
-  private static Path workspaceRoot;
   private static AirbyteVersion airbyteVersion;
   private static HttpClient httpClient;
   private static EventRunner eventRunner;
@@ -69,7 +68,6 @@ public static void setValues(
     ConfigurationApiFactory.trackingClient = trackingClient;
     ConfigurationApiFactory.workerEnvironment = workerEnvironment;
     ConfigurationApiFactory.logConfigs = logConfigs;
-    ConfigurationApiFactory.workspaceRoot = workspaceRoot;
     ConfigurationApiFactory.airbyteVersion = airbyteVersion;
     ConfigurationApiFactory.httpClient = httpClient;
     ConfigurationApiFactory.eventRunner = eventRunner;
@@ -91,7 +89,6 @@ public ConfigurationApi provide() {
         ConfigurationApiFactory.workerEnvironment,
         ConfigurationApiFactory.logConfigs,
         ConfigurationApiFactory.airbyteVersion,
-        ConfigurationApiFactory.workspaceRoot,
         ConfigurationApiFactory.httpClient,
         ConfigurationApiFactory.eventRunner);
   }

airbyte-server/src/main/java/io/airbyte/server/ServerApp.java

@@ -60,8 +60,12 @@
 import io.airbyte.server.handlers.DestinationDefinitionsHandler;
 import io.airbyte.server.handlers.DestinationHandler;
 import io.airbyte.server.handlers.HealthCheckHandler;
+import io.airbyte.server.handlers.JobHistoryHandler;
+import io.airbyte.server.handlers.LogsHandler;
 import io.airbyte.server.handlers.OperationsHandler;
 import io.airbyte.server.handlers.SchedulerHandler;
+import io.airbyte.server.handlers.SourceDefinitionsHandler;
+import io.airbyte.server.handlers.SourceHandler;
 import io.airbyte.server.scheduler.DefaultSynchronousSchedulerClient;
 import io.airbyte.server.scheduler.EventRunner;
 import io.airbyte.server.scheduler.TemporalEventRunner;
@@ -302,6 +306,28 @@ public static ServerRunnable getServer(final ServerFactory apiFactory,
 
     final HealthCheckHandler healthCheckHandler = new HealthCheckHandler(configRepository);
 
+    final SourceHandler sourceHandler = new SourceHandler(
+        configRepository,
+        secretsRepositoryReader,
+        secretsRepositoryWriter,
+        schemaValidator,
+        connectionsHandler);
+
+    final SourceDefinitionsHandler sourceDefinitionsHandler = new SourceDefinitionsHandler(configRepository, syncSchedulerClient, sourceHandler);
+
+    final JobHistoryHandler jobHistoryHandler = new JobHistoryHandler(
+        jobPersistence,
+        configs.getWorkerEnvironment(),
+        configs.getLogConfigs(),
+        connectionsHandler,
+        sourceHandler,
+        sourceDefinitionsHandler,
+        destinationHandler,
+        destinationDefinitionsHandler,
+        configs.getAirbyteVersion());
+
+    final LogsHandler logsHandler = new LogsHandler(configs);
+
     LOGGER.info("Starting server...");
 
     return apiFactory.create(
@@ -327,6 +353,8 @@ public static ServerRunnable getServer(final ServerFactory apiFactory,
         destinationDefinitionsHandler,
         destinationHandler,
         healthCheckHandler,
+        jobHistoryHandler,
+        logsHandler,
         operationsHandler,
         schedulerHandler);
   }

airbyte-server/src/main/java/io/airbyte/server/ServerFactory.java

@@ -22,26 +22,34 @@
 import io.airbyte.server.apis.DestinationDefinitionApiController;
 import io.airbyte.server.apis.DestinationDefinitionSpecificationApiController;
 import io.airbyte.server.apis.HealthApiController;
+import io.airbyte.server.apis.JobsApiController;
+import io.airbyte.server.apis.LogsApiController;
 import io.airbyte.server.apis.binders.AttemptApiBinder;
 import io.airbyte.server.apis.binders.ConnectionApiBinder;
 import io.airbyte.server.apis.binders.DbMigrationBinder;
 import io.airbyte.server.apis.binders.DestinationApiBinder;
 import io.airbyte.server.apis.binders.DestinationDefinitionApiBinder;
 import io.airbyte.server.apis.binders.DestinationDefinitionSpecificationApiBinder;
 import io.airbyte.server.apis.binders.HealthApiBinder;
+import io.airbyte.server.apis.binders.JobsApiBinder;
+import io.airbyte.server.apis.binders.LogsApiBinder;
 import io.airbyte.server.apis.factories.AttemptApiFactory;
 import io.airbyte.server.apis.factories.ConnectionApiFactory;
 import io.airbyte.server.apis.factories.DbMigrationApiFactory;
 import io.airbyte.server.apis.factories.DestinationApiFactory;
 import io.airbyte.server.apis.factories.DestinationDefinitionApiFactory;
 import io.airbyte.server.apis.factories.DestinationDefinitionSpecificationApiFactory;
 import io.airbyte.server.apis.factories.HealthApiFactory;
+import io.airbyte.server.apis.factories.JobsApiFactory;
+import io.airbyte.server.apis.factories.LogsApiFactory;
 import io.airbyte.server.handlers.AttemptHandler;
 import io.airbyte.server.handlers.ConnectionsHandler;
 import io.airbyte.server.handlers.DbMigrationHandler;
 import io.airbyte.server.handlers.DestinationDefinitionsHandler;
 import io.airbyte.server.handlers.DestinationHandler;
 import io.airbyte.server.handlers.HealthCheckHandler;
+import io.airbyte.server.handlers.JobHistoryHandler;
+import io.airbyte.server.handlers.LogsHandler;
 import io.airbyte.server.handlers.OperationsHandler;
 import io.airbyte.server.handlers.SchedulerHandler;
 import io.airbyte.server.scheduler.EventRunner;
@@ -77,6 +85,8 @@ ServerRunnable create(final SynchronousSchedulerClient synchronousSchedulerClien
                         final DestinationDefinitionsHandler destinationDefinitionsHandler,
                         final DestinationHandler destinationApiHandler,
                         final HealthCheckHandler healthCheckHandler,
+                        final JobHistoryHandler jobHistoryHandler,
+                        final LogsHandler logsHandler,
                         final OperationsHandler operationsHandler,
                         final SchedulerHandler schedulerHandler);
 
@@ -105,6 +115,8 @@ public ServerRunnable create(final SynchronousSchedulerClient synchronousSchedul
                                  final DestinationDefinitionsHandler destinationDefinitionsHandler,
                                  final DestinationHandler destinationApiHandler,
                                  final HealthCheckHandler healthCheckHandler,
+                                 final JobHistoryHandler jobHistoryHandler,
+                                 final LogsHandler logsHandler,
                                  final OperationsHandler operationsHandler,
                                  final SchedulerHandler schedulerHandler) {
       final Map<String, String> mdc = MDC.getCopyOfContextMap();
@@ -148,6 +160,10 @@ public ServerRunnable create(final SynchronousSchedulerClient synchronousSchedul
 
       HealthApiFactory.setValues(healthCheckHandler);
 
+      JobsApiFactory.setValues(jobHistoryHandler, schedulerHandler);
+
+      LogsApiFactory.setValues(logsHandler);
+
       // server configurations
       final Set<Class<?>> componentClasses = Set.of(
           ConfigurationApi.class,
@@ -157,7 +173,9 @@ public ServerRunnable create(final SynchronousSchedulerClient synchronousSchedul
           DestinationApiController.class,
           DestinationDefinitionApiController.class,
           DestinationDefinitionSpecificationApiController.class,
-          HealthApiController.class);
+          HealthApiController.class,
+          JobsApiController.class,
+          LogsApiController.class);
 
       final Set<Object> components = Set.of(
           new CorsFilter(),
@@ -168,7 +186,9 @@ public ServerRunnable create(final SynchronousSchedulerClient synchronousSchedul
           new DestinationApiBinder(),
           new DestinationDefinitionApiBinder(),
           new DestinationDefinitionSpecificationApiBinder(),
-          new HealthApiBinder());
+          new HealthApiBinder(),
+          new JobsApiBinder(),
+          new LogsApiBinder());
 
       // construct server
       return new ServerApp(airbyteVersion, componentClasses, components);

airbyte-server/src/main/java/io/airbyte/server/apis/ConfigurationApi.java

@@ -117,7 +117,6 @@
 import io.airbyte.server.handlers.DestinationDefinitionsHandler;
 import io.airbyte.server.handlers.DestinationHandler;
 import io.airbyte.server.handlers.JobHistoryHandler;
-import io.airbyte.server.handlers.LogsHandler;
 import io.airbyte.server.handlers.OAuthHandler;
 import io.airbyte.server.handlers.OpenApiConfigHandler;
 import io.airbyte.server.handlers.OperationsHandler;
@@ -135,7 +134,6 @@
 import java.io.File;
 import java.io.IOException;
 import java.net.http.HttpClient;
-import java.nio.file.Path;
 import java.util.Map;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.NotImplementedException;
@@ -156,12 +154,8 @@ public class ConfigurationApi implements io.airbyte.api.generated.V1Api {
   private final JobHistoryHandler jobHistoryHandler;
   private final WebBackendConnectionsHandler webBackendConnectionsHandler;
   private final WebBackendGeographiesHandler webBackendGeographiesHandler;
-  private final LogsHandler logsHandler;
   private final OpenApiConfigHandler openApiConfigHandler;
   private final OAuthHandler oAuthHandler;
-  private final WorkerEnvironment workerEnvironment;
-  private final LogConfigs logConfigs;
-  private final Path workspaceRoot;
 
   public ConfigurationApi(final ConfigRepository configRepository,
                           final JobPersistence jobPersistence,
@@ -173,12 +167,8 @@ public ConfigurationApi(final ConfigRepository configRepository,
                           final WorkerEnvironment workerEnvironment,
                           final LogConfigs logConfigs,
                           final AirbyteVersion airbyteVersion,
-                          final Path workspaceRoot,
                           final HttpClient httpClient,
                           final EventRunner eventRunner) {
-    this.workerEnvironment = workerEnvironment;
-    this.logConfigs = logConfigs;
-    this.workspaceRoot = workspaceRoot;
 
     final JsonSchemaValidator schemaValidator = new JsonSchemaValidator();
 
@@ -235,7 +225,6 @@ public ConfigurationApi(final ConfigRepository configRepository,
         eventRunner,
         configRepository);
     webBackendGeographiesHandler = new WebBackendGeographiesHandler();
-    logsHandler = new LogsHandler();
     openApiConfigHandler = new OpenApiConfigHandler();
   }
 
@@ -889,41 +878,69 @@ public SourceDiscoverSchemaRead executeSourceDiscoverSchema(final SourceCoreConf
     return execute(() -> schedulerHandler.discoverSchemaForSourceFromSourceCreate(sourceCreate));
   }
 
+  /**
+   * This implementation has been moved to {@link JobsApiController}. Since the path of
+   * {@link JobsApiController} is more granular, it will override this implementation
+   */
   @Override
   public JobInfoRead cancelJob(final JobIdRequestBody jobIdRequestBody) {
-    return execute(() -> schedulerHandler.cancelJob(jobIdRequestBody));
+    throw new NotImplementedException();
   }
 
   // JOB HISTORY
 
+  /**
+   * This implementation has been moved to {@link JobsApiController}. Since the path of
+   * {@link JobsApiController} is more granular, it will override this implementation
+   */
   @Override
   public JobReadList listJobsFor(final JobListRequestBody jobListRequestBody) {
-    return execute(() -> jobHistoryHandler.listJobsFor(jobListRequestBody));
+    throw new NotImplementedException();
   }
 
+  /**
+   * This implementation has been moved to {@link JobsApiController}. Since the path of
+   * {@link JobsApiController} is more granular, it will override this implementation
+   */
   @Override
   public JobInfoRead getJobInfo(final JobIdRequestBody jobIdRequestBody) {
-    return execute(() -> jobHistoryHandler.getJobInfo(jobIdRequestBody));
+    throw new NotImplementedException();
   }
 
+  /**
+   * This implementation has been moved to {@link JobsApiController}. Since the path of
+   * {@link JobsApiController} is more granular, it will override this implementation
+   */
   @Override
   public JobInfoLightRead getJobInfoLight(final JobIdRequestBody jobIdRequestBody) {
-    return execute(() -> jobHistoryHandler.getJobInfoLight(jobIdRequestBody));
+    throw new NotImplementedException();
   }
 
+  /**
+   * This implementation has been moved to {@link JobsApiController}. Since the path of
+   * {@link JobsApiController} is more granular, it will override this implementation
+   */
   @Override
   public JobDebugInfoRead getJobDebugInfo(final JobIdRequestBody jobIdRequestBody) {
-    return execute(() -> jobHistoryHandler.getJobDebugInfo(jobIdRequestBody));
+    throw new NotImplementedException();
   }
 
+  /**
+   * This implementation has been moved to {@link JobsApiController}. Since the path of
+   * {@link JobsApiController} is more granular, it will override this implementation
+   */
   @Override
   public AttemptNormalizationStatusReadList getAttemptNormalizationStatusesForJob(final JobIdRequestBody jobIdRequestBody) {
     return execute(() -> jobHistoryHandler.getAttemptNormalizationStatuses(jobIdRequestBody));
   }
 
+  /**
+   * This implementation has been moved to {@link LogsApiController}. Since the path of
+   * {@link LogsApiController} is more granular, it will override this implementation
+   */
   @Override
   public File getLogs(final LogsRequestBody logsRequestBody) {
-    return execute(() -> logsHandler.getLogs(workspaceRoot, workerEnvironment, logConfigs, logsRequestBody));
+    throw new NotImplementedException();
   }
 
   @Override

airbyte-server/src/main/java/io/airbyte/server/apis/JobsApiController.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.server.apis;
+
+import io.airbyte.api.generated.JobsApi;
+import io.airbyte.api.model.generated.AttemptNormalizationStatusReadList;
+import io.airbyte.api.model.generated.JobDebugInfoRead;
+import io.airbyte.api.model.generated.JobIdRequestBody;
+import io.airbyte.api.model.generated.JobInfoLightRead;
+import io.airbyte.api.model.generated.JobInfoRead;
+import io.airbyte.api.model.generated.JobListRequestBody;
+import io.airbyte.api.model.generated.JobReadList;
+import io.airbyte.server.handlers.JobHistoryHandler;
+import io.airbyte.server.handlers.SchedulerHandler;
+import javax.ws.rs.Path;
+import lombok.AllArgsConstructor;
+
+@Path("/v1/jobs")
+@AllArgsConstructor
+public class JobsApiController implements JobsApi {
+
+  private final JobHistoryHandler jobHistoryHandler;
+  private final SchedulerHandler schedulerHandler;
+
+  @Override
+  public JobInfoRead cancelJob(final JobIdRequestBody jobIdRequestBody) {
+    return ConfigurationApi.execute(() -> schedulerHandler.cancelJob(jobIdRequestBody));
+  }
+
+  @Override
+  public AttemptNormalizationStatusReadList getAttemptNormalizationStatusesForJob(final JobIdRequestBody jobIdRequestBody) {
+    return ConfigurationApi.execute(() -> jobHistoryHandler.getAttemptNormalizationStatuses(jobIdRequestBody));
+  }
+
+  @Override
+  public JobDebugInfoRead getJobDebugInfo(final JobIdRequestBody jobIdRequestBody) {
+    return ConfigurationApi.execute(() -> jobHistoryHandler.getJobDebugInfo(jobIdRequestBody));
+  }
+
+  @Override
+  public JobInfoRead getJobInfo(final JobIdRequestBody jobIdRequestBody) {
+    return ConfigurationApi.execute(() -> jobHistoryHandler.getJobInfo(jobIdRequestBody));
+  }
+
+  @Override
+  public JobInfoLightRead getJobInfoLight(final JobIdRequestBody jobIdRequestBody) {
+    return ConfigurationApi.execute(() -> jobHistoryHandler.getJobInfoLight(jobIdRequestBody));
+  }
+
+  @Override
+  public JobReadList listJobsFor(final JobListRequestBody jobListRequestBody) {
+    return ConfigurationApi.execute(() -> jobHistoryHandler.listJobsFor(jobListRequestBody));
+  }
+
+}

airbyte-server/src/main/java/io/airbyte/server/apis/LogsApiController.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.server.apis;
+
+import io.airbyte.api.generated.LogsApi;
+import io.airbyte.api.model.generated.LogsRequestBody;
+import io.airbyte.server.handlers.LogsHandler;
+import java.io.File;
+import javax.ws.rs.Path;
+import lombok.AllArgsConstructor;
+
+@Path("/v1/logs/get")
+@AllArgsConstructor
+public class LogsApiController implements LogsApi {
+
+  private final LogsHandler logsHandler;
+
+  @Override
+  public File getLogs(final LogsRequestBody logsRequestBody) {
+    return ConfigurationApi.execute(() -> logsHandler.getLogs(logsRequestBody));
+  }
+
+}

airbyte-server/src/main/java/io/airbyte/server/apis/binders/JobsApiBinder.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.server.apis.binders;
+
+import io.airbyte.server.apis.JobsApiController;
+import io.airbyte.server.apis.factories.JobsApiFactory;
+import org.glassfish.hk2.utilities.binding.AbstractBinder;
+import org.glassfish.jersey.process.internal.RequestScoped;
+
+public class JobsApiBinder extends AbstractBinder {
+
+  @Override
+  protected void configure() {
+    bindFactory(JobsApiFactory.class)
+        .to(JobsApiController.class)
+        .in(RequestScoped.class);
+  }
+
+}