diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,3 +1,7 @@ # Changelog for grpc-etcd-client -## Unreleased changes +## 0.1.1.2 + +Transactions. +Watchers. +Leader election. diff --git a/grpc-etcd-client.cabal b/grpc-etcd-client.cabal --- a/grpc-etcd-client.cabal +++ b/grpc-etcd-client.cabal @@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack -- --- hash: fe4df847fc02c4a21ca937d7018145dd7276df51b662536b3c17dcbc79eb6ee4 +-- hash: bf4e5b4e26d379b8f85cd5a804c8ec115551d82a89f6e6b5463144e67d7c3c32 name: grpc-etcd-client -version: 0.1.1.1 +version: 0.1.1.2 synopsis: gRPC client for etcd description: Please see the README on GitHub at category: Distributed Computing @@ -40,7 +40,7 @@ , data-default-class >=0.1 && <0.2 , grpc-api-etcd ==0.1.0.1 , http2-client >=0.8 && <0.9 - , http2-client-grpc >=0.5.0.2 && <0.6 + , http2-client-grpc >=0.5.0.3 && <0.6 , lens >=4.16 && <4.17 , network >=2.6 && <2.7 , proto-lens >=0.3 && <0.4 diff --git a/src/Network/EtcdV3.hs b/src/Network/EtcdV3.hs --- a/src/Network/EtcdV3.hs +++ b/src/Network/EtcdV3.hs @@ -1,6 +1,27 @@ +{-# LANGUAGE TypeFamilies #-} {-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} -- | Simple lib to access Etcd over gRPC. +-- +-- This library offers a low-on-details EtcdV3 APIs to easily add distributed +-- system mechanisms into your Haskell applications. +-- +-- This library uses the 'Network.GRPC.Client.Helpers' module from +-- 'http2-client-grpc', which may trade some functionalities for the sake of +-- simplicity. A typically hidden functionality is the possibility to abort a +-- query upon a client decision (e.g., while waiting for some Lock). +-- +-- In general, this library provides some simplification to express the input +-- messages of RPCs. For instance, the MVCC capabilities are mostly hidden from +-- the _querying_ side of this library (i.e., queries will requet the latest +-- revision of a key/value pair). It is pretty simple, however, to reuse the +-- smart constructor (such as 'putReq' and continue editing values with lens) +-- and then call the gRPC RPC in a one liner. +-- +-- A reason for this design is that the intended user of this library will run +-- a continuous loop to acquire some leadership or monitor some values for the +-- whole duration of a program. module Network.EtcdV3 ( -- * Generalities. @@ -9,6 +30,7 @@ -- * Reading. , KeyRange(..) , range + , rangeReq , rangeResponsePairs -- * Granting leases. , grantLease @@ -17,30 +39,61 @@ , keepAlive -- * Writing. , put + , putReq , delete + , delReq -- * Locking. , AcquiredLock , fromLockResponse , lock , unlock + -- * Transactions. + , transaction + -- * A Few Smart Constructors To Make Life Easy When Building Transactions. + , successTxnReq + , kvEq + , kvNeq + , kvGt + , kvLt + -- * Watches. + , watchForever + , watchReq + -- * Leader election. + , Election(..) + , LeaderEvidence + , runForLeadership + , fromCampaignResponse + , proclaim + , resign + , getProclaimedValue + , observeProclaimedValues + , proclaimReq + , campaignReq + , resignReq + , leaderReq -- * re-exports , def , module Control.Lens ) where import Control.Lens +import Control.Exception (throwIO) import Data.Default.Class (def) import Data.ByteString (ByteString) import qualified Data.ByteString.Char8 as C8 import Data.Semigroup (Endo) +import Data.String (IsString) import GHC.Int (Int64) import Network.GRPC.Client import Network.GRPC.Client.Helpers +import Network.HTTP2.Client (TooMuchConcurrency) import Network.Socket (HostName, PortNumber) import qualified Proto.Etcd.Etcdserver.Etcdserverpb.Rpc as EtcdRPC import qualified Proto.Etcd.Etcdserver.Etcdserverpb.Rpc_Fields as EtcdPB import qualified Proto.Etcd.Etcdserver.Api.V3lock.V3lockpb.V3lock as LockRPC import qualified Proto.Etcd.Etcdserver.Api.V3lock.V3lockpb.V3lock_Fields as LockPB +import qualified Proto.Etcd.Etcdserver.Api.V3election.V3electionpb.V3election as ElectionRPC +import qualified Proto.Etcd.Etcdserver.Api.V3election.V3electionpb.V3election_Fields as ElectionPB -- | EtcdClient configuration. etcdClientConfigSimple :: HostName -> PortNumber -> UseTlsOrNot -> GrpcClientConfig @@ -70,7 +123,12 @@ -- ^ Looked-up range. -> EtcdQuery EtcdRPC.RangeResponse range grpc r = preview unaryOutput <$> - rawUnary (RPC :: RPC EtcdRPC.KV "range") grpc (def & EtcdPB.key .~ k0 & EtcdPB.rangeEnd .~ kend) + rawUnary (RPC :: RPC EtcdRPC.KV "range") grpc (rangeReq r) + +rangeReq :: KeyRange -> EtcdRPC.RangeRequest +rangeReq r = def + & EtcdPB.key .~ k0 + & EtcdPB.rangeEnd .~ kend where (k0, kend) = rangePairForRangeQuery r @@ -141,16 +199,19 @@ -> ByteString -- ^ Key. -> ByteString - -- ^ Value. -> Maybe GrantedLease -- ^ Lease on the key. -> EtcdQuery EtcdRPC.PutResponse put grpc k v gl = - preview unaryOutput <$> rawUnary (RPC :: RPC EtcdRPC.KV "put") grpc (def & EtcdPB.key .~ k & EtcdPB.value .~ v & EtcdPB.lease .~ l) - where - l = maybe 0 _getGrantedLeaseId gl + preview unaryOutput <$> rawUnary (RPC :: RPC EtcdRPC.KV "put") grpc (putReq k v gl) +putReq :: ByteString -> ByteString -> Maybe GrantedLease -> EtcdRPC.PutRequest +putReq k v gl = def + & EtcdPB.key .~ k + & EtcdPB.value .~ v + & EtcdPB.lease .~ (maybe 0 _getGrantedLeaseId gl) + -- | Delete a range of values. delete :: GrpcClient @@ -159,7 +220,12 @@ -- ^ Deleted range. -> EtcdQuery EtcdRPC.DeleteRangeResponse delete grpc r = preview unaryOutput <$> - rawUnary (RPC :: RPC EtcdRPC.KV "deleteRange") grpc (def & EtcdPB.key .~ k0 & EtcdPB.rangeEnd .~ kend) + rawUnary (RPC :: RPC EtcdRPC.KV "deleteRange") grpc (delReq r) + +delReq :: KeyRange -> EtcdRPC.DeleteRangeRequest +delReq r = def + & EtcdPB.key .~ k0 + & EtcdPB.rangeEnd .~ kend where (k0, kend) = rangePairForRangeQuery r @@ -198,9 +264,219 @@ ----------------------------------------------------------------------------------------- -watch +kvEq :: ByteString -> ByteString -> EtcdRPC.Compare +kvEq k v = def + & EtcdPB.target .~ EtcdRPC.Compare'VALUE + & EtcdPB.result .~ EtcdRPC.Compare'EQUAL + & EtcdPB.key .~ k + & EtcdPB.value .~ v + +kvNeq :: ByteString -> ByteString -> EtcdRPC.Compare +kvNeq k v = def + & EtcdPB.target .~ EtcdRPC.Compare'VALUE + & EtcdPB.result .~ EtcdRPC.Compare'NOT_EQUAL + & EtcdPB.key .~ k + & EtcdPB.value .~ v + +kvGt :: ByteString -> ByteString -> EtcdRPC.Compare +kvGt k v = def + & EtcdPB.target .~ EtcdRPC.Compare'VALUE + & EtcdPB.result .~ EtcdRPC.Compare'GREATER + & EtcdPB.key .~ k + & EtcdPB.value .~ v + +kvLt :: ByteString -> ByteString -> EtcdRPC.Compare +kvLt k v = def + & EtcdPB.target .~ EtcdRPC.Compare'VALUE + & EtcdPB.result .~ EtcdRPC.Compare'LESS + & EtcdPB.key .~ k + & EtcdPB.value .~ v + +-- | Smart constructor, to use in conjunction with other smart constructor +-- which give a small DSL for the common cases. The proto-lens generated code +-- is fine but requires qualified imports to avoid namespace clashes. +-- +-- Example use: +-- @ +-- putStrLn $ Data.ProtoLens.showMessage$ successTxnReq [kvEq "hello" "world"] [putReq "txn-written" "1234" Nothing] [] [] +-- compare { target: VALUE key: "hello" value: "world" } +-- success { request_put { key: "txn-written" value: "1234" } } +-- @ +successTxnReq + :: [EtcdRPC.Compare] + -> [EtcdRPC.PutRequest] + -> [EtcdRPC.DeleteRangeRequest] + -> [EtcdRPC.RangeRequest] + -> EtcdRPC.TxnRequest +successTxnReq cmps rps drrs rrs = def + & EtcdPB.compare .~ cmps + & EtcdPB.success .~ (mconcat [ + [ def & EtcdPB.maybe'requestPut ?~ rp | rp <- rps ] + , [ def & EtcdPB.maybe'requestDeleteRange ?~ drr | drr <- drrs ] + , [ def & EtcdPB.maybe'requestRange ?~ rr | rr <- rrs ] + ]) + +-- | Issue a transaction request, performing many modifications at once. +-- +-- See 'successTxnReq' for a smart constructor of transaction request. +transaction :: GrpcClient + -- ^ Initialized gRPC client. + -> EtcdRPC.TxnRequest + -- ^ Transaction + -> EtcdQuery EtcdRPC.TxnResponse +transaction grpc tx = preview unaryOutput <$> + rawUnary (RPC :: RPC EtcdRPC.KV "txn") grpc tx + +----------------------------------------------------------------------------------------- + +watchReq + :: KeyRange + -> [EtcdRPC.WatchCreateRequest'FilterType] + -> Int64 + -> EtcdRPC.WatchRequest +watchReq r fs wId = def + & EtcdPB.maybe'createRequest ?~ (def + & EtcdPB.key .~ k0 + & EtcdPB.rangeEnd .~ kend + & EtcdPB.filters .~ fs + & EtcdPB.fragment .~ True + & EtcdPB.watchId .~ wId) + where + (k0, kend) = rangePairForRangeQuery r + +-- | Starts some watches until an exception occurs. +-- +-- Alas there is no simple way to discard a watch or stop a stream with this +-- method as http2-client-grpc does not yet expose an 'OutgoingEvent' to +-- forcefully close the client stream. +-- +-- See 'watchReq' for building watch requests. +watchForever + :: GrpcClient + -- ^ Initialized gRPC client. + -> [EtcdRPC.WatchRequest] + -- ^ List of watches to register. -> (a -> EtcdRPC.WatchResponse -> IO a) + -- ^ State-passing handler for handling watches. -> a - -> EtcdQuery (a) -watch grpc foo = undefined + -- ^ Initial state. + -> IO (Either TooMuchConcurrency ()) +watchForever grpc wcs handle v0 = do + fmap (const ()) <$> rawGeneralStream (RPC :: RPC EtcdRPC.Watch "watch") grpc v0 handleWatch wcs registerAndWait + where + handleWatch v (RecvMessage msg) = handle v msg + handleWatch _ (Invalid err) = throwIO err + handleWatch v _ = pure v + registerAndWait (x:xs) = pure (xs, SendMessage Compressed x) + registerAndWait [] = pure ([], Finalize) + +----------------------------------------------------------------------------------------- + +-- | A newtype around ByteString to speak about election names. +newtype Election = Election { _getElectionName :: ByteString } + deriving (Show, Eq, Ord, IsString) + +-- | An Opaque type around returned leader key. +newtype LeaderEvidence = LeaderEvidence { _getLeaderKey :: ElectionRPC.LeaderKey } + +-- | Waits until elected. +runForLeadership + :: GrpcClient + -- ^ Initialized gRPC client. + -> Election + -- ^ An election to run for. + -> GrantedLease + -- ^ A lease delimiting the leadership duration. + -> ByteString + -- ^ The initially-proclaimed value. + -> EtcdQuery ElectionRPC.CampaignResponse +runForLeadership grpc el gl v = preview unaryOutput <$> + rawUnary (RPC :: RPC ElectionRPC.Election "campaign") grpc (campaignReq el gl v) + +campaignReq + :: Election + -> GrantedLease + -> ByteString + -> ElectionRPC.CampaignRequest +campaignReq el gl v = def + & ElectionPB.name .~ _getElectionName el + & ElectionPB.lease .~ _getGrantedLeaseId gl + & ElectionPB.value .~ v + +-- | Helper to retrieve an evidence that we are a leader from a CampaignResponse. +fromCampaignResponse :: ElectionRPC.CampaignResponse -> Maybe LeaderEvidence +fromCampaignResponse cr = cr ^? ElectionPB.leader . to LeaderEvidence + +-- | As a leader, proclaim a new value. +-- +-- Observers will be notified. +proclaim + :: GrpcClient + -- ^ Initialized gRPC client. + -> LeaderEvidence + -- ^ Proof of recent successful participation to a leader-election. + -> ByteString + -- ^ New value to proclaim. + -> EtcdQuery ElectionRPC.ProclaimResponse +proclaim grpc le v = preview unaryOutput <$> + rawUnary (RPC :: RPC ElectionRPC.Election "proclaim") grpc (proclaimReq le v) + +proclaimReq + :: LeaderEvidence + -> ByteString + -> ElectionRPC.ProclaimRequest +proclaimReq le v = def + & ElectionPB.leader .~ _getLeaderKey le + & ElectionPB.value .~ v + +-- | Resign from leadership. +-- +-- If other campaigners are waiting for the role, they will get elected. +resign + :: GrpcClient + -- ^ Initialized gRPC client. + -> LeaderEvidence + -- ^ Proof of recent successful participation to a leader-election. + -> EtcdQuery ElectionRPC.ResignResponse +resign grpc le = preview unaryOutput <$> + rawUnary (RPC :: RPC ElectionRPC.Election "resign") grpc (resignReq le) + +resignReq + :: LeaderEvidence + -> ElectionRPC.ResignRequest +resignReq le = def + & ElectionPB.leader .~ _getLeaderKey le + +-- | As a bystander of an election, get the proclaimed value. +getProclaimedValue + :: GrpcClient + -- ^ Initialized gRPC client. + -> Election + -- ^ Election to read value from. + -> EtcdQuery ElectionRPC.LeaderResponse +getProclaimedValue grpc el = preview unaryOutput <$> + rawUnary (RPC :: RPC ElectionRPC.Election "leader") grpc (leaderReq el) + +leaderReq + :: Election + -> ElectionRPC.LeaderRequest +leaderReq el = def + & ElectionPB.name .~ _getElectionName el + +-- | As a bystander of an election, get notified for every proclaimed value. +observeProclaimedValues + :: GrpcClient + -- ^ Initialized gRPC client. + -> Election + -- ^ Election to wait for values from. + -> (a -> ElectionRPC.LeaderResponse -> IO a) + -- ^ State-passing handler for handling iteratively updated proclaimed values. + -> a + -- ^ An initial state. + -> IO (Either TooMuchConcurrency a) +observeProclaimedValues grpc el handler v0 = fmap (view _1) <$> + rawStreamServer (RPC :: RPC ElectionRPC.Election "observe") grpc v0 (observeReq el) handle2 + where + observeReq = leaderReq + handle2 v1 _ msg = handler v1 msg