vmm: support new sandbox controller api and streaming io

vmm/common/Cargo.toml

@@ -8,7 +8,7 @@ edition = "2021"
 panic = 'abort'
 
 [dependencies]
-containerd-sandbox = {git="https://github.com/kuasar-io/rust-extensions.git"}
+containerd-sandbox = { git = "https://github.com/kuasar-io/rust-extensions.git" }
 serde = "1.0.139"
 lazy_static = "1.4.0"
 nix = "0.24.1"
@@ -20,4 +20,4 @@ async-trait = "0.1"
 regex = "1.5.6"
 
 [build-dependencies]
-ttrpc-codegen = "0.4"
+ttrpc-codegen = { git = "https://github.com/kuasar-io/ttrpc-rust.git", branch = "v0.7.1-kuasar" }

vmm/common/build.rs

@@ -25,6 +25,8 @@ fn main() {
         "src/protos/google/protobuf/descriptor.proto",
         "src/protos/google/protobuf/empty.proto",
         "src/protos/google/protobuf/timestamp.proto",
+        "src/protos/streaming.proto",
+        "src/protos/containerd/types/transfer/data.proto",
     ];
 
     Codegen::new()

vmm/common/src/api/mod.rs

@@ -15,10 +15,13 @@ limitations under the License.
 */
 
 pub mod any;
+pub mod data;
 pub mod descriptor;
 pub mod empty;
 pub mod events;
 pub mod fieldpath;
 pub mod sandbox;
 pub mod sandbox_ttrpc;
+pub mod streaming;
+pub mod streaming_ttrpc;
 pub mod timestamp;

vmm/common/src/protos/containerd/types/transfer/data.proto

@@ -0,0 +1,29 @@
+/*
+	Copyright The containerd Authors.
+
+	Licensed under the Apache License, Version 2.0 (the "License");
+	you may not use this file except in compliance with the License.
+	You may obtain a copy of the License at
+
+		http://www.apache.org/licenses/LICENSE-2.0
+
+	Unless required by applicable law or agreed to in writing, software
+	distributed under the License is distributed on an "AS IS" BASIS,
+	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+	See the License for the specific language governing permissions and
+	limitations under the License.
+*/
+
+syntax = "proto3";
+
+package containerd.types.transfer;
+
+option go_package = "github.com/containerd/containerd/v2/api/types/transfer";
+
+message Data {
+	bytes data = 1;
+}
+
+message WindowUpdate {
+	int32 update = 1;
+}

vmm/common/src/protos/google/protobuf/any.proto

@@ -33,7 +33,7 @@ syntax = "proto3";
 package google.protobuf;
 
 option csharp_namespace = "Google.Protobuf.WellKnownTypes";
-option go_package = "google.golang.org/protobuf/types/known/anypb";
+option go_package = "github.com/golang/protobuf/ptypes/any";
 option java_package = "com.google.protobuf";
 option java_outer_classname = "AnyProto";
 option java_multiple_files = true;
@@ -77,13 +77,10 @@ option objc_class_prefix = "GPB";
 //  Example 4: Pack and unpack a message in Go
 //
 //      foo := &pb.Foo{...}
-//      any, err := anypb.New(foo)
-//      if err != nil {
-//        ...
-//      }
+//      any, err := ptypes.MarshalAny(foo)
 //      ...
 //      foo := &pb.Foo{}
-//      if err := any.UnmarshalTo(foo); err != nil {
+//      if err := ptypes.UnmarshalAny(any, foo); err != nil {
 //        ...
 //      }
 //

vmm/common/src/protos/streaming.proto

@@ -0,0 +1,31 @@
+/*
+	Copyright The containerd Authors.
+
+	Licensed under the Apache License, Version 2.0 (the "License");
+	you may not use this file except in compliance with the License.
+	You may obtain a copy of the License at
+
+		http://www.apache.org/licenses/LICENSE-2.0
+
+	Unless required by applicable law or agreed to in writing, software
+	distributed under the License is distributed on an "AS IS" BASIS,
+	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+	See the License for the specific language governing permissions and
+	limitations under the License.
+*/
+
+syntax = "proto3";
+
+package containerd.services.streaming.v1;
+
+import "google/protobuf/any.proto";
+
+option go_package = "github.com/containerd/containerd/v2/api/services/streaming/v1;streaming";
+
+service Streaming {
+	rpc Stream(stream google.protobuf.Any) returns (stream google.protobuf.Any);
+}
+
+message StreamInit {
+	string id = 1;
+}

