mz_cluster_client/
lib.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! The public API for both compute and storage.
11
12#![warn(missing_docs)]
13
14use std::fmt;
15use std::ops::Deref;
16use std::str::FromStr;
17use std::sync::Arc;
18use std::time::Duration;
19
20use anyhow::bail;
21use mz_ore::now::{AsEpochMillis, NowFn};
22use serde::{Deserialize, Serialize};
23
24pub mod client;
25pub mod instances;
26pub mod metrics;
27pub mod params;
28
29/// A function that computes the lag between the given time and wallclock time.
30///
31/// Because sources usually tick once per second and we collect wallclock lag measurements once per
32/// second, the measured lag can be off by up to 1s. To reflect this uncertainty, we report lag
33/// values rounded to seconds. We always round up to avoid underreporting.
34#[derive(Clone)]
35pub struct WallclockLagFn<T>(Arc<dyn Fn(T) -> Duration + Send + Sync>);
36
37impl<T: AsEpochMillis> WallclockLagFn<T> {
38    /// Create a new [`WallclockLagFn`].
39    pub fn new(now: NowFn) -> Self {
40        let inner = Arc::new(move |time: T| {
41            let time_ms = time.as_epoch_millis();
42            let lag_ms = now().saturating_sub(time_ms);
43            let lag_s = lag_ms.div_ceil(1000);
44            Duration::from_secs(lag_s)
45        });
46        Self(inner)
47    }
48}
49
50impl<T> Deref for WallclockLagFn<T> {
51    type Target = dyn Fn(T) -> Duration + Send + Sync;
52
53    fn deref(&self) -> &Self::Target {
54        &(*self.0)
55    }
56}
57
58/// Identifier of a replica.
59#[derive(
60    Clone,
61    Copy,
62    Debug,
63    Eq,
64    PartialEq,
65    Ord,
66    PartialOrd,
67    Hash,
68    Serialize,
69    Deserialize
70)]
71pub enum ReplicaId {
72    /// A user replica.
73    User(u64),
74    /// A system replica.
75    System(u64),
76}
77
78impl ReplicaId {
79    /// Return the inner numeric ID value.
80    pub fn inner_id(&self) -> u64 {
81        match self {
82            ReplicaId::User(id) => *id,
83            ReplicaId::System(id) => *id,
84        }
85    }
86
87    /// Whether this value identifies a user replica.
88    pub fn is_user(&self) -> bool {
89        matches!(self, Self::User(_))
90    }
91
92    /// Whether this value identifies a system replica.
93    pub fn is_system(&self) -> bool {
94        matches!(self, Self::System(_))
95    }
96}
97
98impl fmt::Display for ReplicaId {
99    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
100        match self {
101            Self::User(id) => write!(f, "u{}", id),
102            Self::System(id) => write!(f, "s{}", id),
103        }
104    }
105}
106
107impl FromStr for ReplicaId {
108    type Err = anyhow::Error;
109
110    fn from_str(s: &str) -> Result<Self, Self::Err> {
111        let first = s.chars().next();
112        let rest = s.get(1..);
113        if let (Some(prefix), Some(num)) = (first, rest) {
114            let id = num.parse()?;
115            match prefix {
116                'u' => return Ok(Self::User(id)),
117                's' => return Ok(Self::System(id)),
118                _ => (),
119            }
120        }
121
122        bail!("invalid replica ID: {}", s);
123    }
124}
mz_cluster_client/lib.rs

mz_cluster_client/
lib.rs
