use access layer for observing enpoint call telemetry

Author: AlexanderKlump

src/access.rs

@@ -0,0 +1,18 @@
+use s3s::S3Result;
+use s3s::access::{S3Access, S3AccessContext};
+
+use crate::telemetry;
+
+pub(crate) struct TelemetryAccess;
+
+#[async_trait::async_trait]
+impl S3Access for TelemetryAccess {
+    async fn check(&self, cx: &mut S3AccessContext<'_>) -> S3Result<()> {
+        telemetry::record_endpoint_call(cx.s3_op().name());
+
+        match cx.credentials() {
+            Some(_) => Ok(()),
+            None => Err(s3s::s3_error!(AccessDenied, "Signature is required")),
+        }
+    }
+}

src/lib.rs

@@ -36,6 +36,7 @@ pub use self::proxy::{S3CachingProxy, range_to_string};
 pub use self::s3_cache::{CacheKey, CachedObject, S3Cache};
 pub use self::statistics::UniqueRequestedObjectsStatisticsTracker;
 
+mod access;
 mod auth;
 mod config;
 mod error;
@@ -148,6 +149,7 @@ where
     let service = {
         let mut b = S3ServiceBuilder::new(caching_proxy);
         b.set_auth(auth::create_auth(&config));
+        b.set_access(access::TelemetryAccess);
         let upstream_health_endpoint = Url::parse(&config.upstream_endpoint)
             .unwrap()
             .join("/minio/health/ready")

src/proxy.rs

@@ -115,7 +115,6 @@ 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);
@@ -373,7 +372,6 @@ 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();
 
@@ -402,7 +400,6 @@ 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();
 
@@ -431,7 +428,6 @@ 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
@@ -468,7 +464,6 @@ 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();
 
@@ -497,15 +492,13 @@ 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();
 
@@ -534,103 +527,90 @@ 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

@@ -683,7 +683,7 @@ pub(crate) fn record_request_duration(data: RequestDuration) {
 
 // MARK: - Endpoint Calls
 
-pub(crate) fn record_endpoint_call(method: &'static str) {
+pub(crate) fn record_endpoint_call(method: &str) {
     static ENDPOINT_TOTAL: LazyLock<Counter<u64>> = LazyLock::new(|| {
         opentelemetry::global::meter(CARGO_CRATE_NAME)
             .u64_counter("s3_cache.endpoint_call_total")
@@ -707,7 +707,7 @@ pub(crate) fn record_endpoint_call(method: &'static str) {
     });
 
     PROM_ENDPOINT_TOTAL.with_label_values(&[method]).inc();
-    ENDPOINT_TOTAL.add(1, &[KeyValue::new("rpc.method", method)]);
+    ENDPOINT_TOTAL.add(1, &[KeyValue::new("rpc.method", method.to_owned())]);
 }
 
 // MARK: Response Body Sizes