krabka_bench_driver/soak/
client.rs

1//! The client side of a contract run: the paced producers, the live consumer
2//! groups, and the read-back after the load.
3//!
4//! Unlike the judge, this half does IO. It is shared by the two executors that
5//! run a contract: the `soak_contract` suite of `krabka-broker`, which runs a
6//! soak or a self-check against its own three-broker cluster, and
7//! `krabka-soak-drill`, which runs a disaster-recovery contract against the
8//! secured cluster of the qualification drill. Both send the same records,
9//! keep the same ledgers and read back the same way, so the reconciler judges
10//! both by the same rules.
11//!
12//! - [`load`]: one paced producer per path, one live consumer group per path
13//!   and isolation level.
14//! - [`readback`]: the read-back from the earliest offset, and the
15//!   committed-offset checks.
16//! - [`windows`]: steady and disturbed time, and the latency split across it.
17
18pub mod load;
19pub mod readback;
20pub mod windows;
21
22use {
23    krabka_client_core::{
24        sasl::SaslCredentials,
25        security::{ClientSecurity, TlsConnectorConfig, TrustStore},
26    },
27    krabka_security::{ListenerProtocol, SaslMechanism},
28    std::{
29        path::PathBuf,
30        time::{SystemTime, UNIX_EPOCH},
31    },
32};
33
34/// The wall clock in Unix milliseconds, the coordinate every timestamp of the
35/// evidence is in.
36#[must_use]
37pub fn now_ms() -> i64 {
38    SystemTime::now()
39        .duration_since(UNIX_EPOCH)
40        .map_or(0, |since| {
41            i64::try_from(since.as_millis()).unwrap_or(i64::MAX)
42        })
43}
44
45/// Where every client of a run connects, and how it authenticates.
46#[derive(Debug, Clone)]
47pub struct Endpoint {
48    pub bootstrap: String,
49    /// `None` is the anonymous `PLAINTEXT` listener of the soak cluster.
50    pub security: Option<ClientSecurity>,
51}
52
53impl Endpoint {
54    /// The anonymous `PLAINTEXT` listener at `bootstrap`.
55    #[must_use]
56    pub fn plaintext(bootstrap: impl Into<String>) -> Self {
57        Self {
58            bootstrap: bootstrap.into(),
59            security: None,
60        }
61    }
62
63    /// A `SASL_SSL` listener at `bootstrap` that authenticates with
64    /// SCRAM-SHA-512 and verifies the broker against `ca`. The broker's
65    /// certificate must name `server_name`: the client dials the address the
66    /// bootstrap resolves to, so the name is not taken from the bootstrap.
67    #[must_use]
68    pub fn sasl_ssl(bootstrap: impl Into<String>, scram: ScramLogin) -> Self {
69        let mut tls = TlsConnectorConfig::default();
70        tls.trust_store = TrustStore::PemFile(scram.ca);
71        tls.server_name = scram.server_name;
72        Self {
73            bootstrap: bootstrap.into(),
74            security: Some(ClientSecurity {
75                protocol: ListenerProtocol::SaslSsl,
76                tls: Some(tls),
77                sasl: Some(SaslCredentials::Scram {
78                    mechanism: SaslMechanism::ScramSha512,
79                    username: scram.username,
80                    password: scram.password,
81                    delegation_token: false,
82                }),
83                sasl_host: None,
84            }),
85        }
86    }
87}
88
89/// A SCRAM-SHA-512 login over TLS.
90#[derive(Debug, Clone)]
91pub struct ScramLogin {
92    /// The PEM CA bundle the broker's certificate chains to.
93    pub ca: PathBuf,
94    /// The name the broker's certificate carries.
95    pub server_name: String,
96    pub username: String,
97    pub password: String,
98}
99
100#[cfg(test)]
101mod tests {
102    use {
103        super::{Endpoint, ScramLogin},
104        assert2::check,
105        krabka_client_core::{sasl::SaslCredentials, security::TrustStore},
106        krabka_security::{ListenerProtocol, SaslMechanism},
107    };
108
109    #[test]
110    fn a_sasl_ssl_endpoint_logs_in_with_scram_over_verified_tls() {
111        let endpoint = Endpoint::sasl_ssl("broker-1:9094", ScramLogin {
112            ca: "/work/tls/ca.pem".into(),
113            server_name: "broker-1".into(),
114            username: "soak".into(),
115            password: "secret".into(),
116        });
117        check!(endpoint.bootstrap == "broker-1:9094");
118        let security = endpoint.security.expect("a SASL_SSL endpoint is secured");
119        check!(security.protocol == ListenerProtocol::SaslSsl);
120        let tls = security.tls.expect("SASL_SSL carries a TLS config");
121        check!(tls.trust_store == TrustStore::PemFile("/work/tls/ca.pem".into()));
122        check!(tls.server_name == "broker-1");
123        let Some(SaslCredentials::Scram {
124            mechanism,
125            username,
126            password,
127            delegation_token,
128        }) = security.sasl
129        else {
130            panic!("a SASL_SSL endpoint logs in with SCRAM");
131        };
132        check!(
133            (
134                mechanism,
135                username.as_str(),
136                password.as_str(),
137                delegation_token
138            ) == (SaslMechanism::ScramSha512, "soak", "secret", false)
139        );
140    }
141
142    #[test]
143    fn a_plaintext_endpoint_is_anonymous() {
144        let endpoint = Endpoint::plaintext("localhost:9092");
145        check!(endpoint.security.is_none());
146    }
147}
krabka_bench_driver/soak/client.rs

krabka_bench_driver/soak/
client.rs
