Kube Logs are stored and retrieved from Cloud Storage.

.env

@@ -21,3 +21,16 @@ WEBAPP_URL=http://localhost:8000/
 API_URL=/api/v1/
 TEMPORAL_HOST=airbyte-temporal:7233
 INTERNAL_API_HOST=airbyte-server:8001
+
+# Fill the below variables to log to cloud storage. The provided credentials require both S3 read/write permissions. The logger attempts to create the bucket if
+# it does not exist.
+# S3
+S3_LOG_BUCKET=
+S3_LOG_BUCKET_REGION=
+AWS_ACCESS_KEY_ID=
+AWS_SECRET_ACCESS_KEY=
+
+# The point to a file that is mounted via volumes. The most convenient is to place this in LOCAL_ROOT.
+# GCS
+GCP_STORAGE_BUCKET=
+GOOGLE_APPLICATION_CREDENTIALS=/tmp/airbyte_local/credentials.json

.github/workflows/gradle.yml

@@ -96,6 +96,11 @@ jobs:
       - name: Build
         run: CORE_ONLY=true ./gradlew --no-daemon build --scan
 
+      - name: Run Logging Tests
+        run: ./tools/bin/logging_test.sh
+        env:
+          AWS_S3_INTEGRATION_TEST_CREDS: ${{ secrets.AWS_S3_INTEGRATION_TEST_CREDS }}
+
       - name: Ensure no file change
         run: git status --porcelain && test -z "$(git status --porcelain)"
 

airbyte-commons/build.gradle

@@ -3,6 +3,5 @@ plugins {
 }
 
 dependencies {
-    implementation 'org.apache.commons:commons-compress:1.20'
-    implementation 'org.apache.commons:commons-lang3:3.11'
+    // Dependencies for this module should be specified in the top-level build.gradle.
 }

airbyte-commons/src/main/resources/log4j2.xml

@@ -1,28 +1,60 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<Configuration status="INFO">
+<Configuration status="TRACE">
+<!--<Configuration status="INFO">-->
     <Properties>
         <Property name="default-pattern">%d{yyyy-MM-dd HH:mm:ss}{GMT+0} %highlight{%p} %C{1.}(%M):%L - %X - %m%n</Property>
         <Property name="default-worker-file-pattern">%d{yyyy-MM-dd HH:mm:ss}{GMT+0} %p (%X{job_root}) %C{1}(%M):%L - %m%n</Property>
+
+        <Property name="s3-bucket">$${env:S3_LOG_BUCKET}</Property>
+        <Property name="s3-region">$${env:S3_LOG_BUCKET_REGION}</Property>
+        <Property name="s3-aws-key">$${env:AWS_ACCESS_KEY_ID}</Property>
+        <Property name="s3-aws-secret">$${env:AWS_SECRET_ACCESS_KEY}</Property>
+
+        <Property name="gcp-storage-bucket">$${env:GCP_STORAGE_BUCKET}</Property>
+
     </Properties>
 
     <Appenders>
         <Console name="Default" target="SYSTEM_OUT">
             <PatternLayout pattern="${default-pattern}"/>
         </Console>
+
         <Routing name="LogSplit">
             <Routes pattern="$${ctx:job_root}">
                 <!-- Don't split logs if job_root isn't defined -->
                 <Route key="$${ctx:job_root}">
                     <Null name="/dev/null"/>
                 </Route>
                 <Route>
-                    <File name="${ctx:job_root}" fileName="${ctx:job_root}/${ctx:job_log_filename}">
+                    <File name="${ctx:job_root}-local" fileName="${ctx:job_root}/${ctx:job_log_filename}">
                         <PatternLayout pattern="${default-worker-file-pattern}"/>
                     </File>
                 </Route>
             </Routes>
             <IdlePurgePolicy timeToLive="15" timeUnit="minutes"/>
         </Routing>
+        <Routing name="LogSplitCloud">
+            <Routes pattern="$${ctx:job_root}">
+                <!-- Don't split logs if job_root isn't defined -->
+                <Route key="$${ctx:job_root}">
+                    <Null name="/dev/null"/>
+                </Route>
+                <Route>
+                    <Log4j2Appender name="${ctx:job_root}-cloud"
+                      verbose="true"
+                      stagingBufferAge="1"
+                      s3Bucket="${s3-bucket}" s3Path="job-logging${ctx:job_root}/${ctx:job_log_filename}" s3Region="${s3-region}"
+                      s3AwsKey="${s3-aws-key}" s3AwsSecret="${s3-aws-secret}"
+                      gcpStorageBucket="${gcp-storage-bucket}" gcpStorageBlobNamePrefix="job-logging${ctx:job_root}/${ctx:job_log_filename}">
+                        <PatternLayout>
+                            <Pattern>%d{yyyy-MM-dd HH:mm:ss} %-5p %m%n</Pattern>
+                        </PatternLayout>
+                    </Log4j2Appender>
+                </Route>
+            </Routes>
+            <IdlePurgePolicy timeToLive="15" timeUnit="minutes"/>
+        </Routing>
+
         <Routing name="AppLogSplit">
             <Routes pattern="$${ctx:workspace_app_root}">
                 <!-- Don't split logs if workspace_app_log_root isn't defined -->
