Merge pull request #30 from bakdata/feature/r-native-runtime

Feature/r native runtime

Author: philipp94831

Pipfile.lock

@@ -2,13 +2,4 @@
 
 set -euo pipefail
 
-# Processing
-while true
-do
-  HEADERS="$(mktemp)"
-  # Get an event
-  EVENT_DATA=$(curl -sS -LD "$HEADERS" -X GET "http://${AWS_LAMBDA_RUNTIME_API}/2018-06-01/runtime/invocation/next")
-  REQUEST_ID=$(grep -Fi Lambda-Runtime-Aws-Request-Id "$HEADERS" | tr -d '[:space:]' | cut -d: -f2)
-  
-  /opt/R/bin/Rscript /opt/runtime.R "$EVENT_DATA" ${REQUEST_ID}
-done
+/opt/R/bin/Rscript /opt/bootstrap.R

runtime/src/bootstrap

@@ -0,0 +1,10 @@
+source('/opt/runtime.R')
+tryCatch({
+    function_name <- initializeRuntime()
+    while (TRUE) {
+        handle_request(function_name)
+        rm(list=ls())
+        source('/opt/runtime.R')
+        function_name <- initializeRuntime()
+    }
+}, error = throwInitError)

runtime/src/bootstrap.R

@@ -1,30 +1,83 @@
-library(httr)
-library(jsonlite)
-library(logging)
-
 to_str <- function(x) {
-  return(paste(capture.output(print(x)), collapse = "\n"))
+    return(paste(capture.output(print(x)), collapse = "\n"))
+}
+
+error_to_payload <- function(error) {
+    return(list(errorMessage = toString(error), errorType = class(error)[1]))
+}
+
+post_error <- function(error, url) {
+    logerror(error)
+    res <- POST(url,
+                add_headers("Lambda-Runtime-Function-Error-Type" = "Unhandled"),
+                body = error_to_payload(error),
+                encode = "json")
+    loginfo("Posted result:\n%s", to_str(res))
+}
+
+get_source_file_name <- function(file_base_name) {
+    file_name <- paste0(file_base_name, ".R")
+    if (! file.exists(file_name)) {
+        file_name <- paste0(file_base_name, ".r")
+    }
+    if (! file.exists(file_name)) {
+        stop(paste0('Source file does not exist: ', file_base_name, '.[R|r]'))
+    }
+    return(file_name)
+}
+
+invoke_lambda <- function(EVENT_DATA, function_name) {
+    params <- fromJSON(EVENT_DATA)
+    loginfo("Invoking function '%s' with parameters:\n%s", function_name, to_str(params))
+    result <- do.call(function_name, params)
+    loginfo("Function returned:\n%s", to_str(result))
+    return(result)
+}
+
+initializeRuntime <- function() {
+    library(httr)
+    library(jsonlite)
+    library(logging)
+
+    HANDLER <- Sys.getenv("_HANDLER")
+    HANDLER_split <- strsplit(HANDLER, ".", fixed = TRUE)[[1]]
+    file_base_name <- HANDLER_split[1]
+    file_name <- get_source_file_name(file_base_name)
+    loginfo("Sourcing '%s'", file_name)
+    source(file_name)
+    return(HANDLER_split[2])
 }
 
-HANDLER <- Sys.getenv("_HANDLER")
 AWS_LAMBDA_RUNTIME_API <- Sys.getenv("AWS_LAMBDA_RUNTIME_API")
