Add Disable Failing Connections feature

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.35.49-alpha
+current_version = 0.35.53-alpha
 commit = False
 tag = False
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(\-[a-z]+)?

.env

@@ -10,7 +10,7 @@
 
 
 ### SHARED ###
-VERSION=0.35.49-alpha
+VERSION=0.35.53-alpha
 
 # When using the airbyte-db via default docker image
 CONFIG_ROOT=/data
@@ -91,3 +91,4 @@ MAX_DISCOVER_WORKERS=5
 
 ### FEATURE FLAGS ###
 NEW_SCHEDULER=false
+AUTO_DISABLE_FAILING_CONNECTIONS=false

.github/workflows/gradle.yml

@@ -368,7 +368,6 @@ jobs:
           label: ${{ needs.start-platform-new-scheduler-acceptance-runner.outputs.label }}
           ec2-instance-id: ${{ needs.start-platform-new-scheduler-acceptance-runner.outputs.ec2-instance-id }}
 
-
   # In case of self-hosted EC2 errors, remove this block.
   stop-platform-build-runner:
     name: "Platform: Stop Build EC2 Runner"
@@ -656,13 +655,16 @@ jobs:
       - name: Run Logging Tests
         run: ./tools/bin/cloud_storage_logging_test.sh
         env:
+          # AWS_S3_INTEGRATION_TEST_CREDS can be found in LastPass as AWS_S3_INTEGRATION_TEST_CREDS
           AWS_S3_INTEGRATION_TEST_CREDS: ${{ secrets.AWS_S3_INTEGRATION_TEST_CREDS }}
+          # GOOGLE_CLOUD_STORAGE_TEST_CREDS can be found in LastPass as "google cloud storage ( gcs ) test creds"
           GOOGLE_CLOUD_STORAGE_TEST_CREDS: ${{ secrets.GOOGLE_CLOUD_STORAGE_TEST_CREDS }}
 
       - name: Run Kubernetes End-to-End Acceptance Tests
         env:
           USER: root
           HOME: /home/runner
+          # AWS_S3_INTEGRATION_TEST_CREDS can be found in LastPass as AWS_S3_INTEGRATION_TEST_CREDS
           AWS_S3_INTEGRATION_TEST_CREDS: ${{ secrets.AWS_S3_INTEGRATION_TEST_CREDS }}
           SECRET_STORE_GCP_CREDENTIALS: ${{ secrets.SECRET_STORE_GCP_CREDENTIALS }}
           SECRET_STORE_GCP_PROJECT_ID: ${{ secrets.SECRET_STORE_GCP_PROJECT_ID }}
@@ -684,6 +686,7 @@ jobs:
         env:
           USER: root
           HOME: /home/runner
+          # AWS_S3_INTEGRATION_TEST_CREDS can be found in LastPass as AWS_S3_INTEGRATION_TEST_CREDS
           AWS_S3_INTEGRATION_TEST_CREDS: ${{ secrets.AWS_S3_INTEGRATION_TEST_CREDS }}
           SECRET_STORE_GCP_CREDENTIALS: ${{ secrets.SECRET_STORE_GCP_CREDENTIALS }}
           SECRET_STORE_GCP_PROJECT_ID: ${{ secrets.SECRET_STORE_GCP_PROJECT_ID }}
@@ -806,6 +809,7 @@ jobs:
         env:
           USER: root
           HOME: /home/runner
+          # AWS_S3_INTEGRATION_TEST_CREDS can be found in LastPass as AWS_S3_INTEGRATION_TEST_CREDS
           AWS_S3_INTEGRATION_TEST_CREDS: ${{ secrets.AWS_S3_INTEGRATION_TEST_CREDS }}
           SECRET_STORE_GCP_CREDENTIALS: ${{ secrets.SECRET_STORE_GCP_CREDENTIALS }}
           SECRET_STORE_GCP_PROJECT_ID: ${{ secrets.SECRET_STORE_GCP_PROJECT_ID }}

.github/workflows/terminate-zombie-build-instances.yml

