Add metric for tracking S3 endpoint calls by RPC method

Author: regexident

src/proxy.rs

@@ -115,6 +115,7 @@ impl<T: S3 + Send + Sync> S3 for S3CachingProxy<T> {
         &self,
         req: S3Request<GetObjectInput>,
     ) -> S3Result<S3Response<GetObjectOutput>> {
+        telemetry::record_endpoint_call("GetObject");
         let start = Instant::now();
         let method = req.method.to_string();
         let scheme = req.uri.scheme_str().map(str::to_owned);
@@ -372,6 +373,7 @@ impl<T: S3 + Send + Sync> S3 for S3CachingProxy<T> {
         &self,
         req: S3Request<PutObjectInput>,
     ) -> S3Result<S3Response<PutObjectOutput>> {
+        telemetry::record_endpoint_call("PutObject");
         let bucket = req.input.bucket.clone();
         let key = req.input.key.clone();
 
@@ -400,6 +402,7 @@ impl<T: S3 + Send + Sync> S3 for S3CachingProxy<T> {
         &self,
         req: S3Request<DeleteObjectInput>,
     ) -> S3Result<S3Response<DeleteObjectOutput>> {
+        telemetry::record_endpoint_call("DeleteObject");
         let bucket = req.input.bucket.clone();
         let key = req.input.key.clone();
 
@@ -428,6 +431,7 @@ impl<T: S3 + Send + Sync> S3 for S3CachingProxy<T> {
         &self,
         req: S3Request<DeleteObjectsInput>,
     ) -> S3Result<S3Response<DeleteObjectsOutput>> {
+        telemetry::record_endpoint_call("DeleteObjects");
         let bucket = req.input.bucket.clone();
         let keys: Vec<String> = req
             .input
@@ -464,6 +468,7 @@ impl<T: S3 + Send + Sync> S3 for S3CachingProxy<T> {
         &self,
         req: S3Request<CopyObjectInput>,
     ) -> S3Result<S3Response<CopyObjectOutput>> {
+        telemetry::record_endpoint_call("CopyObject");
         let dest_bucket = req.input.bucket.clone();
         let dest_key = req.input.key.clone();
 
@@ -492,13 +497,15 @@ impl<T: S3 + Send + Sync> S3 for S3CachingProxy<T> {
         &self,
         req: S3Request<AbortMultipartUploadInput>,
     ) -> S3Result<S3Response<AbortMultipartUploadOutput>> {
+        telemetry::record_endpoint_call("AbortMultipartUpload");
         self.inner.abort_multipart_upload(req).await
     }
 
     async fn complete_multipart_upload(
         &self,
         req: S3Request<CompleteMultipartUploadInput>,
     ) -> S3Result<S3Response<CompleteMultipartUploadOutput>> {
+        telemetry::record_endpoint_call("CompleteMultipartUpload");
         let bucket = req.input.bucket.clone();
         let key = req.input.key.clone();
 
@@ -527,90 +534,103 @@ impl<T: S3 + Send + Sync> S3 for S3CachingProxy<T> {
         &self,
         req: S3Request<CreateBucketInput>,
     ) -> S3Result<S3Response<CreateBucketOutput>> {
+        telemetry::record_endpoint_call("CreateBucket");
         self.inner.create_bucket(req).await
     }
 
     async fn create_multipart_upload(
         &self,
         req: S3Request<CreateMultipartUploadInput>,
     ) -> S3Result<S3Response<CreateMultipartUploadOutput>> {
+        telemetry::record_endpoint_call("CreateMultipartUpload");
         self.inner.create_multipart_upload(req).await
     }
 
     async fn delete_bucket(
         &self,
         req: S3Request<DeleteBucketInput>,
     ) -> S3Result<S3Response<DeleteBucketOutput>> {
+        telemetry::record_endpoint_call("DeleteBucket");
         self.inner.delete_bucket(req).await
     }
 
     async fn get_bucket_location(
         &self,
         req: S3Request<GetBucketLocationInput>,
     ) -> S3Result<S3Response<GetBucketLocationOutput>> {
+        telemetry::record_endpoint_call("GetBucketLocation");
         self.inner.get_bucket_location(req).await
     }
 
     async fn head_bucket(
         &self,
         req: S3Request<HeadBucketInput>,
     ) -> S3Result<S3Response<HeadBucketOutput>> {
+        telemetry::record_endpoint_call("HeadBucket");
         self.inner.head_bucket(req).await
     }
 
     async fn head_object(
         &self,
         req: S3Request<HeadObjectInput>,
     ) -> S3Result<S3Response<HeadObjectOutput>> {
+        telemetry::record_endpoint_call("HeadObject");
         self.inner.head_object(req).await
     }
 
     async fn list_buckets(
         &self,
         req: S3Request<ListBucketsInput>,
     ) -> S3Result<S3Response<ListBucketsOutput>> {
+        telemetry::record_endpoint_call("ListBuckets");
         self.inner.list_buckets(req).await
     }
 
     async fn list_multipart_uploads(
         &self,
         req: S3Request<ListMultipartUploadsInput>,
     ) -> S3Result<S3Response<ListMultipartUploadsOutput>> {
+        telemetry::record_endpoint_call("ListMultipartUploads");
         self.inner.list_multipart_uploads(req).await
     }
 
     async fn list_objects(
         &self,
         req: S3Request<ListObjectsInput>,
     ) -> S3Result<S3Response<ListObjectsOutput>> {
+        telemetry::record_endpoint_call("ListObjects");
         self.inner.list_objects(req).await
     }
 
     async fn list_objects_v2(
         &self,
         req: S3Request<ListObjectsV2Input>,
     ) -> S3Result<S3Response<ListObjectsV2Output>> {
+        telemetry::record_endpoint_call("ListObjectsV2");
         self.inner.list_objects_v2(req).await
     }
 
     async fn list_parts(
         &self,
         req: S3Request<ListPartsInput>,
     ) -> S3Result<S3Response<ListPartsOutput>> {
+        telemetry::record_endpoint_call("ListParts");
         self.inner.list_parts(req).await
     }
 
     async fn upload_part(
         &self,
         req: S3Request<UploadPartInput>,
     ) -> S3Result<S3Response<UploadPartOutput>> {
+        telemetry::record_endpoint_call("UploadPart");
         self.inner.upload_part(req).await
     }
 
     async fn upload_part_copy(
         &self,
         req: S3Request<UploadPartCopyInput>,
     ) -> S3Result<S3Response<UploadPartCopyOutput>> {
+        telemetry::record_endpoint_call("UploadPartCopy");
         self.inner.upload_part_copy(req).await
     }
 }

src/telemetry.rs

@@ -5,7 +5,7 @@ use opentelemetry::KeyValue;
 use opentelemetry::metrics::{Counter, Gauge, Histogram};
 use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
 use opentelemetry_otlp::{Compression, WithExportConfig, WithTonicConfig};
-use prometheus::{HistogramOpts, IntCounter, IntGauge, Registry};
+use prometheus::{HistogramOpts, IntCounter, IntCounterVec, IntGauge, Opts, Registry};
 use tracing::{error, info};
 use tracing_subscriber::{EnvFilter, layer::SubscriberExt, util::SubscriberInitExt};
 
@@ -681,6 +681,35 @@ pub(crate) fn record_request_duration(data: RequestDuration) {
     REQUEST_DURATION_MS.record(milliseconds, &attributes);
 }
 
+// MARK: - Endpoint Calls
+
+pub(crate) fn record_endpoint_call(method: &'static str) {
+    static ENDPOINT_TOTAL: LazyLock<Counter<u64>> = LazyLock::new(|| {
+        opentelemetry::global::meter(CARGO_CRATE_NAME)
+            .u64_counter("s3_cache.endpoint_call_total")
+            .with_description("Number of S3 endpoint method calls")
+            .build()
+    });
+
+    static PROM_ENDPOINT_TOTAL: LazyLock<IntCounterVec> = LazyLock::new(|| {
+        let counter = IntCounterVec::new(
+            Opts::new(
+                "s3_cache_endpoint_call_total",
+                "Number of S3 endpoint method calls",
+            ),
+            &["rpc_method"],
+        )
+        .unwrap();
+        PROMETHEUS_REGISTRY
+            .register(Box::new(counter.clone()))
+            .unwrap();
+        counter
+    });
+
+    PROM_ENDPOINT_TOTAL.with_label_values(&[method]).inc();
+    ENDPOINT_TOTAL.add(1, &[KeyValue::new("rpc.method", method)]);
+}
+
 // MARK: Response Body Sizes
 
 /// Attributes based on: https://opentelemetry.io/docs/specs/semconv/http/http-metrics/#http-server