Struct Etcd 

pub struct Etcd { /* private fields */ }

A key in etcd, as a configuration source.

Implementations

impl Etcd

pub async fn new<E, S>( endpoints: E, keys: impl Into<Keys>, ) -> Result<Self, Error>

where E: IntoIterator<Item = S>, S: Into<String>,

Connects to endpoints and reads keys.

keys is a key — "myapp/db.json" — or a Keys, for the several-keys and prefix forms.

The format is taken from the key’s extension — myapp/db.json is JSON. A key without one, and every prefix, needs with_format.

Errors

If the endpoints cannot be parsed. Not if they are unreachable: the client connects lazily, so that surfaces on the first fetch.

pub async fn with_options<E, S>( endpoints: E, keys: impl Into<Keys>, options: ConnectOptions, ) -> Result<Self, Error>

where E: IntoIterator<Item = S>, S: Into<String>,

As new, with etcd’s own connection options.

This is where authentication and TLS live, because that is where etcd-client puts them — there is no second vocabulary to learn, and options this crate has never heard of keep working.

let etcd = Etcd::with_options(
    ["https://etcd.internal:2379"],
    "myapp/db.json",
    ConnectOptions::new()
        .with_user("myapp", std::env::var("ETCD_PASSWORD").unwrap())
        .with_keep_alive(
            std::time::Duration::from_secs(30),
            std::time::Duration::from_secs(5),
        ),
)
.await?;

The credentials live in the client afterwards, which is what lets an expired auth token be replaced without rebuilding anything.

Errors

As new.

pub async fn with_tls<E, S>( endpoints: E, keys: impl Into<Keys>, options: ConnectOptions, tls: &TlsConfig, ) -> Result<Self, Error>

where E: IntoIterator<Item = S>, S: Into<String>,

Available on crate feature tls only.

As with_options, with a private certificate authority or a client certificate from the shared vocabulary.

The same three settings, spelled the same way, in all seven store crates — and spelled as data, so nothing here names a tonic type:

let etcd = Etcd::with_tls(
    ["https://etcd.internal:2379"],
    "myapp/db.json",
    ConnectOptions::new().with_user("myapp", std::env::var("ETCD_PASSWORD").unwrap()),
    &TlsConfig::new()
        .with_ca_certificate_file("/etc/etcd/ca.pem")
        .with_client_certificate_files("/etc/etcd/client.crt", "/etc/etcd/client.key"),
)
.await?;

etcd expresses all of it: a CA from a file or from bytes, and a client certificate from either. mTLS is not an afterthought here the way it is for the HTTP stores — an etcd cluster with --client-cert-auth is the ordinary hardened deployment.

options carries everything that is not TLS: the user and password, keep-alive, whatever etcd-client grows next. The tls argument owns the TLS slot. If options also carries a TlsOptions of its own, this one replaces it — etcd-client exposes no way to ask whether that slot is already filled, so the interaction is documented rather than refused. Use one door or the other, never both.

There is no way to turn verification off; TlsConfig’s own documentation argues that one, and tonic offers no such switch to forward even if this crate wanted to.

Errors

If a PEM file cannot be read, if what was read is not PEM, or as new.

pub fn from_client(client: Client, keys: impl Into<Keys>) -> Self

Uses a client the program already has.

For a caller that already talks to etcd and would rather not open a second connection to it. The client is Clone — cheaply, it is a handle — so sharing one costs nothing.

let etcd = Etcd::from_client(client, "myapp/db.json");

A shared client recovers from an expired auth token like any other: the credentials live in the client, so refreshing the token needs nothing this source would have to own.

pub fn with_timeout(self, timeout: Duration) -> Self

How long a single fetch may take before it is given up on. Ten seconds by default.

The deadline for one fetch attempt, excluding retries the underlying client performs — the same sentence every store in this family answers to.

It is applied here as a tokio::time::timeout around the request rather than through ConnectOptions::with_timeout, and the difference is the whole point: etcd’s own option bounds connecting, and a connection that was established minutes ago cannot be bounded by it. A member that accepts the request and then never answers is the failure worth having a deadline for, and only the wrap catches it.

