refactor(storage): refresh crate across drivers and metrics
zhenyi committed
Modified lib/storage/lib.rs lib/storage/lib.rs +1 −0 View lib/storage/lib.rs
@@ -1,5 +1,6 @@
11pub mod config;
22pub mod local;
3pub mod metrics;
34pub mod repo;
45pub mod s3;
56
Modified lib/storage/local.rs lib/storage/local.rs +70 −18 View lib/storage/local.rs
@@ -11,6 +11,7 @@
1111use url::Url;
1212
1313use crate::config::StorageSettings;
14use crate::metrics;
1415use crate::repo::{StorageError, StorageRepository};
1516
1617#[derive(Debug)]
@@ -42,42 +43,83 @@
4243#[async_trait]
4344impl StorageRepository for LocalStorage {
4445 async fn put(&self, path: &Path, data: Bytes) -> Result<PutResult, StorageError> {
45 Ok(self.store.put(path, data.into()).await?)
46 let _timer = metrics::OpTimer::start();
47 let len = data.len();
48 let result: Result<PutResult, StorageError> = async {
49 Ok(self.store.put(path, data.into()).await?)
50 }
51 .await;
52 metrics::record_put_bytes(len);
53 metrics::record_op("put", &result);
54 result
4655 }
4756
4857 async fn get(&self, path: &Path) -> Result<GetResult, StorageError> {
49 Ok(self.store.get(path).await?)
58 let _timer = metrics::OpTimer::start();
59 let result: Result<GetResult, StorageError> = async {
60 Ok(self.store.get(path).await?)
61 }
62 .await;
63 metrics::record_op("get", &result);
64 result
5065 }
5166
5267 async fn delete(&self, path: &Path) -> Result<(), StorageError> {
53 Ok(self.store.delete(path).await?)
68 let result: Result<(), StorageError> = async { Ok(self.store.delete(path).await?) }.await;
69 metrics::record_op("delete", &result);
70 result
5471 }
5572
5673 async fn head(&self, path: &Path) -> Result<ObjectMeta, StorageError> {
57 Ok(self.store.head(path).await?)
74 let result: Result<ObjectMeta, StorageError> = async {
75 Ok(self.store.head(path).await?)
76 }
77 .await;
78 metrics::record_op("head", &result);
79 result
5880 }
5981
6082 async fn exists(&self, path: &Path) -> Result<bool, StorageError> {
61 Ok(self.store.head(path).await.is_ok())
83 let result: Result<bool, StorageError> = async {
84 Ok(self.store.head(path).await.is_ok())
85 }
86 .await;
87 metrics::record_op("exists", &result);
88 result
6289 }
6390
6491 async fn list(&self, prefix: Option<&Path>) -> Result<Vec<ObjectMeta>, StorageError> {
65 let stream = self.store.list(prefix);
66 Ok(stream.try_collect::<Vec<_>>().await?)
92 let result: Result<Vec<ObjectMeta>, StorageError> = async {
93 let stream = self.store.list(prefix);
94 Ok(stream.try_collect::<Vec<_>>().await?)
95 }
96 .await;
97 metrics::record_op("list", &result);
98 result
6799 }
68100
69101 async fn copy(&self, from: &Path, to: &Path) -> Result<(), StorageError> {
70 Ok(self.store.copy(from, to).await?)
102 let result: Result<(), StorageError> = async {
103 Ok(self.store.copy(from, to).await?)
104 }
105 .await;
106 metrics::record_op("copy", &result);
107 result
71108 }
72109
73110 async fn begin_multipart(
74111 &self,
75112 path: &Path,
76113 ) -> Result<Box<dyn object_store::MultipartUpload>, StorageError> {
77 Ok(self
78 .store
79 .put_multipart_opts(path, object_store::PutMultipartOptions::default())
80 .await?)
114 let result: Result<Box<dyn object_store::MultipartUpload>, StorageError> = async {
115 Ok(self
116 .store
117 .put_multipart_opts(path, object_store::PutMultipartOptions::default())
118 .await?)
119 }
120 .await;
121 metrics::record_op("begin_multipart", &result);
122 result
81123 }
82124
83125 async fn presigned_upload(
@@ -85,9 +127,14 @@
85127 _path: &Path,
86128 _expires_in: Duration,
87129 ) -> Result<Url, StorageError> {
88 Err(StorageError::Unsupported(
89 "presigned upload URLs are not supported by the local backend".to_string(),
90 ))
130 let result: Result<Url, StorageError> = async {
131 Err(StorageError::Unsupported(
132 "presigned upload URLs are not supported by the local backend".to_string(),
133 ))
134 }
135 .await;
136 metrics::record_op("presigned_upload", &result);
137 result
91138 }
92139
93140 async fn presigned_download(
@@ -95,9 +142,14 @@
95142 _path: &Path,
96143 _expires_in: Duration,
97144 ) -> Result<Url, StorageError> {
98 Err(StorageError::Unsupported(
99 "presigned download URLs are not supported by the local backend".to_string(),
100 ))
145 let result: Result<Url, StorageError> = async {
146 Err(StorageError::Unsupported(
147 "presigned download URLs are not supported by the local backend".to_string(),
148 ))
149 }
150 .await;
151 metrics::record_op("presigned_download", &result);
152 result
101153 }
102154}
103155
Added lib/storage/metrics.rs lib/storage/metrics.rs +64 −0 View lib/storage/metrics.rs
@@ -0,0 +1,64 @@
1use std::sync::Once;
2use std::time::Instant;
3
4use metrics::{Unit, counter, describe_counter, describe_histogram, histogram};
5
6use crate::repo::StorageError;
7
8pub const OUTCOME_SUCCESS: &str = "success";
9
10static DESCRIBE_ONCE: Once = Once::new();
11
12pub fn ensure_described() {
13 DESCRIBE_ONCE.call_once(describe_all);
14}
15
16fn describe_all() {
17 describe_counter!(
18 "storage_ops_total",
19 "Storage operations partitioned by op and outcome (success, error)."
20 );
21 describe_counter!("storage_put_bytes_total", "Total bytes written via storage put.");
22 describe_histogram!(
23 "storage_op_duration_seconds",
24 Unit::Seconds,
25 "Duration of storage put and get operations."
26 );
27}
28
29pub fn outcome_for<T>(result: &Result<T, StorageError>) -> &'static str {
30 if result.is_ok() {
31 OUTCOME_SUCCESS
32 } else {
33 "error"
34 }
35}
36
37pub struct OpTimer {
38 start: Instant,
39}
40
41impl OpTimer {
42 pub fn start() -> Self {
43 ensure_described();
44 Self {
45 start: Instant::now(),
46 }
47 }
48}
49
50impl Drop for OpTimer {
51 fn drop(&mut self) {
52 histogram!("storage_op_duration_seconds").record(self.start.elapsed().as_secs_f64());
53 }
54}
55
56pub(crate) fn record_op<T>(op: &'static str, result: &Result<T, StorageError>) {
57 ensure_described();
58 counter!("storage_ops_total", "op" => op, "outcome" => outcome_for(result)).increment(1);
59}
60
61pub(crate) fn record_put_bytes(n: usize) {
62 ensure_described();
63 counter!("storage_put_bytes_total").increment(n as u64);
64}
No newline at end of file
Modified lib/storage/s3.rs lib/storage/s3.rs +66 −14 View lib/storage/s3.rs
@@ -14,6 +14,7 @@
1414use url::Url;
1515
1616use crate::config::StorageSettings;
17use crate::metrics;
1718use crate::repo::{StorageError, StorageRepository};
1819
1920#[derive(Debug)]
@@ -69,42 +70,83 @@
6970#[async_trait]
7071impl StorageRepository for S3Storage {
7172 async fn put(&self, path: &Path, data: Bytes) -> Result<PutResult, StorageError> {
72 Ok(self.store.put(path, data.into()).await?)
73 let _timer = metrics::OpTimer::start();
74 let len = data.len();
75 let result: Result<PutResult, StorageError> = async {
76 Ok(self.store.put(path, data.into()).await?)
77 }
78 .await;
79 metrics::record_put_bytes(len);
80 metrics::record_op("put", &result);
81 result
7382 }
7483
7584 async fn get(&self, path: &Path) -> Result<GetResult, StorageError> {
76 Ok(self.store.get(path).await?)
85 let _timer = metrics::OpTimer::start();
86 let result: Result<GetResult, StorageError> = async {
87 Ok(self.store.get(path).await?)
88 }
89 .await;
90 metrics::record_op("get", &result);
91 result
7792 }
7893
7994 async fn delete(&self, path: &Path) -> Result<(), StorageError> {
80 Ok(self.store.delete(path).await?)
95 let result: Result<(), StorageError> = async { Ok(self.store.delete(path).await?) }.await;
96 metrics::record_op("delete", &result);
97 result
8198 }
8299
83100 async fn head(&self, path: &Path) -> Result<ObjectMeta, StorageError> {
84 Ok(self.store.head(path).await?)
101 let result: Result<ObjectMeta, StorageError> = async {
102 Ok(self.store.head(path).await?)
103 }
104 .await;
105 metrics::record_op("head", &result);
106 result
85107 }
86108
87109 async fn exists(&self, path: &Path) -> Result<bool, StorageError> {
88 Ok(self.store.head(path).await.is_ok())
110 let result: Result<bool, StorageError> = async {
111 Ok(self.store.head(path).await.is_ok())
112 }
113 .await;
114 metrics::record_op("exists", &result);
115 result
89116 }
90117
91118 async fn list(&self, prefix: Option<&Path>) -> Result<Vec<ObjectMeta>, StorageError> {
92 let stream = self.store.list(prefix);
93 Ok(stream.try_collect::<Vec<_>>().await?)
119 let result: Result<Vec<ObjectMeta>, StorageError> = async {
120 let stream = self.store.list(prefix);
121 Ok(stream.try_collect::<Vec<_>>().await?)
122 }
123 .await;
124 metrics::record_op("list", &result);
125 result
94126 }
95127
96128 async fn copy(&self, from: &Path, to: &Path) -> Result<(), StorageError> {
97 Ok(self.store.copy(from, to).await?)
129 let result: Result<(), StorageError> = async {
130 Ok(self.store.copy(from, to).await?)
131 }
132 .await;
133 metrics::record_op("copy", &result);
134 result
98135 }
99136
100137 async fn begin_multipart(
101138 &self,
102139 path: &Path,
103140 ) -> Result<Box<dyn object_store::MultipartUpload>, StorageError> {
104 Ok(self
105 .store
106 .put_multipart_opts(path, object_store::PutMultipartOptions::default())
107 .await?)
141 let result: Result<Box<dyn object_store::MultipartUpload>, StorageError> = async {
142 Ok(self
143 .store
144 .put_multipart_opts(path, object_store::PutMultipartOptions::default())
145 .await?)
146 }
147 .await;
148 metrics::record_op("begin_multipart", &result);
149 result
108150 }
109151
110152 async fn presigned_upload(
@@ -112,7 +154,12 @@
112154 path: &Path,
113155 expires_in: Duration,
114156 ) -> Result<Url, StorageError> {
115 Ok(self.store.signed_url(Method::PUT, path, expires_in).await?)
157 let result: Result<Url, StorageError> = async {
158 Ok(self.store.signed_url(Method::PUT, path, expires_in).await?)
159 }
160 .await;
161 metrics::record_op("presigned_upload", &result);
162 result
116163 }
117164
118165 async fn presigned_download(
@@ -120,6 +167,11 @@
120167 path: &Path,
121168 expires_in: Duration,
122169 ) -> Result<Url, StorageError> {
123 Ok(self.store.signed_url(Method::GET, path, expires_in).await?)
170 let result: Result<Url, StorageError> = async {
171 Ok(self.store.signed_url(Method::GET, path, expires_in).await?)
172 }
173 .await;
174 metrics::record_op("presigned_download", &result);
175 result
124176 }
125177}