MirrorMirror
/
Signal-Server
mirror of https://github.com/signalapp/Signal-Server.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Record a distribution of initial queue depths broken down by platform.

This commit is contained in:
Jon Chambers 2021-02-08 15:26:28 -05:00 committed by Jon Chambers
parent 9c6ce08db0
commit bee9b61831
1 changed files with 9 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
service/src/main/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnection.java
View File

@ -46,6 +46,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Semaphore; import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.LongAdder;
import static com.codahale.metrics.MetricRegistry.name; import static com.codahale.metrics.MetricRegistry.name;
import static org.whispersystems.textsecuregcm.entities.MessageProtos.Envelope; import static org.whispersystems.textsecuregcm.entities.MessageProtos.Envelope;
@ -64,10 +65,11 @@ public class WebSocketConnection implements MessageAvailabilityListener, Displac
private static final Meter sendFailuresMeter = metricRegistry.meter(name(WebSocketConnection.class, "send_failures")); private static final Meter sendFailuresMeter = metricRegistry.meter(name(WebSocketConnection.class, "send_failures"));
private static final Meter discardedMessagesMeter = metricRegistry.meter(name(WebSocketConnection.class, "discardedMessages")); private static final Meter discardedMessagesMeter = metricRegistry.meter(name(WebSocketConnection.class, "discardedMessages"));
private static final String DISPLACEMENT_COUNTER_NAME = name(WebSocketConnection.class, "displacement"); private static final String INITIAL_QUEUE_LENGTH_DISTRIBUTION_NAME = name(WebSocketConnection.class, "initialQueueLength");
private static final String NON_SUCCESS_RESPONSE_COUNTER_NAME = name(WebSocketConnection.class, "clientNonSuccessResponse"); private static final String DISPLACEMENT_COUNTER_NAME = name(WebSocketConnection.class, "displacement");
private static final String STATUS_CODE_TAG = "status"; private static final String NON_SUCCESS_RESPONSE_COUNTER_NAME = name(WebSocketConnection.class, "clientNonSuccessResponse");
private static final String STATUS_MESSAGE_TAG = "message"; private static final String STATUS_CODE_TAG = "status";
private static final String STATUS_MESSAGE_TAG = "message";
@VisibleForTesting @VisibleForTesting
static final int MAX_DESKTOP_MESSAGE_SIZE = 1024 * 1024; static final int MAX_DESKTOP_MESSAGE_SIZE = 1024 * 1024;
@ -86,6 +88,7 @@ public class WebSocketConnection implements MessageAvailabilityListener, Displac
private final Semaphore processStoredMessagesSemaphore = new Semaphore(1); private final Semaphore processStoredMessagesSemaphore = new Semaphore(1);
private final AtomicReference<StoredMessageState> storedMessageState = new AtomicReference<>(StoredMessageState.PERSISTED_NEW_MESSAGES_AVAILABLE); private final AtomicReference<StoredMessageState> storedMessageState = new AtomicReference<>(StoredMessageState.PERSISTED_NEW_MESSAGES_AVAILABLE);
private final AtomicBoolean sentInitialQueueEmptyMessage = new AtomicBoolean(false); private final AtomicBoolean sentInitialQueueEmptyMessage = new AtomicBoolean(false);
private final LongAdder sentMessageCounter = new LongAdder();
private enum StoredMessageState { private enum StoredMessageState {
EMPTY, EMPTY,
@ -128,6 +131,7 @@ public class WebSocketConnection implements MessageAvailabilityListener, Displac
final Optional<byte[]> body = Optional.ofNullable(message.toByteArray()); final Optional<byte[]> body = Optional.ofNullable(message.toByteArray());
sendMessageMeter.mark(); sendMessageMeter.mark();
sentMessageCounter.increment();
bytesSentMeter.mark(body.map(bytes -> bytes.length).orElse(0)); bytesSentMeter.mark(body.map(bytes -> bytes.length).orElse(0));
// X-Signal-Key: false must be sent until Android stops assuming it missing means true // X-Signal-Key: false must be sent until Android stops assuming it missing means true
@ -193,6 +197,7 @@ public class WebSocketConnection implements MessageAvailabilityListener, Displac
queueClearedFuture.whenComplete((v, cause) -> { queueClearedFuture.whenComplete((v, cause) -> {
if (cause == null && sentInitialQueueEmptyMessage.compareAndSet(false, true)) { if (cause == null && sentInitialQueueEmptyMessage.compareAndSet(false, true)) {
Metrics.summary(INITIAL_QUEUE_LENGTH_DISTRIBUTION_NAME, List.of(UserAgentTagUtil.getPlatformTag(client.getUserAgent()))).record(sentMessageCounter.sum());
client.sendRequest("PUT", "/api/v1/queue/empty", Collections.singletonList(TimestampHeaderUtil.getTimestampHeader()), Optional.empty()); client.sendRequest("PUT", "/api/v1/queue/empty", Collections.singletonList(TimestampHeaderUtil.getTimestampHeader()), Optional.empty());
} }