view release on metacpan or  search on metacpan

lib/EV/Etcd.pm  view on Meta::CPAN

*txn = sub {
    my $self = shift;

    if (@_ == 4
        && $reftype->($_[0]) eq 'ARRAY'
        && $reftype->($_[1]) eq 'ARRAY'
        && $reftype->($_[2]) eq 'ARRAY'
        && $reftype->($_[3]) eq 'CODE') {
        return $_xs_txn->($self, @_);
    }
    if (@_ == 4
        && $reftype->($_[0]) eq 'ARRAY'
        && $reftype->($_[1]) eq 'ARRAY'
        && $reftype->($_[2]) eq 'ARRAY') {
        Carp::croak("txn: callback must be a code reference");
    }

    my $callback;
    if (@_ % 2 == 1) {
        $reftype->($_[-1]) eq 'CODE'
            or Carp::croak("txn: expected positional (compare, success, failure, callback) or name => value pairs");
        $callback = pop;
    }

    my %args = @_;
    for my $k (keys %args) {
        Carp::croak("Unknown option '$k' in \$client->txn")
            unless $k eq 'compare' || $k eq 'success' || $k eq 'failure' || $k eq 'callback';
    }
    $callback //= $args{callback};

    for my $k (qw(compare success failure)) {
        next unless defined $args{$k};
        $reftype->($args{$k}) eq 'ARRAY'
            or Carp::croak("txn '$k' must be an array reference");
    }

    my $compare = $args{compare} // [];
    my $success = $args{success} // [];
    my $failure = $args{failure} // [];

    return $_xs_txn->($self, $compare, $success, $failure, $callback);
};
use warnings 'redefine';

1;

__END__

=head1 NAME

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) = @_;
        die $err->{message} if $err;
        say "Value: $resp->{kvs}[0]{value}";
    });

    $client->watch('/my/key', sub {
        my ($resp, $err) = @_;
        return warn "Watch error: $err->{message}\n" if $err;
        for my $event (@{$resp->{events}}) {
            say "Event: $event->{type} on $event->{kv}{key}";
        }
    });

    EV::run;

=head1 DESCRIPTION

An asynchronous etcd v3 client on the gRPC Core C API and the EV event
loop: a gRPC thread waits for completions and wakes the loop, and callbacks
run in the Perl thread. It needs etcd 3.4 or later (C<auth_status> needs
3.5).

Every method takes a callback last; methods with options take them as a
hash reference just before it. Invalid arguments croak. The callback
receives C<($response, $error)>: the response hash and C<undef>, or
C<undef> and an error hash (see L</ERRORS>). Responses carry a C<header>
with C<cluster_id>, C<member_id>, C<revision> and C<raft_term> unless noted
otherwise. A kv hash has C<key>, C<value>, C<create_revision>,
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',
        message   => 'Connection refused',
        source    => 'range',               # the call that failed
        retryable => 1,
    }

C<source> is the method name, except C<range> for C<get>, C<lease_ttl> for
C<lease_time_to_live>, C<keepalive> for C<lease_keepalive>, C<campaign>,
C<proclaim>, C<leader>, C<resign> and C<observe> for the C<election_*>
calls, and C<internal> for a unary response the client could not read.

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
keys) or let it expire, as a stale candidate on it can block the new
attempt.

=head1 CALLBACKS AND HANDLES

A callback that dies stops neither the loop nor the client: the exception
goes to C<$EV::DIED>, a warning by default.

Callbacks live in C structures Perl cannot see, so a closure that captures
its client or stream handle keeps it alive until the stream is cancelled
or the client destroyed; capture a weakened copy instead. Dropping a handle
does not cancel its stream. Destroying a client cancels its calls and
streams without calling their callbacks. A client keeps C<EV::run> running
while it exists: destroy it or call C<EV::break> to leave the loop. Only
EV's default loop runs the callbacks, never a loop from
C<< EV::Loop->new >>.

    use Scalar::Util 'weaken';
    weaken(my $weak = $client);
    my $watch = $client->watch('/jobs', sub {
        my ($resp, $err) = @_;
        $weak->put('/seen', 1, sub {}) if $weak && !$err;
    });

=head2 cancel

    $handle->cancel($callback);

C<watch>, C<lease_keepalive> and C<election_observe> return a handle
(C<EV::Etcd::Watch>, C<EV::Etcd::Keepalive>, C<EV::Etcd::Observe>) whose
C<cancel> ends the stream. The callback runs before C<cancel> returns, with
an empty hash as the response, and the stream delivers nothing after it.
Cancelling again is safe and does the same.

=head1 ENCODING

etcd stores keys and values as bytes and the client does no encoding: a
string with the UTF-8 flag is stored as its UTF-8 bytes, and responses hold
byte strings. Use C<encode_utf8> and C<decode_utf8> from L<Encode> at the
boundary for character data.

A key or value over 1 MiB croaks before anything is sent. etcd refuses a
request over its C<--max-request-bytes> (1.5 MiB by default) with
INVALID_ARGUMENT C<etcdserver: request is too large>, and gRPC one 512 KiB
larger still with RESOURCE_EXHAUSTED.

=head1 KEY-VALUE

=head2 put