vmm/sandbox/Cargo.toml

@@ -8,14 +8,14 @@ edition = "2021"
 panic = 'abort'
 
 [build-dependencies]
-built = { version = "0.7.0", features=["cargo-lock", "dependency-tree", "git2", "chrono", "semver"] }
+built = { version = "0.7.0", features = ["cargo-lock", "dependency-tree", "git2", "chrono", "semver"] }
 
 [dependencies]
-built = { version = "0.7.0", features=["cargo-lock", "dependency-tree", "git2", "chrono", "semver"] }
-clap = { version="4.4.2", features=["derive"] }
+built = { version = "0.7.0", features = ["cargo-lock", "dependency-tree", "git2", "chrono", "semver"] }
+clap = { version = "4.4.2", features = ["derive"] }
 tokio = { version = "1.19.2", features = ["full"] }
-containerd-sandbox = {git="https://github.com/kuasar-io/rust-extensions.git"}
-containerd-shim = { git="https://github.com/kuasar-io/rust-extensions.git", features=["async"] }
+containerd-sandbox = { git = "https://github.com/kuasar-io/rust-extensions.git" }
+containerd-shim = { git = "https://github.com/kuasar-io/rust-extensions.git", features = ["async"] }
 vmm-common = { path = "../common" }
 bytefmt = "0.1.7"
 async-trait = "0.1.56"
@@ -38,7 +38,7 @@ uuid = { version = "1.1.2", features = ["v4"] }
 unshare = { version = "0.7.0" }
 os_pipe = { version = "0.9.2" }
 qapi = { version = "0.8.0", features = ["qmp", "async-tokio-all"] }
-qapi-spec = {version = "0.3.1"}
+qapi-spec = { version = "0.3.1" }
 sandbox-derive = { path = "derive" }
 api_client = { git = "https://github.com/cloud-hypervisor/cloud-hypervisor.git" }
 rtnetlink = "0.13.1"
@@ -52,17 +52,16 @@ hostname = "0.3"
 path-clean = "1.0.1"
 
 [[bin]]
-name = "qemu" 
+name = "qemu"
 path = "src/bin/qemu/main.rs"
 
 [[bin]]
-name = "cloud_hypervisor" 
+name = "cloud_hypervisor"
 path = "src/bin/cloud_hypervisor/main.rs"
 
 [[bin]]
-name = "stratovirt" 
+name = "stratovirt"
 path = "src/bin/stratovirt/main.rs"
 
 [dev-dependencies]
 temp-dir = "0.1.11"
-

vmm/sandbox/src/cloud_hypervisor/hooks.rs

@@ -32,7 +32,7 @@ impl Hooks<CloudHypervisorVM> for CloudHypervisorHooks {
     }
 
     async fn post_start(&self, sandbox: &mut KuasarSandbox<CloudHypervisorVM>) -> Result<()> {
-        sandbox.data.task_address = sandbox.vm.agent_socket.to_string();
+        sandbox.data.task_address = format!("ttrpc+{}", sandbox.vm.agent_socket);
         // sync clock
         sandbox.sync_clock().await;
         Ok(())

vmm/sandbox/src/container/handler/io.rs

@@ -90,7 +90,7 @@ pub async fn attach_pipe<T: VM + Sync + Send>(
     sandbox: &mut KuasarSandbox<T>,
     io_devices: &mut Vec<String>,
 ) -> Result<String> {
-    let name = if !path.is_empty() && !path.contains("vsock") {
+    let name = if !path.is_empty() && !path.contains("://") {
         let (id, chardev_id) = sandbox.hot_attach_pipe(path).await?;
         io_devices.push(id);
         chardev_id

vmm/sandbox/src/qemu/hooks.rs

@@ -44,7 +44,7 @@ impl Hooks<QemuVM> for QemuHooks {
     }
 
     async fn post_start(&self, sandbox: &mut KuasarSandbox<QemuVM>) -> Result<()> {
-        sandbox.data.task_address = sandbox.vm.agent_socket.to_string();
+        sandbox.data.task_address = format!("ttrpc+{}", sandbox.vm.agent_socket);
         // sync clock
         sandbox.sync_clock().await;
         Ok(())