diff --git a/om-legion.cabal b/om-legion.cabal --- a/om-legion.cabal +++ b/om-legion.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: om-legion -version: 6.9.0.4 +version: 6.9.0.5 synopsis: Legion Framework. description: Framework for managing shared, replicated state across a number of homogeneous nodes. @@ -15,7 +15,6 @@ extra-source-files: LICENSE README.md - test/k8s/k8s.mustache common dependencies build-depends: @@ -25,7 +24,6 @@ , binary >= 0.8.9.1 && < 0.9 , bytestring >= 0.12.0.2 && < 0.13 , clock >= 0.8.4 && < 0.9 - , conduit >= 1.3.5 && < 1.4 , containers >= 0.6.8 && < 0.7 , crdt-event-fold >= 1.8.0.2 && < 1.9 , data-default-class >= 0.1.2.0 && < 0.2 @@ -36,11 +34,12 @@ , om-fork >= 0.7.1.9 && < 0.8 , om-logging >= 1.1.0.8 && < 1.2 , om-show >= 0.1.2.9 && < 0.2 - , om-socket >= 0.11.0.5 && < 0.12 + , om-socket >= 1.0.0.0 && < 1.1 , om-time >= 0.3.0.4 && < 0.4 , random-shuffle >= 0.0.4 && < 0.1 , safe-exceptions >= 0.1.7.4 && < 0.2 , stm >= 2.5.2.1 && < 2.6 + , streaming >= 0.2.4.0 && < 0.3 , text >= 2.1 && < 2.2 , time >= 1.12.2 && < 1.13 , transformers >= 0.6.1.0 && < 0.7 @@ -54,84 +53,18 @@ -Wmissing-export-lists -Wmissing-import-lists -Wredundant-constraints + -Wunused-packages + 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.11.7 && < 2.12 - , lens >= 5.2.3 && < 5.3 - , lens-aeson >= 1.2.3 && < 1.3 - , mustache >= 2.4.2 && < 2.5 - , om-kubernetes >= 2.3.1.8 && < 2.4 - , safe >= 0.3.20 && < 0.4 - , template-haskell >= 2.21.0.0 && < 2.22 - , unix >= 2.8.5.0 && < 2.9 - , unliftio >= 0.2.25.0 && < 0.3 - , vector >= 0.13.1.0 && < 0.14 - -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/Conduit.hs b/src/OM/Legion/Conduit.hs deleted file mode 100644 --- a/src/OM/Legion/Conduit.hs +++ /dev/null @@ -1,40 +0,0 @@ -{- | 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 --- a/src/OM/Legion/Connection.hs +++ b/src/OM/Legion/Connection.hs @@ -31,8 +31,8 @@ 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.Function (($), (&), (.)) import Data.Map (Map) import GHC.Generics (Generic) import Network.Socket (PortNumber) @@ -41,11 +41,11 @@ PeerMessage, close, enqueueMsg, newMsgChan, stream) import OM.Show (showt) import OM.Socket (AddressDescription(AddressDescription), openEgress) -import Prelude (Applicative(pure), Bool(False, True), Either(Left, - Right), Maybe(Just, Nothing), Monad((>>=)), Semigroup((<>)), ($), - (.), Eq, IO, Show) +import Prelude (Applicative(pure), Bool(False, True), Either(Left, Right), + Maybe(Just, Nothing), Monad((>>=)), Semigroup((<>)), Eq, IO, Show) import System.Clock (TimeSpec) import qualified Data.Map as Map +import qualified Streaming.Prelude as Stream {- | A handle on the connection to a peer. -} @@ -87,10 +87,10 @@ in finally ( - (tryAny . runConduit) ( + tryAny ( stream rsSelf msgChan - .| logMessageSend - .| openEgress addy + & Stream.mapM logMessageSend + & openEgress addy ) >>= \case Left err -> logInfo $ "Disconnecting because of error: " <> showt err @@ -110,15 +110,13 @@ 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 - ) + => (Peer, PeerMessage e) + -> w (Peer, PeerMessage e) + logMessageSend msg = do + logDebug + $ "Sending Message to Peer (peer, msg): " + <> showt (peer, msg) + pure msg {- | diff --git a/src/OM/Legion/MsgChan.hs b/src/OM/Legion/MsgChan.hs --- a/src/OM/Legion/MsgChan.hs +++ b/src/OM/Legion/MsgChan.hs @@ -29,7 +29,6 @@ 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) @@ -37,6 +36,8 @@ import Data.UUID (UUID) import GHC.Generics (Generic) import Web.HttpApiData (FromHttpApiData, ToHttpApiData) +import Streaming.Prelude (Stream, Of) +import qualified Streaming.Prelude as Stream {- | A specialized channel for outgoing peer messages. -} @@ -60,7 +61,7 @@ :: MonadIO m => Peer -> MsgChan e - -> ConduitT void (Peer, PeerMessage e) m () + -> Stream (Of (Peer, PeerMessage e)) m () stream self (MsgChan chan) = (join . liftIO . atomically) $ readTVar chan >>= \case @@ -77,7 +78,7 @@ -} writeTVar chan (Just []) pure $ do - traverse_ (yield . (self,)) messages + traverse_ (Stream.yield . (self,)) messages stream self (MsgChan chan) diff --git a/src/OM/Legion/Runtime.hs b/src/OM/Legion/Runtime.hs --- a/src/OM/Legion/Runtime.hs +++ b/src/OM/Legion/Runtime.hs @@ -38,7 +38,7 @@ Stats(..), ) where -import Control.Arrow ((&&&)) +import Control.Arrow (Arrow((&&&))) import Control.Concurrent (Chan, newChan, readChan, threadDelay, writeChan) import Control.Exception.Safe (MonadCatch, tryAny) @@ -50,7 +50,7 @@ logInfo) import Control.Monad.State (MonadState(get, put), StateT, evalStateT, gets, modify') -import Control.Monad.Trans.Class (lift) +import Control.Monad.Trans.Class (MonadTrans(lift)) import Data.Aeson (ToJSON) import Data.Binary (Binary) import Data.ByteString.Lazy (ByteString) @@ -59,8 +59,8 @@ 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.Function ((&)) import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef) import Data.Map (Map) import Data.Set ((\\), Set) @@ -70,7 +70,6 @@ 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), @@ -82,6 +81,11 @@ import OM.Socket (AddressDescription(AddressDescription), connectServer, openIngress, openServer) import OM.Time (MonadTimeSpec(getTime), addTime, diffTimeSpec) +import Prelude (Applicative(pure), Either(Left, Right), Enum(succ), + Eq((/=)), Functor(fmap), Maybe(Just, Nothing), Monad((>>), (>>=), + return), Monoid(mempty), Ord((<=), (>=)), Semigroup((<>)), ($), (.), + (<$>), (=<<), IO, Int, MonadFail, Show, String, fst, mapM_, maybe, + seq, sequence_) import System.Clock (TimeSpec) import System.Random.Shuffle (shuffleM) import qualified Data.Binary as Binary @@ -89,6 +93,7 @@ import qualified Data.Map as Map import qualified Data.Set as Set import qualified OM.Fork as Fork +import qualified Streaming.Prelude as Stream {-# ANN module ("HLint: ignore Redundant <$>" :: String) #-} {-# ANN module ("HLint: ignore Use underscore" :: String) #-} @@ -198,7 +203,7 @@ put (Stats timeWithoutProgress) = Binary.put (diffTimeToPicoseconds <$> timeWithoutProgress) - + {- | The type of the runtime message channel. -} newtype RChan e = RChan { unRChan :: Chan (RuntimeMessage e) @@ -250,7 +255,8 @@ {- | Send the result of a call back to the peer that originated it. -} -sendCallResponse :: (MonadIO m) +sendCallResponse + :: (MonadIO m) => RChan e -> Peer -> MessageId @@ -410,33 +416,39 @@ <> ":" <> 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) - ) + openIngress addy + `Stream.for` + (\ (msgSource, msg) -> do + liftIO $ do + now <- getTime + atomicModifyIORef' + peerStats + (\peerTimes -> (Map.insert msgSource now peerTimes, ())) + lift $ logDebug $ "Handling: " <> showt (msgSource :: Peer, msg) + case msg of + PMFullMerge ps -> Stream.yield (FullMerge ps) + PMOutputs outputs -> Stream.yield (Outputs outputs) + PMMerge ps -> Stream.yield (Merge ps) + PMCall source mid callMsg -> + (liftIO . tryAny) (handleUserCall callMsg) >>= \case + Left err -> + lift . logError + $ "User call handling failed with: " <> showt err + Right v -> sendCallResponse runtimeChan source mid v + PMCast castMsg -> liftIO (handleUserCast castMsg) + PMCallResponse source mid responseMsg -> + Stream.yield (HandleCallResponse source mid responseMsg) + ) + & Stream.mapM_ (liftIO . writeChan (unRChan runtimeChan)) - runJoinListener :: (MonadLoggerIO m, MonadFail m) => m () + runJoinListener + :: ( MonadCatch m + , MonadFail m + , MonadLogger m + , MonadUnliftIO m + , Race + ) + => m () runJoinListener = let addy :: AddressDescription @@ -447,13 +459,11 @@ <> ":" <> showt joinMessagePort ) in - runConduit ( - pure () - .| openServer addy Nothing - .| awaitForever (\(req, respond_) -> lift $ - Fork.call runtimeChan (Join req) >>= respond_ - ) - ) + openServer addy Nothing + & Stream.mapM_ + (\(req, respond_) -> + Fork.call runtimeChan (Join req) >>= respond_ + ) runPeriodicResent :: (MonadIO m) => m () runPeriodicResent = @@ -751,7 +761,7 @@ modify' ( let - doModify state = + doModify state = newCluster `seq` state { rsClusterState = newCluster diff --git a/test/Test/OM/Legion.hs b/test/Test/OM/Legion.hs deleted file mode 100644 --- a/test/Test/OM/Legion.hs +++ /dev/null @@ -1,136 +0,0 @@ -{-# 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 deleted file mode 100644 --- a/test/k8s/k8s.mustache +++ /dev/null @@ -1,111 +0,0 @@ -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 deleted file mode 100644 --- a/test/om-legion-test-driver.hs +++ /dev/null @@ -1,426 +0,0 @@ -{-# 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 (AsValue(_Array, _String), key) -import Data.Foldable (for_) -import Data.String (IsString(fromString)) -import OM.Fork (Race, runRace) -import OM.Logging (standardLogging) -import OM.Show (showt) -import Prelude (Applicative((<*>), pure), Bool(False, True), Either(Left, - Right), Eq((==)), Foldable(foldl, null), Maybe(Just, Nothing), - Monad((>>=)), MonadFail(fail), Num((-)), Semigroup((<>)), Show(show), - ($), (&&), (.), (<$>), IO, Int, String, flip, id, not) -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 deleted file mode 100644 --- a/test/om-legion-test-inc.hs +++ /dev/null @@ -1,34 +0,0 @@ -{-# 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 deleted file mode 100644 --- a/test/om-legion-test-node.hs +++ /dev/null @@ -1,157 +0,0 @@ -{-# 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 (AsValue(_Array, _String), 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 Prelude (Applicative(pure), Eq((==)), Maybe(Nothing), Monad((>>=)), - Num((+)), Semigroup((<>)), ($), (&&), (.), (<$>), (=<<), Bool, IO, - filter, flip, fromIntegral, not) -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 deleted file mode 100644 --- a/test/om-legion-test-profile.hs +++ /dev/null @@ -1,197 +0,0 @@ -{-# 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 deleted file mode 100644 --- a/test/om-legion-test-run.hs +++ /dev/null @@ -1,42 +0,0 @@ -{-# 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 deleted file mode 100644 --- a/test/om-legion-test-stable.hs +++ /dev/null @@ -1,61 +0,0 @@ -{-# LANGUAGE NumericUnderscores #-} - -{-# OPTIONS_GHC -Wno-incomplete-uni-patterns #-} - -{- | 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) - -