feat: add upsert all support

config-service-api/src/main/proto/org/hypertrace/config/service/v1/config_service.proto

@@ -19,6 +19,8 @@ service ConfigService {
 
   // Deletes the config for the specified request
   rpc DeleteConfig(DeleteConfigRequest) returns (DeleteConfigResponse) {}
+
+  rpc UpsertAllConfigs(UpsertAllConfigsRequest) returns (UpsertAllConfigsResponse) {}
 }
 
 message UpsertConfigRequest {
@@ -110,3 +112,26 @@ message DeleteConfigRequest {
 message DeleteConfigResponse {
   ContextSpecificConfig deleted_config = 1;
 }
+
+message UpsertAllConfigsRequest {
+  repeated ConfigToUpsert configs = 1;
+
+  message ConfigToUpsert {
+    string resource_name = 1;
+    string resource_namespace = 2;
+    string context = 3;
+    google.protobuf.Value config = 4;
+  }
+}
+
+message UpsertAllConfigsResponse {
+  repeated UpsertedConfig upserted_configs = 1;
+
+  message UpsertedConfig {
+    string context = 1;
+    google.protobuf.Value config = 2;
+    int64 creation_timestamp = 3;
+    int64 update_timestamp = 4;
+    optional google.protobuf.Value prev_config = 5;
+  }
+}

config-service-api/src/testFixtures/java/org/hypertrace/config/service/test/MockGenericConfigService.java

@@ -23,6 +23,7 @@
 import io.grpc.inprocess.InProcessServerBuilder;
 import io.grpc.stub.StreamObserver;
 import java.io.IOException;
+import java.time.Clock;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Objects;
@@ -37,6 +38,9 @@
 import org.hypertrace.config.service.v1.GetAllConfigsResponse;
 import org.hypertrace.config.service.v1.GetConfigRequest;
 import org.hypertrace.config.service.v1.GetConfigResponse;
+import org.hypertrace.config.service.v1.UpsertAllConfigsRequest;
+import org.hypertrace.config.service.v1.UpsertAllConfigsResponse;
+import org.hypertrace.config.service.v1.UpsertAllConfigsResponse.UpsertedConfig;
 import org.hypertrace.config.service.v1.UpsertConfigRequest;
 import org.hypertrace.config.service.v1.UpsertConfigResponse;
 import org.hypertrace.core.grpcutils.context.RequestContext;
@@ -51,6 +55,7 @@
 public class MockGenericConfigService {
 
   private Server grpcServer;
+  private Clock clock = Clock.systemUTC();
   private final InProcessServerBuilder serverBuilder;
   private final ManagedChannel configChannel;
   private final RequestContext context = RequestContext.forTenantId("default tenant");
@@ -102,35 +107,34 @@ public void shutdown() {
     this.configChannel.shutdownNow();
   }
 
+  public MockGenericConfigService withClock(Clock clock) {
+    this.clock = clock;
+    return this;
+  }
+
   @SuppressWarnings("unchecked")
   public MockGenericConfigService mockUpsert() {
     Mockito.doAnswer(
             invocation -> {
               UpsertConfigRequest request = invocation.getArgument(0, UpsertConfigRequest.class);
               StreamObserver<UpsertConfigResponse> responseStreamObserver =
                   invocation.getArgument(1, StreamObserver.class);
-              ResourceType resourceType =
-                  ResourceType.of(request.getResourceNamespace(), request.getResourceName());
-              String configContext = configContextOrDefault(request.getContext());
-              ContextSpecificConfig existingConfig = currentValues.get(resourceType, configContext);
-              long updateTimestamp = System.currentTimeMillis();
-              long creationTimestamp =
-                  existingConfig == null ? updateTimestamp : existingConfig.getCreationTimestamp();
-              currentValues.put(
-                  resourceType,
-                  configContext,
-                  ContextSpecificConfig.newBuilder()
-                      .setContext(configContext)
-                      .setConfig(request.getConfig())
-                      .setCreationTimestamp(creationTimestamp)
-                      .setUpdateTimestamp(updateTimestamp)
-                      .build());
-              responseStreamObserver.onNext(
+              UpsertedConfig upsertedConfig =
+                  this.writeToMap(
+                      request.getResourceNamespace(),
+                      request.getResourceName(),
+                      request.getContext(),
+                      request.getConfig());
+              UpsertConfigResponse.Builder responseBuilder =
                   UpsertConfigResponse.newBuilder()
-                      .setConfig(request.getConfig())
-                      .setCreationTimestamp(creationTimestamp)
-                      .setUpdateTimestamp(updateTimestamp)
-                      .build());
+                      .setConfig(upsertedConfig.getConfig())
+                      .setCreationTimestamp(upsertedConfig.getCreationTimestamp())
+                      .setUpdateTimestamp(upsertedConfig.getUpdateTimestamp());
+              if (upsertedConfig.hasPrevConfig()) {
+                responseBuilder.setPrevConfig(upsertedConfig.getPrevConfig());
+              }
+
+              responseStreamObserver.onNext(responseBuilder.build());
               responseStreamObserver.onCompleted();
               return null;
             })
@@ -243,6 +247,35 @@ public MockGenericConfigService mockGet() {
     return this;
   }
 
+  public MockGenericConfigService mockUpsertAll() {
+    Mockito.doAnswer(
+            invocation -> {
+              UpsertAllConfigsRequest request =
+                  invocation.getArgument(0, UpsertAllConfigsRequest.class);
+              StreamObserver<UpsertAllConfigsResponse> responseStreamObserver =
+                  invocation.getArgument(1, StreamObserver.class);
+
+              List<UpsertedConfig> configs =
+                  request.getConfigsList().stream()
+                      .map(
+                          configToUpsert ->
+                              this.writeToMap(
+                                  configToUpsert.getResourceNamespace(),
+                                  configToUpsert.getResourceName(),
+                                  configToUpsert.getContext(),
+                                  configToUpsert.getConfig()))
+                      .collect(Collectors.toUnmodifiableList());
+              responseStreamObserver.onNext(
+                  UpsertAllConfigsResponse.newBuilder().addAllUpsertedConfigs(configs).build());
+              responseStreamObserver.onCompleted();
+              return null;
+            })
+        .when(this.mockConfigService)
+        .upsertConfig(ArgumentMatchers.any(), ArgumentMatchers.any());
+
+    return this;
+  }
+
   private Optional<Value> mergeValues(List<Value> values) {
     if (values.isEmpty()) {
       return Optional.empty();
@@ -259,6 +292,40 @@ private boolean isValidValue(Optional<Value> value) {
     return value.isPresent() && value.get().getKindCase() != Value.KindCase.NULL_VALUE;
   }
 
+  private UpsertedConfig writeToMap(String namespace, String name, String context, Value config) {
+    ResourceType resourceType = ResourceType.of(namespace, name);
+    String configContext = configContextOrDefault(context);
+    ContextSpecificConfig existingConfig = currentValues.get(resourceType, configContext);
+    long updateTimestamp = clock.millis();
+    long creationTimestamp =
+        Optional.ofNullable(existingConfig)
+            .map(ContextSpecificConfig::getCreationTimestamp)
+            .orElse(updateTimestamp);
+    Optional<Value> previousConfig =
+        Optional.ofNullable(
+                currentValues.put(
+                    resourceType,
+                    configContext,
+                    ContextSpecificConfig.newBuilder()
+                        .setContext(configContext)
+                        .setConfig(config)
+                        .setCreationTimestamp(creationTimestamp)
+                        .setUpdateTimestamp(updateTimestamp)
+                        .build()))
+            .map(ContextSpecificConfig::getConfig);
+
+    UpsertedConfig.Builder resultBuilder =
+        UpsertedConfig.newBuilder()
+            .setConfig(config)
+            .setContext(configContext)
+            .setCreationTimestamp(creationTimestamp)
+            .setUpdateTimestamp(updateTimestamp);
+
+    previousConfig.ifPresent(resultBuilder::setPrevConfig);
+
+    return resultBuilder.build();
+  }
+
   private class TestInterceptor implements ServerInterceptor {
     @Override
     public <ReqT, RespT> Listener<ReqT> interceptCall(

config-service-impl/src/main/java/org/hypertrace/config/service/ConfigResource.java

@@ -3,14 +3,12 @@
 import lombok.Value;
 
 /**
- * Identifies the configuration resource which you want to deal with. A single config resource can
- * have multiple versions of config values associated with it.
+ * Identifies the configuration resource which you want to deal with. Multiple contexts may exist
+ * for this resource, each potentially with multiple versions.
  */
 @Value
 public class ConfigResource {
-
   String resourceName;
   String resourceNamespace;
   String tenantId;
-  String context;
 }

config-service-impl/src/main/java/org/hypertrace/config/service/ConfigResourceContext.java

@@ -0,0 +1,23 @@
+package org.hypertrace.config.service;
+
+import com.google.common.base.Strings;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import lombok.Value;
+
+/** A specific context within a configuration resource. */
+@Value
+public class ConfigResourceContext {
+  private static final String DEFAULT_CONTEXT = "DEFAULT-CONTEXT";
+  ConfigResource configResource;
+  String context;
+
+  public ConfigResourceContext(@Nonnull ConfigResource configResource) {
+    this(configResource, null);
+  }
+
+  public ConfigResourceContext(@Nonnull ConfigResource configResource, @Nullable String context) {
+    this.context = Strings.isNullOrEmpty(context) ? DEFAULT_CONTEXT : context;
+    this.configResource = configResource;
+  }
+}