-
Notifications
You must be signed in to change notification settings - Fork 4.2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
switch catalog discovery and syncs to temporal (#2333)
* add discover worker * oops forgot to merge this * add sync workflow * add sync to job submitter * working after these I mean * make workerrun use shared path, run formatting
- Loading branch information
Showing
12 changed files
with
351 additions
and
44 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
119 changes: 119 additions & 0 deletions
119
airbyte-scheduler/src/main/java/io/airbyte/scheduler/temporal/DiscoverCatalogWorkflow.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,119 @@ | ||
| /* | ||
| * MIT License | ||
| * | ||
| * Copyright (c) 2020 Airbyte | ||
| * | ||
| * Permission is hereby granted, free of charge, to any person obtaining a copy | ||
| * of this software and associated documentation files (the "Software"), to deal | ||
| * in the Software without restriction, including without limitation the rights | ||
| * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell | ||
| * copies of the Software, and to permit persons to whom the Software is | ||
| * furnished to do so, subject to the following conditions: | ||
| * | ||
| * The above copyright notice and this permission notice shall be included in all | ||
| * copies or substantial portions of the Software. | ||
| * | ||
| * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR | ||
| * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, | ||
| * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE | ||
| * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER | ||
| * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | ||
| * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE | ||
| * SOFTWARE. | ||
| */ | ||
|
|
||
| package io.airbyte.scheduler.temporal; | ||
|
|
||
| import com.google.common.base.Preconditions; | ||
| import io.airbyte.config.IntegrationLauncherConfig; | ||
| import io.airbyte.config.JobOutput; | ||
| import io.airbyte.config.StandardDiscoverCatalogInput; | ||
| import io.airbyte.workers.DefaultDiscoverCatalogWorker; | ||
| import io.airbyte.workers.JobStatus; | ||
| import io.airbyte.workers.OutputAndStatus; | ||
| import io.airbyte.workers.WorkerUtils; | ||
| import io.airbyte.workers.process.AirbyteIntegrationLauncher; | ||
| import io.airbyte.workers.process.IntegrationLauncher; | ||
| import io.airbyte.workers.process.ProcessBuilderFactory; | ||
| import io.airbyte.workers.protocols.airbyte.AirbyteStreamFactory; | ||
| import io.airbyte.workers.protocols.airbyte.DefaultAirbyteStreamFactory; | ||
| import io.airbyte.workers.wrappers.JobOutputDiscoverSchemaWorker; | ||
| import io.temporal.activity.ActivityInterface; | ||
| import io.temporal.activity.ActivityMethod; | ||
| import io.temporal.activity.ActivityOptions; | ||
| import io.temporal.workflow.Workflow; | ||
| import io.temporal.workflow.WorkflowInterface; | ||
| import io.temporal.workflow.WorkflowMethod; | ||
| import java.nio.file.Path; | ||
| import java.time.Duration; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| @WorkflowInterface | ||
| public interface DiscoverCatalogWorkflow { | ||
|
|
||
| @WorkflowMethod | ||
| JobOutput run(IntegrationLauncherConfig launcherConfig, StandardDiscoverCatalogInput config) throws TemporalJobException; | ||
|
|
||
| class WorkflowImpl implements DiscoverCatalogWorkflow { | ||
|
|
||
| final ActivityOptions options = ActivityOptions.newBuilder() | ||
| .setScheduleToCloseTimeout(Duration.ofHours(2)) | ||
| .build(); | ||
| private final DiscoverCatalogActivity activity = Workflow.newActivityStub(DiscoverCatalogActivity.class, options); | ||
|
|
||
| @Override | ||
| public JobOutput run(IntegrationLauncherConfig launcherConfig, StandardDiscoverCatalogInput config) throws TemporalJobException { | ||
| return activity.run(launcherConfig, config); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| @ActivityInterface | ||
| interface DiscoverCatalogActivity { | ||
|
|
||
| @ActivityMethod | ||
| JobOutput run(IntegrationLauncherConfig launcherConfig, StandardDiscoverCatalogInput config) throws TemporalJobException; | ||
|
|
||
| } | ||
|
|
||
| class DiscoverCatalogActivityImpl implements DiscoverCatalogActivity { | ||
|
|
||
| private static final Logger LOGGER = LoggerFactory.getLogger(DiscoverCatalogActivityImpl.class); | ||
|
|
||
| private final ProcessBuilderFactory pbf; | ||
| private final Path workspaceRoot; | ||
|
|
||
| public DiscoverCatalogActivityImpl(ProcessBuilderFactory pbf, Path workspaceRoot) { | ||
| this.pbf = pbf; | ||
| this.workspaceRoot = workspaceRoot; | ||
| } | ||
|
|
||
| public JobOutput run(IntegrationLauncherConfig launcherConfig, StandardDiscoverCatalogInput config) { | ||
| try { | ||
| final Path jobRoot = WorkerUtils.getJobRoot(workspaceRoot, launcherConfig); | ||
| WorkerUtils.setJobMdc(jobRoot, launcherConfig.getJobId()); | ||
|
|
||
| final IntegrationLauncher integrationLauncher = | ||
| new AirbyteIntegrationLauncher(launcherConfig.getJobId(), launcherConfig.getAttemptId().intValue(), launcherConfig.getDockerImage(), pbf); | ||
| final AirbyteStreamFactory streamFactory = new DefaultAirbyteStreamFactory(); | ||
|
|
||
| final OutputAndStatus<JobOutput> run = | ||
| new JobOutputDiscoverSchemaWorker( | ||
| new DefaultDiscoverCatalogWorker(integrationLauncher, streamFactory)).run(config, jobRoot); | ||
| if (run.getStatus() == JobStatus.SUCCEEDED) { | ||
| Preconditions.checkState(run.getOutput().isPresent()); | ||
| LOGGER.info("job output {}", run.getOutput().get()); | ||
| return run.getOutput().get(); | ||
| } else { | ||
| throw new RuntimeException("Discover catalog worker completed with a FAILED status."); | ||
| } | ||
|
|
||
| } catch (Exception e) { | ||
| throw new RuntimeException("Discover catalog job failed with an exception", e); | ||
| } | ||
| } | ||
|
|
||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
139 changes: 139 additions & 0 deletions
139
airbyte-scheduler/src/main/java/io/airbyte/scheduler/temporal/SyncWorkflow.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,139 @@ | ||
| /* | ||
| * MIT License | ||
| * | ||
| * Copyright (c) 2020 Airbyte | ||
| * | ||
| * Permission is hereby granted, free of charge, to any person obtaining a copy | ||
| * of this software and associated documentation files (the "Software"), to deal | ||
| * in the Software without restriction, including without limitation the rights | ||
| * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell | ||
| * copies of the Software, and to permit persons to whom the Software is | ||
| * furnished to do so, subject to the following conditions: | ||
| * | ||
| * The above copyright notice and this permission notice shall be included in all | ||
| * copies or substantial portions of the Software. | ||
| * | ||
| * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR | ||
| * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, | ||
| * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE | ||
| * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER | ||
| * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | ||
| * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE | ||
| * SOFTWARE. | ||
| */ | ||
|
|
||
| package io.airbyte.scheduler.temporal; | ||
|
|
||
| import com.google.common.base.Preconditions; | ||
| import io.airbyte.config.JobOutput; | ||
| import io.airbyte.config.StandardSyncInput; | ||
| import io.airbyte.workers.DefaultSyncWorker; | ||
| import io.airbyte.workers.JobStatus; | ||
| import io.airbyte.workers.OutputAndStatus; | ||
| import io.airbyte.workers.WorkerUtils; | ||
| import io.airbyte.workers.normalization.NormalizationRunnerFactory; | ||
| import io.airbyte.workers.process.AirbyteIntegrationLauncher; | ||
| import io.airbyte.workers.process.IntegrationLauncher; | ||
| import io.airbyte.workers.process.ProcessBuilderFactory; | ||
| import io.airbyte.workers.protocols.airbyte.AirbyteMessageTracker; | ||
| import io.airbyte.workers.protocols.airbyte.DefaultAirbyteDestination; | ||
| import io.airbyte.workers.protocols.airbyte.DefaultAirbyteSource; | ||
| import io.airbyte.workers.protocols.airbyte.NamespacingMapper; | ||
| import io.airbyte.workers.wrappers.JobOutputSyncWorker; | ||
| import io.temporal.activity.ActivityInterface; | ||
| import io.temporal.activity.ActivityMethod; | ||
| import io.temporal.activity.ActivityOptions; | ||
| import io.temporal.workflow.Workflow; | ||
| import io.temporal.workflow.WorkflowInterface; | ||
| import io.temporal.workflow.WorkflowMethod; | ||
| import java.nio.file.Path; | ||
| import java.time.Duration; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| @WorkflowInterface | ||
| public interface SyncWorkflow { | ||
|
|
||
| @WorkflowMethod | ||
| JobOutput run(long jobId, long attemptId, String sourceDockerImage, String destinationDockerImage, StandardSyncInput syncInput) | ||
| throws TemporalJobException; | ||
|
|
||
| class WorkflowImpl implements SyncWorkflow { | ||
|
|
||
| final ActivityOptions options = ActivityOptions.newBuilder() | ||
| .setScheduleToCloseTimeout(Duration.ofDays(3)) | ||
| .build(); | ||
| private final SyncActivity activity = Workflow.newActivityStub(SyncActivity.class, options); | ||
|
|
||
| @Override | ||
| public JobOutput run(long jobId, long attemptId, String sourceDockerImage, String destinationDockerImage, StandardSyncInput syncInput) | ||
| throws TemporalJobException { | ||
| return activity.run(jobId, attemptId, sourceDockerImage, destinationDockerImage, syncInput); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| @ActivityInterface | ||
| interface SyncActivity { | ||
|
|
||
| @ActivityMethod | ||
| JobOutput run(long jobId, long attemptId, String sourceDockerImage, String destinationDockerImage, StandardSyncInput syncInput) | ||
| throws TemporalJobException; | ||
|
|
||
| } | ||
|
|
||
| class SyncActivityImpl implements SyncActivity { | ||
|
|
||
| private static final Logger LOGGER = LoggerFactory.getLogger(SyncActivityImpl.class); | ||
|
|
||
| private final ProcessBuilderFactory pbf; | ||
| private final Path workspaceRoot; | ||
|
|
||
| public SyncActivityImpl(ProcessBuilderFactory pbf, Path workspaceRoot) { | ||
| this.pbf = pbf; | ||
| this.workspaceRoot = workspaceRoot; | ||
| } | ||
|
|
||
| public JobOutput run(long jobId, long attemptId, String sourceDockerImage, String destinationDockerImage, StandardSyncInput syncInput) { | ||
| try { | ||
| final Path jobRoot = WorkerUtils.getJobRoot(workspaceRoot, jobId, attemptId); | ||
| WorkerUtils.setJobMdc(jobRoot, jobId); | ||
|
|
||
| final int intAttemptId = Math.toIntExact(attemptId); | ||
|
|
||
| final IntegrationLauncher sourceLauncher = | ||
| new AirbyteIntegrationLauncher(jobId, intAttemptId, sourceDockerImage, pbf); | ||
| final IntegrationLauncher destinationLauncher = | ||
| new AirbyteIntegrationLauncher(jobId, intAttemptId, destinationDockerImage, pbf); | ||
|
|
||
| final DefaultAirbyteSource airbyteSource = new DefaultAirbyteSource(sourceLauncher); | ||
|
|
||
| final OutputAndStatus<JobOutput> run = | ||
| new JobOutputSyncWorker( | ||
| new DefaultSyncWorker( | ||
| jobId, | ||
| intAttemptId, | ||
| airbyteSource, | ||
| new NamespacingMapper(syncInput.getDefaultNamespace()), | ||
| new DefaultAirbyteDestination(destinationLauncher), | ||
| new AirbyteMessageTracker(), | ||
| NormalizationRunnerFactory.create( | ||
| destinationDockerImage, | ||
| pbf, | ||
| syncInput.getDestinationConfiguration()))).run(syncInput, jobRoot); | ||
| if (run.getStatus() == JobStatus.SUCCEEDED) { | ||
| Preconditions.checkState(run.getOutput().isPresent()); | ||
| LOGGER.info("job output {}", run.getOutput().get()); | ||
| return run.getOutput().get(); | ||
| } else { | ||
| throw new RuntimeException("Sync worker completed with a FAILED status."); | ||
| } | ||
|
|
||
| } catch (Exception e) { | ||
| throw new RuntimeException("Sync job failed with an exception", e); | ||
| } | ||
| } | ||
|
|
||
| } | ||
|
|
||
| } |
Oops, something went wrong.