legion 0.1.0.0 → 0.1.0.1

raw patch · 11 files changed

+239/−65 lines, 11 filesdep +canteven-httpdep +http-typesdep +wai

Dependencies added: canteven-http, http-types, wai, wai-extra

Files

• legion.cabal +6/−1
• src/Network/Legion/Admin.hs +68/−21
• src/Network/Legion/Application.hs +1/−2
• src/Network/Legion/ClusterState.hs +18/−0
• src/Network/Legion/Conduit.hs +4/−10
• src/Network/Legion/ConnectionManager.hs +5/−6
• src/Network/Legion/Distribution.hs +4/−0
• src/Network/Legion/PowerState.hs +44/−8
• src/Network/Legion/Propagation.hs +25/−5
• src/Network/Legion/Runtime.hs +1/−4
• src/Network/Legion/StateMachine.hs +63/−8

@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/  name:                legion-version:             0.1.0.0+version:             0.1.0.1 synopsis:            Distributed, stateful, homogeneous microservice framework. description:         Legion is a framework for writing distributed,                      homogeneous, stateful microservices in Haskell.@@ -44,6 +44,7 @@     Network.Legion.Settings     Network.Legion.StateMachine     Network.Legion.UUID+    Paths_legion   -- other-extensions:       build-depends:     Ranged-sets >= 0.3.0 && < 0.4,@@ -52,6 +53,7 @@     binary >= 0.7.5 && < 0.9,     binary-conduit >= 1.2.3 && < 1.3,     bytestring >= 0.10.4.0 && < 0.11,+    canteven-http >= 0.1.1.1 && < 0.2,     conduit >= 1.2.4 && < 1.3,     conduit-extra >= 1.1.9 && < 1.2,     containers >= 0.5.5.1 && < 0.6,@@ -59,6 +61,7 @@     data-dword >= 0.3 && < 0.4,     directory >= 1.2.1.0 && < 1.3,     exceptions >= 0.8 && < 0.9,+    http-types >= 0.9.1 && < 0.10,     monad-logger >= 0.3.17 && < 0.4,     network >= 2.6.2.1 && < 2.7,     scotty >= 0.11.0 && < 0.12,@@ -69,6 +72,8 @@     transformers >= 0.3.0.0 && < 0.5,     unix >= 2.7 && < 2.8,     uuid >= 1.3.11 && < 1.4,+    wai >= 3.2.1.1 && < 3.3,+    wai-extra >= 3.0.16.1 && < 3.1,     warp >= 3.2 && < 3.3   hs-source-dirs:      src   default-language:    Haskell2010

