hashgraph · steven-sheehy · Jul 20, 2021 · Jul 19, 2021
diff --git a/.github/workflows/charts.yml b/.github/workflows/charts.yml
@@ -24,24 +24,30 @@ jobs:
 
   install:
     runs-on: ubuntu-latest
+    env:
+      IMAGE_TAG: ci-${{ github.sha }}
     steps:
       - name: Checkout
         uses: actions/checkout@v2
         with:
           fetch-depth: 0
 
       - name: Build images
-        run: ./mvnw ${MAVEN_CLI_OPTS} install -pl hedera-mirror-grpc,hedera-mirror-importer,hedera-mirror-rest --also-make -Dskip.npm -DskipTests -Ddocker.tag.version=main
+        run: ./mvnw ${MAVEN_CLI_OPTS} install -pl hedera-mirror-grpc,hedera-mirror-importer,hedera-mirror-rest --also-make -Dskip.npm -DskipTests -Ddocker.tag.version=${IMAGE_TAG}
 
       - name: Create k3d cluster
         run: |
           set -ex
           curl -s https://raw.githubusercontent.com/rancher/k3d/main/install.sh | bash
           k3d cluster create mirror --agents 1 --timeout 5m
-          k3d image import -c mirror "gcr.io/mirrornode/hedera-mirror-grpc:main" "gcr.io/mirrornode/hedera-mirror-importer:main" "gcr.io/mirrornode/hedera-mirror-rest:main"
+          k3d image import -c mirror "gcr.io/mirrornode/hedera-mirror-grpc:${IMAGE_TAG}"
+          k3d image import -c mirror "gcr.io/mirrornode/hedera-mirror-importer:${IMAGE_TAG}"
+          k3d image import -c mirror "gcr.io/mirrornode/hedera-mirror-rest:${IMAGE_TAG}"
 
       - name: Install ct
         uses: helm/chart-testing-action@v2.0.1
 
       - name: Install chart
-        run: ct install --config .github/ct.yaml --charts=charts/hedera-mirror
+        run: |
+          echo -e "global:\n  image:\n    tag: ${IMAGE_TAG}" | tee -a charts/hedera-mirror/ci/*
+          ct install --config .github/ct.yaml --charts=charts/hedera-mirror
diff --git a/...r-importer/src/main/java/com/hedera/mirror/importer/config/CloudStorageConfiguration.java b/...r-importer/src/main/java/com/hedera/mirror/importer/config/CloudStorageConfiguration.java
@@ -0,0 +1,119 @@
+package com.hedera.mirror.importer.config;
+
+/*-
+ * ‌
+ * Hedera Mirror Node
+ * 
+ * Copyright (C) 2019 - 2021 Hedera Hashgraph, LLC
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ‍
+ */
+
+import java.net.URI;
+import java.time.Duration;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.interceptor.Context;
+import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
+import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
+import software.amazon.awssdk.http.SdkHttpRequest;
+import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
+import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.S3AsyncClientBuilder;
+
+import com.hedera.mirror.importer.downloader.CommonDownloaderProperties;
+
+@Configuration
+@Log4j2
+@RequiredArgsConstructor
+public class CloudStorageConfiguration {
+
+    private final CommonDownloaderProperties downloaderProperties;
+    private final MetricsExecutionInterceptor metricsExecutionInterceptor;
+
+    @Bean
+    AwsCredentialsProvider awsCredentialsProvider() {
+        if (downloaderProperties.isAnonymousCredentials()) {
+            log.info("Setting up S3 async client using anonymous credentials");
+            return AnonymousCredentialsProvider.create();
+        } else if (downloaderProperties.isStaticCredentials()) {
+            log.info("Setting up S3 async client using provided access/secret key");
+            return StaticCredentialsProvider.create(AwsBasicCredentials.create(downloaderProperties.getAccessKey(),
+                    downloaderProperties.getSecretKey()));
+        }
+        log.info("Setting up S3 async client using AWS Default Credentials Provider");
+        return DefaultCredentialsProvider.create();
+    }
+
+    @Bean
+    @ConditionalOnProperty(prefix = "hedera.mirror.importer.downloader", name = "cloudProvider", havingValue = "GCP")
+    S3AsyncClient gcpCloudStorageClient() {
+        log.info("Configured to download from GCP with bucket name '{}'", downloaderProperties.getBucketName());
+        // Any valid region for aws client. Ignored by GCP.
+        S3AsyncClientBuilder clientBuilder = asyncClientBuilder("us-east-1")
+                .endpointOverride(URI.create(downloaderProperties.getCloudProvider().getEndpoint()));
+        String projectId = downloaderProperties.getGcpProjectId();
+        if (StringUtils.isNotBlank(projectId)) {
+            clientBuilder.overrideConfiguration(builder -> builder.addExecutionInterceptor(new ExecutionInterceptor() {
+                @Override
+                public SdkHttpRequest modifyHttpRequest(
+                        Context.ModifyHttpRequest context, ExecutionAttributes executionAttributes) {
+                    return context.httpRequest().toBuilder()
+                            .appendRawQueryParameter("userProject", projectId).build();
+                }
+            }));
+        }
+        return clientBuilder.build();
+    }
+
+    @Bean
+    @ConditionalOnProperty(prefix = "hedera.mirror.importer.downloader", name = "cloudProvider", havingValue = "S3",
+            matchIfMissing = true)
+    public S3AsyncClient s3CloudStorageClient() {
+        log.info("Configured to download from S3 in region {} with bucket name '{}'",
+                downloaderProperties.getRegion(), downloaderProperties.getBucketName());
+        S3AsyncClientBuilder clientBuilder = asyncClientBuilder(
+                downloaderProperties.getRegion());
+        String endpointOverride = downloaderProperties.getEndpointOverride();
+        if (endpointOverride != null) {
+            log.info("Overriding s3 client endpoint to {}", endpointOverride);
+            clientBuilder.endpointOverride(URI.create(endpointOverride));
+        }
+        return clientBuilder.build();
+    }
+
+    private S3AsyncClientBuilder asyncClientBuilder(String region) {
+        SdkAsyncHttpClient httpClient = NettyNioAsyncHttpClient.builder()
+                .maxConcurrency(downloaderProperties.getMaxConcurrency())
+                .connectionMaxIdleTime(Duration.ofSeconds(5))  // https://github.com/aws/aws-sdk-java-v2/issues/1122
+                .build();
+
+        return S3AsyncClient.builder()
+                .credentialsProvider(awsCredentialsProvider())
+                .region(Region.of(region))
+                .httpClient(httpClient)
+                .overrideConfiguration(c -> c.addExecutionInterceptor(metricsExecutionInterceptor));
+    }
+}
diff --git a/...ter/src/main/java/com/hedera/mirror/importer/config/CredentialsProviderConfiguration.java b/...ter/src/main/java/com/hedera/mirror/importer/config/CredentialsProviderConfiguration.java
diff --git a/...or-importer/src/main/java/com/hedera/mirror/importer/config/HealthCheckConfiguration.java b/...or-importer/src/main/java/com/hedera/mirror/importer/config/HealthCheckConfiguration.java
@@ -32,19 +32,22 @@
 import org.springframework.context.annotation.Configuration;
 
 import com.hedera.mirror.importer.MirrorProperties;
