diff --git a/main/Act.hs b/main/Act.hs --- a/main/Act.hs +++ b/main/Act.hs @@ -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 diff --git a/main/Options.hs b/main/Options.hs --- a/main/Options.hs +++ b/main/Options.hs @@ -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" + diff --git a/src/Network/AWS/Flow.hs b/src/Network/AWS/Flow.hs --- a/src/Network/AWS/Flow.hs +++ b/src/Network/AWS/Flow.hs @@ -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{..} = diff --git a/wolf.cabal b/wolf.cabal --- a/wolf.cabal +++ b/wolf.cabal @@ -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