@@ -1,5 +1,7 @@+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-} {- |   This module contains the admin interface code. -}@@ -7,24 +9,35 @@   runAdmin, ) where +import Canteven.HTTP (requestLogging, logExceptionsAndContinue) import Control.Concurrent (forkIO, newChan, newEmptyMVar, writeChan,   putMVar, takeMVar, Chan) import Control.Monad (void)-import Control.Monad.Logger (askLoggerIO, runLoggingT)+import Control.Monad.Logger (askLoggerIO, runLoggingT, logDebug) import Control.Monad.Trans.Class (lift) import Data.Conduit (Source) import Data.Default.Class (def)+import Data.Text.Encoding (encodeUtf8) import Data.Text.Lazy (Text, pack)+import Data.Version (showVersion)+import Network.HTTP.Types (notFound404) import Network.Legion.Application (LegionConstraints) import Network.Legion.Conduit (chanToSource) import Network.Legion.LIO (LIO) import Network.Legion.PartitionKey (PartitionKey(K))-import Network.Legion.StateMachine (AdminMessage(GetState, GetPart))+import Network.Legion.StateMachine (AdminMessage(GetState, GetPart,+  Eject))+import Network.Wai (Middleware, modifyResponse) import Network.Wai.Handler.Warp (HostPreference, defaultSettings, Port,   setHost, setPort)-import Web.Scotty.Resource.Trans (resource, get)+import Network.Wai.Middleware.AddHeaders (addHeaders)+import Network.Wai.Middleware.StripHeaders (stripHeader)+import Paths_legion (version)+import Text.Read (readMaybe)+import Web.Scotty.Resource.Trans (resource, get, delete) import Web.Scotty.Trans (Options, scottyOptsT, settings, ScottyT, text,-  ActionT, param)+  ActionT, param, middleware, status)+import qualified Data.Text as T  {- |   Start the admin service in a background thread.@@ -34,23 +47,36 @@   -> HostPreference   -> LIO (Source LIO (AdminMessage i o s)) runAdmin addr host = do-  logging <- askLoggerIO-  chan <- lift newChan-  void . lift . forkIO . (`runLoggingT` logging) $-    let-      website :: ScottyT Text LIO ()-      website = do-        resource "/clusterstate" $-          get $ do-            val <- send chan GetState-            text (pack (show val))-        resource "/propstate/:key" $-          get $ do-            key <- K . read <$> param "key"-            val <- send chan (GetPart key)-            text (pack (show val))-    in scottyOptsT (options addr host) (`runLoggingT` logging) website-  return (chanToSource chan)+    logging <- askLoggerIO+    chan <- lift newChan+    void . lift . forkIO . (`runLoggingT` logging) $+      let+        website :: ScottyT Text LIO ()+        website = do+          middleware+            $ requestLogging logging+            . setServer+            . logExceptionsAndContinue logging++          resource "/clusterstate" $+            get $ do+              val <- send chan GetState+              text (pack (show val))+          resource "/propstate/:key" $+            get $ do+              key <- K . read <$> param "key"+              val <- send chan (GetPart key)+              text (pack (show val))+          resource "/peers/:peer" $+            delete $+              readMaybe <$> param "peer" >>= \case+                Nothing -> status notFound404+                Just peer -> do+                  lift . $(logDebug) . T.pack $ "Ejecting peer: " ++ show peer+                  send chan (Eject peer)++      in scottyOptsT (options addr host) (`runLoggingT` logging) website+    return (chanToSource chan)   where     send       :: Chan (AdminMessage i o s)@@ -73,4 +99,25 @@       $ defaultSettings   } ++setServer :: Middleware+setServer = addServerHeader . stripServerHeader+  where+    {- |+      Strip the server header+    -}+    stripServerHeader :: Middleware+    stripServerHeader = modifyResponse (stripHeader "Server") ++    {- |+      Add our own server header.+    -}+    addServerHeader :: Middleware+    addServerHeader = addHeaders [("Server", serverValue)]++    {- |+      The value of the @Server:@ header.+    -}+    serverValue =+      encodeUtf8 (T.pack ("legion-admin/" ++ showVersion version)) 

