Runtime config retrieval and application

conf/default-config.json

@@ -35,6 +35,6 @@
   "enclave_platform": null,
   "failure_shutdown_wait_hours": 120,
   "sharing_token_expiry_seconds": 2592000,
-  "operator_type": "public"
-
+  "operator_type": "public",
+  "config_scan_period_ms": 300000
 }

conf/feat-flag/feat-flag.json

@@ -0,0 +1,3 @@
+{
+  "remote_config_feature_flag": false
+}

conf/integ-config.json

@@ -14,6 +14,6 @@
   "optout_api_token": "test-operator-key",
   "optout_api_uri": "http://localhost:8081/optout/replicate",
   "salts_expired_shutdown_hours": 12,
-  "operator_type": "public"
-
+  "operator_type": "public",
+  "core_operator_config_path": "http://localhost:8088/operator/config"
 }

conf/local-config.json

@@ -36,5 +36,6 @@
   "key_sharing_endpoint_provide_app_names": true,
   "client_side_token_generate_log_invalid_http_origins": true,
   "salts_expired_shutdown_hours": 12,
-  "operator_type": "public"
+  "operator_type": "public",
+  "core_operator_config_path": "http://localhost:8088/operator/config"
 }

src/main/java/com/uid2/operator/Const.java

@@ -29,5 +29,10 @@ public class Config extends com.uid2.shared.Const.Config {
         public static final String OptOutStatusMaxRequestSize = "optout_status_max_request_size";
         public static final String MaxInvalidPaths = "logging_limit_max_invalid_paths_per_interval";
         public static final String MaxVersionBucketsPerSite = "logging_limit_max_version_buckets_per_site";
+
+        public static final String CoreConfigPath = "core_operator_config_path";
+        public static final String ConfigScanPeriodMs = "config_scan_period_ms";
+        public static final String IdentityV3 = "identity_v3";
+        public static final String RemoteConfigFeatureFlag = "remote_config_feature_flag";
     }
 }

src/main/java/com/uid2/operator/Main.java

@@ -8,8 +8,7 @@
 import com.uid2.operator.monitoring.IStatsCollectorQueue;
 import com.uid2.operator.monitoring.OperatorMetrics;
 import com.uid2.operator.monitoring.StatsCollectorVerticle;
-import com.uid2.operator.service.SecureLinkValidatorService;
-import com.uid2.operator.service.ShutdownService;
+import com.uid2.operator.service.*;
 import com.uid2.operator.vertx.Endpoints;
 import com.uid2.operator.vertx.OperatorShutdownHandler;
 import com.uid2.operator.store.CloudSyncOptOutStore;
@@ -38,9 +37,9 @@
 import io.micrometer.core.instrument.distribution.DistributionStatisticConfig;
 import io.micrometer.prometheus.PrometheusMeterRegistry;
 import io.micrometer.prometheus.PrometheusRenameFilter;
+import io.vertx.config.ConfigRetriever;
 import io.vertx.core.*;
 import io.vertx.core.http.HttpServerOptions;
-import io.vertx.core.http.impl.HttpUtils;
 import io.vertx.core.json.JsonObject;
 import io.vertx.micrometer.*;
 import io.vertx.micrometer.backends.BackendRegistries;
@@ -204,7 +203,9 @@ else if (!Utils.isProductionEnvironment()) {
         }
 
         Vertx vertx = createVertx();
