wolf 0.2.0 → 0.2.1
raw patch · 20 files changed
+573/−376 lines, 20 filesdep +basic-preludedep +formattingdep +tastydep ~base
Dependencies added: basic-prelude, formatting, tasty, tasty-hunit, time
Dependency ranges changed: base
Files
- main/Act.hs +143/−0
- main/Decide.hs +34/−0
- main/Execute.hs +40/−0
- main/Options.hs +58/−0
- main/Register.hs +34/−0
- src/Network/AWS/Flow.hs +41/−19
- src/Network/AWS/Flow/Env.hs +4/−7
- src/Network/AWS/Flow/Logger.hs +43/−0
- src/Network/AWS/Flow/Prelude.hs +11/−0
- src/Network/AWS/Flow/S3.hs +37/−9
- src/Network/AWS/Flow/SWF.hs +3/−4
- src/Network/AWS/Flow/Types.hs +31/−35
- src/Network/AWS/Flow/Uid.hs +12/−0
- src/main/Act.hs +0/−146
- src/main/Decide.hs +0/−48
- src/main/Execute.hs +0/−57
- src/main/Register.hs +0/−45
- test/Test.hs +12/−0
- test/Test/Network/AWS/Flow.hs +29/−0
- wolf.cabal +41/−6
+ main/Act.hs view
@@ -0,0 +1,143 @@+{-# LANGUAGE ScopedTypeVariables #-}++module Act+ ( main+ ) where++import BasicPrelude hiding ( ByteString, (</>), hash, length, readFile, find )+import Control.Monad.Trans.Resource+import Data.Aeson.Encode+import Data.ByteString ( length )+import qualified Data.ByteString.Lazy as BL+import Data.Text ( pack, strip )+import Data.Text.Lazy ( toStrict )+import Data.Text.Lazy.Builder hiding ( fromText )+import Data.Yaml hiding ( Parser )+import Network.AWS.Data.Crypto+import Network.AWS.Flow+import Options+import Options.Applicative hiding ( action )+import Shelly hiding ( FilePath, bash )++data Args = Args+ { aConfig :: FilePath+ , aQueue :: Queue+ , aContainer :: FilePath+ , aContainerless :: Maybe String+ } deriving ( Eq, Read, Show )++args :: Parser Args+args = Args <$> configFile <*> (pack <$> queue) <*> containerFile <*> containerless++parser :: ParserInfo Args+parser =+ info ( helper <*> args ) $ fullDesc+ <> header "act: Workflow activity"+ <> progDesc "Workflow activity"++data Container = Container+ { cImage :: Text+ , cCommand :: Text+ , cVolumes :: [Text]+ , cDevices :: [Text]+ , cEnvironment :: [Text]+ , cLink :: [Text]+ } deriving ( Eq, Read, Show )++instance FromJSON Container where+ parseJSON (Object v) =+ Container <$>+ v .: "image" <*>+ v .: "command" <*>+ v .:? "volumes" .!= [] <*>+ v .:? "devices" .!= [] <*>+ v .:? "environment" .!= [] <*>+ v .:? "links" .!= []+ parseJSON _ = mzero++data Control = Control+ { cUid :: Uid+ } deriving ( Eq, Read, Show )++instance ToJSON Control where+ toJSON Control{..} = object+ [ "run_uid" .= cUid+ ]++encodeText :: ToJSON a => a -> Text+encodeText = toStrict . toLazyText . encodeToTextBuilder . toJSON++exec :: MonadIO m => Container -> Maybe String -> Uid -> Metadata -> [Blob] -> m (Metadata, [Artifact])+exec container dockerless uid metadata blobs =+ shelly $ withDir $ \dir dataDir storeDir -> do+ control $ dataDir </> pack "control.json"+ storeInput $ storeDir </> pack "input"+ dataInput $ dataDir </> pack "input.json"+ maybe (docker dataDir storeDir container) (bash dir container) dockerless+-- if dockerless then+-- bash dir container+-- else+-- docker dataDir storeDir container+ result <- dataOutput $ dataDir </> pack "output.json"+ artifacts <- storeOutput $ storeDir </> pack "output"+ return (result, artifacts) where+ withDir action =+ withTmpDir $ \dir -> do+ mkdir $ dir </> pack "data"+ mkdir $ dir </> pack "store"+ mkdir $ dir </> pack "store/input"+ mkdir $ dir </> pack "store/output"+ action dir (dir </> pack "data") (dir </> pack "store")+ control file =+ writefile file $ encodeText $ Control uid+ dataInput file =+ maybe (return ()) (writefile file) metadata+ dataOutput file =+ catch_sh_maybe (readfile file) where+ catch_sh_maybe action =+ catch_sh (liftM Just action) $ \(_ :: SomeException) -> return Nothing+ storeInput dir =+ forM_ blobs $ \(key, blob) -> do+ paths <- liftM strip $ run "dirname" [key]+ mkdir_p $ dir </> paths+ writeBinary (dir </> key) (BL.toStrict blob)+ storeOutput dir = do+ artifacts <- findWhen test_f dir+ forM artifacts $ \artifact -> do+ key <- relativeTo dir artifact+ blob <- readBinary artifact+ return ( toTextIgnore key+ , hash blob+ , fromIntegral $ length blob+ , BL.fromStrict blob+ )+ docker dataDir storeDir Container{..} = do+ devices <- forM cDevices $ \device ->+ liftM strip $ run "readlink" ["-f", device]+ run_ "docker" $ concat+ [["run"]+ , concatMap (("--device" :) . return) devices+ , concatMap (("--env" :) . return) cEnvironment+ , concatMap (("--link" :) . return) cLink+ , concatMap (("--volume" :) . return) $+ toTextIgnore dataDir <> ":/app/data" :+ toTextIgnore storeDir <> ":/app/store" : cVolumes+ , [cImage]+ , words cCommand+ ]+ bash dir Container{..} bashDir = do+ files <- ls $ fromText $ pack bashDir+ forM_ files $ flip cp_r dir+ cd dir+ maybe (return ()) (uncurry $ run_ . fromText) $ uncons $ words cCommand++call :: Args -> IO ()+call Args{..} = do+ config <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")+ container <- decodeFile aContainer >>= maybeThrow (userError "Bad Container")+ env <- flowEnv config+ forever $ runResourceT $ runFlowT env $+ act aQueue $ exec container aContainerless++main :: IO ()+main = execParser parser >>= call
+ main/Decide.hs view
@@ -0,0 +1,34 @@+module Decide+ ( main+ ) where++import BasicPrelude+import Control.Monad.Trans.Resource+import Data.Yaml hiding ( Parser )+import Network.AWS.Flow+import Options+import Options.Applicative++data Args = Args+ { aConfig :: FilePath+ , aPlan :: FilePath+ } deriving ( Eq, Read, Show )++args :: Parser Args+args = Args <$> configFile <*> planFile++parser :: ParserInfo Args+parser =+ info ( helper <*> args ) $ fullDesc+ <> header "decide: Decide a workflow"+ <> progDesc "Decide a workflow"++call :: Args -> IO ()+call Args{..} = do+ config <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")+ plan <- decodeFile aPlan >>= maybeThrow (userError "Bad Plan")+ env <- flowEnv config+ forever $ runResourceT $ runFlowT env $ decide plan++main :: IO ()+main = execParser parser >>= call
+ main/Execute.hs view
@@ -0,0 +1,40 @@+module Execute+ ( main+ ) where++import BasicPrelude hiding ( readFile )+import Control.Monad.Trans.Resource+import Data.Text.IO+import Data.Yaml hiding ( Parser )+import Network.AWS.Flow+import Options+import Options.Applicative++data Args = Args+ { aConfig :: FilePath+ , aPlan :: FilePath+ , aInput :: Maybe FilePath+ } deriving ( Eq, Read, Show )++args :: Parser Args+args = Args <$> configFile <*> planFile <*> inputFile++parser :: ParserInfo Args+parser =+ info ( helper <*> args ) $ fullDesc+ <> header "execute: Execute a workflow"+ <> progDesc "Execute a workflow"++call :: Args -> IO ()+call Args{..} = do+ config <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")+ plan <- decodeFile aPlan >>= maybeThrow (userError "Bad Plan")+ input <- readFileMaybe aInput+ env <- flowEnv config+ runResourceT $ runFlowT env $+ execute (strtTask $ plnStart plan) input where+ readFileMaybe =+ maybe (return Nothing) ((>>= return . Just) . readFile)++main :: IO ()+main = execParser parser >>= call
+ main/Options.hs view
@@ -0,0 +1,58 @@+module Options+ ( configFile+ , planFile+ , inputFile+ , containerFile+ , queue+ , containerless+ ) where++import BasicPrelude+import Options.Applicative++configFile :: Parser String+configFile =+ strOption+ $ long "config"+ <> short 'c'+ <> metavar "FILE"+ <> help "AWS SWF Service Flow config"++planFile :: Parser String+planFile =+ strOption+ $ long "plan"+ <> short 'p'+ <> metavar "FILE"+ <> help "AWS SWF Service Flow plan"++inputFile :: Parser (Maybe String)+inputFile =+ optional $ strOption+ $ long "input"+ <> short 'i'+ <> metavar "FILE"+ <> help "AWS SWF Service Flow input"++containerFile :: Parser String+containerFile =+ strOption+ $ long "container"+ <> short 'x'+ <> metavar "FILE"+ <> help "AWS SWF Service Flow worker container"++queue :: Parser String+queue =+ strOption+ $ long "queue"+ <> short 'q'+ <> metavar "NAME"+ <> help "AWS SWF Service Flow queue"++containerless :: Parser (Maybe String)+containerless =+ optional $ strOption+ $ long "containerless"+ <> metavar "DIR"+ <> help "Run outside of container in directory"
+ main/Register.hs view
@@ -0,0 +1,34 @@+module Register+ ( main+ ) where++import BasicPrelude+import Control.Monad.Trans.Resource hiding ( register )+import Data.Yaml hiding ( Parser )+import Network.AWS.Flow+import Options+import Options.Applicative++data Args = Args+ { aConfig :: FilePath+ , aPlan :: FilePath+ } deriving ( Eq, Read, Show )++args :: Parser Args+args = Args <$> configFile <*> planFile++parser :: ParserInfo Args+parser =+ info ( helper <*> args ) $ fullDesc+ <> header "register: Register a workflow"+ <> progDesc "Register a workflow"++call :: Args -> IO ()+call Args{..} = do+ config' <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")+ plan <- decodeFile aPlan >>= maybeThrow (userError "Bad Plan")+ env <- flowEnv config'+ runResourceT $ runFlowT env $ register plan++main :: IO ()+main = execParser parser >>= call
src/Network/AWS/Flow.hs view
@@ -5,11 +5,15 @@ , decide , flowEnv , runFlowT+ , runDecide+ , nextEvent+ , select , maybeThrow , Uid , Queue , Metadata , Artifact+ , Blob , Task (..) , Timer (..) , Start (..)@@ -18,24 +22,25 @@ , Plan (..) ) where -import Control.Lens-import Control.Monad.Catch-import Control.Monad.Reader-import qualified Data.HashMap.Strict as Map-import Data.List-import Data.UUID-import Data.UUID.V4-import Network.AWS.SWF import Network.AWS.Flow.Env+import Network.AWS.Flow.Logger import Network.AWS.Flow.S3 import Network.AWS.Flow.SWF import Network.AWS.Flow.Types+import Network.AWS.Flow.Uid+import Network.AWS.Flow.Prelude hiding ( ByteString, Metadata )++import Control.Monad.Catch+import qualified Data.HashMap.Strict as Map+import Formatting+import Network.AWS.SWF import Safe -- Interface register :: MonadFlow m => Plan -> m () register Plan{..} = do+ logInfo' "event=register" r <- registerDomainAction s <- registerWorkflowTypeAction (tskName $ strtTask plnStart)@@ -52,18 +57,28 @@ execute :: MonadFlow m => Task -> Metadata -> m () execute Task{..} input = do- uid <- liftIO $ newUid+ uid <- newUid+ logInfo' $ sformat ("event=execute uid=" % stext) uid startWorkflowExecutionAction uid tskName tskVersion tskQueue input -act :: MonadFlow m => Queue -> (Uid -> Metadata -> m (Metadata, [Artifact])) -> m ()+act :: MonadFlow m => Queue -> (Uid -> Metadata -> [Blob] -> m (Metadata, [Artifact])) -> m () act queue action = do+ logInfo' "event=act" (token, uid, input) <- pollForActivityTaskAction queue- (output, artifacts) <- action uid input- forM_ artifacts putObjectAction+ logInfo' $ sformat ("event=act-begin uid=" % stext) uid+ keys <- listObjectsAction uid+ unless (null keys) $ logInfo' $ sformat ("event=list-blobs uid=" % stext) uid+ blobs <- forM keys $ getObjectAction uid+ unless (null blobs) $ logInfo' $ sformat ("event=blobs uid=" % stext) uid+ (output, artifacts) <- action uid input blobs+ logInfo' $ sformat ("event=act-finish uid=" % stext) uid+ forM_ artifacts $ putObjectAction uid+ unless (null artifacts) $ logInfo' $ sformat ("event=artifacts uid=" % stext) uid respondActivityTaskCompletedAction token output decide :: MonadFlow m => Plan -> m () decide plan@Plan{..} = do+ logInfo' "event=decide" (token', events) <- pollForDecisionTaskAction (tskQueue $ strtTask plnStart) token <- maybeThrow (userError "No Token") token' logger <- asks feLogger@@ -73,8 +88,8 @@ -- Decisions runDecide :: Log -> Plan -> [HistoryEvent] -> DecideT m a -> m a-runDecide logger plan events action =- runDecideT env action where+runDecide logger plan events =+ runDecideT env where env = DecideEnv logger plan events findEvent where findEvent = flip Map.lookup $ Map.fromList $ flip map events $ \e ->@@ -115,6 +130,7 @@ start :: MonadDecide m => HistoryEvent -> m [Decision] start event = do+ logInfo' "event=start" input <- maybeThrow (userError "No Start Information") $ do attrs <- event ^. heWorkflowExecutionStartedEventAttributes return $ attrs ^. weseaInput@@ -123,6 +139,7 @@ completed :: MonadDecide m => HistoryEvent -> m [Decision] completed event = do+ logInfo' "event=completed" findEvent <- asks deFindEvent (input, name) <- maybeThrow (userError "No Completed Information") $ do attrs <- event ^. heActivityTaskCompletedEventAttributes@@ -134,6 +151,7 @@ timer :: MonadDecide m => HistoryEvent -> m [Decision] timer event = do+ logInfo' "event=timer" findEvent <- asks deFindEvent name <- maybeThrow (userError "No Timer Information") $ do attrs <- event ^. heTimerFiredEventAttributes@@ -148,6 +166,7 @@ timerStart :: MonadDecide m => HistoryEvent -> Name -> m [Decision] timerStart event name = do+ logInfo' $ sformat ("event=timer-start name=" % stext) name input <- maybeThrow (userError "No Timer Start Information") $ do attrs <- event ^. heWorkflowExecutionStartedEventAttributes return $ attrs ^. weseaInput@@ -156,6 +175,7 @@ timerCompleted :: MonadDecide m => HistoryEvent -> Name -> m [Decision] timerCompleted event name = do+ logInfo' $ sformat ("event=timer-completed name=" % stext) name input <- maybeThrow (userError "No Timer Completed Information") $ do attrs <- event ^. heActivityTaskCompletedEventAttributes return $ attrs ^. atceaResult@@ -167,7 +187,8 @@ scheduleSpec :: MonadDecide m => Metadata -> Spec -> m [Decision] scheduleSpec input spec = do- uid <- liftIO $ newUid+ uid <- newUid+ logInfo' $ sformat ("event=schedule-spec uid=" % stext) uid case spec of Work{..} -> return [scheduleActivityTaskDecision uid@@ -182,6 +203,7 @@ scheduleEnd :: MonadDecide m => Metadata -> m [Decision] scheduleEnd input = do+ logInfo' "event=schedule-end" end <- asks (plnEnd . dePlan) case end of Stop -> return [completeWorkflowExecutionDecision input]@@ -189,11 +211,12 @@ scheduleContinue :: MonadDecide m => m [Decision] scheduleContinue = do+ logInfo' "event=schedule-continue" event <- nextEvent [WorkflowExecutionStarted] input <- maybeThrow (userError "No Continue Start Information") $ do attrs <- event ^. heWorkflowExecutionStartedEventAttributes return $ attrs ^. weseaInput- uid <- liftIO $ newUid+ uid <- newUid task <- asks (strtTask . plnStart . dePlan) return [startChildWorkflowExecutionDecision uid (tskName task)@@ -211,6 +234,7 @@ childStart :: MonadDecide m => HistoryEvent -> m [Decision] childStart event = do+ logInfo' "event=child-start" input <- maybeThrow (userError "No Child Start Information") $ do attrs <- event ^. heWorkflowExecutionStartedEventAttributes return $ attrs ^. weseaInput@@ -218,6 +242,7 @@ childCompleted :: MonadDecide m => HistoryEvent -> m [Decision] childCompleted event = do+ logInfo' "event=child-completed" input <- maybeThrow (userError "No Child Completed Information") $ do attrs <- event ^. heActivityTaskCompletedEventAttributes return $ attrs ^. atceaResult@@ -227,6 +252,3 @@ maybeThrow :: (MonadThrow m, Exception e) => e -> Maybe a -> m a maybeThrow e = maybe (throwM e) return--newUid :: IO Uid-newUid = toText <$> nextRandom
src/Network/AWS/Flow/Env.hs view
@@ -2,17 +2,14 @@ ( flowEnv ) where -import Control.Lens-import Control.Monad.Trans.AWS+import Network.AWS.Flow.Logger+import Network.AWS.Flow.Prelude import Network.AWS.Flow.Types-import System.Log.FastLogger+ import System.IO flowEnv :: FlowConfig -> IO FlowEnv flowEnv FlowConfig{..} = do- loggerSet <- newStderrLoggerSet defaultBufSize logger <- newLogger Info stderr env <- newEnv fcRegion fcCredentials <&> envLogger .~ logger- return $ FlowEnv (logStrLn loggerSet) env (fromIntegral fcTimeout) (fromIntegral fcPollTimeout) fcDomain fcBucket fcPrefix where- logStrLn ls s =- pushLogStr ls s >> flushLogStr ls+ return $ FlowEnv logStrLn env (fromIntegral fcTimeout) (fromIntegral fcPollTimeout) fcDomain fcBucket fcPrefix
+ src/Network/AWS/Flow/Logger.hs view
@@ -0,0 +1,43 @@+module Network.AWS.Flow.Logger+ ( logStrLn+ , logDebug'+ , logInfo'+ , logWarn'+ , logError'+ ) where++import Network.AWS.Flow.Prelude++import Control.Monad.Logger+import Data.Time.Clock+import Data.Time.Format+import Data.Version+import Formatting hiding ( now )+import Paths_wolf+import System.Log.FastLogger++prefix :: IO LogStr+prefix = do+ now <- getCurrentTime+ return $ toLogStr $ sformat (string % " name=wolf v=" % string % " ")+ (formatTime defaultTimeLocale "%FT%T%z" now)+ (showVersion version)++logStrLn :: LogStr -> IO ()+logStrLn s = do+ loggerSet <- newStderrLoggerSet defaultBufSize+ p <- prefix+ mapM_ (pushLogStr loggerSet) [p, s, "\n"]+ flushLogStr loggerSet++logDebug' :: MonadLogger m => Text -> m ()+logDebug' = logDebugN++logInfo' :: MonadLogger m => Text -> m ()+logInfo' = logInfoN++logWarn' :: MonadLogger m => Text -> m ()+logWarn' = logWarnN++logError' :: MonadLogger m => Text -> m ()+logError' = logErrorN
+ src/Network/AWS/Flow/Prelude.hs view
@@ -0,0 +1,11 @@+module Network.AWS.Flow.Prelude+ ( module BasicPrelude+ , module Control.Lens+ , module Control.Monad.Reader+ , module Control.Monad.Trans.AWS+ ) where++import BasicPrelude hiding ( (<.>), uncons )+import Control.Lens+import Control.Monad.Reader+import Control.Monad.Trans.AWS
src/Network/AWS/Flow/S3.hs view
@@ -1,21 +1,49 @@ module Network.AWS.Flow.S3- ( putObjectAction+ ( listObjectsAction+ , getObjectAction+ , putObjectAction ) where -import Network.AWS.Data.Body-import Control.Monad.Reader-import Control.Monad.Trans.AWS-import Data.Conduit.Binary-import Data.Monoid+import Network.AWS.Flow.Prelude hiding ( ByteString, hash, stripPrefix ) import Network.AWS.Flow.Types++import Data.Conduit+import Data.Conduit.List hiding ( concatMap, map, mapMaybe )+import Data.Conduit.Binary+import Data.Text hiding ( concatMap, map )+import Network.AWS.Data.Body+import Network.AWS.Data.Text import Network.AWS.S3 -- Actions -putObjectAction :: MonadFlow m => Artifact -> m ()-putObjectAction (key, hash, size, blob) = do+listObjectsAction :: MonadFlow m => Uid -> m [Text]+listObjectsAction uid = do timeout' <- asks feTimeout bucket' <- asks feBucket prefix <- asks fePrefix+ timeout timeout' $ do+ rs <- paginate (listObjects (BucketName bucket') &+ loPrefix .~ Just (prefix <> "/" <> uid))+ $$ consume+ return $+ mapMaybe (stripPrefix (prefix <> "/" <> uid <> "/") . toText . (^. oKey)) $+ concatMap (^. lorsContents) rs++getObjectAction :: MonadFlow m => Uid -> Text -> m Blob+getObjectAction uid key = do+ timeout' <- asks feTimeout+ bucket' <- asks feBucket+ prefix <- asks fePrefix+ timeout timeout' $ do+ r <- send $ getObject (BucketName bucket') (ObjectKey $ prefix <> "/" <> uid <> "/" <> key)+ blob <- sinkBody (r ^. gorsBody) sinkLbs+ return (key, blob)++putObjectAction :: MonadFlow m => Uid -> Artifact -> m ()+putObjectAction uid (key, hash, size, blob) = do+ timeout' <- asks feTimeout+ bucket' <- asks feBucket+ prefix <- asks fePrefix void $ timeout timeout' $- send $ putObject (BucketName $ bucket') (ObjectKey $ prefix <> "/" <> key) (Hashed $ hashedBody hash size $ sourceLbs blob)+ send $ putObject (BucketName bucket') (ObjectKey $ prefix <> "/" <> uid <> "/" <> key) (Hashed $ hashedBody hash size $ sourceLbs blob)
src/Network/AWS/Flow/SWF.hs view
@@ -15,12 +15,11 @@ , startChildWorkflowExecutionDecision ) where -import Control.Lens-import Control.Monad.Reader-import Control.Monad.Trans.AWS hiding ( Metadata )+import Network.AWS.Flow.Prelude hiding ( Metadata )+import Network.AWS.Flow.Types+ import Data.Conduit import Data.Conduit.List hiding ( concatMap )-import Network.AWS.Flow.Types import Network.AWS.SWF import Safe
src/Network/AWS/Flow/Types.hs view
@@ -7,19 +7,15 @@ module Network.AWS.Flow.Types where -import Control.Lens+import Network.AWS.Flow.Prelude hiding ( ByteString, catch )+ import Control.Monad.Base import Control.Monad.Catch-import Control.Monad.Except import Control.Monad.Logger-import Control.Monad.Reader import Control.Monad.Trans.Control import Control.Monad.Trans.Resource-import Control.Monad.Trans.AWS import Data.Aeson import Data.ByteString.Lazy-import Data.Conduit.Lazy-import Data.Text import Network.AWS.Data.Crypto import Network.AWS.SWF.Types @@ -31,6 +27,7 @@ type Timeout = Text type Metadata = Maybe Text type Artifact = (Text, Digest SHA256, Integer, ByteString)+type Blob = (Text, ByteString) type Log = LogStr -> IO () data FlowConfig = FlowConfig@@ -91,19 +88,19 @@ } newtype FlowT m a = FlowT- { unFlowT :: AWST' FlowEnv m a+ { unFlowT :: LoggingT (AWST' FlowEnv m) a } deriving ( Functor , Applicative , Monad , MonadIO- , MonadActive- , MonadTrans+ , MonadLogger ) type MonadFlow m = ( MonadCatch m , MonadThrow m , MonadResource m+ , MonadLogger m , MonadReader FlowEnv m ) @@ -116,12 +113,19 @@ instance MonadBase b m => MonadBase b (FlowT m) where liftBase = liftBaseDefault +instance MonadTrans FlowT where+ lift = FlowT . lift . lift+ instance MonadTransControl FlowT where type StT FlowT a = StT (ReaderT FlowEnv) a - liftWith = defaultLiftWith FlowT unFlowT- restoreT = defaultRestoreT FlowT+ liftWith f = FlowT $+ liftWith $ \g ->+ liftWith $ \h ->+ f (h . g . unFlowT) + restoreT = FlowT . restoreT . restoreT+ instance MonadBaseControl b m => MonadBaseControl b (FlowT m) where type StM (FlowT m) a = ComposeSt FlowT m a @@ -131,10 +135,6 @@ instance MonadResource m => MonadResource (FlowT m) where liftResourceT = lift . liftResourceT -instance MonadError e m => MonadError e (FlowT m) where- throwError = lift . throwError- catchError m f = FlowT (catchError (unFlowT m) (unFlowT . f))- instance Monad m => MonadReader FlowEnv (FlowT m) where ask = FlowT ask local f = FlowT . local f . unFlowT@@ -144,7 +144,7 @@ environment = lens feEnv (\s a -> s { feEnv = a }) runFlowT :: FlowEnv -> FlowT m a -> m a-runFlowT e (FlowT m) = runAWST e m+runFlowT e (FlowT m) = runAWST e (runLoggingT m (const . const . const $ feLogger e)) data DecideEnv = DecideEnv { deLogger :: Log@@ -154,57 +154,53 @@ } newtype DecideT m a = DecideT- { unDecideT :: ReaderT DecideEnv m a+ { unDecideT :: LoggingT (ReaderT DecideEnv m) a } deriving ( Functor , Applicative , Monad , MonadIO- , MonadActive- , MonadTrans+ , MonadLogger ) type MonadDecide m =- ( MonadCatch m- , MonadThrow m- , MonadResource m+ ( MonadThrow m+ , MonadLogger m+ , MonadIO m , MonadReader DecideEnv m ) instance MonadThrow m => MonadThrow (DecideT m) where throwM = lift . throwM -instance MonadCatch m => MonadCatch (DecideT m) where- catch (DecideT m) f = DecideT (catch m (unDecideT . f))- instance MonadBase b m => MonadBase b (DecideT m) where liftBase = liftBaseDefault +instance MonadTrans DecideT where+ lift = DecideT . lift . lift+ instance MonadTransControl DecideT where type StT DecideT a = StT (ReaderT DecideEnv) a - liftWith = defaultLiftWith DecideT unDecideT- restoreT = defaultRestoreT DecideT+ liftWith f = DecideT $+ liftWith $ \g ->+ liftWith $ \h ->+ f (h . g . unDecideT) + restoreT = DecideT . restoreT . restoreT+ instance MonadBaseControl b m => MonadBaseControl b (DecideT m) where type StM (DecideT m) a = ComposeSt DecideT m a liftBaseWith = defaultLiftBaseWith restoreM = defaultRestoreM -instance MonadResource m => MonadResource (DecideT m) where- liftResourceT = lift . liftResourceT--instance MonadError e m => MonadError e (DecideT m) where- throwError = lift . throwError- catchError m f = DecideT (catchError (unDecideT m) (unDecideT . f))- instance Monad m => MonadReader DecideEnv (DecideT m) where ask = DecideT ask local f = DecideT . local f . unDecideT reader = DecideT . reader runDecideT :: DecideEnv -> DecideT m a -> m a-runDecideT e (DecideT m) = runReaderT m e+runDecideT e (DecideT m) = runReaderT (runLoggingT m (const . const . const $ deLogger e)) e data Task = Task { tskName :: Name
+ src/Network/AWS/Flow/Uid.hs view
@@ -0,0 +1,12 @@+module Network.AWS.Flow.Uid+ ( newUid+ ) where++import Network.AWS.Flow.Prelude+import Network.AWS.Flow.Types++import Data.UUID+import Data.UUID.V4++newUid :: MonadIO m => m Uid+newUid = toText <$> liftIO nextRandom
− src/main/Act.hs
@@ -1,146 +0,0 @@-{-# LANGUAGE ScopedTypeVariables #-}--module Act- ( main- ) where--import Control.Exception-import Control.Monad-import Control.Monad.Trans.Resource-import Control.Monad.IO.Class-import Data.Aeson.Encode-import Data.ByteString ( length )-import Data.ByteString.Lazy ( fromStrict )-import Data.Text ( Text, pack, append, words, strip )-import Data.Text.Lazy ( toStrict )-import Data.Text.Lazy.Builder-import Data.Yaml-import Network.AWS.Data.Crypto-import Network.AWS.Flow-import Options.Applicative hiding ( action )-import Shelly hiding ( FilePath )-import Prelude hiding ( length, readFile, words, writeFile )--data Args = Args- { aConfig :: FilePath- , aQueue :: Queue- , aContainer :: FilePath- } deriving ( Eq, Read, Show )--argsPI :: ParserInfo Args-argsPI =- info ( helper <*> argsP )- ( fullDesc- <> header "act: Workflow activity"- <> progDesc "Workflow activity" ) where- argsP = args- <$> strOption- ( long "config"- <> short 'c'- <> metavar "FILE"- <> help "AWS SWF Service Flow config" )- <*> strOption- ( long "queue"- <> short 'q'- <> metavar "NAME"- <> help "AWS SWF Service Flow queue" )- <*> strOption- ( long "container"- <> short 'x'- <> metavar "FILE"- <> help "AWS SWF Service Flow worker container" ) where- args config queue container = Args- { aConfig = config- , aQueue = pack queue- , aContainer = container- }--data Container = Container- { cImage :: Text- , cCommand :: Text- , cVolumes :: [Text]- , cDevices :: [Text]- , cEnvironment :: [Text]- } deriving ( Eq, Read, Show )--instance FromJSON Container where- parseJSON (Object v) =- Container <$>- v .: "image" <*>- v .: "command" <*>- v .:? "volumes" .!= [] <*>- v .:? "devices" .!= [] <*>- v .:? "environment" .!= []- parseJSON _ = mzero--data Control = Control- { cUid :: Uid- } deriving ( Eq, Read, Show )--instance ToJSON Control where- toJSON Control{..} = object- [ "run_uid" .= cUid- ]--encodeText :: ToJSON a => a -> Text-encodeText =- toStrict . toLazyText . encodeToTextBuilder . toJSON--exec :: MonadIO m => Container -> Uid -> Metadata -> m (Metadata, [Artifact])-exec container uid metadata =- shelly $ withDir $ \dataDir storeDir -> do- control dataDir $ encodeText $ Control uid- input dataDir metadata- docker dataDir storeDir container- result <- output dataDir- artifacts <- store storeDir- return (result, artifacts) where- withDir action =- withTmpDir $ \dir -> do- mkdir $ dir </> pack "data"- mkdir $ dir </> pack "store"- action (dir </> pack "data") (dir </> pack "store")- control dir =- writefile (dir </> pack "control.json")- input dir =- maybe_ (writefile $ dir </> pack "input.json") where- maybe_ =- maybe (return ())- output dir =- catch_sh_maybe (readfile $ dir </> pack "output.json") where- catch_sh_maybe action =- catch_sh (liftM Just action) $ \(_ :: SomeException) -> return Nothing- store dir = do- artifacts <- findWhen test_f dir- forM artifacts $ \artifact -> do- key <- relativeTo dir artifact- blob <- readBinary artifact- return ( toTextIgnore $ uid </> key- , hash blob- , fromIntegral $ length blob- , fromStrict blob- )- docker dataDir storeDir Container{..} = do- devices <- forM cDevices $ \device ->- liftM strip $ run "readlink" ["-f", device]- run_ "docker" $ concat- [["run"]- , concatMap (("--device" :) . return) devices- , concatMap (("--env" :) . return) cEnvironment- , concatMap (("--volume" :) . return) $- append (toTextIgnore dataDir) ":/app/data" :- append (toTextIgnore storeDir) ":/app/store" : cVolumes- , [cImage]- , words cCommand- ]--main :: IO ()-main =- execParser argsPI >>= call where- call Args{..} = do- config <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")- container <- decodeFile aContainer >>= maybeThrow (userError "Bad Container")- env <- flowEnv config- forever $- runResourceT $ runFlowT env $- act aQueue $ exec container
− src/main/Decide.hs
@@ -1,48 +0,0 @@-module Decide- ( main- ) where--import Control.Monad-import Control.Monad.Trans.Resource-import Data.Yaml-import Network.AWS.Flow-import Options.Applicative-import Prelude hiding ( readFile )--data Args = Args- { aConfig :: FilePath- , aPlan :: FilePath- } deriving ( Eq, Read, Show )--argsPI :: ParserInfo Args-argsPI =- info ( helper <*> argsP )- ( fullDesc- <> header "decide: Decide a workflow"- <> progDesc "Decide a workflow" ) where- argsP = args- <$> strOption- ( long "config"- <> short 'c'- <> metavar "FILE"- <> help "AWS SWF Service Flow config" )- <*> strOption- ( long "plan"- <> short 'p'- <> metavar "FILE"- <> help "AWS SWF Service Flow plan" ) where- args config plan = Args- { aConfig = config- , aPlan = plan- }--main :: IO ()-main = do- execParser argsPI >>= call where- call Args{..} = do- config <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")- plan <- decodeFile aPlan >>= maybeThrow (userError "Bad Plan")- env <- flowEnv config- forever $- runResourceT $ runFlowT env $- decide plan
− src/main/Execute.hs
@@ -1,57 +0,0 @@-module Execute- ( main- ) where--import Control.Monad.Trans.Resource-import Data.Text.IO-import Data.Yaml-import Network.AWS.Flow-import Options.Applicative-import Prelude hiding ( readFile )--data Args = Args- { aConfig :: FilePath- , aPlan :: FilePath- , aInput :: Maybe FilePath- } deriving ( Eq, Read, Show )--argsPI :: ParserInfo Args-argsPI =- info ( helper <*> argsP )- ( fullDesc- <> header "execute: Execute a workflow"- <> progDesc "Execute a workflow" ) where- argsP = args- <$> strOption- ( long "config"- <> short 'c'- <> metavar "FILE"- <> help "AWS SWF Service Flow config" )- <*> strOption- ( long "plan"- <> short 'p'- <> metavar "FILE"- <> help "AWS SWF Service Flow plan" )- <*> optional ( strOption- ( long "input"- <> short 'i'- <> metavar "FILE"- <> help "AWS SWF Service Flow input" ) ) where- args config plan input = Args- { aConfig = config- , aPlan = plan- , aInput = input- }--main :: IO ()-main =- execParser argsPI >>= call where- call Args{..} = do- config <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")- plan <- decodeFile aPlan >>= maybeThrow (userError "Bad Plan")- input <- readFileMaybe aInput- env <- flowEnv config- runResourceT $ runFlowT env $- execute (strtTask $ plnStart plan) input where- readFileMaybe =- maybe (return Nothing) ((>>= return . Just) . readFile)
− src/main/Register.hs
@@ -1,45 +0,0 @@-module Register- ( main- ) where--import Control.Monad.Trans.Resource hiding ( register )-import Data.Yaml-import Network.AWS.Flow-import Options.Applicative--data Args = Args- { aConfig :: FilePath- , aPlan :: FilePath- } deriving ( Eq, Read, Show )--argsPI :: ParserInfo Args-argsPI =- info ( helper <*> argsP )- ( fullDesc- <> header "register: Register a workflow"- <> progDesc "Register a workflow" ) where- argsP = args- <$> strOption- ( long "config"- <> short 'c'- <> metavar "FILE"- <> help "AWS SWF Service Flow config" )- <*> strOption- ( long "plan"- <> short 'p'- <> metavar "FILE"- <> help "AWS SWF Service Flow plan" ) where- args config plan = Args- { aConfig = config- , aPlan = plan- }--main :: IO ()-main = do- execParser argsPI >>= call where- call Args{..} = do- config <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")- plan <- decodeFile aPlan >>= maybeThrow (userError "Bad Plan")- env <- flowEnv config- runResourceT $ runFlowT env $- register plan
+ test/Test.hs view
@@ -0,0 +1,12 @@+import BasicPrelude+import qualified Test.Network.AWS.Flow as Flow+import Test.Tasty++tests :: TestTree+tests =+ testGroup "Tests"+ [ Flow.tests+ ]++main :: IO ()+main = defaultMain tests
+ test/Test/Network/AWS/Flow.hs view
@@ -0,0 +1,29 @@+module Test.Network.AWS.Flow+ ( tests+ ) where++import Network.AWS.Flow++import BasicPrelude+import Test.Tasty+import Test.Tasty.HUnit++assertUserError :: String -> IO a -> IO ()+assertUserError s action =+ handleJust check (const $ return ()) $ do+ void $ action+ assertFailure $ "missed user error: " ++ s where+ check e = guard $ userError s == e++testSelect :: TestTree+testSelect =+ testGroup "Select tests"+ [ testCase "No events" $+ assertUserError "No Next Event" $ runDecide undefined undefined [] select+ ]++tests :: TestTree+tests =+ testGroup "Flow tests"+ [ testSelect+ ]
wolf.cabal view
@@ -1,5 +1,5 @@ name: wolf-version: 0.2.0+version: 0.2.1 synopsis: Amazon Simple Workflow Service Wrapper. homepage: https://github.com/swift-nav/wolf license: MIT@@ -22,9 +22,13 @@ library exposed-modules: Network.AWS.Flow other-modules: Network.AWS.Flow.Env+ , Network.AWS.Flow.Logger+ , Network.AWS.Flow.Prelude , Network.AWS.Flow.S3 , Network.AWS.Flow.SWF , Network.AWS.Flow.Types+ , Network.AWS.Flow.Uid+ , Paths_wolf default-language: Haskell2010 hs-source-dirs: src ghc-options: -Wall -fno-warn-orphans@@ -34,11 +38,13 @@ , amazonka-s3 , amazonka-swf , base >= 4.7 && < 5+ , basic-prelude , bytestring , conduit , conduit-extra , exceptions , fast-logger+ , formatting , http-conduit , lens , monad-control@@ -49,6 +55,7 @@ , resourcet , safe , text+ , time , transformers , transformers-base , unordered-containers@@ -57,13 +64,31 @@ default-extensions: OverloadedStrings FlexibleContexts RecordWildCards+ NoImplicitPrelude +test-suite test+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: Test.hs+ other-modules: Test.Network.AWS.Flow+ build-depends: base+ , basic-prelude+ , tasty+ , tasty-hunit+ , wolf+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall+ default-language: Haskell2010+ default-extensions: NoImplicitPrelude+ OverloadedStrings+ executable wolf-register default-language: Haskell2010 main-is: Register.hs- hs-source-dirs: src/main+ other-modules: Options+ hs-source-dirs: main ghc-options: -Wall -main-is Register build-depends: base+ , basic-prelude , optparse-applicative , resourcet , text@@ -71,13 +96,16 @@ , yaml default-extensions: OverloadedStrings RecordWildCards+ NoImplicitPrelude executable wolf-execute default-language: Haskell2010 main-is: Execute.hs- hs-source-dirs: src/main+ other-modules: Options+ hs-source-dirs: main ghc-options: -Wall -main-is Execute build-depends: base+ , basic-prelude , optparse-applicative , resourcet , text@@ -85,13 +113,16 @@ , yaml default-extensions: OverloadedStrings RecordWildCards+ NoImplicitPrelude executable wolf-decide default-language: Haskell2010 main-is: Decide.hs- hs-source-dirs: src/main+ other-modules: Options+ hs-source-dirs: main ghc-options: -Wall -main-is Decide build-depends: base+ , basic-prelude , optparse-applicative , resourcet , text@@ -99,22 +130,26 @@ , yaml default-extensions: OverloadedStrings RecordWildCards+ NoImplicitPrelude executable wolf-act default-language: Haskell2010 main-is: Act.hs- hs-source-dirs: src/main+ other-modules: Options+ hs-source-dirs: main ghc-options: -Wall -main-is Act build-depends: aeson , amazonka-core , base+ , basic-prelude , bytestring , optparse-applicative- , shelly , resourcet+ , shelly , text , transformers , wolf , yaml default-extensions: OverloadedStrings RecordWildCards+ NoImplicitPrelude