wolf 0.2.6 → 0.2.7
raw patch · 6 files changed
+200/−6 lines, 6 filesdep +regex-applicativedep +regex-compatdep ~basenew-component:exe:wolf-act2
Dependencies added: regex-applicative, regex-compat
Dependency ranges changed: base
Files
- main/Act2.hs +111/−0
- main/Options.hs +9/−0
- src/Network/AWS/Flow.hs +29/−5
- src/Network/AWS/Flow/Prelude.hs +3/−0
- src/Network/AWS/Flow/SWF.hs +21/−0
- wolf.cabal +27/−1
+ main/Act2.hs view
@@ -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
main/Options.hs view
@@ -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"
src/Network/AWS/Flow.hs view
@@ -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
src/Network/AWS/Flow/Prelude.hs view
@@ -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
src/Network/AWS/Flow/SWF.hs view
@@ -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 =
wolf.cabal view
@@ -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