Implement Pull

What a doozy! Getting this working was incredibly interesting, and I
learned a whole lot about what it's like working with streams and stream
combinators. i.e: it's haaaaaard.

This current implementation is pretty slick. It streams data directly
from a containrs blob response into containerd's content store via gRPC,
which means that memory usage stays low, and the only disk IO happens
on the containerd end.

Plus, resumable downloads! If a download fails mid-transfer, containerd
supports resuming the write at a later time. Nice!

shellrt/containerd-grpc/Cargo.toml

@@ -14,13 +14,13 @@ prost-types = "0.5.0"
 tokio = "0.2.0-alpha.6"
 
 [dependencies.tonic]
-version = "0.1.0-alpha.4"
+version = "0.1.0-alpha.5"
 # default-features = false
 # features = ["codegen"]
 
 [build-dependencies]
 
 [build-dependencies.tonic-build]
-version = "0.1.0-alpha.4"
+version = "0.1.0-alpha.5"
 # default-features = false
 # features = ["rustfmt"]

shellrt/shellrt-api/src/v0/rtversion.rs

@@ -5,7 +5,7 @@ use serde::{Deserialize, Serialize};
 #[serde(tag = "type", rename = "version")]
 pub struct RuntimeVersionRequest {}
 
-/// Returned once a PullRequest completes successfully
+/// Returned once a RuntimeVersionRequest completes successfully
 #[derive(Debug, Serialize, Deserialize)]
 pub struct RuntimeVersionResponse {
     pub info: String,

shellrt/shellrt-containerd/Cargo.toml

@@ -5,15 +5,17 @@ authors = ["Azure IoT Edge Devs"]
 edition = "2018"
 
 [dependencies]
-containrs = { path = "../../containrs/containrs" }
 containerd-grpc = { path = "../containerd-grpc" }
+containrs = { path = "../../containrs/containrs" }
+
 shellrt-api = { path = "../shellrt-api" }
 
 bytes = "0.4"
 failure = "0.1"
 futures-preview = { version = "0.3.0-alpha.19", features = ["async-await"]}
+lazy_static = "1.4"
 log = "0.4"
 pretty_env_logger = "0.3"
 serde_json = "1.0"
 tokio = "0.2.0-alpha.6"
-tonic = "0.1.0-alpha.4"
+tonic = "0.1.0-alpha.5"

shellrt/shellrt-containerd/src/error.rs

@@ -1,7 +1,8 @@
 use std::fmt;
 use std::fmt::Display;
 
-use failure::{Backtrace, Context, Fail};
+pub use failure::{Backtrace, Context, Fail, ResultExt};
+use log::*;
 
 use shellrt_api::v0::{Error as ApiError, ErrorCode as ApiErrorCode};
 
@@ -22,6 +23,26 @@ pub enum ErrorKind {
 
     #[fail(display = "Unimplemented request kind")]
     UnimplementedReq,
+
+    #[fail(display = "Could not connect to containerd")]
+    GrpcConnect,
+
+    #[fail(display = "Unexpected error while communicating with containerd")]
+    GrpcUnexpectedErr,
+
+    #[fail(display = "Malformed image reference")]
+    MalformedReference,
+
+    #[fail(display = "Malformed credentials")]
+    MalformedCredentials,
+
+    #[fail(display = "Registry returned a malformed image manifest")]
+    MalformedManifest,
+
+    // TODO: this error is too broad. it might be better to map several of the more "informative"
+    // containrs errors to specific ErrorKinds / ErrorCodes.
+    #[fail(display = "Error while communicating with registry")]
+    RegistryError,
 }
 
 impl Fail for Error {
@@ -72,15 +93,23 @@ impl Into<ApiError> for Error {
             code: match self.kind() {
                 IncompatibleVersion => ApiErrorCode::IncompatibleVersion,
                 InvalidRequest => ApiErrorCode::InvalidRequest,
-                UnimplementedReq => ApiErrorCode::Other(100),
+                // XXX: assign specific error codes to all ErrorKind variants
+                _ => ApiErrorCode::Other(999),
+            },
+            message: {
+                error!("{:?}", self.to_string());
+                self.to_string()
             },
-            message: self.to_string(),
             // TODO: make the details nicer
             detail: {
                 let err: failure::Error = self.into();
                 let msg = err
                     .iter_causes()
                     .map(|cause| format!("\tcaused by: {}", cause))
+                    .map(|cause| {
+                        error!("{}", cause);
+                        cause
+                    })
                     .collect::<Vec<String>>()
                     .join("\n");
                 if msg.is_empty() {

shellrt/shellrt-containerd/src/handler.rs

@@ -0,0 +1,5 @@
+mod pull;
+mod rtversion;
+
+pub use pull::PullHandler as Pull;
+pub use rtversion::RuntimeVersionHandler as RuntimeVersion;