EmitAPIOutgoingEnvelopes per API method (#1765)

Label xmtp_api_outgoing_envelopes_total counter by API method
Changes apiOutgoingEnvelopesTotal in pkg/metrics/api.go from a plain Counter to a CounterVec with a method label, so envelope counts are broken down per API procedure.
Updates EmitAPIOutgoingEnvelopes (renamed from EmitApiOutgoingEnvelopes) to accept a method string parameter and call WithLabelValues(method) before incrementing.
Updates all call sites in pkg/api/message/service.go and pkg/api/message/subscribe_topics.go to pass the procedure from the request or stream spec.
Also renames EmitApiWaitForGatewayPublish and EmitApiStagedEnvelopeProcessingDelay to follow the EmitAPI… naming convention.
Behavioral Change: existing dashboards or alerts querying xmtp_api_outgoing_envelopes_total without a method label filter will need to be updated to aggregate across the new label dimension.

Author: fbac

pkg/api/message/publish_worker.go

@@ -117,7 +117,7 @@ func (p *publishWorker) start() {
 					time.Sleep(p.sleepOnFailureTime)
 				}
 				p.lastProcessed.Store(stagedEnv.ID)
-				metrics.EmitApiStagedEnvelopeProcessingDelay(time.Since(stagedEnv.OriginatorTime))
+				metrics.EmitAPIStagedEnvelopeProcessingDelay(time.Since(stagedEnv.OriginatorTime))
 			}
 		}
 	}

pkg/api/message/service.go

@@ -302,7 +302,7 @@ func (s *Service) sendEnvelopes(
 			)
 		}
 
-		metrics.EmitApiOutgoingEnvelopes(len(batch))
+		metrics.EmitAPIOutgoingEnvelopes(stream.Conn().Spec().Procedure, len(batch))
 
 		batchWireBytes = 0
 		batch = batch[:0]
@@ -434,7 +434,7 @@ func (s *Service) QueryEnvelopes(
 		response.Msg.Envelopes = append(response.Msg.Envelopes, originatorEnv)
 	}
 
-	metrics.EmitApiOutgoingEnvelopes(len(response.Msg.GetEnvelopes()))
+	metrics.EmitAPIOutgoingEnvelopes(req.Spec().Procedure, len(response.Msg.GetEnvelopes()))
 
 	return response, nil
 }
@@ -975,7 +975,7 @@ func (s *Service) GetNewestEnvelope(
 		sent++
 	}
 
-	metrics.EmitApiOutgoingEnvelopes(sent)
+	metrics.EmitAPIOutgoingEnvelopes(req.Spec().Procedure, sent)
 
 	return response, nil
 }
@@ -1149,7 +1149,7 @@ func (s *Service) waitForGatewayPublish(
 
 	startTime := time.Now()
 	defer func() {
-		metrics.EmitApiWaitForGatewayPublish(time.Since(startTime))
+		metrics.EmitAPIWaitForGatewayPublish(time.Since(startTime))
 	}()
 
 	timeout := time.After(30 * time.Second)

pkg/api/message/subscribe_topics.go

@@ -348,7 +348,7 @@ func (s *Service) sendTopicEnvelopes(
 		)
 	}
 
-	metrics.EmitApiOutgoingEnvelopes(len(envs))
+	metrics.EmitAPIOutgoingEnvelopes(stream.Conn().Spec().Procedure, len(envs))
 
 	return nil
 }

pkg/metrics/api.go

@@ -96,7 +96,7 @@ var apiWaitForGatewayPublish = prometheus.NewHistogram(
 	},
 )
 
-func EmitApiWaitForGatewayPublish(
+func EmitAPIWaitForGatewayPublish(
 	duration time.Duration,
 ) {
 	apiWaitForGatewayPublish.Observe(duration.Seconds())
@@ -109,17 +109,18 @@ var apiStagedEnvelopeProcessingDelay = prometheus.NewHistogram(
 	},
 )
 
-func EmitApiStagedEnvelopeProcessingDelay(duration time.Duration) {
+func EmitAPIStagedEnvelopeProcessingDelay(duration time.Duration) {
 	apiStagedEnvelopeProcessingDelay.Observe(duration.Seconds())
 }
 
-var apiOutgoingEnvelopesTotal = prometheus.NewCounter(
+var apiOutgoingEnvelopesTotal = prometheus.NewCounterVec(
 	prometheus.CounterOpts{
 		Name: "xmtp_api_outgoing_envelopes_total",
 		Help: "Total number of envelopes delivered to clients via subscribe and query APIs",
 	},
+	[]string{"method"},
 )
 
-func EmitApiOutgoingEnvelopes(n int) {
-	apiOutgoingEnvelopesTotal.Add(float64(n))
+func EmitAPIOutgoingEnvelopes(method string, n int) {
+	apiOutgoingEnvelopesTotal.WithLabelValues(method).Add(float64(n))
 }