@@ -31,7 +63,7 @@
                 </Route>
                 <Route>
                     <RollingFile
-                            name="${ctx:workspace_app_root}"
+                            name="${ctx:workspace_app_root}-local"
                             fileName="${ctx:workspace_app_root}/logs.log"
                             filePattern="${ctx:workspace_app_root}/logs.%i.log.gz"
                             ignoreExceptions="false">
@@ -47,13 +79,35 @@
             </Routes>
             <IdlePurgePolicy timeToLive="15" timeUnit="minutes"/>
         </Routing>
+        <Routing name="AppLogSplitCloud">
+            <Routes pattern="$${ctx:workspace_app_root}">
+                <!-- Don't split logs if workspace_app_log_root isn't defined -->
+                <Route key="$${ctx:workspace_app_root}">
+                    <Null name="/dev/null"/>
+                </Route>
+                <Route>
+                    <Log4j2Appender name="app-logging/${ctx:workspace_app_root}-cloud/"
+                      stagingBufferAge="1"
+                      s3Bucket="${s3-bucket}" s3Path="app-logging${ctx:workspace_app_root}" s3Region="${s3-region}"
+                      s3AwsKey="${s3-aws-key}" s3AwsSecret="${s3-aws-secret}"
+                      gcpStorageBucket="${gcp-storage-bucket}" gcpStorageBlobNamePrefix="app-logging${ctx:workspace_app_root}">
+                        <PatternLayout>
+                            <Pattern>%d{yyyy-MM-dd HH:mm:ss} %-5p %m%n</Pattern>
+                        </PatternLayout>
+                    </Log4j2Appender>
+                </Route>
+            </Routes>
+            <IdlePurgePolicy timeToLive="15" timeUnit="minutes"/>
+        </Routing>
     </Appenders>
 
     <Loggers>
         <Root level="INFO">
             <AppenderRef ref="Default"/>
             <AppenderRef ref="LogSplit"/>
+            <AppenderRef ref="LogSplitCloud"/>
             <AppenderRef ref="AppLogSplit"/>
+            <AppenderRef ref="AppLogSplitCloud"/>
         </Root>
 
         <Logger name="org.eclipse.jetty" level="INFO" />

airbyte-config/models/build.gradle

