Error reporting and retry fixes (awslabs#5)

sapessi · web-flow · commit b787b703b00c · 2018-11-27T09:19:37.000-08:00
* Updated header values for provided runtime and switched to ms time remaining

* Fixed issue with deadline not being set in context and switched to realtime clock to get correct epoch

* Removed libc dependency and switched to chrono for deadline calculation

* Fixed docs example not to import HttpRuntimeClient

* New error reporting method with correct content type and error type header.

* Two fixes to honor max retries in the get_next_event method and call the error_reponse rather than init_fail when an error is thrown after the call to /next
diff --git a/lambda-runtime-client/src/client.rs b/lambda-runtime-client/src/client.rs
@@ -1,4 +1,4 @@
-use error::{ApiError, RuntimeApiError};
+use error::{ApiError, ErrorResponse, RuntimeApiError};
 use hyper::{
     client::HttpConnector,
     header::{self, HeaderMap, HeaderValue},
@@ -11,6 +11,8 @@ use tokio::runtime::Runtime;
 
 const RUNTIME_API_VERSION: &str = "2018-06-01";
 const API_CONTENT_TYPE: &str = "application/json";
+const API_ERROR_CONTENT_TYPE: &str = "application/vnd.aws.lambda.error+json";
+const RUNTIME_ERROR_HEADER: &str = "Lambda-Runtime-Function-Error-Type";
 
 /// Enum of the headers returned by Lambda's `/next` API call.
 pub enum LambdaHeaders {
@@ -141,7 +143,8 @@ impl RuntimeClient {
         let uri = format!(
             "http://{}/{}/runtime/invocation/next",
             self.endpoint, RUNTIME_API_VERSION
-        ).parse()?;
+        )
+        .parse()?;
         trace!("Polling for next event");
 
         // We wait instead of processing the future asynchronously because AWS Lambda
@@ -203,7 +206,8 @@ impl RuntimeClient {
         let uri: Uri = format!(
             "http://{}/{}/runtime/invocation/{}/response",
             self.endpoint, RUNTIME_API_VERSION, request_id
-        ).parse()?;
+        )
+        .parse()?;
         trace!(
             "Posting response for request {} to Runtime API. Response length {} bytes",
             request_id,
@@ -250,14 +254,14 @@ impl RuntimeClient {
         let uri: Uri = format!(
             "http://{}/{}/runtime/invocation/{}/error",
             self.endpoint, RUNTIME_API_VERSION, request_id
-        ).parse()?;
+        )
+        .parse()?;
         trace!(
             "Posting error to runtime API for request {}: {}",
             request_id,
             e.to_response().error_message
         );
-        let err_body = serde_json::to_vec(&e.to_response()).expect("Could not serialize error object");
-        let req = self.get_runtime_post_request(&uri, err_body);
+        let req = self.get_runtime_error_request(&uri, &e.to_response());
 
         match self.http_client.request(req).wait() {
             Ok(resp) => {
@@ -298,19 +302,19 @@ impl RuntimeClient {
             .parse()
             .expect("Could not generate Runtime URI");
         error!("Calling fail_init Runtime API: {}", e.to_response().error_message);
-        let err_body = serde_json::to_vec(&e.to_response()).expect("Could not serialize error object");
-        let req = self.get_runtime_post_request(&uri, err_body);
+        let req = self.get_runtime_error_request(&uri, &e.to_response());
 
-        let resp = self
-            .http_client
+        self.http_client
             .request(req)
+            .wait()
             .map_err(|e| {
                 error!("Error while sending init failed message: {}", e);
                 panic!("Error while sending init failed message: {}", e);
-            }).map(|resp| {
+            })
+            .map(|resp| {
                 info!("Successfully sent error response to the runtime API: {:?}", resp);
-            });
-        tokio::spawn(resp);
+            })
+            .expect("Could not complete init_fail request");
     }
 
     /// Returns the endpoint configured for this HTTP Runtime client.
@@ -339,6 +343,20 @@ impl RuntimeClient {
             .unwrap()
     }
 
+    fn get_runtime_error_request(&self, uri: &Uri, e: &ErrorResponse) -> Request<Body> {
+        let body = serde_json::to_vec(e).expect("Could not turn error object into response JSON");
+        Request::builder()
+            .method(Method::POST)
+            .uri(uri.clone())
+            .header(
+                header::CONTENT_TYPE,
+                header::HeaderValue::from_static(API_ERROR_CONTENT_TYPE),
+            )
+            .header(RUNTIME_ERROR_HEADER, HeaderValue::from_static("RuntimeError")) // TODO: We should add this code to the error object.
+            .body(Body::from(body))
+            .unwrap()
+    }
+
     /// Creates an `EventContext` object based on the response returned by the Runtime
     /// API `/next` endpoint.
     ///
@@ -383,7 +401,8 @@ impl RuntimeClient {
                 error!("Response headers do not contain deadline header");
                 return Err(ApiError::new(&format!("Missing {} header", LambdaHeaders::Deadline)));
             }
-        }.parse::<u128>()?;
+        }
+        .parse::<u128>()?;
 
         let mut ctx = EventContext {
             aws_request_id,
diff --git a/lambda-runtime/src/error.rs b/lambda-runtime/src/error.rs
@@ -15,6 +15,8 @@ use serde_json;
 pub struct RuntimeError {
     msg: String,
     stack_trace: Option<backtrace::Backtrace>,
+    /// The request id that generated this error
+    pub(crate) request_id: Option<String>,
     /// Whether the error is recoverable or not.
     pub(crate) recoverable: bool,
 }
@@ -58,6 +60,7 @@ impl RuntimeError {
             msg: String::from(msg),
             stack_trace: trace,
             recoverable: true,
+            request_id: None,
         }
     }
 }
diff --git a/lambda-runtime/src/runtime.rs b/lambda-runtime/src/runtime.rs
@@ -174,7 +174,7 @@ where
     fn start(&self) {
         debug!("Beginning main event loop");
         loop {
-            let (event, ctx) = self.get_next_event(0);
+            let (event, ctx) = self.get_next_event(0, None);
             let request_id = ctx.aws_request_id.clone();
             info!("Received new event with AWS request id: {}", request_id);
             let function_outcome = self.invoke(event, ctx);
@@ -247,7 +247,27 @@ where
     ///
     /// # Return
     /// The next `Event` object to be processed.
-    pub(super) fn get_next_event(&self, retries: i8) -> (E, Context) {
+    pub(super) fn get_next_event(&self, retries: i8, e: Option<RuntimeError>) -> (E, Context) {
+        if let Some(err) = e {
+            if retries > self.max_retries {
+                error!("Unrecoverable error while fetching next event: {}", err);
+                match err.request_id.clone() {
+                    Some(req_id) => {
+                        self.runtime_client
+                            .event_error(&req_id, &err)
+                            .expect("Could not send event error response");
+                    }
+                    None => {
+                        self.runtime_client.fail_init(&err);
+                    }
+                }
+
+                // these errors are not recoverable. Either we can't communicate with the runtie APIs
+                // or we cannot parse the event. panic to restart the environment.
+                panic!("Could not retrieve next event");
+            }
+        }
+
         match self.runtime_client.next_event() {
             Ok((ev_data, invocation_ctx)) => {
                 let parse_result = serde_json::from_slice(&ev_data);
@@ -263,29 +283,15 @@ where
 
                         (ev, handler_ctx)
                     }
-                    Err(e) => {
+                    Err(mut e) => {
                         error!("Could not parse event to type: {}", e);
-                        self.get_next_event(retries + 1)
+                        let mut runtime_err = RuntimeError::from(e);
+                        runtime_err.request_id = Option::from(invocation_ctx.aws_request_id);
+                        self.get_next_event(retries + 1, Option::from(runtime_err))
                     }
                 }
             }
-            Err(e) => {
-                if !e.recoverable {
-                    error!("Unrecoverable error while fetching next event: {}", e);
-                    self.runtime_client.fail_init(&e);
-                    panic!("Could not retrieve next event");
-                }
-
-                // if the error is recoverable we retry up to max_retries time
-                if retries <= self.max_retries {
-                    //let next_retries = retries + 1;
-                    self.get_next_event(retries + 1)
-                } else {
-                    error!("Exceeded maximum number of retries: {}", e);
-                    self.runtime_client.fail_init(&e);
-                    panic!("Could not retrieve next event");
-                }
-            }
+            Err(e) => self.get_next_event(retries + 1, Option::from(RuntimeError::from(e))),
         }
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -15,6 +15,8 @@ use serde_json;`
`15`	`15`	`pub struct RuntimeError {`
`16`	`16`	`msg: String,`
`17`	`17`	`stack_trace: Option<backtrace::Backtrace>,`
	`18`	`+ /// The request id that generated this error`
	`19`	`+ pub(crate) request_id: Option<String>,`
`18`	`20`	`/// Whether the error is recoverable or not.`
`19`	`21`	`pub(crate) recoverable: bool,`
`20`	`22`	`}`
`@@ -58,6 +60,7 @@ impl RuntimeError {`
`58`	`60`	`msg: String::from(msg),`
`59`	`61`	`stack_trace: trace,`
`60`	`62`	`recoverable: true,`
	`63`	`+ request_id: None,`
`61`	`64`	`}`
`62`	`65`	`}`
`63`	`66`	`}`