fix(platform): use updated configuration after check during syncs #21899

pedroslopez · 2023-01-26T05:31:12Z

What

fixes #20912

This PR builds on the changes introduced in #21629. While the aforementioned PR addressed using updated state/config values between attempts, this PR solves the issue of updating configuration within an attempt. This can happen when, for example, an oauth token is refreshed during the check operation that can run prior to the SyncWorkflow.

How

This implements the second part of the solution as laid out in the tech spec, changing when we're generating the sync input and how we are generating the check inputs.

Before this change, the SyncWorkflow input was being generated at the start of the ConnectionManagerWorkflow, and was then converted into Check inputs if needed. The activities look something like this:

(these are snippets of diagrams, the full views can be found in the tech spec)

After this change, we separate the check input generation into its own activity method. Then, we can move the SyncWorkflow input generation to after the check activity runs. Because the previous PR updated the SyncWorkflow input generation activity to fetch latest configs, this will take into consideration any changes that happened during check.

This ends up looking something like this:

The change is versioned, so ConnectionManagerWorkflows that executed before this change will still follow the old path/behavior.

🚨 User Impact 🚨

No breaking change. Configs updated during check will now correctly be retrieved for the sync.

…after checks

pedroslopez · 2023-02-02T04:36:30Z

...kers/src/main/java/io/airbyte/workers/temporal/scheduling/ConnectionManagerWorkflowImpl.java

+    final SyncJobCheckConnectionInputs checkInputs;
+    if (generateCheckInputVersion < GENERATE_CHECK_INPUT_CURRENT_VERSION && jobInputs != null) {
+      checkInputs = getCheckConnectionInputFromSync(jobInputs);
+    } else {
+      checkInputs = getCheckConnectionInput();
+    }


Previous versions of the workflow will have retrieved the sync input, which we can use to build the check inputs like was being done before.

For new workflow executions, we will instead run an activity to generate the Check inputs specifically.

pedroslopez · 2023-02-03T00:56:22Z

...kers/src/main/java/io/airbyte/workers/temporal/scheduling/ConnectionManagerWorkflowImpl.java

