Signal/Android
1
0
Fork 0
mirror of https://github.com/signalapp/Signal-Android.git synced 2025-12-22 12:08:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add local metrics for message processing.

Browse Source
This commit is contained in:
Clark
2023-07-11 18:00:01 -04:00
committed by Clark Chen
parent 8fc03a67b9
commit ac4d8679a1
10 changed files with 129 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
app/src/androidTest/java/org/thoughtcrime/securesms/messages/TimingMessageContentProcessorV2.kt
View File

@@ -3,6 +3,7 @@ package org.thoughtcrime.securesms.messages
import android.content.Context
import org.signal.core.util.logging.Log
import org.thoughtcrime.securesms.testing.LogPredicate
import org.thoughtcrime.securesms.util.SignalLocalMetrics
import org.whispersystems.signalservice.api.crypto.EnvelopeMetadata
import org.whispersystems.signalservice.internal.push.SignalServiceProtos
@@ -18,9 +19,9 @@ class TimingMessageContentProcessorV2(context: Context) : MessageContentProcesso
fun endTag(timestamp: Long) = "$timestamp end"
}
override fun process(envelope: SignalServiceProtos.Envelope, content: SignalServiceProtos.Content, metadata: EnvelopeMetadata, serverDeliveredTimestamp: Long, processingEarlyContent: Boolean) {
override fun process(envelope: SignalServiceProtos.Envelope, content: SignalServiceProtos.Content, metadata: EnvelopeMetadata, serverDeliveredTimestamp: Long, processingEarlyContent: Boolean, localMetric: SignalLocalMetrics.MessageReceive?) {
Log.d(TAG, startTag(envelope.timestamp))
super.process(envelope, content, metadata, serverDeliveredTimestamp, processingEarlyContent)
super.process(envelope, content, metadata, serverDeliveredTimestamp, processingEarlyContent, localMetric)
Log.d(TAG, endTag(envelope.timestamp))
}
}

2
app/src/main/java/org/thoughtcrime/securesms/database/LocalMetricsDatabase.kt
View File

@@ -136,7 +136,7 @@ class LocalMetricsDatabase private constructor(
put(EVENT_ID, event.eventId)
put(EVENT_NAME, event.eventName)
put(SPLIT_NAME, split.name)
put(DURATION, split.duration)
put(DURATION, event.timeunit.convert(split.duration, TimeUnit.NANOSECONDS))
}
)
}

7
app/src/main/java/org/thoughtcrime/securesms/database/model/LocalMetricsEvent.kt
View File

@@ -1,12 +1,15 @@
package org.thoughtcrime.securesms.database.model
import java.util.concurrent.TimeUnit
data class LocalMetricsEvent(
val createdAt: Long,
val eventId: String,
val eventName: String,
val splits: MutableList<LocalMetricsSplit>
val splits: MutableList<LocalMetricsSplit>,
val timeunit: TimeUnit
) {
override fun toString(): String {
return "[$eventName] total: ${splits.sumOf { it.duration }} | ${splits.map { it.toString() }.joinToString(", ")}"
return "[$eventName] total: ${splits.sumOf { timeunit.convert(it.duration, TimeUnit.NANOSECONDS) }} | ${splits.map { it.toString() }.joinToString(", ")}"
}
}

7
app/src/main/java/org/thoughtcrime/securesms/database/model/LocalMetricsSplit.kt
View File

@@ -1,10 +1,13 @@
package org.thoughtcrime.securesms.database.model
import java.util.concurrent.TimeUnit
data class LocalMetricsSplit(
val name: String,
val duration: Long
val duration: Long,
val timeunit: TimeUnit = TimeUnit.MILLISECONDS
) {
override fun toString(): String {
return "$name: $duration"
return "$name: ${timeunit.convert(duration, TimeUnit.NANOSECONDS)}"
}
}

5
app/src/main/java/org/thoughtcrime/securesms/jobs/PushProcessMessageJobV2.kt
View File

