felixctl 0.6.0

Command-line tool for Felix: publish, subscribe, read caches, manage the control plane and run benchmarks
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

//! Reaching the brokers: addresses, TLS and the client configuration, built
//! from resolved [`Settings`] with `felix-client`'s public API only.

use std::net::SocketAddr;
use std::sync::Arc;

use felix_client::{ClientConfig, ClientIdentity, ClusterClient, ReconnectPolicy, Seed};

use crate::context::Settings;
use crate::error::{Exit, MarkExit, fail};

/// A connection to the cluster, and the tenant and namespace to use it in.
pub(crate) struct Broker {
    pub(crate) cluster: Arc<ClusterClient>,
    pub(crate) tenant: String,
    pub(crate) namespace: String,
}

impl Broker {
    /// Connect to whichever configured broker answers first.
    pub(crate) async fn connect(settings: &Settings) -> anyhow::Result<Self> {
        let seeds = seeds(settings.brokers()?)?;
        let config = client_config(settings)?;
        let server_name = server_name(settings);
        let cluster =
            ClusterClient::connect_seeds(&seeds, &server_name, config, ReconnectPolicy::default())
                .await
                .mark(
                    Exit::Connection,
                    format!("connect to {}", settings.brokers.join(", ")),
                )?;
        Ok(Self {
            cluster: Arc::new(cluster),
            tenant: settings.tenant()?.to_string(),
            namespace: settings.namespace.clone(),
        })
    }
}

/// The client configuration for `settings`: the library's defaults and
/// `FELIX_*` tuning overrides, then this tenant and token, over QUIC TLS
/// built from the CA and client certificate files.
pub(crate) fn client_config(settings: &Settings) -> anyhow::Result<ClientConfig> {
    let tenant = settings.tenant()?.to_string();
    let token = settings.token()?;
    let quinn = quic_tls(settings)?;
    let mut config = ClientConfig::from_env_or_yaml(quinn, None)
        .mark(Exit::Usage, "read the client configuration")?;
    config.auth_tenant_id = Some(tenant);
    config.auth_token = Some(token);
    Ok(config)
}

/// The TLS server name: explicit, else the first broker's host name, else
/// `localhost`, which is what a broker's generated certificate names.
pub(crate) fn server_name(settings: &Settings) -> String {
    if let Some(name) = &settings.server_name {
        return name.clone();
    }
    settings
        .brokers
        .first()
        .and_then(|broker| broker.rsplit_once(':').map(|(host, _)| host))
        .map(|host| host.trim_start_matches('[').trim_end_matches(']'))
        .filter(|host| host.parse::<std::net::IpAddr>().is_err() && !host.is_empty())
        .unwrap_or("localhost")
        .to_string()
}

/// The brokers as cluster client seeds. Names stay names, so the client can
/// resolve them again after every broker behind them has moved.
pub(crate) fn seeds(brokers: &[String]) -> anyhow::Result<Vec<Seed>> {
    brokers
        .iter()
        .map(|broker| {
            broker
                .parse::<Seed>()
                .mark(Exit::Usage, "read the broker address")
        })
        .collect()
}

/// Resolve `host:port` strings. Every name is resolved and every address it
/// yields is kept, so a DNS name for several brokers works as a seed list.
pub(crate) async fn addresses(brokers: &[String]) -> anyhow::Result<Vec<SocketAddr>> {
    let mut addrs = Vec::new();
    for broker in brokers {
        if !broker.contains(':') {
            return Err(fail(
                Exit::Usage,
                format!("broker address {broker:?} needs a port, as host:port"),
            ));
        }
        let resolved = tokio::net::lookup_host(broker.as_str())
            .await
            .mark(Exit::Connection, format!("resolve {broker}"))?;
        addrs.extend(resolved);
    }
    Ok(addrs)
}

fn quic_tls(settings: &Settings) -> anyhow::Result<quinn::ClientConfig> {
    let roots = settings
        .ca_file
        .as_deref()
        .map(felix_client::root_store_from_pem_file)
        .transpose()
        .mark(Exit::Usage, "read the CA file")?
        .map(Arc::new);
    match (&settings.client_cert_file, &settings.client_key_file) {
        (None, None) => felix_client::quic_client_config(roots, settings.alpn),
        (Some(cert), Some(key)) => {
            let identity = ClientIdentity::from_pem_files(cert, key)
                .mark(Exit::Usage, "read the client certificate")?;
            felix_client::quic_client_config_with_identity(roots, identity, settings.alpn)
        }
        _ => {
            return Err(fail(
                Exit::Usage,
                "a client certificate needs both --client-cert-file and --client-key-file",
            ));
        }
    }
    .mark(Exit::Usage, "set up TLS")
}

#[cfg(test)]
mod tests;