+import com.hedera.mirror.importer.leader.LeaderService;
 import com.hedera.mirror.importer.parser.ParserProperties;
 
 @Configuration
 @RequiredArgsConstructor
 public class HealthCheckConfiguration {
+
+    private final LeaderService leaderService;
     private final MirrorProperties mirrorProperties;
+    private final Collection<ParserProperties> parserProperties;
 
     @Bean
-    CompositeHealthContributor streamFileActivity(@Named("prometheusMeterRegistry") MeterRegistry meterRegistry,
-                                                  Collection<ParserProperties> parserProperties) {
+    CompositeHealthContributor streamFileActivity(@Named("prometheusMeterRegistry") MeterRegistry meterRegistry) {
         Map<String, HealthIndicator> healthIndicators = parserProperties.stream().collect(Collectors.toMap(
                 k -> k.getStreamType().toString(),
-                v -> new StreamFileHealthIndicator(v, meterRegistry, mirrorProperties)));
+                v -> new StreamFileHealthIndicator(leaderService, meterRegistry, mirrorProperties, v)));
 
         return CompositeHealthContributor.fromMap(healthIndicators);
     }

diff --git a/...importer/src/main/java/com/hedera/mirror/importer/config/MirrorImporterConfiguration.java b/...importer/src/main/java/com/hedera/mirror/importer/config/MirrorImporterConfiguration.java
@@ -20,12 +20,9 @@
  * ‍
  */
 
-import java.net.URI;
-import java.time.Duration;
 import javax.annotation.PostConstruct;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
-import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
@@ -39,20 +36,10 @@
 import org.springframework.retry.annotation.EnableRetry;
 import org.springframework.scheduling.annotation.EnableAsync;
 import org.springframework.scheduling.annotation.EnableScheduling;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.core.interceptor.Context;
-import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
-import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
-import software.amazon.awssdk.http.SdkHttpRequest;
-import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
-import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
-import software.amazon.awssdk.regions.Region;
-import software.amazon.awssdk.services.s3.S3AsyncClient;
-import software.amazon.awssdk.services.s3.S3AsyncClientBuilder;
 
 import com.hedera.mirror.importer.MirrorProperties;
-import com.hedera.mirror.importer.downloader.CommonDownloaderProperties;
 import com.hedera.mirror.importer.leader.LeaderAspect;
+import com.hedera.mirror.importer.leader.LeaderService;
 
 @Configuration
 @EnableAsync
