Handle asynchronous errors from pulling container images (#904)

If a container image pull fails asynchronously, such as if the image exists
but does not have the right architecture, the HTTP response has
an OK status code but the last status object in the response body is
an error.

Before this change, only the status code was used to determine the overall
success of the operation. With this change, the last status object is also
used.

edgelet/docker-rs/src/apis/image_api.rs

@@ -433,14 +433,30 @@ where
                         .and_then(move |body| Ok((status, body)))
                         .map_err(|e| Error::from(e))
                 })
-                .and_then(|(status, body)| {
-                    if status.is_success() {
-                        Ok(body)
+                .and_then(|(status, body)| -> Result<(), Error<serde_json::Value>> {
+                    if !status.is_success() {
+                        return Err(Error::from((status, &*body)));
+                    }
+
+                    // Response body is a sequence of JSON objects.
+                    // Each object is either a `{ "status": ... }` or an `{ "errorDetail": ... }`
+                    //
+                    // The overall success or failure of the operation is determined by which one
+                    // the last object is.
+
+                    let mut deserializer = serde_json::Deserializer::from_slice(&body).into_iter();
+                    let mut last_response: serde_json::Map<String, serde_json::Value> =
+                        deserializer.last().ok_or_else(|| {
+                            Error::Serde(serde::de::Error::custom(
+                                "empty response from container runtime",
+                            ))
+                        })??;
+                    if let Some(error_detail) = last_response.remove("errorDetail") {
+                        Err(Error::from((status, error_detail)))
                     } else {
-                        Err(Error::from((status, &*body)))
+                        Ok(())
                     }
-                })
-                .and_then(|_| futures::future::ok(())),
+                }),
         )
     }
 

edgelet/docker-rs/src/apis/mod.rs

@@ -4,9 +4,9 @@ use serde_json;
 
 #[derive(Debug)]
 pub enum Error<T> {
+    Api(ApiError<T>),
     Hyper(hyper::Error),
     Serde(serde_json::Error),
-    Api(ApiError<T>),
 }
 
 #[derive(Debug)]
@@ -36,6 +36,15 @@ where
     }
 }
 
+impl<'de> From<(hyper::StatusCode, serde_json::Value)> for Error<serde_json::Value> {
+    fn from(e: (hyper::StatusCode, serde_json::Value)) -> Self {
+        Error::Api(ApiError {
+            code: e.0,
+            content: Some(e.1),
+        })
+    }
+}
+
 impl<T> From<hyper::Error> for Error<T> {
     fn from(e: hyper::Error) -> Self {
         return Error::Hyper(e);

edgelet/edgelet-docker/src/error.rs

@@ -26,7 +26,7 @@ fn get_message(
 
     match content {
         Some(serde_json::Value::Object(props)) => {
-            if let serde_json::Value::String(message) = &props["message"] {
+            if let Some(serde_json::Value::String(message)) = props.get("message") {
                 return Ok(message.clone());
             }