EV-Etcd
view release on metacpan or search on metacpan
.github/workflows/ci.yml view on Meta::CPAN
run: |
ARCH=$(uname -m | sed 's/x86_64/amd64/;s/aarch64/arm64/')
curl -fsSL "https://github.com/etcd-io/etcd/releases/download/v3.5.26/etcd-v3.5.26-linux-${ARCH}.tar.gz" | tar xz
sudo mv "etcd-v3.5.26-linux-${ARCH}/etcd" /usr/local/bin/
sudo mv "etcd-v3.5.26-linux-${ARCH}/etcdctl" /usr/local/bin/
- name: Start etcd
run: |
etcd --data-dir /tmp/etcd-data &
for i in $(seq 1 30); do
etcdctl endpoint health 2>/dev/null && break
sleep 1
done
etcdctl endpoint health
- name: Test
env:
EV_ETCD_TEST_ETCD: 1
ETCD_TEST_AUTH_ENABLE_DISABLE: 1
run: make test
test-debian:
runs-on: ubuntu-latest
container: debian:bookworm
.github/workflows/ci.yml view on Meta::CPAN
run: |
ARCH=$(uname -m | sed 's/x86_64/amd64/;s/aarch64/arm64/')
curl -fsSL "https://github.com/etcd-io/etcd/releases/download/v3.5.26/etcd-v3.5.26-linux-${ARCH}.tar.gz" | tar xz
mv "etcd-v3.5.26-linux-${ARCH}/etcd" /usr/local/bin/
mv "etcd-v3.5.26-linux-${ARCH}/etcdctl" /usr/local/bin/
- name: Start etcd
run: |
etcd --data-dir /tmp/etcd-data > /tmp/etcd.log 2>&1 &
for i in $(seq 1 30); do
etcdctl endpoint health 2>/dev/null && break
sleep 1
done
etcdctl endpoint health
- name: Test
env:
EV_ETCD_TEST_ETCD: 1
ETCD_TEST_AUTH_ENABLE_DISABLE: 1
run: make test
test-macos:
runs-on: macos-latest
strategy:
.github/workflows/ci.yml view on Meta::CPAN
ARCH=$(uname -m | sed 's/x86_64/amd64/;s/aarch64/arm64/;s/arm64/arm64/')
curl -fsSL "https://github.com/etcd-io/etcd/releases/download/v3.5.26/etcd-v3.5.26-darwin-${ARCH}.zip" -o etcd.zip
unzip -q etcd.zip
sudo mv "etcd-v3.5.26-darwin-${ARCH}/etcd" /usr/local/bin/
sudo mv "etcd-v3.5.26-darwin-${ARCH}/etcdctl" /usr/local/bin/
- name: Start etcd
run: |
etcd --data-dir /tmp/etcd-data &
for i in $(seq 1 30); do
etcdctl endpoint health 2>/dev/null && break
sleep 1
done
etcdctl endpoint health
- name: Test
env:
EV_ETCD_TEST_ETCD: 1
ETCD_TEST_AUTH_ENABLE_DISABLE: 1
run: make test
test-freebsd:
runs-on: ubuntu-latest
name: FreeBSD 14.2
.github/workflows/ci.yml view on Meta::CPAN
usesh: true
prepare: |
pkg install -y grpc protobuf-c pkgconf perl5 p5-EV gmake
pkg install -y etcd3 || pkg install -y etcd || true
run: |
perl Makefile.PL
gmake
if command -v etcd >/dev/null 2>&1; then
etcd --data-dir /tmp/etcd-data &
sleep 5
etcdctl endpoint health 2>/dev/null || true
fi
gmake test
test-openbsd:
runs-on: ubuntu-latest
name: OpenBSD
continue-on-error: true
steps:
- uses: actions/checkout@v7
- uses: vmactions/openbsd-vm@v1
.github/workflows/ci.yml view on Meta::CPAN
run: |
ARCH=$(uname -m | sed 's/x86_64/amd64/;s/aarch64/arm64/')
curl -fsSL "https://github.com/etcd-io/etcd/releases/download/v3.5.26/etcd-v3.5.26-linux-${ARCH}.tar.gz" | tar xz -C "$RUNNER_TEMP"
sudo mv "$RUNNER_TEMP/etcd-v3.5.26-linux-${ARCH}/etcd" /usr/local/bin/
sudo mv "$RUNNER_TEMP/etcd-v3.5.26-linux-${ARCH}/etcdctl" /usr/local/bin/
- name: Start etcd
run: |
etcd --data-dir /tmp/etcd-data &
for i in $(seq 1 30); do
etcdctl endpoint health 2>/dev/null && break
sleep 1
done
etcdctl endpoint health
- name: Dist from clean checkout
run: |
perl Makefile.PL
make
make disttest
- name: Author tests
run: prove -b xt/
Revision history for EV-Etcd
0.11 2026-10-07
- Fixes for watch replay, failover, leaderless members, reconnects, crashes
- Invalid arguments and unknown options croak
- max_retries defaults to 30: streams ride out about two minutes of outage
0.10 2026-09-23
- Add TLS and mutual TLS (tls_* options, https:// endpoints)
- Fail over to the next endpoint when the current one is unreachable
- Keepalive pings (keepalive_time, keepalive_timeout)
- Forked children can use etcd if the parent held no client, else croak
- health_interval accepts fractions
- Fix crashes and leaks on client/handle destroy, txn and ithreads
0.09 2026-07-21
- Fix lease_keepalive never renewing the lease (resend at TTL/3)
- Reconnect streams on any stream end, not only batch failure
- Fix NULL EV API table crash in stream reconnect timers
- Fix cancel() during reconnect backoff leaking the handle
- KV operations: get, put, delete, range, txn (compare-and-swap)
- Watch with bidirectional streaming and auto-reconnect
- Lease: grant, revoke, keepalive, time-to-live, leases
- Lock and unlock tied to leases
- Election: campaign, proclaim, leader, resign, observe
- Cluster: member list/add/remove/update/promote
- Maintenance: status, compact, defragment, alarm, hash_kv, move_leader
- Auth: user/role management, authenticate, enable/disable
- Health monitoring with configurable interval and callback
- Automatic retries for transient gRPC failures
- Multiple endpoint support with failover
- Structured error callbacks ({code, message, source})
}
static int64_t hv_fetch_i64(pTHX_ HV *hv, const char *key, I32 klen) {
SV **svp = hv_fetch(hv, key, klen, 0);
if (!svp)
return 0;
SvGETMAGIC(*svp);
return SvOK(*svp) ? SvI64_nomg(*svp) : 0;
}
static int endpoint_scheme_len(const char *ep, int *is_https) {
*is_https = strnEQ(ep, "https://", 8);
return *is_https ? 8 : strnEQ(ep, "http://", 7) ? 7 : 0;
}
static void process_txn_response(pTHX_ pending_call_t *pc);
static void process_auth_response(pTHX_ pending_call_t *pc);
static void process_user_add_response(pTHX_ pending_call_t *pc);
static void process_user_delete_response(pTHX_ pending_call_t *pc);
static void process_user_change_password_response(pTHX_ pending_call_t *pc);
static void process_auth_enable_response(pTHX_ pending_call_t *pc);
(void)revents;
if (!client->active) {
return;
}
grpc_connectivity_state state = grpc_channel_check_connectivity_state(client->channel, 0);
int was_healthy = client->is_healthy;
int is_healthy = state != GRPC_CHANNEL_TRANSIENT_FAILURE && state != GRPC_CHANNEL_SHUTDOWN;
int checked = client->current_endpoint;
if (was_healthy != is_healthy) {
client->is_healthy = is_healthy;
if (!is_healthy && client->endpoint_count > 1) {
etcd_rotate_endpoint(client);
}
if (client->health_callback) {
dSP;
CALLBACK_WINDOW_BEGIN(client);
ENTER;
SAVETMPS;
PUSHMARK(SP);
EXTEND(SP, 2);
PUSHs(sv_2mortal(newSViv(is_healthy)));
PUSHs(sv_2mortal(newSVpv(client->endpoints[checked], 0)));
PUTBACK;
CALL_SV_SAFE(client->health_callback, G_DISCARD);
FREETMPS;
LEAVE;
if (CALLBACK_WINDOW_END(client))
return;
}
}
}
/* Before the handler: a DESTROY from its callback frees what is still listed */
unlink_pending_call(pc);
/* Freed at scope end, which an exit() from the callback also reaches */
ENTER;
SAVEDESTRUCTOR_X(free_handled_call, pc);
/* A failed batch can leave the zero-initialised GRPC_STATUS_OK */
grpc_status_code status = !success && pc->status == GRPC_STATUS_OK
? GRPC_STATUS_UNAVAILABLE : pc->status;
/* Before the callback, so a retry from it reaches the next endpoint */
etcd_endpoint_failed(client, &pc->base, status, pc->status_details);
if (success) {
switch (pc->base.type) {
case CALL_TYPE_RANGE:
process_range_response(aTHX_ pc);
break;
case CALL_TYPE_PUT:
process_put_response(aTHX_ pc);
break;
case CALL_TYPE_DELETE:
init_method_slices();
ev_etcd_pid = getpid();
pthread_atfork(ev_etcd_atfork_prepare, NULL, ev_etcd_atfork_child);
SV *
ev_etcd_new(class, ...)
char *class
CODE:
{
ev_etcd_t *client;
AV *endpoints_av = NULL;
int timeout_seconds = 30;
int max_retries = 30;
NV health_interval = 0;
NV keepalive_time = 10, keepalive_timeout = 10;
SV *health_callback = NULL;
char *init_auth_token = NULL;
STRLEN init_auth_token_len = 0;
int tls = 0, saw_http = 0;
const char *tls_ca_file = NULL, *tls_cert_file = NULL, *tls_key_file = NULL;
const char *tls_server_name = NULL;
SV *ca_pem = NULL, *cert_pem = NULL, *key_pem = NULL;
int i;
if ((items - 1) % 2) croak("Odd number of options in EV::Etcd->new");
for (i = 1; i < items; i += 2) {
if (i + 1 < items) {
const char *key = SvPV_nolen(ST(i));
SV *val = ST(i + 1);
SvGETMAGIC(val);
if (strEQ(key, "endpoints")) {
if (SvOK(val)) {
if (!SvROK(val) || SvTYPE(SvRV(val)) != SVt_PVAV)
croak("EV::Etcd: endpoints must be an array reference");
endpoints_av = (AV *)sv_2mortal(SvREFCNT_inc(SvRV(val)));
if (av_len(endpoints_av) < 0)
croak("EV::Etcd: endpoints is empty");
}
} else if (strEQ(key, "timeout")) {
if (SvOK(val)) {
IV v = SvIV_nomg(val);
timeout_seconds = v < 1 ? 1 : v > INT_MAX ? INT_MAX : (int)v;
}
} else if (strEQ(key, "max_retries")) {
if (SvOK(val)) {
IV v = SvIV_nomg(val);
max_retries = v < 0 ? 0 : v > INT_MAX ? INT_MAX : (int)v;
} else if (strEQ(key, "keepalive_timeout")) {
if (SvOK(val)) keepalive_timeout = SvNV_nomg(val);
} else {
croak("Unknown option '%s' in EV::Etcd->new", key);
}
}
}
/* Only these copies are used: a tied array could hand back something else */
AV *checked = NULL;
if (endpoints_av && av_len(endpoints_av) >= 0) {
int count = av_len(endpoints_av) + 1;
checked = (AV *)sv_2mortal((SV *)newAV());
for (i = 0; i < count; i++) {
SV **ep = av_fetch(endpoints_av, i, 0);
STRLEN len;
const char *str;
int is_https, skip;
if (ep)
SvGETMAGIC(*ep);
if (!ep || !SvOK(*ep))
croak("EV::Etcd: endpoints element %d is undefined", i);
str = SvPV_nomg(*ep, len);
VALIDATE_URL_SIZE(len);
VALIDATE_NO_NUL(str, len, "endpoint");
skip = endpoint_scheme_len(str, &is_https);
if (len <= (STRLEN)skip)
croak("EV::Etcd: endpoints element %d is empty", i);
av_push(checked, newSVpvn(str + skip, len - skip));
if (!skip)
continue;
if (is_https)
tls = 1;
else
saw_http = 1;
}
}
if (tls_ca_file || tls_cert_file || tls_key_file || tls_server_name)
tls = 1;
if (tls && saw_http)
croak("EV::Etcd: http:// endpoint on a TLS client");
if (!tls_cert_file != !tls_key_file)
croak("EV::Etcd: tls_cert_file and tls_key_file must be given together");
if (tls_ca_file) ca_pem = slurp_pem(aTHX_ "tls_ca_file", tls_ca_file);
if (tls_cert_file) cert_pem = slurp_pem(aTHX_ "tls_cert_file", tls_cert_file);
if (tls_key_file) key_pem = slurp_pem(aTHX_ "tls_key_file", tls_key_file);
/* Its fork-safety croak must come before anything is allocated */
grpc_acquire(aTHX);
Newxz(client, 1, ev_etcd_t);
if (checked) {
int count = av_len(checked) + 1;
Newx(client->endpoints, count, char *);
client->endpoint_count = count;
for (i = 0; i < count; i++) {
SV *ep = *av_fetch(checked, i, 0);
client->endpoints[i] = savepvn(SvPVX(ep), SvCUR(ep));
}
} else {
Newx(client->endpoints, 1, char *);
client->endpoints[0] = savepv("127.0.0.1:2379");
client->endpoint_count = 1;
}
client->current_endpoint = 0;
client->keepalive_ms = seconds_to_ms(keepalive_time);
client->keepalive_timeout_ms = seconds_to_ms(keepalive_timeout);
if (!client->keepalive_timeout_ms)
client->keepalive_timeout_ms = 10000;
if (tls) {
grpc_ssl_pem_key_cert_pair pair;
if (cert_pem) {
pair.private_key = SvPV_nolen(key_pem);
pair.cert_chain = SvPV_nolen(cert_pem);
}
client->creds = grpc_ssl_credentials_create(
ca_pem ? SvPV_nolen(ca_pem) : NULL, cert_pem ? &pair : NULL, NULL, NULL);
if (!client->creds) {
for (int j = 0; j < client->endpoint_count; j++) {
Safefree(client->endpoints[j]);
}
Safefree(client->endpoints);
Safefree(client);
grpc_release();
croak("EV::Etcd: failed to create TLS credentials");
}
if (key_pem)
memset(SvPVX(key_pem), 0, SvCUR(key_pem));
if (tls_server_name)
client->tls_server_name = savepv(tls_server_name);
}
client->channel_ref = etcd_create_channel(client, client->endpoints[0]);
if (!client->channel_ref) {
for (int j = 0; j < client->endpoint_count; j++) {
Safefree(client->endpoints[j]);
}
Safefree(client->endpoints);
if (client->creds) grpc_channel_credentials_release(client->creds);
Safefree(client->tls_server_name);
Safefree(client);
grpc_release();
croak("Failed to create gRPC channel");
}
client->channel = client->channel_ref->channel;
client->cq = grpc_completion_queue_create_for_next(NULL);
WITH_SIGNALS_BLOCKED(thread_error = pthread_create(&client->cq_thread, NULL, cq_thread_func, client));
if (thread_error) {
ev_async_stop(EV_DEFAULT, &client->cq_async);
pthread_mutex_destroy(&client->queue_mutex);
grpc_completion_queue_shutdown(client->cq);
while (grpc_completion_queue_next(client->cq,
gpr_inf_past(GPR_CLOCK_REALTIME), NULL).type != GRPC_QUEUE_SHUTDOWN)
;
grpc_completion_queue_destroy(client->cq);
etcd_channel_release(client->channel_ref, 1);
for (int j = 0; j < client->endpoint_count; j++) {
Safefree(client->endpoints[j]);
}
Safefree(client->endpoints);
if (client->creds) grpc_channel_credentials_release(client->creds);
Safefree(client->tls_server_name);
Safefree(client);
grpc_release();
croak("Failed to create gRPC completion queue thread");
}
if (init_auth_token && init_auth_token_len > 0) {
Newx(client->auth_token, init_auth_token_len + 1, char);
Copy(init_auth_token, client->auth_token, init_auth_token_len, char);
if (client->health_callback) {
SvREFCNT_dec(client->health_callback);
client->health_callback = NULL;
}
clear_auth_token(client);
Safefree(client->tls_server_name);
client->tls_server_name = NULL;
if (client->endpoints) {
int i;
for (i = 0; i < client->endpoint_count; i++) {
if (client->endpoints[i]) {
Safefree(client->endpoints[i]);
}
}
Safefree(client->endpoints);
client->endpoints = NULL;
}
/* Inside a callback, finish_client_destroy frees it later */
if (!client->in_callback) {
Safefree(client);
}
}
MODULE = EV::Etcd PACKAGE = EV::Etcd::Watch PREFIX = ev_etcd_watch_
NAME
EV::Etcd - Async etcd v3 client using native gRPC and EV/libev
SYNOPSIS
use v5.10;
use EV;
use EV::Etcd;
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
);
$client->put('/my/key', 'value', sub {
my ($resp, $err) = @_;
die $err->{message} if $err;
say "Put succeeded, revision: $resp->{header}{revision}";
});
$client->get('/my/key', sub {
my ($resp, $err) = @_;
receives "($response, $error)": the response hash and "undef", or
"undef" and an error hash (see "ERRORS"). Responses carry a "header"
with "cluster_id", "member_id", "revision" and "raft_term" unless noted
otherwise. A kv hash has "key", "value", "create_revision",
"mod_revision", "version" and "lease".
CONSTRUCTOR
new
my $client = EV::Etcd->new(%options);
endpoints => [ 'host:port', ... ]
Default "['127.0.0.1:2379']"; an empty list croaks. An endpoint may
carry the "http://" or "https://" scheme etcd prints; "https://"
turns on "tls".
The client uses one endpoint at a time and moves to the next when it
cannot be reached: a unary call fails with UNAVAILABLE, or with
DEADLINE_EXCEEDED before a connection was made, a stream has to
reconnect because its connection is down, or a keepalive ping goes
unanswered. The failing call still reports its error; retrying it
reaches the next endpoint. Several failures on one endpoint move the
client once.
A member that has lost its leader fails linearizable reads and
writes after etcd's request timeout (7 seconds by default), and
these failures move the client on too, though a shorter "timeout"
ends the call on the client first and does not. Errors a working
member returns during an election, such as "etcdserver: leader
changed", do not move it.
Streams need a leader as well: a member without one refuses new
streams and, after a few election timeouts, ends running watches and
keepalives with UNAVAILABLE "etcdserver: no leader". Streams retry
such answers a second apart until a leader is back, on the next
endpoint when there are several; each restarts the "max_retries"
count. Once a unary call, or a stream that was running, reports
"etcdserver: no leader", pending "lock" and "election_campaign"
calls sent over the same connection fail with UNAVAILABLE, as they
do when a connection that never became ready is abandoned.
An endpoint may also be a gRPC target listing several addresses,
such as "ipv4:10.0.0.1:2379,10.0.0.2:2379". gRPC moves between them
within one connection, so a refused address fails no call, but never
away from a member that has lost its leader: list members as
separate endpoints to leave one.
gRPC connects through a proxy named in "grpc_proxy", "https_proxy"
or "http_proxy", even to a loopback address; list etcd's hosts in
"no_proxy" to connect directly.
timeout => $seconds
RPC timeout in whole seconds, at least 1; default 30. "lock" and
"election_campaign" wait without one, as a timeout could not tell
whether they succeeded.
max_retries => $count
Reconnection attempts for a stream ("watch", "lease_keepalive",
"election_observe") after a connection failure; default 30, 0
disables reconnecting. Attempts are 0.5 seconds further apart each
time, up to 5 seconds, so the default gives up about two minutes
after the endpoints start refusing connections, plus gRPC's
20-second connect timeout for one that accepts them but stays
silent. Lower it to learn sooner that the endpoints are gone.
keepalive_time => $seconds
keepalive_timeout => $seconds
Seconds between pings while calls or streams are open (default 10,
fractions allowed, 0 disables them), and how long a ping may go
unanswered (default 10). An unanswered ping closes the connection:
its calls fail with UNAVAILABLE and the client moves to the next
endpoint. Without pings, an endpoint that goes silent (a hung host,
a partition) goes unnoticed until the operating system gives up on
the connection. etcd closes connections that ping more often than
its "--grpc-keepalive-min-time" (5 seconds by default).
health_interval => $seconds
on_health_change => sub { my ($healthy, $endpoint) = @_; ... }
Every "health_interval" seconds (fractions allowed; default 0, off)
the client checks its connection state without sending a request,
and calls "on_health_change" when health changes. Only a failed
connection attempt counts as unhealthy; with several endpoints it
also moves the client on. Switching endpoints after a failed call is
not reported.
auth_token => $token
A token from an earlier "authenticate", used without authenticating
again.
tls => $bool
Connect with TLS; any "tls_*" option or an "https://" endpoint turns
it on too. Without "tls_ca_file", gRPC uses its default roots, or
the PEM bundle named by "GRPC_DEFAULT_SSL_ROOTS_FILE_PATH".
tls_ca_file => $path
PEM CA certificates to verify the servers with (etcd's
"--trusted-ca-file"), in place of the default roots. Verification
cannot be turned off; for a self-signed server, give its certificate
here.
tls_cert_file => $path
tls_key_file => $path
Client certificate and key, given together, for servers started with
"--client-cert-auth".
tls_server_name => $name
Name to verify the server certificate against and send as SNI, in
place of the endpoint's host.
my $client = EV::Etcd->new(
endpoints => ['https://10.0.0.1:2379', 'https://10.0.0.2:2379'],
tls_ca_file => '/etc/etcd/ca.crt',
tls_cert_file => '/etc/etcd/client.crt',
tls_key_file => '/etc/etcd/client.key',
);
ERRORS
{
code => 14, # gRPC status code
status => 'UNAVAILABLE',
message => 'Connection refused',
"retryable" is set for UNAVAILABLE, ABORTED, DEADLINE_EXCEEDED and
RESOURCE_EXHAUSTED "etcdserver: too many requests", but never for a
failed "lock" or "election_campaign". Other RESOURCE_EXHAUSTED errors
last until someone intervenes, such as "etcdserver: mvcc: database space
exceeded" (see "alarm"). Retryable means transient, not without effect:
a write can be applied after its deadline has passed, so retrying
"lease_grant", "lease_revoke" or "delete" can apply it twice or report
what the first attempt did.
Unary calls are not retried. Streams reconnect on their own (see
"max_retries"), to the next endpoint once the current one has failed,
and the count restarts once a stream is working again. A stream reports
an error when reconnecting is disabled or exhausted, at once for a
status a new connection cannot fix (UNAUTHENTICATED, PERMISSION_DENIED,
INVALID_ARGUMENT, NOT_FOUND, ALREADY_EXISTS, FAILED_PRECONDITION,
OUT_OF_RANGE, UNIMPLEMENTED), and for an error the server sends on it,
such as a cancelled or compacted watch or an expired lease, which ends
it.
The server can still clean up after a failed "lock" or
"election_campaign" has reported, deleting the key it made on the lease.
use Time::HiRes qw(time);
use lib 'blib/lib', 'blib/arch';
$| = 1; # Autoflush
use EV;
use EV::Etcd;
# Check if etcd is running
my $etcd_running = 0;
eval {
my $result = `etcdctl endpoint health 2>&1`;
$etcd_running = 1 if $result =~ /is healthy/;
};
die "etcd not running\n" unless $etcd_running;
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
);
my $prefix = "/bench_$$";
my $iterations = $ENV{BENCH_ITER} || 1000;
print "EV::Etcd Benchmark\n";
print "==================\n\n";
# Benchmark 1: Sequential puts
{
eg/auth_test.pl view on Meta::CPAN
# 2. Or create a non-root user:
# etcdctl user add testuser:testpass
# etcdctl role add testrole
# etcdctl role grant-permission testrole readwrite /authtest --prefix
# etcdctl user grant-role testuser testrole
my $username = $ENV{ETCD_USER} // 'root';
my $password = $ENV{ETCD_PASSWORD} // 'rootpassword';
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
);
print "=== Authentication Test ===\n\n";
# Step 1: Authenticate
print "1. Authenticating as '$username'...\n";
$client->authenticate($username, $password, sub {
my ($resp, $err) = @_;
if ($err) {
eg/batch_watch_example.pl view on Meta::CPAN
my $prefix = "/config/app1/";
# Track state
my $watcher1_batches = 0;
my $watcher2_batches = 0;
my $writes_done = 0;
my $cleaning_up = 0;
# === Client 1: First watcher ===
my $watcher_client1 = EV::Etcd->new(endpoints => ['127.0.0.1:2379']);
my $watch1 = $watcher_client1->watch($prefix, { prefix => 1 }, sub {
my ($resp, $err) = @_;
if ($err) {
print "Watcher 1 error: $err->{message}\n";
return;
}
$watcher1_batches++;
my $events = $resp->{events} || [];
eg/batch_watch_example.pl view on Meta::CPAN
my $type = $event->{type} // 'PUT';
my $key = $event->{kv}{key} // '';
my $value = $event->{kv}{value} // '';
print " - $type: $key = $value\n";
}
check_done();
});
# === Client 2: Second watcher ===
my $watcher_client2 = EV::Etcd->new(endpoints => ['127.0.0.1:2379']);
my $watch2 = $watcher_client2->watch($prefix, { prefix => 1 }, sub {
my ($resp, $err) = @_;
if ($err) {
print "Watcher 2 error: $err->{message}\n";
return;
}
$watcher2_batches++;
my $events = $resp->{events} || [];
eg/batch_watch_example.pl view on Meta::CPAN
my $type = $event->{type} // 'PUT';
my $key = $event->{kv}{key} // '';
my $value = $event->{kv}{value} // '';
print " - $type: $key = $value\n";
}
check_done();
});
# === Client 3: Writer (performs transactional batch write) ===
my $writer_client = EV::Etcd->new(endpoints => ['127.0.0.1:2379']);
# Wait a moment for watches to be established, then write
my $write_timer = EV::timer(0.5, 0, sub {
print "Writer: Performing transactional batch write to $prefix\n";
# Transaction that writes 5 keys atomically
# All watchers will receive these as ONE batch
$writer_client->txn(
# No compare conditions - always succeeds
compare => [],
eg/crud_test.pl view on Meta::CPAN
#!/usr/bin/env perl
use strict;
use warnings;
use lib 'blib/lib', 'blib/arch';
use EV;
use EV::Etcd;
use Data::Dumper;
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
);
print "=== CRUD Test ===\n\n";
# Step 1: Put
print "1. PUT /test = 'hello world'\n";
$client->put('/test', 'hello world', sub {
my ($resp, $err) = @_;
die "Put error: $err->{message}" if $err;
print " Put OK, revision: $resp->{header}{revision}\n\n";
eg/distributed_mutex.pl view on Meta::CPAN
use warnings;
use lib 'blib/lib', 'blib/arch';
use EV;
use EV::Etcd;
my $name = $ARGV[0] // '/jobs/example';
my $work_msg = $ARGV[1] // 'critical section';
my $work_time = $ARGV[2] // 5; # seconds the lock is held
my $lock_ttl = 10; # seconds; must exceed work_time + slack
my $client = EV::Etcd->new(endpoints => ['127.0.0.1:2379'], max_retries => 5);
my ($lease_id, $lock_key, $keepalive, $work_timer, $lease_died, $acquire_timer);
# 1. Lease for the lock — if we crash, the lock auto-releases after lock_ttl
$client->lease_grant($lock_ttl, sub {
my ($r, $err) = @_;
die "lease_grant: $err->{message}\n" if $err;
$lease_id = $r->{id};
say "[$$] lease=$lease_id ttl=${lock_ttl}s";
# 2. Refresh the lease while we work
eg/leader_cron.pl view on Meta::CPAN
use lib 'blib/lib', 'blib/arch';
use EV;
use EV::Etcd;
my $election = "/cron/nightly-aggregator";
my $lease_ttl = 10; # seconds; shorter = faster failover, more chatter
my $tick_period = 2; # how often the leader runs the job
my $value = "$0/$$"; # what other observers see for this leader
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
max_retries => 5,
);
my ($lease_id, $leader, $keepalive, $tick_timer);
sub fail { my $m = shift; warn "[$$] $m\n"; cleanup_and_exit(1) }
# 1. Grant an election lease
$client->lease_grant($lease_ttl, sub {
my ($r, $err) = @_;
eg/lease_test.pl view on Meta::CPAN
#!/usr/bin/env perl
use strict;
use warnings;
use lib 'blib/lib', 'blib/arch';
use EV;
use EV::Etcd;
use Data::Dumper;
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
);
print "=== Lease Test ===\n\n";
my $lease_id;
my $keepalive_count = 0;
# Step 1: Grant a lease with 10 second TTL
print "1. Granting lease with TTL=10s...\n";
$client->lease_grant(10, sub {
eg/put_get_test.pl view on Meta::CPAN
#!/usr/bin/env perl
use strict;
use warnings;
use lib 'blib/lib', 'blib/arch';
use EV;
use EV::Etcd;
use Data::Dumper;
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
);
print "Created client\n";
# First, put a value
print "Putting key '/test' with value 'hello from perl'...\n";
$client->put('/test', 'hello from perl', sub {
my ($resp, $err) = @_;
eg/resumable_watch.pl view on Meta::CPAN
use strict;
use warnings;
use lib 'blib/lib', 'blib/arch';
use EV;
use EV::Etcd;
my $prefix = $ARGV[0] // '/myapp/config/';
my $state_dir = $ENV{RESUMABLE_WATCH_STATE_DIR} || '/tmp';
my $state_path = "$state_dir/resumable_watch_" . _safe_name($prefix) . ".rev";
my $client = EV::Etcd->new(endpoints => ['127.0.0.1:2379'], max_retries => 5);
my $last_rev = read_last_rev();
my ($watch, $retry_timer);
if ($last_rev) {
say "[resume] last seen revision: $last_rev, replaying what came after";
start_watch($last_rev + 1);
} else {
say "[resume] no saved revision, listing the current state";
relist();
}
eg/service_registry.pl view on Meta::CPAN
use EV;
use EV::Etcd;
my $service_type = $ARGV[0] // 'web';
my $self_id = sprintf "%s-%d", ($ENV{HOSTNAME} || 'localhost'), $$;
my $self_value = "host=$self_id pid=$$ started=" . time();
my $prefix = "/services/$service_type/";
my $self_key = "$prefix$self_id";
my $lease_ttl = 15;
my $client = EV::Etcd->new(endpoints => ['127.0.0.1:2379'], max_retries => 5);
my ($lease_id, $keepalive, $watch);
# 1. Lease — keys self-evict if we crash
$client->lease_grant($lease_ttl, sub {
my ($r, $err) = @_;
die "lease_grant: $err->{message}\n" if $err;
$lease_id = $r->{id};
say "[$self_id] lease=$lease_id ttl=${lease_ttl}s";
# 2. Heartbeat
eg/sync_and_watch_example.pl view on Meta::CPAN
use warnings;
use lib 'blib/lib', 'blib/arch';
use EV;
use EV::Etcd;
my $prefix = "/myapp/config/";
# === Setup: Create some initial data ===
print "=== Setup: Creating initial data ===\n";
my $setup_client = EV::Etcd->new(endpoints => ['127.0.0.1:2379']);
my $setup_done = 0;
# Use transaction to create initial config atomically
$setup_client->txn(
compare => [],
success => [
{ put => { key => "${prefix}db/host", value => "localhost" } },
{ put => { key => "${prefix}db/port", value => "5432" } },
{ put => { key => "${prefix}db/name", value => "myapp" } },
{ put => { key => "${prefix}cache/host", value => "localhost" } },
eg/sync_and_watch_example.pl view on Meta::CPAN
$setup_done = 1;
EV::break;
}
);
my $t_setup = EV::timer(5, 0, sub { die "Setup timeout" });
EV::run;
# === Client 1: Subscriber that loads then watches ===
print "=== Client 1: Loading existing data and watching ===\n";
my $subscriber = EV::Etcd->new(endpoints => ['127.0.0.1:2379']);
my %local_cache; # Local copy of the config branch
my $watch_revision; # Revision to start watching from
my $watch_handle;
my $load_done = 0;
# Step 1: Load all existing data from the branch
$subscriber->get($prefix, { prefix => 1 }, sub {
my ($resp, $err) = @_;
die "Load failed: $err->{message}" if $err;
eg/sync_and_watch_example.pl view on Meta::CPAN
print "\nNow watching for changes...\n";
EV::break;
});
my $t_load = EV::timer(5, 0, sub { die "Load timeout" });
EV::run;
# === Client 2: Publisher that makes changes ===
print "\n=== Client 2: Making changes ===\n";
my $publisher = EV::Etcd->new(endpoints => ['127.0.0.1:2379']);
my $changes_done = 0;
# Schedule a series of changes
my @changes = (
sub {
print "\nPublisher: Updating db/host...\n";
$publisher->put("${prefix}db/host", "db.production.local", sub {
my ($resp, $err) = @_;
print " Updated (rev $resp->{header}{revision})\n" unless $err;
EV::break;
eg/txn_test.pl view on Meta::CPAN
#!/usr/bin/env perl
use strict;
use warnings;
use lib 'blib/lib', 'blib/arch';
use EV;
use EV::Etcd;
use Data::Dumper;
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
);
print "=== Transaction Test ===\n\n";
# Test 1: Compare-and-swap - create key if it doesn't exist (version == 0)
print "1. Compare-and-swap: Create '/txntest' only if it doesn't exist...\n";
$client->txn(
# Compare: version == 0 means key doesn't exist
[
eg/watch_config_tree.pl view on Meta::CPAN
use warnings;
use lib 'blib/lib', 'blib/arch';
use EV;
use EV::Etcd;
use Data::Path::XS qw(path_get path_set path_delete);
use Data::Dumper;
my $prefix = "/myapp/";
my %config;
my $client = EV::Etcd->new(endpoints => ['127.0.0.1:2379']);
# Seed some data
print "=== Seeding config ===\n";
$client->txn(
compare => [],
success => [
{ put => { key => "${prefix}db/host", value => "localhost" } },
{ put => { key => "${prefix}db/port", value => "5432" } },
{ put => { key => "${prefix}cache/host", value => "redis.local" } },
{ put => { key => "${prefix}cache/ttl", value => "3600" } },
eg/watch_extract_fields.pl view on Meta::CPAN
use Data::Path::XS qw(pathc_get path_compile);
# Pre-compile paths for fields we extract from every event
my $p_type = path_compile('/type');
my $p_key = path_compile('/kv/key');
my $p_value = path_compile('/kv/value');
my $p_rev = path_compile('/kv/mod_revision');
my $p_lease = path_compile('/kv/lease');
my $p_prev = path_compile('/prev_kv/value');
my $client = EV::Etcd->new(endpoints => ['127.0.0.1:2379']);
my $prefix = "/events-demo/";
# Seed data
$client->put("${prefix}sensor/temp", "22.5", sub { EV::break });
my $t0 = EV::timer(5, 0, sub { die "timeout" });
EV::run;
print "=== Watching $prefix with prev_kv ===\n";
print "(Compiled paths extract fields from each event)\n\n";
eg/watch_test.pl view on Meta::CPAN
#!/usr/bin/env perl
use strict;
use warnings;
use lib 'blib/lib', 'blib/arch';
use EV;
use EV::Etcd;
use Data::Dumper;
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
);
print "=== Watch Test ===\n\n";
print "Watching key '/watchtest'...\n";
print "(In another terminal, run: etcdctl put /watchtest value1)\n\n";
my $event_count = 0;
$client->watch('/watchtest', sub {
my ($resp, $err) = @_;
etcd_common.c view on Meta::CPAN
}
static void check_leader_failure(ev_etcd_t *client, unsigned channel_gen,
grpc_status_code status, grpc_slice status_details) {
if (etcd_is_no_leader(status, status_details)) {
/* Retired channels too: another failure may have moved the client already */
cancel_blocking_calls(client, channel_gen, "etcdserver: no leader");
}
}
void etcd_rotate_endpoint(ev_etcd_t *client) {
int next = client->current_endpoint;
if (client->endpoint_count > 1)
next = (next + 1) % client->endpoint_count;
channel_ref_t *ref = etcd_create_channel(client, client->endpoints[next]);
if (!ref) return;
channel_ref_t *old = client->channel_ref;
unsigned old_gen = client->channel_gen;
int was_ready = grpc_channel_check_connectivity_state(old->channel, 0) == GRPC_CHANNEL_READY;
client->channel_ref = ref;
client->channel = ref->channel;
client->current_endpoint = next;
client->channel_gen++;
/* Calls without a deadline must not stay queued on a dead connection */
if (!was_ready) {
cancel_blocking_calls(client, old_gen, "Endpoint changed");
for (watch_call_t *wc = client->watches; wc; wc = wc->next) {
if (wc->active && wc->auto_reconnect && wc->call && wc->base.channel_gen == old_gen)
grpc_call_cancel_with_status(wc->call, GRPC_STATUS_UNAVAILABLE, "Endpoint changed", NULL);
}
for (keepalive_call_t *kc = client->keepalives; kc; kc = kc->next) {
etcd_common.c view on Meta::CPAN
if (status != GRPC_STATUS_UNAVAILABLE)
return 0;
for (int i = 0; messages[i]; i++) {
if (grpc_slice_eq(details, grpc_slice_from_static_string(messages[i])))
return 1;
}
return 0;
}
static int failed_on_current(ev_etcd_t *client, unsigned channel_gen) {
return client->endpoint_count > 1 && channel_gen == client->channel_gen;
}
static int connected(ev_etcd_t *client) {
return grpc_channel_check_connectivity_state(client->channel, 0) == GRPC_CHANNEL_READY;
}
/* etcd's own request timeout on a member without a leader, not the client's deadline */
static int member_timed_out(grpc_status_code status, grpc_slice details) {
if ((status == GRPC_STATUS_DEADLINE_EXCEEDED || status == GRPC_STATUS_UNKNOWN)
&& grpc_slice_eq(details, grpc_slice_from_static_string("context deadline exceeded")))
return 1;
return status == GRPC_STATUS_UNKNOWN
&& (grpc_slice_eq(details, grpc_slice_from_static_string("etcdserver: request timed out"))
|| grpc_slice_eq(details, grpc_slice_from_static_string(
"etcdserver: request timed out, possibly due to connection lost")));
}
/* Only failures on the current channel rotate, so a burst moves the client once */
void etcd_endpoint_failed(ev_etcd_t *client, const call_base_t *base,
grpc_status_code status, grpc_slice status_details) {
/* A lock or campaign ended by the sweep carries the same message */
if (base->type != CALL_TYPE_LOCK && base->type != CALL_TYPE_ELECTION_CAMPAIGN)
check_leader_failure(client, base->channel_gen, status, status_details);
if (etcd_is_no_leader(status, status_details))
client->no_leader_epoch++;
if (!failed_on_current(client, base->channel_gen))
return;
if ((status == GRPC_STATUS_UNAVAILABLE && !cluster_wide_unavailable(status, status_details))
|| (status == GRPC_STATUS_DEADLINE_EXCEEDED && !connected(client))
|| member_timed_out(status, status_details))
etcd_rotate_endpoint(client);
}
/* etcd refuses new streams during any election, but ends an established one
* only after several election timeouts without a leader */
void etcd_stream_failed(ev_etcd_t *client, unsigned channel_gen, int established,
grpc_status_code status, grpc_slice status_details) {
if (established)
check_leader_failure(client, channel_gen, status, status_details);
if (etcd_is_no_leader(status, status_details))
client->no_leader_epoch++;
if (failed_on_current(client, channel_gen)
&& ((status == GRPC_STATUS_UNAVAILABLE && !cluster_wide_unavailable(status, status_details))
|| !connected(client)))
etcd_rotate_endpoint(client);
}
SV* create_error_hv(pTHX_ grpc_status_code code, const char *message, size_t message_len, const char *source) {
HV *err = newHV();
hv_store(err, "code", 4, newSViv(code), 0);
hv_store(err, "status", 6, newSVpv(grpc_status_name(code), 0), 0);
if (message && message_len > 0) {
hv_store(err, "message", 7, newSVpvn(message, message_len), 0);
} else {
hv_store(err, "message", 7, newSVpv("", 0), 0);
etcd_common.h view on Meta::CPAN
pending_call_t *pending_calls;
watch_call_t *watches;
keepalive_call_t *keepalives;
observe_call_t *observes;
int active;
int in_callback; /* Guard against freeing client during event processing */
char *auth_token;
size_t auth_token_len;
int timeout_seconds;
char **endpoints;
int endpoint_count;
int current_endpoint;
unsigned channel_gen; /* bumped on every endpoint switch */
grpc_channel_credentials *creds; /* NULL = insecure */
char *tls_server_name;
int keepalive_ms; /* 0 = no keepalive pings */
int keepalive_timeout_ms;
int max_retries;
/* Bumped by each answer without a leader: voids reconnects counted before it */
unsigned no_leader_epoch;
etcd_common.h view on Meta::CPAN
SV* create_error_hv(pTHX_ grpc_status_code code, const char *message, size_t message_len, const char *source);
SV* create_pending_error_hv(pTHX_ pending_call_t *pc, grpc_status_code code, const char *source);
SV* kv_to_hashref(pTHX_ Mvccpb__KeyValue *kv);
SV* event_to_hashref(pTHX_ Mvccpb__Event *event);
void add_header_to_hv(pTHX_ HV *result, Etcdserverpb__ResponseHeader *header);
channel_ref_t *etcd_create_channel(ev_etcd_t *client, const char *target);
void etcd_channel_release(channel_ref_t *ref, int destroy);
void etcd_call_release(call_base_t *base);
void etcd_rotate_endpoint(ev_etcd_t *client);
void etcd_endpoint_failed(ev_etcd_t *client, const call_base_t *base,
grpc_status_code status, grpc_slice status_details);
void etcd_stream_failed(ev_etcd_t *client, unsigned channel_gen, int established,
grpc_status_code status, grpc_slice status_details);
void setup_auth_metadata(ev_etcd_t *client, grpc_op *op, call_base_t *base);
void setup_stream_metadata(ev_etcd_t *client, grpc_op *op, call_base_t *base);
extern grpc_slice METHOD_KV_RANGE;
extern grpc_slice METHOD_KV_PUT;
extern grpc_slice METHOD_KV_DELETE;
lib/EV/Etcd.pm view on Meta::CPAN
EV::Etcd - Async etcd v3 client using native gRPC and EV/libev
=head1 SYNOPSIS
use v5.10;
use EV;
use EV::Etcd;
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
);
$client->put('/my/key', 'value', sub {
my ($resp, $err) = @_;
die $err->{message} if $err;
say "Put succeeded, revision: $resp->{header}{revision}";
});
$client->get('/my/key', sub {
my ($resp, $err) = @_;
lib/EV/Etcd.pm view on Meta::CPAN
C<mod_revision>, C<version> and C<lease>.
=head1 CONSTRUCTOR
=head2 new
my $client = EV::Etcd->new(%options);
=over 4
=item endpoints => [ 'host:port', ... ]
Default C<['127.0.0.1:2379']>; an empty list croaks. An endpoint may carry
the C<http://> or C<https://> scheme etcd prints; C<https://> turns on
C<tls>.
The client uses one endpoint at a time and moves to the next when it cannot
be reached: a unary call fails with UNAVAILABLE, or with DEADLINE_EXCEEDED
before a connection was made, a stream has to reconnect because its
connection is down, or a keepalive ping goes unanswered. The failing call
still reports its error; retrying it reaches the next endpoint. Several
failures on one endpoint move the client once.
A member that has lost its leader fails linearizable reads and writes
after etcd's request timeout (7 seconds by default), and these failures
move the client on too, though a shorter C<timeout> ends the call on the
client first and does not. Errors a working member returns during an election, such as
C<etcdserver: leader changed>, do not move it.
Streams need a leader as well: a member without one refuses new streams
and, after a few election timeouts, ends running watches and keepalives
with UNAVAILABLE C<etcdserver: no leader>. Streams retry such answers a
second apart until a leader is back, on the next endpoint when there are
several; each restarts the C<max_retries> count. Once a unary call, or a
stream that was running, reports C<etcdserver: no leader>, pending C<lock>
and C<election_campaign> calls sent over the same connection fail with
UNAVAILABLE, as they do when a connection that never became ready is
abandoned.
An endpoint may also be a gRPC target listing several addresses, such as
C<ipv4:10.0.0.1:2379,10.0.0.2:2379>. gRPC moves between them within one
connection, so a refused address fails no call, but never away from a
member that has lost its leader: list members as separate endpoints to
leave one.
gRPC connects through a proxy named in C<grpc_proxy>, C<https_proxy> or
C<http_proxy>, even to a loopback address; list etcd's hosts in
C<no_proxy> to connect directly.
=item timeout => $seconds
RPC timeout in whole seconds, at least 1; default 30. C<lock> and
C<election_campaign> wait without one, as a timeout could not tell whether
they succeeded.
=item max_retries => $count
Reconnection attempts for a stream (C<watch>, C<lease_keepalive>,
C<election_observe>) after a connection failure; default 30, 0 disables
reconnecting. Attempts are 0.5 seconds further apart each time, up to 5
seconds, so the default gives up about two minutes after the endpoints
start refusing connections, plus gRPC's 20-second connect timeout for one
that accepts them but stays silent. Lower it to learn sooner that the
endpoints are gone.
=item keepalive_time => $seconds
=item keepalive_timeout => $seconds
Seconds between pings while calls or streams are open (default 10,
fractions allowed, 0 disables them), and how long a ping may go unanswered
(default 10). An unanswered ping closes the connection: its calls fail with
UNAVAILABLE and the client moves to the next endpoint. Without pings, an
endpoint that goes silent (a hung host, a partition) goes unnoticed until
the operating system gives up on the connection. etcd closes connections
that ping more often than its C<--grpc-keepalive-min-time> (5 seconds by
default).
=item health_interval => $seconds
=item on_health_change => sub { my ($healthy, $endpoint) = @_; ... }
Every C<health_interval> seconds (fractions allowed; default 0, off) the
client checks its connection state without sending a request, and calls
C<on_health_change> when health changes. Only a failed connection attempt
counts as unhealthy; with several endpoints it also moves the client on.
Switching endpoints after a failed call is not reported.
=item auth_token => $token
A token from an earlier C<authenticate>, used without authenticating again.
=item tls => $bool
Connect with TLS; any C<tls_*> option or an C<https://> endpoint turns it
on too. Without C<tls_ca_file>, gRPC uses its default roots, or the PEM
bundle named by C<GRPC_DEFAULT_SSL_ROOTS_FILE_PATH>.
=item tls_ca_file => $path
PEM CA certificates to verify the servers with (etcd's
C<--trusted-ca-file>), in place of the default roots. Verification cannot
be turned off; for a self-signed server, give its certificate here.
=item tls_cert_file => $path
=item tls_key_file => $path
Client certificate and key, given together, for servers started with
C<--client-cert-auth>.
=item tls_server_name => $name
Name to verify the server certificate against and send as SNI, in place of
the endpoint's host.
=back
my $client = EV::Etcd->new(
endpoints => ['https://10.0.0.1:2379', 'https://10.0.0.2:2379'],
tls_ca_file => '/etc/etcd/ca.crt',
tls_cert_file => '/etc/etcd/client.crt',
tls_key_file => '/etc/etcd/client.key',
);
=head1 ERRORS
{
code => 14, # gRPC status code
status => 'UNAVAILABLE',
lib/EV/Etcd.pm view on Meta::CPAN
C<retryable> is set for UNAVAILABLE, ABORTED, DEADLINE_EXCEEDED and
RESOURCE_EXHAUSTED C<etcdserver: too many requests>, but never for a failed
C<lock> or C<election_campaign>. Other RESOURCE_EXHAUSTED errors last until
someone intervenes, such as C<etcdserver: mvcc: database space exceeded> (see
C<alarm>). Retryable means transient, not without effect: a write can be
applied after its deadline has passed, so retrying C<lease_grant>,
C<lease_revoke> or C<delete> can apply it twice or report what the first
attempt did.
Unary calls are not retried. Streams reconnect on their own (see
C<max_retries>), to the next endpoint once the current one has failed, and
the count restarts once a stream is working again. A stream reports an
error when reconnecting is disabled or exhausted, at once for a status a
new connection cannot fix (UNAUTHENTICATED, PERMISSION_DENIED,
INVALID_ARGUMENT, NOT_FOUND, ALREADY_EXISTS, FAILED_PRECONDITION,
OUT_OF_RANGE, UNIMPLEMENTED), and for an error the server sends on it, such
as a cancelled or compacted watch or an expired lease, which ends it.
The server can still clean up after a failed C<lock> or
C<election_campaign> has reported, deleting the key it made on the lease.
Retry with a fresh lease, and revoke the old one (which deletes all its
scripts/local_cluster.sh view on Meta::CPAN
start_node "$NODE1_NAME" "$NODE1_CLIENT" "$NODE1_PEER"
start_node "$NODE2_NAME" "$NODE2_CLIENT" "$NODE2_PEER"
start_node "$NODE3_NAME" "$NODE3_CLIENT" "$NODE3_PEER"
echo ""
echo "Waiting for cluster to form..."
sleep 3
# Check health
if etcdctl --endpoints="http://${NODE1_CLIENT}" endpoint health 2>/dev/null; then
echo ""
echo "Cluster is healthy!"
echo ""
echo "Endpoints:"
echo " Node 1: http://${NODE1_CLIENT}"
echo " Node 2: http://${NODE2_CLIENT}"
echo " Node 3: http://${NODE3_CLIENT}"
echo ""
echo "To use with EV::Etcd:"
echo " my \$client = EV::Etcd->new(endpoints => ['${NODE1_CLIENT}', '${NODE2_CLIENT}', '${NODE3_CLIENT}']);"
else
echo "Warning: Cluster may not be fully healthy yet"
fi
}
stop_cluster() {
echo "Stopping etcd cluster..."
for name in "$NODE1_NAME" "$NODE2_NAME" "$NODE3_NAME"; do
local pidfile="${CLUSTER_DIR}/${name}/etcd.pid"
scripts/local_cluster.sh view on Meta::CPAN
fi
else
echo " $name: NOT STARTED"
fi
done
echo ""
if [ $running -eq 3 ]; then
echo "Endpoint health:"
etcdctl --endpoints="http://${NODE1_CLIENT},http://${NODE2_CLIENT},http://${NODE3_CLIENT}" endpoint health 2>/dev/null || echo " (health check failed)"
echo ""
echo "Endpoint status:"
etcdctl --endpoints="http://${NODE1_CLIENT},http://${NODE2_CLIENT},http://${NODE3_CLIENT}" endpoint status --write-out=table 2>/dev/null || echo " (status check failed)"
fi
}
case "${1:-}" in
start)
stop_cluster 2>/dev/null || true
start_cluster
;;
stop)
stop_cluster
t/00-load.t view on Meta::CPAN
plan tests => 4;
my @load_warnings;
{
local $SIG{__WARN__} = sub { push @load_warnings, @_ };
use_ok('EV::Etcd');
}
is_deeply(\@load_warnings, [], 'loads without warnings');
my $client = EV::Etcd->new(endpoints => ['127.0.0.1:2379']);
isa_ok($client, 'EV::Etcd');
# The etcd tests skip when EV::Etcd cannot reach etcd, so with one known to be
# running a broken new() or status() must fail here instead
SKIP: {
skip 'set EV_ETCD_TEST_ETCD=1 when an etcd for testing runs on 127.0.0.1:2379', 1
unless $ENV{EV_ETCD_TEST_ETCD};
my ($ok, $err);
$client->status(sub { ($ok, $err) = (!$_[1], $_[1]); EV::break });
my $t = EV::timer(10, 0, sub { EV::break });
eval { require EV };
plan skip_all => 'EV required' if $@;
}
use EV;
use EV::Etcd;
my $etcd_available = 0;
eval {
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
timeout => 2,
);
$client->status(sub {
my ($resp, $err) = @_;
$etcd_available = 1 if !$err;
EV::break;
});
my $t = EV::timer(3, 0, sub { EV::break });
EV::run;
};
plan skip_all => 'etcd not available on 127.0.0.1:2379' unless $etcd_available;
# Never toggles auth; t/auth_enable_disable.t does that
plan tests => 30;
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
);
my $test_user = "test-user-$$-" . time();
my $test_role = "test-role-$$-" . time();
$client->auth_status(sub {
my ($resp, $err) = @_;
ok(!$err, 'auth_status succeeded');
ok(defined $resp->{enabled}, 'auth_status has enabled field');
diag("Auth enabled: " . ($resp->{enabled} ? "yes" : "no"));
t/auth_enable_disable.t view on Meta::CPAN
eval { require EV };
plan skip_all => 'EV required' if $@;
}
use EV;
use EV::Etcd;
my $etcd_available = 0;
eval {
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
timeout => 2,
);
$client->status(sub {
my ($resp, $err) = @_;
$etcd_available = 1 if !$err;
EV::break;
});
my $t = EV::timer(3, 0, sub { EV::break });
EV::run;
};
t/auth_enable_disable.t view on Meta::CPAN
($resp, $err) = @_;
$done = 1;
EV::break;
});
my $t = EV::timer(10, 0, sub { EV::break });
EV::run;
$live = 0;
return $done ? ($resp, $err) : (undef, { status => 'TIMEOUT', message => "$method: no response" });
}
my ($resp, $err) = call(EV::Etcd->new(endpoints => ['127.0.0.1:2379']), 'auth_status');
plan skip_all => 'Auth already enabled - cannot run enable/disable tests'
if $resp && $resp->{enabled};
plan tests => 18;
my $client = EV::Etcd->new(endpoints => ['127.0.0.1:2379']);
my $root_password = "root-test-pwd-$$-" . time();
my $auth_on;
# Disable auth again if the run dies while it may be on
END {
if ($auth_on) {
my $cleanup_client = EV::Etcd->new(endpoints => ['127.0.0.1:2379']);
my (undef, $err) = call($cleanup_client, 'authenticate', 'root', $root_password);
(undef, $err) = call($cleanup_client, 'auth_disable') unless $err;
diag("auth may still be enabled: $err->{message}")
if $err && $err->{status} ne 'FAILED_PRECONDITION';
}
}
($resp, $err) = call($client, 'user_add', 'root', $root_password);
ok(!$err, 'root user created') or diag explain $err;
($resp, $err) = call($client, 'user_grant_role', 'root', 'root');
ok(!$err, 'root role granted') or diag explain $err;
$auth_on = 1;
($resp, $err) = call($client, 'auth_enable');
ok(!$err, 'auth_enable succeeded') or diag explain $err;
($resp, $err) = call($client, 'authenticate', 'root', $root_password);
my $auth_token = $resp && $resp->{token};
ok($auth_token, 'authenticated and got token') or diag explain $err;
my $auth_client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
auth_token => $auth_token,
);
($resp, $err) = call($auth_client, 'auth_status');
ok($resp && $resp->{enabled}, 'auth_status confirms enabled') or diag explain $err;
# Stream leader metadata must coexist with the saved authentication token.
($resp) = call($auth_client, 'lease_grant', 30);
my $lease = $resp && $resp->{id};
my (%ready, %errors, @handles);
if ($lease) {
t/auth_errors.t view on Meta::CPAN
BEGIN { delete @ENV{qw(http_proxy https_proxy grpc_proxy)} }
use lib 'blib/lib', 'blib/arch';
use Test::More;
BEGIN { eval { require EV }; plan skip_all => 'EV required' if $@ }
use EV;
use EV::Etcd;
my $available = 0;
eval {
my $c = EV::Etcd->new(endpoints => ['127.0.0.1:2379'], timeout => 2);
$c->status(sub { $available = 1 if !$_[1]; EV::break });
my $t = EV::timer(3, 0, sub { EV::break });
EV::run;
};
plan skip_all => 'etcd not available on 127.0.0.1:2379' unless $available;
my $client = EV::Etcd->new(endpoints => ['127.0.0.1:2379']);
my $user = "test_autherr_$$";
my $err;
$client->user_add($user, "rightpw", sub { $err = $_[1]; EV::break });
my $t1 = EV::timer(3, 0, sub { EV::break });
EV::run;
plan skip_all => "user_add failed: $err->{message}" if $err && $err->{status} ne 'ALREADY_EXISTS';
my $dup_err;
$client->user_add($user, "anything", sub { $dup_err = $_[1]; EV::break });
t/auto_reconnect.t view on Meta::CPAN
BEGIN {
eval { require EV };
plan skip_all => 'EV required' if $@;
}
use EV;
use EV::Etcd;
my $etcd_available = 0;
eval {
my $c = EV::Etcd->new(endpoints => ['127.0.0.1:2379'], timeout => 2);
$c->status(sub { $etcd_available = 1 if !$_[1]; EV::break });
my $t = EV::timer(3, 0, sub { EV::break });
EV::run;
};
plan skip_all => 'etcd not available on 127.0.0.1:2379' unless $etcd_available;
plan tests => 5;
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
max_retries => 5,
);
ok($client, 'client created');
my $test_key = "/test_auto_reconnect_$$";
my $watch;
my $events_received = 0;
my $test_done = 0;
my $watch_created = 0;
t/binary_data.t view on Meta::CPAN
eval { require EV };
plan skip_all => 'EV required' if $@;
}
use EV;
use EV::Etcd;
my $etcd_available = 0;
eval {
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
timeout => 2,
);
$client->status(sub {
my ($resp, $err) = @_;
$etcd_available = 1 if !$err;
EV::break;
});
my $t = EV::timer(3, 0, sub { EV::break });
EV::run;
};
plan skip_all => 'etcd not available on 127.0.0.1:2379' unless $etcd_available;
plan tests => 18;
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
);
my $prefix = "/test-binary-$$-" . time();
{
my $utf8_key = "$prefix/utf8-\x{4e2d}\x{6587}";
my $utf8_value = "value-\x{65e5}\x{672c}\x{8a9e}";
my $put_ok = 0;
$client->put(encode_utf8($utf8_key), encode_utf8($utf8_value), sub {
t/callback_validation.t view on Meta::CPAN
plan skip_all => 'EV module not available';
exit;
}
}
plan tests => 27;
use_ok('EV::Etcd');
my $client = EV::Etcd->new(
endpoints => ['127.0.0.1:2379'],
);
ok($client, 'client created');
eval { $client->get('/test/key', 'not_a_callback'); };
like($@, qr/callback must be a code reference/, 'get rejects non-coderef callback');
eval { $client->put('/test/key', 'value', 'not_a_callback'); };
like($@, qr/callback must be a code reference/, 'put rejects non-coderef callback');
t/callback_validation.t view on Meta::CPAN
eval { $client->txn([{ key => '/k', value => 'v', target => [] }], [], [], sub {}); };
like($@, qr/invalid compare target/, 'txn rejects ref target');
eval { $client->txn([{ key => '/k', value => 'v', result => 5 }], [], [], sub {}); };
like($@, qr/invalid compare result/, 'txn rejects numeric result');
eval { $client->txn([{ key => '/k', value => 'v', target => undef, result => undef }], [], [], sub {}); };
ok(!$@, 'txn undef target/result behave as absent') or diag($@);
eval { EV::Etcd->new(endpoints => ['127.0.0.1:2379'], bogus_opt => 1); };
like($@, qr/Unknown option/, 'new rejects unknown option');
eval { $client->get('/k', { sort_order => 'sideways' }, sub {}); };
like($@, qr/Invalid sort_order/, 'get rejects invalid sort_order');
eval { $client->get('/k', { sort_target => 'BOGUS' }, sub {}); };
like($@, qr/Invalid sort_target/, 'get rejects invalid sort_target');
eval { $client->txn([{ key => '/k', value => 'v', result => '=', target => 'value' }], [], [], sub {}); };
ok(!$@, 'valid txn compare does not croak') or diag($@);
( run in 2.804 seconds using v1.01-cache-2.11-cpan-cb04b646d8c )