partially supports etcdv3 client(kv, lease, watch)

lualib/core/etcd.lua

@@ -0,0 +1,210 @@
+local core = require "core"
+local json = require "core.json"
+local pb = require "pb"
+local grpc = require "core.grpc"
+local proto = require "core.etcd.v3.proto"
+local M = {}
+
+local pairs = pairs
+local assert = assert
+local sub = string.sub
+local char = string.char
+local remove = table.remove
+local setmetatable = setmetatable
+
+local no_prefix_end = "\0"
+
+local sort_order_num<const> = {
+	NONE = pb.enum(".etcdserverpb.RangeRequest.SortOrder", "NONE"),
+	ASCEND = pb.enum(".etcdserverpb.RangeRequest.SortOrder", "ASCEND"),
+	DESCEND = pb.enum(".etcdserverpb.RangeRequest.SortOrder", "DESCEND"),
+}
+
+local sort_target_num<const> = {
+	KEY = pb.enum(".etcdserverpb.RangeRequest.SortTarget", "KEY"),
+	VERSION = pb.enum(".etcdserverpb.RangeRequest.SortTarget", "VERSION"),
+	CREATE = pb.enum(".etcdserverpb.RangeRequest.SortTarget", "CREATE"),
+	MOD = pb.enum(".etcdserverpb.RangeRequest.SortTarget", "MOD"),
+	VALUE = pb.enum(".etcdserverpb.RangeRequest.SortTarget", "VALUE"),
+}
+
+local function opt_prefix(key, _, options)
+	options.prefix = nil
+	if not key or #key == 0 then
+		options.key = no_prefix_end
+		options.range_end = no_prefix_end
+		return
+	end
+	for i = #key, 1, -1 do
+		if key[i] < 0xff then
+			options.range_end = sub(key, 1, i - 1) .. char(key[i] + 1)
+			return
+		end
+	end
+	-- next prefix does not exist (e.g., 0xffff);
+	-- default to WithFromKey policy
+	options.range_end = no_prefix_end
+end
+
+local function opt_fromkey(key, opt, options)
+	if #key == 0 then
+		options.key = no_prefix_end
+		return
+	end
+	options.range_end = no_prefix_end
+end
+
+local function opt_options(_, sort_target, options)
+	if not sort_target then
+		return
+	end
+	local sort_order
+	if sort_target == "KEY" then
+		sort_order = options.sort_order
+		if sort_order == "ASCEND" then
+			sort_order = "NONE"
+		end
+	end
+	options.sort_order = sort_order_num[sort_order]
+	options.sort_target = sort_target_num[sort_target]
+end
+
+local option_fn = {
+	prefix = opt_prefix,
+	fromkey = opt_fromkey,
+	sort_target = opt_options,
+}
+
+local function apply_options(key, req, options)
+	if not options then
+		return
+	end
+	for k, fn in pairs(option_fn) do
+		local v = options[k]
+		if v then
+			fn(key, v, options)
+		end
+	end
+	for k, v in pairs(options) do
+		req[k] = v
+	end
+end
+
+local mt = {__index = M}
+function M.newclient(conf)
+	return setmetatable({
+		kv = grpc.newclient {
+			service = "KV",
+			endpoints = conf.endpoints,
+			proto = proto.loaded['rpc.proto'],
+			timeout = conf.timeout,
+		},
+		lease = grpc.newclient {
+			service = "Lease",
+			endpoints = conf.endpoints,
+			proto = proto.loaded['rpc.proto'],
+			timeout = conf.timeout,
+		},
+		watcher = grpc.newclient {
+			service = "Watch",
+			endpoints = conf.endpoints,
+			proto = proto.loaded['rpc.proto'],
+			timeout = conf.timeout,
+		},
+	}, mt)
+end
+
+------KV
+function M.put(self, key, value, options)
+	local req = {
+		key = key,
+		value = value,
+	}
+	apply_options(key, req, options)
+	return self.kv.Put(req)
+end
+
+function M.get(self, key, options)
+	local req = {
+		key = key,
+	}
+	apply_options(key, req, options)
+	return self.kv.Range(req)
+end
+
+function M.delete(self, key, options)
+	local req = {
+		key = key,
+	}
+	apply_options(key, req, options)
+	return self.kv.DeleteRange(req)
+end
+
+function M.compact(self, rev, options)
+	local req ={
+		revision = rev,
+	}
+	apply_options(rev, req, options)
+	return self.kv.Compact(req)
+end
+
+--Lease
+function M.grant(self, ttl)
+	local req = {
+		TTL = ttl,
+	}
+	return self.kv.LeaseGrant(req)
+end
+
+function M.revoke(self, leaseid)
+	local req = {
+		lease_id = leaseid,
+	}
+	return self.kv.LeaseRevoke(req)
+end
+
+function M.ttl(self, leaseid, options)
+	local req = {
+		lease_id = leaseid,
+	}
+	if options then
+		for k, v in pairs(options) do
+			req[k] = v
+		end
+	end
+	return self.kv.LeaseTimeToLive(req)
+end
+
+function M.leases(self)
+	return self.kv.LeaseLeases()
+end
+
+function M.keepalive(self, leaseid)
+	local req = {
+		ID = leaseid,
+	}
+	return self.kv.LeaseKeepAlive(req)
+end
+
+--watcher
+function M.watch(self, key, options)
+	local req = {
+		key = key,
+	}
+	apply_options(key, req, options)
+	local send, recv, err = self.watcher.Watch(req)
+	if not send then
+		return nil, err
+	end
+	local ok, err = send({create_request = req})
+	if not ok then
+		return nil, err
+	end
+	local ack, err = recv()
+	if not ack then
+		return nil, err
+	end
+	return recv, nil
+end
+
+return M