@@ -62,9 +49,6 @@
 public class MirrorImporterConfiguration {
 
     private final MirrorProperties mirrorProperties;
-    private final CommonDownloaderProperties downloaderProperties;
-    private final MetricsExecutionInterceptor metricsExecutionInterceptor;
-    private final AwsCredentialsProvider awsCredentialsProvider;
 
     @Autowired(required = false)
     @Qualifier("webServerStartStop")
@@ -80,58 +64,14 @@ void init() {
 
     @Bean
     @Profile("kubernetes")
-    LeaderAspect leaderAspect() {
+    LeaderService leaderAspect() {
         return new LeaderAspect();
     }
 
     @Bean
-    @ConditionalOnProperty(prefix = "hedera.mirror.importer.downloader", name = "cloudProvider", havingValue = "GCP")
-    public S3AsyncClient gcpCloudStorageClient() {
-        log.info("Configured to download from GCP with bucket name '{}'", downloaderProperties.getBucketName());
-        // Any valid region for aws client. Ignored by GCP.
-        S3AsyncClientBuilder clientBuilder = asyncClientBuilder("us-east-1")
-                .endpointOverride(URI.create(downloaderProperties.getCloudProvider().getEndpoint()));
-        String projectId = downloaderProperties.getGcpProjectId();
-        if (StringUtils.isNotBlank(projectId)) {
-            clientBuilder.overrideConfiguration(builder -> builder.addExecutionInterceptor(new ExecutionInterceptor() {
-                @Override
-                public SdkHttpRequest modifyHttpRequest(
-                        Context.ModifyHttpRequest context, ExecutionAttributes executionAttributes) {
-                    return context.httpRequest().toBuilder()
-                            .appendRawQueryParameter("userProject", projectId).build();
-                }
-            }));
-        }
-        return clientBuilder.build();
-    }
-
-    @Bean
-    @ConditionalOnProperty(prefix = "hedera.mirror.importer.downloader", name = "cloudProvider", havingValue = "S3",
-            matchIfMissing = true)
-    public S3AsyncClient s3CloudStorageClient() {
-        log.info("Configured to download from S3 in region {} with bucket name '{}'",
-                downloaderProperties.getRegion(), downloaderProperties.getBucketName());
-        S3AsyncClientBuilder clientBuilder = asyncClientBuilder(
-                downloaderProperties.getRegion());
-        String endpointOverride = downloaderProperties.getEndpointOverride();
-        if (endpointOverride != null) {
-            log.info("Overriding s3 client endpoint to {}", endpointOverride);
-            clientBuilder.endpointOverride(URI.create(endpointOverride));
-        }
-        return clientBuilder.build();
-    }
-
-    private S3AsyncClientBuilder asyncClientBuilder(String region) {
-        SdkAsyncHttpClient httpClient = NettyNioAsyncHttpClient.builder()
-                .maxConcurrency(downloaderProperties.getMaxConcurrency())
-                .connectionMaxIdleTime(Duration.ofSeconds(5))  // https://github.com/aws/aws-sdk-java-v2/issues/1122
-                .build();
-
-        return S3AsyncClient.builder()
-                .credentialsProvider(awsCredentialsProvider)
-                .region(Region.of(region))
-                .httpClient(httpClient)
-                .overrideConfiguration(c -> c.addExecutionInterceptor(metricsExecutionInterceptor));
+    @Profile("!kubernetes")
+    LeaderService leaderService() {
+        return Boolean.TRUE::booleanValue; // Leader election not available outside Kubernetes
     }
 
     @Bean

diff --git a/...r-importer/src/main/java/com/hedera/mirror/importer/config/StreamFileHealthIndicator.java b/...r-importer/src/main/java/com/hedera/mirror/importer/config/StreamFileHealthIndicator.java
@@ -36,6 +36,7 @@
 import org.springframework.boot.actuate.health.Status;
 
 import com.hedera.mirror.importer.MirrorProperties;
+import com.hedera.mirror.importer.leader.LeaderService;
 import com.hedera.mirror.importer.parser.ParserProperties;
 
 @RequiredArgsConstructor
@@ -54,16 +55,19 @@ public class StreamFileHealthIndicator implements HealthIndicator {
             getHealthWithReason(Status.UNKNOWN, STREAM_CLOSE_LATENCY_METRIC_NAME + MISSING_TIMER_REASON);
     private static final Health endDateInPastHealth =
             getHealthWithReason(Status.UP, "EndDate has passed, stream files are no longer expected");
+    private static final Health notLeaderHealth = Health.unknown().withDetail(REASON_KEY, "Not currently leader")
+            .build();
     private final AtomicReference<Health> lastHealthStatus = new AtomicReference<>(Health
             .unknown()
             .withDetail(COUNT_KEY, 0L)
             .withDetail(LAST_CHECK_KEY, Instant.now())
             .withDetail(REASON_KEY, "Starting up, no files parsed yet")
             .build()); // unknown until at least 1 stream file is parsed
 
-    private final ParserProperties parserProperty;
+    private final LeaderService leaderService;
     private final MeterRegistry meterRegistry;
     private final MirrorProperties mirrorProperties;
+    private final ParserProperties parserProperty;
 
     @Override
     public Health health() {
@@ -79,6 +83,10 @@ public Health health() {
             return endDateInPastHealth;
         }
 
+        if (!leaderService.isLeader()) {
+            return notLeaderHealth;
+        }
+
         Timer streamParseDurationTimer = meterRegistry
                 .find(STREAM_PARSE_DURATION_METRIC_NAME)
                 .tags(Tags.of(