Fix metrics counter reporting.

transport/transport-runtime/src/androidTest/java/com/google/android/datatransport/runtime/TestDestination.java

@@ -0,0 +1,40 @@
+// Copyright 2021 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.android.datatransport.runtime;
+
+import androidx.annotation.NonNull;
+import androidx.annotation.Nullable;
+
+public class TestDestination implements Destination {
+  private final String name;
+  private final byte[] extras;
+
+  public TestDestination(String name, byte[] extras) {
+    this.name = name;
+    this.extras = extras;
+  }
+
+  @NonNull
+  @Override
+  public String getName() {
+    return name;
+  }
+
+  @Nullable
+  @Override
+  public byte[] getExtras() {
+    return extras;
+  }
+}

transport/transport-runtime/src/androidTest/java/com/google/android/datatransport/runtime/UploaderIntegrationTest.java

@@ -36,13 +36,19 @@
 import com.google.android.datatransport.runtime.backends.BackendRequest;
 import com.google.android.datatransport.runtime.backends.BackendResponse;
 import com.google.android.datatransport.runtime.backends.TransportBackend;
+import com.google.android.datatransport.runtime.firebase.transport.LogEventDropped;
+import com.google.android.datatransport.runtime.firebase.transport.LogSourceMetrics;
 import com.google.android.datatransport.runtime.scheduling.jobscheduling.WorkScheduler;
 import com.google.android.datatransport.runtime.scheduling.persistence.EventStore;
 import com.google.android.datatransport.runtime.scheduling.persistence.PersistedEvent;
+import com.google.android.datatransport.runtime.scheduling.persistence.SQLiteEventStore;
 import com.google.android.datatransport.runtime.synchronization.SynchronizationException;
 import java.nio.charset.Charset;
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 import java.util.UUID;
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -77,6 +83,12 @@ public void setUp() {
                 invocation -> invocation.<EventInternal>getArgument(0).toBuilder().build());
   }
 
+  @After
+  public void tearDown() {
+    component.getEventStore().clearDb();
+    component.getEventStore().resetClientMetrics();
+  }
+
   private String generateBackendName() {
     return UUID.randomUUID().toString().replace("-", "");
   }
