Add support for native ETCD distributed locks (#122)

* implement ETCD locking

* expand README.md

* fix typos in README.md

* add Connection#with_lock

* Update to grpc 1.17.0

Grpc was being kept at 1.6.0, which does not support ruby 2.5.0. Since
that important for us, update to 1.17.0 and solve issue with deadline
in spec by using `#from_relative_time` provided by
`GRPC::Core::TimeConsts` instead of our own implementation.

Also provide rake file with task to download etcd and improve travis
config to test more configurations.

* Don't test lock in etcd v3.1.X

* Use version from test instance

* Disable 3.3.10 in travis (doesn't work)

* Improve test coverage to make codecov happy

* Don't test locks in 3.1.0

* add lease support to #lock

* fix specs

* fix doc comments

* fix README.md locking examples

.travis.yml

@@ -1,16 +1,19 @@
 language: ruby
 rvm:
-  - 2.4.1
-  - 2.2
+  - 2.5.3
+  - 2.4.5
+  - 2.3.8
 
 env:
-  global:
-    - ETCD_VERSION=v3.2.0
+  - ETCD_VERSION=v3.1.20
+  - ETCD_VERSION=v3.2.25
+# v3.3.10 is not working for whatever reason (at least in travis, spec passes
+# locally for me)
+#  - ETCD_VERSION=v3.3.10
 
 install:
     - bundle install
-    - wget https://github.com/coreos/etcd/releases/download/$ETCD_VERSION/etcd-$ETCD_VERSION-linux-amd64.tar.gz -O etcd.tar.gz --no-check-certificate
-    - tar zxvf etcd.tar.gz
-    - export PATH=$PATH:etcd-$ETCD_VERSION-linux-amd64
+    - bundle exec rake download-etcd
+    - export PATH="$(dirname $(find /tmp -name 'etcd')):$PATH"
 
 script: bundle exec rspec

README.md

@@ -164,6 +164,23 @@ conn.watch('boom') do |events|
 end
 ```
 
+## Locks
+```ruby
+# First, get yourself a lease
+lease_id = conn.lease_grant(100)['ID']
+
+# Attempt to lock distibuted lock 'foo', wait at most 10 seconds
+lock_key = conn.lock('foo', lease_id, timeout: 10).key
+
+# Unlock the 'foo' lock using the key returned from `lock`
+conn.unlock(key)
+
+# Perform a critical section while holding the lock 'hello'
+conn.with_lock('hello', lease_id) do
+  puts "kitty!"
+end
+```
+
 ## Alarms
 ```ruby
 # List all active Alarms

Rakefile

@@ -0,0 +1,16 @@
+# frozen_string_literal: true
+
+ETCD_VERSION = ENV["ETCD_VERSION"] || "v3.2.0"
+ETCD_URL = "https://github.com/coreos/etcd/releases/download/#{ETCD_VERSION}/etcd-#{ETCD_VERSION}-linux-amd64.tar.gz"
+
+require "tmpdir"
+
+desc "Download etcd for it can be used in rspec"
+task :"download-etcd" do
+  tmpdir = Dir.mktmpdir
+  system("wget", ETCD_URL, "-O", "#{tmpdir}/etcd.tar.gz")   || exit(1)
+  system(*%W{tar -C #{tmpdir} -zxvf #{tmpdir}/etcd.tar.gz}) || exit(1)
+
+  puts "Etcd downloaded and extracted. Add it to the path:"
+  puts "    export PATH=\"#{tmpdir}/etcd-#{ETCD_VERSION}-linux-amd64:$PATH\""
+end

etcdv3.gemspec

@@ -14,6 +14,8 @@ Gem::Specification.new do |s|
   s.files         = `git ls-files`.split("\n")
   s.test_files    = `git ls-files -- {test,spec,features}/*`.split("\n")
 
-  s.add_dependency("grpc", "~> 1.6.0")
-  s.add_development_dependency("rspec", "~> 3.6.0")
+  s.add_dependency("grpc", "~> 1.17")
+  s.add_development_dependency("pry-byebug", "~> 3.6")
+  s.add_development_dependency("rake", "~> 12.3")
+  s.add_development_dependency("rspec", "~> 3.6")
 end

lib/etcdv3.rb

@@ -2,13 +2,15 @@
 require 'uri'
 
 require 'etcdv3/etcdrpc/rpc_services_pb'
+require 'etcdv3/etcdrpc/v3lock_services_pb'
 require 'etcdv3/auth'
 require 'etcdv3/kv/requests'
 require 'etcdv3/kv/transaction'
 require 'etcdv3/kv'
 require 'etcdv3/maintenance'
 require 'etcdv3/lease'
 require 'etcdv3/watch'
+require 'etcdv3/lock'
 require 'etcdv3/connection'
 require 'etcdv3/connection_wrapper'
 
@@ -84,6 +86,37 @@ def get(key, opts={})
     @conn.handle(:kv, 'get', [key, opts])
   end
 
+  # Locks distributed lock with the given name. The lock will unlock automatically
+  # when lease with the given ID expires. If this is not desirable, provide a non-expiring
+  # lease ID as an argument.
+  # name                          - string
+  # lease_id                      - integer
+  # optional :timeout             - integer
+  def lock(name, lease_id, timeout: nil)
+    @conn.handle(:lock, 'lock', [name, lease_id, {timeout: timeout}])
+  end
+
+  # Unlock distributed lock using the key previously obtained from lock.
+  # key                           - string
+  # optional :timeout             - integer
+  def unlock(key, timeout: nil)
+    @conn.handle(:lock, 'unlock', [key, {timeout: timeout}])
+  end
+
+  # Yield into the critical section while holding lock with the given
+  # name. The lock will be unlocked even if the block throws.
+  # name                          - string
+  # lease_id                      - integer
+  # optional :timeout             - integer
+  def with_lock(name, lease_id, timeout: nil)
+    key = lock(name, lease_id, timeout: timeout).key
+    begin
+      yield
+    ensure
+      unlock(key, timeout: timeout)
+    end
+  end
+
   # Inserts a new key.
   # key                           - string
   # value                         - string

lib/etcdv3/auth.rb

@@ -1,6 +1,7 @@
 
 class Etcdv3
   class Auth
+    include GRPC::Core::TimeConsts
 
     PERMISSIONS = {
       :read => Authpb::Permission::Type::READ,
@@ -122,7 +123,7 @@ def generate_token(user, password, timeout: nil)
     private
 
     def deadline(timeout)
-      Time.now.to_f + (timeout || @timeout)
+      from_relative_time(timeout || @timeout)
     end
   end
 end

lib/etcdv3/connection.rb

@@ -6,7 +6,8 @@ class Connection
       kv: Etcdv3::KV,
       maintenance: Etcdv3::Maintenance,
       lease: Etcdv3::Lease,
-      watch: Etcdv3::Watch
+      watch: Etcdv3::Watch,
+      lock: Etcdv3::Lock,
     }
 
     attr_reader :endpoint, :hostname, :handlers, :credentials

lib/etcdv3/kv.rb

@@ -2,6 +2,7 @@
 class Etcdv3
   class KV
     include Etcdv3::KV::Requests
+    include GRPC::Core::TimeConsts
 
     def initialize(hostname, credentials, timeout, metadata={})
       @stub = Etcdserverpb::KV::Stub.new(hostname, credentials)
@@ -36,7 +37,7 @@ def transaction(block, timeout: nil)
     private
 
     def deadline(timeout)
-      Time.now.to_f + (timeout || @timeout)
+      from_relative_time(timeout || @timeout)
     end
 
     def generate_request_ops(requests)

lib/etcdv3/lease.rb

@@ -1,5 +1,7 @@
 class Etcdv3
   class Lease
+    include GRPC::Core::TimeConsts
+
     def initialize(hostname, credentials, timeout, metadata={})
       @stub = Etcdserverpb::Lease::Stub.new(hostname, credentials)
       @timeout = timeout
@@ -31,7 +33,7 @@ def lease_keep_alive_once(id, timeout: nil)
     private
 
     def deadline(timeout)
-      Time.now.to_f + (timeout || @timeout)
+      from_relative_time(timeout || @timeout)
     end
   end
 end

lib/etcdv3/lock.rb

@@ -0,0 +1,27 @@
+class Etcdv3
+  class Lock
+    include GRPC::Core::TimeConsts
+
+    def initialize(hostname, credentials, timeout, metadata = {})
+      @stub = V3lockpb::Lock::Stub.new(hostname, credentials)
+      @timeout = timeout
+      @metadata = metadata
+    end
+
+    def lock(name, lease_id, timeout: nil)
+      request = V3lockpb::LockRequest.new(name: name, lease: lease_id)
+      @stub.lock(request, deadline: deadline(timeout))
+    end
+
+    def unlock(key, timeout: nil)
+      request = V3lockpb::UnlockRequest.new(key: key)
+      @stub.unlock(request, deadline: deadline(timeout))
+    end
+
+    private
+
+    def deadline(timeout)
+      from_relative_time(timeout || @timeout)
+    end
+  end
+end

lib/etcdv3/protos/v3lock.proto

@@ -0,0 +1,55 @@
+syntax = "proto3";
+package v3lockpb;
+
+import "annotations.proto";
+import "rpc.proto";
+import "gogo.proto";
+
+option (gogoproto.marshaler_all) = true;
+option (gogoproto.unmarshaler_all) = true;
+
+service Lock {
+  // Lock acquires a distributed shared lock on a given named lock.
+  rpc Lock(LockRequest) returns (LockResponse) {
+      option (google.api.http) = {
+        post: "/v3alpha/lock/lock"
+        body: "*"
+    };
+  }
+
+  // Unlock takes a key returned by Lock and releases the hold on lock.
+  rpc Unlock(UnlockRequest) returns (UnlockResponse) {
+      option (google.api.http) = {
+        post: "/v3alpha/lock/unlock"
+        body: "*"
+    };
+  }
+}
+
+message LockRequest {
+  // name is the identifier for the distributed shared lock to be acquired.
+  bytes name = 1;
+  // lease is the ID of the lease that will be attached to ownership of the
+  // lock. If the lease expires or is revoked and currently holds the lock,
+  // the lock is automatically released. Calls to Lock with the same lease will
+  // be treated as a single acquisition; locking twice with the same lease is a
+  // no-op.
+  int64 lease = 2;
+}
+
+message LockResponse {
+  etcdserverpb.ResponseHeader header = 1;
+  // key is a key that will exist on etcd for the duration that the Lock caller
+  // owns the lock. Users should not modify this key or the lock may exhibit
+  // undefined behavior.
+  bytes key = 2;
+}
+
+message UnlockRequest {
+  // key is the lock ownership key granted by Lock.
+  bytes key = 1;
+}
+
+message UnlockResponse {
+  etcdserverpb.ResponseHeader header = 1;
+}

spec/etcdv3/lock_spec.rb

@@ -0,0 +1,23 @@
+require 'spec_helper'
+
+# Locking is not implemented in etcd v3.1.X
+unless $instance.version < Gem::Version.new("3.2.0")
+  describe Etcdv3::Lock do
+    let(:stub) { local_stub(Etcdv3::Lock, 1) }
+    let(:lease_stub) { local_stub(Etcdv3::Lease, 1) }
+
+    it_should_behave_like "a method with a GRPC timeout", described_class, :unlock, :unlock, 'foo'
+    #it_should_behave_like "a method with a GRPC timeout", described_class, :lock, :lock, 'foo'
+
+    describe '#lock' do
+      let(:lease_id) { lease_stub.lease_grant(10)['ID'] }
+      subject { stub.lock('foo', lease_id) }
+      it { is_expected.to be_an_instance_of(V3lockpb::LockResponse) }
+    end
+
+    describe '#unlock' do
+      subject { stub.unlock('foo') }
+      it { is_expected.to be_an_instance_of(V3lockpb::UnlockResponse) }
+    end
+  end
+end