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

#![cfg(feature = "discovery-etcd")]

use std::time::{Duration, SystemTime, UNIX_EPOCH};

use rs_zero::discovery::{Discovery, InstanceEndpoint, Registry, ServiceInstance};
use rs_zero::discovery_etcd::{
    BackoffConfig, EtcdDiscoveryConfig, EtcdRegistry, WatchEvent, WatchEventKind,
};

#[tokio::test]
#[ignore = "requires external etcd from examples/production-adapters/docker-compose.yml"]
async fn etcd_registry_external_lifecycle_and_watch() {
    let registry = EtcdRegistry::connect(test_config())
        .await
        .expect("connect etcd");
    let mut watch = registry.watch_service("api").await.expect("watch api");
    let instance = ServiceInstance::new(
        "api",
        "api-1",
        InstanceEndpoint::new("127.0.0.1", 18080).expect("endpoint"),
    )
    .with_metadata("zone", "local")
    .with_weight(3);

    registry.register(instance.clone()).await.expect("register");
    let discovered = discover_until_ready(&registry, "api").await;
    assert_eq!(discovered, vec![instance.clone()]);
    let put = next_event(&mut watch, WatchEventKind::Put, "api-1").await;
    assert_eq!(put.instance.as_ref(), Some(&instance));
    assert!(registry.lease_status("api", "api-1").await.is_some());

    let unhealthy = registry
        .update_health("api", "api-1", false)
        .await
        .expect("mark unhealthy");
    assert!(!unhealthy.healthy);
    let health_event = next_event(&mut watch, WatchEventKind::Put, "api-1").await;
    assert_eq!(health_event.instance.as_ref(), Some(&unhealthy));
    assert_discover_empty(&registry, "api").await;

    let healthy = registry
        .update_health("api", "api-1", true)
        .await
        .expect("mark healthy");
    assert!(healthy.healthy);
    let discovered = discover_until_ready(&registry, "api").await;
    assert_eq!(discovered, vec![healthy.clone()]);

    let removed = registry
        .deregister("api", "api-1")
        .await
        .expect("deregister");
    assert_eq!(removed.id, "api-1");
    let delete = next_event(&mut watch, WatchEventKind::Delete, "api-1").await;
    assert_eq!(delete.service, "api");
    assert_eq!(delete.id, "api-1");
    assert_discover_empty(&registry, "api").await;
}

fn test_config() -> EtcdDiscoveryConfig {
    EtcdDiscoveryConfig {
        endpoints: vec![
            std::env::var("RS_ZERO_ETCD_ENDPOINT")
                .unwrap_or_else(|_| "http://127.0.0.1:2379".to_string()),
        ],
        prefix: format!("/rs-zero-test/{}", unique_suffix()),
        lease_ttl: 5,
        reconnect_interval: Duration::from_millis(100),
        connect_timeout: Duration::from_secs(3),
        operation_timeout: Duration::from_secs(3),
        keep_alive_interval: Duration::from_millis(500),
        keep_alive_timeout: Duration::from_secs(2),
        watch_backoff: BackoffConfig {
            initial: Duration::from_millis(50),
            max: Duration::from_millis(200),
        },
        auth: None,
    }
}

fn unique_suffix() -> String {
    let millis = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis();
    format!("{millis}-{}", std::process::id())
}

async fn discover_until_ready(registry: &EtcdRegistry, service: &str) -> Vec<ServiceInstance> {
    for _ in 0..20 {
        if let Ok(instances) = registry.discover(service).await {
            return instances;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    registry
        .discover(service)
        .await
        .expect("discover eventually")
}

async fn assert_discover_empty(registry: &EtcdRegistry, service: &str) {
    for _ in 0..20 {
        if registry.discover(service).await.is_err() {
            return;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    panic!("service {service} still discoverable");
}

async fn next_event(
    watch: &mut rs_zero::discovery_etcd::EtcdWatchStream,
    kind: WatchEventKind,
    id: &str,
) -> WatchEvent {
    let deadline = tokio::time::sleep(Duration::from_secs(5));
    tokio::pin!(deadline);
    loop {
        tokio::select! {
            _ = &mut deadline => panic!("timed out waiting for {kind:?} event for {id}"),
            item = watch.recv() => {
                let event = item.expect("watch open").expect("watch event");
                if event.kind == kind && event.id == id {
                    return event;
                }
            }
        }
    }
}