It does not cover watch, which is long-lived by definition; a watch that stops after ten seconds would be a watch that does not work. It does bound each range read a prefix watch performs in answer to an event — that is a request like any other, and one that hangs would wedge the loop for good.

pub fn reporting_to(self, sink: RemoteSink) -> Self

Reports the watch loop’s failed attempts to sink.

Without this a watch is the half of a store dynamic-config cannot see. RemoteSink::apply records a delivery, so a working watch keeps the status current — but a loop whose stream broke, whose watch was cancelled or whose credential was refused delivers nothing, and so says nothing: dynamic_config_remote_up reports the last delivery rather than the last attempt, and a store that stopped answering an hour ago looks healthy until something calls refresh_remote_async().

let sink = DbConfig::remote_sink();

// The same sink delivers and reports: one generation, one fence.
etcd.reporting_to(sink)
    .watch(move |document| sink.apply(document))
    .await

A sink is Copy and captures its source’s generation when it is taken, which is what keeps a loop winding down after its source was replaced from charging its failures to the replacement — so take it once, where the watch is wired, exactly as the delivering half already does.

Only the watch. A fetch records itself through refresh_remote_async() already, and what is reported here is the failure streak and the last failure and nothing else: the staleness clock keeps ageing while remote_up goes to zero, which is the pair an alert wants.

The error’s kind is what travels. Nothing that names this store — no endpoint, no key, no credential — enters a RemoteStatus.

pub fn with_format(self, format: Format) -> Self

States the format, for a key whose name does not.

Required for Keys::Prefix — a prefix has no extension — and it also settles a list whose keys name two different formats.

pub async fn watch<F>(&self, on_change: F) -> Result<(), Error>

where F: FnMut(Fetched) -> Result<(), Error> + Send,

Calls on_change every time what this source reads moves, forever.

One key or a prefix. A prefix watch is the multi-key case that can be answered honestly: etcd’s watch says the range moved and carries the revision it moved at, and one range read at that revision is the whole set as of one instant. So the document delivered is a state the cluster really was in, never a merge of one key’s new value with another’s old one. A named list is still refused; the reason is on Keys::Several.

The first call happens when the first change arrives, not at startup: a watch reports changes, and reporting the current value as one would make every restart look like an edit. Fetch first if the starting value matters, which it usually does:

sink.apply(etcd.fetch().await?)?;
etcd.watch(move |document| sink.apply(document)).await

Cancellation is dropping the future. There is no stop flag, because there is nothing to poll one between: this suspends on the stream, so any executor’s cancellation already ends it immediately.

A deletion is not a change this reports for a single key. The key holding no value is not a configuration, and calling back with the last one — or with nothing — would both be worse than leaving the running snapshot alone. Under a prefix a deletion is a change like any other: the set is what it is after the delete, and the re-read reports it — unless nothing is left under the prefix, which is the same no-configuration case and is skipped for the same reason.

Errors

If the watch cannot be established, if the connection fails or ends, if etcd cancels the watch — compaction is the usual reason — or if on_change returns an error, which ends the watch, so a caller that wants to survive a bad document should log it and return Ok.

Under a prefix, also if the range read at an event’s revision fails, or if two keys under the prefix supply the same path — that is a deployment bug rather than a blip, and retrying it forever with nothing said would leave the configuration frozen and silent.

This never returns Ok: a watch either runs or has failed, and a silent success would leave a spawned task finished and a configuration frozen with nothing said about either. Callers that want to reconnect should loop around it.

Every one of those failures is also reported to the sink reporting_to was given, if one was — because a watch is normally spawned and its JoinHandle dropped, so the error returned here has nowhere else to go. The one failure not charged to the store is on_change’s own refusal: the store answered, apply recorded the delivery, and whether the document then installs is ConfigStatus’s business.

Trait Implementations

impl AsyncRemoteSource for Etcd

fn fetch( &self, ) -> Pin<Box<dyn Future<Output = Result<Fetched, Error>> + Send + '_>>

Reads the current document.

fn describe(&self) -> String

How to name this source in an error or a report.

impl Debug for Etcd

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.