+      final int generateCheckInputVersion =
+          Workflow.getVersion(GENERATE_CHECK_INPUT_TAG, Workflow.DEFAULT_VERSION, GENERATE_CHECK_INPUT_CURRENT_VERSION);
+
+      GeneratedJobInput jobInputs = null;
+      if (generateCheckInputVersion < GENERATE_CHECK_INPUT_CURRENT_VERSION) {
+        jobInputs = getJobInput();
+      }

      reportJobStarting(connectionUpdaterInput.getConnectionId());
      StandardSyncOutput standardSyncOutput = null;

      try {
-        final SyncCheckConnectionFailure syncCheckConnectionFailure = checkConnections(jobInputs);
+        final SyncCheckConnectionFailure syncCheckConnectionFailure = checkConnections(getJobRunConfig(), jobInputs);
        if (syncCheckConnectionFailure.isFailed()) {
          final StandardSyncOutput checkFailureOutput = syncCheckConnectionFailure.buildFailureOutput();
          workflowState.setFailed(getFailStatus(checkFailureOutput));
          reportFailure(connectionUpdaterInput, checkFailureOutput, FailureCause.CONNECTION);
        } else {
+          if (generateCheckInputVersion >= GENERATE_CHECK_INPUT_CURRENT_VERSION) {
+            jobInputs = getJobInput();
+          }


This is where we are moving the activity that generates the SyncWorkflow inputs to after the potential checks. Based on the workflow version, we keep the old position of the activity.

We do this so that any config updates that happen during check are freshly retrieved before running the sync.

pedroslopez · 2023-02-03T01:10:06Z

...c/main/java/io/airbyte/workers/temporal/scheduling/activities/GenerateInputActivityImpl.java

+    return sourceLauncherConfig;
+  }
+
+  private IntegrationLauncherConfig getDestinationIntegrationLauncherConfig(final long jobId,


most of the changes in this file are pulling out some common logic used for both SyncWorkflow and Check input generation.

evantahler · 2023-02-03T01:06:45Z

...kers/src/main/java/io/airbyte/workers/temporal/scheduling/ConnectionManagerWorkflowImpl.java

+  private static final String GENERATE_CHECK_INPUT_TAG = "generate_check_input";
+  private static final int GENERATE_CHECK_INPUT_CURRENT_VERSION = 1;


... time to version I guess!

If you didn't version, I think that running syncs would fail, but perhaps only before the sync stage, not during or afterwords? If that's the case, and they retry, that's not so bad?

Yeah, that sounds right. Avoiding the failure wasn't too bad here with the version though so seemed worth it to keep history compatibility. FWIW there's also a test that verifies history can be replayed that not versioning would break 😄

evantahler · 2023-02-03T01:09:15Z

...s/src/main/java/io/airbyte/workers/temporal/scheduling/activities/GenerateInputActivity.java

@@ -36,6 +37,18 @@ class SyncInputWithAttemptNumber {

  }

+  @Data
+  @NoArgsConstructor
+  @AllArgsConstructor


@AllArgsConstructor - today I learned

evantahler · 2023-02-03T01:12:01Z

...kers/src/main/java/io/airbyte/workers/temporal/scheduling/ConnectionManagerWorkflowImpl.java

+    if (!isLastJobOrAttemptFailure) {
+      log.info("SOURCE CHECK: Skipped, last attempt was not a failure");
+      log.info("DESTINATION CHECK: Skipped, last attempt was not a failure");
+      return checkFailure;
+    }


I think I'm confused by this conditional - if we are skipping CHECKS because the previous job/attempt didn't fail, why would we return a failure? Or is checkFailure initialized with isFailed(false)

This was already the case before, I just moved this higher up to be clearer / more intentional about it.

Yes, a SyncCheckConnectionFailure is always returned by this method. When an actual failure happens, checkFailure.setFailureOrigin and checkFailure.setFailureOutput are called. The SyncCheckConnectionFailure class has a method .isFailed() that actually returns whether it failed or not if these are set.

Can we add a comment on the method? This is confusing.

Ok, this was definitely confusing :D

Renamed SyncCheckConnectionFailure to SyncCheckConnectionResult to be more clear that it doesn't always represent a failure.

mfsiega-airbyte · 2023-02-08T00:22:47Z

High-level comment - I think it would be worthwhile to break this down into smaller PRs, to (1) reduce the likelihood that something goes wrong; (2) minimise the impact if something does go wrong.

The db migration should be on its own, for rollback safety.
I haven't taken a super close look but I'd imagine some of the workflow/persistence/API changes can be broken into separate changes?

pedroslopez · 2023-02-08T00:54:06Z

@mfsiega-airbyte whoops, 100% - I did separate this into a smaller set of changes but when I merged the other piece this one now includes everything. Sorting that out now.

pedroslopez · 2023-02-08T01:03:41Z

@mfsiega-airbyte OK - this should be more manageable now

pedroslopez · 2023-02-21T23:01:45Z

superseded by airbytehq/airbyte-platform#91

pedroslopez added 24 commits January 19, 2023 19:01

add AttemptSyncConfig, move info out of JobSyncConfig

c65c003

get build working

5dbea60

Merge branch 'master' into pedroslopez/attempt-sync-config

7c82560

add db migration

5aa96b9

load config when building attempts

2bfe226

persist AttemptSyncConfig

6c4e0cd

it compiles

0ea4e4b

fix job persistence test

8d97db0

implement submitSync with attempt config

2349a44

fix TemporalClientTest

de41dc6

reorganizing some code

74578dc

add GenerateInputActivity test

fbaf075

verify AttemptSyncConfig is persisted

fef3e5d

add test for persistence changes

532e8fd

add test for getAttemptByNumber

f31e201

Merge branch 'master' into pedroslopez/attempt-sync-config

6431d5c

use apis rather than direct db access

1411e04

Merge branch 'master' into pedroslopez/attempt-sync-config

2f2eaa3

fix compatibility with master

9ecd1b2

copy update

773d4b6

fix tests for allowed hosts addition

f5ed8de

remove debug logging

9dbf869

Merge branch 'master' into pedroslopez/attempt-sync-config

1541875

run activity to generate check inputs, move sync input generation to …

9e8c33f

…after checks

pedroslopez changed the base branch from master to pedroslopez/attempt-sync-config January 26, 2023 05:31

octavia-squidington-iv added area/platform issues related to the platform area/worker Related to worker labels Jan 26, 2023

pedroslopez temporarily deployed to more-secrets January 26, 2023 05:33 — with GitHub Actions Inactive

only run activity to generate check input if necessary

6891917

pedroslopez requested a review from a team February 3, 2023 00:55

pedroslopez commented Feb 3, 2023

View reviewed changes

pedroslopez changed the title ~~fix(platform): handle updating configuration during check~~ fix(platform): use updated configuration after check during syncs Feb 3, 2023

pedroslopez commented Feb 3, 2023

View reviewed changes

evantahler reviewed Feb 3, 2023

View reviewed changes

SyncCheckConnectionFailure -> SyncCheckConnectionResult for clarity

77d2195

pedroslopez temporarily deployed to more-secrets February 6, 2023 18:00 — with GitHub Actions Inactive

Base automatically changed from revert-22211-revert-21629-pedroslopez/attempt-sync-config to master February 7, 2023 23:42

pedroslopez requested a review from a team as a code owner February 7, 2023 23:42

Merge branch 'master' into pedroslopez/check-input-gen

f7894cd

pedroslopez temporarily deployed to more-secrets February 8, 2023 00:59 — with GitHub Actions Inactive

fix merge issue

1267d3e

pedroslopez temporarily deployed to more-secrets February 8, 2023 01:10 — with GitHub Actions Inactive

Merge branch 'master' into pedroslopez/check-input-gen

e243745

pedroslopez temporarily deployed to more-secrets February 10, 2023 19:51 — with GitHub Actions Inactive

consider oauth

d24e428

pedroslopez temporarily deployed to more-secrets February 10, 2023 20:16 — with GitHub Actions Inactive

This was referenced Feb 15, 2023

add activity to fetch feature flags #22960

Closed

add activity to fetch feature flags airbytehq/airbyte-platform#79

Merged

fix(platform): use updated configuration after check during syncs airbytehq/airbyte-platform#91

Merged

pedroslopez closed this Feb 21, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix(platform): use updated configuration after check during syncs #21899

fix(platform): use updated configuration after check during syncs #21899

pedroslopez commented Jan 26, 2023 •

edited

Loading

pedroslopez Feb 2, 2023

pedroslopez Feb 3, 2023

pedroslopez Feb 3, 2023

evantahler Feb 3, 2023

evantahler Feb 3, 2023

pedroslopez Feb 3, 2023

evantahler Feb 3, 2023

evantahler Feb 3, 2023

pedroslopez Feb 3, 2023

benmoriceau Feb 6, 2023

pedroslopez Feb 6, 2023

mfsiega-airbyte commented Feb 8, 2023

pedroslopez commented Feb 8, 2023

pedroslopez commented Feb 8, 2023

pedroslopez commented Feb 21, 2023

		private static final String GENERATE_CHECK_INPUT_TAG = "generate_check_input";
		private static final int GENERATE_CHECK_INPUT_CURRENT_VERSION = 1;

fix(platform): use updated configuration after check during syncs #21899

fix(platform): use updated configuration after check during syncs #21899

Conversation

pedroslopez commented Jan 26, 2023 • edited Loading

What

How

🚨 User Impact 🚨

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mfsiega-airbyte commented Feb 8, 2023

pedroslopez commented Feb 8, 2023

pedroslopez commented Feb 8, 2023

pedroslopez commented Feb 21, 2023

pedroslopez commented Jan 26, 2023 •

edited

Loading