@@ -40,8 +40,7 @@     {- |       The request handler, implemented by the user to service requests. -      Returns a response to the request, together with the new partition-      state.+      Given a request and a state, returns a response to the request.     -}     handleRequest :: PartitionKey -> i -> s -> o,     {- |

@@ -17,6 +17,7 @@   findPartition,   getDistribution,   joinCluster,+  eject,   mergeEither,   actions,   allParticipants,@@ -83,6 +84,7 @@ data Update   = PeerJoined Peer BSockAddr   | Participating Peer KeySet+  | PeerEjected Peer   deriving (Show, Generic) instance Binary Update instance ApplyDelta Update ClusterState where@@ -90,6 +92,11 @@     cs {peers = Map.insert peer addr peers}   apply (Participating peer ks) cs@ClusterState {distribution} =     cs {distribution = modify (Set.insert peer) ks distribution}+  apply (PeerEjected peer) cs@ClusterState {distribution, peers} =+    cs {+        distribution = modify (Set.delete peer) full distribution,+        peers = Map.delete peer peers+      }   {- |@@ -170,6 +177,17 @@   ClusterPropState   . P.delta (PeerJoined peer addy)   . P.participate peer+  . unPropState+++{- |+  Eject a peer from the cluster.+-}+eject :: Peer -> ClusterPropState -> ClusterPropState+eject peer =+  ClusterPropState+  . P.delta (PeerEjected peer)+  . P.disassociate peer   . unPropState  

@@ -11,27 +11,21 @@ import Control.Concurrent.Chan (Chan, newChan, writeChan, readChan) import Control.Monad (void, forever) import Control.Monad.IO.Class (MonadIO, liftIO)-import Data.Conduit (Source, Sink, ($$), await, ($=), yield, await)+import Data.Conduit (Source, Sink, ($$), ($=), yield, awaitForever) import qualified Data.Conduit.List as CL (map)  {- |-  Convert a chanel into a Source.+  Convert a channel into a Source. -} chanToSource :: (MonadIO io) => Chan a -> Source io a chanToSource chan = forever $ yield =<< liftIO (readChan chan)   {- |- Convert an chanel into a Sink.+ Convert a channel into a Sink. -} chanToSink :: (MonadIO io) => Chan a -> Sink a io ()-chanToSink chan = do-  val <- await-  case val of-    Nothing -> return ()-    Just v -> do-      liftIO (writeChan chan v)-      chanToSink chan+chanToSink chan = awaitForever (liftIO . writeChan chan)   {- |

@@ -1,4 +1,5 @@ {-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE TemplateHaskell #-} {- |@@ -116,9 +117,8 @@       "working" socket is.     -}     sendWithRetry :: Maybe Socket -> ByteString -> LIO (Maybe Socket)-    sendWithRetry Nothing payload = do-      result <- (lift . try) openSocket-      case result of+    sendWithRetry Nothing payload =+      (lift . try) openSocket >>= \case         Left err -> do           $(logWarn) . pack             $ "Can't connect to: " ++ show addr ++ ". Dropping message on "@@ -136,9 +136,8 @@               ++ "The message was: " ++ show payload             Right _ -> return ()           return (Just so)-    sendWithRetry (Just so) payload = do-      result <- (lift . try) (sendAll so payload)-      case result of+    sendWithRetry (Just so) payload =+      (lift . try) (sendAll so payload) >>= \case         Left err -> do           $(logInfo) . pack             $ "Socket to " ++ show addr ++ " died. Retrying on a new "

@@ -27,6 +27,7 @@ import Network.Legion.LIO (LIO) import Network.Legion.PartitionKey (PartitionKey) import Network.Legion.UUID (getUUID)+import Text.Read (readPrec) import qualified Data.Set as Set import qualified Network.Legion.KeySet as KS @@ -35,6 +36,9 @@   The way to identify a peer. -} newtype Peer = Peer UUID deriving (Show, Binary, Eq, Ord)+instance Read Peer where+  readPrec = Peer <$> readPrec+     {- |

@@ -36,8 +36,8 @@ import Data.Set (Set, union, (\\), null, member) import Data.Word (Word64) import GHC.Generics (Generic)-import qualified Data.Map as Map (insert, empty)-import qualified Data.Set as Set (insert, empty, delete)+import qualified Data.Map as Map+import qualified Data.Set as Set   {- |@@ -164,13 +164,49 @@   -> PowerState o s p d   -> Either String (PowerState o s p d) mergeEither (PowerState o1 i1 d1) (PowerState o2 i2 d2) | o1 == o2 =-    Right . reduce $ PowerState {origin = o1, infimum, deltas}+    Right . reduce . removeRenegade $ PowerState {+        origin = o1,+        infimum,+        deltas = removeObsolete (unionWith mergeAcks d1 d2)+      }   where     infimum = max i1 i2-    deltas = removeObsolete (unionWith mergeKnowns d1 d2)++    {- |+      Obsolete deltas are deltas that are already included in the latest+      infimum.+    -}     removeObsolete = filterWithKey (\k _ -> k > stateId infimum)-    mergeKnowns (d, s1) (_, s2) = (d, s1 `union` s2) +    {- |+      Renegade deltas are deltas that originate from a non-participating+      peer.  This might happen in a network partition situation, where+      the cluster ejected a peer that later reappears on the network,+      broadcasting updates.++      In reality, this will probably always be a no-op because the+      message dispatcher in the main state machine will immediately+      drop messages that originate from unknown peers (where "unknown"+      includes peers that have been ejected), so it is unlikely that any+      renegade merge requests will make it this far, but you can never+      be too paranoid I guess.+    -}+    removeRenegade ps =+        ps {+            deltas =+              fromAscList+              . filter nonRenegade+              . toAscList+              . deltas+              $ ps+          }+      where+        nonRenegade (BottomSid, _) = True+        nonRenegade (Sid _ p, _) = p `member` peers+        peers = allParticipants ps++    mergeAcks (d, s1) (_, s2) = (d, s1 `union` s2)+ mergeEither a b = Left   $ "PowerStates " ++ show a ++ " and " ++ show b ++ " do not share the "   ++ "same origin, and cannot be merged."@@ -198,7 +234,7 @@   => p   -> PowerState o s p d   -> PowerState o s p d-participate p ps@PowerState {deltas} = reduce ps {+participate p ps@PowerState {deltas} = acknowledge p $ ps {     deltas = Map.insert (nextId p ps) (Join p, Set.empty) deltas   } @@ -211,7 +247,7 @@   => p   -> PowerState o s p d   -> PowerState o s p d-disassociate p ps@PowerState {deltas} = reduce ps {+disassociate p ps@PowerState {deltas} = acknowledge p $ ps {     deltas = Map.insert (nextId p ps) (UnJoin p, Set.empty) deltas   } @@ -224,7 +260,7 @@   -> d   -> PowerState o s p d   -> PowerState o s p d-delta p d ps@PowerState {deltas} = reduce ps {+delta p d ps@PowerState {deltas} = acknowledge p $ ps {     deltas = Map.insert (nextId p ps) (Delta d, Set.empty) deltas   } 

@@ -18,6 +18,7 @@   getPowerState,   ask,   participate,+  disassociate,   getSelf,   divergences,   participating,@@ -141,10 +142,6 @@   Like `merge`, but total. `mergeEither` returns a human readable reason why   the foreign powerstate can't be merged in the event of an error. -}-{--  This algorithm is weaksauce. We need to find someone who knows a lot about-  gossip protocols to fix this.--} mergeEither :: (Eq o, Ord p, Show o, Show s, Show p, Show d, ApplyDelta d s)   => p   -> PropPowerState o s p d@@ -156,6 +153,11 @@     Left err -> Left err     Right merged -> Right prop {         powerState = merged,++        {-+          This algorithm is weaksauce. We need to find someone who knows+          a lot about gossip protocols to fix this.+        -}         peerStates =           Map.fromList $ [               (p, ns)@@ -219,7 +221,7 @@   -> PropState o s p d   -> PropState o s p d delta d prop@PropState {self, powerState, now} =-  let newPowerState = acknowledge self (PS.delta self d powerState)+  let newPowerState = PS.delta self d powerState   in prop {       powerState = newPowerState,       peerStates = Map.fromAscList [@@ -283,6 +285,24 @@   -> PropState o s p d participate peer prop@PropState {powerState, now} =   let newPowerState = PS.participate peer powerState+  in prop {+      powerState = newPowerState,+      peerStates = Map.fromAscList [+          (p, NeedsSendAt now)+          | p <- Set.toAscList (divergent newPowerState)+        ]+    }+++{- |+  Eject a participant from the power state.+-}+disassociate :: (Ord p, ApplyDelta d s)+  => p+  -> PropState o s p d+  -> PropState o s p d+disassociate peer prop@PropState {powerState, now} =+  let newPowerState = PS.disassociate peer powerState   in prop {       powerState = newPowerState,       peerStates = Map.fromAscList [

@@ -133,10 +133,7 @@   = NewCluster     -- ^ Indicates that we should bootstrap a new cluster at startup. The     --   persistence layer may be safely pre-populated because the new-    --   node will claim the entire keyspace. Future plans include-    --   implementing some safeguards to make sure only one node in-    --   a cluster was started using this startup mode, but for now,-    --   we are counting on you, the user, to do the right thing.+    --   node will claim the entire keyspace.    | JoinCluster SockAddr     -- ^ Indicates that the node should try to join an existing cluster,     --   either by starting fresh, or by recovering from a shutdown

@@ -8,6 +8,15 @@ {-# LANGUAGE TemplateHaskell #-} {- |   This module contains the state machine implementation of a legion node.++  Discussion:++  This is a first attempt to discover a pure legion state machine and isolated+  it from the runtime IO considerations. It is obviously not perfect, because+  everything still lives in 'LIO', which is 'IO'-backed; but mostly this is+  because access to the persistence layer still happens here. Once we pull that+  out into the 'Network.Legion.Runtime' module we should be clear to remove IO+  and make this thing look more like a pure state machine. - Rick -} module Network.Legion.StateMachine (   stateMachine,@@ -69,7 +78,13 @@ import qualified Network.Legion.PartitionState as P  -{- | This conduit houses the main legionary state machine.  -}+{- |+  This conduit houses the main legionary state machine. The conduit's+  input, internal state, and output are analogous to a "real" state+  machine's input, state, and output. If this seems like an odd use of+  conduit, that's ok.  Hopefully we can make this look more like a pure+  state machine once we remove 'IO' from this module.+-} stateMachine :: (LegionConstraints i o s)   => Legionary i o s   -> NodeState i o s@@ -117,7 +132,7 @@         Just (peer, _) ->           forward peer key request respond     J m -> handleJoinRequest m-    A m -> lift . handleAdminMessage l m =<< getS+    A m -> handleAdminMessage l m   {- | Handles one incomming message from a peer. -}@@ -272,15 +287,42 @@ handleAdminMessage   :: Legionary i o s   -> AdminMessage i o s-  -> NodeState i o s-  -> LIO ()-handleAdminMessage _ (GetState respond) ns =-  respond ns-handleAdminMessage Legionary {persistence} (GetPart key respond) _ = do+  -> StateM i o s ()+handleAdminMessage _ (GetState respond) =+  lift . respond =<< getS+handleAdminMessage Legionary {persistence} (GetPart key respond) = lift $ do   partitionVal <- lift (getState persistence key)   respond partitionVal+handleAdminMessage _ (Eject peer respond) = do+    {-+      TODO: we should attempt to notify the ejected peer that it has+      been ejected instead of just cutting it off and washing our hands+      of it. I have a vague notion that maybe ejected peers should be+      permanently recorded in the cluster state so that if they ever+      reconnect then we can notify them that they are no longer welcome+      to participate. +      On a related note, we need to think very hard about the split brain+      problem. A random thought about that is that we should consider the+      extreme case where the network just fails completely and every node+      believes that every other node should be or has been ejected. This+      would obviously be catastrophic in terms of data durability unless+      we have some way to reintegrate an ejected node. So, either we+      have to guarantee that such a situation can never happen, or else+      implement a reintegration strategy.  It might be acceptable for+      the reintegration strategy to be very costly if it is characterized+      as an extreme recovery scenario. +      Question: would a reintegration strategy become less costly if the+      "next state id" for a peer were global across all power states+      instead of local to each power state?+    -}+    modifyS eject+    lift $ respond ()+  where+    eject ns@NodeState {cluster} = ns {cluster = C.eject peer cluster}++ {- | Update all of the propagation states with the current time.  -} heartbeat :: StateM i o s () heartbeat = do@@ -461,10 +503,12 @@ data AdminMessage i o s   = GetState (NodeState i o s -> LIO ())   | GetPart PartitionKey (Maybe (PartitionPowerState i s) -> LIO ())+  | Eject Peer (() -> LIO ())  instance Show (AdminMessage i o s) where   show (GetState _) = "(GetState _)"   show (GetPart k _) = "(GetPart " ++ show k ++ " _)"+  show (Eject p _) = "(Eject " ++ show p ++ " _)"   {- | Defines the local state of a node in the cluster.  -}@@ -479,7 +523,7 @@   deriving (Show)  -{- | A set of forwardmed messages.  -}+{- | A set of forwarded messages.  -} newtype Forwarded o = F {unF :: Map MessageId (o -> LIO ())} instance Show (Forwarded o) where   show = show . Map.keys . unF@@ -601,6 +645,10 @@     Functor, Applicative, Monad, MonadLogger, MonadCatch,     MonadThrow, MonadIO   )+{-+  We can lift things from the underlying monad straight to 'StateT',+  bypassing the `CondutM` layer.+-} instance MonadTrans (StateMT i o s) where   lift = StateMT . lift . lift @@ -640,5 +688,12 @@ -} putS :: NodeState i o s -> StateMT i o s m () putS = StateMT . put+++{- |+  Modify the node state.+-}+modifyS :: (NodeState i o s -> NodeState i o s) -> StateMT i o s m ()+modifyS f = putS . f =<< getS