picodata-pike 4.0.0

Cargo plugin for Picodata plugin development
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

use crate::commands::lib::run_query_in_picodata_admin;
use crate::commands::run::Params;
use crate::healthcheck::api;
use anyhow::{bail, Result};
use log::{debug, info};
use std::thread;
use std::time::{Duration, Instant};

use super::PicodataInstance;

const HEALTH_CHECK_TIMEOUT: Duration = Duration::from_secs(60);
const CHECK_INTERVAL: Duration = Duration::from_millis(500);

/// Polls `GET /api/v1/health/ready` on each instance until all return 200,
/// or until the timeout is exceeded.
pub(super) fn wait_instances_ready(instances: &[PicodataInstance]) -> Result<()> {
    if instances.is_empty() {
        return Ok(());
    }

    info!(
        "Waiting for {} instance(s) to become ready (timeout {}s)",
        instances.len(),
        HEALTH_CHECK_TIMEOUT.as_secs()
    );

    let start = Instant::now();

    loop {
        if start.elapsed() >= HEALTH_CHECK_TIMEOUT {
            bail!(
                "cluster setup timed out: not all instances became ready within {}s",
                HEALTH_CHECK_TIMEOUT.as_secs()
            );
        }

        let ready_count = instances
            .iter()
            .map(api::is_instance_ready)
            .collect::<Result<Vec<_>>>()?
            .into_iter()
            .filter(|&ready| ready)
            .count();

        if ready_count == instances.len() {
            info!("All {} instance(s) are ready", instances.len());
            return Ok(());
        }

        debug!(
            "{}/{} instance(s) ready, retrying...",
            ready_count,
            instances.len()
        );
        thread::sleep(CHECK_INTERVAL);
    }
}

/// Polls vshard router state via admin socket on each instance until
/// `bucket_count: {active_buckets}` appears in `vshard.router` output,
/// indicating that vshard discovery has settled.
pub(super) fn wait_vshard_discovery(instances: &[PicodataInstance], params: &Params) -> Result<()> {
    let timeout = Duration::from_secs(params.wait_vshard_discovery_timeout);
    info!(
        "Waiting for vshard discovery on {} instance(s) (timeout {}s)",
        instances.len(),
        timeout.as_secs()
    );

    for instance in instances {
        let socket_path = instance.data_dir().join("admin.sock");
        let start = Instant::now();
        let active_buckets = api::get_health_status(instance)?.buckets.active;
        let needle = format!("bucket_count: {active_buckets}");

        loop {
            if start.elapsed() >= timeout {
                bail!(
                    "vshard discovery timed out: '{needle}' not found in vshard.router output \
                     on instance {} within {}s",
                    instance.http_port(),
                    timeout.as_secs()
                );
            }

            let output = run_query_in_picodata_admin(
                &params.picodata_path,
                &socket_path,
                "\\lua\nvshard.router",
            );

            match output {
                Ok(stdout) if stdout.contains(&needle) => break,
                Ok(_) => {}
                Err(e) => {
                    debug!(
                        "vshard.router query failed on instance {}: {e}",
                        instance.http_port()
                    );
                }
            }

            thread::sleep(CHECK_INTERVAL);
        }
    }

    info!("vshard discovery has been completed on all instances");
    Ok(())
}