-args = commandArgs(trailingOnly = TRUE)
-EVENT_DATA <- args[1]
-REQUEST_ID <- args[2]
-
-HANDLER_split <- strsplit(HANDLER, ".", fixed = TRUE)[[1]]
-file_name <- paste0(HANDLER_split[1], ".R")
-function_name <- HANDLER_split[2]
-loginfo("Sourcing '%s'", file_name)
-source(file_name)
-params <- fromJSON(EVENT_DATA)
-loginfo("Invoking function '%s' with parameters: %s", function_name, to_str(params))
-result <- do.call(function_name, params)
-loginfo("Function returned: %s", to_str(result))
-url <- paste0("http://",
-              AWS_LAMBDA_RUNTIME_API,
-              "/2018-06-01/runtime/invocation/",
-              REQUEST_ID,
-              "/response")
-res <- POST(url, body = list(result = result), encode = "json")
-loginfo("Posted result: %s", to_str(res))
+API_ENDPOINT <- paste0("http://", AWS_LAMBDA_RUNTIME_API, "/2018-06-01/runtime/")
+
+throwInitError <- function(error) {
+    url <- paste0(API_ENDPOINT, "init/error")
+    post_error(error, url)
+    stop()
+}
+
+throwRuntimeError <- function(error, REQUEST_ID) {
+    url <- paste0(API_ENDPOINT, "invocation/", REQUEST_ID, "/error")
+    post_error(error, url)
+}
+
+postResult <- function(result, REQUEST_ID) {
+    url <- paste0(API_ENDPOINT, "invocation/", REQUEST_ID, "/response")
+    res <- POST(url, body = list(result = result), encode = "json")
+    loginfo("Posted result:\n%s", to_str(res))
+}
+
+handle_request <- function(function_name) {
+    event_url <- paste0(API_ENDPOINT, "invocation/next")
+    event_response <- GET(event_url)
+    REQUEST_ID <- event_response$headers$`Lambda-Runtime-Aws-Request-Id`
+    tryCatch({
+        EVENT_DATA <- rawToChar(event_response$content)
+        result <- invoke_lambda(EVENT_DATA, function_name)
+        postResult(result, REQUEST_ID)
+    },
+    error = function(error) {
+        throwRuntimeError(error, REQUEST_ID)
+    })
+}

runtime/src/runtime.R

@@ -13,32 +13,54 @@ Resources:
     Type: 'AWS::Serverless::Function'
     Properties:
       Handler: script.handler
+  LowerCaseExtensionFunction:
+    Type: 'AWS::Serverless::Function'
+    Properties:
+      Handler: lowercase.handler
+  MissingFunctionFunction:
+    Type: 'AWS::Serverless::Function'
+    Properties:
+      Handler: script.handler_missing
+  MissingSourceFileFunction:
+    Type: 'AWS::Serverless::Function'
+    Properties:
+      Handler: missing.handler
+  MultipleArgumentsFunction:
+    Type: 'AWS::Serverless::Function'
+    Properties:
+      Handler: script.handler_with_multiple_arguments
+  VariableArgumentsFunction:
+    Type: 'AWS::Serverless::Function'
+    Properties:
+      Handler: script.handler_with_variable_arguments
   MatrixFunction:
     Type: 'AWS::Serverless::Function'
     Properties:
       Handler: matrix.handler
       Layers:
-        - !Ref RuntimeLayer
         - !Ref RecommendedLayer
+  MissingLibraryFunction:
+    Type: 'AWS::Serverless::Function'
+    Properties:
+      Handler: matrix.handler
   AWSFunction:
     Type: 'AWS::Serverless::Function'
     Properties:
       Handler: aws.handler
       Layers:
-        - !Ref RuntimeLayer
         - !Ref AWSLayer
   RuntimeLayer:
     Type: AWS::Serverless::LayerVersion
     Properties:
-      LayerName: runtime
+      LayerName: r-runtime
       ContentUri: runtime/build/layer/
   RecommendedLayer:
     Type: AWS::Serverless::LayerVersion
     Properties:
-      LayerName: runtime
+      LayerName: r-recommended
       ContentUri: recommended/build/layer/
   AWSLayer:
     Type: AWS::Serverless::LayerVersion
     Properties:
-      LayerName: runtime
+      LayerName: r-awspack
       ContentUri: awspack/build/layer/

template.yaml

@@ -0,0 +1,3 @@
+handler <- function(x) {
+    return(x + 1)
+}

tests/R/lowercase.r

@@ -1,3 +1,11 @@
 handler <- function(x) {
     return(x + 1)
 }
+
+handler_with_multiple_arguments <- function(x, y) {
+    return(list(x = x, y = y))
+}
+
+handler_with_variable_arguments <- function(...) {
+    return(1)
+}