diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,20 @@ +Copyright (C) 2015 Swift Navigation Inc. + +Permission is hereby granted, free of charge, to any person obtaining +a copy of this software and associated documentation files (the +"Software"), to deal in the Software without restriction, including +without limitation the rights to use, copy, modify, merge, publish, +distribute, sublicense, and/or sell copies of the Software, and to +permit persons to whom the Software is furnished to do so, subject to +the following conditions: + +The above copyright notice and this permission notice shall be included +in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. +IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, +TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE +SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,3 @@ +# Wolf + +Wolf is a wrapper around Amazon Simple Workflow Service. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/src/Network/AWS/Flow.hs b/src/Network/AWS/Flow.hs new file mode 100644 --- /dev/null +++ b/src/Network/AWS/Flow.hs @@ -0,0 +1,229 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ConstraintKinds #-} +{-# LANGUAGE RecordWildCards #-} + +module Network.AWS.Flow + ( register + , execute + , act + , decide + , runFlowT + , throwStringError + , hoistStringEither + , maybeToFlowError + , Uid + , Name + , Version + , Queue + , Token + , Timeout + , Metadata + , Artifact + , FlowConfig (..) + , FlowEnv (..) + , FlowError + , FlowT + , MonadFlow + , Task (..) + , Timer (..) + , Start (..) + , Spec (..) + , End (..) + , Plan (..) + ) where + +import Control.Lens ( (^.) ) +import Control.Monad ( foldM, forM_ ) +import Control.Monad.Reader ( asks ) +import Data.List ( find ) +import Network.AWS.SWF +import Network.AWS.Flow.Internal +import Network.AWS.Flow.S3 +import Network.AWS.Flow.SWF +import Network.AWS.Flow.Types +import Safe ( headMay, tailMay ) + +-- Interface + +register :: MonadFlow m => Plan -> m [()] +register Plan{..} = do + 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 + +execute :: MonadFlow m => Task -> Metadata -> m () +execute Task{..} input = do + uid <- newUid + startWorkflowExecutionAction uid tskName tskVersion tskQueue input + +act :: MonadFlow m => Queue -> (Uid -> Metadata -> m (Metadata, [Artifact])) -> m () +act queue action = do + (token, uid, input) <- pollForActivityTaskAction queue + (output, artifacts) <- action uid input + forM_ artifacts putObjectAction + respondActivityTaskCompletedAction token output + +decide :: MonadFlow m => Plan -> m () +decide plan@Plan{..} = do + (token', events) <- pollForDecisionTaskAction (tskQueue $ strtTask plnStart) + token <- maybeToFlowError "No Token" token' + logger <- asks feLogger + decisions <- runDecide logger plan events select + respondDecisionTaskCompletedAction token decisions + +-- Helpers + +nextEvent :: MonadDecide m => [EventType] -> m HistoryEvent +nextEvent ets = do + events <- asks deEvents + maybeToFlowError "No Next Event" $ flip find events $ \e -> + e ^. heEventType `elem` ets + +workNext :: MonadDecide m => Name -> m (Maybe Spec) +workNext name = do + specs <- asks (plnSpecs . dePlan) + return $ tailMay (dropWhile p specs) >>= headMay where + p Work{..} = tskName wrkTask /= name + p _ = True + +sleepNext :: MonadDecide m => Name -> m (Maybe Spec) +sleepNext name = do + specs <- asks (plnSpecs . dePlan) + return $ tailMay (dropWhile p specs) >>= headMay where + p Sleep{..} = tmrName slpTimer /= name + p _ = True + +select :: MonadDecide m => m [Decision] +select = do + event <- nextEvent [ WorkflowExecutionStarted + , ActivityTaskCompleted + , TimerFired + , StartChildWorkflowExecutionInitiated ] + case event ^. heEventType of + WorkflowExecutionStarted -> start event + ActivityTaskCompleted -> completed event + TimerFired -> timer event + StartChildWorkflowExecutionInitiated -> child + _ -> throwStringError "Unknown Select Event" + +start :: MonadDecide m => HistoryEvent -> m [Decision] +start event = do + input <- maybeToFlowError "No Start Information" $ do + attrs <- event ^. heWorkflowExecutionStartedEventAttributes + return $ attrs ^. weseaInput + specs <- asks (plnSpecs . dePlan) + schedule input $ headMay specs + +completed :: MonadDecide m => HistoryEvent -> m [Decision] +completed event = do + findEvent <- asks deFindEvent + (input, name) <- maybeToFlowError "No Completed Information" $ do + attrs <- event ^. heActivityTaskCompletedEventAttributes + event' <- findEvent $ attrs ^. atceaScheduledEventId + attrs' <- event' ^. heActivityTaskScheduledEventAttributes + return (attrs ^. atceaResult, attrs' ^. atseaActivityType ^. atName) + next <- workNext name + schedule input next + +timer :: MonadDecide m => HistoryEvent -> m [Decision] +timer event = do + findEvent <- asks deFindEvent + name <- maybeToFlowError "No Timer Information" $ do + attrs <- event ^. heTimerFiredEventAttributes + event' <- findEvent $ attrs ^. tfeaStartedEventId + attrs' <- event' ^. heTimerStartedEventAttributes + attrs' ^. tseaControl + event' <- nextEvent [WorkflowExecutionStarted, ActivityTaskCompleted] + case event' ^. heEventType of + WorkflowExecutionStarted -> timerStart event' name + ActivityTaskCompleted -> timerCompleted event' name + _ -> throwStringError "Unknown Timer Event" + +timerStart :: MonadDecide m => HistoryEvent -> Name -> m [Decision] +timerStart event name = do + input <- maybeToFlowError "No Timer Start Information" $ do + attrs <- event ^. heWorkflowExecutionStartedEventAttributes + return $ attrs ^. weseaInput + next <- sleepNext name + schedule input next + +timerCompleted :: MonadDecide m => HistoryEvent -> Name -> m [Decision] +timerCompleted event name = do + input <- maybeToFlowError "No Timer Completed Information" $ do + attrs <- event ^. heActivityTaskCompletedEventAttributes + return $ attrs ^. atceaResult + next <- sleepNext name + schedule input next + +schedule :: MonadDecide m => Metadata -> Maybe Spec -> m [Decision] +schedule input = maybe (scheduleEnd input) (scheduleSpec input) + +scheduleSpec :: MonadDecide m => Metadata -> Spec -> m [Decision] +scheduleSpec input spec = do + uid <- newUid + case spec of + Work{..} -> + return [scheduleActivityTaskDecision uid + (tskName wrkTask) + (tskVersion wrkTask) + (tskQueue wrkTask) + input] + Sleep{..} -> + return [startTimerDecision uid + (tmrName slpTimer) + (tmrTimeout slpTimer)] + +scheduleEnd :: MonadDecide m => Metadata -> m [Decision] +scheduleEnd input = do + end <- asks (plnEnd . dePlan) + case end of + Stop -> return [completeWorkflowExecutionDecision input] + Continue -> scheduleContinue + +scheduleContinue :: MonadDecide m => m [Decision] +scheduleContinue = do + event <- nextEvent [WorkflowExecutionStarted] + input <- maybeToFlowError "No Continue Start Information" $ do + attrs <- event ^. heWorkflowExecutionStartedEventAttributes + return $ attrs ^. weseaInput + uid <- newUid + task <- asks (strtTask . plnStart . dePlan) + return [startChildWorkflowExecutionDecision uid + (tskName task) + (tskVersion task) + (tskQueue task) + input] + +child :: MonadDecide m => m [Decision] +child = do + event <- nextEvent [WorkflowExecutionStarted, ActivityTaskCompleted] + case event ^. heEventType of + WorkflowExecutionStarted -> childStart event + ActivityTaskCompleted -> childCompleted event + _ -> throwStringError "Unknown Child Event" + +childStart :: MonadDecide m => HistoryEvent -> m [Decision] +childStart event = do + input <- maybeToFlowError "No Child Start Information" $ do + attrs <- event ^. heWorkflowExecutionStartedEventAttributes + return $ attrs ^. weseaInput + return [completeWorkflowExecutionDecision input] + +childCompleted :: MonadDecide m => HistoryEvent -> m [Decision] +childCompleted event = do + input <- maybeToFlowError "No Child Completed Information" $ do + attrs <- event ^. heActivityTaskCompletedEventAttributes + return $ attrs ^. atceaResult + return [completeWorkflowExecutionDecision input] + diff --git a/src/Network/AWS/Flow/Env.hs b/src/Network/AWS/Flow/Env.hs new file mode 100644 --- /dev/null +++ b/src/Network/AWS/Flow/Env.hs @@ -0,0 +1,75 @@ +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ConstraintKinds #-} +{-# LANGUAGE OverloadedStrings #-} + +module Network.AWS.Flow.Env + ( flowEnv + ) where + +import Control.Applicative ( (<$>), (<*>) ) +import Control.Lens ( (.~), (<&>) ) +import Control.Monad ( mzero ) +import Control.Monad.Except ( runExceptT ) +import Control.Monad.Trans.AWS +import Data.Aeson +import Network.AWS.Flow +import Network.HTTP.Conduit ( conduitManagerSettings + , managerResponseTimeout + , newManager ) +import System.Log.FastLogger ( defaultBufSize + , flushLogStr + , newStderrLoggerSet + , pushLogStr ) +import System.IO ( stderr ) + +instance FromJSON Region where + parseJSON (String v) + | v == "eu-west-1" = return Ireland + | v == "eu-central-1" = return Frankfurt + | v == "ap-northeast-1" = return Tokyo + | v == "ap-southeast-1" = return Singapore + | v == "ap-southeast-2" = return Sydney + | v == "cn-north-1" = return Beijing + | v == "us-east-1" = return NorthVirginia + | v == "us-west-1" = return NorthCalifornia + | v == "us-west-2" = return Oregon + | v == "us-gov-west-1" = return GovCloud + | v == "fips-us-gov-west-1" = return GovCloudFIPS + | v == "sa-east-1" = return SaoPaulo + | otherwise = mzero + parseJSON _ = mzero + +instance FromJSON Credentials where + parseJSON (Object v) = + FromEnv <$> + v .: "access-key-env-var" <*> + v .: "secret-key-env-var" + parseJSON _ = mzero + +instance FromJSON FlowConfig where + parseJSON (Object v) = + FlowConfig <$> + v .: "region" <*> + v .: "credentials" <*> + v .: "timeout" <*> + v .: "poll-timeout" <*> + v .: "domain" <*> + v .: "bucket" + parseJSON _ = mzero + +flowEnv :: FlowConfig -> IO FlowEnv +flowEnv FlowConfig{..} = do + loggerSet <- newStderrLoggerSet defaultBufSize + logger <- newLogger Info stderr + manager <- newManager (managerSettings fcTimeout) + pollManager <- newManager (managerSettings fcPollTimeout) + env <- newEnv' manager <&> envLogger .~ logger + pollEnv <- newEnv' pollManager <&> envLogger .~ logger + return $ FlowEnv (logStrLn loggerSet) env pollEnv fcDomain fcBucket where + managerSettings timeout = + conduitManagerSettings { managerResponseTimeout = Just timeout } + newEnv' m = + runExceptT (newEnv fcRegion fcCredentials m) >>= either error return + logStrLn ls s = + pushLogStr ls s >> flushLogStr ls diff --git a/src/Network/AWS/Flow/Internal.hs b/src/Network/AWS/Flow/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Network/AWS/Flow/Internal.hs @@ -0,0 +1,207 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE UndecidableInstances #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE ConstraintKinds #-} + +module Network.AWS.Flow.Internal + ( runAWS + , runFlowT + , runDecide + , throwStringError + , hoistStringEither + , maybeToFlowError + , newUid + ) where + +import Control.Applicative ( (<$>), (<*>) ) +import Control.Lens ( (^.) ) +import Control.Monad ( msum, mzero ) +import Control.Monad.Base ( MonadBase, liftBase, liftBaseDefault ) +import Control.Monad.Except ( MonadError, ExceptT, runExceptT, throwError ) +import Control.Monad.IO.Class ( MonadIO, liftIO ) +import Control.Monad.Logger ( LogStr, runLoggingT ) +import Control.Monad.Reader ( MonadReader, ReaderT, ask, asks, local, runReaderT ) +import Control.Monad.Trans.AWS ( AWST, Env, Error, runAWST ) +import Control.Monad.Trans.Class ( MonadTrans, lift ) +import Control.Monad.Trans.Control ( MonadBaseControl + , MonadTransControl + , StM + , StT + , ComposeSt + , liftBaseWith + , liftWith + , defaultLiftBaseWith + , defaultRestoreM + , restoreM + , restoreT ) +import Data.Aeson ( FromJSON, Value(..), parseJSON, (.:) ) +import Data.HashMap.Strict ( fromList, lookup ) +import Data.Text ( pack ) +import Data.UUID ( toString ) +import Data.UUID.V4 ( nextRandom ) +import Network.AWS.SWF +import Network.AWS.Flow.Types +import Prelude hiding ( lookup ) + +-- FlowT + +runFlowT :: MonadIO m => FlowEnv -> FlowT m a -> m (Either FlowError a) +runFlowT e (FlowT k) = runExceptT (runReaderT (runLoggingT k l) e) where + l = const . const . const $ feLogger e + +instance MonadBase b m => MonadBase b (FlowT m) where + liftBase = liftBaseDefault + +instance MonadBaseControl b m => MonadBaseControl b (FlowT m) where + type StM (FlowT m) a = ComposeSt FlowT m a + + liftBaseWith = defaultLiftBaseWith + + restoreM = defaultRestoreM + +instance MonadTrans FlowT where + lift = FlowT . lift . lift . lift + +instance MonadTransControl FlowT where + type StT FlowT a = + StT (ExceptT FlowError) (StT (ReaderT FlowEnv) a) + + liftWith f = FlowT $ + liftWith $ \g -> + liftWith $ \h -> + liftWith $ \i -> + f (i . h . g . unFlowT) + + restoreT = FlowT . restoreT . restoreT . restoreT + +instance Monad m => MonadReader FlowEnv (FlowT m) where + ask = FlowT ask + local f = FlowT . local f . unFlowT + +-- DecideT + +runDecideT :: MonadIO m => DecideEnv -> DecideT m a -> m (Either FlowError a) +runDecideT e (DecideT k) = runExceptT (runReaderT (runLoggingT k l) e) where + l = const . const . const $ deLogger e + +instance MonadBase b m => MonadBase b (DecideT m) where + liftBase = liftBaseDefault + +instance MonadBaseControl b m => MonadBaseControl b (DecideT m) where + type StM (DecideT m) a = ComposeSt DecideT m a + + liftBaseWith = defaultLiftBaseWith + + restoreM = defaultRestoreM + +instance MonadTrans DecideT where + lift = DecideT . lift . lift . lift + +instance MonadTransControl DecideT where + type StT DecideT a = + StT (ExceptT FlowError) (StT (ReaderT DecideEnv) a) + + liftWith f = DecideT $ + liftWith $ \g -> + liftWith $ \h -> + liftWith $ \i -> + f (i . h . g . unDecideT) + + restoreT = DecideT . restoreT . restoreT . restoreT + +instance Monad m => MonadReader DecideEnv (DecideT m) where + ask = DecideT ask + local f = DecideT . local f . unDecideT + +-- Planning + +instance FromJSON Plan where + parseJSON (Object v) = + Plan <$> + v .: "start" <*> + v .: "specs" <*> + v .: "end" + parseJSON _ = mzero + +instance FromJSON Spec where + parseJSON (Object v) = + msum + [ Work <$> + v .: "work" + , Sleep <$> + v .: "sleep" + ] + parseJSON _ = + mzero + +instance FromJSON End where + parseJSON (String v) + | v == "stop" = return Stop + | v == "continue" = return Continue + | otherwise = mzero + parseJSON _ = mzero + +instance FromJSON Start where + parseJSON (Object v) = + Start <$> + v .: "flow" + parseJSON _ = mzero + +instance FromJSON Task where + parseJSON (Object v) = + Task <$> + v .: "name" <*> + v .: "version" <*> + v .: "queue" <*> + v .: "timeout" + parseJSON _ = mzero + +instance FromJSON Timer where + parseJSON (Object v) = + Timer <$> + v .: "name" <*> + v .: "timeout" + parseJSON _ = mzero + +-- Helpers + +hoistAWSEither :: MonadError FlowError m => Either Error a -> m a +hoistAWSEither = either (throwError . AWSError) return + +runAWS :: MonadFlow m => (FlowEnv -> Env) -> AWST m a -> m a +runAWS env action = do + e <- asks env + r <- runAWST e action + hoistAWSEither r + +throwStringError :: MonadError FlowError m => String -> m a +throwStringError = throwError . FlowError + +hoistStringEither :: MonadError FlowError m => Either String a -> m a +hoistStringEither = either throwStringError return + +runDecide :: (MonadError FlowError m, MonadIO m) + => (LogStr -> IO ()) -> Plan -> [HistoryEvent] -> DecideT m a -> m a +runDecide logger plan events action = + runDecideT env action >>= hoistFlowEither + where + env = DecideEnv logger plan events findEvent where + findEvent = + flip lookup $ fromList $ flip map events $ \e -> + (e ^. heEventId, e) + hoistFlowEither = either throwError return + +maybeToEither :: e -> Maybe a -> Either e a +maybeToEither e = maybe (Left e) Right + +maybeToFlowError :: MonadError FlowError m => String -> Maybe a -> m a +maybeToFlowError e = hoistStringEither . maybeToEither e + +newUid :: MonadIO m => m Uid +newUid = + liftIO $ do + r <- nextRandom + return $ pack $ toString r diff --git a/src/Network/AWS/Flow/S3.hs b/src/Network/AWS/Flow/S3.hs new file mode 100644 --- /dev/null +++ b/src/Network/AWS/Flow/S3.hs @@ -0,0 +1,22 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ConstraintKinds #-} + +module Network.AWS.Flow.S3 + ( putObjectAction + ) where + +import Control.Monad.Reader ( asks ) +import Control.Monad.Trans.AWS ( send_, sourceBody ) +import Data.Conduit.Binary ( sourceLbs ) +import Network.AWS.Flow.Types +import Network.AWS.Flow.Internal ( runAWS ) +import Network.AWS.S3 hiding ( bucket ) + +-- Actions + +putObjectAction :: MonadFlow m => Artifact -> m () +putObjectAction (key, hash, size, blob) = do + bucket <- asks feBucket + runAWS feEnv $ + send_ $ putObject body bucket key where + body = sourceBody hash size $ sourceLbs blob diff --git a/src/Network/AWS/Flow/SWF.hs b/src/Network/AWS/Flow/SWF.hs new file mode 100644 --- /dev/null +++ b/src/Network/AWS/Flow/SWF.hs @@ -0,0 +1,147 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ConstraintKinds #-} + +module Network.AWS.Flow.SWF + ( registerDomainAction + , registerActivityTypeAction + , registerWorkflowTypeAction + , startWorkflowExecutionAction + , pollForActivityTaskAction + , respondActivityTaskCompletedAction + , respondActivityTaskFailedAction + , pollForDecisionTaskAction + , respondDecisionTaskCompletedAction + , scheduleActivityTaskDecision + , completeWorkflowExecutionDecision + , startTimerDecision + , continueAsNewWorkflowExecutionDecision + , startChildWorkflowExecutionDecision + ) where + +import Control.Lens ( (^.), (.~), (&) ) +import Control.Monad ( liftM ) +import Control.Monad.Reader ( asks ) +import Control.Monad.Trans.AWS ( paginate, send, send_ ) +import Data.Conduit ( ($$) ) +import Data.Conduit.List ( consume ) +import Network.AWS.Flow.Types +import Network.AWS.Flow.Internal ( runAWS ) +import Network.AWS.SWF +import Safe ( headMay ) + +-- Actions + +registerDomainAction :: MonadFlow m => m () +registerDomainAction = do + domain <- asks feDomain + runAWS feEnv $ + send_ $ registerDomain domain "30" + +registerActivityTypeAction :: MonadFlow m => Name -> Version -> Timeout -> m () +registerActivityTypeAction name version timeout = do + domain <- asks feDomain + runAWS feEnv $ + send_ $ registerActivityType domain name version & + ratDefaultTaskHeartbeatTimeout .~ Just "NONE" & + ratDefaultTaskScheduleToCloseTimeout .~ Just "NONE" & + ratDefaultTaskScheduleToStartTimeout .~ Just "60" & + ratDefaultTaskStartToCloseTimeout .~ Just timeout + +registerWorkflowTypeAction :: MonadFlow m => Name -> Version -> Timeout -> m () +registerWorkflowTypeAction name version timeout = do + domain <- asks feDomain + runAWS feEnv $ + send_ $ registerWorkflowType domain name version & + rwtDefaultChildPolicy .~ Just Abandon & + rwtDefaultExecutionStartToCloseTimeout .~ Just timeout & + rwtDefaultTaskStartToCloseTimeout .~ Just "60" + +startWorkflowExecutionAction :: MonadFlow m + => Uid -> Name -> Version -> Queue -> Metadata -> m () +startWorkflowExecutionAction uid name version queue input = do + domain <- asks feDomain + runAWS feEnv $ + send_ $ startWorkflowExecution domain uid (workflowType name version) & + swe1TaskList .~ Just (taskList queue) & + swe1Input .~ input + +pollForActivityTaskAction :: MonadFlow m => Queue -> m (Token, Uid, Metadata) +pollForActivityTaskAction queue = do + domain <- asks feDomain + runAWS fePollEnv $ do + r <- send $ pollForActivityTask domain (taskList queue) + return + ( r ^. pfatrTaskToken + , r ^. pfatrWorkflowExecution ^. weWorkflowId + , r ^. pfatrInput ) + +respondActivityTaskCompletedAction :: MonadFlow m => Token -> Metadata -> m () +respondActivityTaskCompletedAction token result = + runAWS feEnv $ + send_ $ respondActivityTaskCompleted token & + ratcResult .~ result + +respondActivityTaskFailedAction :: MonadFlow m => Token -> m () +respondActivityTaskFailedAction token = + runAWS feEnv $ + send_ $ respondActivityTaskFailed token + +pollForDecisionTaskAction :: MonadFlow m => Queue -> m (Maybe Token, [HistoryEvent]) +pollForDecisionTaskAction queue = do + domain <- asks feDomain + runAWS fePollEnv $ do + rs <- paginate (pollForDecisionTask domain (taskList queue) & + pfdtReverseOrder .~ Just True & + pfdtMaximumPageSize .~ Just 100) + $$ consume + return + ( liftM (^. pfdtrTaskToken) (headMay rs) + , concatMap (^. pfdtrEvents) rs) + +respondDecisionTaskCompletedAction :: MonadFlow m => Token -> [Decision] -> m () +respondDecisionTaskCompletedAction token decisions = + runAWS feEnv $ + send_ $ respondDecisionTaskCompleted token & + rdtcDecisions .~ decisions + +-- Decisions + +scheduleActivityTaskDecision :: Uid -> Name -> Version -> Queue -> Metadata -> Decision +scheduleActivityTaskDecision uid name version list input = + decision ScheduleActivityTask & + dScheduleActivityTaskDecisionAttributes .~ Just attrs where + attrs = scheduleActivityTaskDecisionAttributes (activityType name version) uid & + satdaTaskList .~ Just (taskList list) & + satdaInput .~ input + +completeWorkflowExecutionDecision :: Metadata -> Decision +completeWorkflowExecutionDecision result = + decision CompleteWorkflowExecution & + dCompleteWorkflowExecutionDecisionAttributes .~ Just attrs where + attrs = completeWorkflowExecutionDecisionAttributes & + cwedaResult .~ result + +startTimerDecision :: Uid -> Name -> Timeout -> Decision +startTimerDecision uid name timeout = + decision StartTimer & + dStartTimerDecisionAttributes .~ Just attrs where + attrs = startTimerDecisionAttributes uid timeout & + stdaControl .~ Just name + +continueAsNewWorkflowExecutionDecision :: Version -> Queue -> Metadata -> Decision +continueAsNewWorkflowExecutionDecision version queue input = + decision ContinueAsNewWorkflowExecution & + dContinueAsNewWorkflowExecutionDecisionAttributes .~ Just attrs where + attrs = continueAsNewWorkflowExecutionDecisionAttributes & + canwedaWorkflowTypeVersion .~ Just version & + canwedaTaskList .~ Just (taskList queue) & + canwedaInput .~ input + +startChildWorkflowExecutionDecision :: Uid -> Name -> Version -> Queue -> Metadata -> Decision +startChildWorkflowExecutionDecision uid name version queue input = + decision StartChildWorkflowExecution & + dStartChildWorkflowExecutionDecisionAttributes .~ Just attrs where + attrs = startChildWorkflowExecutionDecisionAttributes (workflowType name version) uid & + scwedaTaskList .~ Just (taskList queue) & + scwedaInput .~ input diff --git a/src/Network/AWS/Flow/Types.hs b/src/Network/AWS/Flow/Types.hs new file mode 100644 --- /dev/null +++ b/src/Network/AWS/Flow/Types.hs @@ -0,0 +1,133 @@ +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE ConstraintKinds #-} + +module Network.AWS.Flow.Types where + +import Control.Applicative ( Applicative ) +import Control.Monad.Catch ( MonadCatch, MonadThrow ) +import Control.Monad.IO.Class ( MonadIO ) +import Control.Monad.Except ( ExceptT, MonadError ) +import Control.Monad.Logger ( LoggingT, MonadLogger, LogStr ) +import Control.Monad.Reader ( ReaderT, MonadReader ) +import Control.Monad.Trans.AWS ( Credentials, Env, Error, Region ) +import Control.Monad.Trans.Control ( MonadBaseControl ) +import Crypto.Hash ( Digest, SHA256 ) +import Data.ByteString.Lazy ( ByteString ) +import Data.Int ( Int64 ) +import Data.Text ( Text ) +import Network.AWS.SWF.Types ( HistoryEvent ) + +type Uid = Text +type Name = Text +type Version = Text +type Queue = Text +type Token = Text +type Timeout = Text +type Metadata = Maybe Text +type Artifact = (Text, Digest SHA256, Int64, ByteString) + +data FlowConfig = FlowConfig + { fcRegion :: Region + , fcCredentials :: Credentials + , fcTimeout :: Int + , fcPollTimeout :: Int + , fcDomain :: Text + , fcBucket :: Text + } + +data FlowEnv = FlowEnv + { feLogger :: LogStr -> IO () + , feEnv :: Env + , fePollEnv :: Env + , feDomain :: Text + , feBucket :: Text + } + +data FlowError + = FlowError String + | AWSError Error + deriving ( Show ) + +newtype FlowT m a = FlowT + { unFlowT :: LoggingT (ReaderT FlowEnv (ExceptT FlowError m)) a + } deriving ( Functor + , Applicative + , Monad + , MonadIO + , MonadLogger + , MonadCatch + , MonadThrow + , MonadError FlowError + ) + +type MonadFlow m = + ( MonadBaseControl IO m + , MonadCatch m + , MonadIO m + , MonadLogger m + , MonadReader FlowEnv m + , MonadError FlowError m + ) + +data DecideEnv = DecideEnv + { deLogger :: LogStr -> IO () + , dePlan :: Plan + , deEvents :: [HistoryEvent] + , deFindEvent :: Integer -> Maybe HistoryEvent + } + +newtype DecideT m a = DecideT + { unDecideT :: LoggingT (ReaderT DecideEnv (ExceptT FlowError m)) a + } deriving ( Functor + , Applicative + , Monad + , MonadIO + , MonadLogger + , MonadCatch + , MonadThrow + , MonadError FlowError + ) + +type MonadDecide m = + ( MonadBaseControl IO m + , MonadCatch m + , MonadIO m + , MonadLogger m + , MonadReader DecideEnv m + , MonadError FlowError m + ) + +data Task = Task + { tskName :: Name + , tskVersion :: Version + , tskQueue :: Queue + , tskTimeout :: Timeout + } deriving ( Eq, Read, Show ) + +data Timer = Timer + { tmrName :: Name + , tmrTimeout :: Timeout + } deriving ( Eq, Read, Show ) + +data Start = Start + { strtTask :: Task + } deriving ( Eq, Read, Show ) + +data Spec + = Work + { wrkTask :: Task + } + | Sleep + { slpTimer :: Timer + } deriving ( Eq, Read, Show ) + +data End + = Stop + | Continue + deriving ( Eq, Read, Show ) + +data Plan = Plan + { plnStart :: Start + , plnSpecs :: [Spec] + , plnEnd :: End + } deriving ( Eq, Read, Show ) diff --git a/src/main/Act.hs b/src/main/Act.hs new file mode 100644 --- /dev/null +++ b/src/main/Act.hs @@ -0,0 +1,129 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Act ( main ) where + +import Control.Exception ( SomeException ) +import Control.Monad ( forever, forM, mzero, liftM ) +import Control.Monad.IO.Class ( MonadIO ) +import Crypto.Hash ( hash ) +import Data.ByteString ( length ) +import Data.ByteString.Lazy ( fromStrict ) +import Data.Text ( Text, pack, append, words ) +import Data.Yaml +import Network.AWS.Flow ( Artifact, Metadata, Queue, Uid, runFlowT, act ) +import Network.AWS.Flow.Env ( flowEnv ) +import Options.Applicative hiding ( action ) +import Shelly hiding ( FilePath ) +import Prelude hiding ( length, readFile, words, writeFile ) + +data Container = Container + { cImage :: Text + , cCommand :: Text + , cVolumes :: [Text] + , cDevices :: [Text] + } deriving ( Eq, Read, Show ) + +instance FromJSON Container where + parseJSON (Object v) = + Container <$> + v .: "image" <*> + v .: "command" <*> + v .:? "volumes" .!= [] <*> + v .:? "devices" .!= [] + parseJSON _ = mzero + +data Args = Args + { aConfig :: FilePath + , aQueue :: Queue + , aContainer :: FilePath + } deriving ( Eq, Read, Show ) + +argsPI :: ParserInfo Args +argsPI = + info ( helper <*> argsP ) + ( fullDesc + <> header "act: Workflow activity" + <> progDesc "Workflow activity" ) where + argsP = args + <$> strOption + ( long "config" + <> short 'c' + <> metavar "FILE" + <> help "AWS SWF Service Flow config" ) + <*> strOption + ( long "queue" + <> short 'q' + <> metavar "NAME" + <> help "AWS SWF Service Flow queue" ) + <*> strOption + ( long "container" + <> short 'x' + <> metavar "FILE" + <> help "AWS SWF Service Flow worker container" ) where + args config queue container = Args + { aConfig = config + , aQueue = pack queue + , aContainer = container + } + +exec :: MonadIO m => Container -> Uid -> Metadata -> m (Metadata, [Artifact]) +exec container uid metadata = + shelly $ withDir $ \dataDir storeDir -> do + input dataDir metadata + docker dataDir storeDir container + result <- output dataDir + artifacts <- store storeDir + return (result, artifacts) where + withDir action = + withTmpDir $ \dir -> do + mkdir $ dir pack "data" + mkdir $ dir pack "store" + action (dir pack "data") (dir pack "store") + input dir = + maybe_ $ writefile $ dir pack "input.json" where + maybe_ = + maybe (return ()) + output dir = + catch_sh_maybe $ readfile $ dir pack "output.json" where + catch_sh_maybe action = + catch_sh (liftM Just action) $ \(_ :: SomeException) -> return Nothing + store dir = do + artifacts <- findWhen test_f dir + forM artifacts $ \artifact -> do + key <- relativeTo dir artifact + blob <- readBinary artifact + return ( toTextIgnore $ uid key + , hash blob + , fromIntegral $ length blob + , fromStrict blob + ) + docker dataDir storeDir Container{..} = + run_ "docker" $ concat + [["run"] + , devices + , volumes + , [cImage] + , words cCommand + ] where + devices = + concatMap (("--device" :) . return) cDevices + volumes = + concatMap (("--volume" :) . return) $ + append (toTextIgnore dataDir) ":/app/data" : + append (toTextIgnore storeDir) ":/app/store" : cVolumes + +main :: IO () +main = + execParser argsPI >>= call where + call Args{..} = do + config <- decodeFile aConfig >>= hoistMaybe "Bad Config" + container <- decodeFile aContainer >>= hoistMaybe "Bad Container" + env <- flowEnv config + forever $ do + r <- runFlowT env $ + act aQueue $ exec container + print r where + hoistMaybe s = + maybe (error s) return diff --git a/src/main/Decide.hs b/src/main/Decide.hs new file mode 100644 --- /dev/null +++ b/src/main/Decide.hs @@ -0,0 +1,51 @@ +{-# LANGUAGE RecordWildCards #-} + +module Decide ( main ) where + +import Control.Monad ( forever ) +import Data.Yaml ( decodeFile ) +import Network.AWS.Flow ( runFlowT, decide ) +import Network.AWS.Flow.Env ( flowEnv ) +import Options.Applicative +import Prelude hiding ( readFile ) + +data Args = Args + { aConfig :: FilePath + , aPlan :: FilePath + } deriving ( Eq, Read, Show ) + +argsPI :: ParserInfo Args +argsPI = + info ( helper <*> argsP ) + ( fullDesc + <> header "decide: Decide a workflow" + <> progDesc "Decide a workflow" ) where + argsP = args + <$> strOption + ( long "config" + <> short 'c' + <> metavar "FILE" + <> help "AWS SWF Service Flow config" ) + <*> strOption + ( long "plan" + <> short 'p' + <> metavar "FILE" + <> help "AWS SWF Service Flow plan" ) where + args config plan = Args + { aConfig = config + , aPlan = plan + } + +main :: IO () +main = + execParser argsPI >>= call where + call Args{..} = do + config <- decodeFile aConfig >>= hoistMaybe "Bad Config" + plan <- decodeFile aPlan >>= hoistMaybe "Bad Plan" + env <- flowEnv config + forever $ do + r <- runFlowT env $ + decide plan + print r where + hoistMaybe s = + maybe (error s) return diff --git a/src/main/Execute.hs b/src/main/Execute.hs new file mode 100644 --- /dev/null +++ b/src/main/Execute.hs @@ -0,0 +1,58 @@ +{-# LANGUAGE RecordWildCards #-} + +module Execute ( main ) where + +import Data.Text.IO ( readFile ) +import Data.Yaml ( decodeFile ) +import Network.AWS.Flow ( Plan (..), Start (..), runFlowT, execute ) +import Network.AWS.Flow.Env ( flowEnv ) +import Options.Applicative +import Prelude hiding ( readFile ) + +data Args = Args + { aConfig :: FilePath + , aPlan :: FilePath + , aInput :: FilePath + } deriving ( Eq, Read, Show ) + +argsPI :: ParserInfo Args +argsPI = + info ( helper <*> argsP ) + ( fullDesc + <> header "execute: Execute a workflow" + <> progDesc "Execute a workflow" ) where + argsP = args + <$> strOption + ( long "config" + <> short 'c' + <> metavar "FILE" + <> help "AWS SWF Service Flow config" ) + <*> strOption + ( long "plan" + <> short 'p' + <> metavar "FILE" + <> help "AWS SWF Service Flow plan" ) + <*> strOption + ( long "input" + <> short 'i' + <> metavar "FILE" + <> help "AWS SWF Service Flow input" ) where + args config plan input = Args + { aConfig = config + , aPlan = plan + , aInput = input + } + +main :: IO () +main = + execParser argsPI >>= call where + call Args{..} = do + config <- decodeFile aConfig >>= hoistMaybe "Bad Config" + plan <- decodeFile aPlan >>= hoistMaybe "Bad Plan" + input <- readFile aInput + env <- flowEnv config + r <- runFlowT env $ + execute (strtTask $ plnStart plan) (Just input) + print r where + hoistMaybe s = + maybe (error s) return diff --git a/src/main/Register.hs b/src/main/Register.hs new file mode 100644 --- /dev/null +++ b/src/main/Register.hs @@ -0,0 +1,48 @@ +{-# LANGUAGE RecordWildCards #-} + +module Register ( main ) where + +import Data.Yaml ( decodeFile ) +import Network.AWS.Flow ( runFlowT, register ) +import Network.AWS.Flow.Env ( flowEnv ) +import Options.Applicative + +data Args = Args + { aConfig :: FilePath + , aPlan :: FilePath + } deriving ( Eq, Read, Show ) + +argsPI :: ParserInfo Args +argsPI = + info ( helper <*> argsP ) + ( fullDesc + <> header "register: Register a workflow" + <> progDesc "Register a workflow" ) where + argsP = args + <$> strOption + ( long "config" + <> short 'c' + <> metavar "FILE" + <> help "AWS SWF Service Flow config" ) + <*> strOption + ( long "plan" + <> short 'p' + <> metavar "FILE" + <> help "AWS SWF Service Flow plan" ) where + args config plan = Args + { aConfig = config + , aPlan = plan + } + +main :: IO () +main = + execParser argsPI >>= call where + call Args{..} = do + config <- decodeFile aConfig >>= hoistMaybe "Bad Config" + plan <- decodeFile aPlan >>= hoistMaybe "Bad Plan" + env <- flowEnv config + r <- runFlowT env $ + register plan + print r where + hoistMaybe s = + maybe (error s) return diff --git a/wolf.cabal b/wolf.cabal new file mode 100644 --- /dev/null +++ b/wolf.cabal @@ -0,0 +1,103 @@ +name: wolf +version: 0.1.0 +synopsis: Amazon Simple Workflow Service Wrapper. +homepage: https://github.com/swift-nav/wolf +license: MIT +license-file: LICENSE +author: Swift Navigation Inc. +maintainer: Mark Fine +copyright: Copyright (C) 2015 Swift Navigation, Inc. +category: Network, AWS, Cloud, Distributed Computing +build-type: Simple +extra-source-files: README.md +cabal-version: >= 1.10 + +description: + Wolf is a wrapper around Amazon Simple Workflow Service. + +source-repository head + type: git + location: git@github.com:swift-nav/wolf.git + +library + exposed-modules: Network.AWS.Flow + , Network.AWS.Flow.Env + other-modules: Network.AWS.Flow.Internal + , Network.AWS.Flow.S3 + , Network.AWS.Flow.SWF + , Network.AWS.Flow.Types + default-language: Haskell2010 + hs-source-dirs: src + ghc-options: -Wall -fno-warn-orphans + build-depends: aeson >= 0.8.0 + , amazonka >= 0.3.4 + , amazonka-s3 >= 0.3.4 + , amazonka-swf >= 0.3.4 + , base >= 4.7 && < 4.8 + , bytestring >= 0.10.4 + , conduit >= 1.2.0 + , conduit-extra >= 1.1.7 + , cryptohash >= 0.11.6 + , exceptions >= 0.8.0 + , fast-logger >= 2.3.1 + , http-conduit >= 2.1.0 + , lens >= 4.9.0 + , monad-control >= 1.0.0 + , monad-logger >= 0.3.0 + , mtl >= 2.2.0 + , optparse-applicative >= 0.11.0 + , safe >= 0.3.0 + , text >= 1.2.0 + , transformers >= 0.4.0 + , transformers-base >= 0.4.0 + , unordered-containers >= 0.2.5 + , uuid >= 1.3.10 + , yaml >= 0.8.11 + +executable wolf-register + default-language: Haskell2010 + main-is: Register.hs + hs-source-dirs: src/main + ghc-options: -Wall -fno-warn-orphans -main-is Register + build-depends: base >= 4.7 && < 4.8 + , wolf + , optparse-applicative >= 0.11.0 + , text >= 1.2.0 + , yaml >= 0.8.0 + +executable wolf-execute + default-language: Haskell2010 + main-is: Execute.hs + hs-source-dirs: src/main + ghc-options: -Wall -fno-warn-orphans -main-is Execute + build-depends: base >= 4.7 && < 4.8 + , wolf + , optparse-applicative >= 0.11.0 + , text >= 1.2.0 + , yaml >= 0.8.0 + +executable wolf-decide + default-language: Haskell2010 + main-is: Decide.hs + hs-source-dirs: src/main + ghc-options: -Wall -fno-warn-orphans -main-is Decide + build-depends: base >= 4.7 && < 4.8 + , wolf + , optparse-applicative >= 0.11.0 + , text >= 1.2.0 + , yaml >= 0.8.0 + +executable wolf-act + default-language: Haskell2010 + main-is: Act.hs + hs-source-dirs: src/main + ghc-options: -Wall -fno-warn-orphans -main-is Act + build-depends: base >= 4.7 && < 4.8 + , wolf + , bytestring >= 0.10.4 + , cryptohash >= 0.11.6 + , optparse-applicative >= 0.11.0 + , shelly >= 1.6.1 + , text >= 1.2.0 + , transformers >= 0.4.0 + , yaml >= 0.8.0