@@ -1,12 +1,16 @@
 import org.jsonschema2pojo.SourceType
 
 plugins {
+    id 'airbyte-docker'
+    id 'airbyte-integration-test-java'
     id "com.github.eirnym.js2p" version "1.0"
 }
 
 dependencies {
     implementation project(':airbyte-json-validation')
     implementation project(':airbyte-protocol:models')
+
+    integrationTestJavaImplementation project(':airbyte-config:models')
 }
 
 jsonSchema2Pojo {

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

@@ -65,6 +65,15 @@ public interface Configs {
 
   Set<Integer> getTemporalWorkerPorts();
 
+  // The following methods retrieve logging related information.
+  String getS3LogBucket();
+
+  String getS3LogBucketRegion();
+
+  String getAwsAccessKey();
+
+  String getAwsSecretAccessKey();
+
   enum TrackingStrategy {
     SEGMENT,
     LOGGING

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

@@ -25,6 +25,7 @@
 package io.airbyte.config;
 
 import com.google.common.base.Preconditions;
+import io.airbyte.config.helpers.LogHelpers;
 import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.Optional;
@@ -177,6 +178,26 @@ public Set<Integer> getTemporalWorkerPorts() {
         .collect(Collectors.toSet());
   }
 
+  @Override
+  public String getS3LogBucket() {
+    return getEnsureEnv(LogHelpers.S3_LOG_BUCKET);
+  }
+
+  @Override
+  public String getS3LogBucketRegion() {
+    return getEnsureEnv(LogHelpers.S3_LOG_BUCKET_REGION);
+  }
+
+  @Override
+  public String getAwsAccessKey() {
+    return getEnsureEnv(LogHelpers.AWS_ACCESS_KEY_ID);
+  }
+
+  @Override
+  public String getAwsSecretAccessKey() {
+    return getEnsureEnv(LogHelpers.AWS_SECRET_ACCESS_KEY);
+  }
+
   private String getEnvOrDefault(String key, String defaultValue) {
     return getEnvOrDefault(key, defaultValue, Function.identity());
   }

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

@@ -0,0 +1,50 @@
+/*
+ * 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.config.helpers;
+
+import io.airbyte.config.Configs;
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Interface for various Cloud Storage clients supporting Cloud log retrieval.
+ *
+ * The underlying assumption 1) each file at the path is part of the entire log file represented by that path 2) log files names start with timestamps,
+ * making it possible extract the time the file was written from it's name.
+ */
+public interface CloudLogs {
+
+  /**
+   * Retrieve all objects at the given path in lexicographical order, and return their contents as one file.
+   */
+  File downloadCloudLog(Configs configs, String logPath) throws IOException;
+
+  /**
+   * Assume all the lexicographically ordered objects at the given path form one giant log file, return the last numLines lines.
+   */
+  List<String> tailCloudLog(Configs configs, String logPath, int numLines) throws IOException;
+
+}

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

@@ -24,15 +24,33 @@
 
 package io.airbyte.config.helpers;
 
+import io.airbyte.commons.io.IOs;
 import io.airbyte.config.Configs;
+import io.airbyte.config.Configs.WorkerEnvironment;
 import java.io.File;
+import java.io.IOException;
 import java.nio.file.Path;
+import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class LogHelpers {
 
-  // if you update these values, you must also update log4j2.xml
+  private static final Logger LOGGER = LoggerFactory.getLogger(LogHelpers.class);
+
+  private static final int LOG_TAIL_SIZE = 1000000;
+  private static final CloudLogs S3 = new S3Logs();
+
+  // Any changes to the following values must also be propagated to the log4j2.xml in main/resources.
   public static String WORKSPACE_MDC_KEY = "workspace_app_root";
   public static String LOG_FILENAME = "logs.log";
+  public static String S3_LOG_BUCKET = "S3_LOG_BUCKET";
+  public static String S3_LOG_BUCKET_REGION = "S3_LOG_BUCKET_REGION";
+  public static String AWS_ACCESS_KEY_ID = "AWS_ACCESS_KEY_ID";
+  public static String AWS_SECRET_ACCESS_KEY = "AWS_SECRET_ACCESS_KEY";
+
+  public static String APP_LOGGING_CLOUD_PREFIX = "app-logging";
+  public static String JOB_LOGGING_CLOUD_PREFIX = "job-logging";
 
   public static Path getServerLogsRoot(Configs configs) {
     return configs.getWorkspaceRoot().resolve("server/logs");
@@ -43,11 +61,42 @@ public static Path getSchedulerLogsRoot(Configs configs) {
   }
 
   public static File getServerLogFile(Configs configs) {
-    return getServerLogsRoot(configs).resolve(LOG_FILENAME).toFile();
+    var logPathBase = getServerLogsRoot(configs);
+
+    if (configs.getWorkerEnvironment().equals(WorkerEnvironment.DOCKER)) {
+      return logPathBase.resolve(LOG_FILENAME).toFile();
+    }
+
+    var cloudLogPath = APP_LOGGING_CLOUD_PREFIX + logPathBase;
+    try {
+      return S3.downloadCloudLog(configs, cloudLogPath);
+    } catch (IOException e) {
+      throw new RuntimeException("Error retrieving log file: " + cloudLogPath + " from S3", e);
+    }
   }
 
   public static File getSchedulerLogFile(Configs configs) {
-    return getSchedulerLogsRoot(configs).resolve(LOG_FILENAME).toFile();
+    var logPathBase = getSchedulerLogsRoot(configs);
+
+    if (configs.getWorkerEnvironment().equals(WorkerEnvironment.DOCKER)) {
+      return logPathBase.resolve(LOG_FILENAME).toFile();
+    }
+
+    var cloudLogPath = APP_LOGGING_CLOUD_PREFIX + logPathBase;
+    try {
+      return S3.downloadCloudLog(configs, cloudLogPath);
+    } catch (IOException e) {
+      throw new RuntimeException("Error retrieving log file: " + cloudLogPath + " from S3", e);
+    }
+  }
+
+  public static List<String> getJobLogFile(Configs configs, Path logPath) throws IOException {
+    if (configs.getWorkerEnvironment().equals(WorkerEnvironment.DOCKER)) {
+      return IOs.getTail(LOG_TAIL_SIZE, logPath);
+    }
+
+    var cloudLogPath = JOB_LOGGING_CLOUD_PREFIX + logPath;
+    return S3.tailCloudLog(configs, cloudLogPath, LOG_TAIL_SIZE);
   }
 
 }