1use crate::attempts::counting_attempts;
5use crate::layout::{parse_object_key, DurableObjectFamily};
6use crate::object_store::Result;
7use crate::{
8 ByteRange, ByteStream, MultipartCompletion, MultipartPart, ObjectBody, ObjectMetadata,
9 ObjectStore, ObjectStoreError, PutMode, StoredObjectChecksum,
10};
11use async_trait::async_trait;
12use bytes::Bytes;
13use futures::stream::{BoxStream, TryStreamExt};
14use loonfs_api::StorageChecksum;
15use serde::{Deserialize, Serialize};
16use std::fmt;
17use std::fs::{self, File};
18use std::io::{self, BufWriter, Write};
19use std::path::Path;
20use std::sync::{Arc, Mutex};
21use std::time::{Duration, Instant};
22
23#[derive(Debug, Clone, Serialize, Deserialize)]
28pub struct ObjectStoreMetricSample {
29 pub operation: ObjectStoreOperation,
31 pub elapsed_micros: u128,
33 pub attempts: u32,
43 pub result: ObjectStoreResultClass,
45 pub bytes_in: Option<u64>,
47 pub bytes_out: Option<u64>,
49 pub item_count: Option<u64>,
51 pub key_class: KeyClass,
53 pub range_class: Option<RangeClass>,
55 pub put_mode: Option<PutModeClass>,
57 pub store_kind: Option<String>,
59}
60
61#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
63#[serde(rename_all = "snake_case")]
64pub enum ObjectStoreOperation {
65 Head,
67 GetWithMetadata,
69 Get,
71 Put,
73 PutStreamed,
75 Delete,
77 CreateMultipartUpload,
79 CompleteMultipartUpload,
81 AbortMultipartUpload,
83 ListPrefix,
85 ListPrefixStream,
87}
88
89impl ObjectStoreOperation {
90 pub fn as_str(self) -> &'static str {
93 match self {
94 Self::Head => "head",
95 Self::GetWithMetadata => "get_with_metadata",
96 Self::Get => "get",
97 Self::Put => "put",
98 Self::PutStreamed => "put_streamed",
99 Self::Delete => "delete",
100 Self::CreateMultipartUpload => "create_multipart_upload",
101 Self::CompleteMultipartUpload => "complete_multipart_upload",
102 Self::AbortMultipartUpload => "abort_multipart_upload",
103 Self::ListPrefix => "list_prefix",
104 Self::ListPrefixStream => "list_prefix_stream",
105 }
106 }
107}
108
109#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
111#[serde(rename_all = "snake_case")]
112pub enum ObjectStoreResultClass {
113 Ok,
115 NotFound,
117 InvalidKey,
119 InvalidContentRef,
121 InvalidRange,
123 PreconditionFailed,
125 PermissionDenied,
127 Unsupported,
129 Transport,
131 OtherError,
133}
134
135impl ObjectStoreResultClass {
136 pub fn as_str(self) -> &'static str {
139 match self {
140 Self::Ok => "ok",
141 Self::NotFound => "not_found",
142 Self::InvalidKey => "invalid_key",
143 Self::InvalidContentRef => "invalid_content_ref",
144 Self::InvalidRange => "invalid_range",
145 Self::PreconditionFailed => "precondition_failed",
146 Self::PermissionDenied => "permission_denied",
147 Self::Unsupported => "unsupported",
148 Self::Transport => "transport",
149 Self::OtherError => "other_error",
150 }
151 }
152}
153
154#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
156#[serde(rename_all = "snake_case")]
157pub enum KeyClass {
158 Content,
160 Metadata,
162 NamespaceHead,
164 WalSegment,
166 NamespaceManifest,
168 MetadataSst,
170 GcControl,
172 Unknown,
174}
175
176#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
178#[serde(rename_all = "snake_case")]
179pub enum RangeClass {
180 FullObject,
182 Prefix,
184 Suffix,
186 Bounded,
188 Empty,
190}
191
192#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
194#[serde(rename_all = "snake_case")]
195pub enum PutModeClass {
196 Overwrite,
198 CreateIfAbsent,
200 CompareAndSwap,
202}
203
204pub trait ObjectStoreMetricsRecorder: Send + Sync + 'static {
208 fn record(&self, sample: ObjectStoreMetricSample);
210}
211
212#[derive(Default)]
214pub struct VecObjectStoreMetricsRecorder {
215 samples: Mutex<Vec<ObjectStoreMetricSample>>,
216}
217
218impl VecObjectStoreMetricsRecorder {
219 pub fn samples(&self) -> Vec<ObjectStoreMetricSample> {
221 self.samples
222 .lock()
223 .unwrap_or_else(|poisoned| poisoned.into_inner())
224 .clone()
225 }
226}
227
228impl ObjectStoreMetricsRecorder for VecObjectStoreMetricsRecorder {
229 fn record(&self, sample: ObjectStoreMetricSample) {
230 self.samples
231 .lock()
232 .unwrap_or_else(|poisoned| poisoned.into_inner())
233 .push(sample);
234 }
235}
236
237pub struct JsonlObjectStoreMetricsRecorder {
242 writer: Mutex<BufWriter<File>>,
243}
244
245impl JsonlObjectStoreMetricsRecorder {
246 pub fn create(path: impl AsRef<Path>) -> io::Result<Self> {
250 let path = path.as_ref();
251 if let Some(parent) = path.parent() {
252 if !parent.as_os_str().is_empty() {
253 fs::create_dir_all(parent)?;
254 }
255 }
256 Ok(Self {
257 writer: Mutex::new(BufWriter::new(File::create(path)?)),
258 })
259 }
260
261 pub fn flush(&self) -> io::Result<()> {
265 self.writer
266 .lock()
267 .unwrap_or_else(|poisoned| poisoned.into_inner())
268 .flush()
269 }
270}
271
272impl ObjectStoreMetricsRecorder for JsonlObjectStoreMetricsRecorder {
273 fn record(&self, sample: ObjectStoreMetricSample) {
274 let mut writer = self
275 .writer
276 .lock()
277 .unwrap_or_else(|poisoned| poisoned.into_inner());
278 let _ = serde_json::to_writer(&mut *writer, &sample);
279 let _ = writer.write_all(b"\n");
280 }
281}
282
283pub struct InstrumentedObjectStore<S> {
288 inner: S,
289 recorder: Arc<dyn ObjectStoreMetricsRecorder>,
290 store_kind: Option<String>,
291}
292
293impl<S: fmt::Debug> fmt::Debug for InstrumentedObjectStore<S> {
294 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
295 f.debug_struct("InstrumentedObjectStore")
296 .field("inner", &self.inner)
297 .field("store_kind", &self.store_kind)
298 .finish_non_exhaustive()
299 }
300}
301
302impl<S> InstrumentedObjectStore<S> {
303 pub fn new(inner: S, recorder: Arc<dyn ObjectStoreMetricsRecorder>) -> Self {
305 Self {
306 inner,
307 recorder,
308 store_kind: None,
309 }
310 }
311
312 pub fn store_kind(mut self, store_kind: impl Into<String>) -> Self {
314 self.store_kind = Some(store_kind.into());
315 self
316 }
317
318 pub fn into_inner(self) -> S {
320 self.inner
321 }
322}
323
324#[allow(clippy::disallowed_methods)]
325fn sample_clock() -> Instant {
326 Instant::now()
329}
330
331#[async_trait]
332impl<S> ObjectStore for InstrumentedObjectStore<S>
333where
334 S: ObjectStore,
335{
336 async fn head(&self, key: &str) -> Result<Option<ObjectMetadata>> {
337 let start = sample_clock();
338 let (result, attempts) = counting_attempts(self.inner.head(key)).await;
339 self.record_head_like(
340 ObjectStoreOperation::Head,
341 key,
342 start.elapsed(),
343 attempts,
344 &result,
345 );
346 result
347 }
348
349 async fn head_stored_checksum(&self, key: &str) -> Result<Option<StoredObjectChecksum>> {
350 let start = sample_clock();
351 let (result, attempts) = counting_attempts(self.inner.head_stored_checksum(key)).await;
352 self.record_head_like(
355 ObjectStoreOperation::Head,
356 key,
357 start.elapsed(),
358 attempts,
359 &result,
360 );
361 result
362 }
363
364 async fn create_multipart_upload(&self, key: &str) -> Result<String> {
365 let start = sample_clock();
366 let (result, attempts) = counting_attempts(self.inner.create_multipart_upload(key)).await;
367 self.record_unit(
371 ObjectStoreOperation::CreateMultipartUpload,
372 key,
373 start.elapsed(),
374 attempts,
375 &result,
376 );
377 result
378 }
379
380 async fn complete_multipart_upload(
381 &self,
382 key: &str,
383 provider_upload_id: &str,
384 parts: &[MultipartPart],
385 full_object_checksum: &StorageChecksum,
386 ) -> Result<MultipartCompletion> {
387 let start = sample_clock();
388 let (result, attempts) = counting_attempts(self.inner.complete_multipart_upload(
389 key,
390 provider_upload_id,
391 parts,
392 full_object_checksum,
393 ))
394 .await;
395 self.record_unit(
396 ObjectStoreOperation::CompleteMultipartUpload,
397 key,
398 start.elapsed(),
399 attempts,
400 &result,
401 );
402 result
403 }
404
405 async fn abort_multipart_upload(&self, key: &str, provider_upload_id: &str) -> Result<()> {
406 let start = sample_clock();
407 let (result, attempts) =
408 counting_attempts(self.inner.abort_multipart_upload(key, provider_upload_id)).await;
409 self.record_unit(
410 ObjectStoreOperation::AbortMultipartUpload,
411 key,
412 start.elapsed(),
413 attempts,
414 &result,
415 );
416 result
417 }
418
419 async fn get(&self, key: &str, range: Option<ByteRange>) -> Result<Option<Bytes>> {
420 let start = sample_clock();
421 let (result, attempts) = counting_attempts(self.inner.get(key, range.clone())).await;
422 self.record_get(key, range.as_ref(), start.elapsed(), attempts, &result);
423 result
424 }
425
426 async fn get_with_metadata(&self, key: &str) -> Result<Option<ObjectBody>> {
427 let start = sample_clock();
428 let (result, attempts) = counting_attempts(self.inner.get_with_metadata(key)).await;
429 self.record_get_with_metadata(key, start.elapsed(), attempts, &result);
430 result
431 }
432
433 async fn put(&self, key: &str, bytes: Bytes, mode: PutMode) -> Result<ObjectMetadata> {
434 let start = sample_clock();
435 let bytes_in = bytes.len() as u64;
436 let (result, attempts) = counting_attempts(self.inner.put(key, bytes, mode.clone())).await;
437 self.record_put(key, bytes_in, &mode, start.elapsed(), attempts, &result);
438 result
439 }
440
441 async fn put_streamed(&self, key: &str, body: ByteStream, mode: PutMode) -> Result<u64> {
446 let start = sample_clock();
447 let (result, attempts) =
448 counting_attempts(self.inner.put_streamed(key, body, mode.clone())).await;
449 self.record(ObjectStoreMetricSample {
450 operation: ObjectStoreOperation::PutStreamed,
451 elapsed_micros: start.elapsed().as_micros(),
452 attempts,
453 result: classify_result(&result),
454 bytes_in: result.as_ref().ok().copied(),
455 bytes_out: None,
456 item_count: None,
457 key_class: classify_key(key),
458 range_class: None,
459 put_mode: Some(classify_put_mode(&mode)),
460 store_kind: self.store_kind.clone(),
461 });
462 result
463 }
464
465 async fn delete(&self, key: &str) -> Result<()> {
466 let start = sample_clock();
467 let (result, attempts) = counting_attempts(self.inner.delete(key)).await;
468 self.record_unit(
469 ObjectStoreOperation::Delete,
470 key,
471 start.elapsed(),
472 attempts,
473 &result,
474 );
475 result
476 }
477
478 fn list_prefix_stream(&self, prefix: &str) -> BoxStream<'static, Result<String>> {
479 Box::pin(RecordedListStream {
484 inner: self.inner.list_prefix_stream(prefix),
485 recorder: Arc::clone(&self.recorder),
486 store_kind: self.store_kind.clone(),
487 key_class: classify_key(prefix),
488 started: sample_clock(),
489 items: 0,
490 first_error: None,
491 })
492 }
493
494 async fn list_prefix(&self, prefix: &str) -> Result<Vec<String>> {
495 let start = sample_clock();
496 let (result, attempts): (Result<Vec<_>>, u32) = counting_attempts(async {
497 self.inner
498 .list_prefix_stream(prefix)
499 .try_collect()
500 .await
501 .map(|mut keys: Vec<String>| {
502 keys.sort();
503 keys
504 })
505 })
506 .await;
507 self.record_list(prefix, start.elapsed(), attempts, &result);
508 result
509 }
510}
511
512impl<S> InstrumentedObjectStore<S> {
513 fn record_head_like<T>(
514 &self,
515 operation: ObjectStoreOperation,
516 key: &str,
517 elapsed: Duration,
518 attempts: u32,
519 result: &Result<Option<T>>,
520 ) {
521 self.record(ObjectStoreMetricSample {
522 operation,
523 elapsed_micros: elapsed.as_micros(),
524 attempts,
525 result: classify_optional_result(result),
526 bytes_in: None,
527 bytes_out: None,
528 item_count: None,
529 key_class: classify_key(key),
530 range_class: None,
531 put_mode: None,
532 store_kind: self.store_kind.clone(),
533 });
534 }
535
536 fn record_get(
537 &self,
538 key: &str,
539 range: Option<&ByteRange>,
540 elapsed: Duration,
541 attempts: u32,
542 result: &Result<Option<Bytes>>,
543 ) {
544 self.record(ObjectStoreMetricSample {
545 operation: ObjectStoreOperation::Get,
546 elapsed_micros: elapsed.as_micros(),
547 attempts,
548 result: classify_optional_result(result),
549 bytes_in: None,
550 bytes_out: result
551 .as_ref()
552 .ok()
553 .and_then(|bytes| bytes.as_ref().map(|bytes| bytes.len() as u64)),
554 item_count: None,
555 key_class: classify_key(key),
556 range_class: Some(classify_range(range)),
557 put_mode: None,
558 store_kind: self.store_kind.clone(),
559 });
560 }
561
562 fn record_get_with_metadata(
563 &self,
564 key: &str,
565 elapsed: Duration,
566 attempts: u32,
567 result: &Result<Option<ObjectBody>>,
568 ) {
569 self.record(ObjectStoreMetricSample {
570 operation: ObjectStoreOperation::GetWithMetadata,
571 elapsed_micros: elapsed.as_micros(),
572 attempts,
573 result: classify_optional_result(result),
574 bytes_in: None,
575 bytes_out: result
576 .as_ref()
577 .ok()
578 .and_then(|body| body.as_ref().map(|body| body.bytes.len() as u64)),
579 item_count: None,
580 key_class: classify_key(key),
581 range_class: Some(RangeClass::FullObject),
582 put_mode: None,
583 store_kind: self.store_kind.clone(),
584 });
585 }
586
587 fn record_put(
588 &self,
589 key: &str,
590 bytes_in: u64,
591 mode: &PutMode,
592 elapsed: Duration,
593 attempts: u32,
594 result: &Result<ObjectMetadata>,
595 ) {
596 self.record(ObjectStoreMetricSample {
597 operation: ObjectStoreOperation::Put,
598 elapsed_micros: elapsed.as_micros(),
599 attempts,
600 result: classify_result(result),
601 bytes_in: Some(bytes_in),
602 bytes_out: None,
603 item_count: None,
604 key_class: classify_key(key),
605 range_class: None,
606 put_mode: Some(classify_put_mode(mode)),
607 store_kind: self.store_kind.clone(),
608 });
609 }
610
611 fn record_unit<T>(
612 &self,
613 operation: ObjectStoreOperation,
614 key: &str,
615 elapsed: Duration,
616 attempts: u32,
617 result: &Result<T>,
618 ) {
619 self.record(ObjectStoreMetricSample {
620 operation,
621 elapsed_micros: elapsed.as_micros(),
622 attempts,
623 result: classify_result(result),
624 bytes_in: None,
625 bytes_out: None,
626 item_count: None,
627 key_class: classify_key(key),
628 range_class: None,
629 put_mode: None,
630 store_kind: self.store_kind.clone(),
631 });
632 }
633
634 fn record_list(
635 &self,
636 prefix: &str,
637 elapsed: Duration,
638 attempts: u32,
639 result: &Result<Vec<String>>,
640 ) {
641 self.record(ObjectStoreMetricSample {
642 operation: ObjectStoreOperation::ListPrefix,
643 elapsed_micros: elapsed.as_micros(),
644 attempts,
645 result: classify_result(result),
646 bytes_in: None,
647 bytes_out: None,
648 item_count: result.as_ref().ok().map(|items| items.len() as u64),
649 key_class: classify_key(prefix),
650 range_class: Some(RangeClass::Prefix),
651 put_mode: None,
652 store_kind: self.store_kind.clone(),
653 });
654 }
655
656 fn record(&self, sample: ObjectStoreMetricSample) {
657 self.recorder.record(sample);
658 }
659}
660
661struct RecordedListStream {
662 inner: BoxStream<'static, Result<String>>,
663 recorder: Arc<dyn ObjectStoreMetricsRecorder>,
664 store_kind: Option<String>,
665 key_class: KeyClass,
666 started: Instant,
667 items: u64,
668 first_error: Option<ObjectStoreResultClass>,
669}
670
671impl futures::Stream for RecordedListStream {
672 type Item = Result<String>;
673
674 fn poll_next(
675 mut self: std::pin::Pin<&mut Self>,
676 cx: &mut std::task::Context<'_>,
677 ) -> std::task::Poll<Option<Self::Item>> {
678 let polled = self.inner.as_mut().poll_next(cx);
679 if let std::task::Poll::Ready(Some(item)) = &polled {
680 match item {
681 Ok(_) => self.items += 1,
682 Err(error) => {
683 if self.first_error.is_none() {
684 self.first_error = Some(classify_error(error));
685 }
686 }
687 }
688 }
689 polled
690 }
691}
692
693impl Drop for RecordedListStream {
694 fn drop(&mut self) {
695 self.recorder.record(ObjectStoreMetricSample {
696 operation: ObjectStoreOperation::ListPrefixStream,
697 elapsed_micros: self.started.elapsed().as_micros(),
698 attempts: 1,
703 result: self.first_error.unwrap_or(ObjectStoreResultClass::Ok),
704 bytes_in: None,
705 bytes_out: None,
706 item_count: Some(self.items),
707 key_class: self.key_class,
708 range_class: None,
709 put_mode: None,
710 store_kind: self.store_kind.clone(),
711 });
712 }
713}
714
715fn classify_key(key: &str) -> KeyClass {
716 let Some(parsed) = parse_object_key(key) else {
717 return KeyClass::Unknown;
718 };
719
720 match parsed.family() {
721 DurableObjectFamily::ContentBlob => KeyClass::Content,
722 DurableObjectFamily::WalHead => KeyClass::NamespaceHead,
723 DurableObjectFamily::WalSegment => KeyClass::WalSegment,
724 DurableObjectFamily::MetadataManifest => KeyClass::NamespaceManifest,
725 DurableObjectFamily::MetadataTable => KeyClass::MetadataSst,
726 DurableObjectFamily::CheckpointRecord | DurableObjectFamily::WalFloor => {
727 KeyClass::GcControl
728 }
729 DurableObjectFamily::MetadataRoot => KeyClass::NamespaceManifest,
730 DurableObjectFamily::UploadSession => KeyClass::Metadata,
731 }
732}
733
734fn classify_optional_result<T>(result: &Result<Option<T>>) -> ObjectStoreResultClass {
735 match result {
736 Ok(Some(_)) => ObjectStoreResultClass::Ok,
737 Ok(None) => ObjectStoreResultClass::NotFound,
738 Err(error) => classify_error(error),
739 }
740}
741
742fn classify_result<T>(result: &Result<T>) -> ObjectStoreResultClass {
743 match result {
744 Ok(_) => ObjectStoreResultClass::Ok,
745 Err(error) => classify_error(error),
746 }
747}
748
749fn classify_error(error: &ObjectStoreError) -> ObjectStoreResultClass {
750 match error {
751 ObjectStoreError::NotFound { .. } => ObjectStoreResultClass::NotFound,
752 ObjectStoreError::InvalidKey { .. } => ObjectStoreResultClass::InvalidKey,
753 ObjectStoreError::InvalidContentRef(_) => ObjectStoreResultClass::InvalidContentRef,
754 ObjectStoreError::InvalidRange { .. } => ObjectStoreResultClass::InvalidRange,
755 ObjectStoreError::PreconditionFailed { .. } => ObjectStoreResultClass::PreconditionFailed,
756 ObjectStoreError::PermissionDenied { .. } => ObjectStoreResultClass::PermissionDenied,
757 ObjectStoreError::Unsupported(_) => ObjectStoreResultClass::Unsupported,
758 ObjectStoreError::Configuration(_) => ObjectStoreResultClass::Transport,
761 ObjectStoreError::Transport { .. } => ObjectStoreResultClass::Transport,
762 }
763}
764
765fn classify_range(range: Option<&ByteRange>) -> RangeClass {
766 let Some(range) = range else {
767 return RangeClass::FullObject;
768 };
769 if range.start_inclusive == range.end_exclusive {
770 RangeClass::Empty
771 } else if range.start_inclusive == 0 {
772 RangeClass::Prefix
773 } else if range.end_exclusive == u64::MAX {
774 RangeClass::Suffix
775 } else {
776 RangeClass::Bounded
777 }
778}
779
780fn classify_put_mode(mode: &PutMode) -> PutModeClass {
781 match mode {
782 PutMode::Overwrite => PutModeClass::Overwrite,
783 PutMode::CreateIfAbsent => PutModeClass::CreateIfAbsent,
784 PutMode::CompareAndSwap { .. } => PutModeClass::CompareAndSwap,
785 }
786}