diff --git a/main/Act2.hs b/main/Act2.hs new file mode 100644 --- /dev/null +++ b/main/Act2.hs @@ -0,0 +1,111 @@ +{-# LANGUAGE ScopedTypeVariables #-} +module Act2 + ( main + ) where + +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 ) +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 Filesystem.Path ( (<.>), dropExtension ) +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 + , aCommandLine :: Text + } deriving ( Eq, Read, Show ) + +args :: Parser Args +args = Args <$> + configFile <*> + (pack <$> queue) <*> + (pack <$> commandLine) + +parser :: ParserInfo Args +parser = + info ( helper <*> args ) $ fullDesc + <> header "act: Workflow activity" + <> progDesc "Workflow activity" + +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 => Text -> Uid -> Metadata -> [Blob] -> m (Metadata, [Artifact]) +exec cmdline uid metadata blobs = + shelly $ withDir $ \dir dataDir storeDir -> do + control $ dataDir pack "control.json" + storeInput $ storeDir pack "input" + dataInput $ dataDir pack "input.json" + bash dir + 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 + writeArtifact file blob = + writeBinary (dropExtension file) $ BL.toStrict $ decompress blob + readArtifact dir file = 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 + ) + 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 + writeArtifact (dir key) blob + storeOutput dir = do + artifacts <- findWhen test_f dir + forM artifacts $ readArtifact dir + bash dir = do + bashDir <- pwd + files <- ls bashDir + forM_ files $ flip cp_r dir + cd dir + maybe (return ()) (uncurry $ run_ . fromText) $ uncons $ words cmdline + +call :: Args -> IO () +call Args{..} = do + config <- decodeFile aConfig >>= maybeThrow (userError "Bad Config") + env <- flowEnv config + forever $ runResourceT $ runFlowT env $ + act aQueue $ exec aCommandLine + +main :: IO () +main = execParser parser >>= call diff --git a/main/Options.hs b/main/Options.hs --- a/main/Options.hs +++ b/main/Options.hs @@ -6,6 +6,7 @@ , queue , containerless , gzip + , commandLine ) where import BasicPrelude @@ -63,4 +64,12 @@ switch $ long "gzip" <> help "GZIP contents of artifacts" + +commandLine :: Parser String +commandLine = + strOption + $ long "command-line" + <> short 'l' + <> metavar "COMMAND" + <> help "Command to run" 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 @@ -33,11 +33,15 @@ import Network.AWS.Flow.Prelude hiding ( ByteString, Metadata, handle ) import Control.Monad.Catch +import Data.Char import qualified Data.HashMap.Strict as Map -import Formatting +import Data.Text ( pack ) +import Data.Typeable +import Formatting hiding ( string ) import Network.AWS.SWF import Network.HTTP.Types import Safe +import Text.Regex.Applicative -- Interface @@ -83,12 +87,18 @@ s ^. serializeAbbrev == "SWF" e -> throwM e +exitCode :: RE Char Int +exitCode = + many anySym *> string "exit status: " *> num <* many anySym where + num = read . pack <$> many (psym isDigit) + 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 + logError' $ sformat ("event=act-exception-type " % stext) $ show $ typeOf e + logError' $ sformat ("event=act-exception " % stext) $ show e + maybe' ((textToString $ show e) =~ exitCode) (respondActivityTaskFailedAction token) $ \code -> do + if code == 255 then respondActivityTaskCanceledAction token else + respondActivityTaskFailedAction token act :: MonadFlow m => Queue -> (Uid -> Metadata -> [Blob] -> m (Metadata, [Artifact])) -> m () act queue action = @@ -154,11 +164,15 @@ select = do event <- nextEvent [ WorkflowExecutionStarted , ActivityTaskCompleted + , ActivityTaskFailed + , ActivityTaskCanceled , TimerFired , StartChildWorkflowExecutionInitiated ] case event ^. heEventType of WorkflowExecutionStarted -> start event ActivityTaskCompleted -> completed event + ActivityTaskFailed -> failed event + ActivityTaskCanceled -> canceled event TimerFired -> timer event StartChildWorkflowExecutionInitiated -> child _ -> throwM (userError "Unknown Select Event") @@ -183,6 +197,16 @@ return (attrs ^. atceaResult, attrs' ^. atseaActivityType ^. atName) next <- workNext name schedule input next + +failed :: MonadDecide m => HistoryEvent -> m [Decision] +failed _event = do + logInfo' "event=failed" + return [failWorkflowExecutionDecision] + +canceled :: MonadDecide m => HistoryEvent -> m [Decision] +canceled _event = do + logInfo' "event=canceled" + return [cancelWorkflowExecutionDecision] timer :: MonadDecide m => HistoryEvent -> m [Decision] timer event = do diff --git a/src/Network/AWS/Flow/Prelude.hs b/src/Network/AWS/Flow/Prelude.hs --- a/src/Network/AWS/Flow/Prelude.hs +++ b/src/Network/AWS/Flow/Prelude.hs @@ -4,6 +4,7 @@ , module Control.Monad.Reader , module Control.Monad.Trans.AWS , maybe_ + , maybe' ) where import BasicPrelude hiding ( (<.>), uncons ) @@ -14,3 +15,5 @@ maybe_ :: Monad m => Maybe a -> (a -> m ()) -> m () maybe_ m f = maybe (return ()) f m +maybe' :: Maybe a -> b -> (a -> b) -> b +maybe' m b a = maybe b a m diff --git a/src/Network/AWS/Flow/SWF.hs b/src/Network/AWS/Flow/SWF.hs --- a/src/Network/AWS/Flow/SWF.hs +++ b/src/Network/AWS/Flow/SWF.hs @@ -6,10 +6,13 @@ , pollForActivityTaskAction , respondActivityTaskCompletedAction , respondActivityTaskFailedAction + , respondActivityTaskCanceledAction , pollForDecisionTaskAction , respondDecisionTaskCompletedAction , scheduleActivityTaskDecision , completeWorkflowExecutionDecision + , failWorkflowExecutionDecision + , cancelWorkflowExecutionDecision , startTimerDecision , continueAsNewWorkflowExecutionDecision , startChildWorkflowExecutionDecision @@ -87,6 +90,12 @@ void $ timeout timeout' $ send $ respondActivityTaskFailed token +respondActivityTaskCanceledAction :: MonadFlow m => Token -> m () +respondActivityTaskCanceledAction token = do + timeout' <- asks feTimeout + void $ timeout timeout' $ + send $ respondActivityTaskCanceled token + pollForDecisionTaskAction :: MonadFlow m => Queue -> m (Maybe Token, [HistoryEvent]) pollForDecisionTaskAction queue = do timeout' <- asks fePollTimeout @@ -123,6 +132,18 @@ dCompleteWorkflowExecutionDecisionAttributes .~ Just attrs where attrs = completeWorkflowExecutionDecisionAttributes & cwedaResult .~ result + +failWorkflowExecutionDecision :: Decision +failWorkflowExecutionDecision = + decision FailWorkflowExecution & + dFailWorkflowExecutionDecisionAttributes .~ Just attrs where + attrs = failWorkflowExecutionDecisionAttributes + +cancelWorkflowExecutionDecision :: Decision +cancelWorkflowExecutionDecision = + decision CancelWorkflowExecution & + dCancelWorkflowExecutionDecisionAttributes .~ Just attrs where + attrs = cancelWorkflowExecutionDecisionAttributes startTimerDecision :: Uid -> Name -> Timeout -> Decision startTimerDecision uid name t = diff --git a/wolf.cabal b/wolf.cabal --- a/wolf.cabal +++ b/wolf.cabal @@ -1,5 +1,5 @@ name: wolf -version: 0.2.6 +version: 0.2.7 synopsis: Amazon Simple Workflow Service Wrapper. homepage: https://github.com/swift-nav/wolf license: MIT @@ -53,6 +53,8 @@ , mtl , mtl-compat , optparse-applicative + , regex-applicative + , regex-compat , resourcet , safe , text @@ -142,6 +144,30 @@ other-modules: Options hs-source-dirs: main ghc-options: -Wall -main-is Act + build-depends: aeson + , amazonka-core + , base + , basic-prelude + , bytestring + , optparse-applicative + , resourcet + , shelly + , system-filepath + , text + , transformers + , wolf + , yaml + , zlib + default-extensions: OverloadedStrings + RecordWildCards + NoImplicitPrelude + +executable wolf-act2 + default-language: Haskell2010 + main-is: Act2.hs + other-modules: Options + hs-source-dirs: main + ghc-options: -Wall -main-is Act2 build-depends: aeson , amazonka-core , base