diff --git a/main/actor.hs b/main/actor.hs --- a/main/actor.hs +++ b/main/actor.hs @@ -12,29 +12,31 @@ -- Program arguments. -- data Args = Args - { config :: FilePath + { config :: FilePath -- ^ Configuration file. , storeconf :: Bool -- ^ Optional copy configuration file to output dorectory. (default: False) - , quiesce :: Maybe FilePath + , quiesce :: Maybe FilePath -- ^ Optional quiesce file to stop actor. - , domain :: Maybe Text + , domain :: Maybe Text -- ^ Optional domain to use. - , bucket :: Maybe Text + , bucket :: Maybe Text -- ^ Optional bucket to use. - , prefix :: Maybe Text + , prefix :: Maybe Text -- ^ Optional prefix to use. - , queue :: [Text] + , queue :: [Text] -- ^ Queue to listen to act on. - , num :: Maybe Int + , num :: Maybe Int -- ^ Number of actors to run concurrently. - , nocopy :: Bool + , interval :: Maybe Int + -- ^ Interval to heartbeat at. + , nocopy :: Bool -- ^ Copy working directory. (default: False) - , local :: Bool + , local :: Bool -- ^ Run locally, not in a temp directory. (default: False) - , include :: [FilePath] + , include :: [FilePath] -- ^ Optional artifacts to filter. - , command :: String + , command :: String -- ^ Command to run. } deriving (Show, Generic) @@ -47,13 +49,13 @@ args <- getRecord "Actor" actMain (config args) - (storeconf args) (quiesce args) (domain args) (bucket args) (prefix args) (queue args) (fromMaybe 1 $ num args) + (interval args) (nocopy args) (local args) (include args) diff --git a/src/Network/AWS/Wolf/Act.hs b/src/Network/AWS/Wolf/Act.hs --- a/src/Network/AWS/Wolf/Act.hs +++ b/src/Network/AWS/Wolf/Act.hs @@ -9,6 +9,8 @@ , actMain ) where +import Control.Concurrent +import Control.Concurrent.Async.Lifted import Data.Aeson import Data.Time import Network.AWS.Wolf.Ctx @@ -68,15 +70,53 @@ traceInfo "end" [ "exception" .= (displayException <$> e) ] pure e --- | Check if quiesce file is present. +-- | Check if optional file is present. -- check :: MonadIO m => Maybe FilePath -> m Bool check = maybe (pure False) (liftIO . doesFileExist) +-- | Run a heartbeat. +-- +startHearbeat :: MonadConf c m => Text -> Int -> Text -> FilePath -> m () +startHearbeat queue interval token wd = do + traceInfo "heartbeat" mempty + sd <- storeDirectory wd + msd <- metaDirectory sd + let f = msd "heartbeat" + writeText f mempty + liftIO $ threadDelay $ interval * 1000000 + ok <- check $ pure f + if ok then do + traceInfo "fail" mempty + failActivity token + statsIncrement "wolf.act.activity.count" [ "queue" =. queue, "status" =. "fail" ] + else do + nok <- heartbeatActivity token + if not nok then startHearbeat queue interval token msd else do + traceInfo "cancel" mempty + cancelActivity token + statsIncrement "wolf.act.activity.count" [ "queue" =. queue, "status" =. "cancel" ] + +-- | Run command. +-- +startCommand :: MonadConf c m => Text -> String -> Text -> FilePath -> m () +startCommand queue command token wd = do + traceInfo "command" mempty + dd <- dataDirectory wd + sd <- storeDirectory wd + osd <- outputDirectory sd + msd <- metaDirectory sd + e <- run command + out <- readText (dd "output.json") + writeText (msd (textToString queue <> "_output.json")) out + maybe (completeActivity token out) (const $ failActivity token) e + let status = textFromString $ maybe "complete" (const "fail") e + statsIncrement "wolf.act.activity.count" [ "queue" =. queue, "status" =. status ] + -- | Actor logic - poll for work, download artifacts, run command, upload artifacts. -- -act :: MonadConf c m => Text -> Bool -> Bool -> [FilePath] -> String -> Bool -> m () -act queue nocopy local includes command storeconf = +act :: MonadConf c m => Text -> Bool -> Bool -> [FilePath] -> String -> Maybe Int -> m () +act queue nocopy local includes command interval = preConfCtx [ "label" .= LabelAct ] $ runAmazonWorkCtx queue $ do traceInfo "poll" mempty @@ -90,34 +130,29 @@ withCurrentWorkDirectory uid' nocopy local $ \wd -> runAmazonStoreCtx uid' $ do traceInfo "start" [ "dir" .= wd ] - t2 <- liftIO getCurrentTime - dd <- dataDirectory wd - sd <- storeDirectory wd - isd <- inputDirectory sd - osd <- outputDirectory sd - msd <- metaDirectory sd + t2 <- liftIO getCurrentTime + dd <- dataDirectory wd + sd <- storeDirectory wd + isd <- inputDirectory sd + osd <- outputDirectory sd + msd <- metaDirectory sd conf <- view ccConf - when storeconf (writeYaml (osd "config.yml") conf) + writeYaml (osd "config.yml") conf writeJson (dd "control.json") (Control uid') writeText (dd "input.json") input writeText (msd (textToString queue <> "_input.json")) input download isd includes - e <- run command + maybe' interval (startCommand queue command token' wd) $ \interval' -> + race_ (startHearbeat queue interval' token' wd) (startCommand queue command token' wd) upload osd - output <- readText (dd "output.json") - writeText (msd (textToString queue <> "_output.json")) output - maybe (completeActivity token' output) (const $ failActivity token') e t3 <- liftIO getCurrentTime - traceInfo "finish" [ "dir" .= wd ] - let status = textFromString $ maybe "complete" (const "fail") e - statsIncrement "wolf.act.activity.count" [ "queue" =. queue, "status" =. status ] statsHistogram "wolf.act.activity.elapsed" (realToFrac (diffUTCTime t3 t2) :: Double) [ "queue" =. queue ] - + traceInfo "finish" [ "dir" .= wd ] -- | Run actor from main with config file. -- -actMain :: MonadControl m => FilePath -> Bool -> Maybe FilePath -> Maybe Text -> Maybe Text -> Maybe Text -> [Text] -> Int -> Bool -> Bool -> [FilePath] -> String -> m () -actMain cf storeconf quiesce domain bucket prefix queues num nocopy local includes command = +actMain :: MonadControl m => FilePath -> Maybe FilePath -> Maybe Text -> Maybe Text -> Maybe Text -> [Text] -> Int -> Maybe Int -> Bool -> Bool -> [FilePath] -> String -> m () +actMain cf quiesce domain bucket prefix queues num interval nocopy local includes command = runCtx $ runTop $ do conf <- readYaml cf let conf' = override cPrefix prefix $ override cBucket bucket $ override cDomain domain conf @@ -127,4 +162,4 @@ ok <- check quiesce when ok $ liftIO exitSuccess - act queue nocopy local includes command storeconf + act queue nocopy local includes command interval diff --git a/src/Network/AWS/Wolf/SWF.hs b/src/Network/AWS/Wolf/SWF.hs --- a/src/Network/AWS/Wolf/SWF.hs +++ b/src/Network/AWS/Wolf/SWF.hs @@ -10,6 +10,8 @@ , countDecisions , completeActivity , failActivity + , cancelActivity + , heartbeatActivity , completeDecision , scheduleWork , completeWork @@ -84,6 +86,21 @@ failActivity token = runResourceT $ runAmazonCtx $ void $ send $ respondActivityTaskFailed token + +-- | Cancel activity. +-- +cancelActivity :: MonadConf c m => Text -> m () +cancelActivity token = + runResourceT $ runAmazonCtx $ + void $ send $ respondActivityTaskCanceled token + +-- | Heartbeat activity. +-- +heartbeatActivity :: MonadConf c m => Text -> m Bool +heartbeatActivity token = + runResourceT $ runAmazonCtx $ do + rathrs <- send $ recordActivityTaskHeartbeat token + pure (rathrs ^. rathrsCancelRequested) -- | Successful decision completion. -- diff --git a/wolf.cabal b/wolf.cabal --- a/wolf.cabal +++ b/wolf.cabal @@ -1,6 +1,6 @@ cabal-version: >=1.22 name: wolf -version: 0.3.47 +version: 0.3.48 license: MIT license-file: LICENSE copyright: Copyright (C) 2015-2016 Swift Navigation, Inc.