🐛 Airbyte Kube will now error on start up if incorrect logs configuration is passed in. Fix scheduler counting bug. (#6188)

davinchia · web-flow · commit 51fadf2b9b74 · 2021-09-18T10:21:04.000+08:00
* Move the cloud client create call so this errors out on app start up instead of when they the log file is retrieved.

* Add tests.

* Fix code that was never being called.
diff --git a/airbyte-config/models/src/main/java/io/airbyte/config/helpers/LogClientSingleton.java b/airbyte-config/models/src/main/java/io/airbyte/config/helpers/LogClientSingleton.java
@@ -91,7 +91,6 @@ public static File getServerLogFile(Configs configs) {
     }
 
     var logConfigs = new LogConfigDelegator(configs);
-    createCloudClientIfNull(logConfigs);
     var cloudLogPath = APP_LOGGING_CLOUD_PREFIX + logPathBase;
     try {
       return logClient.downloadCloudLog(logConfigs, cloudLogPath);
@@ -107,7 +106,6 @@ public static File getSchedulerLogFile(Configs configs) {
     }
 
     var logConfigs = new LogConfigDelegator(configs);
-    createCloudClientIfNull(logConfigs);
     var cloudLogPath = APP_LOGGING_CLOUD_PREFIX + logPathBase;
     try {
       return logClient.downloadCloudLog(logConfigs, cloudLogPath);
@@ -122,7 +120,6 @@ public static List<String> getJobLogFile(Configs configs, Path logPath) throws I
     }
 
     var logConfigs = new LogConfigDelegator(configs);
-    createCloudClientIfNull(logConfigs);
     var cloudLogPath = JOB_LOGGING_CLOUD_PREFIX + logPath;
     return logClient.tailCloudLog(logConfigs, cloudLogPath, LOG_TAIL_SIZE);
   }
@@ -136,33 +133,38 @@ public static void deleteLogs(Configs configs, String logPath) {
       throw new NotImplementedException("Local log deletes not supported.");
     }
     var logConfigs = new LogConfigDelegator(configs);
-    createCloudClientIfNull(logConfigs);
     var cloudLogPath = JOB_LOGGING_CLOUD_PREFIX + logPath;
     logClient.deleteLogs(logConfigs, cloudLogPath);
   }
 
   public static void setJobMdc(Path path) {
-    if (shouldUseLocalLogs(new EnvConfigs())) {
+    var configs = new EnvConfigs();
+    if (shouldUseLocalLogs(configs)) {
       LOGGER.debug("Setting docker job mdc");
       MDC.put(LogClientSingleton.JOB_LOG_PATH_MDC_KEY, path.resolve(LogClientSingleton.LOG_FILENAME).toString());
     } else {
       LOGGER.debug("Setting kube job mdc");
+      var logConfigs = new LogConfigDelegator(configs);
+      createCloudClientIfNull(logConfigs);
       MDC.put(LogClientSingleton.CLOUD_JOB_LOG_PATH_MDC_KEY, path.resolve(LogClientSingleton.LOG_FILENAME).toString());
     }
   }
 
   public static void setWorkspaceMdc(Path path) {
-    if (shouldUseLocalLogs(new EnvConfigs())) {
+    var configs = new EnvConfigs();
+    if (shouldUseLocalLogs(configs)) {
       LOGGER.debug("Setting docker workspace mdc");
       MDC.put(LogClientSingleton.WORKSPACE_MDC_KEY, path.toString());
     } else {
       LOGGER.debug("Setting kube workspace mdc");
+      var logConfigs = new LogConfigDelegator(configs);
+      createCloudClientIfNull(logConfigs);
       MDC.put(LogClientSingleton.CLOUD_WORKSPACE_MDC_KEY, path.toString());
     }
   }
 
   private static boolean shouldUseLocalLogs(Configs configs) {
-    return configs.getWorkerEnvironment().equals(WorkerEnvironment.DOCKER) || CloudLogs.hasEmptyConfigs(new LogConfigDelegator(configs));
+    return configs.getWorkerEnvironment().equals(WorkerEnvironment.DOCKER);
   }
 
   private static void createCloudClientIfNull(LogConfigs configs) {
diff --git a/airbyte-config/models/src/test/java/io/airbyte/config/helpers/CloudLogsClientTest.java b/airbyte-config/models/src/test/java/io/airbyte/config/helpers/CloudLogsClientTest.java
@@ -25,11 +25,59 @@
 package io.airbyte.config.helpers;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
+import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 
-public class CloudLogsTest {
+public class CloudLogsClientTest {
+
+  @Nested
+  class CloudLogClientMissingConfiguration {
+
+    @Test
+    public void testMinio() {
+      var configs = Mockito.mock(LogConfigs.class);
+      // Mising bucket.
+      Mockito.when(configs.getS3MinioEndpoint()).thenReturn("minio-endpoint");
+      Mockito.when(configs.getAwsAccessKey()).thenReturn("access-key");
+      Mockito.when(configs.getAwsSecretAccessKey()).thenReturn("access-key-secret");
+      Mockito.when(configs.getS3LogBucket()).thenReturn("");
+      Mockito.when(configs.getS3LogBucketRegion()).thenReturn("");
+
+      assertThrows(RuntimeException.class, () -> CloudLogs.createCloudLogClient(configs));
+    }
+
+    @Test
+    public void testAws() {
+      var configs = Mockito.mock(LogConfigs.class);
+      // Missing bucket and access key.
+      Mockito.when(configs.getS3MinioEndpoint()).thenReturn("");
+      Mockito.when(configs.getAwsAccessKey()).thenReturn("");
+      Mockito.when(configs.getAwsSecretAccessKey()).thenReturn("access-key-secret");
+      Mockito.when(configs.getS3LogBucket()).thenReturn("");
+      Mockito.when(configs.getS3LogBucketRegion()).thenReturn("");
+
+      assertThrows(RuntimeException.class, () -> CloudLogs.createCloudLogClient(configs));
+    }
+
+    @Test
+    public void testGcs() {
+      var configs = Mockito.mock(LogConfigs.class);
+      Mockito.when(configs.getAwsAccessKey()).thenReturn("");
+      Mockito.when(configs.getAwsSecretAccessKey()).thenReturn("");
+      Mockito.when(configs.getS3LogBucket()).thenReturn("");
+      Mockito.when(configs.getS3LogBucketRegion()).thenReturn("");
+
+      // Missing bucket.
+      Mockito.when(configs.getGcpStorageBucket()).thenReturn("");
+      Mockito.when(configs.getGoogleApplicationCredentials()).thenReturn("path/to/google/secret");
+
+      assertThrows(RuntimeException.class, () -> CloudLogs.createCloudLogClient(configs));
+    }
+
+  }
 
   @Test
   public void createCloudLogClientTestMinio() {
diff --git a/airbyte-scheduler/app/src/main/java/io/airbyte/scheduler/app/JobScheduler.java b/airbyte-scheduler/app/src/main/java/io/airbyte/scheduler/app/JobScheduler.java
@@ -40,7 +40,6 @@
 import java.time.Instant;
 import java.util.List;
 import java.util.Optional;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiPredicate;
 import java.util.stream.Collectors;
 import org.slf4j.Logger;
@@ -92,19 +91,20 @@ public void run() {
   }
 
   private void scheduleSyncJobs() throws IOException {
-    final AtomicInteger jobsScheduled = new AtomicInteger();
+    int jobsScheduled = 0;
     final List<StandardSync> activeConnections = getAllActiveConnections();
 
     for (StandardSync connection : activeConnections) {
       final Optional<Job> previousJobOptional = jobPersistence.getLastReplicationJob(connection.getConnectionId());
 
       if (scheduleJobPredicate.test(previousJobOptional, connection)) {
         jobFactory.create(connection.getConnectionId());
+        jobsScheduled++;
       }
     }
-    int jobsScheduledCount = jobsScheduled.get();
-    if (jobsScheduledCount > 0) {
-      LOGGER.info("Job-Scheduler Summary. Active connections: {}, Jobs scheduler: {}", activeConnections.size(), jobsScheduled.get());
+
+    if (jobsScheduled > 0) {
+      LOGGER.info("Job-Scheduler Summary. Active connections: {}, Jobs scheduled this cycle: {}", activeConnections.size(), jobsScheduled);
     }
   }
 
diff --git a/build.gradle b/build.gradle
@@ -71,7 +71,7 @@ def createSpotlessTarget = { pattern ->
         'normalization_test_output',
         'tools',
         'secrets',
-        'charts'
+        'charts' // Helm charts often have injected template strings that will fail general linting. Helm linting is done separately.
     ]
 
     if (System.getenv().containsKey("SUB_BUILD")) {

Original file line number	Diff line number	Diff line change
`@@ -91,7 +91,6 @@ public static File getServerLogFile(Configs configs) {`
`91`	`91`	`}`
`92`	`92`
`93`	`93`	`var logConfigs = new LogConfigDelegator(configs);`
`94`		`- createCloudClientIfNull(logConfigs);`
`95`	`94`	`var cloudLogPath = APP_LOGGING_CLOUD_PREFIX + logPathBase;`
`96`	`95`	`try {`
`97`	`96`	`return logClient.downloadCloudLog(logConfigs, cloudLogPath);`
`@@ -107,7 +106,6 @@ public static File getSchedulerLogFile(Configs configs) {`
`107`	`106`	`}`
`108`	`107`
`109`	`108`	`var logConfigs = new LogConfigDelegator(configs);`
`110`		`- createCloudClientIfNull(logConfigs);`
`111`	`109`	`var cloudLogPath = APP_LOGGING_CLOUD_PREFIX + logPathBase;`
`112`	`110`	`try {`
`113`	`111`	`return logClient.downloadCloudLog(logConfigs, cloudLogPath);`
`@@ -122,7 +120,6 @@ public static List<String> getJobLogFile(Configs configs, Path logPath) throws I`
`122`	`120`	`}`
`123`	`121`
`124`	`122`	`var logConfigs = new LogConfigDelegator(configs);`
`125`		`- createCloudClientIfNull(logConfigs);`
`126`	`123`	`var cloudLogPath = JOB_LOGGING_CLOUD_PREFIX + logPath;`
`127`	`124`	`return logClient.tailCloudLog(logConfigs, cloudLogPath, LOG_TAIL_SIZE);`
`128`	`125`	`}`
`@@ -136,33 +133,38 @@ public static void deleteLogs(Configs configs, String logPath) {`
`136`	`133`	`throw new NotImplementedException("Local log deletes not supported.");`
`137`	`134`	`}`
`138`	`135`	`var logConfigs = new LogConfigDelegator(configs);`
`139`		`- createCloudClientIfNull(logConfigs);`
`140`	`136`	`var cloudLogPath = JOB_LOGGING_CLOUD_PREFIX + logPath;`
`141`	`137`	`logClient.deleteLogs(logConfigs, cloudLogPath);`
`142`	`138`	`}`
`143`	`139`
`144`	`140`	`public static void setJobMdc(Path path) {`
`145`		`- if (shouldUseLocalLogs(new EnvConfigs())) {`
	`141`	`+ var configs = new EnvConfigs();`
	`142`	`+ if (shouldUseLocalLogs(configs)) {`
`146`	`143`	`LOGGER.debug("Setting docker job mdc");`
`147`	`144`	`MDC.put(LogClientSingleton.JOB_LOG_PATH_MDC_KEY, path.resolve(LogClientSingleton.LOG_FILENAME).toString());`
`148`	`145`	`} else {`
`149`	`146`	`LOGGER.debug("Setting kube job mdc");`
	`147`	`+ var logConfigs = new LogConfigDelegator(configs);`
	`148`	`+ createCloudClientIfNull(logConfigs);`
`150`	`149`	`MDC.put(LogClientSingleton.CLOUD_JOB_LOG_PATH_MDC_KEY, path.resolve(LogClientSingleton.LOG_FILENAME).toString());`
`151`	`150`	`}`
`152`	`151`	`}`
`153`	`152`
`154`	`153`	`public static void setWorkspaceMdc(Path path) {`
`155`		`- if (shouldUseLocalLogs(new EnvConfigs())) {`
	`154`	`+ var configs = new EnvConfigs();`
	`155`	`+ if (shouldUseLocalLogs(configs)) {`
`156`	`156`	`LOGGER.debug("Setting docker workspace mdc");`
`157`	`157`	`MDC.put(LogClientSingleton.WORKSPACE_MDC_KEY, path.toString());`
`158`	`158`	`} else {`
`159`	`159`	`LOGGER.debug("Setting kube workspace mdc");`
	`160`	`+ var logConfigs = new LogConfigDelegator(configs);`
	`161`	`+ createCloudClientIfNull(logConfigs);`
`160`	`162`	`MDC.put(LogClientSingleton.CLOUD_WORKSPACE_MDC_KEY, path.toString());`
`161`	`163`	`}`
`162`	`164`	`}`
`163`	`165`
`164`	`166`	`private static boolean shouldUseLocalLogs(Configs configs) {`
`165`		`- return configs.getWorkerEnvironment().equals(WorkerEnvironment.DOCKER) \|\| CloudLogs.hasEmptyConfigs(new LogConfigDelegator(configs));`
	`167`	`+ return configs.getWorkerEnvironment().equals(WorkerEnvironment.DOCKER);`
`166`	`168`	`}`
`167`	`169`
`168`	`170`	`private static void createCloudClientIfNull(LogConfigs configs) {`
Original file line number	Diff line number	Diff line change
`@@ -40,7 +40,6 @@`
`40`	`40`	`import java.time.Instant;`
`41`	`41`	`import java.util.List;`
`42`	`42`	`import java.util.Optional;`
`43`		`-import java.util.concurrent.atomic.AtomicInteger;`
`44`	`43`	`import java.util.function.BiPredicate;`
`45`	`44`	`import java.util.stream.Collectors;`
`46`	`45`	`import org.slf4j.Logger;`
`@@ -92,19 +91,20 @@ public void run() {`
`92`	`91`	`}`
`93`	`92`
`94`	`93`	`private void scheduleSyncJobs() throws IOException {`
`95`		`- final AtomicInteger jobsScheduled = new AtomicInteger();`
	`94`	`+ int jobsScheduled = 0;`
`96`	`95`	`final List<StandardSync> activeConnections = getAllActiveConnections();`
`97`	`96`
`98`	`97`	`for (StandardSync connection : activeConnections) {`
`99`	`98`	`final Optional<Job> previousJobOptional = jobPersistence.getLastReplicationJob(connection.getConnectionId());`
`100`	`99`
`101`	`100`	`if (scheduleJobPredicate.test(previousJobOptional, connection)) {`
`102`	`101`	`jobFactory.create(connection.getConnectionId());`
	`102`	`+ jobsScheduled++;`
`103`	`103`	`}`
`104`	`104`	`}`
`105`		`- int jobsScheduledCount = jobsScheduled.get();`
`106`		`- if (jobsScheduledCount > 0) {`
`107`		`- LOGGER.info("Job-Scheduler Summary. Active connections: {}, Jobs scheduler: {}", activeConnections.size(), jobsScheduled.get());`
	`105`	`+`
	`106`	`+ if (jobsScheduled > 0) {`
	`107`	`+ LOGGER.info("Job-Scheduler Summary. Active connections: {}, Jobs scheduled this cycle: {}", activeConnections.size(), jobsScheduled);`
`108`	`108`	`}`
`109`	`109`	`}`
`110`	`110`
Original file line number	Diff line number	Diff line change
`@@ -71,7 +71,7 @@ def createSpotlessTarget = { pattern ->`
`71`	`71`	`'normalization_test_output',`
`72`	`72`	`'tools',`
`73`	`73`	`'secrets',`
`74`		`- 'charts'`
	`74`	`+ 'charts' // Helm charts often have injected template strings that will fail general linting. Helm linting is done separately.`
`75`	`75`	`]`
`76`	`76`
`77`	`77`	`if (System.getenv().containsKey("SUB_BUILD")) {`