@@ -0,0 +1,27 @@
+name: Terminate Zombie Build Instances
+
+on:
+  push:
+  schedule:
+    - cron: "0 */1 * * *"
+
+jobs:
+  terminate:
+    runs-on: ubuntu-latest
+    steps:
+      - name: List Instances Older Than an Hour
+        env:
+          AWS_ACCESS_KEY_ID: ${{ secrets.SELF_RUNNER_AWS_ACCESS_KEY_ID }}
+          AWS_SECRET_ACCESS_KEY: ${{ secrets.SELF_RUNNER_AWS_SECRET_ACCESS_KEY }}
+          # See https://github.com/aws/aws-cli/issues/5623
+          AWS_EC2_METADATA_DISABLED: true
+        run: |
+          set -euxo pipefail
+
+          aws configure set default.region us-east-2
+
+          # Since the AWS cli returns an ISO HH:MM timestamp, and Jq only accepts Z timestamps, we define a function toZ to convert this.
+          aws ec2 describe-instances --no-paginate --filters Name=instance-type,Values=c5.2xlarge Name=instance-state-name,Values=running \
+            --query 'Reservations[*].Instances[*].{Instance:InstanceId,LaunchTime:LaunchTime}' --output json \
+          | jq 'def toZ(str): str | (split("+")[0] + "Z") | fromdate ;
+              flatten | map( { InstanceId: .Instance, LaunchTime: toZ(.LaunchTime) } ) | map( select ( .LaunchTime < (now - 3600) ) )'

airbyte-analytics/readme.md

@@ -0,0 +1,3 @@
+# airbyte-analytics
+
+Java library with shared code for telemetry tracking including Segment.

airbyte-api/readme.md

@@ -0,0 +1,11 @@
+# airbyte-api
+
+Defines the OpenApi configuration for the Airbyte Configuration API. It also is responsible for generating the following from the API spec:
+* Java API client
+* Java API server - this generated code is used in `airbyte-server` to allow us to implement the Configuration API in a type safe way. See `ConfigurationApi.java` in `airbyte-server`
+* API docs
+
+## Key Files
+* src/openapi/config.yaml - Defines the config API interface using OpenApi3
+* AirbyteApiClient.java - wraps all api clients so that they can be dependency injected together
+* PatchedLogsApi.java - fixes generated code for log api.

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

@@ -3343,19 +3343,16 @@ components:
       description: Indicates where the error originated. If not set, the origin of error is not well known.
       type: string
       enum:
-        - unknown # todo (parker) remove this in favor of leaving the failureOrigin unset
         - source
         - destination
         - replication
-        - replication_worker # todo (parker) remove this in favor of replication
         - persistence
         - normalization
         - dbt
     AttemptFailureType:
       description: Categorizes well known errors into types for programmatic handling. If not set, the type of error is not well known.
       type: string
       enum:
-        - unknown # todo (parker) remove this in favor of leaving the failureType unset
         - config_error
         - system_error
         - manual_cancellation

airbyte-bootloader/Dockerfile

@@ -5,6 +5,6 @@ ENV APPLICATION airbyte-bootloader
 
 WORKDIR /app
 
-ADD bin/${APPLICATION}-0.35.49-alpha.tar /app
+ADD bin/${APPLICATION}-0.35.53-alpha.tar /app
 
-ENTRYPOINT ["/bin/bash", "-c", "${APPLICATION}-0.35.49-alpha/bin/${APPLICATION}"]
+ENTRYPOINT ["/bin/bash", "-c", "${APPLICATION}-0.35.53-alpha/bin/${APPLICATION}"]

airbyte-bootloader/readme.md

@@ -0,0 +1,6 @@
+# airbyte-bootloader
+
+This application runs at start up for Airbyte. It is responsible for making sure that the environment is upgraded and in a good state. e.g. It makes sure the database has been migrated to the correct version.
+
+## Entrypoint
+* BootloaderApp.java - has the main method for running the bootloader.

airbyte-bootloader/src/main/java/io/airbyte/bootloader/BootloaderApp.java

