om-legion 6.9.0.4 → 6.9.0.5
raw patch · 13 files changed
+73/−1335 lines, 13 filesdep +streamingdep −conduitdep −hostnamedep −hspecdep ~om-socketPVP ok
version bump matches the API change (PVP)
Dependencies added: streaming
Dependencies removed: conduit, hostname, hspec, lens, lens-aeson, mustache, om-kubernetes, om-legion, safe, template-haskell, unix, unliftio, vector
Dependency ranges changed: om-socket
API changes (from Hackage documentation)
Files
- om-legion.cabal +5/−72
- src/OM/Legion/Conduit.hs +0/−40
- src/OM/Legion/Connection.hs +14/−16
- src/OM/Legion/MsgChan.hs +4/−3
- src/OM/Legion/Runtime.hs +50/−40
- test/Test/OM/Legion.hs +0/−136
- test/k8s/k8s.mustache +0/−111
- test/om-legion-test-driver.hs +0/−426
- test/om-legion-test-inc.hs +0/−34
- test/om-legion-test-node.hs +0/−157
- test/om-legion-test-profile.hs +0/−197
- test/om-legion-test-run.hs +0/−42
- test/om-legion-test-stable.hs +0/−61
om-legion.cabal view
@@ -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
− src/OM/Legion/Conduit.hs
@@ -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)--
src/OM/Legion/Connection.hs view
@@ -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 {- |
src/OM/Legion/MsgChan.hs view
@@ -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)
src/OM/Legion/Runtime.hs view
@@ -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
− test/Test/OM/Legion.hs
@@ -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--
− test/k8s/k8s.mustache
@@ -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
− test/om-legion-test-driver.hs
@@ -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}) "<namespace>")- "The k8s namespace to deploy to."- , Option- "i"- ["image"]- (ReqArg (\val cfg -> cfg {image = Just val}) "<docker-image>")- "The docker image to deploy."- , Option- ""- ["log-level"]- (- OptArg- (\mLevel cfg ->- case mLevel of- Nothing -> cfg {logLevel = LevelInfo}- Just "DEBUG" -> cfg {logLevel = LevelDebug}- Just "INFO" -> cfg {logLevel = LevelInfo}- Just "WARN" -> cfg {logLevel = LevelWarn}- Just "ERROR" -> cfg {logLevel = LevelError}- Just other -> cfg {logLevel = LevelOther (fromString other)}- )- "<log-level>"- )- "The logging level to choose. [DEBUG, INFO, WARN, ERROR]"- ]---runTests :: Spec -> IO ()-runTests spec =- runSpec spec defaultConfig >>= evaluateSummary---terminateCluster- :: forall m.- ( MonadFail m- , MonadLoggerIO m- , MonadUnliftIO m- )- => Config Identity- -> m ()-terminateCluster Config {namespace} =- finally- (do- logDumpedState 0- logDumpedState 1- logDumpedState 2- ) (- cmd- "kubectl"- [ "delete"- , "namespace"- , namespace- ]- )- where- logDumpedState :: Int -> m ()- logDumpedState n = do- (_, out, _) <-- readProcessWithExitCode- "bash"- [- "-c",- "kubectl exec -i -n "- <> show namespace- <> " "- <> show namespace- <> "-"- <> show n- <> " bash -- -c 'cat log | gzip' | gzip -d"- ]- ""- logInfo $ "logs " <> showt n <> ": " <> showt out---launchCluster- :: ( MonadFail m- , MonadLogger m- , MonadUnliftIO m- )- => Config Identity- -> m ()-launchCluster Config { namespace, image } = do- readProcessWithExitCode- "kubectl"- [- "apply",- "-f",- "-"- ]- (Text.unpack (k8sConfig namespace image))- >>= \case- (ExitSuccess, out, err) -> do- logInfo (fromString out)- unless (null err) $ logError (fromString err)- err -> fail ("Got apply error: " <> show err)- waitForReady namespace 0---cmd- :: ( MonadFail m- , MonadLogger m- , MonadUnliftIO m- )- => String- -> [String]- -> m ()-cmd prog args = do- withCreateProcess- (proc prog args)- (\_ _ _ handle ->- waitForProcess handle >>= \case- result@(ExitFailure _) ->- fail- $ "Failed:\n"- <> showCommandForUser prog args- <> "\nwith: " <> show result- result@ExitSuccess ->- logDebug ("got: " <> showt result)- )---{- | Check that n nodes are reporting stable stats. -}-checkStable- :: ( MonadFail m- , MonadLogger m- , MonadUnliftIO m- )- => Config Identity- -> Int- -> Int- -> m ()-checkStable config n v = do- cmd- "kubectl"- [ "-n"- , (namespace config)- , "run"- , "test-driver"- , "--image=us.gcr.io/friendlee/om-legion-test:latest"- , "-i"- , "--rm"- , "--restart=Never"- , "--env=NAMESPACE=" <> namespace config- , "--env=LOG_LEVEL=DEBUG"- , "--serviceaccount=" <> namespace config- , "--wait=true"- , "--"- , "/bin/om-legion-test-stable"- , show n- , show v- ]---waitForStable- :: ( MonadFail m- , MonadLogger m- , MonadUnliftIO m- )- => Config Identity- -> Int- -> Int- -> Int- -> m ()-waitForStable config n v tries =- case tries of- 0 -> fail "The cluster doesn't seem like it's going to stabilize."- _ -> do- runOps config 0 [ReadState, ReadStats]- try (checkStable config n v) >>= \case- Left (_err :: SomeException) -> do- logInfo $ "Not yet stable."- waitForStable config n v (tries - 1)- Right () -> pure ()---data Config f = Config- { namespace :: F f String- , image :: F f String- , logLevel :: LogLevel- }-deriving stock instance Show (Config Maybe)---type family F a b where- F Identity a = a- F f a = f a---validate :: Config Maybe -> Maybe (Config Identity)-validate (Config namespace image logLevel) =- Config- <$> namespace- <*> image- <*> pure logLevel--
− test/om-legion-test-inc.hs
@@ -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."---
− test/om-legion-test-node.hs
@@ -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--
− test/om-legion-test-profile.hs
@@ -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}) "<namespace>")- "The k8s namespace to deploy to."- , Option- ""- ["num-nodes"]- (ReqArg (\val cfg -> cfg {numNodes = readMay val}) "<num-nodes>")- "The number of nodes in the cluster."- , Option- ""- ["num-ops"]- (ReqArg (\val cfg -> cfg {numOps = readMay val}) "<num-ops>")- "The number of increments to perform on each node, in each thread."- , Option- ""- ["num-threads"]- (ReqArg (\val cfg -> cfg {threads = readMay val}) "<num-threads>")- "The number of request threads each node should be hit with."- , Option- ""- ["log-level"]- (- OptArg- (\mLevel cfg ->- case mLevel of- Nothing -> cfg {logLevel = LevelInfo}- Just "DEBUG" -> cfg {logLevel = LevelDebug}- Just "INFO" -> cfg {logLevel = LevelInfo}- Just "WARN" -> cfg {logLevel = LevelWarn}- Just "ERROR" -> cfg {logLevel = LevelError}- Just other -> cfg {logLevel = LevelOther (fromString other)}- )- "<log-level>"- )- "The logging level to choose. [DEBUG, INFO, WARN, ERROR]"- ]---cmd- :: ( MonadFail m- , MonadLogger m- , MonadUnliftIO m- )- => String- -> [String]- -> m ()-cmd prog args = do- withCreateProcess- (proc prog args)- (\_ _ _ handle ->- waitForProcess handle >>= \case- result@(ExitFailure _) ->- fail- $ "Failed:\n"- <> showCommandForUser prog args- <> "\nwith: " <> show result- result@ExitSuccess ->- logDebug ("got: " <> showt result)- )---data Config f = Config- { namespace :: F f String- , logLevel :: LogLevel- , numNodes :: F f Int- , numOps :: F f Int- , threads :: F f Int- }-deriving stock instance Show (Config Maybe)---type family F a b where- F Identity a = a- F f a = f a---validate :: Config Maybe -> Maybe (Config Identity)-validate (Config namespace logLevel numNodes numOps threads) =- Config- <$> namespace- <*> pure logLevel- <*> numNodes- <*> numOps- <*> threads--
− test/om-legion-test-run.hs
@@ -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)--
− test/om-legion-test-stable.hs
@@ -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)--