wolf 0.2.2 → 0.2.3
raw patch · 4 files changed
+87/−39 lines, 4 filesdep +system-filepathdep +zlib
Dependencies added: system-filepath, zlib
Files
- main/Act.hs +40/−20
- main/Options.hs +8/−0
- src/Network/AWS/Flow.hs +36/−18
- wolf.cabal +3/−1
main/Act.hs view
@@ -4,7 +4,8 @@ ( main ) where -import BasicPrelude hiding ( ByteString, (</>), hash, length, readFile, find )+import BasicPrelude hiding ( ByteString, (</>), (<.>), hash, length, readFile, find )+import Codec.Compression.GZip import Control.Monad.Trans.Resource import Data.Aeson.Encode import Data.ByteString ( length )@@ -13,22 +14,30 @@ import Data.Text.Lazy ( toStrict ) import Data.Text.Lazy.Builder hiding ( fromText ) import Data.Yaml hiding ( Parser )+import Filesystem.Path ( (<.>), dropExtension ) import Network.AWS.Data.Crypto import Network.AWS.Flow import Options import Options.Applicative hiding ( action )-import Shelly hiding ( FilePath, bash )+import Shelly hiding ( FilePath, (<.>), bash ) data Args = Args { aConfig :: FilePath , aQueue :: Queue , aContainer :: FilePath , aContainerless :: Maybe String+ , aGzip :: Bool } deriving ( Eq, Read, Show ) args :: Parser Args-args = Args <$> configFile <*> (pack <$> queue) <*> containerFile <*> containerless+args = Args <$>+ configFile <*>+ (pack <$> queue) <*>+ containerFile <*>+ containerless <*>+ gzip + parser :: ParserInfo Args parser = info ( helper <*> args ) $ fullDesc@@ -67,17 +76,13 @@ 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 =+exec :: MonadIO m => Args -> Container -> Uid -> Metadata -> [Blob] -> m (Metadata, [Artifact])+exec Args{..} container 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+ maybe (docker dataDir storeDir container) (bash dir container) aContainerless result <- dataOutput $ dataDir </> pack "output.json" artifacts <- storeOutput $ storeDir </> pack "output" return (result, artifacts) where@@ -90,6 +95,28 @@ action dir (dir </> pack "data") (dir </> pack "store") control file = writefile file $ encodeText $ Control uid+ writeArtifact file blob =+ if aGzip then+ writeBinary (dropExtension file) $ BL.toStrict $ decompress blob+ else+ writeBinary file $ BL.toStrict blob+ readArtifact dir file =+ if aGzip then do+ key <- relativeTo dir file+ blob <- BL.toStrict . compress . BL.fromStrict <$> readBinary file+ return ( toTextIgnore (key <.> "gz")+ , hash blob+ , fromIntegral $ length blob+ , BL.fromStrict blob+ )+ else do+ key <- relativeTo dir file+ blob <- readBinary file+ return ( toTextIgnore key+ , hash blob+ , fromIntegral $ length blob+ , BL.fromStrict blob+ ) dataInput file = maybe (return ()) (writefile file) metadata dataOutput file =@@ -100,17 +127,10 @@ forM_ blobs $ \(key, blob) -> do paths <- liftM strip $ run "dirname" [key] mkdir_p $ dir </> paths- writeBinary (dir </> key) (BL.toStrict blob)+ writeArtifact (dir </> key) 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- )+ forM artifacts $ readArtifact dir docker dataDir storeDir Container{..} = do devices <- forM cDevices $ \device -> liftM strip $ run "readlink" ["-f", device]@@ -137,7 +157,7 @@ container <- decodeFile aContainer >>= maybeThrow (userError "Bad Container") env <- flowEnv config forever $ runResourceT $ runFlowT env $- act aQueue $ exec container aContainerless+ act aQueue $ exec Args{..} container main :: IO () main = execParser parser >>= call
main/Options.hs view
@@ -5,6 +5,7 @@ , containerFile , queue , containerless+ , gzip ) where import BasicPrelude@@ -56,3 +57,10 @@ $ long "containerless" <> metavar "DIR" <> help "Run outside of container in directory"++gzip :: Parser Bool+gzip =+ switch+ $ long "gzip"+ <> help "GZIP contents of artifacts"+
src/Network/AWS/Flow.hs view
@@ -41,22 +41,32 @@ -- Interface +serviceError :: MonadFlow m => ErrorCode -> Error -> m ()+serviceError code = \case+ e@(ServiceError s) ->+ unless check $ throwM e where+ check =+ s ^. serviceStatus == badRequest400 &&+ s ^. serviceAbbrev == "SWF" &&+ s ^. serviceCode == code+ e -> throwM e+ register :: MonadFlow m => Plan -> m () register Plan{..} = do logInfo' "event=register"- r <- registerDomainAction- s <- registerWorkflowTypeAction- (tskName $ strtTask plnStart)- (tskVersion $ strtTask plnStart)- (tskTimeout $ strtTask plnStart)- foldM_ go [s, r] plnSpecs where- go rs Work{..} = do- r <- registerActivityTypeAction- (tskName wrkTask)- (tskVersion wrkTask)- (tskTimeout wrkTask)- return (r : rs)- go rs Sleep{..} = return rs+ handle (serviceError "DomainAlreadyExists") registerDomainAction+ handle (serviceError "TypeAlreadyExists") $ registerWorkflowTypeAction+ (tskName $ strtTask plnStart)+ (tskVersion $ strtTask plnStart)+ (tskTimeout $ strtTask plnStart)+ mapM_ go plnSpecs where+ go Work{..} =+ handle (serviceError "TypeAlreadyExists") $ registerActivityTypeAction+ (tskName wrkTask)+ (tskVersion wrkTask)+ (tskTimeout wrkTask)+ go Sleep{..} =+ return () execute :: MonadFlow m => Task -> Metadata -> m () execute Task{..} input = do@@ -74,6 +84,13 @@ s ^. serializeMessage == "key \"taskToken\" not present" e -> throwM e +actException :: MonadFlow m => Token -> SomeException -> m ()+actException token e = do+ logError' "event=act-exception-begin"+ logError' $ show e+ logError' "event=act-exception-finish"+ respondActivityTaskFailedAction token+ act :: MonadFlow m => Queue -> (Uid -> Metadata -> [Blob] -> m (Metadata, [Artifact])) -> m () act queue action = handle serializeError $ do@@ -84,11 +101,12 @@ 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+ handle (actException token) $ do+ (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{..} =
wolf.cabal view
@@ -1,5 +1,5 @@ name: wolf-version: 0.2.2+version: 0.2.3 synopsis: Amazon Simple Workflow Service Wrapper. homepage: https://github.com/swift-nav/wolf license: MIT@@ -147,10 +147,12 @@ , optparse-applicative , resourcet , shelly+ , system-filepath , text , transformers , wolf , yaml+ , zlib default-extensions: OverloadedStrings RecordWildCards NoImplicitPrelude