rs-zero 0.2.8

Rust-first microservice framework inspired by go-zero engineering practices
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161

use std::{
    future::Future,
    time::{Duration, Instant},
};

use crate::{
    cache_redis::{
        RedisCacheError, RedisCacheResult, RedisClusterRedirectKind, RedisCommandEvent,
        RedisCommandEventKind, RedisCommandOutcome, RedisDegradationEvent,
    },
    resil::BreakerError,
};

/// Records Redis command, pool, redirect and degradation metrics.
#[derive(Debug, Clone, Default)]
pub struct RedisCommandRecorder {
    #[cfg(feature = "observability")]
    metrics: Option<crate::observability::MetricsRegistry>,
    shard: String,
}

impl RedisCommandRecorder {
    /// Creates a recorder without attached metrics.
    pub fn new() -> Self {
        Self {
            #[cfg(feature = "observability")]
            metrics: None,
            shard: "default".to_string(),
        }
    }

    /// Sets the low-cardinality shard label.
    pub fn with_shard(mut self, shard: impl Into<String>) -> Self {
        self.shard = shard.into();
        self
    }

    /// Attaches metrics.
    #[cfg(feature = "observability")]
    pub fn with_metrics(mut self, metrics: crate::observability::MetricsRegistry) -> Self {
        self.metrics = Some(metrics);
        self
    }

    /// Records a command duration and outcome.
    pub fn record_command(
        &self,
        operation: &str,
        outcome: RedisCommandOutcome,
        duration: Duration,
    ) {
        #[cfg(feature = "observability")]
        crate::observability::record_redis_command(
            self.metrics.as_ref(),
            operation,
            &self.shard,
            outcome.as_str(),
            duration,
        );

        #[cfg(not(feature = "observability"))]
        let _ = (operation, outcome, duration);
    }

    /// Records a Redis low-cardinality event.
    pub fn record_event(&self, event: RedisCommandEvent) {
        #[cfg(feature = "observability")]
        crate::observability::record_redis_event(
            self.metrics.as_ref(),
            event.event.as_str(),
            &self.shard,
            event.outcome.as_str(),
        );

        #[cfg(not(feature = "observability"))]
        let _ = event;
    }

    /// Records a Redis Cluster redirect.
    pub fn record_redirect(&self, kind: RedisClusterRedirectKind) {
        self.record_event(RedisCommandEvent::redirect(kind));
    }

    /// Records a degradation event.
    pub fn record_degradation(&self, event: RedisDegradationEvent) {
        #[cfg(feature = "observability")]
        crate::observability::record_redis_degradation(
            self.metrics.as_ref(),
            event.operation.as_str(),
            event.action.as_str(),
            &self.shard,
        );
        #[cfg(not(feature = "observability"))]
        let _ = event;

        self.record_event(RedisCommandEvent::new(
            RedisCommandEventKind::Degradation,
            RedisCommandOutcome::Degraded,
        ));
    }

    /// Records a Redis breaker event.
    pub fn record_breaker(&self, outcome: RedisCommandOutcome) {
        self.record_event(RedisCommandEvent::new(
            RedisCommandEventKind::Breaker,
            outcome,
        ));
    }
}

/// Runs a Redis future with timeout and records the classified outcome.
pub async fn run_redis_command<T, F>(
    recorder: &RedisCommandRecorder,
    operation: &'static str,
    command_timeout: Duration,
    future: F,
) -> RedisCacheResult<T>
where
    F: Future<Output = redis::RedisResult<T>>,
{
    run_redis_cache_command(recorder, operation, command_timeout, async {
        future.await.map_err(redis_error)
    })
    .await
}

pub(crate) async fn run_redis_cache_command<T, F>(
    recorder: &RedisCommandRecorder,
    operation: &'static str,
    command_timeout: Duration,
    future: F,
) -> RedisCacheResult<T>
where
    F: Future<Output = RedisCacheResult<T>>,
{
    let started = Instant::now();
    let result = match tokio::time::timeout(command_timeout, future).await {
        Ok(result) => result,
        Err(_) => Err(RedisCacheError::Timeout(operation.to_string())),
    };
    let outcome = match &result {
        Ok(_) => RedisCommandOutcome::Success,
        Err(error) => crate::cache_redis::classify_redis_error(error),
    };
    recorder.record_command(operation, outcome, started.elapsed());
    if let RedisCommandOutcome::Redirect(kind) = outcome {
        recorder.record_redirect(kind);
    }
    result
}

pub(crate) fn redis_breaker_rejection(
    operation: &'static str,
    error: BreakerError,
) -> RedisCacheError {
    RedisCacheError::BreakerOpen(format!("{operation}: {error}"))
}

pub(crate) fn redis_error(error: redis::RedisError) -> RedisCacheError {
    RedisCacheError::Backend(error.to_string())
}