diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -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. + + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/om-legion.cabal b/om-legion.cabal new file mode 100644 --- /dev/null +++ b/om-legion.cabal @@ -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 + diff --git a/src/OM/Legion.hs b/src/OM/Legion.hs new file mode 100644 --- /dev/null +++ b/src/OM/Legion.hs @@ -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) + + diff --git a/src/OM/Legion/Conduit.hs b/src/OM/Legion/Conduit.hs new file mode 100644 --- /dev/null +++ b/src/OM/Legion/Conduit.hs @@ -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) + + diff --git a/src/OM/Legion/Connection.hs b/src/OM/Legion/Connection.hs new file mode 100644 --- /dev/null +++ b/src/OM/Legion/Connection.hs @@ -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 + + diff --git a/src/OM/Legion/MsgChan.hs b/src/OM/Legion/MsgChan.hs new file mode 100644 --- /dev/null +++ b/src/OM/Legion/MsgChan.hs @@ -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 + + diff --git a/src/OM/Legion/Runtime.hs b/src/OM/Legion/Runtime.hs new file mode 100644 --- /dev/null +++ b/src/OM/Legion/Runtime.hs @@ -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 + } + + diff --git a/test/Test/OM/Legion.hs b/test/Test/OM/Legion.hs new file mode 100644 --- /dev/null +++ b/test/Test/OM/Legion.hs @@ -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 + + diff --git a/test/k8s/k8s.mustache b/test/k8s/k8s.mustache new file mode 100644 --- /dev/null +++ b/test/k8s/k8s.mustache @@ -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 diff --git a/test/om-legion-test-driver.hs b/test/om-legion-test-driver.hs new file mode 100644 --- /dev/null +++ b/test/om-legion-test-driver.hs @@ -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}) "") + "The k8s namespace to deploy to." + , Option + "i" + ["image"] + (ReqArg (\val cfg -> cfg {image = Just val}) "") + "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)} + ) + "" + ) + "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 + + diff --git a/test/om-legion-test-inc.hs b/test/om-legion-test-inc.hs new file mode 100644 --- /dev/null +++ b/test/om-legion-test-inc.hs @@ -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." + + + diff --git a/test/om-legion-test-node.hs b/test/om-legion-test-node.hs new file mode 100644 --- /dev/null +++ b/test/om-legion-test-node.hs @@ -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 + + diff --git a/test/om-legion-test-profile.hs b/test/om-legion-test-profile.hs new file mode 100644 --- /dev/null +++ b/test/om-legion-test-profile.hs @@ -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}) "") + "The k8s namespace to deploy to." + , Option + "" + ["num-nodes"] + (ReqArg (\val cfg -> cfg {numNodes = readMay val}) "") + "The number of nodes in the cluster." + , Option + "" + ["num-ops"] + (ReqArg (\val cfg -> cfg {numOps = readMay val}) "") + "The number of increments to perform on each node, in each thread." + , Option + "" + ["num-threads"] + (ReqArg (\val cfg -> cfg {threads = readMay val}) "") + "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)} + ) + "" + ) + "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 + + diff --git a/test/om-legion-test-run.hs b/test/om-legion-test-run.hs new file mode 100644 --- /dev/null +++ b/test/om-legion-test-run.hs @@ -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) + + diff --git a/test/om-legion-test-stable.hs b/test/om-legion-test-stable.hs new file mode 100644 --- /dev/null +++ b/test/om-legion-test-stable.hs @@ -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) + +