grpc-etcd-client 0.1.1.1 → 0.1.1.2

raw patch · 3 files changed

+293/−13 lines, 3 filesdep ~http2-client-grpcPVP: minor bump suggested

API additions: PVP suggests at least a minor version bump

Dependency ranges changed: http2-client-grpc

API changes (from Hackage documentation)

Files

• ChangeLog.md +5/−1
• grpc-etcd-client.cabal +3/−3
• src/Network/EtcdV3.hs +285/−9

@@ -1,3 +1,7 @@ # Changelog for grpc-etcd-client -## Unreleased changes+## 0.1.1.2++Transactions.+Watchers.+Leader election.

@@ -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 <https://github.com/lucasdicioccio/etcd-grpc#readme> 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

@@ -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