@@ -114,7 +114,7 @@ public void load() throws Exception {
 
       final ConfigPersistence configPersistence = DatabaseConfigPersistence.createWithValidation(configDatabase);
       final ConfigRepository configRepository =
-          new ConfigRepository(configPersistence, null, Optional.empty(), Optional.empty(), configDatabase);
+          new ConfigRepository(configPersistence, configDatabase);
 
       createWorkspaceIfNoneExists(configRepository);
       LOGGER.info("Default workspace created..");

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

@@ -72,7 +72,7 @@ void testBootloaderAppBlankDb() throws Exception {
         container.getPassword(),
         container.getJdbcUrl()).getInitialized();
     val jobsMigrator = new JobsDatabaseMigrator(jobDatabase, this.getClass().getName());
-    assertEquals("0.35.5.001", jobsMigrator.getLatestMigration().getVersion().getVersion());
+    assertEquals("0.35.40.001", jobsMigrator.getLatestMigration().getVersion().getVersion());
 
     val configDatabase = new ConfigsDatabaseInstance(
         mockedConfigs.getConfigDatabaseUser(),

airbyte-cli/readme.md

@@ -0,0 +1,3 @@
+# airbyte-cli
+
+Thin CLI over the Airbyte Configuration API to make it easier to interact with the API from the command line.

airbyte-commons-cli/readme.md

@@ -1 +1,3 @@
+# airbyte-commons-cli
+
 This module houses utility functions for the `commons-cli` library. It is separate from `commons`, because it depends on external library `commons-cli` which we do not want to introduce as a dependency to every module.

airbyte-commons-docker/readme.md

@@ -0,0 +1,3 @@
+# airbyte-commons-docker
+
+This module contains common helpers for interacting with Docker and Docker images from Java.

airbyte-commons/build.gradle

@@ -3,5 +3,5 @@ plugins {
 }
 
 dependencies {
-    // Dependencies for this module should be specified in the top-level build.gradle.
+    // Dependencies for this module should be specified in the top-level build.gradle. See readme for more explanation.
 }

airbyte-commons/readme.md

@@ -0,0 +1,7 @@
+# airbyte-commons
+
+Common java helpers.
+
+This submodule is inherited by all other java modules in the monorepo! It is therefore important that we do not add dependencies to it, as those dependencies will also be added to every java module. The only dependencies that this module uses are the ones declared in the `build.gradle` at the root of the Airbyte monorepo. In other words it only uses dependencies that are already shared across all modules. The `dependencies` section of the `build.gradle` of `airbyte-commons` should always be empty.
+
+For other common java code that needs to be shared across modules that requires additional dependencies, we follow this convention: `airbyte-commons-<name of lib>`. See for example `airbyte-commons-cli` and `airbyte-commons-docker`.

airbyte-commons/src/main/java/io/airbyte/commons/features/EnvVariableFeatureFlags.java

@@ -16,4 +16,11 @@ public boolean usesNewScheduler() {
     return Boolean.parseBoolean(System.getenv("NEW_SCHEDULER"));
   }
 
+  @Override
+  public boolean autoDisablesFailingConnections() {
+    log.info("Auto Disable Failing Connections: " + Boolean.parseBoolean(System.getenv("AUTO_DISABLE_FAILING_CONNECTIONS")));
+
+    return Boolean.parseBoolean(System.getenv("AUTO_DISABLE_FAILING_CONNECTIONS"));
+  }
+
 }

airbyte-commons/src/main/java/io/airbyte/commons/features/FeatureFlags.java

@@ -12,4 +12,6 @@ public interface FeatureFlags {
 
   boolean usesNewScheduler();
 
+  boolean autoDisablesFailingConnections();
+
 }

airbyte-commons/src/main/java/io/airbyte/commons/lang/Exceptions.java

@@ -6,9 +6,11 @@
 
 import java.util.concurrent.Callable;
 import java.util.function.Function;
+import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+@Slf4j
 public class Exceptions {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(Exceptions.class);

airbyte-commons/src/test/java/io/airbyte/commons/io/IOsTest.java

@@ -75,7 +75,7 @@ public void testGetTailExists() throws IOException {
         "line7",
         "line8");
 
-    final Writer writer = new BufferedWriter(new FileWriter(stdoutFile.toString(), true));
+    final Writer writer = new BufferedWriter(new FileWriter(stdoutFile.toString(), StandardCharsets.UTF_8, true));
 
     for (final String line : Iterables.concat(head, expectedTail)) {
       writer.write(line + "\n");

airbyte-commons/src/test/java/io/airbyte/commons/json/JsonsTest.java

@@ -18,6 +18,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
+import java.nio.charset.StandardCharsets;
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
@@ -55,7 +56,7 @@ void testSerializeJsonNode() {
     assertEquals(
         "{\"test\":\"dGVzdA==\"}",
         Jsons.serialize(Jsons.jsonNode(ImmutableMap.of(
-            "test", new BinaryNode("test".getBytes())))));
+            "test", new BinaryNode("test".getBytes(StandardCharsets.UTF_8))))));
   }
 
   @Test

airbyte-config/init/readme.md

