rs-zero 0.2.6

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

use axum::extract::MatchedPath;
use axum::{
    body::Body,
    extract::Request,
    http::StatusCode,
    middleware::Next,
    response::{IntoResponse, Response},
};

use crate::{
    resil::{BreakerConfig, BreakerPolicyConfig, BreakerRegistry},
    rest::{
        RestError, RestResilienceConfig,
        middleware::{MiddlewareMetrics, record_resilience_event},
    },
};

/// Applies route-level circuit breaking to one request.
pub async fn breaker_middleware(
    registry: BreakerRegistry,
    service: String,
    config: RestResilienceConfig,
    metrics: MiddlewareMetrics,
    request: Request<Body>,
    next: Next,
) -> Response {
    let key = breaker_key(&service, &request);
    let breaker = registry
        .get_or_insert_with_policy(
            key,
            BreakerConfig {
                failure_threshold: config.breaker_failure_threshold,
                reset_timeout: config.breaker_reset_timeout,
            },
            breaker_policy(&config),
        )
        .await;

    let guard = match breaker.allow().await {
        Ok(guard) => {
            record_resilience_event(&metrics, "breaker", "allowed");
            guard
        }
        Err(error) => {
            record_resilience_event(&metrics, "breaker", "dropped");
            return RestError::ServiceUnavailable(error.to_string()).into_response();
        }
    };

    let response = next.run(request).await;
    if response.status().is_server_error() {
        record_resilience_event(&metrics, "breaker", "failure");
        guard.record_failure().await;
    } else {
        record_resilience_event(&metrics, "breaker", "success");
        guard.record_success().await;
    }
    response
}

fn breaker_key(service: &str, request: &Request<Body>) -> String {
    let method = request.method().as_str();
    let route = request
        .extensions()
        .get::<MatchedPath>()
        .map(|path| path.as_str())
        .unwrap_or("unknown");
    format!("{service}:{method}:{route}")
}

/// Returns whether a response status should count as a breaker failure.
pub fn is_failure_status(status: StatusCode) -> bool {
    status.is_server_error()
}

/// Builds a breaker policy from REST resilience config.
pub fn breaker_policy(config: &RestResilienceConfig) -> BreakerPolicyConfig {
    let mut policy = if config.breaker_sre_enabled {
        BreakerPolicyConfig::google_sre()
    } else {
        BreakerPolicyConfig::default()
    };
    policy.sre_k_millis = config.breaker_sre_k_millis;
    policy.sre_protection = config.breaker_sre_protection;
    policy
}