remove appconfig in favor of S3ObjectMonitor

service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerConfiguration.java

@@ -27,7 +27,6 @@ import org.whispersystems.textsecuregcm.configuration.DatadogConfiguration;
 import org.whispersystems.textsecuregcm.configuration.DefaultAwsCredentialsFactory;
 import org.whispersystems.textsecuregcm.configuration.DirectoryV2Configuration;
 import org.whispersystems.textsecuregcm.configuration.DogstatsdConfiguration;
-import org.whispersystems.textsecuregcm.configuration.DynamicConfigurationManagerFactory;
 import org.whispersystems.textsecuregcm.configuration.DynamoDbClientFactory;
 import org.whispersystems.textsecuregcm.configuration.DynamoDbTables;
 import org.whispersystems.textsecuregcm.configuration.ExternalRequestFilterConfiguration;
@@ -244,7 +243,7 @@ public class WhisperServerConfiguration extends Configuration {
   @Valid
   @NotNull
   @JsonProperty
-  private DynamicConfigurationManagerFactory appConfig;
+  private S3ObjectMonitorFactory dynamicConfig;
 
   @Valid
   @NotNull
@@ -487,8 +486,8 @@ public class WhisperServerConfiguration extends Configuration {
     return remoteConfig;
   }
 
-  public DynamicConfigurationManagerFactory getAppConfig() {
-    return appConfig;
+  public S3ObjectMonitorFactory getDynamicConfig() {
+    return dynamicConfig;
   }
 
   public BadgesConfiguration getBadges() {

service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java

@@ -351,8 +351,9 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     ScheduledExecutorService dynamicConfigurationExecutor = environment.lifecycle()
         .scheduledExecutorService(name(getClass(), "dynamicConfiguration-%d")).threads(1).build();
 
-    DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager = config.getAppConfig()
-        .build(DynamicConfiguration.class, dynamicConfigurationExecutor, awsCredentialsProvider);
+    DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager =
+        new DynamicConfigurationManager<>(
+            config.getDynamicConfig().build(awsCredentialsProvider, dynamicConfigurationExecutor), DynamicConfiguration.class);
     dynamicConfigurationManager.start();
 
     MetricsUtil.configureRegistries(config, environment, dynamicConfigurationManager);

service/src/main/java/org/whispersystems/textsecuregcm/configuration/AppConfigConfiguration.java

@@ -1,43 +0,0 @@
-package org.whispersystems.textsecuregcm.configuration;
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-import java.util.concurrent.ScheduledExecutorService;
-import javax.validation.constraints.NotEmpty;
-import com.fasterxml.jackson.annotation.JsonTypeName;
-import org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-
-@JsonTypeName("default")
-public class AppConfigConfiguration implements DynamicConfigurationManagerFactory {
-
-  @JsonProperty
-  @NotEmpty
-  private String application;
-
-  @JsonProperty
-  @NotEmpty
-  private String environment;
-
-  @JsonProperty
-  @NotEmpty
-  private String configuration;
-
-  public String getApplication() {
-    return application;
-  }
-
-  public String getEnvironment() {
-    return environment;
-  }
-
-  public String getConfigurationName() {
-    return configuration;
-  }
-
-  @Override
-  public <T> DynamicConfigurationManager<T> build(Class<T> klazz, ScheduledExecutorService scheduledExecutorService,
-      AwsCredentialsProvider awsCredentialsProvider) {
-    return new DynamicConfigurationManager<>(application, environment, configuration, awsCredentialsProvider, klazz,
-        scheduledExecutorService);
-  }
-}

service/src/main/java/org/whispersystems/textsecuregcm/configuration/DynamicConfigurationManagerFactory.java

@@ -1,19 +0,0 @@
-/*
- * Copyright 2024 Signal Messenger, LLC
- * SPDX-License-Identifier: AGPL-3.0-only
- */
-
-package org.whispersystems.textsecuregcm.configuration;
-
-import com.fasterxml.jackson.annotation.JsonTypeInfo;
-import io.dropwizard.jackson.Discoverable;
-import org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import java.util.concurrent.ScheduledExecutorService;
-
-@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type", defaultImpl = AppConfigConfiguration.class)
-public interface DynamicConfigurationManagerFactory extends Discoverable {
-
-  <T> DynamicConfigurationManager<T> build(Class<T> configurationClass,
-      ScheduledExecutorService scheduledExecutorService, AwsCredentialsProvider awsCredentialsProvider);
-}

service/src/main/java/org/whispersystems/textsecuregcm/storage/DynamicConfigurationManager.java

@@ -7,44 +7,31 @@ package org.whispersystems.textsecuregcm.storage;
 
 import static org.whispersystems.textsecuregcm.metrics.MetricsUtil.name;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import com.google.common.annotations.VisibleForTesting;
 import io.micrometer.core.instrument.Metrics;
-import java.time.Duration;
+
+import java.io.IOException;
+import java.io.InputStream;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import javax.validation.ConstraintViolation;
 import javax.validation.Validation;
 import javax.validation.Validator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.s3.S3ObjectMonitor;
 import org.whispersystems.textsecuregcm.util.SystemMapper;
-import org.whispersystems.textsecuregcm.util.Util;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
-import software.amazon.awssdk.services.appconfigdata.AppConfigDataClient;
-import software.amazon.awssdk.services.appconfigdata.model.GetLatestConfigurationRequest;
-import software.amazon.awssdk.services.appconfigdata.model.GetLatestConfigurationResponse;
-import software.amazon.awssdk.services.appconfigdata.model.StartConfigurationSessionRequest;
-import software.amazon.awssdk.services.appconfigdata.model.StartConfigurationSessionResponse;
 
 public class DynamicConfigurationManager<T> {
 
-  private final String application;
-  private final String environment;
-  private final String configurationName;
-  private final AppConfigDataClient appConfigClient;
+  private final S3ObjectMonitor configMonitor;
   private final Class<T> configurationClass;
 
   // Set on initial config fetch
   private final AtomicReference<T> configuration = new AtomicReference<>();
   private final CountDownLatch initialized = new CountDownLatch(1);
-  private final ScheduledExecutorService scheduledExecutorService;
-  private String configurationToken = null;
 
   private static final Validator VALIDATOR = Validation.buildDefaultValidatorFactory().getValidator();
 
@@ -54,28 +41,9 @@ public class DynamicConfigurationManager<T> {
 
   private static final Logger logger = LoggerFactory.getLogger(DynamicConfigurationManager.class);
 
-  public DynamicConfigurationManager(String application, String environment, String configurationName,
-      AwsCredentialsProvider awsCredentialsProvider, Class<T> configurationClass,
-      ScheduledExecutorService scheduledExecutorService) {
-    this(AppConfigDataClient
-            .builder()
-            .credentialsProvider(awsCredentialsProvider)
-            .overrideConfiguration(ClientOverrideConfiguration.builder()
-                .apiCallTimeout(Duration.ofSeconds(10))
-                .apiCallAttemptTimeout(Duration.ofSeconds(10)).build())
-            .build(),
-        application, environment, configurationName, configurationClass, scheduledExecutorService);
-  }
-
-  @VisibleForTesting
-  DynamicConfigurationManager(AppConfigDataClient appConfigClient, String application, String environment,
-      String configurationName, Class<T> configurationClass, ScheduledExecutorService scheduledExecutorService) {
-    this.appConfigClient = appConfigClient;
-    this.application = application;
-    this.environment = environment;
-    this.configurationName = configurationName;
+  public DynamicConfigurationManager(final S3ObjectMonitor configMonitor, final Class<T> configurationClass) {
+    this.configMonitor = configMonitor;
     this.configurationClass = configurationClass;
-    this.scheduledExecutorService = scheduledExecutorService;
   }
 
   public T getConfiguration() {
@@ -88,92 +56,63 @@ public class DynamicConfigurationManager<T> {
     return configuration.get();
   }
 
-  public void start() {
+  public synchronized void start() {
     if (initialized.getCount() == 0) {
       return;
     }
-    configuration.set(retrieveInitialDynamicConfiguration());
-    initialized.countDown();
 
-    scheduledExecutorService.scheduleWithFixedDelay(() -> {
+    this.configMonitor.start(this::receiveConfiguration);
+
+    // Starting an S3ObjectMonitor immediately does a blocking retrieve of the data, but it might
+    // fail to parse, in which case we wait for an update (which will happen on another thread) to
+    // give us a valid configuration before marking ourselves ready
+    while (configuration.get() == null) {
+      logger.warn("Failed to retrieve or parse initial dynamic configuration");
       try {
-        retrieveDynamicConfiguration().ifPresent(configuration::set);
-      } catch (Exception e) {
-        logger.warn("Error retrieving dynamic configuration", e);
-      }
-    }, 0, 5, TimeUnit.SECONDS);
+        this.wait();
+      } catch (InterruptedException e) {}
+    }
+    initialized.countDown();
   }
 
-  private Optional<T> retrieveDynamicConfiguration() throws JsonProcessingException {
-    if (configurationToken == null) {
-        logger.error("Invalid configuration token, will not be able to fetch configuration updates");
-    }
-    GetLatestConfigurationResponse latestConfiguration;
+  private synchronized void receiveConfiguration(InputStream configDataStream) {
+    final String configData;
     try {
-      latestConfiguration = appConfigClient.getLatestConfiguration(GetLatestConfigurationRequest.builder()
-          .configurationToken(configurationToken)
-          .build());
-      // token to use in the next fetch
-      configurationToken = latestConfiguration.nextPollConfigurationToken();
-      logger.debug("next token: {}", configurationToken);
-    } catch (final RuntimeException e) {
+      configData = new String(configDataStream.readAllBytes());
+    } catch (IOException e) {
       Metrics.counter(ERROR_COUNTER_NAME, ERROR_TYPE_TAG_NAME, "fetch").increment();
-      throw e;
+      return;
     }
 
-    if (!latestConfiguration.configuration().asByteBuffer().hasRemaining()) {
-      // empty configuration means nothing has changed
-      return Optional.empty();
-    }
-    logger.info("Received new config of length {}, next configuration token: {}",
-        latestConfiguration.configuration().asByteBuffer().remaining(),
-        configurationToken);
-
-    try {
-      return parseConfiguration(latestConfiguration.configuration().asUtf8String(), configurationClass);
-    } catch (final JsonProcessingException e) {
-      Metrics.counter(ERROR_COUNTER_NAME,
-          ERROR_TYPE_TAG_NAME, "parse",
-          CONFIG_CLASS_TAG_NAME, configurationClass.getName()).increment();
-      throw e;
-    }
+    logger.info("Received new dynamic configuration of length {}", configData.length());
+    parseConfiguration(configData, configurationClass).ifPresent(configuration::set);
+    this.notify();
   }
 
   @VisibleForTesting
-  public static <T> Optional<T> parseConfiguration(final String configurationYaml, final Class<T> configurationClass)
-      throws JsonProcessingException {
-    final T configuration = SystemMapper.yamlMapper().readValue(configurationYaml, configurationClass);
+  public static <T> Optional<T> parseConfiguration(final String configurationYaml, final Class<T> configurationClass) {
+    final T configuration;
+    try {
+      configuration = SystemMapper.yamlMapper().readValue(configurationYaml, configurationClass);
+    } catch (final IOException e) {
+      logger.warn("Failed to parse dynamic configuration", e);
+      Metrics.counter(ERROR_COUNTER_NAME,
+          ERROR_TYPE_TAG_NAME, "parse",
+          CONFIG_CLASS_TAG_NAME, configurationClass.getName()).increment();
+      return Optional.empty();
+    }
+
     final Set<ConstraintViolation<T>> violations = VALIDATOR.validate(configuration);
 
-    final Optional<T> maybeDynamicConfiguration;
-
-    if (violations.isEmpty()) {
-      maybeDynamicConfiguration = Optional.of(configuration);
-    } else {
+    if (!violations.isEmpty()) {
       logger.warn("Failed to validate configuration: {}", violations);
-      maybeDynamicConfiguration = Optional.empty();
+      Metrics.counter(ERROR_COUNTER_NAME,
+          ERROR_TYPE_TAG_NAME, "validate",
+          CONFIG_CLASS_TAG_NAME, configurationClass.getName()).increment();
+      return Optional.empty();
     }
 
-    return maybeDynamicConfiguration;
+    return Optional.of(configuration);
   }
 
-  private T retrieveInitialDynamicConfiguration() {
-    for (;;) {
-      try {
-        if (configurationToken == null) {
-          // first time around, start the configuration session
-          final StartConfigurationSessionResponse startResponse = appConfigClient
-              .startConfigurationSession(StartConfigurationSessionRequest.builder()
-                  .applicationIdentifier(application)
-                  .environmentIdentifier(environment)
-                  .configurationProfileIdentifier(configurationName).build());
-          configurationToken = startResponse.initialConfigurationToken();
-        }
-        return retrieveDynamicConfiguration().orElseThrow(() -> new IllegalStateException("No initial configuration available"));
-      } catch (Exception e) {
-        logger.warn("Error retrieving initial dynamic configuration", e);
-        Util.sleep(1000);
-      }
-    }
-  }
 }

service/src/main/java/org/whispersystems/textsecuregcm/workers/CheckDynamicConfigurationCommand.java

@@ -5,7 +5,6 @@
 
 package org.whispersystems.textsecuregcm.workers;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import io.dropwizard.core.cli.Command;
 import io.dropwizard.core.setup.Bootstrap;
 import java.nio.file.Files;
@@ -37,12 +36,7 @@ public class CheckDynamicConfigurationCommand extends Command {
   }
 
   private boolean isValid(final Class<?> configurationClass, final String yamlConfig) {
-    try {
-      return DynamicConfigurationManager.parseConfiguration(yamlConfig, configurationClass).isPresent();
-    } catch (JsonProcessingException e) {
-      System.err.println(e.getMessage());
-      return false;
-    }
+    return DynamicConfigurationManager.parseConfiguration(yamlConfig, configurationClass).isPresent();
   }
 
   /**

service/src/main/java/org/whispersystems/textsecuregcm/workers/CommandDependencies.java

@@ -103,8 +103,9 @@ record CommandDependencies(
     ScheduledExecutorService dynamicConfigurationExecutor = environment.lifecycle()
         .scheduledExecutorService(name(name, "dynamicConfiguration-%d")).threads(1).build();
 
-    DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager = configuration.getAppConfig().build(
-        DynamicConfiguration.class, dynamicConfigurationExecutor, awsCredentialsProvider);
+    DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager =
+        new DynamicConfigurationManager<>(
+            configuration.getDynamicConfig().build(awsCredentialsProvider, dynamicConfigurationExecutor), DynamicConfiguration.class);
     dynamicConfigurationManager.start();
 
     final ClientResources.Builder redisClientResourcesBuilder = ClientResources.builder();