-        VertxUtils.createConfigRetriever(vertx).getConfig(ar -> {
+        ConfigRetriever configRetriever = VertxUtils.createConfigRetriever(vertx);
+
+        configRetriever.getConfig(ar -> {
             if (ar.failed()) {
                 LOGGER.error("Unable to read config: " + ar.cause().getMessage(), ar.cause());
                 return;
@@ -268,39 +269,96 @@ private ICloudStorage wrapCloudStorageForOptOut(ICloudStorage cloudStorage) {
     private void run() throws Exception {
         this.createVertxInstancesMetric();
         this.createVertxEventLoopsMetric();
-        Supplier<Verticle> operatorVerticleSupplier = () -> {
-            UIDOperatorVerticle verticle = new UIDOperatorVerticle(config, this.clientSideTokenGenerate, siteProvider, clientKeyProvider, clientSideKeypairProvider, getKeyManager(), saltProvider, optOutStore, Clock.systemUTC(), _statsCollectorQueue, new SecureLinkValidatorService(this.serviceLinkProvider, this.serviceProvider), this.shutdownHandler::handleSaltRetrievalResponse);
-            return verticle;
-        };
-
-        DeploymentOptions options = new DeploymentOptions();
-        int svcInstances = this.config.getInteger(Const.Config.ServiceInstancesProp);
-        options.setInstances(svcInstances);
 
-        Promise<Void> compositePromise = Promise.promise();
-        List<Future> fs = new ArrayList<>();
-        fs.add(createAndDeployStatsCollector());
-        fs.add(createStoreVerticles());
+        ConfigRetrieverFactory configRetrieverFactory = new ConfigRetrieverFactory();
+        ConfigRetriever staticConfigRetriever = configRetrieverFactory.createJsonRetriever(vertx, config);
+        ConfigRetriever dynamicConfigRetriever = configRetrieverFactory.createRemoteConfigRetriever(vertx, config, this.createOperatorKeyRetriever().retrieve());
+        Future<ConfigService> staticConfigFuture = ConfigService.create(staticConfigRetriever);
+        Future<ConfigService> dynamicConfigFuture = ConfigService.create(dynamicConfigRetriever);
+
+        ConfigRetriever featureFlagConfigRetriever = configRetrieverFactory.createFileRetriever(
+                vertx,
+                "conf/feat-flag/feat-flag.json"
+        );
+        Future<JsonObject> featureFlagFuture = featureFlagConfigRetriever.getConfig();
+
+        featureFlagFuture.compose(featureFlagConfig -> {
+                    boolean featureFlag = featureFlagConfig.getBoolean(Const.Config.RemoteConfigFeatureFlag, true);
+                    // use static config if dynamic config fails and feature flag toggled off
+                    return dynamicConfigFuture
+                            .recover(throwable -> {
+                                if (!featureFlag) {
+                                    LOGGER.warn("Dynamic config service creation failed: ", throwable);
+                                    return staticConfigFuture;
+                                } else {
+                                    return Future.failedFuture(new Exception("Dynamic config service creation failed and feature flag is enabled: ", throwable));
+                                }
+                            })
+                            .compose(dynamicConfigService -> Future.all(Future.succeededFuture(dynamicConfigService), staticConfigFuture));
+                })
+                .compose(configServiceManagerCompositeFuture -> {
+                    ConfigService dynamicConfigService = configServiceManagerCompositeFuture.resultAt(0);
+                    ConfigService staticConfigService = configServiceManagerCompositeFuture.resultAt(1);
+
+                    boolean featureFlag = featureFlagConfigRetriever.getCachedConfig().getBoolean(Const.Config.RemoteConfigFeatureFlag, false);
+
+                    ConfigServiceManager configServiceManager = new ConfigServiceManager(
+                            vertx, dynamicConfigService, staticConfigService, featureFlag);
+
+                    setupFeatureFlagListener(configServiceManager, featureFlagConfigRetriever);
+
+                    IConfigService configService = configServiceManager.getDelegatingConfigService();
+                    Supplier<Verticle> operatorVerticleSupplier = () -> {
+                        UIDOperatorVerticle verticle = new UIDOperatorVerticle(configService, config, this.clientSideTokenGenerate, siteProvider, clientKeyProvider, clientSideKeypairProvider, getKeyManager(), saltProvider, optOutStore, Clock.systemUTC(), _statsCollectorQueue, new SecureLinkValidatorService(this.serviceLinkProvider, this.serviceProvider), this.shutdownHandler::handleSaltRetrievalResponse);
+                        return verticle;
+                    };
+
+                    DeploymentOptions options = new DeploymentOptions();
+                    int svcInstances = this.config.getInteger(Const.Config.ServiceInstancesProp);
+                    options.setInstances(svcInstances);
+
+                    Promise<Void> compositePromise = Promise.promise();
+                    List<Future> fs = new ArrayList<>();
+                    fs.add(createAndDeployStatsCollector());
+                    try {
+                        fs.add(createStoreVerticles());
+                    } catch (Exception e) {
+                        throw new RuntimeException(e);
+                    }
 
-        CompositeFuture.all(fs).onComplete(ar -> {
-            if (ar.failed()) compositePromise.fail(new Exception(ar.cause()));
-            else compositePromise.complete();
-        });
+                    CompositeFuture.all(fs).onComplete(ar -> {
+                        if (ar.failed()) compositePromise.fail(new Exception(ar.cause()));
+                        else compositePromise.complete();
+                    });
+
+                    return compositePromise.future()
+                            .compose(v -> {
+                                metrics.setup();
+                                vertx.setPeriodic(60000, id -> metrics.update());
+                                Promise<String> promise = Promise.promise();
+                                vertx.deployVerticle(operatorVerticleSupplier, options, promise);
+                                return promise.future();
+                            })
+                            .onFailure(t -> {
+                                LOGGER.error("Failed to bootstrap operator: " + t.getMessage(), new Exception(t));
+                                vertx.close();
+                                System.exit(1);
+                            });
+                });
+    }
 
-        compositePromise.future()
-            .compose(v -> {
-                metrics.setup();
-                vertx.setPeriodic(60000, id -> metrics.update());
-
-                Promise<String> promise = Promise.promise();
-                vertx.deployVerticle(operatorVerticleSupplier, options, promise);
-                return promise.future();
-            })
-            .onFailure(t -> {
-                LOGGER.error("Failed to bootstrap operator: " + t.getMessage(), new Exception(t));
-                vertx.close();
-                System.exit(1);
+    private void setupFeatureFlagListener(ConfigServiceManager manager, ConfigRetriever retriever) {
+        retriever.listen(change -> {
+            JsonObject newConfig = change.getNewConfiguration();
+            boolean useDynamicConfig = newConfig.getBoolean(Const.Config.RemoteConfigFeatureFlag, true);
+            manager.updateConfigService(useDynamicConfig).onComplete(update -> {
+                if (update.succeeded()) {
+                    LOGGER.info("Remote config feature flag toggled successfully");
+                } else {
+                    LOGGER.error("Failed to toggle remote config feature flag: " + update.cause());
+                }
             });
+        });
     }
 
     private Future<Void> createStoreVerticles() throws Exception {

src/main/java/com/uid2/operator/service/ConfigRetrieverFactory.java

@@ -0,0 +1,62 @@
+package com.uid2.operator.service;
+
+import io.vertx.config.ConfigRetriever;
+import io.vertx.config.ConfigRetrieverOptions;
+import io.vertx.config.ConfigStoreOptions;
+import io.vertx.core.Vertx;
+import io.vertx.core.json.JsonObject;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+
+import static com.uid2.operator.Const.Config.ConfigScanPeriodMs;
+import static com.uid2.operator.Const.Config.CoreConfigPath;
+
+public class ConfigRetrieverFactory {
+    public ConfigRetriever createRemoteConfigRetriever(Vertx vertx, JsonObject bootstrapConfig, String operatorKey) throws URISyntaxException {
+        String configPath = bootstrapConfig.getString(CoreConfigPath);
+        URI uri = new URI(configPath);
+
+        ConfigStoreOptions httpStore = new ConfigStoreOptions()
+                .setType("http")
+                .setOptional(true)
+                .setConfig(new JsonObject()
+                        .put("host", uri.getHost())
+                        .put("port", uri.getPort())
+                        .put("path", uri.getPath())
+                        .put("headers", new JsonObject()
+                                .put("Authorization", "Bearer " + operatorKey)));
+
+        ConfigRetrieverOptions retrieverOptions = new ConfigRetrieverOptions()
+                .setScanPeriod(bootstrapConfig.getLong(ConfigScanPeriodMs))
+                .addStore(httpStore);
+
+        return ConfigRetriever.create(vertx, retrieverOptions);
+    }
+
+    public ConfigRetriever createJsonRetriever(Vertx vertx, JsonObject config) {
+        ConfigStoreOptions jsonStore = new ConfigStoreOptions()
+                .setType("json")
+                .setConfig(config);
+
+        ConfigRetrieverOptions retrieverOptions = new ConfigRetrieverOptions()
+                .setScanPeriod(-1)
+                .addStore(jsonStore);
+
+
+        return ConfigRetriever.create(vertx, retrieverOptions);
+    }
+
+    public ConfigRetriever createFileRetriever(Vertx vertx, String path) {
+        ConfigStoreOptions fileStore = new ConfigStoreOptions()
+                .setType("file")
+                .setConfig(new JsonObject()
+                        .put("path", path)
+                        .put("format", "json"));
+
+        ConfigRetrieverOptions retrieverOptions = new ConfigRetrieverOptions()
+                .addStore(fileStore);
+
+        return ConfigRetriever.create(vertx, retrieverOptions);
+    }
+}

src/main/java/com/uid2/operator/service/ConfigService.java

@@ -0,0 +1,70 @@
+package com.uid2.operator.service;
+
+import com.uid2.operator.Const;
+import io.vertx.config.ConfigRetriever;
+import io.vertx.core.Future;
+import io.vertx.core.Promise;
+import io.vertx.core.json.JsonObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static com.uid2.operator.service.ConfigValidatorUtil.*;
+import static com.uid2.operator.service.UIDOperatorService.*;
+
+public class ConfigService implements IConfigService {
+
+    private final ConfigRetriever configRetriever;
+    private static final Logger logger = LoggerFactory.getLogger(ConfigService.class);
+
+    private ConfigService(ConfigRetriever configRetriever) {
+        this.configRetriever = configRetriever;
+        this.configRetriever.setConfigurationProcessor(this::configValidationHandler);
+    }
+
+    public static Future<ConfigService> create(ConfigRetriever configRetriever) {
+        Promise<ConfigService> promise = Promise.promise();
+
+        ConfigService instance = new ConfigService(configRetriever);
+
+        configRetriever.getConfig(ar -> {
+            if (ar.succeeded()) {
+                logger.info("Successfully loaded config");
+                promise.complete(instance);
+            } else {
+                logger.error("Failed to load config: {}", ar.cause().getMessage());
+                promise.fail(ar.cause());
+            }
+        });
+
+        return promise.future();
+    }
+
+    @Override
+    public JsonObject getConfig() {
+        return configRetriever.getCachedConfig();
+    }
+
+    private JsonObject configValidationHandler(JsonObject config) {
+        boolean isValid = true;
+        Integer identityExpiresAfter = config.getInteger(IDENTITY_TOKEN_EXPIRES_AFTER_SECONDS);
+        Integer refreshExpiresAfter = config.getInteger(REFRESH_TOKEN_EXPIRES_AFTER_SECONDS);
+        Integer refreshIdentityAfter = config.getInteger(REFRESH_IDENTITY_TOKEN_AFTER_SECONDS);
+        Integer maxBidstreamLifetimeSeconds = config.getInteger(Const.Config.MaxBidstreamLifetimeSecondsProp, identityExpiresAfter);
+
+        isValid &= validateIdentityRefreshTokens(identityExpiresAfter, refreshExpiresAfter, refreshIdentityAfter);
+
+        isValid &= validateBidstreamLifetime(maxBidstreamLifetimeSeconds, identityExpiresAfter);
+
+        if (!isValid) {
+            logger.error("Failed to update config");
+            JsonObject lastConfig = this.getConfig();
+            if (lastConfig == null || lastConfig.isEmpty()) {
+                throw new RuntimeException("Invalid config retrieved and no previous config to revert to");
+            }
+            return lastConfig;
+        }
+
+        logger.info("Successfully updated config");
+        return config;
+    }
+}

src/main/java/com/uid2/operator/service/ConfigServiceManager.java

@@ -0,0 +1,56 @@
+package com.uid2.operator.service;
+
+import io.vertx.core.Future;
+import io.vertx.core.Promise;
+import io.vertx.core.Vertx;
+import io.vertx.core.shareddata.Lock;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ConfigServiceManager {
+    private final Vertx vertx;
+    private final DelegatingConfigService delegatingConfigService;
+    private final IConfigService dynamicConfigService;
+    private final IConfigService staticConfigService;
+    private static final Logger logger = LoggerFactory.getLogger(ConfigServiceManager.class);
+
+    public ConfigServiceManager(Vertx vertx, IConfigService dynamicConfigService, IConfigService staticConfigService, boolean useDynamicConfig) {
+        this.vertx = vertx;
+        this.dynamicConfigService = dynamicConfigService;
+        this.staticConfigService = staticConfigService;
+        this.delegatingConfigService = new DelegatingConfigService(useDynamicConfig ? dynamicConfigService : staticConfigService);
+    }
+
+    public Future<Void> updateConfigService(boolean useDynamicConfig) {
+        Promise<Void> promise = Promise.promise();
+        vertx.sharedData().getLocalLock("updateConfigServiceLock", lockAsyncResult -> {
+            if (lockAsyncResult.succeeded()) {
+                Lock lock = lockAsyncResult.result();
+                try {
+                    if (useDynamicConfig) {
+                        logger.info("Switching to DynamicConfigService");
+                        delegatingConfigService.updateConfigService(dynamicConfigService);
+                    } else {
+                        logger.info("Switching to StaticConfigService");
+                        delegatingConfigService.updateConfigService(staticConfigService);
+                    }
+                    promise.complete();
+                } catch (Exception e) {
+                    promise.fail(e);
+                } finally {
+                    lock.release();
+                }
+            } else {
+                logger.error("Failed to acquire lock for updating active ConfigService", lockAsyncResult.cause());
+                promise.fail(lockAsyncResult.cause());
+            }
+        });
+
+        return promise.future();
+    }
+
+    public IConfigService getDelegatingConfigService() {
+        return delegatingConfigService;
+    }
+
+}