S3 application settings

docs/application-settings.md

@@ -259,6 +259,51 @@ Here's an example YAML file containing account-specific settings:
           default: true
 ```
 
+## Setting Account Configuration in S3
+
+This is identical to the account configuration in a file system, with the main difference that your file system is
+[AWS S3](https://aws.amazon.com/de/s3/) or any S3 compatible storage, such as [MinIO](https://min.io/).
+
+
+The general idea is that you'll place all the account-specific settings in a separate YAML file and point to that file.
+
+```yaml
+settings:
+  s3:
+      accessKeyId: <S3 access key>
+      secretAccessKey: <S3 access key>
+      endpoint: <endpoint> # http://s3.storage.com
+      bucket: <bucket name> # prebid-application-settings
+      region: <region name> # if not provided AWS_GLOBAL will be used. Example value: 'eu-central-1'
+      accounts-dir: accounts
+      stored-imps-dir: stored-impressions
+      stored-requests-dir: stored-requests
+      stored-responses-dir: stored-responses
+
+  # recommended to configure an in memory cache, but this is optional
+  in-memory-cache:
+      # example settings, tailor to your needs
+      cache-size: 100000
+      ttl-seconds: 1200 # 20 minutes
+      # recommended to configure
+      s3-update:
+          refresh-rate: 900000 # Refresh every 15 minutes
+          timeout: 5000
+```
+
+### File format
+
+We recommend using the `json` format for your account configuration. A minimal configuration may look like this.
+
+```json
+{
+  "id" : "979c7116-1f5a-43d4-9a87-5da3ccc4f52c",
+  "status" : "active"
+}
+```
+
+This pairs nicely if you have a default configuration defined in your prebid server config under `settings.default-account-config`.
+
 ## Setting Account Configuration in the Database
 
 In database approach account properties are stored in database table(s).

extra/pom.xml

@@ -52,6 +52,7 @@
         <protobuf.version>3.21.7</protobuf.version>
         <protoc.version>3.17.3</protoc.version>
         <json-logic.version>1.0.7</json-logic.version>
+        <aws.awssdk.version>2.26.24</aws.awssdk.version>
 
         <!-- Project test dependency versions -->
         <wiremock.version>3.9.1</wiremock.version>
@@ -212,6 +213,11 @@
                 <artifactId>geoip2</artifactId>
                 <version>${maxmind-client.version}</version>
             </dependency>
+            <dependency>
+                <groupId>software.amazon.awssdk</groupId>
+                <artifactId>s3</artifactId>
+                <version>${aws.awssdk.version}</version>
+            </dependency>
             <dependency>
                 <groupId>com.google.protobuf</groupId>
                 <artifactId>protobuf-java</artifactId>

pom.xml

@@ -170,6 +170,10 @@
             <groupId>org.postgresql</groupId>
             <artifactId>postgresql</artifactId>
         </dependency>
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>s3</artifactId>
+        </dependency>
         <dependency>
             <groupId>com.github.ben-manes.caffeine</groupId>
             <artifactId>caffeine</artifactId>

sample/configs/prebid-config-s3.yaml

@@ -0,0 +1,60 @@
+status-response: "ok"
+
+server:
+  enable-quickack: true
+  enable-reuseport: true
+
+adapters:
+  appnexus:
+    enabled: true
+  ix:
+    enabled: true
+  openx:
+    enabled: true
+  pubmatic:
+    enabled: true
+  rubicon:
+    enabled: true
+metrics:
+  prefix: prebid
+cache:
+  scheme: http
+  host: localhost
+  path: /cache
+  query: uuid=
+settings:
+  enforce-valid-account: false
+  generate-storedrequest-bidrequest-id: true
+  s3:
+    accessKeyId: prebid-server-test
+    secretAccessKey: nq9h6whXQURNL2NnWg3rcMlLMtGGDJeWrdl8hC9g
+    endpoint: http://localhost:9000
+    bucket: prebid-server-configs.example.com # prebid-application-settings
+    force-path-style: true # virtual bucketing
+    # region: <region name> # if not provided AWS_GLOBAL will be used. Example value: 'eu-central-1'
+    accounts-dir: accounts
+    stored-imps-dir: stored-impressions
+    stored-requests-dir: stored-requests
+    stored-responses-dir: stored-responses
+
+  in-memory-cache:
+    cache-size: 10000
+    ttl-seconds: 1200 # 20 minutes
+    s3-update:
+      refresh-rate: 900000 # Refresh every 15 minutes
+      timeout: 5000
+
+gdpr:
+  default-value: 1
+  vendorlist:
+    v2:
+      cache-dir: /var/tmp/vendor2
+    v3:
+      cache-dir: /var/tmp/vendor3
+
+admin-endpoints:
+  logging-changelevel:
+    enabled: true
+    path: /logging/changelevel
+    on-application-port: true
+    protected: false

src/main/java/org/prebid/server/settings/S3ApplicationSettings.java

@@ -0,0 +1,227 @@
+package org.prebid.server.settings;
+
+import io.vertx.core.CompositeFuture;
+import io.vertx.core.Future;
+import io.vertx.core.Promise;
+import io.vertx.core.Vertx;
+import org.apache.commons.collections4.SetUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.prebid.server.auction.model.Tuple2;
+import org.prebid.server.exception.PreBidException;
+import org.prebid.server.execution.Timeout;
+import org.prebid.server.json.DecodeException;
+import org.prebid.server.json.JacksonMapper;
+import org.prebid.server.settings.model.Account;
+import org.prebid.server.settings.model.StoredDataResult;
+import org.prebid.server.settings.model.StoredResponseDataResult;
+import software.amazon.awssdk.core.BytesWrapper;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Implementation of {@link ApplicationSettings}.
+ * <p>
+ * Reads an application settings from JSON file in a s3 bucket, stores and serves them in and from the memory.
+ * <p>
+ * Immediately loads stored request data from local files. These are stored in memory for low-latency reads.
+ * This expects each file in the directory to be named "{config_id}.json".
+ */
+public class S3ApplicationSettings implements ApplicationSettings {
+
+    private static final String JSON_SUFFIX = ".json";
+
+    final S3AsyncClient asyncClient;
+    final String bucket;
+    final String accountsDirectory;
+    final String storedImpressionsDirectory;
+    final String storedRequestsDirectory;
+    final String storedResponsesDirectory;
+    final JacksonMapper jacksonMapper;
+    final Vertx vertx;
+
+    public S3ApplicationSettings(S3AsyncClient asyncClient,
+                                 String bucket,
+                                 String accountsDirectory,
+                                 String storedImpressionsDirectory,
+                                 String storedRequestsDirectory,
+                                 String storedResponsesDirectory,
+                                 JacksonMapper jacksonMapper,
+                                 Vertx vertx) {
+
+        this.asyncClient = Objects.requireNonNull(asyncClient);
+        this.bucket = Objects.requireNonNull(bucket);
+        this.accountsDirectory = Objects.requireNonNull(accountsDirectory);
+        this.storedImpressionsDirectory = Objects.requireNonNull(storedImpressionsDirectory);
+        this.storedRequestsDirectory = Objects.requireNonNull(storedRequestsDirectory);
+        this.storedResponsesDirectory = Objects.requireNonNull(storedResponsesDirectory);
+        this.jacksonMapper = Objects.requireNonNull(jacksonMapper);
+        this.vertx = Objects.requireNonNull(vertx);
+    }
+
+    @Override
+    public Future<Account> getAccountById(String accountId, Timeout timeout) {
+        return withTimeout(() -> downloadFile(accountsDirectory + "/" + accountId + JSON_SUFFIX), timeout)
+                .map(fileContent -> decodeAccount(fileContent, accountId));
+    }
+
+    private Account decodeAccount(String fileContent, String requestedAccountId) {
+        if (fileContent == null) {
+            throw new PreBidException("Account with id %s not found".formatted(requestedAccountId));
+        }
+
+        final Account account;
+        try {
+            account = jacksonMapper.decodeValue(fileContent, Account.class);
+        } catch (DecodeException e) {
+            throw new PreBidException("Invalid json for account with id %s".formatted(requestedAccountId));
+        }
+
+        validateAccount(account, requestedAccountId);
+        return account;
+    }
+
+    private static void validateAccount(Account account, String requestedAccountId) {
+        final String receivedAccountId = account != null ? account.getId() : null;
+        if (!StringUtils.equals(receivedAccountId, requestedAccountId)) {
+            throw new PreBidException(
+                    "Account with id %s does not match id %s in file".formatted(requestedAccountId, receivedAccountId));
+        }
+    }
+
+    @Override
+    public Future<StoredDataResult> getStoredData(String accountId,
+                                                  Set<String> requestIds,
+                                                  Set<String> impIds,
+                                                  Timeout timeout) {
+
+        return withTimeout(
+                () -> Future.all(
+                        getFileContents(storedRequestsDirectory, requestIds),
+                        getFileContents(storedImpressionsDirectory, impIds)),
+                timeout)
+                .map(results -> buildStoredDataResult(
+                        results.resultAt(0),
+                        results.resultAt(1),
+                        requestIds,
+                        impIds));
+    }
+
+    private StoredDataResult buildStoredDataResult(Map<String, String> storedIdToRequest,
+                                                   Map<String, String> storedIdToImp,
+                                                   Set<String> requestIds,
+                                                   Set<String> impIds) {
+
+        final List<String> errors = Stream.concat(
+                        missingStoredDataIds(storedIdToImp, impIds).stream()
+                                .map("No stored impression found for id: %s"::formatted),
+                        missingStoredDataIds(storedIdToRequest, requestIds).stream()
+                                .map("No stored request found for id: %s"::formatted))
+                .toList();
+
+        return StoredDataResult.of(storedIdToRequest, storedIdToImp, errors);
+    }
+
+    private Set<String> missingStoredDataIds(Map<String, String> fileContents, Set<String> responseIds) {
+        return SetUtils.difference(responseIds, fileContents.keySet());
+    }
+
+    @Override
+    public Future<StoredDataResult> getAmpStoredData(String accountId,
+                                                     Set<String> requestIds,
+                                                     Set<String> impIds,
+                                                     Timeout timeout) {
+
+        return getStoredData(accountId, requestIds, Collections.emptySet(), timeout);
+    }
+
+    @Override
+    public Future<StoredDataResult> getVideoStoredData(String accountId,
+                                                       Set<String> requestIds,
+                                                       Set<String> impIds,
+                                                       Timeout timeout) {
+
+        return getStoredData(accountId, requestIds, impIds, timeout);
+    }
+
+    @Override
+    public Future<StoredResponseDataResult> getStoredResponses(Set<String> responseIds, Timeout timeout) {
+        return withTimeout(() -> getFileContents(storedResponsesDirectory, responseIds), timeout)
+                .map(storedIdToResponse -> StoredResponseDataResult.of(
+                        storedIdToResponse,
+                        missingStoredDataIds(storedIdToResponse, responseIds).stream()
+                                .map("No stored response found for id: %s"::formatted)
+                                .toList()));
+    }
+
+    @Override
+    public Future<Map<String, String>> getCategories(String primaryAdServer, String publisher, Timeout timeout) {
+        return Future.succeededFuture(Collections.emptyMap());
+    }
+
+    private Future<Map<String, String>> getFileContents(String directory, Set<String> ids) {
+        return Future.join(ids.stream()
+                        .map(impId -> downloadFile(directory + withInitialSlash(impId) + JSON_SUFFIX)
+                                .map(fileContent -> Tuple2.of(impId, fileContent)))
+                        .toList())
+                .map(CompositeFuture::<Tuple2<String, String>>list)
+                .map(impIdToFileContent -> impIdToFileContent.stream()
+                        .filter(tuple -> tuple.getRight() != null)
+                        .collect(Collectors.toMap(Tuple2::getLeft, Tuple2::getRight)));
+    }
+
+    /**
+     * When the impression id is the ad unit path it may already start with a slash and there's no need to add
+     * another one.
+     *
+     * @param impressionId from the bid request
+     * @return impression id with only a single slash at the beginning
+     */
+    private static String withInitialSlash(String impressionId) {
+        return impressionId.startsWith("/") ? impressionId : "/" + impressionId;
+    }
+
+    private Future<String> downloadFile(String key) {
+        final GetObjectRequest request = GetObjectRequest.builder().bucket(bucket).key(key).build();
+
+        return Future.fromCompletionStage(
+                        asyncClient.getObject(request, AsyncResponseTransformer.toBytes()),
+                        vertx.getOrCreateContext())
+                .map(BytesWrapper::asUtf8String)
+                .otherwiseEmpty();
+    }
+
+    private <T> Future<T> withTimeout(Supplier<Future<T>> futureFactory, Timeout timeout) {
+        final long remainingTime = timeout.remaining();
+        if (remainingTime <= 0L) {
+            return Future.failedFuture(new TimeoutException("Timeout has been exceeded"));
+        }
+
+        final Promise<T> promise = Promise.promise();
+        final Future<T> future = futureFactory.get();
+
+        final long timerId = vertx.setTimer(remainingTime, id ->
+                promise.tryFail(new TimeoutException("Timeout has been exceeded")));
+
+        future.onComplete(result -> {
+            vertx.cancelTimer(timerId);
+            if (result.succeeded()) {
+                promise.tryComplete(result.result());
+            } else {
+                promise.tryFail(result.cause());
+            }
+        });
+
+        return promise.future();
+    }
+}