packages feed

loup 0.0.1 → 0.0.2

raw patch · 12 files changed

+237/−37 lines, 12 filesdep +bytestringdep +timedep ~basenew-component:exe:loup-converger

Dependencies added: bytestring, time

Dependency ranges changed: base

Files

loup.cabal view
@@ -1,5 +1,5 @@ name:                  loup-version:               0.0.1+version:               0.0.2 synopsis:              Amazon Simple Workflow Service Wrapper for Work Pools. description:           Loup is a wrapper around Amazon Simple Workflow Service for Work Pools. homepage:              https://github.com/swift-nav/loup@@ -19,10 +19,12 @@ library   exposed-modules:     Network.AWS.Loup   other-modules:       Network.AWS.Loup.Act+                     , Network.AWS.Loup.Converge                      , Network.AWS.Loup.Ctx                      , Network.AWS.Loup.Decide                      , Network.AWS.Loup.Prelude                      , Network.AWS.Loup.Types+                     , Network.AWS.Loup.Types.Alias                      , Network.AWS.Loup.Types.Ctx                      , Network.AWS.Loup.Types.Product                      , Network.AWS.Loup.Types.Sum@@ -33,10 +35,12 @@                      , amazonka                      , amazonka-swf                      , base == 4.8.*+                     , bytestring                      , conduit                      , lifted-async                      , monad-control                      , preamble+                     , time                      , turtle                      , uuid                      , yaml@@ -53,6 +57,15 @@ executable loup-decider   hs-source-dirs:      main   main-is:             decider.hs+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N -Wall+  build-depends:       base+                     , loup+                     , optparse-generic+  default-language:    Haskell2010++executable loup-converger+  hs-source-dirs:      main+  main-is:             converger.hs   ghc-options:         -threaded -rtsopts -with-rtsopts=-N -Wall   build-depends:       base                      , loup
+ main/converger.hs view
@@ -0,0 +1,30 @@+{-# LANGUAGE DeriveGeneric     #-}+{-# LANGUAGE NoImplicitPrelude #-}+{-# LANGUAGE OverloadedStrings #-}++-- | Run converger.+--+import Network.AWS.Loup+import Options.Generic++-- | Args+--+-- Program arguments.+--+data Args = Args+  { domain :: Text+    -- ^ Configuration file.+  , pool   :: FilePath+    -- ^ Pool file to converge on.+  } deriving (Show, Generic)++instance ParseRecord Args++-- | Run converger.+--+main :: IO ()+main = do+  args <- getRecord "Converger"+  convergeMain+    (domain args)+    (pool args)
main/decider.hs view
@@ -2,7 +2,7 @@ {-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE OverloadedStrings #-} --- | Run actor.+-- | Run decider. -- import Network.AWS.Loup import Options.Generic
src/Network/AWS/Loup.hs view
@@ -4,7 +4,8 @@   ( module Exports   ) where -import Network.AWS.Loup.Act     as Exports-import Network.AWS.Loup.Decide  as Exports-import Network.AWS.Loup.Prelude as Exports-import Network.AWS.Loup.Types   as Exports+import Network.AWS.Loup.Act      as Exports+import Network.AWS.Loup.Converge as Exports+import Network.AWS.Loup.Decide   as Exports+import Network.AWS.Loup.Prelude  as Exports+import Network.AWS.Loup.Types    as Exports
src/Network/AWS/Loup/Act.hs view
@@ -9,6 +9,7 @@   , actMain   ) where +import Control.Concurrent import Control.Concurrent.Async.Lifted import Control.Monad.Trans.AWS import Network.AWS.Loup.Ctx@@ -30,31 +31,50 @@ completeActivity token =   void $ send $ respondActivityTaskCompleted token +-- | Cancel activity.+--+cancelActivity :: MonadAmazonCtx c m => Text -> m ()+cancelActivity token =+  void $ send $ respondActivityTaskCanceled token+ -- | Hearbeat. ---heartbeat :: MonadAmazonCtx c m => Text -> m ()-heartbeat token =-  void $ send $ recordActivityTaskHeartbeat token+heartbeat :: MonadAmazonCtx c m => Text -> m Bool+heartbeat token = do+  rathrs <- send $ recordActivityTaskHeartbeat token+  return $ rathrs ^. rathrsCancelRequested  -- | Run a managed action inside a temp directory. ---intempdir :: MonadIO m => Managed a -> m ()+intempdir :: MonadControl m => Managed a -> m () intempdir action =-  sh $ using $ do-    fromdir <- pwd-    todir   <- mktempdir "/tmp" "loup-"-    cptree fromdir todir-    cd todir-    action+  bracket pwd cd $ \fromdir ->+    sh $ using $ do+      todir <- mktempdir "/tmp" "loup-"+      cptree fromdir todir+      cd todir+      action +-- | Run heartbeat.+--+runHeartbeat :: MonadAmazonCtx c m => Text -> Int -> m ()+runHeartbeat token interval = do+  traceInfo "heartbeat" mempty+  liftIO $ threadDelay $ interval * 1000000+  nok <- heartbeat token+  if not nok then runHeartbeat token interval else do+    traceInfo "cancel" mempty+    cancelActivity token+ -- | Run command with input. ---runActivity :: MonadCtx c m => Text -> Maybe Text -> m ()-runActivity command input = do+runActivity :: MonadAmazonCtx c m => Text -> Text -> Maybe Text -> m ()+runActivity token command input = do   traceInfo "run" [ "command" .= command, "input" .= input]   intempdir $ do     liftIO $ maybe_ input $ writeTextFile "input.json"     stdout $ inshell command mempty+  completeActivity token  -- | Actor logic - poll for work, download artifacts, run command, upload artifacts. --@@ -65,8 +85,7 @@     (token, input) <- pollActivity domain (taskList queue)     maybe_ token $ \token' -> do       traceInfo "start" mempty-      race_ (runEvery (interval * 1000000) $ heartbeat token') (runActivity command input)-      completeActivity token'+      race_ (runHeartbeat token' interval) (runActivity token' command input)       traceInfo "finish" mempty  -- | Run actor from main with configuration.
+ src/Network/AWS/Loup/Converge.hs view
@@ -0,0 +1,68 @@+{-# LANGUAGE FlexibleContexts  #-}+{-# LANGUAGE NoImplicitPrelude #-}+{-# LANGUAGE OverloadedStrings #-}++-- | SWF Converger logic.+--+module Network.AWS.Loup.Converge+  ( converge+  , convergeMain+  ) where++import Control.Monad.Trans.AWS+import Data.Time.Clock.POSIX+import Data.Yaml+import Network.AWS.Loup.Ctx+import Network.AWS.Loup.Prelude+import Network.AWS.Loup.Types+import Network.AWS.SWF++-- | List open workflows.+--+listWorkflows :: MonadAmazonCtx c m => Text -> ActivityType -> m [Text]+listWorkflows domain activity = do+  let etf = executionTimeFilter $ posixSecondsToUTCTime $ fromIntegral (0 :: Int)+      wtf = workflowTypeFilter (activity ^. atName)+  weis <- pages $ set loweTypeFilter (return wtf) $ listOpenWorkflowExecutions domain etf+  let predicate wei = maybe True not $ wei ^. weiCancelRequested+  return $ view weWorkflowId . view weiExecution <$> filter predicate (join $ view weiExecutionInfos <$> weis)++-- | Start a workflow.+--+startWorkflow :: MonadAmazonCtx c m => Text -> ActivityType -> TaskList -> Text -> Maybe Text -> m ()+startWorkflow domain activity list wid input = do+  let wt = workflowType (activity ^. atName) (activity ^. atVersion)+  void $ send $ startWorkflowExecution domain wid wt+    & sTaskList .~ return list+    & sInput .~ input++-- | Cancel a workflow.+--+cancelWorkflow :: MonadAmazonCtx c m => Text -> Text -> m ()+cancelWorkflow domain wid =+  void $ send $ requestCancelWorkflowExecution domain wid++-- | Converger logic - get running workers and converge against pool.+--+converge :: MonadAmazonCtx c m => Text -> Pool -> m ()+converge domain pool =+  preAmazonCtx [ "label" .= LabelDecide, "domain" .= domain, "pool" .= pool ] $ do+    let activity = pool ^. pTask ^. tActivityType+    wids  <- listWorkflows domain activity+    let fold kvs as action = do+          let g k v bs = if k `elem` bs then return $ k `delete` bs else action k v >> return bs+          ifoldrM g as kvs+    wids' <- fold (pool ^. pWorkers) wids $ \wid input -> do+      traceInfo "start" [ "wid" .= wid, "input" .= input ]+      startWorkflow domain activity (pool ^. pTask ^. tTaskList) wid (encode' <$> input)+    forM_ wids' $ \wid -> do+      traceInfo "cancel" [ "wid" .= wid ]+      cancelWorkflow domain wid++-- | Run converger from main with configuration.+--+convergeMain :: MonadControl m => Text -> FilePath -> m ()+convergeMain domain file =+  runResourceT $ runCtx $ runStatsCtx $ runAmazonCtx $ do+    pools <- liftIO $ join . maybeToList <$> decodeFile file+    runConcurrent $ converge domain <$> pools
src/Network/AWS/Loup/Decide.hs view
@@ -10,8 +10,6 @@   ) where  import Control.Monad.Trans.AWS-import Data.Conduit-import Data.Conduit.List        hiding (foldM) import Data.UUID import Data.UUID.V4 import Data.Yaml@@ -24,37 +22,52 @@ -- pollDecision :: MonadAmazonCtx c m => Text -> TaskList -> m (Maybe Text, [HistoryEvent]) pollDecision domain list = do-  pfdtrs <- paginate (pollForDecisionTask domain list) $$ consume+  pfdtrs <- pages $ pollForDecisionTask domain list   return (join $ headMay $ view pfdtrsTaskToken <$> pfdtrs, reverse $ join $ view pfdtrsEvents <$> pfdtrs)  -- | Successful decision completion. -- completeDecision :: MonadAmazonCtx c m => Text -> [Decision] -> m () completeDecision token decisions =-  void $ send $ set rdtcDecisions decisions $ respondDecisionTaskCompleted token+  void $ send $ respondDecisionTaskCompleted token+    & rdtcDecisions .~ decisions +-- | Schedule activity decision.+-- scheduleActivity :: UUID -> ActivityType -> TaskList -> Maybe Text -> Decision scheduleActivity uid activity list input = do   let satda = scheduleActivityTaskDecisionAttributes activity (toText uid)         & satdaTaskList .~ return list         & satdaInput    .~ input-  set dScheduleActivityTaskDecisionAttributes (return satda) $ decision ScheduleActivityTask+  decision ScheduleActivityTask+    & dScheduleActivityTaskDecisionAttributes .~ return satda +-- | Complete activity decision.+-- completeActivity :: Decision completeActivity = do   let cweda = completeWorkflowExecutionDecisionAttributes-  set dCompleteWorkflowExecutionDecisionAttributes (return cweda) $ decision CompleteWorkflowExecution+  decision CompleteWorkflowExecution+    & dCompleteWorkflowExecutionDecisionAttributes .~ return cweda +-- | Cancel activity decision.+-- cancelActivity :: Decision cancelActivity = do   let cweda = cancelWorkflowExecutionDecisionAttributes-  set dCancelWorkflowExecutionDecisionAttributes (return cweda) $ decision CancelWorkflowExecution+  decision CancelWorkflowExecution+    & dCancelWorkflowExecutionDecisionAttributes .~ return cweda +-- | Request to cancel activity decision.+-- requestCancel :: UUID -> Decision requestCancel uid = do   let rcatda = requestCancelActivityTaskDecisionAttributes (toText uid)-  set dRequestCancelActivityTaskDecisionAttributes (return rcatda) $ decision RequestCancelActivityTask+  decision RequestCancelActivityTask+    & dRequestCancelActivityTaskDecisionAttributes .~ return rcatda +-- | Find a matching event type.+-- findEvent :: MonadDecisionCtx c m => EventType -> m (Maybe HistoryEvent) findEvent eventType = do   let f []     = return Nothing@@ -62,6 +75,8 @@   events <- view dcEvents   f events +-- | Begin workflow step.+-- begin :: MonadDecisionCtx c m => HistoryEvent -> m [Decision] begin event = do   traceInfo "begin" mempty@@ -70,6 +85,8 @@   let input = join $ view weseaInput <$> event ^. heWorkflowExecutionStartedEventAttributes   return [ scheduleActivity uid (task ^. tActivityType) (task ^. tTaskList) input ] +-- | Cancel workflow step.+-- cancel :: MonadDecisionCtx c m => m [Decision] cancel = do   traceInfo "cancel" mempty@@ -77,16 +94,22 @@   let uid = join $ fromText . view atseaActivityId <$> join (view heActivityTaskScheduledEventAttributes <$> event)   return [ maybe cancelActivity requestCancel uid ] +-- | Completed workflow step.+-- completed :: MonadDecisionCtx c m => m [Decision] completed = do   traceInfo "completed" mempty   return [ completeActivity ] +-- | Canceled workflow step.+-- canceled :: MonadDecisionCtx c m => m [Decision] canceled = do   traceInfo "canceled" mempty   return [ completeActivity ] +-- | Timed out workflow step.+-- timedout :: MonadDecisionCtx c m => m [Decision] timedout = do   traceInfo "timedout" mempty@@ -96,6 +119,8 @@   let input = join $ view weseaInput <$> join (view heWorkflowExecutionStartedEventAttributes <$> event)   return [ scheduleActivity uid (task ^. tActivityType) (task ^. tTaskList) input ] +-- | Failed workflow step.+-- failed :: MonadDecisionCtx c m => m [Decision] failed = do   traceInfo "failed" mempty
src/Network/AWS/Loup/Prelude.hs view
@@ -6,24 +6,32 @@ -- module Network.AWS.Loup.Prelude   ( module Exports+  , encode'+  , pages   , runConcurrent-  , runEvery   ) where -import Control.Concurrent import Control.Concurrent.Async.Lifted+import Control.Monad.Trans.AWS import Control.Monad.Trans.Control+import Data.Aeson+import Data.ByteString.Lazy            hiding (map)+import Data.Conduit+import Data.Conduit.List               hiding (map) import Preamble                        as Exports +-- | Encode JSON to Text.+--+encode' :: ToJSON a => a -> Text+encode' = decodeUtf8 . toStrict . encode++-- | Paginate all.+--+pages :: (AWSConstraint c m, AWSPager a) => a -> m [Rs a]+pages a = paginate a $$ consume+ -- | Run a list of actions concurrently. -- runConcurrent :: MonadBaseControl IO m => [m a] -> m () runConcurrent = void . runConcurrently . sequenceA . map Concurrently --- | Run an action every interval.----runEvery :: MonadIO m => Int -> m a -> m b-runEvery ms action =-  forever $ do-    liftIO $ threadDelay ms-    action
src/Network/AWS/Loup/Types.hs view
@@ -4,6 +4,7 @@   ( module Exports   ) where +import Network.AWS.Loup.Types.Alias   as Exports import Network.AWS.Loup.Types.Ctx     as Exports import Network.AWS.Loup.Types.Product as Exports import Network.AWS.Loup.Types.Sum     as Exports
+ src/Network/AWS/Loup/Types/Alias.hs view
@@ -0,0 +1,14 @@+{-# LANGUAGE NoImplicitPrelude #-}++-- | Various alias types.+--+module Network.AWS.Loup.Types.Alias+  ( module Network.AWS.Loup.Types.Alias+  ) where++import Data.Aeson+import Network.AWS.Loup.Prelude++-- | InputMap+--+type InputMap = HashMap Text (Maybe Value)
src/Network/AWS/Loup/Types/Product.hs view
@@ -5,10 +5,16 @@ -- module Network.AWS.Loup.Types.Product   ( module Network.AWS.Loup.Types.Product+  , ActivityType+  , atName+  , atVersion+  , TaskList+  , tlName   ) where  import Data.Aeson.TH import Network.AWS.Loup.Prelude+import Network.AWS.Loup.Types.Alias import Network.AWS.SWF  -- | Task@@ -38,3 +44,17 @@  $(makeLenses ''Plan) $(deriveJSON spinalOptions ''Plan)++-- | Pool+--+-- Pool of workers associated with a task.+--+data Pool = Pool+  { _pTask    :: Task+    -- ^ Workers task.+  , _pWorkers :: InputMap+    -- ^ Workers input map.+  } deriving (Show, Eq)++$(makeLenses ''Pool)+$(deriveJSON spinalOptions ''Pool)
src/Network/AWS/Loup/Types/Sum.hs view
@@ -15,6 +15,7 @@ data LabelType   = LabelAct   | LabelDecide+  | LabelConverge   deriving (Show, Eq)  $(deriveJSON spinalOptions ''LabelType)