Break down push latency metrics by VOIP/not-VOIP and optionally by client version

service/src/main/java/org/whispersystems/textsecuregcm/metrics/PushLatencyManager.java

@@ -6,17 +6,34 @@
 package org.whispersystems.textsecuregcm.metrics;
 
 import com.codahale.metrics.MetricRegistry;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.google.common.annotations.VisibleForTesting;
+import com.vdurmont.semver4j.Semver;
 import io.lettuce.core.SetArgs;
-import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
 import io.micrometer.core.instrument.Metrics;
-import io.micrometer.core.instrument.Tags;
-import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
-
+import io.micrometer.core.instrument.Tag;
+import java.time.Clock;
 import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
+import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
+import org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
+import org.whispersystems.textsecuregcm.util.SystemMapper;
+import org.whispersystems.textsecuregcm.util.ua.UnrecognizedUserAgentException;
+import org.whispersystems.textsecuregcm.util.ua.UserAgent;
+import org.whispersystems.textsecuregcm.util.ua.UserAgentUtil;
 
 /**
  * Measures and records the latency between sending a push notification to a device and that device draining its queue
@@ -28,48 +45,157 @@ import java.util.concurrent.TimeUnit;
  * device and records the time elapsed since the push notification timestamp as a latency observation.
  */
 public class PushLatencyManager {
-    private static final String TIMER_NAME = MetricRegistry.name(PushLatencyManager.class, "latency");
-    private static final int    TTL        = (int)Duration.ofDays(1).toSeconds();
 
-    private final FaultTolerantRedisCluster redisCluster;
+  private final FaultTolerantRedisCluster redisCluster;
+  private final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager;
 
-    public PushLatencyManager(final FaultTolerantRedisCluster redisCluster) {
-        this.redisCluster = redisCluster;
+  private final Clock clock;
+
+  private static final String TIMER_NAME = MetricRegistry.name(PushLatencyManager.class, "latency");
+  private static final int TTL = (int) Duration.ofDays(1).toSeconds();
+
+  private static final Logger log = LoggerFactory.getLogger(PushLatencyManager.class);
+
+  @VisibleForTesting
+  enum PushType {
+    STANDARD,
+    VOIP
+  }
+
+  @VisibleForTesting
+  static class PushRecord {
+    private final Instant timestamp;
+
+    @Nullable
+    private final PushType pushType;
+
+    @JsonCreator
+    PushRecord(@JsonProperty("timestamp") final Instant timestamp,
+        @JsonProperty("pushType") @Nullable final PushType pushType) {
+
+      this.timestamp = timestamp;
+      this.pushType = pushType;
     }
 
-    public void recordPushSent(final UUID accountUuid, final long deviceId) {
-        recordPushSent(accountUuid, deviceId, System.currentTimeMillis());
+    public Instant getTimestamp() {
+      return timestamp;
     }
 
-    @VisibleForTesting
-    void recordPushSent(final UUID accountUuid, final long deviceId, final long currentTime) {
-        redisCluster.useCluster(connection ->
-                connection.async().set(getFirstUnacknowledgedPushKey(accountUuid, deviceId), String.valueOf(currentTime), SetArgs.Builder.nx().ex(TTL)));
+    @Nullable
+    public PushType getPushType() {
+      return pushType;
     }
+  }
 
-    public void recordQueueRead(final UUID accountUuid, final long deviceId, final String userAgent) {
-        getLatencyAndClearTimestamp(accountUuid, deviceId, System.currentTimeMillis()).thenAccept(latency -> {
-            if (latency != null) {
-                Metrics.timer(TIMER_NAME, Tags.of(UserAgentTagUtil.getPlatformTag(userAgent))).record(latency, TimeUnit.MILLISECONDS);
+  public PushLatencyManager(final FaultTolerantRedisCluster redisCluster,
+      final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager) {
+
+    this(redisCluster, dynamicConfigurationManager, Clock.systemUTC());
+  }
+
+  @VisibleForTesting
+  PushLatencyManager(final FaultTolerantRedisCluster redisCluster,
+      final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager,
+      final Clock clock) {
+
+    this.redisCluster = redisCluster;
+    this.dynamicConfigurationManager = dynamicConfigurationManager;
+    this.clock = clock;
+  }
+
+  public void recordPushSent(final UUID accountUuid, final long deviceId, final boolean isVoip) {
+    try {
+      final String recordJson = SystemMapper.getMapper().writeValueAsString(
+          new PushRecord(Instant.now(clock), isVoip ? PushType.VOIP : PushType.STANDARD));
+
+      redisCluster.useCluster(connection ->
+          connection.async().set(getFirstUnacknowledgedPushKey(accountUuid, deviceId),
+              recordJson,
+              SetArgs.Builder.nx().ex(TTL)));
+    } catch (final JsonProcessingException e) {
+      // This should never happen
+      log.error("Failed to write push latency record JSON", e);
+    }
+  }
+
+  public void recordQueueRead(final UUID accountUuid, final long deviceId, final String userAgentString) {
+    takePushRecord(accountUuid, deviceId).thenAccept(pushRecord -> {
+      if (pushRecord != null) {
+        final Duration latency = Duration.between(pushRecord.getTimestamp(), Instant.now());
+
+        final List<Tag> tags = new ArrayList<>(2);
+
+        tags.add(UserAgentTagUtil.getPlatformTag(userAgentString));
+
+        if (pushRecord.getPushType() != null) {
+          tags.add(Tag.of("pushType", pushRecord.getPushType().name().toLowerCase()));
+        }
+
+        try {
+          final UserAgent userAgent = UserAgentUtil.parseUserAgentString(userAgentString);
+
+          final Set<Semver> instrumentedVersions =
+              dynamicConfigurationManager.getConfiguration().getPushLatencyConfiguration().getInstrumentedVersions()
+                  .getOrDefault(userAgent.getPlatform(), Collections.emptySet());
+
+          if (instrumentedVersions.contains(userAgent.getVersion())) {
+            tags.add(Tag.of("clientVersion", userAgent.getVersion().toString()));
+          }
+        } catch (UnrecognizedUserAgentException ignored) {
+        }
+
+        Metrics.timer(TIMER_NAME, tags).record(latency);
+      }
+    });
+  }
+
+  @VisibleForTesting
+  CompletableFuture<PushRecord> takePushRecord(final UUID accountUuid, final long deviceId) {
+    final String key = getFirstUnacknowledgedPushKey(accountUuid, deviceId);
+    final String legacyKey = getLegacyFirstUnacknowledgedPushKey(accountUuid, deviceId);
+
+    return redisCluster.withCluster(connection -> {
+      final CompletableFuture<PushRecord> getFuture = connection.async().get(key).toCompletableFuture()
+          .thenApply(recordJson -> {
+            if (StringUtils.isNotEmpty(recordJson)) {
+              try {
+                return SystemMapper.getMapper().readValue(recordJson, PushRecord.class);
+              } catch (JsonProcessingException e) {
+                return null;
+              }
+            } else {
+              return null;
             }
-        });
-    }
+          });
 
-    @VisibleForTesting
-    CompletableFuture<Long> getLatencyAndClearTimestamp(final UUID accountUuid, final long deviceId, final long currentTimeMillis) {
-        final String key = getFirstUnacknowledgedPushKey(accountUuid, deviceId);
+      final CompletableFuture<PushRecord> legacyGetFuture = connection.async().get(legacyKey).toCompletableFuture()
+          .thenApply(timestampString -> {
+            if (StringUtils.isNotEmpty(timestampString)) {
+              return new PushRecord(Instant.ofEpochMilli(Long.parseLong(timestampString)), null);
+            } else {
+              return null;
+            }
+          });
 
-        return redisCluster.withCluster(connection -> {
-            final RedisAdvancedClusterAsyncCommands<String, String> commands = connection.async();
+      final CompletableFuture<PushRecord> pushRecordFuture = getFuture.thenCombine(legacyGetFuture,
+          (a, b) -> a != null ? a : b);
 
-            final CompletableFuture<String> getFuture = commands.get(key).toCompletableFuture();
-            commands.del(key);
+      pushRecordFuture.whenComplete((record, cause) -> {
+        if (cause == null) {
+          connection.async().del(key, legacyKey);
+        }
+      });
 
-            return getFuture.thenApply(timestampString -> timestampString != null ? currentTimeMillis - Long.parseLong(timestampString, 10) : null);
-        });
-    }
+      return pushRecordFuture;
+    });
+  }
 
-    private static String getFirstUnacknowledgedPushKey(final UUID accountUuid, final long deviceId) {
-        return "push_latency::" + accountUuid.toString() + "::" + deviceId;
-    }
+  private static String getFirstUnacknowledgedPushKey(final UUID accountUuid, final long deviceId) {
+    return "push_latency::v2::" + accountUuid.toString() + "::" + deviceId;
+  }
+
+  @VisibleForTesting
+  static String getLegacyFirstUnacknowledgedPushKey(final UUID accountUuid, final long deviceId) {
+    return "push_latency::" + accountUuid.toString() + "::" + deviceId;
+  }
 }