refactor: Add message container to payload

spotify · Apr 23, 2024 · 2a89b67 · 2a89b67
1 parent 394e377
commit 2a89b67
Show file tree

Hide file tree

Showing 6 changed files with 67 additions and 19 deletions.
diff --git a/src/main/java/com/spotify/confidence/Confidence.java b/src/main/java/com/spotify/confidence/Confidence.java
@@ -81,18 +81,18 @@ public Confidence withContext(Map<String, ConfidenceValue> context) {
   }
 
   @Override
-  public void send(String name) {
+  public void send(String eventName) {
     try {
-      client().send(name, getContext(), Optional.empty());
+      client().send(eventName, getContext(), Optional.empty());
     } catch (IllegalStateException e) {
       // swallow this exception
     }
   }
 
   @Override
-  public void send(String name, ConfidenceValue.Struct message) {
+  public void send(String eventName, ConfidenceValue.Struct message) {
     try {
-      client().send(name, getContext(), Optional.of(message));
+      client().send(eventName, getContext(), Optional.of(message));
     } catch (IllegalStateException e) {
       // swallow this exception
     }

diff --git a/src/main/java/com/spotify/confidence/EventSender.java b/src/main/java/com/spotify/confidence/EventSender.java
@@ -5,9 +5,9 @@
 
 @Beta
 public interface EventSender extends Contextual {
-  public void send(String name, ConfidenceValue.Struct message);
+  public void send(String eventName, ConfidenceValue.Struct message);
 
-  public void send(String name);
+  public void send(String eventName);
 
   @Override
   EventSender withContext(ConfidenceValue.Struct context);

diff --git a/src/main/java/com/spotify/confidence/EventUploader.java b/src/main/java/com/spotify/confidence/EventUploader.java
@@ -1,7 +1,5 @@
 package com.spotify.confidence;
 
-import static com.spotify.confidence.GrpcEventUploader.CONTEXT;
-
 import com.google.protobuf.Struct;
 import com.spotify.confidence.events.v1.Event;
 import java.util.List;
@@ -15,8 +13,8 @@ static Event.Builder event(
         .setEventDefinition(EventSenderEngineImpl.EVENT_NAME_PREFIX + name)
         .setPayload(
             Struct.newBuilder()
-                .putAllFields(message.orElse(ConfidenceValue.Struct.EMPTY).asProtoMap())
-                .putFields(CONTEXT, context.toProto()));
+                .putAllFields(context.asProtoMap())
+                .putFields("message", message.orElse(ConfidenceValue.Struct.EMPTY).toProto()));
   }
 
   CompletableFuture<Boolean> upload(List<Event> events);

diff --git a/src/main/java/com/spotify/confidence/GrpcEventUploader.java b/src/main/java/com/spotify/confidence/GrpcEventUploader.java
@@ -26,7 +26,6 @@ class GrpcEventUploader implements EventUploader {
           Status.Code.ABORTED,
           Status.Code.INTERNAL,
           Status.Code.DATA_LOSS);
-  static final String CONTEXT = "context";
   private final String clientSecret;
   private final Sdk sdk;
   private final ManagedChannel managedChannel;

diff --git a/src/test/java/com/spotify/confidence/EventSenderEngineTest.java b/src/test/java/com/spotify/confidence/EventSenderEngineTest.java
@@ -6,6 +6,7 @@
 
 import com.google.common.base.Stopwatch;
 import com.google.common.collect.ImmutableMap;
+import com.google.protobuf.Struct;
 import com.google.protobuf.Value;
 import java.io.IOException;
 import java.time.Duration;
@@ -81,6 +82,48 @@ public void testEngineUploads() throws IOException {
     assertThat(uploadCallsCount - fullBatchCount).isEqualTo(additionalBatch);
   }
 
+  @Test
+  public void testOverlappingKeysInPayload() throws InterruptedException {
+    final FakeUploader alwaysSucceedUploader = new FakeUploader(List.of());
+    final EventSenderEngine engine =
+        new EventSenderEngineImpl(
+            1,
+            alwaysSucceedUploader,
+            clock,
+            DEFAULT_MAX_FLUSH_INTERVAL,
+            DEFAULT_MAX_MEMORY_CONSUMPTION);
+    // wait for the flush timeout to trigger the upload
+    engine.send(
+        "my_event",
+        ConfidenceValue.of(
+            ImmutableMap.of(
+                "a", ConfidenceValue.of(2),
+                "message", ConfidenceValue.of(3))),
+        Optional.of(
+            ConfidenceValue.Struct.of(
+                Map.of(
+                    "a", ConfidenceValue.of(0),
+                    "message", ConfidenceValue.of(1)))));
+    Thread.sleep(300);
+    assertThat(alwaysSucceedUploader.uploadCalls.peek().get(0).getPayload())
+        .isEqualTo(
+            Struct.newBuilder()
+                .putAllFields(
+                    Map.of(
+                        "message",
+                            Value.newBuilder()
+                                .setStructValue(
+                                    Struct.newBuilder()
+                                        .putFields(
+                                            "a", Value.newBuilder().setNumberValue(0).build())
+                                        .putFields(
+                                            "message",
+                                            Value.newBuilder().setNumberValue(1).build()))
+                                .build(),
+                        "a", Value.newBuilder().setNumberValue(2).build()))
+                .build());
+  }
+
   @Test
   public void testEngineCloseSuccessfullyWithoutEventsQueued() throws IOException {
     final FakeUploader alwaysSucceedUploader = new FakeUploader(List.of());
@@ -156,7 +199,15 @@ public void testEngineUploadsWhenIntermittentErrorWillRetry() throws IOException
     final Set<Value> uniqueEventIds =
         fakeUploader.uploadCalls.stream()
             .flatMap(Collection::stream)
-            .map(event -> event.getPayload().getFieldsMap().get("id"))
+            .map(
+                event ->
+                    event
+                        .getPayload()
+                        .getFieldsMap()
+                        .get("message")
+                        .getStructValue()
+                        .getFieldsMap()
+                        .get("id"))
             .collect(Collectors.toSet());
     // Verify all events reached the uploader
     assertThat(uniqueEventIds.size()).isEqualTo(numEvents);

diff --git a/src/test/java/com/spotify/confidence/GrpcEventUploaderTest.java b/src/test/java/com/spotify/confidence/GrpcEventUploaderTest.java
@@ -97,15 +97,15 @@ public void testMapsSingleEventBatchToProtobuf() throws ExecutionException, Inte
     assertThat(protoEvent.getEventDefinition()).isEqualTo("eventDefinitions/event1");
 
     final Map<String, com.google.protobuf.Value> fieldsMap = protoEvent.getPayload().getFieldsMap();
-    assertThat(fieldsMap.get("messageKey").getStringValue()).isEqualTo("value_1");
     assertThat(
             fieldsMap
-                .get(CONTEXT)
+                .get("message")
                 .getStructValue()
                 .getFieldsMap()
-                .get("contextKey")
+                .get("messageKey")
                 .getStringValue())
         .isEqualTo("value_1");
+    assertThat(fieldsMap.get("contextKey").getStringValue()).isEqualTo("value_1");
   }
 
   @Test
@@ -136,15 +136,15 @@ public void testMapsMultiEventBatchToProtobuf() {
 
       final Map<String, com.google.protobuf.Value> fieldsMap =
           protoEvent.getPayload().getFieldsMap();
-      assertThat(fieldsMap.get("messageKey").getStringValue()).isEqualTo("value_m" + (i + 1));
       assertThat(
               fieldsMap
-                  .get(CONTEXT)
+                  .get("message")
                   .getStructValue()
                   .getFieldsMap()
-                  .get("contextKey")
+                  .get("messageKey")
                   .getStringValue())
-          .isEqualTo("value_c" + (i + 1));
+          .isEqualTo("value_m" + (i + 1));
+      assertThat(fieldsMap.get("contextKey").getStringValue()).isEqualTo("value_c" + (i + 1));
     }
   }