@@ -0,0 +1,8 @@
+# airbyte-config:init
+
+This module fulfills two responsibilities:
+1. It is where we declare what connectors should ship with the Platform. See below for more instruction on how it works.
+2. It contains the scripts and Dockerfile that allow the `docker-compose` version of Airbyte to mount the local filesystem. This is helpful in cases where a user wants to use a connector that interacts with (reads data from or writes data to) the local filesystem. e.g. `destination-local-json`.
+
+## Declaring connectors that ship with the Platform
+In order to have a connector ship with the Platform is must be present in the respective `source_definitions.yaml` or `destination_definitions.yaml` files in `src/main/resources/seed`. If a connector is added there, the build system will handle fetching its spec and adding it to `source_specs.yaml` or `destination_specs.yaml`. See the gradle tasks to understand how this all works. The logic for fetching the specs is in `airbyte-config:specs`.

airbyte-config/init/src/main/java/io/airbyte/config/init/YamlSeedConfigPersistence.java

@@ -35,7 +35,7 @@
  */
 public class YamlSeedConfigPersistence implements ConfigPersistence {
 
-  public static Class<?> DEFAULT_SEED_DEFINITION_RESOURCE_CLASS = SeedType.class;
+  public static final Class<?> DEFAULT_SEED_DEFINITION_RESOURCE_CLASS = SeedType.class;
 
   private static final Map<AirbyteConfig, SeedType> CONFIG_SCHEMA_MAP = Map.of(
       ConfigSchema.STANDARD_SOURCE_DEFINITION, SeedType.STANDARD_SOURCE_DEFINITION,

airbyte-config/init/src/main/resources/seed/destination_definitions.yaml

@@ -203,7 +203,7 @@
 - name: S3
   destinationDefinitionId: 4816b78f-1489-44c1-9060-4b19d5fa9362
   dockerRepository: airbyte/destination-s3
-  dockerImageTag: 0.2.9
+  dockerImageTag: 0.2.10
   documentationUrl: https://docs.airbyte.io/integrations/destinations/s3
   icon: s3.svg
   resourceRequirements:
@@ -221,7 +221,7 @@
 - name: Snowflake
   destinationDefinitionId: 424892c4-daac-4491-b35d-c6688ba547ba
   dockerRepository: airbyte/destination-snowflake
-  dockerImageTag: 0.4.17
+  dockerImageTag: 0.4.19
   documentationUrl: https://docs.airbyte.io/integrations/destinations/snowflake
   icon: snowflake.svg
   resourceRequirements:

airbyte-config/init/src/main/resources/seed/destination_specs.yaml

@@ -3454,7 +3454,7 @@
     supported_destination_sync_modes:
     - "append"
     - "overwrite"
-- dockerImage: "airbyte/destination-s3:0.2.9"
+- dockerImage: "airbyte/destination-s3:0.2.10"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/s3"
     connectionSpecification:
@@ -3825,7 +3825,7 @@
     supported_destination_sync_modes:
     - "overwrite"
     - "append"
-- dockerImage: "airbyte/destination-snowflake:0.4.17"
+- dockerImage: "airbyte/destination-snowflake:0.4.19"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/snowflake"
     connectionSpecification:
@@ -3844,9 +3844,12 @@
       properties:
         host:
           description: "The host domain of the snowflake instance (must include the\
-            \ account, region, cloud environment, and end with snowflakecomputing.com)."
+            \ account, region, cloud environment, and end with snowflakecomputing.com).\
+            \ The account identifier differs depending on your cloud region, be sure\
+            \ to verify with Snowflake's documentation."
           examples:
           - "accountname.us-east-2.aws.snowflakecomputing.com"
+          - "accountname.snowflakecomputing.com"
           type: "string"
           title: "Host"
           order: 0

airbyte-config/models/src/main/java/io/airbyte/config/Configs.java

@@ -249,6 +249,18 @@ public interface Configs {
    */
   Map<String, String> getJobDefaultEnvMap();
 
+  /**
+   * Defines the number of consecutive job failures required before a connection is auto-disabled if
+   * the AUTO_DISABLE_FAILING_CONNECTIONS flag is set to true.
+   */
+  int getMaxFailedJobsInARowBeforeConnectionDisable();
+
+  /**
+   * Defines the required number of days with only failed jobs before a connection is auto-disabled if
+   * the AUTO_DISABLE_FAILING_CONNECTIONS flag is set to true.
+   */
+  int getMaxDaysOfOnlyFailedJobsBeforeConnectionDisable();
+
   // Jobs - Kube only
   /**
    * Define the check job container's minimum CPU request. Defaults to

airbyte-config/models/src/main/java/io/airbyte/config/EnvConfigs.java

@@ -121,6 +121,9 @@ public class EnvConfigs implements Configs {
   private static final String SHOULD_RUN_SYNC_WORKFLOWS = "SHOULD_RUN_SYNC_WORKFLOWS";
   private static final String SHOULD_RUN_CONNECTION_MANAGER_WORKFLOWS = "SHOULD_RUN_CONNECTION_MANAGER_WORKFLOWS";
 
+  private static final String MAX_FAILED_JOBS_IN_A_ROW_BEFORE_CONNECTION_DISABLE = "MAX_FAILED_JOBS_IN_A_ROW_BEFORE_CONNECTION_DISABLE";
+  private static final String MAX_DAYS_OF_ONLY_FAILED_JOBS_BEFORE_CONNECTION_DISABLE = "MAX_DAYS_OF_ONLY_FAILED_JOBS_BEFORE_CONNECTION_DISABLE";
+
   // job-type-specific overrides
   public static final String SPEC_JOB_KUBE_NODE_SELECTORS = "SPEC_JOB_KUBE_NODE_SELECTORS";
   public static final String CHECK_JOB_KUBE_NODE_SELECTORS = "CHECK_JOB_KUBE_NODE_SELECTORS";
@@ -178,6 +181,9 @@ public class EnvConfigs implements Configs {
 
   public static final int DEFAULT_TEMPORAL_HISTORY_RETENTION_IN_DAYS = 30;
 
+  public static final int DEFAULT_FAILED_JOBS_IN_A_ROW_BEFORE_CONNECTION_DISABLE = 100;
+  public static final int DEFAULT_DAYS_OF_ONLY_FAILED_JOBS_BEFORE_CONNECTION_DISABLE = 14;
+
   private final Function<String, String> getEnv;
   private final Supplier<Set<String>> getAllEnvKeys;
   private final LogConfigs logConfigs;
@@ -661,6 +667,16 @@ public Map<String, String> getJobDefaultEnvMap() {
     return MoreMaps.merge(jobPrefixedEnvMap, jobSharedEnvMap);
   }
 
+  @Override
+  public int getMaxFailedJobsInARowBeforeConnectionDisable() {
+    return getEnvOrDefault(MAX_FAILED_JOBS_IN_A_ROW_BEFORE_CONNECTION_DISABLE, DEFAULT_FAILED_JOBS_IN_A_ROW_BEFORE_CONNECTION_DISABLE);
+  }
+
+  @Override
+  public int getMaxDaysOfOnlyFailedJobsBeforeConnectionDisable() {
+    return getEnvOrDefault(MAX_DAYS_OF_ONLY_FAILED_JOBS_BEFORE_CONNECTION_DISABLE, DEFAULT_DAYS_OF_ONLY_FAILED_JOBS_BEFORE_CONNECTION_DISABLE);
+  }
+
   @Override
   public String getCheckJobMainContainerCpuRequest() {
     return getEnvOrDefault(CHECK_JOB_MAIN_CONTAINER_CPU_REQUEST, getJobMainContainerCpuRequest());

airbyte-config/models/src/main/java/io/airbyte/config/helpers/GcsLogs.java

@@ -15,6 +15,7 @@
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Supplier;
@@ -89,7 +90,7 @@ public List<String> tailCloudLog(final LogConfigs configs, final String logPath,
       final var poppedBlob = descendingTimestampBlobs.remove(0);
       try (final var inMemoryData = new ByteArrayOutputStream()) {
         poppedBlob.downloadTo(inMemoryData);
-        final var currFileLines = inMemoryData.toString().split("\n");
+        final var currFileLines = inMemoryData.toString(StandardCharsets.UTF_8).split("\n");
         final List<String> currFileLinesReversed = Lists.reverse(List.of(currFileLines));
         for (final var line : currFileLinesReversed) {
           if (linesRead == numLines) {

airbyte-config/models/src/main/java/io/airbyte/config/helpers/LogClientSingleton.java

@@ -117,7 +117,7 @@ public List<String> getJobLogFile(final WorkerEnvironment workerEnvironment, fin
    */
   @VisibleForTesting
   public void deleteLogs(final WorkerEnvironment workerEnvironment, final LogConfigs logConfigs, final String logPath) {
-    if (logPath == null || logPath.equals(Path.of(""))) {
+    if (logPath == null || logPath.equals("")) {
       return;
     }