om-legion (empty) → 6.9.0.3
raw patch · 17 files changed
+2881/−0 lines, 17 filesdep +aesondep +asyncdep +basesetup-changed
Dependencies added: aeson, async, base, binary, bytestring, clock, conduit, containers, crdt-event-fold, data-default-class, hostname, hspec, http-api-data, lens, lens-aeson, monad-logger, mtl, mustache, network, om-fork, om-kubernetes, om-legion, om-logging, om-show, om-socket, om-time, random-shuffle, safe, safe-exceptions, stm, template-haskell, text, time, transformers, unix, unliftio, unliftio-core, uuid, vector
Files
- LICENSE +19/−0
- README.md +33/−0
- Setup.hs +2/−0
- om-legion.cabal +137/−0
- src/OM/Legion.hs +76/−0
- src/OM/Legion/Conduit.hs +40/−0
- src/OM/Legion/Connection.hs +230/−0
- src/OM/Legion/MsgChan.hs +163/−0
- src/OM/Legion/Runtime.hs +1026/−0
- test/Test/OM/Legion.hs +136/−0
- test/k8s/k8s.mustache +111/−0
- test/om-legion-test-driver.hs +422/−0
- test/om-legion-test-inc.hs +34/−0
- test/om-legion-test-node.hs +154/−0
- test/om-legion-test-profile.hs +197/−0
- test/om-legion-test-run.hs +42/−0
- test/om-legion-test-stable.hs +59/−0
+ LICENSE view
@@ -0,0 +1,19 @@+Copyright 2022 Rick Owens++Permission is hereby granted, free of charge, to any person obtaining a+copy of this software and associated documentation files (the "Software"),+to deal in the Software without restriction, including without limitation+the rights to use, copy, modify, merge, publish, distribute, sublicense,+and/or sell copies of the Software, and to permit persons to whom the+Software is furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included+in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL+THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR+OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,+ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR+OTHER DEALINGS IN THE SOFTWARE.
+ README.md view
@@ -0,0 +1,33 @@+# om-legion++Legion is framework for building clusters of homogenous nodes that must+maintain a shared, replicated state.++For instance, maybe in order to scale a service you are using a partitioning+strategy to direct traffic for users A-H to node 1, users I-Q to node 2, and+R-Z to node 3. The partition table itself is something all the nodes have to+agree on and would make an excellent candidate for the "shared, replicated+state" managed by this framework.++To update the shared state, whatever it is, use:++* `applyFast`+* or `applyConsistent`++Additionally, since Legion already has to understand about different nodes in+the cluster, and how to talk to them, we provide a variety of tools to help+facilitate inter-node communication:++* `cast`: We support sending arbitrary messages to other peers without+ waiting on a response.++* `call`: We support sending requests to other peers that block on a+ response.++* `broadcast`: We support sending a message to _every_ other peer without+ waiting on a response++* `broadcall`: We support sending a request to every other peer, and blocking+ until we recieve a response from them all.++
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ om-legion.cabal view
@@ -0,0 +1,137 @@+cabal-version: 3.0+name: om-legion+version: 6.9.0.3+synopsis: Legion Framework.+description: Framework for managing shared, replicated state across a+ number of homogeneous nodes.+homepage: https://github.com/owensmurray/om-legion+license: MIT+license-file: LICENSE+author: Rick Owens+maintainer: rick@owensmurray.com+copyright: 2022 Owens Murray, LLC.+category: Network+build-type: Simple+extra-source-files:+ LICENSE+ README.md+ test/k8s/k8s.mustache++common dependencies+ build-depends:+ , aeson >= 2.0.3.0 && < 2.1+ , async >= 2.2.4 && < 2.3+ , base >= 4.15.0.0 && < 4.16+ , binary >= 0.8.8.0 && < 0.9+ , bytestring >= 0.10.12.1 && < 0.11+ , clock >= 0.8.3 && < 0.9+ , conduit >= 1.3.4.2 && < 1.4+ , containers >= 0.6.4.1 && < 0.7+ , crdt-event-fold >= 1.8.0.0 && < 1.9+ , data-default-class >= 0.1.2.0 && < 0.2+ , http-api-data >= 0.4.3 && < 0.5+ , monad-logger >= 0.3.36 && < 0.4+ , mtl >= 2.2.2 && < 2.3+ , network >= 3.1.2.7 && < 3.2+ , om-fork >= 0.7.1.3 && < 0.8+ , om-logging >= 1.1.0.1 && < 1.2+ , om-show >= 0.1.2.1 && < 0.2+ , om-socket >= 0.11.0.1 && < 0.12+ , om-time >= 0.3.0.0 && < 0.4+ , random-shuffle >= 0.0.4 && < 0.1+ , safe-exceptions >= 0.1.7.3 && < 0.2+ , stm >= 2.5.0.0 && < 2.6+ , text >= 1.2.5.0 && < 1.3+ , time >= 1.9.3 && < 1.10+ , transformers >= 0.5.6.2 && < 0.6+ , unliftio-core >= 0.2.0.1 && < 0.3+ , uuid >= 1.3.15 && < 1.4++common warnings+ ghc-options:+ -Wall+ -Wmissing-deriving-strategies+ -Wmissing-export-lists+ -Wmissing-import-lists+ -Wredundant-constraints++library+ import: dependencies, warnings+ exposed-modules: + OM.Legion+ other-modules: + OM.Legion.Conduit+ OM.Legion.Connection+ OM.Legion.MsgChan+ OM.Legion.Runtime+ -- other-extensions: + hs-source-dirs: src+ default-language: Haskell2010++common test-dependencies+ build-depends:+ , om-legion++ , hostname >= 1.0 && < 1.1+ , hspec >= 2.8.5 && < 2.9+ , lens >= 5.0.1 && < 5.1+ , lens-aeson >= 1.1.3 && < 1.2+ , mustache >= 2.4.1 && < 2.5+ , om-kubernetes >= 2.3.1.3 && < 2.4+ , safe >= 0.3.19 && < 0.4+ , template-haskell >= 2.17.0.0 && < 2.18+ , unix >= 2.7.2.2 && < 2.8+ , unliftio >= 0.2.22.0 && < 0.3+ , vector >= 0.12.3.1 && < 0.13++executable om-legion-test-node+ import: warnings, dependencies, test-dependencies+ main-is: om-legion-test-node.hs+ hs-source-dirs: test+ other-modules:+ Test.OM.Legion+ default-language: Haskell2010++executable om-legion-test-driver+ import: warnings, dependencies, test-dependencies+ main-is: om-legion-test-driver.hs+ hs-source-dirs: test+ other-modules:+ Test.OM.Legion+ default-language: Haskell2010++executable om-legion-test-run+ import: warnings, dependencies, test-dependencies+ main-is: om-legion-test-run.hs+ hs-source-dirs: test+ other-modules:+ Test.OM.Legion+ default-language: Haskell2010++executable om-legion-test-inc+ import: warnings, dependencies, test-dependencies+ main-is: om-legion-test-inc.hs+ hs-source-dirs: test+ other-modules:+ Test.OM.Legion+ default-language: Haskell2010+ ghc-options: -threaded++executable om-legion-test-stable+ import: warnings, dependencies, test-dependencies+ main-is: om-legion-test-stable.hs+ hs-source-dirs: test+ other-modules:+ Test.OM.Legion+ default-language: Haskell2010++executable om-legion-test-profile+ import: warnings, dependencies, test-dependencies+ main-is: om-legion-test-profile.hs+ hs-source-dirs: test+ other-modules:+ Test.OM.Legion+ default-language: Haskell2010+ ghc-options:+ -threaded+
+ src/OM/Legion.hs view
@@ -0,0 +1,76 @@++{- |+ Legion is framework for building clusters of homogenous nodes that must+ maintain a shared, replicated state.++ For instance, maybe in order to scale a service you are using a partitioning+ strategy to direct traffic for users A-H to node 1, users I-Q to node 2, and+ R-Z to node 3. The partition table itself is something all the nodes have to+ agree on and would make an excellent candidate for the "shared, replicated+ state" managed by this framework.++ To update the shared state, whatever it is, use:++ * 'applyFast'+ * or 'applyConsistent'++ Additionally, since Legion already has to understand about different nodes in+ the cluster, and how to talk to them, we provide a variety of tools to help+ facilitate inter-node communication:++ * 'cast': We support sending arbitrary messages to other peers without+ waiting on a response.++ * 'call': We support sending requests to other peers that block on a+ response.++ * 'broadcast': We support sending a message to _every_ other peer without+ waiting on a response++ * 'broadcall': We support sending a request to every other peer, and blocking+ until we recieve a response from them all.+-}+module OM.Legion (+ -- * Starting up the runtime.+ forkLegionary,+ EventConstraints,+ MonadConstraints,+ StartupMode(..),+ Runtime,++ -- * Applying state changes.+ applyFast,+ applyConsistent,++ -- * Sending messages around the cluster.+ cast,+ call,+ broadcast,+ broadcall,++ -- * Inspecting the current state.+ readState,+ getSelf,+ getClusterName,++ -- * Observability+ getStats,+ Stats(..),++ -- * Cluster Topology+ eject,+ Peer(..),+ ClusterName(..),++) where+++import OM.Legion.Connection (EventConstraints)+import OM.Legion.MsgChan (ClusterName(ClusterName, unClusterName),+ Peer(Peer, unPeer))+import OM.Legion.Runtime (StartupMode(JoinCluster, NewCluster, Recover),+ Stats(Stats, timeWithoutProgress), MonadConstraints, Runtime,+ applyConsistent, applyFast, broadcall, broadcast, call, cast, eject,+ forkLegionary, getClusterName, getSelf, getStats, readState)++
+ src/OM/Legion/Conduit.hs view
@@ -0,0 +1,40 @@+{- | This module contains some handy conduit abstractions. -}+module OM.Legion.Conduit (+ chanToSource,+ chanToSink,+) where+++import Control.Concurrent.Chan (Chan, readChan, writeChan)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Data.Conduit (ConduitT, awaitForever, yield)+++{- | Convert a channel into a source conduit. -}+chanToSource :: (MonadIO m) => Chan a -> ConduitT void a m ()+chanToSource chan = do+ {-+ Don't use 'Control.Monad.forever' here. For some reason that is unclear to+ me, use of 'forever' creates a space leak, despite the comments in the+ 'forever' source code.+ + The code:++ > forever $ yield =<< liftIO (readChan chan)++ will reliably leak several megabytes of memory over the course of 10k+ messages when tested using the 'legion-discovery' project. This was+ discovered by @-hr@ heap profiling, which pointed to 'chanToSource'+ as the retainer. I think it didn't point to 'forever' as the retainer+ because 'forever' is inlined, and thus does not have a cost-centre+ associated with it.+ -}+ yield =<< liftIO (readChan chan)+ chanToSource chan+++{- | Convert a channel into a sink conduit. -}+chanToSink :: (MonadIO m) => Chan a -> ConduitT a void m ()+chanToSink chan = awaitForever (liftIO . writeChan chan)++
+ src/OM/Legion/Connection.hs view
@@ -0,0 +1,230 @@+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE UndecidableInstances #-}++{- | Description: Manage connections to other peers. -}+module OM.Legion.Connection (+ JoinResponse(..),+ RuntimeState(..),+ EventConstraints,+ disconnect,+ peerMessagePort,+ sendPeer,+) where++import Control.Concurrent.Async (async)+import Control.Exception.Safe (MonadCatch, finally, tryAny)+import Control.Monad (void)+import Control.Monad.IO.Class (MonadIO(liftIO))+import Control.Monad.Logger.CallStack (LoggingT(runLoggingT),+ MonadLoggerIO(askLoggerIO), MonadLogger, logDebug, logInfo)+import Control.Monad.State (MonadState(get), modify)+import Data.Aeson (ToJSON)+import Data.Binary (Binary)+import Data.ByteString.Lazy (ByteString)+import Data.CRDT.EventFold (Event(Output, State), EventFold, EventId)+import Data.Conduit ((.|), ConduitT, awaitForever, runConduit, yield)+import Data.Default.Class (Default)+import Data.Map (Map)+import GHC.Generics (Generic)+import Network.Socket (PortNumber)+import OM.Fork (Responder)+import OM.Legion.MsgChan (Peer(unPeer), ClusterName, MessageId,+ PeerMessage, close, enqueueMsg, newMsgChan, stream)+import OM.Show (showt)+import OM.Socket (AddressDescription(AddressDescription), openEgress)+import System.Clock (TimeSpec)+import qualified Data.Map as Map+++{- | A handle on the connection to a peer. -}+newtype Connection e = Connection+ { _unConnection+ :: forall m.+ ( MonadIO m+ , MonadLogger m+ , MonadState (RuntimeState e) m+ )+ => PeerMessage e+ -> m Bool+ }+++{- | Create a connection to a peer. -}+createConnection+ :: forall m e.+ ( EventConstraints e+ , MonadCatch m+ , MonadLoggerIO m+ , MonadState (RuntimeState e) m+ )+ => Peer+ -> m (Connection e)+createConnection peer = do+ logInfo $ "Creating connection to: " <> showt peer+ RuntimeState {rsSelf} <- get+ msgChan <- newMsgChan+ logging <- askLoggerIO+ liftIO . void . async . (`runLoggingT` logging) $+ let+ addy :: AddressDescription+ addy =+ AddressDescription+ (+ unPeer peer+ <> ":" <> showt peerMessagePort+ )+ in+ finally + (+ (tryAny . runConduit) (+ stream rsSelf msgChan+ .| logMessageSend+ .| openEgress addy+ ) >>= \case+ Left err ->+ logInfo $ "Disconnecting because of error: " <> showt err+ Right () -> logInfo "Disconnecting because source dried up."+ )+ (close msgChan)+ + let+ conn :: Connection e+ conn = Connection (enqueueMsg msgChan)+ modify+ (\state -> state {+ rsConnections = Map.insert peer conn (rsConnections state)+ })+ pure conn+ where+ logMessageSend+ :: forall w.+ (MonadLogger w)+ => ConduitT (Peer, PeerMessage e) (Peer, PeerMessage e) w ()+ logMessageSend =+ awaitForever+ (\msg -> do+ logDebug+ $ "Sending Message to Peer (peer, msg): "+ <> showt (peer, msg)+ yield msg+ )+++{- |+ Shorthand for all the constraints needed for the event type. Mainly+ used so that documentation renders better.+-}+type EventConstraints e =+ ( Binary (Output e)+ , Binary (State e)+ , Binary e+ , Default (State e)+ , Eq (Output e)+ , Eq e+ , Event Peer e+ , Show (Output e)+ , Show (State e)+ , Show e+ , ToJSON (Output e)+ , ToJSON (State e)+ , ToJSON e+ )+++{- | The Legionary runtime state. -}+data RuntimeState e = RuntimeState+ { rsSelf :: Peer+ , rsClusterState :: EventFold ClusterName Peer e+ , rsConnections :: Map Peer (Connection e)+ , rsWaiting :: Map (EventId Peer) (Responder (Output e))+ , rsCalls :: Map MessageId (Responder ByteString)+ , rsBroadcalls :: Map+ MessageId+ (+ Map Peer (Maybe ByteString),+ Responder (Map Peer (Maybe ByteString)),+ TimeSpec+ )+ , rsNextId :: MessageId+ , rsNotify :: EventFold ClusterName Peer e -> IO ()+ , rsJoins :: Map+ (EventId Peer)+ (Responder (JoinResponse e))+ {- ^+ The infimum of the eventfold we send to a+ new participant must have moved past the+ participation event itself. In other words,+ the join must be totally consistent across the+ cluster. The reason is that we can't make the+ new participant responsible for applying events+ that occur before it joined the cluster, because+ it has no way to ensure that it can collect all+ such events. Therefore, this field tracks the+ outstanding joins until they become consistent.+ -}+ }+++{- | The response to a JoinRequest message -}+newtype JoinResponse e+ = JoinOk (EventFold ClusterName Peer e)+ deriving stock (Generic)+deriving stock instance (EventConstraints e) => Show (JoinResponse e)+instance (EventConstraints e) => Binary (JoinResponse e)+++{- | The peer message port. -}+peerMessagePort :: PortNumber+peerMessagePort = 5288+++{- | Disconnect the connection to a peer. -}+disconnect+ :: ( MonadLogger m+ , MonadState (RuntimeState e) m+ )+ => Peer+ -> m ()+disconnect peer = do+ logInfo $ "Disconnecting: " <> showt peer+ modify (\state@RuntimeState {rsConnections} -> state {+ rsConnections = Map.delete peer rsConnections+ })+++{- | Send a peer message, creating a new connection if need be. -}+sendPeer+ :: forall m e.+ ( EventConstraints e+ , MonadCatch m+ , MonadLoggerIO m+ , MonadState (RuntimeState e) m+ )+ => PeerMessage e+ -> Peer+ -> m ()+sendPeer msg peer = do+ RuntimeState {rsConnections} <- get+ case Map.lookup peer rsConnections of+ Nothing -> do+ conn <- createConnection peer+ sendTheMessage conn+ Just conn ->+ sendTheMessage conn+ where+ sendTheMessage :: Connection e -> m ()+ sendTheMessage (Connection conn) =+ conn msg >>= \case+ True -> pure ()+ False -> disconnect peer++
+ src/OM/Legion/MsgChan.hs view
@@ -0,0 +1,163 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE UndecidableInstances #-}++{- | Description: A specialized channel for outgoing peer messages. -}+module OM.Legion.MsgChan (+ MsgChan,+ Peer(..),+ ClusterName(..),+ MessageId(..),+ PeerMessage(..),+ close,+ enqueueMsg,+ newMsgChan,+ stream,+) where++import Control.Concurrent.STM (TVar, atomically, newTVarIO, readTVar,+ retry, writeTVar)+import Control.Monad (join)+import Control.Monad.IO.Class (MonadIO(liftIO))+import Data.Aeson (FromJSON, FromJSONKey, ToJSON, ToJSONKey)+import Data.Binary (Binary, Word64)+import Data.ByteString.Lazy (ByteString)+import Data.CRDT.EventFold (Event(Output, State), Diff, EventFold,+ EventId)+import Data.Conduit (ConduitT, yield)+import Data.Foldable (traverse_)+import Data.Map (Map)+import Data.String (IsString)+import Data.Text (Text)+import Data.UUID (UUID)+import GHC.Generics (Generic)+import Web.HttpApiData (FromHttpApiData, ToHttpApiData)+++{- | A specialized channel for outgoing peer messages. -}+newtype MsgChan e = MsgChan+ { _unMsgChan :: TVar (Maybe [PeerMessage e])+ }+++{- | Create a new 'MsgChan'. -}+newMsgChan :: (MonadIO m) => m (MsgChan e)+newMsgChan = liftIO $ MsgChan <$> newTVarIO (Just [])+++{- | Close a MsgChan. -}+close :: (MonadIO m) => MsgChan e -> m ()+close (MsgChan chan) =+ (liftIO . atomically) (writeTVar chan Nothing)+++stream+ :: MonadIO m+ => Peer+ -> MsgChan e+ -> ConduitT void (Peer, PeerMessage e) m ()+stream self (MsgChan chan) =+ (join . liftIO . atomically) $+ readTVar chan >>= \case+ Nothing ->+ {- The stream is closed, so stop. -}+ pure (pure ())+ Just [] ->+ {- The stream is empty, so wait. -}+ retry+ Just messages -> do+ {-+ The stream is not empty, consume the messages and set the+ stream to empty.+ -}+ writeTVar chan (Just []) + pure $ do+ traverse_ (yield . (self,)) messages+ stream self (MsgChan chan)+++{- | Enqueue a message to be sent. Return 'False' if the 'MsgChan' is closed. -}+enqueueMsg :: (MonadIO m) => MsgChan e -> PeerMessage e -> m Bool+enqueueMsg (MsgChan chan) msg =+ join . liftIO . atomically $+ readTVar chan >>= \case+ Nothing -> pure (pure False)+ Just msgs -> do+ writeTVar chan (+ Just $ case msg of+ PMMerge _ ->+ msg : filter+ (\case {PMMerge _ -> False; _ -> True})+ msgs+ PMFullMerge _ ->+ {-+ Full merges override both older full merges and+ older partial merges.+ -}+ msg : filter+ (\case+ PMMerge _ -> False+ PMFullMerge _ -> False+ _ -> True+ )+ msgs+ _ -> msgs ++ [msg] + )+ pure (pure True)+++{- | The types of messages that can be sent from one peer to another. -}+data PeerMessage e+ = PMMerge (Diff ClusterName Peer e)+ {- ^ Send a powerstate merge. -}+ | PMFullMerge (EventFold ClusterName Peer e)+ {- ^ Send a full merge. -}+ | PMCall Peer MessageId ByteString+ {- ^ Send a user call message from one peer to another. -}+ | PMCast ByteString+ {- ^ Send a user cast message from one peer to another. -}+ | PMCallResponse Peer MessageId ByteString+ {- ^ Send a response to a user call message. -}+ | PMOutputs (Map (EventId Peer) (Output e))+ {- ^ Send outputs to the appropriate recipient. -}+ deriving stock (Generic)+deriving stock instance+ ( Show e+ , Show (Output e)+ , Show (State e)+ )+ =>+ Show (PeerMessage e)+instance (Binary e, Binary (Output e), Binary (State e)) => Binary (PeerMessage e)+++{- | The identification of a node within the legion cluster. -}+newtype Peer = Peer+ { unPeer :: Text+ }+ deriving newtype (+ Eq, Ord, Show, ToJSON, Binary, ToJSONKey, FromJSON, FromJSONKey,+ FromHttpApiData, ToHttpApiData+ )+++{- | The name of a cluster. -}+newtype ClusterName = ClusterName+ { unClusterName :: Text+ }+ deriving newtype (+ IsString, Show, Binary, Eq, ToJSON, FromJSON, Ord, ToHttpApiData,+ FromHttpApiData+ )+++{- | Message Identifier. -}+data MessageId = M UUID Word64 deriving stock (Generic, Show, Eq, Ord)+instance Binary MessageId++
+ src/OM/Legion/Runtime.hs view
@@ -0,0 +1,1026 @@+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}++{- | The meat of the Legion runtime implementation. -}+module OM.Legion.Runtime (+ -- * Starting the framework runtime.+ forkLegionary,+ Runtime,+ StartupMode(..),++ -- * Constraints+ MonadConstraints,++ -- * Runtime Interface.+ applyFast,+ applyConsistent,+ readState,+ call,+ cast,+ broadcall,+ broadcast,+ eject,+ getSelf,+ getClusterName,+ getStats,+ Stats(..),+) where++import Control.Arrow ((&&&))+import Control.Concurrent (Chan, newChan, readChan, threadDelay,+ writeChan)+import Control.Exception.Safe (MonadCatch, tryAny)+import Control.Monad (unless, void, when)+import Control.Monad.IO.Class (MonadIO(liftIO))+import Control.Monad.IO.Unlift (MonadUnliftIO)+import Control.Monad.Logger.CallStack (LoggingT(runLoggingT),+ MonadLoggerIO(askLoggerIO), LogStr, MonadLogger, logDebug, logError,+ logInfo)+import Control.Monad.State (MonadState(get, put), StateT, evalStateT,+ gets, modify')+import Control.Monad.Trans.Class (lift)+import Data.Aeson (ToJSON)+import Data.Binary (Binary)+import Data.ByteString.Lazy (ByteString)+import Data.CRDT.EventFold (Event(Output, State),+ UpdateResult(urEventFold, urOutputs), Diff, EventFold, EventId,+ infimumId, projParticipants)+import Data.CRDT.EventFold.Monad (MonadUpdateEF(diffMerge, disassociate,+ event, fullMerge, participate), EventFoldT, runEventFoldT)+import Data.Conduit ((.|), awaitForever, runConduit, yield)+import Data.Default.Class (Default)+import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef)+import Data.Map (Map)+import Data.Set ((\\), Set)+import Data.Time (DiffTime, diffTimeToPicoseconds, picosecondsToDiffTime)+import Data.UUID (UUID)+import Data.UUID.V1 (nextUUID)+import GHC.Generics (Generic)+import Network.Socket (PortNumber)+import OM.Fork (Actor(Msg, actorChan), Race, Responder, race)+import OM.Legion.Conduit (chanToSink)+import OM.Legion.Connection (JoinResponse(JoinOk),+ RuntimeState(RuntimeState, rsBroadcalls, rsCalls, rsClusterState,+ rsConnections, rsJoins, rsNextId, rsNotify, rsSelf, rsWaiting),+ EventConstraints, disconnect, peerMessagePort, sendPeer)+import OM.Legion.MsgChan (MessageId(M), Peer(unPeer), PeerMessage(PMCall,+ PMCallResponse, PMCast, PMFullMerge, PMMerge, PMOutputs), ClusterName)+import OM.Logging (withPrefix)+import OM.Show (showj, showt)+import OM.Socket (AddressDescription(AddressDescription), connectServer,+ openIngress, openServer)+import OM.Time (MonadTimeSpec(getTime), addTime, diffTimeSpec)+import System.Clock (TimeSpec)+import System.Random.Shuffle (shuffleM)+import qualified Data.Binary as Binary+import qualified Data.CRDT.EventFold as EF+import qualified Data.Map as Map+import qualified Data.Set as Set+import qualified OM.Fork as Fork++{-# ANN module ("HLint: ignore Redundant <$>" :: String) #-}+{-# ANN module ("HLint: ignore Use underscore" :: String) #-}++{- |+ Shorthand for all the monad constraints, mainly use so that+ documentation renders better.+-}+type MonadConstraints m =+ ( MonadCatch m+ , MonadFail m+ , MonadLoggerIO m+ , MonadTimeSpec m+ , MonadUnliftIO m+ , Race+ )+++{- | Fork the Legion runtime system. -}+forkLegionary+ :: ( EventConstraints e+ , MonadConstraints m+ )+ => (ByteString -> IO ByteString) {- ^ Handle a user call request. -}+ -> (ByteString -> IO ()) {- ^ Handle a user cast message. -}+ -> (Peer -> EventFold ClusterName Peer e -> IO ())+ {- ^ Callback when the cluster-wide eventfold changes. -}+ -> Int+ {- ^+ The propagation interval, in microseconds (for use with+ `threadDelay`).+ -}+ -> StartupMode e+ {- ^+ How to start the runtime, by creating new cluster or joining an+ existing cluster.+ -}+ -> m (Runtime e)+forkLegionary+ handleUserCall+ handleUserCast+ notify+ resendInterval+ startupMode+ = do+ logInfo $ "Starting up with the following Mode: " <> showt startupMode+ rts <- makeRuntimeState notify startupMode+ runtimeChan <- RChan <$> liftIO newChan+ logging <- withPrefix (logPrefix (rsSelf rts)) <$> askLoggerIO+ rStats <- liftIO $ newIORef mempty+ (`runLoggingT` logging) $+ executeRuntime+ handleUserCall+ handleUserCast+ resendInterval+ rts+ runtimeChan+ rStats+ let+ clusterId :: ClusterName+ clusterId = EF.origin (rsClusterState rts)+ return+ Runtime+ { rChan = runtimeChan+ , rSelf = rsSelf rts+ , rClusterId = clusterId+ , rStats+ }+ where+ logPrefix :: Peer -> LogStr+ logPrefix self_ = "[" <> showt self_ <> "] "+++{- | A handle on the Legion runtime. -}+data Runtime e = Runtime+ { rChan :: RChan e+ , rSelf :: Peer+ , rClusterId :: ClusterName+ , rStats :: IORef (Map Peer TimeSpec)+ }+instance Actor (Runtime e) where+ type Msg (Runtime e) = RuntimeMessage e+ actorChan = actorChan . rChan+++{- |+ Some basic stats that can be used to intuit the health of the cluster.+ We currently only report on how long it has been since some peer has+ made some progress.+-}+newtype Stats = Stats+ { timeWithoutProgress :: Map Peer DiffTime+ {- ^+ How long it has been since a 'Peer' has+ made progress (if it is 'divergent'). If+ the peer is completely up to date as far as+ we know, then it does not appear in the map+ at all. Only peers which we are expecting+ to make progress appear.+ -}+ }+ deriving stock (Generic, Show, Eq)+ deriving anyclass (ToJSON)+instance Binary Stats where+ get =+ Stats <$> (fmap picosecondsToDiffTime <$> Binary.get)+ put (Stats timeWithoutProgress) =+ Binary.put (diffTimeToPicoseconds <$> timeWithoutProgress)++ +{- | The type of the runtime message channel. -}+newtype RChan e = RChan {+ unRChan :: Chan (RuntimeMessage e)+ }+instance Actor (RChan e) where+ type Msg (RChan e) = RuntimeMessage e+ actorChan = writeChan . unRChan+++{- |+ Update the distributed cluster state by applying an event. The event+ output will be returned immediately and may not reflect a totally+ consistent view of the cluster. The state update itself, however,+ is guaranteed to be applied atomically and consistently throughout+ the cluster.+-}+applyFast :: (MonadIO m)+ => Runtime e {- ^ The runtime handle. -}+ -> e {- ^ The event to be applied. -}+ -> m (Output e) {- ^ Returns the possibly inconsistent event output. -}+applyFast runtime e = Fork.call runtime (ApplyFast e)+++{- |+ Update the distributed cluster state by applying an event. Both the+ event output and resulting state will be totally consistent throughout+ the cluster.+-}+applyConsistent :: (MonadIO m)+ => Runtime e {- ^ The runtime handle. -}+ -> e {- ^ The event to be applied. -}+ -> m (Output e) {- ^ Returns the strongly consistent event output. -}+applyConsistent runtime e = Fork.call runtime (ApplyConsistent e)+++{- | Read the current powerstate value. -}+readState :: (MonadIO m)+ => Runtime e+ -> m (EventFold ClusterName Peer e)+readState runtime = Fork.call runtime ReadState+++{- |+ Send a user message to some other peer, and block until a response+ is received.+-}+call :: (MonadIO m) => Runtime e -> Peer -> ByteString -> m ByteString+call runtime target msg = Fork.call runtime (Call target msg)+++{- | Send the result of a call back to the peer that originated it. -}+sendCallResponse :: (MonadIO m)+ => RChan e+ -> Peer+ -> MessageId+ -> ByteString+ -> m ()+sendCallResponse runtimeChan target mid msg =+ Fork.cast runtimeChan (SendCallResponse target mid msg)+++{- | Send a user message to some other peer, without waiting on a response. -}+cast :: (MonadIO m) => Runtime e -> Peer -> ByteString -> m ()+cast runtime target message = Fork.cast runtime (Cast target message)+++{- |+ Send a user message to all peers, and block until a response is received+ from all of them.+-}+broadcall :: (MonadIO m)+ => Runtime e+ -> DiffTime {- ^ The timeout. -}+ -> ByteString+ -> m (Map Peer (Maybe ByteString))+broadcall runtime timeout msg = Fork.call runtime (Broadcall timeout msg)+++{- | Send a user message to all peers, without wating on a response. -}+broadcast :: (MonadIO m) => Runtime e -> ByteString -> m ()+broadcast runtime msg = Fork.cast runtime (Broadcast msg)+++{- | Eject a peer from the cluster. -}+eject :: (MonadIO m) => Runtime e -> Peer -> m ()+eject runtime peer = Fork.call runtime (Eject peer)+++{- | Get the identifier for the local peer. -}+getSelf :: Runtime e -> Peer+getSelf = rSelf+++{- | The types of messages that can be sent to the runtime. -}+data RuntimeMessage e+ = ApplyFast e (Responder (Output e))+ | ApplyConsistent e (Responder (Output e))+ | Eject Peer (Responder ())+ | Merge (Diff ClusterName Peer e)+ | FullMerge (EventFold ClusterName Peer e)+ | Outputs (Map (EventId Peer) (Output e))+ | Join JoinRequest (Responder (JoinResponse e))+ | ReadState (Responder (EventFold ClusterName Peer e))+ | Call Peer ByteString (Responder ByteString)+ | Cast Peer ByteString+ | Broadcall+ DiffTime+ ByteString+ (Responder (Map Peer (Maybe ByteString)))+ | Broadcast ByteString+ | SendCallResponse Peer MessageId ByteString+ | HandleCallResponse Peer MessageId ByteString+ | Resend (Responder ())+ | GetStats (Responder (EventFold ClusterName Peer e))+deriving stock instance+ ( Show e+ , Show (Output e)+ , Show (State e)+ )+ =>+ Show (RuntimeMessage e)+++{- |+ Execute the Legion runtime, with the given user definitions, and+ framework settings. This function never returns (except maybe with an+ exception if something goes horribly wrong).+-}+executeRuntime+ :: ( Binary (Output e)+ , Binary (State e)+ , Binary e+ , Default (State e)+ , Eq (Output e)+ , Eq e+ , Event Peer e+ , MonadCatch m+ , MonadFail m+ , MonadLoggerIO m+ , MonadTimeSpec m+ , MonadUnliftIO m+ , Race+ , Show (Output e)+ , Show (State e)+ , Show e+ , ToJSON (Output e)+ , ToJSON (State e)+ , ToJSON e+ )+ => (ByteString -> IO ByteString)+ {- ^ Handle a user call request. -}+ -> (ByteString -> IO ())+ {- ^ Handle a user cast message. -}+ -> Int+ {- ^+ The propagation interval, in microseconds (for use with+ `threadDelay`).+ -}+ -> RuntimeState e+ -> RChan e+ {- ^+ A source of requests, together with a way to respond to the+ requets.+ -}+ -> IORef (Map Peer TimeSpec)+ -> m ()+executeRuntime+ handleUserCall+ handleUserCast+ resendInterval+ rts+ runtimeChan+ peerStats+ = do+ {- Start the various messages sources. -}+ race "om-legion peer listener" runPeerListener+ race "om-legion join listener" runJoinListener+ race "om-legion periodic resend" runPeriodicResent+ race "om-legion message handler" $+ (`evalStateT` rts)+ (+ let+ -- handleMessages :: StateT (RuntimeState e3) m Void+ handleMessages = do+ msg <- liftIO $ readChan (unRChan runtimeChan)+ RuntimeState {rsClusterState = cluster1} <- get+ logDebug $ "Handling: " <> showt msg+ handleRuntimeMessage msg+ RuntimeState {rsClusterState = cluster2} <- get+ when (cluster1 /= cluster2) $+ logDebug $ "New Cluster State: " <> showj cluster2+ -- propagate+ handleBroadcallTimeouts+ handleOutstandingJoins+ handleMessages+ in do+ liftIO $ rsNotify rts (rsClusterState rts)+ handleMessages+ )+ where+ runPeerListener :: (MonadLoggerIO m, MonadFail m) => m ()+ runPeerListener =+ let+ addy :: AddressDescription+ addy =+ AddressDescription+ (+ unPeer (rsSelf rts)+ <> ":" <> showt peerMessagePort+ )+ in+ runConduit (+ openIngress addy+ .| awaitForever (\ (msgSource, msg) -> do+ liftIO $ do+ now <- getTime+ atomicModifyIORef'+ peerStats+ (\peerTimes -> (Map.insert msgSource now peerTimes, ()))+ logDebug $ "Handling: " <> showt (msgSource :: Peer, msg)+ case msg of+ PMFullMerge ps -> yield (FullMerge ps)+ PMOutputs outputs -> yield (Outputs outputs)+ PMMerge ps -> yield (Merge ps)+ PMCall source mid callMsg ->+ (liftIO . tryAny) (handleUserCall callMsg) >>= \case+ Left err ->+ logError+ $ "User call handling failed with: " <> showt err+ Right v -> sendCallResponse runtimeChan source mid v+ PMCast castMsg -> liftIO (handleUserCast castMsg)+ PMCallResponse source mid responseMsg ->+ yield (HandleCallResponse source mid responseMsg)+ )+ .| chanToSink (unRChan runtimeChan)+ )++ runJoinListener :: (MonadLoggerIO m, MonadFail m) => m ()+ runJoinListener =+ let+ addy :: AddressDescription+ addy =+ AddressDescription+ (+ unPeer (rsSelf rts)+ <> ":" <> showt joinMessagePort+ )+ in+ runConduit (+ pure ()+ .| openServer addy Nothing+ .| awaitForever (\(req, respond_) -> lift $+ Fork.call runtimeChan (Join req) >>= respond_+ )+ )++ runPeriodicResent :: (MonadIO m) => m ()+ runPeriodicResent =+ let+ periodicResend :: (MonadIO m) => m ()+ periodicResend = do+ liftIO $ threadDelay resendInterval+ Fork.call runtimeChan Resend+ periodicResend+ in+ periodicResend+++{- | Handle any outstanding joins. -}+handleOutstandingJoins :: (MonadLoggerIO m) => StateT (RuntimeState e) m ()+handleOutstandingJoins = do+ state@RuntimeState {rsJoins, rsClusterState} <- get+ let+ (consistent, pending) =+ Map.partitionWithKey+ (\k _ -> k <= infimumId rsClusterState)+ rsJoins+ put state {rsJoins = pending}+ sequence_ [+ do+ logInfo $ "Completing join (" <> showt sid <> ")."+ respond responder (JoinOk rsClusterState)+ | (sid, responder) <- Map.toList consistent+ ]+++{- | Handle any broadcall timeouts. -}+handleBroadcallTimeouts+ :: ( MonadIO m+ , MonadTimeSpec m+ )+ => StateT (RuntimeState e) m ()+handleBroadcallTimeouts = do+ broadcalls <- gets rsBroadcalls+ now <- getTime+ sequence_ [+ do+ respond responder responses+ modify' (\rs -> rs {+ rsBroadcalls = Map.delete messageId (rsBroadcalls rs)+ })+ | (messageId, (responses, responder, expiry)) <- Map.toList broadcalls+ , now >= expiry+ ]+++{- | Execute the incoming messages. -}+handleRuntimeMessage+ :: ( Binary (Output e)+ , Binary (State e)+ , Binary e+ , Default (State e)+ , Eq (Output e)+ , Eq e+ , Event Peer e+ , MonadCatch m+ , MonadLoggerIO m+ , MonadTimeSpec m+ , Show (Output e)+ , Show (State e)+ , Show e+ , ToJSON (Output e)+ , ToJSON (State e)+ , ToJSON e+ )+ => RuntimeMessage e+ -> StateT (RuntimeState e) m ()++handleRuntimeMessage (Outputs outputs) =+ {-# SCC "Outputs" #-}+ respondToWaiting outputs++handleRuntimeMessage (GetStats responder) =+ {-# SCC "GetStats" #-}+ respond responder =<< gets rsClusterState++handleRuntimeMessage (ApplyFast e responder) =+ {-# SCC "ApplyFast" #-}+ updateCluster $+ fst <$> event e >>= respond responder++handleRuntimeMessage (ApplyConsistent e responder) =+ {-# SCC "ApplyConsistent" #-}+ (+ do+ updateCluster $ do+ (_v, sid) <- event e+ lift (waitOn sid responder)+ rs <- get+ logDebug $ "Waiting: " <> showt (rsWaiting rs)+ )++handleRuntimeMessage (Eject peer responder) =+ {-# SCC "Eject" #-}+ do+ updateClusterAs peer $+ void $ disassociate peer+ propagate+ {- ↓+ This is an awful hack. The problem is that 'propagate' uses+ 'sendPeer', but 'sendPeer' itself is asynchronous (though it should be+ very fast). The correct solution is a bit tricky. We can either figure+ out some way to block all the way down through the internals of the+ connection management, or else maybe extend the "join port" server+ endpoint to accept eject notifications as well as join requests.+ -}+ liftIO $ threadDelay 500_000+ respond responder ()++handleRuntimeMessage (Merge other) =+ {-# SCC "Merge" #-}+ updateCluster $+ diffMerge other >>= \case+ Left err -> logError $ "Bad cluster merge: " <> showt err+ Right () -> return ()++handleRuntimeMessage (FullMerge other) =+ {-# SCC "FullMerge" #-}+ updateCluster $+ fullMerge other >>= \case+ Left err -> logError $ "Bad cluster merge: " <> showt err+ Right () -> return ()++handleRuntimeMessage (Join (JoinRequest peer) responder) =+ {-# SCC "Join" #-}+ do+ logInfo $ "Handling join from peer: " <> showt peer+ updateCluster (do+ void $ disassociate peer+ void $ participate peer+ )+ RuntimeState {rsClusterState} <- get+ logInfo $ "Join immediately with: " <> showt rsClusterState+ respond responder (JoinOk rsClusterState)++handleRuntimeMessage (ReadState responder) =+ {-# SCC "ReadState" #-}+ respond responder . rsClusterState =<< get++handleRuntimeMessage (Call target msg responder) =+ {-# SCC "Call" #-}+ do+ mid <- newMessageId+ source <- gets rsSelf+ setCallResponder mid+ sendPeer (PMCall source mid msg) target+ where+ setCallResponder :: (Monad m)+ => MessageId+ -> StateT (RuntimeState e) m ()+ setCallResponder mid = do+ state@RuntimeState {rsCalls} <- get+ put state {+ rsCalls = Map.insert mid responder rsCalls+ }++handleRuntimeMessage (Cast target msg) =+ {-# SCC "Cast" #-}+ sendPeer (PMCast msg) target++handleRuntimeMessage (Broadcall timeout msg responder) =+ {-# SCC "Broadcall" #-}+ do+ expiry <- addTime timeout <$> getTime+ mid <- newMessageId+ source <- gets rsSelf+ setBroadcallResponder expiry mid+ mapM_ (sendPeer (PMCall source mid msg)) =<< getPeers+ where+ setBroadcallResponder :: (Monad m)+ => TimeSpec+ -> MessageId+ -> StateT (RuntimeState e) m ()+ setBroadcallResponder expiry mid = do+ peers <- getPeers+ state@RuntimeState {rsBroadcalls} <- get+ put state {+ rsBroadcalls =+ Map.insert+ mid+ (+ Map.fromList [(peer, Nothing) | peer <- Set.toList peers],+ responder,+ expiry+ )+ rsBroadcalls+ }++handleRuntimeMessage (Broadcast msg) =+ {-# SCC "Broadcast" #-}+ mapM_ (sendPeer (PMCast msg)) =<< getPeers++handleRuntimeMessage (SendCallResponse target mid msg) =+ {-# SCC "SendCallResponse" #-}+ do+ source <- gets rsSelf+ sendPeer (PMCallResponse source mid msg) target++handleRuntimeMessage (HandleCallResponse source mid msg) =+ {-# SCC "HandleCallResponse" #-}+ do+ state@RuntimeState {rsCalls, rsBroadcalls} <- get+ case Map.lookup mid rsCalls of+ Nothing ->+ case Map.lookup mid rsBroadcalls of+ Nothing -> return ()+ Just (responses, responder, expiry) ->+ let+ responses2 = Map.insert source (Just msg) responses+ response = Map.fromList [+ (peer, r)+ | (peer, Just r) <- Map.toList responses2+ ]+ peers = Map.keysSet responses2+ in+ if Set.null (peers \\ Map.keysSet response)+ then do+ respond responder (Just <$> response)+ put state {+ rsBroadcalls = Map.delete mid rsBroadcalls+ }+ else+ put state {+ rsBroadcalls =+ Map.insert+ mid+ (responses2, responder, expiry)+ rsBroadcalls+ }+ Just responder -> do+ respond responder msg+ put state {rsCalls = Map.delete mid rsCalls}++handleRuntimeMessage (Resend responder) =+ {-# SCC "Resend" #-}+ propagate >>= respond responder+++{- | Get the projected peers. -}+getPeers :: (Monad m) => StateT (RuntimeState e) m (Set Peer)+getPeers = gets (projParticipants . rsClusterState)+++{- | Get a new messageId. -}+newMessageId :: (Monad m) => StateT (RuntimeState e) m MessageId+newMessageId = do+ state@RuntimeState {rsNextId} <- get+ put state {rsNextId = nextMessageId rsNextId}+ return rsNextId+++{- |+ Like 'runEventFoldT', plus automatically take care of doing necessary+ IO implied by the cluster update.+-}+updateCluster+ :: ( EventConstraints e+ , MonadCatch m+ , MonadLoggerIO m+ , MonadState (RuntimeState e) m+ )+ => EventFoldT ClusterName Peer e m a+ -> m a+updateCluster action = do+ RuntimeState {rsSelf} <- get+ updateClusterAs rsSelf action+++{- |+ Like 'updateCluster', but perform the operation on behalf of a specified+ peer. This is required for e.g. the peer eject case, when the ejected peer+ may not be able to perform acknowledgements on its own behalf.+-}+updateClusterAs+ :: forall e m a.+ ( EventConstraints e+ , MonadCatch m+ , MonadLoggerIO m+ , MonadState (RuntimeState e) m+ )+ => Peer+ -> EventFoldT ClusterName Peer e m a+ -> m a+updateClusterAs asPeer action = do+ (oldCluster, notify)+ <- gets (rsClusterState &&& rsNotify)+ (v, ur) <- runEventFoldT asPeer oldCluster action+ do {- Update the cluster -}+ let+ newCluster :: EventFold ClusterName Peer e+ newCluster = urEventFold ur+ when (oldCluster /= newCluster) $ liftIO (notify newCluster)+ modify'+ (+ let+ doModify state = + newCluster `seq`+ state+ { rsClusterState = newCluster+ }+ in+ doModify+ )+ do {- Dispatch outputs. -}+ self <- gets rsSelf+ let+ outputs :: Map (EventId Peer) (Output e)+ outputs = urOutputs ur++ byRemotePeer :: Map Peer (Map (EventId Peer) (Output e))+ byRemotePeer =+ Map.unionsWith+ (<>)+ [ Map.singleton peer (Map.singleton eid o)+ | (eid, o) <- Map.toList outputs+ , Just peer <- [EF.source eid]+ ]+ respondToWaiting outputs+ sequence_+ [ do+ logDebug $ "Sending remote outputs: " <> showt byRemotePeer+ sendPeer (PMOutputs outputsForPeer) peer+ | (peer, outputsForPeer) <- Map.toList byRemotePeer+ , peer /= self+ ]+ pure v+++{- | Wait on a consistent response for the given state id. -}+waitOn :: (Monad m)+ => EventId Peer+ -> Responder (Output e)+ -> StateT (RuntimeState e) m ()+waitOn sid responder =+ modify' (\state@RuntimeState {rsWaiting} -> state {+ rsWaiting = Map.insert sid responder rsWaiting+ })+++{- | Propagates cluster information if necessary. -}+propagate+ :: ( EventConstraints e+ , MonadCatch m+ , MonadLoggerIO m+ , MonadState (RuntimeState e) m+ )+ => m ()+propagate = do+ (self, cluster) <- gets (rsSelf &&& rsClusterState)+ let+ targets = Set.delete self $+ EF.allParticipants cluster++ liftIO (shuffleM (Set.toList targets)) >>= \case+ [] -> return ()+ target:_ ->+ -- case events target cluster of+ -- Nothing -> do+ -- logInfo+ -- $ "Sending full merge because the target's join event "+ -- <> "has not reaached the infimum"+ -- sendPeer (PMFullMerge cluster) target+ -- Just diff ->+ -- sendPeer (PMMerge diff) target+ sendPeer (PMFullMerge cluster) target+ disconnectObsolete+ where+ {- |+ Shut down connections to peers that are no longer participating+ in the cluster.+ -}+ disconnectObsolete+ :: ( MonadState (RuntimeState e) m+ , MonadLogger m+ )+ => m ()+ disconnectObsolete = do+ (cluster, conns) <- gets (rsClusterState &&& rsConnections)+ let obsolete = Map.keysSet conns \\ EF.allParticipants cluster+ unless (Set.null obsolete) $+ logInfo $ "Disconnecting obsolete: " <> showt obsolete+ mapM_ disconnect obsolete+++{- |+ Respond to event applications that are waiting on a consistent result,+ if such a result is available.+-}+respondToWaiting+ :: forall m e.+ ( MonadLoggerIO m+ , MonadState (RuntimeState e) m+ , Show (Output e)+ )+ => Map (EventId Peer) (Output e)+ -> m ()+respondToWaiting available = do+ rs <- get+ logDebug+ $ "Responding to: " <> showt (available, Map.keysSet (rsWaiting rs))+ mapM_ respondToOne (Map.toList available)+ where+ respondToOne+ :: (EventId Peer, Output e)+ -> m ()+ respondToOne (sid, output) = do+ state@RuntimeState {rsWaiting} <- get+ case Map.lookup sid rsWaiting of+ Nothing -> return ()+ Just responder -> do+ respond responder output+ put state {rsWaiting = Map.delete sid rsWaiting}+++{- | This defines the various ways a node can be spun up. -}+data StartupMode e+ {- | Indicates that we should bootstrap a new cluster at startup. -}+ = NewCluster+ Peer {- ^ The peer being launched. -}+ ClusterName {- ^ The name of the cluster being launched. -}++ {- | Indicates that the node should try to join an existing cluster. -}+ | JoinCluster+ Peer {- ^ The peer being launched. -}+ ClusterName {- ^ The name of the cluster we are trying to join. -}+ Peer {- ^ The existing peer we are attempting to join with. -}++ {- | Resume operation given the previously saved state. -}+ | Recover+ Peer {- ^ The Peer being recovered. -}+ (EventFold ClusterName Peer e)+ {- ^ The last acknowledged state we had before we crashed. -}+deriving stock instance+ ( Show e+ , Show (Output e)+ , Show (State e)+ )+ =>+ Show (StartupMode e)+++{- | Initialize the runtime state. -}+makeRuntimeState :: (EventConstraints e, MonadLoggerIO m)+ => (Peer -> EventFold ClusterName Peer e -> IO ())+ {- ^ Callback when the cluster-wide powerstate changes. -}+ -> StartupMode e+ -> m (RuntimeState e)++makeRuntimeState+ notify+ (NewCluster self clusterId)+ = do+ logInfo "Starting a new cluster."+ {- Build a brand new node state, for the first node in a cluster. -}+ makeRuntimeState+ notify+ (Recover self (EF.new clusterId self))++makeRuntimeState+ notify+ (JoinCluster self _clusterName targetPeer)+ = do+ {- Join a cluster an existing cluster. -}+ logInfo $ "Trying to join an existing cluster on " <> showt addr+ JoinOk cluster <-+ requestJoin+ . JoinRequest+ $ self+ logInfo $ "Join response with cluster: " <> showt cluster+ makeRuntimeState notify (Recover self cluster)+ where+ requestJoin :: (EventConstraints e, MonadLoggerIO m)+ => JoinRequest+ -> m (JoinResponse e)+ requestJoin joinMsg = ($ joinMsg) =<< connectServer addr Nothing++ addr :: AddressDescription+ addr =+ AddressDescription+ (+ unPeer targetPeer+ <> ":" <> showt joinMessagePort+ )++makeRuntimeState+ notify+ (Recover self clusterState)+ = do+ rsNextId <- newSequence+ return+ RuntimeState+ { rsSelf = self+ , rsClusterState = clusterState+ , rsConnections = mempty+ , rsWaiting = mempty+ , rsCalls = mempty+ , rsBroadcalls = mempty+ , rsNextId+ , rsNotify = notify self+ , rsJoins = mempty+ }+++{- | This is the type of a join request message. -}+newtype JoinRequest = JoinRequest Peer+ deriving stock (Generic, Show)+instance Binary JoinRequest+++{- |+ Initialize a new sequence of messageIds. It would be perfectly fine to ensure+ unique message ids by generating a unique UUID for each one, but generating+ UUIDs is not free, and we are probably going to be generating a lot of these.+-}+newSequence :: (MonadIO m) => m MessageId+newSequence = do+ sid <- getUUID+ pure (M sid 0)+ where+ {- | A utility function that makes a UUID, no matter what. -}+ getUUID :: (MonadIO m) => m UUID+ getUUID = liftIO nextUUID >>= maybe (wait >> getUUID) return+ where+ wait = liftIO (threadDelay oneMillisecond)+ oneMillisecond = 1000+++{- |+ Generate the next message id in the sequence. We would normally use+ `succ` for this kind of thing, but making `MessageId` an instance of+ `Enum` really isn't appropriate.+-}+nextMessageId :: MessageId -> MessageId+nextMessageId (M sequenceId ord) = M sequenceId (succ ord)+++{- | Obtain the 'ClusterName'. -}+getClusterName :: Runtime e -> ClusterName+getClusterName = rClusterId+++{- | The join message port -}+joinMessagePort :: PortNumber+joinMessagePort = 5289+++{- | Like 'Fork.respond', but returns '()'. -}+respond :: (MonadIO m) => Responder a -> a -> m ()+respond responder = void . Fork.respond responder+++{- |+ Retrieve some basic stats that can be used to intuit the health of+ the cluster.+-}+getStats :: (MonadIO m) => Runtime e -> m Stats+getStats runtime = do+ now <- liftIO getTime+ stats <- liftIO $ readIORef (rStats runtime)+ pure+ Stats+ { timeWithoutProgress = diffTimeSpec now <$> stats+ }++
+ test/Test/OM/Legion.hs view
@@ -0,0 +1,136 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}++module Test.OM.Legion (+ S(..),+ Request(..),+ Response(..),+ Op(..),+ k8sConfig,+ Client,+ send,+ makeClient,+ makeClientLocal,+) where+++import Control.Monad.IO.Class (MonadIO(liftIO))+import Control.Monad.Logger.CallStack (MonadLoggerIO, logInfo)+import Data.Aeson (KeyValue((.=)), ToJSON, object)+import Data.Binary (Binary)+import Data.CRDT.EventFold (Event(Output, State, apply),+ EventResult(Pure), EventFold)+import Data.Default.Class (Default(def))+import Data.Text (Text)+import GHC.Generics (Generic)+import Language.Haskell.TH.Syntax (addDependentFile)+import Numeric.Natural (Natural)+import OM.Kubernetes (Namespace(Namespace))+import OM.Legion (ClusterName, Peer, Stats)+import OM.Show (showt)+import OM.Socket (AddressDescription(AddressDescription), connectServer)+import qualified Text.Mustache as Mustache (Template, substitute)+import qualified Text.Mustache.Compile as Mustache (embedSingleTemplate)+++newtype S = S Int+ deriving newtype (Enum, Binary, Show, ToJSON, Eq, Num)+instance Default S where+ def = 0+++data Request+ = OpReq Op+ | ReadState+ | ReadStats+ deriving stock (Generic, Show, Read)+ deriving anyclass (Binary)+++data Response+ = OpR (Output Op)+ | ReadStateR (EventFold ClusterName Peer Op)+ | ReadStatsR Stats+ deriving stock (Generic, Show, Eq)+ deriving anyclass (Binary, ToJSON)+++data Op+ = Inc+ | Dec+ | Get+ deriving stock (Eq, Generic, Show, Read)+ deriving anyclass (Binary, ToJSON)+instance Event Peer Op where+ type Output Op = (S, S)+ type State Op = S++ apply op val =+ let+ newVal :: S+ newVal =+ case op of+ Inc -> succ val+ Dec -> pred val+ Get -> val+ in+ Pure (val, newVal) newVal+++k8sConfig+ :: String {- ^ cluster name. -}+ -> String {- ^ Docker image name. -}+ -> Text {- ^ k8s config. -}+k8sConfig name image =+ Mustache.substitute+ template+ (object ["name" .= name, "image" .= image])+ where+ template :: Mustache.Template+ template =+ $(do+ addDependentFile "test/k8s/k8s.mustache"+ Mustache.embedSingleTemplate "test/k8s/k8s.mustache"+ )+++newtype Client = Client+ { unClient :: Request -> IO Response+ }+++send :: (MonadIO m) => Client -> Request -> m Response+send client = liftIO . unClient client+++makeClient :: (MonadLoggerIO m) => Namespace -> Natural -> m Client+makeClient (Namespace namespace) ord = do+ let+ targetPeer =+ namespace <> "-" <> showt ord+ <> "." <> namespace+ <> "." <> namespace+ <> ".svc.cluster.local"+ logInfo $ "Target: " <> showt targetPeer+ Client <$>+ connectServer+ (AddressDescription (targetPeer <> ":9999"))+ Nothing+++makeClientLocal :: (MonadLoggerIO m) => m Client+makeClientLocal= do+ Client <$>+ connectServer+ (AddressDescription ("localhost:9999"))+ Nothing++
+ test/k8s/k8s.mustache view
@@ -0,0 +1,111 @@+apiVersion: v1+kind: Namespace+metadata:+ annotations:+ description: |+ The namespace under which {{name}} stuff runs.+ name: {{name}}+---+apiVersion: v1+kind: ServiceAccount+metadata:+ annotations:+ description: |+ This defines the service account as which legion pods run.+ name: {{name}}+ namespace: {{name}}+---+apiVersion: rbac.authorization.k8s.io/v1+kind: Role+metadata:+ annotations:+ description: |+ A role that has pod management permissions, allowing the {{name}}+ service to self-manage itself.+ namespace: {{name}}+ name: peer-discovery+rules:+- apiGroups: [""]+ resources: ["pods"]+ verbs: ["list"]+---+apiVersion: rbac.authorization.k8s.io/v1+kind: RoleBinding+metadata:+ annotations:+ description: |+ Bind legion pods to peer-discovery roles.+ name: {{name}}+ namespace: {{name}}+subjects:+- kind: ServiceAccount+ name: {{name}}+roleRef:+ kind: Role+ name: peer-discovery+ apiGroup: rbac.authorization.k8s.io+---+apiVersion: v1+kind: Service+metadata:+ annotations:+ description: |+ This is a headless service required by the {{name}} StatefulSet+ that causes pods to be exposed via DNS records.+ name: {{name}}+ namespace: {{name}}+spec:+ selector:+ app: {{name}}+ type: ClusterIP+ clusterIP: None+---+apiVersion: apps/v1+kind: StatefulSet+metadata:+ annotations:+ description: |+ {{name}} deployment.+ name: {{name}}+ namespace: {{name}}+ labels:+ app: {{name}}+spec:+ replicas: 1+ selector:+ matchLabels:+ app: {{name}}+ updateStrategy:+ type: RollingUpdate+ serviceName: {{name}}+ template:+ metadata:+ annotations:+ description: |+ Pod of the {{name}} service.+ name: {{name}}+ namespace: {{name}}+ labels:+ app: {{name}}+ spec:+ containers:+ - name: {{name}}-container+ image: {{image}}+ imagePullPolicy: Always+ readinessProbe:+ tcpSocket:+ port: 5289+ livenessProbe:+ tcpSocket:+ port: 5289+ env:+ - name: NAMESPACE+ valueFrom:+ fieldRef:+ fieldPath: metadata.namespace+ - name: LOG_LEVEL+ value: DEBUG+ - name: DOMAIN+ value: {{name}}.{{name}}.svc.cluster.local+ serviceAccountName: {{name}}+ terminationGracePeriodSeconds: 1200
+ test/om-legion-test-driver.hs view
@@ -0,0 +1,422 @@+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}++{- | Description: Run an om-legion tests. -}+module Main (+ main,+) where+++import Control.Concurrent (threadDelay)+import Control.Lens (Identity, view)+import Control.Monad (unless)+import Control.Monad.Logger.CallStack (LogLevel(LevelDebug, LevelError,+ LevelInfo, LevelOther, LevelWarn), LoggingT(runLoggingT), MonadLogger,+ MonadLoggerIO, logDebug, logError, logInfo)+import Data.Aeson (Value, eitherDecode)+import Data.Aeson.Lens (AsPrimitive(_String), AsValue(_Array), key)+import Data.Foldable (for_)+import Data.String (IsString(fromString))+import OM.Fork (Race, runRace)+import OM.Logging (standardLogging)+import OM.Show (showt)+import System.Console.GetOpt (ArgDescr(OptArg, ReqArg), ArgOrder(Permute),+ OptDescr(Option), getOpt, usageInfo)+import System.Environment (getArgs)+import System.Exit (ExitCode(ExitFailure, ExitSuccess))+import Test.Hspec (Spec, describe, it, shouldReturn)+import Test.Hspec.Runner (defaultConfig, evaluateSummary, runSpec)+import Test.OM.Legion (Op(Inc), Request(OpReq, ReadState, ReadStats),+ k8sConfig)+import UnliftIO (MonadIO(liftIO), MonadUnliftIO, SomeException, finally,+ try)+import UnliftIO.Process (proc, readProcess, readProcessWithExitCode,+ showCommandForUser, waitForProcess, withCreateProcess)+import qualified Data.Text as Text+import qualified Data.Vector as Vector+++main :: IO ()+main = do+ config <- getConfig+ let+ logging = standardLogging (logLevel config)+ it_ :: String -> (Race => LoggingT IO ()) -> Spec+ it_ doesWhat action =+ it doesWhat+ . flip shouldReturn ()+ . flip runLoggingT logging+ $ runRace+ $ flip finally (terminateCluster config)+ $ action+ runTests $+ describe "om-legion" $ do+ it_ "scales up" $ do+ launchCluster config+ runOps config 0 [OpReq Inc, OpReq Inc, OpReq Inc]+ resizeCluster (namespace config) 3+ runOps config 1 [OpReq Inc, OpReq Inc, OpReq Inc]+ runOps config 2 [OpReq Inc, OpReq Inc, OpReq Inc]+ for_ [0..2] $ \n ->+ runOps config n [ ReadState, ReadStats ]+ checkStable config 3 9++ it_ "scales down" $ do+ launchCluster config+ resizeCluster (namespace config) 3+ for_ [0..2] $ \n ->+ runOps config n [OpReq Inc, OpReq Inc, OpReq Inc]+ resizeCluster (namespace config) 1+ runOps config 0 [ ReadState, ReadStats ]+ checkStable config 1 9++ it_ "recovers from pod deletion" $ do+ launchCluster config+ resizeCluster (namespace config) 3+ runOps config 2 [OpReq Inc, OpReq Inc, OpReq Inc, ReadStats]+ runOps config 2 [ReadState]+ cmd "kubectl" ["-n", namespace config, "delete", "pod", "test-2"]+ runOps config 0 [OpReq Inc, OpReq Inc, OpReq Inc, ReadStats]+ waitForStable config 3 50 6+++runOps+ :: ( MonadFail m+ , MonadLogger m+ , MonadUnliftIO m+ )+ => Config Identity+ -> Int+ -> [Request]+ -> m ()+runOps config n ops = do+ cmd+ "kubectl"+ (+ [ "-n"+ , (namespace config)+ , "run"+ , "test-driver"+ , "--image=us.gcr.io/friendlee/om-legion-test:latest"+ , "-i"+ , "--rm"+ , "--restart=Never"+ , "--env=NAMESPACE=" <> namespace config+ , "--env=LOG_LEVEL=DEBUG"+ , "--serviceaccount=" <> namespace config+ , "--wait=true"+ , "--"+ , "/bin/om-legion-test-run"+ , show n+ ] <> (show <$> ops)+ )+++resizeCluster+ :: ( MonadFail m+ , MonadLogger m+ , MonadUnliftIO m+ )+ => String+ -> Int+ -> m ()+resizeCluster namespace n = do+ cmd+ "kubectl"+ [+ "-n",+ namespace,+ "patch",+ "statefulset",+ namespace,+ "-p",+ "{\"spec\": {\"replicas\": " <> showt n <> "}}"+ ]+ for_ [0..n-1] (waitForReady namespace)+++waitForReady+ :: ( MonadFail m+ , MonadLogger m+ , MonadUnliftIO m+ )+ => String+ -> Int+ -> m ()+waitForReady namespace n = do+ cmd "kubectl" ["-n", namespace, "get", "pods"]+ json <-+ readProcess+ "kubectl"+ [ "-n"+ , namespace+ , "get"+ , "pod"+ , namespace <> "-" <> show n+ , "-o"+ , "json"+ ]+ ""+ case eitherDecode (fromString json) of+ Left err -> fail $ "Can't check ready: " <> show (n, err)+ Right val ->+ case isReady val of+ True -> pure ()+ False -> do+ logInfo $ "Not yet ready: " <> showt n+ liftIO $ threadDelay 1_000_000+ waitForReady namespace n+ where+ isReady :: Value -> Bool+ isReady =+ not+ . Vector.null+ . Vector.filter+ (\condition ->+ view (key "status" . _String) condition == "True"+ && view (key "type" . _String) condition == "Ready"+ )+ . view (key "status" . key "conditions" . _Array)+++getConfig+ :: ( MonadFail m+ , MonadIO m+ )+ => m (Config Identity)+getConfig = do+ let logging = standardLogging LevelDebug+ flip runLoggingT logging $ do+ args <- liftIO $ getArgs+ case getOpt Permute options args of+ (o, [], []) ->+ let+ config = foldl (.) id o (Config Nothing Nothing LevelInfo)+ in+ case validate config of+ Nothing -> do+ logError ("Invalid config: " <> showt config)+ usage+ Just cfg -> pure cfg+ (_, _, err) -> do+ logError ("opt-result: " <> showt err)+ usage+ where+ usage :: (MonadLogger m, MonadFail m) => m a+ usage = do+ logError $ fromString (usageInfo "Usage:" options)+ fail "Invalid usage."++ options :: [OptDescr (Config Maybe -> Config Maybe)]+ options =+ [ Option+ "n"+ ["namespace"]+ (ReqArg (\val cfg -> cfg {namespace = Just val}) "<namespace>")+ "The k8s namespace to deploy to."+ , Option+ "i"+ ["image"]+ (ReqArg (\val cfg -> cfg {image = Just val}) "<docker-image>")+ "The docker image to deploy."+ , Option+ ""+ ["log-level"]+ (+ OptArg+ (\mLevel cfg ->+ case mLevel of+ Nothing -> cfg {logLevel = LevelInfo}+ Just "DEBUG" -> cfg {logLevel = LevelDebug}+ Just "INFO" -> cfg {logLevel = LevelInfo}+ Just "WARN" -> cfg {logLevel = LevelWarn}+ Just "ERROR" -> cfg {logLevel = LevelError}+ Just other -> cfg {logLevel = LevelOther (fromString other)}+ )+ "<log-level>"+ )+ "The logging level to choose. [DEBUG, INFO, WARN, ERROR]"+ ]+++runTests :: Spec -> IO ()+runTests spec =+ runSpec spec defaultConfig >>= evaluateSummary+++terminateCluster+ :: forall m.+ ( MonadFail m+ , MonadLoggerIO m+ , MonadUnliftIO m+ )+ => Config Identity+ -> m ()+terminateCluster Config {namespace} =+ finally+ (do+ logDumpedState 0+ logDumpedState 1+ logDumpedState 2+ ) (+ cmd+ "kubectl"+ [ "delete"+ , "namespace"+ , namespace+ ]+ )+ where+ logDumpedState :: Int -> m ()+ logDumpedState n = do+ (_, out, _) <-+ readProcessWithExitCode+ "bash"+ [+ "-c",+ "kubectl exec -i -n "+ <> show namespace+ <> " "+ <> show namespace+ <> "-"+ <> show n+ <> " bash -- -c 'cat log | gzip' | gzip -d"+ ]+ ""+ logInfo $ "logs " <> showt n <> ": " <> showt out+++launchCluster+ :: ( MonadFail m+ , MonadLogger m+ , MonadUnliftIO m+ )+ => Config Identity+ -> m ()+launchCluster Config { namespace, image } = do+ readProcessWithExitCode+ "kubectl"+ [+ "apply",+ "-f",+ "-"+ ]+ (Text.unpack (k8sConfig namespace image))+ >>= \case+ (ExitSuccess, out, err) -> do+ logInfo (fromString out)+ unless (null err) $ logError (fromString err)+ err -> fail ("Got apply error: " <> show err)+ waitForReady namespace 0+++cmd+ :: ( MonadFail m+ , MonadLogger m+ , MonadUnliftIO m+ )+ => String+ -> [String]+ -> m ()+cmd prog args = do+ withCreateProcess+ (proc prog args)+ (\_ _ _ handle ->+ waitForProcess handle >>= \case+ result@(ExitFailure _) ->+ fail+ $ "Failed:\n"+ <> showCommandForUser prog args+ <> "\nwith: " <> show result+ result@ExitSuccess ->+ logDebug ("got: " <> showt result)+ )+++{- | Check that n nodes are reporting stable stats. -}+checkStable+ :: ( MonadFail m+ , MonadLogger m+ , MonadUnliftIO m+ )+ => Config Identity+ -> Int+ -> Int+ -> m ()+checkStable config n v = do+ cmd+ "kubectl"+ [ "-n"+ , (namespace config)+ , "run"+ , "test-driver"+ , "--image=us.gcr.io/friendlee/om-legion-test:latest"+ , "-i"+ , "--rm"+ , "--restart=Never"+ , "--env=NAMESPACE=" <> namespace config+ , "--env=LOG_LEVEL=DEBUG"+ , "--serviceaccount=" <> namespace config+ , "--wait=true"+ , "--"+ , "/bin/om-legion-test-stable"+ , show n+ , show v+ ]+++waitForStable+ :: ( MonadFail m+ , MonadLogger m+ , MonadUnliftIO m+ )+ => Config Identity+ -> Int+ -> Int+ -> Int+ -> m ()+waitForStable config n v tries =+ case tries of+ 0 -> fail "The cluster doesn't seem like it's going to stabilize."+ _ -> do+ runOps config 0 [ReadState, ReadStats]+ try (checkStable config n v) >>= \case+ Left (_err :: SomeException) -> do+ logInfo $ "Not yet stable."+ waitForStable config n v (tries - 1)+ Right () -> pure ()+++data Config f = Config+ { namespace :: F f String+ , image :: F f String+ , logLevel :: LogLevel+ }+deriving stock instance Show (Config Maybe)+++type family F a b where+ F Identity a = a+ F f a = f a+++validate :: Config Maybe -> Maybe (Config Identity)+validate (Config namespace image logLevel) =+ Config+ <$> namespace+ <*> image+ <*> pure logLevel++
+ test/om-legion-test-inc.hs view
@@ -0,0 +1,34 @@+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE OverloadedStrings #-}++{- |+ Description: a program that runs some operations against a legion+ test node.+-}+module Main (main) where++import Control.Concurrent.Async (mapConcurrently_)+import Control.Monad.Logger.CallStack (LoggingT(runLoggingT), logInfo)+import Data.String (IsString(fromString))+import OM.Logging (parseLevel, standardLogging)+import OM.Show (showj)+import System.Environment (getArgs, getEnv)+import Test.OM.Legion (Op(Inc), Request(OpReq), makeClientLocal, send)+++main :: IO ()+main = do+ level <- parseLevel . fromString <$> getEnv "LOG_LEVEL" + [numIncrements, threads] <- getArgs+ let logging = standardLogging level+ mapConcurrently_ id . replicate (read threads) $ do+ flip runLoggingT logging $ do+ client <- makeClientLocal+ logInfo "Sending sequence."+ sequence_ . replicate (read numIncrements) $ do+ logInfo "Sending req"+ send client (OpReq Inc) >>= logInfo . ("Recieved: " <>) . showj+ logInfo "Sequence sent."+++
+ test/om-legion-test-node.hs view
@@ -0,0 +1,154 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}++module Main (main) where+++import Conduit (MonadTrans(lift), (.|), awaitForever, runConduit)+import Control.Concurrent (myThreadId, throwTo)+import Control.Lens (view)+import Control.Monad (void)+import Control.Monad.IO.Class (MonadIO(liftIO))+import Control.Monad.Logger.CallStack (LoggingT(runLoggingT), MonadLogger,+ logDebug)+import Data.Aeson.Lens (AsPrimitive(_String), AsValue(_Array), key)+import Data.String (IsString(fromString))+import Network.HostName (HostName, getHostName)+import OM.Fork (race, runRace, wait)+import OM.Kubernetes (Pod(unPod), newK8s, queryPods)+import OM.Legion (Peer(Peer), StartupMode(JoinCluster, NewCluster),+ Runtime, applyConsistent, eject, forkLegionary, getSelf, getStats,+ readState)+import OM.Logging (fdLogging, parseLevel, stdoutLogging, teeLogging,+ withStandardFormat)+import OM.Show (showt)+import OM.Socket (openServer)+import System.Environment (getEnv)+import System.Exit (ExitCode(ExitFailure))+import System.IO (IOMode(WriteMode), withFile)+import Test.OM.Legion (Request(OpReq, ReadState, ReadStats), Response(OpR,+ ReadStateR, ReadStatsR), Op)+import qualified Data.Text as Text+import qualified Data.Vector as Vector+import qualified System.Posix.Signals as Signal+++main :: IO ()+main = do+ withFile "log" WriteMode $ \logFD ->+ runRace $ do+ logLevel <- (parseLevel . fromString) <$> getEnv "LOG_LEVEL"+ let+ logging =+ withStandardFormat+ logLevel+ (teeLogging stdoutLogging (fdLogging logFD))+ flip runLoggingT logging $ do+ startupMode <- getStartupMode+ runtime <-+ forkLegionary+ pure+ (\_ -> pure ())+ (\_ _ -> pure ())+ 500_000+ startupMode++ liftIO . void $ do+ mainThread <- myThreadId+ Signal.installHandler+ Signal.sigTERM+ (Signal.Catch (do+ eject runtime (getSelf runtime)+ throwTo+ mainThread+ (ExitFailure (fromIntegral Signal.sigTERM + 128))+ ))+ Nothing++ race "service endpoint" $+ runConduit+ (+ pure ()+ .| openServer "0.0.0.0:9999" Nothing+ .| awaitForever (lift . void . handle runtime)+ )+ wait+ where+ handle :: (MonadIO m) => Runtime Op -> (Request, Response -> m b) -> m b+ handle runtime (req, respond) =+ case req of+ OpReq op -> respond . OpR =<< applyConsistent runtime op+ ReadState -> respond . ReadStateR =<< readState runtime+ ReadStats -> respond . ReadStatsR =<< getStats runtime++ getStartupMode+ :: ( MonadIO m+ , MonadLogger m+ )+ => m (StartupMode Op)+ getStartupMode = do+ domain <- liftIO $ getEnv "DOMAIN"+ hostName <- liftIO $ getHostName+ let self = Peer (fromString (hostName <> "." <> domain))+ nsString <- liftIO $ getEnv "NAMESPACE"+ let+ clusterName :: (IsString a) => a+ clusterName = fromString nsString+ getExistingNodes clusterName >>= \case+ [] -> pure $ NewCluster self clusterName+ node:_ ->+ let+ joinPeer :: Peer+ joinPeer = Peer . fromString $ node <> "." <> domain+ in+ pure $+ JoinCluster+ self+ clusterName+ joinPeer+ where+ getExistingNodes+ :: ( MonadIO m+ , MonadLogger m+ )+ => (forall ns. IsString ns => ns)+ -> m [HostName]+ getExistingNodes namespace = do+ k8s <- newK8s+ logDebug "Querying pods"+ pods <- queryPods k8s namespace [("app", namespace)]+ logDebug $ "Got pods: " <> showt pods+ let+ readyPods :: [Pod]+ readyPods = filter isReady pods+ logDebug $ "Ready pods: " <> showt readyPods+ pure $ getName <$> readyPods+ where+ isReady :: Pod -> Bool+ isReady =+ not+ . Vector.null+ . Vector.filter+ (\condition ->+ view (key "status" . _String) condition == "True"+ && view (key "type" . _String) condition == "Ready"+ )+ . view (key "status" . key "conditions" . _Array)+ . unPod++ getName :: Pod -> HostName+ getName =+ Text.unpack+ . view (key "metadata" . key "name" . _String)+ . unPod++
+ test/om-legion-test-profile.hs view
@@ -0,0 +1,197 @@+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}++{- | Description: Run an om-legion tests. -}+module Main (+ main,+) where+++import Control.Lens (Identity)+import Control.Monad.Logger.CallStack (LogLevel(LevelDebug, LevelError,+ LevelInfo, LevelOther, LevelWarn), LoggingT(runLoggingT), MonadLogger,+ logDebug, logError)+import Data.String (IsString(fromString))+import OM.Logging (standardLogging)+import OM.Show (showt)+import Safe (readMay)+import System.Console.GetOpt (ArgDescr(OptArg, ReqArg), ArgOrder(Permute),+ OptDescr(Option), getOpt, usageInfo)+import System.Environment (getArgs)+import System.Exit (ExitCode(ExitFailure, ExitSuccess))+import UnliftIO (MonadIO(liftIO), MonadUnliftIO, forConcurrently_)+import UnliftIO.Process (proc, showCommandForUser, waitForProcess,+ withCreateProcess)+++main :: IO ()+main = do+ + Config+ { namespace+ , logLevel+ , numNodes+ , threads+ , numOps+ }+ <- getConfig+ flip runLoggingT (standardLogging logLevel) $+ forConcurrently_ [0..numNodes - 1] $ \node ->+ runIncs namespace node threads numOps+++runIncs+ :: ( MonadFail m+ , MonadLogger m+ , MonadUnliftIO m+ )+ => String+ -> Int+ -> Int+ -> Int+ -> m ()+runIncs namespace node threads numOps = do+ cmd+ "kubectl"+ [ "-n"+ , namespace+ , "exec"+ , "-i"+ , namespace <> "-" <> show node+ , "--"+ , "/bin/om-legion-test-inc"+ , show numOps+ , show threads+ ]+++getConfig+ :: ( MonadFail m+ , MonadIO m+ )+ => m (Config Identity)+getConfig = do+ let logging = standardLogging LevelDebug+ flip runLoggingT logging $ do+ args <- liftIO $ getArgs+ case getOpt Permute options args of+ (o, [], []) ->+ let+ config = foldl (.) id o (Config Nothing LevelInfo Nothing Nothing Nothing)+ in+ case validate config of+ Nothing -> do+ logError ("Invalid config: " <> showt config)+ usage+ Just cfg -> pure cfg+ (_, _, err) -> do+ logError ("opt-result: " <> showt err)+ usage+ where+ usage :: (MonadLogger m, MonadFail m) => m a+ usage = do+ logError $ fromString (usageInfo "Usage:" options)+ fail "Invalid usage."++ options :: [OptDescr (Config Maybe -> Config Maybe)]+ options =+ [ Option+ "n"+ ["namespace"]+ (ReqArg (\val cfg -> cfg {namespace = Just val}) "<namespace>")+ "The k8s namespace to deploy to."+ , Option+ ""+ ["num-nodes"]+ (ReqArg (\val cfg -> cfg {numNodes = readMay val}) "<num-nodes>")+ "The number of nodes in the cluster."+ , Option+ ""+ ["num-ops"]+ (ReqArg (\val cfg -> cfg {numOps = readMay val}) "<num-ops>")+ "The number of increments to perform on each node, in each thread."+ , Option+ ""+ ["num-threads"]+ (ReqArg (\val cfg -> cfg {threads = readMay val}) "<num-threads>")+ "The number of request threads each node should be hit with."+ , Option+ ""+ ["log-level"]+ (+ OptArg+ (\mLevel cfg ->+ case mLevel of+ Nothing -> cfg {logLevel = LevelInfo}+ Just "DEBUG" -> cfg {logLevel = LevelDebug}+ Just "INFO" -> cfg {logLevel = LevelInfo}+ Just "WARN" -> cfg {logLevel = LevelWarn}+ Just "ERROR" -> cfg {logLevel = LevelError}+ Just other -> cfg {logLevel = LevelOther (fromString other)}+ )+ "<log-level>"+ )+ "The logging level to choose. [DEBUG, INFO, WARN, ERROR]"+ ]+++cmd+ :: ( MonadFail m+ , MonadLogger m+ , MonadUnliftIO m+ )+ => String+ -> [String]+ -> m ()+cmd prog args = do+ withCreateProcess+ (proc prog args)+ (\_ _ _ handle ->+ waitForProcess handle >>= \case+ result@(ExitFailure _) ->+ fail+ $ "Failed:\n"+ <> showCommandForUser prog args+ <> "\nwith: " <> show result+ result@ExitSuccess ->+ logDebug ("got: " <> showt result)+ )+++data Config f = Config+ { namespace :: F f String+ , logLevel :: LogLevel+ , numNodes :: F f Int+ , numOps :: F f Int+ , threads :: F f Int+ }+deriving stock instance Show (Config Maybe)+++type family F a b where+ F Identity a = a+ F f a = f a+++validate :: Config Maybe -> Maybe (Config Identity)+validate (Config namespace logLevel numNodes numOps threads) =+ Config+ <$> namespace+ <*> pure logLevel+ <*> numNodes+ <*> numOps+ <*> threads++
+ test/om-legion-test-run.hs view
@@ -0,0 +1,42 @@+{-# LANGUAGE NumericUnderscores #-}++{- |+ Description: a program that runs some operations against a legion+ test node.+-}+module Main (main) where++import Control.Monad.Logger.CallStack (LoggingT(runLoggingT))+import Data.Maybe (maybeToList)+import Data.String (IsString(fromString))+import Numeric.Natural (Natural)+import OM.Logging (parseLevel, standardLogging)+import OM.Show (showj)+import Safe (readMay)+import System.Environment (getArgs, getEnv)+import Test.OM.Legion (Request, makeClient, send)+++main :: IO ()+main = do+ namespace <- fromString <$> getEnv "NAMESPACE"+ level <- parseLevel . fromString <$> getEnv "LOG_LEVEL" + (ord, reqs) <- parseArgs =<< getArgs+ let logging = standardLogging level+ client <- flip runLoggingT logging $ makeClient namespace ord+ sequence_+ [ send client req >>= putStrLn . showj+ | req <- reqs+ ]+ +parseArgs :: (MonadFail m) => [String] -> m (Natural, [Request])+parseArgs args =+ case args of+ [] -> fail "Needs some arguments."+ n:ops ->+ case readMay n of+ Nothing -> fail "Bad first argment."+ Just ord ->+ pure $ (ord, readMay <$> ops >>= maybeToList)++
+ test/om-legion-test-stable.hs view
@@ -0,0 +1,59 @@+{-# LANGUAGE NumericUnderscores #-}++{- | Description: "Scale up" test driver. -}+module Main (main) where++import Control.Monad (unless, when)+import Control.Monad.Logger.CallStack (LoggingT(runLoggingT))+import Data.CRDT.EventFold (infimumValue, projectedValue)+import Data.String (IsString(fromString))+import OM.Legion (Stats(Stats))+import OM.Logging (parseLevel, standardLogging)+import System.Environment (getArgs, getEnv)+import Test.OM.Legion (Request(ReadState, ReadStats), Response(ReadStateR,+ ReadStatsR), S(S), makeClient, send)+++main :: IO ()+main = do+ [n, val] <- fmap read <$> getArgs+ namespace <- fromString <$> getEnv "NAMESPACE"+ level <- parseLevel . fromString <$> getEnv "LOG_LEVEL" + let logging = standardLogging level+ flip runLoggingT logging $ do+ actual <-+ sequence+ [ do+ client <- makeClient namespace ord+ (,)+ <$> send client ReadStats+ <*> send client ReadState+ | ord <- [0..fromIntegral n - 1]+ ]++ let+ actualStats :: [Response]+ actualStats = fst <$> actual+ expectedStats :: [Response]+ expectedStats = replicate n (ReadStatsR (Stats mempty))+ state:states = snd <$> actual+ when (actualStats /= expectedStats) $+ fail+ $ "(actualStats /= expectedStats): "+ <> show (actualStats, expectedStats)+ unless (all (state ==) states) $+ fail+ $ "Not all states the same: "+ <> show actual++ let+ ReadStateR ef = state+ expectedVals = (S val, S val)+ actualVals = (projectedValue ef, infimumValue ef)++ unless (actualVals == expectedVals) $+ fail+ $ "(actualVals /= expectedVals): "+ <> show (actualVals, expectedVals)++