@@ -201,6 +213,61 @@ public void uploader_dbException_shouldReschedule() {
     Transport<String> transport =
         factory.getTransport(testTransport, String.class, String::getBytes);
     Event<String> stringEvent = Event.ofTelemetry("TelemetryData");
+
+    transport.send(stringEvent);
+    verify(spyScheduler, times(1)).schedule(any(), eq(2));
+    assertThat(store.getNextCallTime(transportContext)).isEqualTo(0);
+  }
+
+  @Test
+  public void uploader_invalidPayload_shouldNotReschedule() {
+    TransportRuntime runtime = TransportRuntime.getInstance();
+    SQLiteEventStore store = component.getEventStore();
+    String mockBackendName = generateBackendName();
+    byte[] mockExtras = "extras".getBytes(Charset.defaultCharset());
+
+    when(mockRegistry.get(mockBackendName)).thenReturn(mockBackend);
+    when(mockBackend.send(any())).thenReturn(BackendResponse.invalidPayload());
+
+    Transport<String> transport =
+        runtime
+            .newFactory(new TestDestination(mockBackendName, mockExtras))
+            .getTransport(testTransport, String.class, String::getBytes);
+    transport.send(Event.ofTelemetry("TelemetryData"));
+
+    verify(spyScheduler, times(0)).schedule(any(), eq(2));
+
+    List<LogSourceMetrics> logSourceMetrics = store.loadClientMetrics().getLogSourceMetricsList();
+    assertThat(logSourceMetrics).hasSize(1);
+    LogSourceMetrics metrics = logSourceMetrics.get(0);
+    List<LogEventDropped> logEventDroppedList = metrics.getLogEventDroppedList();
+    assertThat(logEventDroppedList).hasSize(1);
+    LogEventDropped logEventDropped = logEventDroppedList.get(0);
+    assertThat(logEventDropped.getEventsDroppedCount()).isEqualTo(1);
+    assertThat(logEventDropped.getReason()).isEqualTo(LogEventDropped.Reason.INVALID_PAYLOD);
+  }
+
+  @Test
+  public void uploader_withPendingClientMetricsAndSuccessfulUpload_shouldResetCounters() {
+    TransportRuntime runtime = TransportRuntime.getInstance();
+    SQLiteEventStore store = component.getEventStore();
+    String anotherTransport = "anotherTransport";
+    store.recordLogEventDropped(20, LogEventDropped.Reason.CACHE_FULL, testTransport);
+    store.recordLogEventDropped(1, LogEventDropped.Reason.MAX_RETRIES_REACHED, testTransport);
+    store.recordLogEventDropped(1, LogEventDropped.Reason.INVALID_PAYLOD, anotherTransport);
+
+    assertThat(store.loadClientMetrics().getLogSourceMetricsList()).hasSize(2);
+
+    String mockBackendName = generateBackendName();
+    byte[] mockExtras = "extras".getBytes(Charset.defaultCharset());
+
+    when(mockRegistry.get(mockBackendName)).thenReturn(mockBackend);
+    when(mockBackend.send(any())).thenReturn(BackendResponse.ok(1));
+
+    Transport<String> transport =
+        runtime
+            .newFactory(new TestDestination(mockBackendName, mockExtras))
+            .getTransport(testTransport, String.class, String::getBytes);
     EventInternal expectedEvent =
         EventInternal.builder()
             .setEventMillis(3)
@@ -210,8 +277,142 @@ public void uploader_dbException_shouldReschedule() {
                 new EncodedPayload(
                     PROTOBUF_ENCODING, "TelemetryData".getBytes(Charset.defaultCharset())))
             .build();
-    transport.send(stringEvent);
-    verify(spyScheduler, times(1)).schedule(any(), eq(2));
-    assertThat(store.getNextCallTime(transportContext)).isEqualTo(0);
+    EventInternal metricsEvent = component.getUploader().createMetricsEvent(mockBackend);
+    transport.send(Event.ofTelemetry("TelemetryData"));
+    verify(mockBackend, times(1))
+        .send(
+            eq(
+                BackendRequest.builder()
+                    .setEvents(Arrays.asList(expectedEvent, metricsEvent))
+                    .setExtras(mockExtras)
+                    .build()));
+
+    verify(spyScheduler, times(0)).schedule(any(), eq(2));
+
+    assertThat(store.loadClientMetrics().getLogSourceMetricsList()).isEmpty();
+  }
+
+  @Test
+  public void uploader_withPendingClientMetricsAndInvalidPayload_shouldNotResetCounters() {
+    TransportRuntime runtime = TransportRuntime.getInstance();
+    SQLiteEventStore store = component.getEventStore();
+    String anotherTransport = "anotherTransport";
+    store.recordLogEventDropped(1, LogEventDropped.Reason.INVALID_PAYLOD, anotherTransport);
+
+    assertThat(store.loadClientMetrics().getLogSourceMetricsList()).hasSize(1);
+
+    String mockBackendName = generateBackendName();
+    byte[] mockExtras = "extras".getBytes(Charset.defaultCharset());
+
+    when(mockRegistry.get(mockBackendName)).thenReturn(mockBackend);
+    when(mockBackend.send(any())).thenReturn(BackendResponse.invalidPayload());
+
+    Transport<String> transport =
+        runtime
+            .newFactory(new TestDestination(mockBackendName, mockExtras))
+            .getTransport(testTransport, String.class, String::getBytes);
+    EventInternal expectedEvent =
+        EventInternal.builder()
+            .setEventMillis(3)
+            .setUptimeMillis(1)
+            .setTransportName(testTransport)
+            .setEncodedPayload(
+                new EncodedPayload(
+                    PROTOBUF_ENCODING, "TelemetryData".getBytes(Charset.defaultCharset())))
+            .build();
+    EventInternal metricsEvent = component.getUploader().createMetricsEvent(mockBackend);
+    transport.send(Event.ofTelemetry("TelemetryData"));
+    verify(mockBackend, times(1))
+        .send(
+            eq(
+                BackendRequest.builder()
+                    .setEvents(Arrays.asList(expectedEvent, metricsEvent))
+                    .setExtras(mockExtras)
+                    .build()));
+
+    verify(spyScheduler, times(0)).schedule(any(), eq(2));
+
+    assertThat(store.loadClientMetrics().getLogSourceMetricsList()).hasSize(2);
+  }
+
+  @Test
+  public void uploader_withPendingClientMetricsAndFatalError_shouldNotResetCounters() {
+    TransportRuntime runtime = TransportRuntime.getInstance();
+    SQLiteEventStore store = component.getEventStore();
+    String anotherTransport = "anotherTransport";
+    store.recordLogEventDropped(1, LogEventDropped.Reason.INVALID_PAYLOD, anotherTransport);
+
+    assertThat(store.loadClientMetrics().getLogSourceMetricsList()).hasSize(1);
+
+    String mockBackendName = generateBackendName();
+    byte[] mockExtras = "extras".getBytes(Charset.defaultCharset());
+
+    when(mockRegistry.get(mockBackendName)).thenReturn(mockBackend);
+    when(mockBackend.send(any())).thenReturn(BackendResponse.fatalError());
+
+    Transport<String> transport =
+        runtime
+            .newFactory(new TestDestination(mockBackendName, mockExtras))
+            .getTransport(testTransport, String.class, String::getBytes);
+    EventInternal expectedEvent =
+        EventInternal.builder()
+            .setEventMillis(3)
+            .setUptimeMillis(1)
+            .setTransportName(testTransport)
+            .setEncodedPayload(
+                new EncodedPayload(
+                    PROTOBUF_ENCODING, "TelemetryData".getBytes(Charset.defaultCharset())))
+            .build();
+    EventInternal metricsEvent = component.getUploader().createMetricsEvent(mockBackend);
+    transport.send(Event.ofTelemetry("TelemetryData"));
+    verify(mockBackend, times(1))
+        .send(
+            eq(
+                BackendRequest.builder()
+                    .setEvents(Arrays.asList(expectedEvent, metricsEvent))
+                    .setExtras(mockExtras)
+                    .build()));
+
+    verify(spyScheduler, times(0)).schedule(any(), eq(2));
+
+    assertThat(store.loadClientMetrics().getLogSourceMetricsList()).hasSize(1);
+  }
+
+  @Test
+  public void
+      uploader_withPendingClientMetricsAndSuccessfulUploadButNotLegacyTarget_shouldNotResetCounters() {
+    TransportRuntime runtime = TransportRuntime.getInstance();
+    SQLiteEventStore store = component.getEventStore();
+    String anotherTransport = "anotherTransport";
+    store.recordLogEventDropped(1, LogEventDropped.Reason.INVALID_PAYLOD, anotherTransport);
+
+    assertThat(store.loadClientMetrics().getLogSourceMetricsList()).hasSize(1);
+
+    String mockBackendName = generateBackendName();
+
+    when(mockRegistry.get(mockBackendName)).thenReturn(mockBackend);
+    when(mockBackend.send(any())).thenReturn(BackendResponse.ok(1));
+
+    Transport<String> transport =
+        runtime
+            .newFactory(new TestDestination(mockBackendName, null))
+            .getTransport(testTransport, String.class, String::getBytes);
+    EventInternal expectedEvent =
+        EventInternal.builder()
+            .setEventMillis(3)
+            .setUptimeMillis(1)
+            .setTransportName(testTransport)
+            .setEncodedPayload(
+                new EncodedPayload(
+                    PROTOBUF_ENCODING, "TelemetryData".getBytes(Charset.defaultCharset())))
+            .build();
+
+    transport.send(Event.ofTelemetry("TelemetryData"));
+    verify(mockBackend, times(1))
+        .send(eq(BackendRequest.create(Collections.singletonList(expectedEvent))));
+
+    verify(spyScheduler, times(0)).schedule(any(), eq(2));
+
+    assertThat(store.loadClientMetrics().getLogSourceMetricsList()).hasSize(1);
   }
 }

