diff --git a/hal.cabal b/hal.cabal --- a/hal.cabal +++ b/hal.cabal @@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack -- --- hash: 2c7baad228ce0a3a9be1e930b2b3f15f122b84118367a2a1efeadb22c7cbe530 +-- hash: 6f7613899646dea541e673e595bdbab5b6d5d435566e444cd4bbfb31377b9890 name: hal -version: 0.1.4 +version: 0.1.5 synopsis: A runtime environment for Haskell applications running on AWS Lambda. description: This library uniquely supports different types of AWS Lambda Handlers for your needs/comfort with advanced Haskell. Instead of exposing a single function @@ -44,9 +44,12 @@ aeson , base >=4.7 && <5 , bytestring + , conduit + , conduit-extra , containers , envy , exceptions + , http-client , http-conduit , http-types , mtl diff --git a/src/AWS/Lambda/Runtime.hs b/src/AWS/Lambda/Runtime.hs --- a/src/AWS/Lambda/Runtime.hs +++ b/src/AWS/Lambda/Runtime.hs @@ -20,8 +20,10 @@ mRuntimeWithContext ) where -import AWS.Lambda.RuntimeClient (getBaseRuntimeRequest, getNextEvent, - sendEventError, sendEventSuccess, sendInitError) +import AWS.Lambda.RuntimeClient (RuntimeClientConfig, getNextEvent, + getRuntimeClientConfig, + sendEventError, sendEventSuccess, + sendInitError) import AWS.Lambda.Context (LambdaContext(..), HasLambdaContext(..)) import AWS.Lambda.Internal (StaticContext, DynamicContext(DynamicContext), mkContext) @@ -40,8 +42,7 @@ import Data.Text (unpack) import Data.Text.Encoding (decodeUtf8) import Data.Time.Clock.POSIX (posixSecondsToUTCTime) -import Network.HTTP.Simple (Request, getResponseBody, - getResponseHeader) +import Network.HTTP.Simple (getResponseBody, getResponseHeader) import System.Environment (setEnv) import System.Envy (decodeEnv, defConfig) @@ -76,11 +77,11 @@ _ -> Nothing -runtimeLoop :: (HasLambdaContext r, MonadReader r m, MonadCatch m, MonadIO m, FromJSON event, ToJSON result) => Request -> StaticContext -> +runtimeLoop :: (HasLambdaContext r, MonadReader r m, MonadCatch m, MonadIO m, FromJSON event, ToJSON result) => RuntimeClientConfig -> StaticContext -> (event -> m result) -> m () -runtimeLoop baseRuntimeRequest staticContext fn = do +runtimeLoop runtimeClientConfig staticContext fn = do -- Get an event - nextRes <- liftIO $ getNextEvent baseRuntimeRequest + nextRes <- liftIO $ getNextEvent runtimeClientConfig -- If we got an event but our requestId is invalid/missing, there's no hope of meaningful recovery let reqIdBS = head $ getResponseHeader "Lambda-Runtime-Aws-Request-Id" nextRes @@ -128,8 +129,8 @@ return $ first (displayException :: SomeException -> String) caughtResult liftIO $ case result of - Right r -> sendEventSuccess baseRuntimeRequest reqIdBS r - Left e -> sendEventError baseRuntimeRequest reqIdBS e + Right r -> sendEventSuccess runtimeClientConfig reqIdBS r + Left e -> sendEventError runtimeClientConfig reqIdBS e --TODO: Revisit all names before we put them under contract -- | For any monad that supports IO/catch/Reader LambdaContext. @@ -172,17 +173,15 @@ mRuntimeWithContext :: (HasLambdaContext r, MonadCatch m, MonadReader r m, MonadIO m, FromJSON event, ToJSON result) => (event -> m result) -> m () mRuntimeWithContext fn = do - -- TODO: Hide the implementation details of Request and StaticContext. - -- If we instead have a method that returns an opaque RuntimeClientConfig - -- that encapsulates these details, and then all clientMethods accept - -- the RuntimeClientConfig instead of a baseRuntimeRequest. - baseRuntimeRequest <- liftIO getBaseRuntimeRequest + -- TODO: Hide the implementation details of StaticContext within + -- RuntimeClientConfig that encapsulates more details + runtimeClientConfig <- liftIO getRuntimeClientConfig possibleStaticCtx <- liftIO $ (decodeEnv :: IO (Either String StaticContext)) case possibleStaticCtx of - Left err -> liftIO $ sendInitError baseRuntimeRequest err - Right staticContext -> forever $ runtimeLoop baseRuntimeRequest staticContext fn + Left err -> liftIO $ sendInitError runtimeClientConfig err + Right staticContext -> forever $ runtimeLoop runtimeClientConfig staticContext fn -- | Helper for using arbitrary monads with only the LambdaContext in its Reader readerTRuntimeWithContext :: ReaderT LambdaContext m a -> m a diff --git a/src/AWS/Lambda/RuntimeClient.hs b/src/AWS/Lambda/RuntimeClient.hs --- a/src/AWS/Lambda/RuntimeClient.hs +++ b/src/AWS/Lambda/RuntimeClient.hs @@ -8,7 +8,8 @@ -} module AWS.Lambda.RuntimeClient ( - getBaseRuntimeRequest, + RuntimeClientConfig, + getRuntimeClientConfig, getNextEvent, sendEventSuccess, sendEventError, @@ -16,16 +17,29 @@ ) where import Control.Concurrent (threadDelay) -import Control.Exception (displayException, try, throw) -import Data.Aeson (encode) +import Control.Exception (Exception, displayException, throw, + try) +import Control.Monad (unless) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Data.Aeson (Result (..), Value, encode, + fromJSON) +import Data.Aeson.Parser (value') import Data.Aeson.Types (FromJSON, ToJSON) import Data.Bifunctor (first) import qualified Data.ByteString as BS +import Data.Conduit (ConduitM, runConduit, yield, (.|)) +import Data.Conduit.Attoparsec (sinkParser) import GHC.Generics (Generic (..)) -import Network.HTTP.Simple (HttpException, JSONException, - Request, Response, - getResponseStatus, httpJSONEither, - httpNoBody, parseRequest, +import Network.HTTP.Client (BodyReader, HttpException, Manager, + Request, Response, brRead, + defaultManagerSettings, httpNoBody, + managerConnCount, + managerIdleConnectionCount, + managerResponseTimeout, + managerSetProxy, newManager, + noProxy, parseRequest, responseBody, + responseTimeoutNone, withResponse) +import Network.HTTP.Simple (getResponseStatus, setRequestBodyJSON, setRequestBodyLBS, setRequestCheckStatus, @@ -43,6 +57,8 @@ instance ToJSON LambdaError +data RuntimeClientConfig = RuntimeClientConfig Request Manager + -- Exposed Handlers -- TODO: It would be interesting if we could make the interface a sort of @@ -50,14 +66,28 @@ -- things off we get a 'getNextEvent' handler and then the 'getNextEvent' -- handler returns both the 'success' and 'error' handlers. So things like -- baseRequest and reqId are pre-injected. -getBaseRuntimeRequest :: IO Request -getBaseRuntimeRequest = do +getRuntimeClientConfig :: IO RuntimeClientConfig +getRuntimeClientConfig = do awsLambdaRuntimeApi <- getEnv "AWS_LAMBDA_RUNTIME_API" - parseRequest $ "http://" ++ awsLambdaRuntimeApi + req <- parseRequest $ "http://" ++ awsLambdaRuntimeApi + man <- newManager + -- In the off chance that they set a proxy value, we don't want to + -- use it. There's also no reason to spend time reading env vars. + $ managerSetProxy noProxy + $ defaultManagerSettings + -- This is the most important setting, we must not timeout requests + { managerResponseTimeout = responseTimeoutNone + -- We only ever need a single connection, because we'll never make + -- concurrent requests and never talk to more than one host. + , managerConnCount = 1 + , managerIdleConnectionCount = 1 + } + return $ RuntimeClientConfig req man -getNextEvent :: FromJSON a => Request -> IO (Response (Either JSONException a)) -getNextEvent baseRuntimeRequest = do - resOrEx <- runtimeClientRetryTry $ httpJSONEither $ toNextEventRequest baseRuntimeRequest + +getNextEvent :: FromJSON a => RuntimeClientConfig -> IO (Response (Either JSONParseException a)) +getNextEvent rcc@(RuntimeClientConfig baseRuntimeRequest manager) = do + resOrEx <- runtimeClientRetryTry $ flip httpJSONEither manager $ toNextEventRequest baseRuntimeRequest let checkStatus res = if not $ statusIsSuccessful $ getResponseStatus res then Left "Unexpected Runtime Error: Could not retrieve next event." else @@ -65,13 +95,13 @@ let resOrMsg = first (displayException :: HttpException -> String) resOrEx >>= checkStatus case resOrMsg of Left msg -> do - _ <- sendInitError baseRuntimeRequest msg + _ <- sendInitError rcc msg error msg Right y -> return y -sendEventSuccess :: ToJSON a => Request -> BS.ByteString -> a -> IO () -sendEventSuccess baseRuntimeRequest reqId json = do - resOrEx <- runtimeClientRetryTry $ httpNoBody $ toEventSuccessRequest reqId json baseRuntimeRequest +sendEventSuccess :: ToJSON a => RuntimeClientConfig -> BS.ByteString -> a -> IO () +sendEventSuccess rcc@(RuntimeClientConfig baseRuntimeRequest manager) reqId json = do + resOrEx <- runtimeClientRetryTry $ flip httpNoBody manager $ toEventSuccessRequest reqId json baseRuntimeRequest let resOrTypedMsg = case resOrEx of Left ex -> @@ -92,37 +122,72 @@ case resOrTypedMsg of Left (Left msg) -> -- If an exception occurs here, we want that to propogate - sendEventError baseRuntimeRequest reqId msg + sendEventError rcc reqId msg Left (Right msg) -> error msg Right () -> return () -sendEventError :: Request -> BS.ByteString -> String -> IO () -sendEventError baseRuntimeRequest reqId e = - fmap (const ()) $ runtimeClientRetry $ httpNoBody $ toEventErrorRequest reqId e baseRuntimeRequest +sendEventError :: RuntimeClientConfig -> BS.ByteString -> String -> IO () +sendEventError (RuntimeClientConfig baseRuntimeRequest manager) reqId e = + fmap (const ()) $ runtimeClientRetry $ flip httpNoBody manager $ toEventErrorRequest reqId e baseRuntimeRequest -sendInitError :: Request -> String -> IO () -sendInitError baseRuntimeRequest e = - fmap (const ()) $ runtimeClientRetry $ httpNoBody $ toInitErrorRequest e baseRuntimeRequest +sendInitError :: RuntimeClientConfig -> String -> IO () +sendInitError (RuntimeClientConfig baseRuntimeRequest manager) e = + fmap (const ()) $ runtimeClientRetry $ flip httpNoBody manager $ toInitErrorRequest e baseRuntimeRequest +-- Helpers for Requests with JSON Bodies + +newtype JSONParseException = JSONParseException { getException :: String } deriving (Show) +instance Exception JSONParseException + +httpJSONEither :: FromJSON a => Request -> Manager -> IO (Response (Either JSONParseException a)) +httpJSONEither request manager = + let toEither v = + case fromJSON v of + Error e -> Left (JSONParseException e) + Success decoded -> Right decoded + in fmap toEither <$> httpValue request manager + +httpValue :: Request -> Manager -> IO (Response Value) +httpValue request manager = + withResponse request manager (\bodyReaderRes -> do + value <- runConduit $ bodyReaderSource (responseBody bodyReaderRes) .| sinkParser value' + return $ fmap (const value) bodyReaderRes + ) + +bodyReaderSource :: MonadIO m + => BodyReader + -> ConduitM i BS.ByteString m () +bodyReaderSource br = + loop + where + loop = do + bs <- liftIO $ brRead br + unless (BS.null bs) $ do + yield bs + loop + -- Retry Helpers -runtimeClientRetryTry' :: Int -> IO (Response a) -> IO (Either HttpException (Response a)) -runtimeClientRetryTry' 1 f = try f -runtimeClientRetryTry' i f = do - resOrEx <- try f - let retry = threadDelay 500 >> runtimeClientRetryTry' (i - 1) f - case resOrEx of - Left (_ :: HttpException) -> retry - Right res -> - -- TODO: Explore this further. - -- Before ~July 22nd 2020 it seemed that if a next event request reached - -- the runtime before a new event was available that there would be a - -- network error. After it appears that a 403 is returned. - if getResponseStatus res == status403 then retry - else return $ Right res +runtimeClientRetryTry' :: Int -> Int -> IO (Response a) -> IO (Either HttpException (Response a)) +runtimeClientRetryTry' retries maxRetries f + | retries == maxRetries = try f + | otherwise = do + resOrEx <- try f + let retry = + threadDelay (500 * 2 ^ retries) + >> runtimeClientRetryTry' (retries + 1) maxRetries f + case resOrEx of + Left (_ :: HttpException) -> retry + Right res -> + -- TODO: Explore this further. + -- Before ~July 22nd 2020 it seemed that if a next event request reached + -- the runtime before a new event was available that there would be a + -- network error. After it appears that a 403 is returned. + if getResponseStatus res == status403 then retry + else return $ Right res runtimeClientRetryTry :: IO (Response a) -> IO (Either HttpException (Response a)) -runtimeClientRetryTry = runtimeClientRetryTry' 3 +runtimeClientRetryTry = runtimeClientRetryTry' 0 10 runtimeClientRetry :: IO (Response a) -> IO (Response a) runtimeClientRetry = fmap (either throw id) . runtimeClientRetryTry