@@ -15,6 +15,7 @@ import org.thoughtcrime.securesms.messages.MessageDecryptor
import org.thoughtcrime.securesms.messages.SignalServiceProtoUtil.groupId
import org.thoughtcrime.securesms.recipients.RecipientId
import org.thoughtcrime.securesms.util.GroupUtil
import org.thoughtcrime.securesms.util.SignalLocalMetrics
import org.whispersystems.signalservice.api.crypto.EnvelopeMetadata
import org.whispersystems.signalservice.api.crypto.protos.CompleteMessage
import org.whispersystems.signalservice.api.groupsv2.NoCredentialForRedemptionTimeException
@@ -111,7 +112,7 @@ class PushProcessMessageJobV2 private constructor(
return QUEUE_PREFIX + recipientId.toQueueKey()
}
fun processOrDefer(messageProcessor: MessageContentProcessorV2, result: MessageDecryptor.Result.Success): PushProcessMessageJobV2? {
fun processOrDefer(messageProcessor: MessageContentProcessorV2, result: MessageDecryptor.Result.Success, localReceiveMetric: SignalLocalMetrics.MessageReceive): PushProcessMessageJobV2? {
val queueName: String
val groupContext = GroupUtil.getGroupContextIfPresent(result.content)
@@ -146,7 +147,7 @@ class PushProcessMessageJobV2 private constructor(
PushProcessMessageJobV2(builder.build(), result.envelope.toBuilder().clearContent().build(), result.content, result.metadata, result.serverDeliveredTimestamp)
} else {
try {
messageProcessor.process(result.envelope, result.content, result.metadata, result.serverDeliveredTimestamp)
messageProcessor.process(result.envelope, result.content, result.metadata, result.serverDeliveredTimestamp, localMetric = localReceiveMetric)
} catch (e: Exception) {
Log.e(TAG, "Failed to process message with timestamp ${result.envelope.timestamp}. Dropping.")
}

19
app/src/main/java/org/thoughtcrime/securesms/messages/DataMessageProcessor.kt
View File

@@ -89,6 +89,7 @@ import org.thoughtcrime.securesms.util.EarlyMessageCacheEntry
import org.thoughtcrime.securesms.util.LinkUtil
import org.thoughtcrime.securesms.util.MediaUtil
import org.thoughtcrime.securesms.util.MessageConstraintsUtil
import org.thoughtcrime.securesms.util.SignalLocalMetrics
import org.thoughtcrime.securesms.util.TextSecurePreferences
import org.thoughtcrime.securesms.util.isStory
import org.whispersystems.signalservice.api.crypto.EnvelopeMetadata
@@ -119,7 +120,8 @@ object DataMessageProcessor {
content: Content,
metadata: EnvelopeMetadata,
receivedTime: Long,
earlyMessageCacheEntry: EarlyMessageCacheEntry?
earlyMessageCacheEntry: EarlyMessageCacheEntry?,
localMetrics: SignalLocalMetrics.MessageReceive?
) {
val message: DataMessage = content.dataMessage
val groupSecretParams = if (message.hasGroupContext) GroupSecretParams.deriveFromMasterKey(message.groupV2.groupMasterKey) else null
@@ -131,6 +133,7 @@ object DataMessageProcessor {
if (groupProcessResult == MessageContentProcessorV2.Gv2PreProcessResult.IGNORE) {
return
}
localMetrics?.onGv2Processed()
}
var messageId: MessageId? = null
@@ -146,8 +149,8 @@ object DataMessageProcessor {
message.hasPayment() -> messageId = handlePayment(context, envelope, metadata, message, senderRecipient.id, receivedTime)
message.hasStoryContext() -> messageId = handleStoryReply(context, envelope, metadata, message, senderRecipient, groupId, receivedTime)
message.hasGiftBadge() -> messageId = handleGiftMessage(context, envelope, metadata, message, senderRecipient, threadRecipient.id, receivedTime)
message.isMediaMessage -> messageId = handleMediaMessage(context, envelope, metadata, message, senderRecipient, threadRecipient, groupId, receivedTime)
message.hasBody() -> messageId = handleTextMessage(context, envelope, metadata, message, senderRecipient, threadRecipient, groupId, receivedTime)
message.isMediaMessage -> messageId = handleMediaMessage(context, envelope, metadata, message, senderRecipient, threadRecipient, groupId, receivedTime, localMetrics)
message.hasBody() -> messageId = handleTextMessage(context, envelope, metadata, message, senderRecipient, threadRecipient, groupId, receivedTime, localMetrics)
message.hasGroupCallUpdate() -> handleGroupCallUpdateMessage(envelope, message, senderRecipient.id, groupId)
}
@@ -191,6 +194,8 @@ object DataMessageProcessor {
}
}
}
localMetrics?.onPostProcessComplete()
localMetrics?.complete(groupId != null)
}
private fun handleProfileKey(
@@ -822,7 +827,8 @@ object DataMessageProcessor {
senderRecipient: Recipient,
threadRecipient: Recipient,
groupId: GroupId.V2?,
receivedTime: Long
receivedTime: Long,
localMetrics: SignalLocalMetrics.MessageReceive?
): MessageId? {
log(envelope.timestamp, "Media message.")
@@ -871,6 +877,7 @@ object DataMessageProcessor {
} finally {
SignalDatabase.messages.endTransaction()
}
localMetrics?.onInsertedMediaMessage()
return if (insertResult != null) {
SignalDatabase.runPostSuccessfulTransaction {
@@ -912,7 +919,8 @@ object DataMessageProcessor {
senderRecipient: Recipient,
threadRecipient: Recipient,
groupId: GroupId.V2?,
receivedTime: Long
receivedTime: Long,
localMetrics: SignalLocalMetrics.MessageReceive?
): MessageId? {
log(envelope.timestamp, "Text message.")
@@ -936,6 +944,7 @@ object DataMessageProcessor {
)
val insertResult: InsertResult? = SignalDatabase.messages.insertMessageInbox(IncomingEncryptedMessage(textMessage, body)).orNull()
localMetrics?.onInsertedTextMessage()
return if (insertResult != null) {
ApplicationDependencies.getMessageNotifier().updateNotification(context, ConversationId.forConversation(insertResult.threadId))

4
app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageObserver.kt
View File

@@ -295,10 +295,12 @@ class IncomingMessageObserver(private val context: Application) {
}
private fun processMessage(bufferedProtocolStore: BufferedProtocolStore, envelope: SignalServiceProtos.Envelope, serverDeliveredTimestamp: Long): List<FollowUpOperation> {
val localReceiveMetric = SignalLocalMetrics.MessageReceive.start()
val result = MessageDecryptor.decrypt(context, bufferedProtocolStore, envelope, serverDeliveredTimestamp)
localReceiveMetric.onEnvelopeDecrypted()
when (result) {
is MessageDecryptor.Result.Success -> {
val job = PushProcessMessageJobV2.processOrDefer(messageContentProcessor, result)
val job = PushProcessMessageJobV2.processOrDefer(messageContentProcessor, result, localReceiveMetric)
if (job != null) {
return result.followUpOperations + FollowUpOperation { job }
}

24
app/src/main/java/org/thoughtcrime/securesms/messages/MessageContentProcessorV2.kt
View File

@@ -1,7 +1,6 @@
package org.thoughtcrime.securesms.messages
import android.content.Context
import org.signal.core.util.Stopwatch
import org.signal.core.util.logging.Log
import org.signal.core.util.orNull
import org.signal.libsignal.protocol.SignalProtocolAddress
@@ -41,6 +40,7 @@ import org.thoughtcrime.securesms.recipients.Recipient
import org.thoughtcrime.securesms.recipients.RecipientId
import org.thoughtcrime.securesms.util.EarlyMessageCacheEntry
import org.thoughtcrime.securesms.util.FeatureFlags
import org.thoughtcrime.securesms.util.SignalLocalMetrics
import org.thoughtcrime.securesms.util.TextSecurePreferences
import org.thoughtcrime.securesms.util.Util
import org.whispersystems.signalservice.api.crypto.EnvelopeMetadata
@@ -303,12 +303,10 @@ open class MessageContentProcessorV2(private val context: Context) {
* store or enqueue early content jobs if we detect this as being early, to avoid recursive scenarios.
*/
@JvmOverloads
open fun process(envelope: Envelope, content: Content, metadata: EnvelopeMetadata, serverDeliveredTimestamp: Long, processingEarlyContent: Boolean = false) {
val stopwatch = Stopwatch("process-content")
open fun process(envelope: Envelope, content: Content, metadata: EnvelopeMetadata, serverDeliveredTimestamp: Long, processingEarlyContent: Boolean = false, localMetric: SignalLocalMetrics.MessageReceive? = null) {
val senderRecipient = Recipient.externalPush(SignalServiceAddress(metadata.sourceServiceId, metadata.sourceE164))
handleMessage(senderRecipient, envelope, content, metadata, serverDeliveredTimestamp, processingEarlyContent, stopwatch)
handleMessage(senderRecipient, envelope, content, metadata, serverDeliveredTimestamp, processingEarlyContent, localMetric)
val earlyCacheEntries: List<EarlyMessageCacheEntry>? = ApplicationDependencies
.getEarlyMessageCache()
@@ -318,11 +316,9 @@ open class MessageContentProcessorV2(private val context: Context) {
if (!processingEarlyContent && earlyCacheEntries != null) {
log(envelope.timestamp, "Found " + earlyCacheEntries.size + " dependent item(s) that were retrieved earlier. Processing.")
for (entry in earlyCacheEntries) {
handleMessage(senderRecipient, entry.envelope, entry.content, entry.metadata, entry.serverDeliveredTimestamp, processingEarlyContent = true, stopwatch)
handleMessage(senderRecipient, entry.envelope, entry.content, entry.metadata, entry.serverDeliveredTimestamp, processingEarlyContent = true, localMetric = null)
}
stopwatch.split("early-entries")
}
stopwatch.stop(TAG)
}
private fun handleMessage(
@@ -332,7 +328,7 @@ open class MessageContentProcessorV2(private val context: Context) {
metadata: EnvelopeMetadata,
serverDeliveredTimestamp: Long,
processingEarlyContent: Boolean,
stopwatch: Stopwatch
localMetric: SignalLocalMetrics.MessageReceive?
) {
val threadRecipient = getMessageDestination(content, senderRecipient)
@@ -345,7 +341,7 @@ open class MessageContentProcessorV2(private val context: Context) {
val receivedTime: Long = handlePendingRetry(pending, envelope.timestamp, threadRecipient)
log(envelope.timestamp, "Beginning message processing. Sender: " + formatSender(senderRecipient.id, metadata.sourceServiceId, metadata.sourceDeviceId))
stopwatch.split("pre-process")
localMetric?.onPreProcessComplete()
when {
content.hasDataMessage() -> {
DataMessageProcessor.process(
@@ -356,9 +352,9 @@ open class MessageContentProcessorV2(private val context: Context) {
content,
metadata,
receivedTime,
if (processingEarlyContent) null else EarlyMessageCacheEntry(envelope, content, metadata, serverDeliveredTimestamp)
if (processingEarlyContent) null else EarlyMessageCacheEntry(envelope, content, metadata, serverDeliveredTimestamp),
localMetric
)
stopwatch.split("data-message")
}
content.hasSyncMessage() -> {
TextSecurePreferences.setMultiDevice(context, true)
@@ -371,7 +367,6 @@ open class MessageContentProcessorV2(private val context: Context) {
metadata,
if (processingEarlyContent) null else EarlyMessageCacheEntry(envelope, content, metadata, serverDeliveredTimestamp)
)
stopwatch.split("sync-message")
}
content.hasCallMessage() -> {
log(envelope.timestamp, "Got call message...")
@@ -395,11 +390,9 @@ open class MessageContentProcessorV2(private val context: Context) {
metadata,
if (processingEarlyContent) null else EarlyMessageCacheEntry(envelope, content, metadata, serverDeliveredTimestamp)
)
stopwatch.split("receipt-message")
}
content.hasTypingMessage() -> {
handleTypingMessage(envelope, metadata, content.typingMessage, senderRecipient)
stopwatch.split("typing-message")
}
content.hasStoryMessage() -> {
StoryMessageProcessor.process(
@@ -409,7 +402,6 @@ open class MessageContentProcessorV2(private val context: Context) {
senderRecipient,
threadRecipient
)
stopwatch.split("story-message")
}
content.hasDecryptionErrorMessage() -> {
handleRetryReceipt(envelope, metadata, content.decryptionErrorMessage!!.toDecryptionErrorMessage(metadata), senderRecipient)

15
app/src/main/java/org/thoughtcrime/securesms/util/LocalMetrics.kt
View File

@@ -1,5 +1,6 @@
package org.thoughtcrime.securesms.util
import android.os.SystemClock
import org.signal.core.util.ThreadUtil
import org.signal.core.util.concurrent.SignalExecutors
import org.signal.core.util.logging.Log
@@ -8,6 +9,7 @@ import org.thoughtcrime.securesms.database.model.LocalMetricsEvent
import org.thoughtcrime.securesms.database.model.LocalMetricsSplit
import org.thoughtcrime.securesms.dependencies.ApplicationDependencies
import java.util.concurrent.Executor
import java.util.concurrent.TimeUnit
/**
* A class for keeping track of local-only metrics.
@@ -44,15 +46,17 @@ object LocalMetrics {
* @param id A constant that should be unique to this *specific event*. You'll use this same id when calling [split] and [end]. e.g. "message-send-1234"
* @param name The name of the event. Does not need to be unique. e.g. "message-send"
*/
fun start(id: String, name: String) {
val time = System.currentTimeMillis()
@JvmOverloads
fun start(id: String, name: String, timeunit: TimeUnit = TimeUnit.MILLISECONDS) {
val time = SystemClock.elapsedRealtimeNanos()
executor.execute {
eventsById[id] = LocalMetricsEvent(
createdAt = System.currentTimeMillis(),
eventId = id,
eventName = name,
splits = mutableListOf()
splits = mutableListOf(),
timeunit = timeunit
)
lastSplitTimeById[id] = time
}
@@ -65,13 +69,14 @@ object LocalMetrics {
* If an event with the provided ID does not exist, this is effectively a no-op.
*/
fun split(id: String, split: String) {
val time = System.currentTimeMillis()
val time = SystemClock.elapsedRealtimeNanos()
executor.execute {
val lastTime: Long? = lastSplitTimeById[id]
val splitDoesNotExist: Boolean = eventsById[id]?.splits?.none { it.name == split } ?: true
if (lastTime != null && splitDoesNotExist) {
eventsById[id]?.splits?.add(LocalMetricsSplit(split, time - lastTime))
val event = eventsById[id]
event?.splits?.add(LocalMetricsSplit(split, time - lastTime, event.timeunit))
lastSplitTimeById[id] = time
}
}

77
app/src/main/java/org/thoughtcrime/securesms/util/SignalLocalMetrics.java
View File

@@ -7,6 +7,7 @@ import androidx.annotation.Nullable;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.TimeUnit;
/**
* A nice interface for {@link LocalMetrics} that gives us a place to define string constants and nicer method names.
@@ -357,4 +358,80 @@ public final class SignalLocalMetrics {
}
}
}
public static final class MessageReceive {
private static final String NAME_GROUP = "group-message-receive";
private static final String NAME_INDIVIDUAL = "individual-message-receive";
private static final String SPLIT_DECRYPTION = "decryption";
private static final String SPLIT_PRE_PROCESS = "pre-process";
private static final String SPLIT_GROUPS_PROCESSING = "groups-v2";
private static final String SPLIT_DB_INSERT_MEDIA = "media-insert";
private static final String SPLIT_DB_INSERT_TEXT = "text-insert";
private static final String SPLIT_POST_PROCESS = "post-process";
private boolean insertedToDb = false;
private final String individualMetricId;
private final String groupMetricId;
public static MessageReceive start() {
return new MessageReceive();
}
private MessageReceive() {
long time = System.currentTimeMillis();
individualMetricId = NAME_INDIVIDUAL + time;
groupMetricId = NAME_GROUP + time;
LocalMetrics.getInstance().start(individualMetricId, NAME_INDIVIDUAL, TimeUnit.MICROSECONDS);
LocalMetrics.getInstance().start(groupMetricId, NAME_GROUP, TimeUnit.MICROSECONDS);
}
public void onEnvelopeDecrypted() {
split(SPLIT_DECRYPTION);
}
public void onPreProcessComplete() {
split(SPLIT_PRE_PROCESS);
}
public void onInsertedMediaMessage() {
split(SPLIT_DB_INSERT_MEDIA);
insertedToDb = true;
}
public void onInsertedTextMessage() {
split(SPLIT_DB_INSERT_TEXT);
insertedToDb = true;
}
public void onPostProcessComplete() {
split(SPLIT_POST_PROCESS);
}
public void onGv2Processed() {
split(SPLIT_GROUPS_PROCESSING);
}
private void split(String name) {
LocalMetrics.getInstance().split(groupMetricId, name);
LocalMetrics.getInstance().split(individualMetricId, name);
}
public void complete(boolean isGroup) {
if (!insertedToDb) {
LocalMetrics.getInstance().cancel(groupMetricId);
LocalMetrics.getInstance().cancel(individualMetricId);
return;
}
if (isGroup) {
LocalMetrics.getInstance().cancel(individualMetricId);
LocalMetrics.getInstance().end(groupMetricId);
} else {
LocalMetrics.getInstance().cancel(groupMetricId);
LocalMetrics.getInstance().end(individualMetricId);
}
}
}
}