Improve first processing lag metrics

GitOrigin-RevId: 672b4b5ad56b38b0d239aecd59f3adb300db0803

Author: mateuszmrozewski

misk-aws2-sqs/api/misk-aws2-sqs.api

@@ -84,9 +84,8 @@ public final class misk/aws2/sqs/jobqueue/SqsMetrics {
 	public final fun getJobsDeadLettered ()Lio/prometheus/client/Counter;
 	public final fun getJobsEnqueued ()Lio/prometheus/client/Counter;
 	public final fun getJobsReceived ()Lio/prometheus/client/Counter;
+	public final fun getQueueFirstProcessingLag ()Lio/prometheus/client/Histogram;
 	public final fun getQueueProcessingLag ()Lio/prometheus/client/Histogram;
-	public final fun getSqsApproxNumberOfMessages ()Lio/prometheus/client/Gauge;
-	public final fun getSqsApproxNumberOfMessagesNotVisible ()Lio/prometheus/client/Gauge;
 	public final fun getSqsDeleteTime ()Lio/prometheus/client/Histogram;
 	public final fun getSqsReceiveTime ()Lio/prometheus/client/Histogram;
 	public final fun getSqsSendTime ()Lio/prometheus/client/Histogram;
@@ -99,7 +98,6 @@ public final class misk/aws2/sqs/jobqueue/StaticDeadLetterQueueProvider : misk/a
 }
 
 public final class misk/aws2/sqs/jobqueue/Subscriber {
-	public static final field Companion Lmisk/aws2/sqs/jobqueue/Subscriber$Companion;
 	public fun <init> (Lmisk/jobqueue/QueueName;Lmisk/aws2/sqs/jobqueue/config/SqsQueueConfig;Lmisk/jobqueue/QueueName;Lmisk/jobqueue/v2/JobHandler;Lkotlinx/coroutines/channels/Channel;Lsoftware/amazon/awssdk/services/sqs/SqsAsyncClient;Lmisk/aws2/sqs/jobqueue/QueueResolver;Lmisk/aws2/sqs/jobqueue/SqsMetrics;Lcom/squareup/moshi/Moshi;Ljava/time/Clock;Lio/opentracing/Tracer;Lmisk/aws2/sqs/jobqueue/VisibilityTimeoutCalculator;)V
 	public final fun getChannel ()Lkotlinx/coroutines/channels/Channel;
 	public final fun getClient ()Lsoftware/amazon/awssdk/services/sqs/SqsAsyncClient;
@@ -117,9 +115,6 @@ public final class misk/aws2/sqs/jobqueue/Subscriber {
 	public final fun run (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 
-public final class misk/aws2/sqs/jobqueue/Subscriber$Companion {
-}
-
 public final class misk/aws2/sqs/jobqueue/SubscriptionService : com/google/common/util/concurrent/AbstractIdleService {
 	public static final field Companion Lmisk/aws2/sqs/jobqueue/SubscriptionService$Companion;
 	public fun <init> (Lmisk/aws2/sqs/jobqueue/SqsJobConsumer;Ljava/util/Map;Lmisk/aws2/sqs/jobqueue/config/SqsConfig;)V

misk-aws2-sqs/src/main/kotlin/misk/aws2/sqs/jobqueue/SqsMetrics.kt

@@ -78,6 +78,12 @@ class SqsMetrics @Inject internal constructor(metrics: Metrics) {
     listOf("QueueName")
   )
 
+  val queueFirstProcessingLag = metrics.histogram(
+    "jobs_sqs_first_processing_lag_v2",
+    "time it took to receive a job from when it was enqueued",
+    listOf("QueueName")
+  )
+
   val queueProcessingLag = metrics.histogram(
     "jobs_sqs_processing_lag_v2",
     "time it took to receive a job from when it was enqueued",
@@ -89,20 +95,4 @@ class SqsMetrics @Inject internal constructor(metrics: Metrics) {
     "time a job spent it the channel between receiver and handler",
     listOf("QueueName")
   )
-
-  val sqsApproxNumberOfMessages = metrics.gauge(
-    "ApproximateNumberOfMessagesVisible",
-    "the approximate number of messages available for retrieval from SQS",
-    // `namespace` and `stat` is to emulate the CloudWatch metrics.
-    listOf("namespace", "stat", "QueueName")
-  )
-
-  val sqsApproxNumberOfMessagesNotVisible = metrics.gauge(
-    "ApproximateNumberOfMessagesNotVisible",
-    "the  approximate number of messages that are in flight. Messages are considered to " +
-      "be in flight if they have been sent to a client but have not yet been deleted or have " +
-      "not yet reached the end of their visibility window.",
-    // `namespace` and `stat` is to emulate the CloudWatch metrics.
-    listOf("namespace", "stat", "QueueName")
-  )
 }

misk-aws2-sqs/src/main/kotlin/misk/aws2/sqs/jobqueue/Subscriber.kt

@@ -166,10 +166,14 @@ class Subscriber(
 
       sqsMetrics.jobsReceived.labels(queueName.value).inc(response.messages().size.toDouble())
       response.messages().forEach { message ->
-        message.messageAttributes()[SQS_ATTRIBUTE_SENT_TIMESTAMP]?.let {
-          val sentTimestamp = it.stringValue().toLong()
-          val processingLag = clock.instant().minusMillis(sentTimestamp).toEpochMilli()
-          sqsMetrics.queueProcessingLag.labels(queueName.value).observe(processingLag.toDouble())
+        message.attributes()[MessageSystemAttributeName.SENT_TIMESTAMP]?.let {
+          val sentTimestamp = it.toLong()
+          val processingLag = clock.instant().minusMillis(sentTimestamp).toEpochMilli().toDouble()
+          val receiveCounter = message.attributes()[MessageSystemAttributeName.APPROXIMATE_RECEIVE_COUNT]?.toInt()
+          if (receiveCounter == 1) {
+            sqsMetrics.queueFirstProcessingLag.labels(queueName.value).observe(processingLag)
+          }
+          sqsMetrics.queueProcessingLag.labels(queueName.value).observe(processingLag)
         }
         val publishToChannelTimestamp = clock.instant()
         emit(
@@ -188,16 +192,12 @@ class Subscriber(
   private fun fetchMessages(queueUrl: String): CompletableFuture<ReceiveMessageResponse> {
     val request = ReceiveMessageRequest.builder()
       .queueUrl(queueUrl)
-      .messageAttributeNames("All")
+      .messageAttributeNames(MessageSystemAttributeName.ALL.name)
       .messageSystemAttributeNames(MessageSystemAttributeName.ALL)
       .maxNumberOfMessages(queueConfig.max_number_of_messages)
       .waitTimeSeconds(queueConfig.wait_timeout)
       .visibilityTimeout(queueConfig.visibility_timeout)
       .build()
     return client.receiveMessage(request)
   }
-
-  companion object {
-    private const val SQS_ATTRIBUTE_SENT_TIMESTAMP = "SentTimestamp"
-  }
 }