transport/transport-runtime/src/androidTest/java/com/google/android/datatransport/runtime/UploaderTestRuntimeComponent.java

@@ -16,6 +16,7 @@
 
 import android.content.Context;
 import com.google.android.datatransport.runtime.backends.BackendRegistry;
+import com.google.android.datatransport.runtime.scheduling.jobscheduling.Uploader;
 import com.google.android.datatransport.runtime.scheduling.jobscheduling.WorkScheduler;
 import com.google.android.datatransport.runtime.scheduling.persistence.SQLiteEventStore;
 import com.google.android.datatransport.runtime.scheduling.persistence.SpyEventStoreModule;
@@ -44,6 +45,8 @@ abstract class UploaderTestRuntimeComponent extends TransportRuntimeComponent {
 
   abstract WorkScheduler getWorkScheduler();
 
+  abstract Uploader getUploader();
+
   @Override
   public void close() throws IOException {
     getEventStore().clearDb();

transport/transport-runtime/src/main/java/com/google/android/datatransport/runtime/scheduling/jobscheduling/Uploader.java

@@ -17,6 +17,7 @@
 import android.content.Context;
 import android.net.ConnectivityManager;
 import android.net.NetworkInfo;
+import androidx.annotation.VisibleForTesting;
 import com.google.android.datatransport.Encoding;
 import com.google.android.datatransport.runtime.EncodedPayload;
 import com.google.android.datatransport.runtime.EventInternal;
@@ -136,18 +137,7 @@ void logAndUpdateState(TransportContext transportContext, int attemptNumber) {
         }
 
         if (transportContext.shouldUploadClientHealthMetrics()) {
-          ClientMetrics clientMetrics =
-              guard.runCriticalSection(clientHealthMetricsStore::loadClientMetrics);
-          EventInternal eventInternal =
-              EventInternal.builder()
-                  .setEventMillis(clock.getTime())
-                  .setUptimeMillis(uptimeClock.getTime())
-                  .setTransportName(CLIENT_HEALTH_METRICS_LOG_SOURCE)
-                  .setEncodedPayload(
-                      new EncodedPayload(Encoding.of("proto"), clientMetrics.toByteArray()))
-                  .build();
-          EventInternal decoratedEvent = backend.decorate(eventInternal);
-          eventInternals.add(decoratedEvent);
+          eventInternals.add(createMetricsEvent(backend));
         }
 
         response =
@@ -177,6 +167,13 @@ void logAndUpdateState(TransportContext transportContext, int attemptNumber) {
         if (response.getStatus() == BackendResponse.Status.OK) {
           maxNextRequestWaitMillis =
               Math.max(maxNextRequestWaitMillis, response.getNextRequestWaitMillis());
+          if (transportContext.shouldUploadClientHealthMetrics()) {
+            guard.runCriticalSection(
+                () -> {
+                  clientHealthMetricsStore.resetClientMetrics();
+                  return null;
+                });
+          }
         } else if (response.getStatus() == BackendResponse.Status.INVALID_PAYLOAD) {
           Map<String, Integer> countMap = new HashMap<>();
           for (PersistedEvent persistedEvent : persistedEvents) {
@@ -206,4 +203,18 @@ void logAndUpdateState(TransportContext transportContext, int attemptNumber) {
           return null;
         });
   }
+
+  @VisibleForTesting
+  public EventInternal createMetricsEvent(TransportBackend backend) {
+    ClientMetrics clientMetrics =
+        guard.runCriticalSection(clientHealthMetricsStore::loadClientMetrics);
+    return backend.decorate(
+        EventInternal.builder()
+            .setEventMillis(clock.getTime())
+            .setUptimeMillis(uptimeClock.getTime())
+            .setTransportName(CLIENT_HEALTH_METRICS_LOG_SOURCE)
+            .setEncodedPayload(
+                new EncodedPayload(Encoding.of("proto"), clientMetrics.toByteArray()))
+            .build());
+  }
 }