chore: release (#354)

Co-authored-by: edera-cultivation[bot] <165992271+edera-cultivation[bot]@users.noreply.github.com>

.github/workflows/check.yml

@@ -26,9 +26,8 @@ jobs:
         rustup component add rustfmt
     - name: install linux dependencies
       run: ./hack/ci/install-linux-deps.sh
-    # Temporarily ignored: https://github.com/edera-dev/krata/issues/206
     - name: cargo fmt
-      run: ./hack/build/cargo.sh fmt --all -- --check || true
+      run: ./hack/build/cargo.sh fmt --all -- --check
   shellcheck:
     name: shellcheck
     runs-on: ubuntu-latest

.github/workflows/release-plz.yml

@@ -37,7 +37,7 @@ jobs:
     - name: install linux dependencies
       run: ./hack/ci/install-linux-deps.sh
     - name: release-plz
-      uses: MarcoIeni/release-plz-action@92ae919a6b3e27c0472659e3a7414ff4a00e833f # v0.5.64
+      uses: MarcoIeni/release-plz-action@e28810957ef1fedfa89b5e9692e750ce45f62a67 # v0.5.65
       env:
         GITHUB_TOKEN: "${{ steps.generate-token.outputs.token }}"
         CARGO_REGISTRY_TOKEN: "${{ secrets.KRATA_RELEASE_CARGO_TOKEN }}"

CHANGELOG.md

@@ -6,6 +6,36 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.0.19](https://github.com/edera-dev/krata/compare/v0.0.18...v0.0.19) - 2024-08-25
+
+### Added
+- *(config)* write default config to config.toml on startup ([#356](https://github.com/edera-dev/krata/pull/356))
+- *(ctl)* add --format option to host status and improve cpu topology format ([#355](https://github.com/edera-dev/krata/pull/355))
+
+### Fixed
+- *(zone-exec)* ensure that the underlying process is killed when rpc is closed ([#361](https://github.com/edera-dev/krata/pull/361))
+- *(rpc)* rename HostStatus to GetHostStatus ([#360](https://github.com/edera-dev/krata/pull/360))
+- *(console)* don't replay history when attaching to the console ([#358](https://github.com/edera-dev/krata/pull/358))
+- *(zone-exec)* catch panic errors and show all errors immediately ([#359](https://github.com/edera-dev/krata/pull/359))
+
+### Other
+- *(control)* split out all of the rpc calls into their own files ([#357](https://github.com/edera-dev/krata/pull/357))
+
+## [0.0.18](https://github.com/edera-dev/krata/compare/v0.0.17...v0.0.18) - 2024-08-22
+
+### Added
+- *(zone)* kernel command line control on launch ([#351](https://github.com/edera-dev/krata/pull/351))
+- *(xen-preflight)* test for hypervisor presence explicitly and error if missing ([#347](https://github.com/edera-dev/krata/pull/347))
+
+### Fixed
+- *(network)* allocate host ip from allocation pool ([#353](https://github.com/edera-dev/krata/pull/353))
+- *(daemon)* turn off trace logging ([#352](https://github.com/edera-dev/krata/pull/352))
+
+### Other
+- Add support for reading hypervisor console ([#344](https://github.com/edera-dev/krata/pull/344))
+- *(ctl)* move logic for branching ctl run steps into ControlCommands ([#342](https://github.com/edera-dev/krata/pull/342))
+- update Cargo.toml dependencies
+
 ## [0.0.17](https://github.com/edera-dev/krata/compare/v0.0.16...v0.0.17) - 2024-08-15
 
 ### Added

Cargo.lock

@@ -17,6 +17,12 @@ version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe"
 
+[[package]]
+name = "adler2"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "512761e0bb2578dd7380c6baaa0f4ce03e84f95e960231d1dec8bf4d7d6e2627"
+
 [[package]]
 name = "ahash"
 version = "0.8.11"
@@ -101,9 +107,9 @@ checksum = "b3d1d046238990b9cf5bcde22a3fb3584ee5cf65fb2765f454ed428c7a0063da"
 
 [[package]]
 name = "arrayvec"
-version = "0.7.4"
+version = "0.7.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "96d30a06541fbafbc7f82ed10c06164cfbd2c401138f6addd8404629c4b16711"
+checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50"
 dependencies = [
  "serde",
 ]
@@ -241,7 +247,7 @@ dependencies = [
  "cc",
  "cfg-if",
  "libc",
- "miniz_oxide",
+ "miniz_oxide 0.7.4",
  "object",
  "rustc-demangle",
 ]
@@ -381,9 +387,9 @@ checksum = "da987586004ae7c43b7df5e3f7693775068522e1086f8d9b2d74c778a0f43313"
 
 [[package]]
 name = "clap"
-version = "4.5.15"
+version = "4.5.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "11d8838454fda655dafd3accb2b6e2bea645b9e4078abe84a22ceb947235c5cc"
+checksum = "ed6719fffa43d0d87e5fd8caeab59be1554fb028cd30edc88fc4369b17971019"
 dependencies = [
  "clap_builder",
  "clap_derive",
@@ -808,12 +814,12 @@ checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80"
 
 [[package]]
 name = "flate2"
-version = "1.0.31"
+version = "1.0.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7f211bbe8e69bbd0cfdea405084f128ae8b4aaa6b0b522fc8f2b009084797920"
+checksum = "9c0596c1eac1f9e04ed902702e9878208b336edc9d6fddc8a48387349bab3666"
 dependencies = [
  "crc32fast",
- "miniz_oxide",
+ "miniz_oxide 0.8.0",
 ]
 
 [[package]]
@@ -1291,7 +1297,7 @@ dependencies = [
 
 [[package]]
 name = "krata"
-version = "0.0.17"
+version = "0.0.19"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -1331,7 +1337,7 @@ dependencies = [
 
 [[package]]
 name = "krata-buildtools"
-version = "0.0.17"
+version = "0.0.19"
 dependencies = [
  "anyhow",
  "env_logger",
@@ -1346,7 +1352,7 @@ dependencies = [
 
 [[package]]
 name = "krata-ctl"
-version = "0.0.17"
+version = "0.0.19"
 dependencies = [
  "anyhow",
  "async-stream",
@@ -1376,7 +1382,7 @@ dependencies = [
 
 [[package]]
 name = "krata-daemon"
-version = "0.0.17"
+version = "0.0.19"
 dependencies = [
  "anyhow",
  "async-stream",
@@ -1408,14 +1414,14 @@ dependencies = [
 
 [[package]]
 name = "krata-loopdev"
-version = "0.0.17"
+version = "0.0.19"
 dependencies = [
  "libc",
 ]
 
 [[package]]
 name = "krata-network"
-version = "0.0.17"
+version = "0.0.19"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -1439,7 +1445,7 @@ dependencies = [
 
 [[package]]
 name = "krata-oci"
-version = "0.0.17"
+version = "0.0.19"
 dependencies = [
  "anyhow",
  "async-compression",
@@ -1466,7 +1472,7 @@ dependencies = [
 
 [[package]]
 name = "krata-runtime"
-version = "0.0.17"
+version = "0.0.19"
 dependencies = [
  "anyhow",
  "backhand",
@@ -1507,7 +1513,7 @@ dependencies = [
 
 [[package]]
 name = "krata-xencall"
-version = "0.0.17"
+version = "0.0.19"
 dependencies = [
  "env_logger",
  "libc",
@@ -1520,7 +1526,7 @@ dependencies = [
 
 [[package]]
 name = "krata-xenclient"
-version = "0.0.17"
+version = "0.0.19"
 dependencies = [
  "async-trait",
  "env_logger",
@@ -1538,7 +1544,7 @@ dependencies = [
 
 [[package]]
 name = "krata-xenevtchn"
-version = "0.0.17"
+version = "0.0.19"
 dependencies = [
  "byteorder",
  "libc",
@@ -1550,7 +1556,7 @@ dependencies = [
 
 [[package]]
 name = "krata-xengnt"
-version = "0.0.17"
+version = "0.0.19"
 dependencies = [
  "libc",
  "nix 0.29.0",
@@ -1559,7 +1565,7 @@ dependencies = [
 
 [[package]]
 name = "krata-xenplatform"
-version = "0.0.17"
+version = "0.0.19"
 dependencies = [
  "async-trait",
  "c2rust-bitfields",
@@ -1582,7 +1588,7 @@ dependencies = [
 
 [[package]]
 name = "krata-xenstore"
-version = "0.0.17"
+version = "0.0.19"
 dependencies = [
  "byteorder",
  "env_logger",
@@ -1594,7 +1600,7 @@ dependencies = [
 
 [[package]]
 name = "krata-zone"
-version = "0.0.17"
+version = "0.0.19"
 dependencies = [
  "anyhow",
  "cgroups-rs",
@@ -1616,6 +1622,7 @@ dependencies = [
  "sys-mount",
  "sysinfo",
  "tokio",
+ "tokio-util",
 ]
 
 [[package]]
@@ -1626,9 +1633,9 @@ checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
 
 [[package]]
 name = "libc"
-version = "0.2.155"
+version = "0.2.158"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "97b3888a4aecf77e811145cadf6eef5901f4782c53886191b2f693f24761847c"
+checksum = "d8adc4bb1803a324070e64a98ae98f38934d91957a99cfb3a43dcbc01bc56439"
 
 [[package]]
 name = "libredox"
@@ -1716,6 +1723,15 @@ dependencies = [
  "adler",
 ]
 
+[[package]]
+name = "miniz_oxide"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e2d80299ef12ff69b16a84bb182e3b9df68b5a91574d3d4fa6e41b65deec4df1"
+dependencies = [
+ "adler2",
+]
+
 [[package]]
 name = "mio"
 version = "1.0.2"
@@ -1929,7 +1945,7 @@ dependencies = [
  "libc",
  "redox_syscall 0.5.3",
  "smallvec",
- "windows-targets 0.52.6",
+ "windows-targets",
 ]
 
 [[package]]
@@ -2385,9 +2401,9 @@ checksum = "7a66a03ae7c801facd77a29370b4faec201768915ac14a721ba36f20bc9c209b"
 
 [[package]]
 name = "reqwest"
-version = "0.12.5"
+version = "0.12.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c7d6d2a27d57148378eb5e111173f4276ad26340ecc5c49a4a2152167a2d6a37"
+checksum = "f8f4955649ef5c38cc7f9e8aa41761d48fb9677197daea9984dc54f56aad5e63"
 dependencies = [
  "base64",
  "bytes",
@@ -2422,7 +2438,7 @@ dependencies = [
  "wasm-bindgen-futures",
  "web-sys",
  "webpki-roots",
- "winreg",
+ "windows-registry",
 ]
 
 [[package]]
@@ -2561,9 +2577,9 @@ checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
 
 [[package]]
 name = "serde"
-version = "1.0.207"
+version = "1.0.208"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5665e14a49a4ea1b91029ba7d3bca9f299e1f7cfa194388ccc20f14743e784f2"
+checksum = "cff085d2cb684faa248efb494c39b68e522822ac0de72ccf08109abde717cfb2"
 dependencies = [
  "serde_derive",
 ]
@@ -2580,9 +2596,9 @@ dependencies = [
 
 [[package]]
 name = "serde_derive"
-version = "1.0.207"
+version = "1.0.208"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6aea2634c86b0e8ef2cfdc0c340baede54ec27b1e46febd7f80dffb2aa44a00e"
+checksum = "24008e81ff7613ed8e5ba0cfaf24e2c2f1e5b8a0495711e44fcd4882fca62bcf"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2832,6 +2848,9 @@ name = "sync_wrapper"
 version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a7065abeca94b6a8a577f9bd45aa0867a2238b74e8eb67cf10d492bc39351394"
+dependencies = [
+ "futures-core",
+]
 
 [[package]]
 name = "sys-mount"
@@ -2922,9 +2941,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
 
 [[package]]
 name = "tokio"
-version = "1.39.2"
+version = "1.39.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "daa4fb1bc778bd6f04cbfc4bb2d06a7396a8f299dc33ea1900cedaa316f467b1"
+checksum = "9babc99b9923bfa4804bd74722ff02c0381021eafa4db9949217e3be8e84fff5"
 dependencies = [
  "backtrace",
  "bytes",
@@ -3419,7 +3438,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "12342cb4d8e3b046f3d80effd474a7a02447231330ef77d71daa6fbc40681143"
 dependencies = [
  "windows-core",
- "windows-targets 0.52.6",
+ "windows-targets",
 ]
 
 [[package]]
@@ -3430,8 +3449,8 @@ checksum = "d2ed2439a290666cd67ecce2b0ffaad89c2a56b976b736e6ece670297897832d"
 dependencies = [
  "windows-implement",
  "windows-interface",
- "windows-result",
+ "windows-result 0.1.2",
- "windows-targets 0.52.6",
+ "windows-targets",
 ]
 
 [[package]]
@@ -3456,22 +3475,43 @@ dependencies = [
  "syn 2.0.74",
 ]
 
+[[package]]
+name = "windows-registry"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e400001bb720a623c1c69032f8e3e4cf09984deec740f007dd2b03ec864804b0"
+dependencies = [
+ "windows-result 0.2.0",
+ "windows-strings",
+ "windows-targets",
+]
+
 [[package]]
 name = "windows-result"
 version = "0.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5e383302e8ec8515204254685643de10811af0ed97ea37210dc26fb0032647f8"
 dependencies = [
- "windows-targets 0.52.6",
+ "windows-targets",
 ]
 
 [[package]]
-name = "windows-sys"
+name = "windows-result"
-version = "0.48.0"
+version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9"
+checksum = "1d1043d8214f791817bab27572aaa8af63732e11bf84aa21a45a78d6c317ae0e"
 dependencies = [
- "windows-targets 0.48.5",
+ "windows-targets",
+]
+
+[[package]]
+name = "windows-strings"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4cd9b125c486025df0eabcb585e62173c6c9eddcec5d117d3b6e8c30e2ee4d10"
+dependencies = [
+ "windows-result 0.2.0",
+ "windows-targets",
 ]
 
 [[package]]
@@ -3480,7 +3520,7 @@ version = "0.52.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d"
 dependencies = [
- "windows-targets 0.52.6",
+ "windows-targets",
 ]
 
 [[package]]
@@ -3489,22 +3529,7 @@ version = "0.59.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b"
 dependencies = [
- "windows-targets 0.52.6",
+ "windows-targets",
-]
-
-[[package]]
-name = "windows-targets"
-version = "0.48.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9a2fa6e2155d7247be68c096456083145c183cbbbc2764150dda45a87197940c"
-dependencies = [
- "windows_aarch64_gnullvm 0.48.5",
- "windows_aarch64_msvc 0.48.5",
- "windows_i686_gnu 0.48.5",
- "windows_i686_msvc 0.48.5",
- "windows_x86_64_gnu 0.48.5",
- "windows_x86_64_gnullvm 0.48.5",
- "windows_x86_64_msvc 0.48.5",
 ]
 
 [[package]]
@@ -3513,46 +3538,28 @@ version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973"
 dependencies = [
- "windows_aarch64_gnullvm 0.52.6",
+ "windows_aarch64_gnullvm",
- "windows_aarch64_msvc 0.52.6",
+ "windows_aarch64_msvc",
- "windows_i686_gnu 0.52.6",
+ "windows_i686_gnu",
  "windows_i686_gnullvm",
- "windows_i686_msvc 0.52.6",
+ "windows_i686_msvc",
- "windows_x86_64_gnu 0.52.6",
+ "windows_x86_64_gnu",
- "windows_x86_64_gnullvm 0.52.6",
+ "windows_x86_64_gnullvm",
- "windows_x86_64_msvc 0.52.6",
+ "windows_x86_64_msvc",
 ]
 
-[[package]]
-name = "windows_aarch64_gnullvm"
-version = "0.48.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2b38e32f0abccf9987a4e3079dfb67dcd799fb61361e53e2882c3cbaf0d905d8"
-
 [[package]]
 name = "windows_aarch64_gnullvm"
 version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3"
 
-[[package]]
-name = "windows_aarch64_msvc"
-version = "0.48.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dc35310971f3b2dbbf3f0690a219f40e2d9afcf64f9ab7cc1be722937c26b4bc"
-
 [[package]]
 name = "windows_aarch64_msvc"
 version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469"
 
-[[package]]
-name = "windows_i686_gnu"
-version = "0.48.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a75915e7def60c94dcef72200b9a8e58e5091744960da64ec734a6c6e9b3743e"
-
 [[package]]
 name = "windows_i686_gnu"
 version = "0.52.6"
@@ -3565,48 +3572,24 @@ version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66"
 
-[[package]]
-name = "windows_i686_msvc"
-version = "0.48.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8f55c233f70c4b27f66c523580f78f1004e8b5a8b659e05a4eb49d4166cca406"
-
 [[package]]
 name = "windows_i686_msvc"
 version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66"
 
-[[package]]
-name = "windows_x86_64_gnu"
-version = "0.48.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "53d40abd2583d23e4718fddf1ebec84dbff8381c07cae67ff7768bbf19c6718e"
-
 [[package]]
 name = "windows_x86_64_gnu"
 version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78"
 
-[[package]]
-name = "windows_x86_64_gnullvm"
-version = "0.48.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0b7b52767868a23d5bab768e390dc5f5c55825b6d30b86c844ff2dc7414044cc"
-
 [[package]]
 name = "windows_x86_64_gnullvm"
 version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d"
 
-[[package]]
-name = "windows_x86_64_msvc"
-version = "0.48.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ed94fce61571a4006852b7389a063ab983c02eb1bb37b47f8272ce92d06d9538"
-
 [[package]]
 name = "windows_x86_64_msvc"
 version = "0.52.6"
@@ -3631,16 +3614,6 @@ dependencies = [
  "memchr",
 ]
 
-[[package]]
-name = "winreg"
-version = "0.52.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a277a57398d4bfa075df44f501a17cfdf8542d224f0d36095a2adc7aee4ef0a5"
-dependencies = [
- "cfg-if",
- "windows-sys 0.48.0",
-]
-
 [[package]]
 name = "wyz"
 version = "0.5.1"

Cargo.toml

@@ -18,14 +18,14 @@ members = [
 resolver = "2"
 
 [workspace.package]
-version = "0.0.17"
+version = "0.0.19"
 homepage = "https://krata.dev"
 license = "Apache-2.0"
 repository = "https://github.com/edera-dev/krata"
 
 [workspace.dependencies]
 anyhow = "1.0"
-arrayvec = "0.7.4"
+arrayvec = "0.7.6"
 async-compression = "0.4.12"
 async-stream = "0.3.5"
 async-trait = "0.1.81"
@@ -85,6 +85,7 @@ sysinfo = "0.31.2"
 termtree = "0.5.1"
 thiserror = "1.0"
 tokio-tun = "0.11.5"
+tokio-util = "0.7.11"
 toml = "0.8.19"
 tonic-build = "0.12.1"
 tower = "0.5.0"
@@ -94,7 +95,7 @@ walkdir = "2"
 xz2 = "0.1"
 
 [workspace.dependencies.clap]
-version = "4.5.15"
+version = "4.5.16"
 features = ["derive"]
 
 [workspace.dependencies.prost-reflect]
@@ -102,12 +103,12 @@ version = "0.14.0"
 features = ["derive"]
 
 [workspace.dependencies.reqwest]
-version = "0.12.5"
+version = "0.12.7"
 default-features = false
 features = ["rustls-tls"]
 
 [workspace.dependencies.serde]
-version = "1.0.207"
+version = "1.0.208"
 features = ["derive"]
 
 [workspace.dependencies.sys-mount]
@@ -115,7 +116,7 @@ version = "3.0.0"
 default-features = false
 
 [workspace.dependencies.tokio]
-version = "1.39.2"
+version = "1.39.3"
 features = ["full"]
 
 [workspace.dependencies.tokio-stream]

crates/build/Cargo.toml

@@ -16,7 +16,7 @@ oci-spec = { workspace = true }
 scopeguard = { workspace = true }
 tokio = { workspace = true }
 tokio-stream = { workspace = true }
-krata-oci = { path = "../oci", version = "^0.0.17" }
+krata-oci = { path = "../oci", version = "^0.0.19" }
 krata-tokio-tar = { workspace = true }
 uuid = { workspace = true }
 

crates/ctl/Cargo.toml

@@ -20,7 +20,7 @@ env_logger = { workspace = true }
 fancy-duration = { workspace = true }
 human_bytes = { workspace = true }
 indicatif = { workspace = true }
-krata = { path = "../krata", version = "^0.0.17" }
+krata = { path = "../krata", version = "^0.0.19" }
 log = { workspace = true }
 prost-reflect = { workspace = true, features = ["serde"] }
 prost-types = { workspace = true }

crates/ctl/src/cli/device/list.rs

@@ -23,7 +23,7 @@ enum DeviceListFormat {
 }
 
 #[derive(Parser)]
-#[command(about = "List the devices on the isolation engine")]
+#[command(about = "List device information")]
 pub struct DeviceListCommand {
     #[arg(short, long, default_value = "table", help = "Output format")]
     format: DeviceListFormat,

crates/ctl/src/cli/host/cpu_topology.rs

@@ -5,7 +5,9 @@ use comfy_table::{Cell, Table};
 use krata::v1::control::{
     control_service_client::ControlServiceClient, GetHostCpuTopologyRequest, HostCpuTopologyClass,
 };
+use serde_json::Value;
 
+use crate::format::{kv2line, proto2dynamic, proto2kv};
 use tonic::{transport::Channel, Request};
 
 fn class_to_str(input: HostCpuTopologyClass) -> String {
@@ -19,6 +21,11 @@ fn class_to_str(input: HostCpuTopologyClass) -> String {
 #[derive(ValueEnum, Clone, Debug, PartialEq, Eq)]
 enum HostCpuTopologyFormat {
     Table,
+    Json,
+    JsonPretty,
+    Jsonl,
+    Yaml,
+    KeyValue,
 }
 
 #[derive(Parser)]
@@ -35,24 +42,61 @@ impl HostCpuTopologyCommand {
             .await?
             .into_inner();
 
-        let mut table = Table::new();
+        match self.format {
-        table.load_preset(UTF8_FULL_CONDENSED);
+            HostCpuTopologyFormat::Table => {
-        table.set_content_arrangement(comfy_table::ContentArrangement::Dynamic);
+                let mut table = Table::new();
-        table.set_header(vec!["id", "node", "socket", "core", "thread", "class"]);
+                table.load_preset(UTF8_FULL_CONDENSED);
+                table.set_content_arrangement(comfy_table::ContentArrangement::Dynamic);
+                table.set_header(vec!["id", "node", "socket", "core", "thread", "class"]);
 
-        for (i, cpu) in response.cpus.iter().enumerate() {
+                for (i, cpu) in response.cpus.iter().enumerate() {
-            table.add_row(vec![
+                    table.add_row(vec![
-                Cell::new(i),
+                        Cell::new(i),
-                Cell::new(cpu.node),
+                        Cell::new(cpu.node),
-                Cell::new(cpu.socket),
+                        Cell::new(cpu.socket),
-                Cell::new(cpu.core),
+                        Cell::new(cpu.core),
-                Cell::new(cpu.thread),
+                        Cell::new(cpu.thread),
-                Cell::new(class_to_str(cpu.class())),
+                        Cell::new(class_to_str(cpu.class())),
-            ]);
+                    ]);
-        }
+                }
 
-        if !table.is_empty() {
+                if !table.is_empty() {
-            println!("{}", table);
+                    println!("{}", table);
+                }
+            }
+
+            HostCpuTopologyFormat::Json
+            | HostCpuTopologyFormat::JsonPretty
+            | HostCpuTopologyFormat::Yaml => {
+                let mut values = Vec::new();
+                for cpu in response.cpus {
+                    let message = proto2dynamic(cpu)?;
+                    values.push(serde_json::to_value(message)?);
+                }
+                let value = Value::Array(values);
+                let encoded = if self.format == HostCpuTopologyFormat::JsonPretty {
+                    serde_json::to_string_pretty(&value)?
+                } else if self.format == HostCpuTopologyFormat::Yaml {
+                    serde_yaml::to_string(&value)?
+                } else {
+                    serde_json::to_string(&value)?
+                };
+                println!("{}", encoded.trim());
+            }
+
+            HostCpuTopologyFormat::Jsonl => {
+                for cpu in response.cpus {
+                    let message = proto2dynamic(cpu)?;
+                    println!("{}", serde_json::to_string(&message)?);
+                }
+            }
+
+            HostCpuTopologyFormat::KeyValue => {
+                for cpu in response.cpus {
+                    let kvs = proto2kv(cpu)?;
+                    println!("{}", kv2line(kvs),);
+                }
+            }
         }
 
         Ok(())

crates/ctl/src/cli/host/hv_console.rs

@@ -0,0 +1,23 @@
+use anyhow::Result;
+use clap::Parser;
+use krata::v1::control::{
+    control_service_client::ControlServiceClient, ReadHypervisorConsoleRequest,
+};
+
+use tonic::{transport::Channel, Request};
+
+#[derive(Parser)]
+#[command(about = "Display hypervisor console output")]
+pub struct HostHvConsoleCommand {}
+
+impl HostHvConsoleCommand {
+    pub async fn run(self, mut client: ControlServiceClient<Channel>) -> Result<()> {
+        let response = client
+            .read_hypervisor_console(Request::new(ReadHypervisorConsoleRequest {}))
+            .await?
+            .into_inner();
+
+        print!("{}", response.data);
+        Ok(())
+    }
+}

crates/ctl/src/cli/host/identify.rs

@@ -1,22 +0,0 @@
-use anyhow::Result;
-use clap::Parser;
-use krata::v1::control::{control_service_client::ControlServiceClient, HostStatusRequest};
-
-use tonic::{transport::Channel, Request};
-
-#[derive(Parser)]
-#[command(about = "Get information about the host")]
-pub struct HostStatusCommand {}
-
-impl HostStatusCommand {
-    pub async fn run(self, mut client: ControlServiceClient<Channel>) -> Result<()> {
-        let response = client
-            .host_status(Request::new(HostStatusRequest {}))
-            .await?
-            .into_inner();
-        println!("Host UUID: {}", response.host_uuid);
-        println!("Host Domain: {}", response.host_domid);
-        println!("Krata Version: {}", response.krata_version);
-        Ok(())
-    }
-}

crates/ctl/src/cli/host/mod.rs

@@ -6,12 +6,14 @@ use krata::events::EventStream;
 use krata::v1::control::control_service_client::ControlServiceClient;
 
 use crate::cli::host::cpu_topology::HostCpuTopologyCommand;
-use crate::cli::host::identify::HostStatusCommand;
+use crate::cli::host::hv_console::HostHvConsoleCommand;
 use crate::cli::host::idm_snoop::HostIdmSnoopCommand;
+use crate::cli::host::status::HostStatusCommand;
 
 pub mod cpu_topology;
-pub mod identify;
+pub mod hv_console;
 pub mod idm_snoop;
+pub mod status;
 
 #[derive(Parser)]
 #[command(about = "Manage the host of the isolation engine")]
@@ -35,6 +37,7 @@ pub enum HostCommands {
     CpuTopology(HostCpuTopologyCommand),
     Status(HostStatusCommand),
     IdmSnoop(HostIdmSnoopCommand),
+    HvConsole(HostHvConsoleCommand),
 }
 
 impl HostCommands {
@@ -49,6 +52,8 @@ impl HostCommands {
             HostCommands::Status(status) => status.run(client).await,
 
             HostCommands::IdmSnoop(snoop) => snoop.run(client, events).await,
+
+            HostCommands::HvConsole(hvconsole) => hvconsole.run(client).await,
         }
     }
 }

crates/ctl/src/cli/host/status.rs

@@ -0,0 +1,60 @@
+use anyhow::Result;
+use clap::{Parser, ValueEnum};
+use krata::v1::control::{control_service_client::ControlServiceClient, GetHostStatusRequest};
+
+use crate::format::{kv2line, proto2dynamic, proto2kv};
+use tonic::{transport::Channel, Request};
+
+#[derive(ValueEnum, Clone, Debug, PartialEq, Eq)]
+enum HostStatusFormat {
+    Simple,
+    Json,
+    JsonPretty,
+    Yaml,
+    KeyValue,
+}
+
+#[derive(Parser)]
+#[command(about = "Get information about the host")]
+pub struct HostStatusCommand {
+    #[arg(short, long, default_value = "simple", help = "Output format")]
+    format: HostStatusFormat,
+}
+
+impl HostStatusCommand {
+    pub async fn run(self, mut client: ControlServiceClient<Channel>) -> Result<()> {
+        let response = client
+            .get_host_status(Request::new(GetHostStatusRequest {}))
+            .await?
+            .into_inner();
+        match self.format {
+            HostStatusFormat::Simple => {
+                println!("Host UUID: {}", response.host_uuid);
+                println!("Host Domain: {}", response.host_domid);
+                println!("Krata Version: {}", response.krata_version);
+                println!("Host IPv4: {}", response.host_ipv4);
+                println!("Host IPv6: {}", response.host_ipv6);
+                println!("Host Ethernet Address: {}", response.host_mac);
+            }
+
+            HostStatusFormat::Json | HostStatusFormat::JsonPretty | HostStatusFormat::Yaml => {
+                let message = proto2dynamic(response)?;
+                let value = serde_json::to_value(message)?;
+                let encoded = if self.format == HostStatusFormat::JsonPretty {
+                    serde_json::to_string_pretty(&value)?
+                } else if self.format == HostStatusFormat::Yaml {
+                    serde_yaml::to_string(&value)?
+                } else {
+                    serde_json::to_string(&value)?
+                };
+                println!("{}", encoded.trim());
+            }
+
+            HostStatusFormat::KeyValue => {
+                let kvs = proto2kv(response)?;
+                println!("{}", kv2line(kvs),);
+            }
+        }
+        Ok(())
+    }
+}

crates/ctl/src/cli/mod.rs

@@ -31,6 +31,7 @@ pub struct ControlCommand {
     command: ControlCommands,
 }
 
+#[allow(clippy::large_enum_variant)]
 #[derive(Parser)]
 pub enum ControlCommands {
     Zone(ZoneCommand),
@@ -43,8 +44,17 @@ impl ControlCommand {
     pub async fn run(self) -> Result<()> {
         let client = ControlClientProvider::dial(self.connection.parse()?).await?;
         let events = EventStream::open(client.clone()).await?;
+        self.command.run(client, events).await
+    }
+}
 
-        match self.command {
+impl ControlCommands {
+    pub async fn run(
+        self,
+        client: ControlServiceClient<Channel>,
+        events: EventStream,
+    ) -> Result<()> {
+        match self {
             ControlCommands::Zone(zone) => zone.run(client, events).await,
 
             ControlCommands::Image(image) => image.run(client, events).await,

crates/ctl/src/cli/zone/attach.rs

@@ -23,7 +23,7 @@ impl ZoneAttachCommand {
         events: EventStream,
     ) -> Result<()> {
         let zone_id: String = resolve_zone(&mut client, &self.zone).await?;
-        let input = StdioConsoleStream::stdin_stream(zone_id.clone()).await;
+        let input = StdioConsoleStream::stdin_stream(zone_id.clone(), false).await;
         let output = client.attach_zone_console(input).await?.into_inner();
         let stdout_handle =
             tokio::task::spawn(async move { StdioConsoleStream::stdout(output, true).await });

crates/ctl/src/cli/zone/launch.rs

@@ -6,8 +6,8 @@ use krata::{
     events::EventStream,
     v1::{
         common::{
-            zone_image_spec::Image, OciImageFormat, ZoneImageSpec, ZoneOciImageSpec,
+            zone_image_spec::Image, OciImageFormat, ZoneImageSpec, ZoneKernelOptionsSpec,
-            ZoneResourceSpec, ZoneSpec, ZoneSpecDevice, ZoneState, ZoneTaskSpec,
+            ZoneOciImageSpec, ZoneResourceSpec, ZoneSpec, ZoneSpecDevice, ZoneState, ZoneTaskSpec,
             ZoneTaskSpecEnvVar,
         },
         control::{
@@ -91,6 +91,10 @@ pub struct ZoneLaunchCommand {
     initrd: Option<String>,
     #[arg(short = 'w', long, help = "Working directory")]
     working_directory: Option<String>,
+    #[arg(long, help = "Enable verbose logging on the kernel")]
+    kernel_verbose: bool,
+    #[arg(long, help = "Additional kernel cmdline options")]
+    kernel_cmdline_append: Option<String>,
     #[arg(help = "Container image for zone to use")]
     oci: String,
     #[arg(
@@ -166,6 +170,10 @@ impl ZoneLaunchCommand {
                     .iter()
                     .map(|name| ZoneSpecDevice { name: name.clone() })
                     .collect(),
+                kernel_options: Some(ZoneKernelOptionsSpec {
+                    verbose: self.kernel_verbose,
+                    cmdline_append: self.kernel_cmdline_append.clone().unwrap_or_default(),
+                }),
             }),
         };
         let response = client
@@ -179,7 +187,7 @@ impl ZoneLaunchCommand {
         }
 
         let code = if self.attach {
-            let input = StdioConsoleStream::stdin_stream(id.clone()).await;
+            let input = StdioConsoleStream::stdin_stream(id.clone(), true).await;
             let output = client.attach_zone_console(input).await?.into_inner();
             let stdout_handle =
                 tokio::task::spawn(async move { StdioConsoleStream::stdout(output, true).await });

crates/ctl/src/cli/zone/list.rs

@@ -29,7 +29,7 @@ enum ZoneListFormat {
 }
 
 #[derive(Parser)]
-#[command(about = "List the zones on the isolation engine")]
+#[command(about = "List zone information")]
 pub struct ZoneListCommand {
     #[arg(short, long, default_value = "table", help = "Output format")]
     format: ZoneListFormat,

crates/ctl/src/cli/zone/logs.rs

@@ -33,7 +33,7 @@ impl ZoneLogsCommand {
         let zone_id_stream = zone_id.clone();
         let follow = self.follow;
         let input = stream! {
-            yield ZoneConsoleRequest { zone_id: zone_id_stream, data: Vec::new() };
+            yield ZoneConsoleRequest { zone_id: zone_id_stream, replay_history: true, data: Vec::new() };
             if follow {
                 let mut pending = pending::<ZoneConsoleRequest>();
                 while let Some(x) = pending.next().await {

crates/ctl/src/cli/zone/mod.rs

@@ -26,7 +26,7 @@ pub mod logs;
 pub mod metrics;
 pub mod resolve;
 pub mod top;
-mod update_resources;
+pub mod update_resources;
 pub mod watch;
 
 #[derive(Parser)]
@@ -46,6 +46,7 @@ impl ZoneCommand {
     }
 }
 
+#[allow(clippy::large_enum_variant)]
 #[derive(Subcommand)]
 pub enum ZoneCommands {
     Attach(ZoneAttachCommand),

crates/ctl/src/console.rs

@@ -1,4 +1,4 @@
-use anyhow::{anyhow, Result};
+use anyhow::Result;
 use async_stream::stream;
 use crossterm::{
     terminal::{disable_raw_mode, enable_raw_mode, is_raw_mode_enabled},
@@ -23,10 +23,13 @@ use tonic::Streaming;
 pub struct StdioConsoleStream;
 
 impl StdioConsoleStream {
-    pub async fn stdin_stream(zone: String) -> impl Stream<Item = ZoneConsoleRequest> {
+    pub async fn stdin_stream(
+        zone: String,
+        replay_history: bool,
+    ) -> impl Stream<Item = ZoneConsoleRequest> {
         let mut stdin = stdin();
         stream! {
-            yield ZoneConsoleRequest { zone_id: zone, data: vec![] };
+            yield ZoneConsoleRequest { zone_id: zone, replay_history, data: vec![] };
 
             let mut buffer = vec![0u8; 60];
             loop {
@@ -41,7 +44,7 @@ impl StdioConsoleStream {
                 if size == 1 && buffer[0] == 0x1d {
                     break;
                 }
-                yield ZoneConsoleRequest { zone_id: String::default(), data };
+                yield ZoneConsoleRequest { zone_id: String::default(), replay_history, data };
             }
         }
     }
@@ -115,7 +118,12 @@ impl StdioConsoleStream {
                 return if reply.error.is_empty() {
                     Ok(reply.exit_code)
                 } else {
-                    Err(anyhow!("exec failed: {}", reply.error))
+                    StdioConsoleStream::restore_terminal_mode();
+                    stderr
+                        .write_all(format!("Error: exec failed: {}\n", reply.error).as_bytes())
+                        .await?;
+                    stderr.flush().await?;
+                    Ok(-1)
                 };
             }
         }

crates/daemon/Cargo.toml

@@ -19,9 +19,9 @@ clap = { workspace = true }
 env_logger = { workspace = true }
 futures = { workspace = true }
 ipnetwork = { workspace = true }
-krata = { path = "../krata", version = "^0.0.17" }
+krata = { path = "../krata", version = "^0.0.19" }
-krata-oci = { path = "../oci", version = "^0.0.17" }
+krata-oci = { path = "../oci", version = "^0.0.19" }
-krata-runtime = { path = "../runtime", version = "^0.0.17" }
+krata-runtime = { path = "../runtime", version = "^0.0.19" }
 log = { workspace = true }
 prost = { workspace = true }
 redb = { workspace = true }

crates/daemon/bin/daemon.rs

@@ -15,7 +15,7 @@ use kratad::command::DaemonCommand;
 async fn main() -> Result<()> {
     let mut builder = env_logger::Builder::new();
     builder
-        .filter_level(LevelFilter::Trace)
+        .filter_level(LevelFilter::Info)
         .parse_default_env()
         .filter(Some("backhand::filesystem::writer"), LevelFilter::Warn);
 

crates/daemon/src/config.rs

@@ -97,7 +97,7 @@ fn default_network_ipv4() -> DaemonIpv4NetworkConfig {
 }
 
 fn default_network_ipv4_subnet() -> String {
-    "10.75.80.0/24".to_string()
+    "10.75.0.0/16".to_string()
 }
 
 fn default_network_ipv6() -> DaemonIpv6NetworkConfig {
@@ -112,13 +112,13 @@ fn default_network_ipv6_subnet() -> String {
 
 impl DaemonConfig {
     pub async fn load(path: &Path) -> Result<DaemonConfig> {
-        if path.exists() {
+        if !path.exists() {
-            let content = fs::read_to_string(path).await?;
+            let config: DaemonConfig = toml::from_str("")?;
-            let config: DaemonConfig = toml::from_str(&content)?;
+            let content = toml::to_string_pretty(&config)?;
-            Ok(config)
+            fs::write(&path, content).await?;
-        } else {
-            fs::write(&path, "").await?;
-            Ok(DaemonConfig::default())
         }
+        let content = fs::read_to_string(path).await?;
+        let config: DaemonConfig = toml::from_str(&content)?;
+        Ok(config)
     }
 }

crates/daemon/src/console.rs

@@ -24,7 +24,7 @@ type BufferMap = Arc<Mutex<HashMap<u32, ConsoleBuffer>>>;
 
 #[derive(Clone)]
 pub struct DaemonConsoleHandle {
-    glt: ZoneLookupTable,
+    zlt: ZoneLookupTable,
     listeners: ListenerMap,
     buffers: BufferMap,
     sender: Sender<(u32, Vec<u8>)>,
@@ -57,7 +57,7 @@ impl DaemonConsoleHandle {
         uuid: Uuid,
         sender: Sender<Vec<u8>>,
     ) -> Result<DaemonConsoleAttachHandle> {
-        let Some(domid) = self.glt.lookup_domid_by_uuid(&uuid).await else {
+        let Some(domid) = self.zlt.lookup_domid_by_uuid(&uuid).await else {
             return Err(anyhow!("unable to find domain {}", uuid));
         };
         let buffers = self.buffers.lock().await;
@@ -84,7 +84,7 @@ impl Drop for DaemonConsoleHandle {
 }
 
 pub struct DaemonConsole {
-    glt: ZoneLookupTable,
+    zlt: ZoneLookupTable,
     listeners: ListenerMap,
     buffers: BufferMap,
     receiver: Receiver<(u32, Option<Vec<u8>>)>,
@@ -93,14 +93,14 @@ pub struct DaemonConsole {
 }
 
 impl DaemonConsole {
-    pub async fn new(glt: ZoneLookupTable) -> Result<DaemonConsole> {
+    pub async fn new(zlt: ZoneLookupTable) -> Result<DaemonConsole> {
         let (service, sender, receiver) =
             ChannelService::new("krata-console".to_string(), Some(0)).await?;
         let task = service.launch().await?;
         let listeners = Arc::new(Mutex::new(HashMap::new()));
         let buffers = Arc::new(Mutex::new(HashMap::new()));
         Ok(DaemonConsole {
-            glt,
+            zlt,
             listeners,
             buffers,
             receiver,
@@ -110,7 +110,7 @@ impl DaemonConsole {
     }
 
     pub async fn launch(mut self) -> Result<DaemonConsoleHandle> {
-        let glt = self.glt.clone();
+        let zlt = self.zlt.clone();
         let listeners = self.listeners.clone();
         let buffers = self.buffers.clone();
         let sender = self.sender.clone();
@@ -120,7 +120,7 @@ impl DaemonConsole {
             }
         });
         Ok(DaemonConsoleHandle {
-            glt,
+            zlt,
             listeners,
             buffers,
             sender,

crates/daemon/src/control.rs

@@ -1,713 +0,0 @@
-use crate::db::zone::ZoneStore;
-use crate::{
-    command::DaemonCommand, console::DaemonConsoleHandle, devices::DaemonDeviceManager,
-    event::DaemonEventContext, idm::DaemonIdmHandle, metrics::idm_metric_to_api,
-    oci::convert_oci_progress, zlt::ZoneLookupTable,
-};
-use async_stream::try_stream;
-use futures::Stream;
-use krata::v1::common::ZoneResourceStatus;
-use krata::v1::control::{
-    GetZoneReply, GetZoneRequest, SetHostPowerManagementPolicyReply,
-    SetHostPowerManagementPolicyRequest,
-};
-use krata::{
-    idm::internal::{
-        exec_stream_request_update::Update, request::Request as IdmRequestType,
-        response::Response as IdmResponseType, ExecEnvVar, ExecStreamRequestStart,
-        ExecStreamRequestStdin, ExecStreamRequestUpdate, MetricsRequest, Request as IdmRequest,
-    },
-    v1::{
-        common::{OciImageFormat, Zone, ZoneState, ZoneStatus},
-        control::{
-            control_service_server::ControlService, CreateZoneReply, CreateZoneRequest,
-            DestroyZoneReply, DestroyZoneRequest, DeviceInfo, ExecInsideZoneReply,
-            ExecInsideZoneRequest, GetHostCpuTopologyReply, GetHostCpuTopologyRequest,
-            HostCpuTopologyInfo, HostStatusReply, HostStatusRequest, ListDevicesReply,
-            ListDevicesRequest, ListZonesReply, ListZonesRequest, PullImageReply, PullImageRequest,
-            ReadZoneMetricsReply, ReadZoneMetricsRequest, ResolveZoneIdReply, ResolveZoneIdRequest,
-            SnoopIdmReply, SnoopIdmRequest, UpdateZoneResourcesReply, UpdateZoneResourcesRequest,
-            WatchEventsReply, WatchEventsRequest, ZoneConsoleReply, ZoneConsoleRequest,
-        },
-    },
-};
-use krataoci::{
-    name::ImageName,
-    packer::{service::OciPackerService, OciPackedFormat, OciPackedImage},
-    progress::{OciProgress, OciProgressContext},
-};
-use kratart::Runtime;
-use std::{pin::Pin, str::FromStr};
-use tokio::{
-    select,
-    sync::mpsc::{channel, Sender},
-    task::JoinError,
-};
-use tokio_stream::StreamExt;
-use tonic::{Request, Response, Status, Streaming};
-use uuid::Uuid;
-
-pub struct ApiError {
-    message: String,
-}
-
-impl From<anyhow::Error> for ApiError {
-    fn from(value: anyhow::Error) -> Self {
-        ApiError {
-            message: value.to_string(),
-        }
-    }
-}
-
-impl From<ApiError> for Status {
-    fn from(value: ApiError) -> Self {
-        Status::unknown(value.message)
-    }
-}
-
-#[derive(Clone)]
-pub struct DaemonControlService {
-    glt: ZoneLookupTable,
-    devices: DaemonDeviceManager,
-    events: DaemonEventContext,
-    console: DaemonConsoleHandle,
-    idm: DaemonIdmHandle,
-    zones: ZoneStore,
-    zone_reconciler_notify: Sender<Uuid>,
-    packer: OciPackerService,
-    runtime: Runtime,
-}
-
-impl DaemonControlService {
-    #[allow(clippy::too_many_arguments)]
-    pub fn new(
-        glt: ZoneLookupTable,
-        devices: DaemonDeviceManager,
-        events: DaemonEventContext,
-        console: DaemonConsoleHandle,
-        idm: DaemonIdmHandle,
-        zones: ZoneStore,
-        zone_reconciler_notify: Sender<Uuid>,
-        packer: OciPackerService,
-        runtime: Runtime,
-    ) -> Self {
-        Self {
-            glt,
-            devices,
-            events,
-            console,
-            idm,
-            zones,
-            zone_reconciler_notify,
-            packer,
-            runtime,
-        }
-    }
-}
-
-enum ConsoleDataSelect {
-    Read(Option<Vec<u8>>),
-    Write(Option<Result<ZoneConsoleRequest, Status>>),
-}
-
-enum PullImageSelect {
-    Progress(Option<OciProgress>),
-    Completed(Result<Result<OciPackedImage, anyhow::Error>, JoinError>),
-}
-
-#[tonic::async_trait]
-impl ControlService for DaemonControlService {
-    type ExecInsideZoneStream =
-        Pin<Box<dyn Stream<Item = Result<ExecInsideZoneReply, Status>> + Send + 'static>>;
-
-    type AttachZoneConsoleStream =
-        Pin<Box<dyn Stream<Item = Result<ZoneConsoleReply, Status>> + Send + 'static>>;
-
-    type PullImageStream =
-        Pin<Box<dyn Stream<Item = Result<PullImageReply, Status>> + Send + 'static>>;
-
-    type WatchEventsStream =
-        Pin<Box<dyn Stream<Item = Result<WatchEventsReply, Status>> + Send + 'static>>;
-
-    type SnoopIdmStream =
-        Pin<Box<dyn Stream<Item = Result<SnoopIdmReply, Status>> + Send + 'static>>;
-
-    async fn host_status(
-        &self,
-        request: Request<HostStatusRequest>,
-    ) -> Result<Response<HostStatusReply>, Status> {
-        let _ = request.into_inner();
-        Ok(Response::new(HostStatusReply {
-            host_domid: self.glt.host_domid(),
-            host_uuid: self.glt.host_uuid().to_string(),
-            krata_version: DaemonCommand::version(),
-        }))
-    }
-
-    async fn create_zone(
-        &self,
-        request: Request<CreateZoneRequest>,
-    ) -> Result<Response<CreateZoneReply>, Status> {
-        let request = request.into_inner();
-        let Some(spec) = request.spec else {
-            return Err(ApiError {
-                message: "zone spec not provided".to_string(),
-            }
-            .into());
-        };
-        let uuid = Uuid::new_v4();
-        self.zones
-            .update(
-                uuid,
-                Zone {
-                    id: uuid.to_string(),
-                    status: Some(ZoneStatus {
-                        state: ZoneState::Creating.into(),
-                        network_status: None,
-                        exit_status: None,
-                        error_status: None,
-                        resource_status: None,
-                        host: self.glt.host_uuid().to_string(),
-                        domid: u32::MAX,
-                    }),
-                    spec: Some(spec),
-                },
-            )
-            .await
-            .map_err(ApiError::from)?;
-        self.zone_reconciler_notify
-            .send(uuid)
-            .await
-            .map_err(|x| ApiError {
-                message: x.to_string(),
-            })?;
-        Ok(Response::new(CreateZoneReply {
-            zone_id: uuid.to_string(),
-        }))
-    }
-
-    async fn exec_inside_zone(
-        &self,
-        request: Request<Streaming<ExecInsideZoneRequest>>,
-    ) -> Result<Response<Self::ExecInsideZoneStream>, Status> {
-        let mut input = request.into_inner();
-        let Some(request) = input.next().await else {
-            return Err(ApiError {
-                message: "expected to have at least one request".to_string(),
-            }
-            .into());
-        };
-        let request = request?;
-
-        let Some(task) = request.task else {
-            return Err(ApiError {
-                message: "task is missing".to_string(),
-            }
-            .into());
-        };
-
-        let uuid = Uuid::from_str(&request.zone_id).map_err(|error| ApiError {
-            message: error.to_string(),
-        })?;
-        let idm = self.idm.client(uuid).await.map_err(|error| ApiError {
-            message: error.to_string(),
-        })?;
-
-        let idm_request = IdmRequest {
-            request: Some(IdmRequestType::ExecStream(ExecStreamRequestUpdate {
-                update: Some(Update::Start(ExecStreamRequestStart {
-                    environment: task
-                        .environment
-                        .into_iter()
-                        .map(|x| ExecEnvVar {
-                            key: x.key,
-                            value: x.value,
-                        })
-                        .collect(),
-                    command: task.command,
-                    working_directory: task.working_directory,
-                    tty: task.tty,
-                })),
-            })),
-        };
-
-        let output = try_stream! {
-            let mut handle = idm.send_stream(idm_request).await.map_err(|x| ApiError {
-                message: x.to_string(),
-            })?;
-
-            loop {
-                select! {
-                    x = input.next() => if let Some(update) = x {
-                        let update: Result<ExecInsideZoneRequest, Status> = update.map_err(|error| ApiError {
-                            message: error.to_string()
-                        }.into());
-
-                        if let Ok(update) = update {
-                            if !update.stdin.is_empty() {
-                                let _ = handle.update(IdmRequest {
-                                    request: Some(IdmRequestType::ExecStream(ExecStreamRequestUpdate {
-                                        update: Some(Update::Stdin(ExecStreamRequestStdin {
-                                            data: update.stdin,
-                                            closed: update.stdin_closed,
-                                        })),
-                                    }))}).await;
-                            }
-                        }
-                    },
-                    x = handle.receiver.recv() => match x {
-                        Some(response) => {
-                            let Some(IdmResponseType::ExecStream(update)) = response.response else {
-                                break;
-                            };
-                            let reply = ExecInsideZoneReply {
-                                exited: update.exited,
-                                error: update.error,
-                                exit_code: update.exit_code,
-                                stdout: update.stdout,
-                                stderr: update.stderr,
-                            };
-                            yield reply;
-                        },
-                        None => {
-                            break;
-                        }
-                    }
-                }
-            }
-        };
-
-        Ok(Response::new(Box::pin(output) as Self::ExecInsideZoneStream))
-    }
-
-    async fn destroy_zone(
-        &self,
-        request: Request<DestroyZoneRequest>,
-    ) -> Result<Response<DestroyZoneReply>, Status> {
-        let request = request.into_inner();
-        let uuid = Uuid::from_str(&request.zone_id).map_err(|error| ApiError {
-            message: error.to_string(),
-        })?;
-        let Some(mut zone) = self.zones.read(uuid).await.map_err(ApiError::from)? else {
-            return Err(ApiError {
-                message: "zone not found".to_string(),
-            }
-            .into());
-        };
-
-        zone.status = Some(zone.status.as_mut().cloned().unwrap_or_default());
-
-        if zone.status.as_ref().unwrap().state() == ZoneState::Destroyed {
-            return Err(ApiError {
-                message: "zone already destroyed".to_string(),
-            }
-            .into());
-        }
-
-        zone.status.as_mut().unwrap().state = ZoneState::Destroying.into();
-        self.zones
-            .update(uuid, zone)
-            .await
-            .map_err(ApiError::from)?;
-        self.zone_reconciler_notify
-            .send(uuid)
-            .await
-            .map_err(|x| ApiError {
-                message: x.to_string(),
-            })?;
-        Ok(Response::new(DestroyZoneReply {}))
-    }
-
-    async fn list_zones(
-        &self,
-        request: Request<ListZonesRequest>,
-    ) -> Result<Response<ListZonesReply>, Status> {
-        let _ = request.into_inner();
-        let zones = self.zones.list().await.map_err(ApiError::from)?;
-        let zones = zones.into_values().collect::<Vec<Zone>>();
-        Ok(Response::new(ListZonesReply { zones }))
-    }
-
-    async fn resolve_zone_id(
-        &self,
-        request: Request<ResolveZoneIdRequest>,
-    ) -> Result<Response<ResolveZoneIdReply>, Status> {
-        let request = request.into_inner();
-        let zones = self.zones.list().await.map_err(ApiError::from)?;
-        let zones = zones
-            .into_values()
-            .filter(|x| {
-                let comparison_spec = x.spec.as_ref().cloned().unwrap_or_default();
-                (!request.name.is_empty() && comparison_spec.name == request.name)
-                    || x.id == request.name
-            })
-            .collect::<Vec<Zone>>();
-        Ok(Response::new(ResolveZoneIdReply {
-            zone_id: zones.first().cloned().map(|x| x.id).unwrap_or_default(),
-        }))
-    }
-
-    async fn attach_zone_console(
-        &self,
-        request: Request<Streaming<ZoneConsoleRequest>>,
-    ) -> Result<Response<Self::AttachZoneConsoleStream>, Status> {
-        let mut input = request.into_inner();
-        let Some(request) = input.next().await else {
-            return Err(ApiError {
-                message: "expected to have at least one request".to_string(),
-            }
-            .into());
-        };
-        let request = request?;
-        let uuid = Uuid::from_str(&request.zone_id).map_err(|error| ApiError {
-            message: error.to_string(),
-        })?;
-        let (sender, mut receiver) = channel(100);
-        let console = self
-            .console
-            .attach(uuid, sender)
-            .await
-            .map_err(|error| ApiError {
-                message: format!("failed to attach to console: {}", error),
-            })?;
-
-        let output = try_stream! {
-            yield ZoneConsoleReply { data: console.initial.clone(), };
-            loop {
-                let what = select! {
-                    x = receiver.recv() => ConsoleDataSelect::Read(x),
-                    x = input.next() => ConsoleDataSelect::Write(x),
-                };
-
-                match what {
-                    ConsoleDataSelect::Read(Some(data)) => {
-                        yield ZoneConsoleReply { data, };
-                    },
-
-                    ConsoleDataSelect::Read(None) => {
-                        break;
-                    }
-
-                    ConsoleDataSelect::Write(Some(request)) => {
-                        let request = request?;
-                        if !request.data.is_empty() {
-                            console.send(request.data).await.map_err(|error| ApiError {
-                                message: error.to_string(),
-                            })?;
-                        }
-                    },
-
-                    ConsoleDataSelect::Write(None) => {
-                        break;
-                    }
-                }
-            }
-        };
-
-        Ok(Response::new(
-            Box::pin(output) as Self::AttachZoneConsoleStream
-        ))
-    }
-
-    async fn read_zone_metrics(
-        &self,
-        request: Request<ReadZoneMetricsRequest>,
-    ) -> Result<Response<ReadZoneMetricsReply>, Status> {
-        let request = request.into_inner();
-        let uuid = Uuid::from_str(&request.zone_id).map_err(|error| ApiError {
-            message: error.to_string(),
-        })?;
-        let client = self.idm.client(uuid).await.map_err(|error| ApiError {
-            message: error.to_string(),
-        })?;
-
-        let response = client
-            .send(IdmRequest {
-                request: Some(IdmRequestType::Metrics(MetricsRequest {})),
-            })
-            .await
-            .map_err(|error| ApiError {
-                message: error.to_string(),
-            })?;
-
-        let mut reply = ReadZoneMetricsReply::default();
-        if let Some(IdmResponseType::Metrics(metrics)) = response.response {
-            reply.root = metrics.root.map(idm_metric_to_api);
-        }
-        Ok(Response::new(reply))
-    }
-
-    async fn pull_image(
-        &self,
-        request: Request<PullImageRequest>,
-    ) -> Result<Response<Self::PullImageStream>, Status> {
-        let request = request.into_inner();
-        let name = ImageName::parse(&request.image).map_err(|err| ApiError {
-            message: err.to_string(),
-        })?;
-        let format = match request.format() {
-            OciImageFormat::Unknown => OciPackedFormat::Squashfs,
-            OciImageFormat::Squashfs => OciPackedFormat::Squashfs,
-            OciImageFormat::Erofs => OciPackedFormat::Erofs,
-            OciImageFormat::Tar => OciPackedFormat::Tar,
-        };
-        let (context, mut receiver) = OciProgressContext::create();
-        let our_packer = self.packer.clone();
-
-        let output = try_stream! {
-            let mut task = tokio::task::spawn(async move {
-                our_packer.request(name, format, request.overwrite_cache, request.update, context).await
-            });
-            let abort_handle = task.abort_handle();
-            let _task_cancel_guard = scopeguard::guard(abort_handle, |handle| {
-                handle.abort();
-            });
-
-            loop {
-                let what = select! {
-                    x = receiver.changed() => match x {
-                        Ok(_) => PullImageSelect::Progress(Some(receiver.borrow_and_update().clone())),
-                        Err(_) => PullImageSelect::Progress(None),
-                    },
-                    x = &mut task => PullImageSelect::Completed(x),
-                };
-                match what {
-                    PullImageSelect::Progress(Some(progress)) => {
-                        let reply = PullImageReply {
-                            progress: Some(convert_oci_progress(progress)),
-                            digest: String::new(),
-                            format: OciImageFormat::Unknown.into(),
-                        };
-                        yield reply;
-                    },
-
-                    PullImageSelect::Completed(result) => {
-                        let result = result.map_err(|err| ApiError {
-                            message: err.to_string(),
-                        })?;
-                        let packed = result.map_err(|err| ApiError {
-                            message: err.to_string(),
-                        })?;
-                        let reply = PullImageReply {
-                            progress: None,
-                            digest: packed.digest,
-                            format: match packed.format {
-                                OciPackedFormat::Squashfs => OciImageFormat::Squashfs.into(),
-                                OciPackedFormat::Erofs => OciImageFormat::Erofs.into(),
-                                OciPackedFormat::Tar => OciImageFormat::Tar.into(),
-                            },
-                        };
-                        yield reply;
-                        break;
-                    },
-
-                    _ => {
-                        continue;
-                    }
-                }
-            }
-        };
-        Ok(Response::new(Box::pin(output) as Self::PullImageStream))
-    }
-
-    async fn watch_events(
-        &self,
-        request: Request<WatchEventsRequest>,
-    ) -> Result<Response<Self::WatchEventsStream>, Status> {
-        let _ = request.into_inner();
-        let mut events = self.events.subscribe();
-        let output = try_stream! {
-            while let Ok(event) = events.recv().await {
-                yield WatchEventsReply { event: Some(event), };
-            }
-        };
-        Ok(Response::new(Box::pin(output) as Self::WatchEventsStream))
-    }
-
-    async fn snoop_idm(
-        &self,
-        request: Request<SnoopIdmRequest>,
-    ) -> Result<Response<Self::SnoopIdmStream>, Status> {
-        let _ = request.into_inner();
-        let mut messages = self.idm.snoop();
-        let glt = self.glt.clone();
-        let output = try_stream! {
-            while let Ok(event) = messages.recv().await {
-                let Some(from_uuid) = glt.lookup_uuid_by_domid(event.from).await else {
-                    continue;
-                };
-                let Some(to_uuid) = glt.lookup_uuid_by_domid(event.to).await else {
-                    continue;
-                };
-                yield SnoopIdmReply { from: from_uuid.to_string(), to: to_uuid.to_string(), packet: Some(event.packet) };
-            }
-        };
-        Ok(Response::new(Box::pin(output) as Self::SnoopIdmStream))
-    }
-
-    async fn list_devices(
-        &self,
-        request: Request<ListDevicesRequest>,
-    ) -> Result<Response<ListDevicesReply>, Status> {
-        let _ = request.into_inner();
-        let mut devices = Vec::new();
-        let state = self.devices.copy().await.map_err(|error| ApiError {
-            message: error.to_string(),
-        })?;
-        for (name, state) in state {
-            devices.push(DeviceInfo {
-                name,
-                claimed: state.owner.is_some(),
-                owner: state.owner.map(|x| x.to_string()).unwrap_or_default(),
-            });
-        }
-        Ok(Response::new(ListDevicesReply { devices }))
-    }
-
-    async fn get_host_cpu_topology(
-        &self,
-        request: Request<GetHostCpuTopologyRequest>,
-    ) -> Result<Response<GetHostCpuTopologyReply>, Status> {
-        let _ = request.into_inner();
-        let power = self
-            .runtime
-            .power_management_context()
-            .await
-            .map_err(ApiError::from)?;
-        let cputopo = power.cpu_topology().await.map_err(ApiError::from)?;
-        let mut cpus = vec![];
-
-        for cpu in cputopo {
-            cpus.push(HostCpuTopologyInfo {
-                core: cpu.core,
-                socket: cpu.socket,
-                node: cpu.node,
-                thread: cpu.thread,
-                class: cpu.class as i32,
-            })
-        }
-
-        Ok(Response::new(GetHostCpuTopologyReply { cpus }))
-    }
-
-    async fn set_host_power_management_policy(
-        &self,
-        request: Request<SetHostPowerManagementPolicyRequest>,
-    ) -> Result<Response<SetHostPowerManagementPolicyReply>, Status> {
-        let policy = request.into_inner();
-        let power = self
-            .runtime
-            .power_management_context()
-            .await
-            .map_err(ApiError::from)?;
-        let scheduler = &policy.scheduler;
-
-        power
-            .set_smt_policy(policy.smt_awareness)
-            .await
-            .map_err(ApiError::from)?;
-        power
-            .set_scheduler_policy(scheduler)
-            .await
-            .map_err(ApiError::from)?;
-
-        Ok(Response::new(SetHostPowerManagementPolicyReply {}))
-    }
-
-    async fn get_zone(
-        &self,
-        request: Request<GetZoneRequest>,
-    ) -> Result<Response<GetZoneReply>, Status> {
-        let request = request.into_inner();
-        let zones = self.zones.list().await.map_err(ApiError::from)?;
-        let zone = zones.get(&Uuid::from_str(&request.zone_id).map_err(|error| ApiError {
-            message: error.to_string(),
-        })?);
-        Ok(Response::new(GetZoneReply {
-            zone: zone.cloned(),
-        }))
-    }
-
-    async fn update_zone_resources(
-        &self,
-        request: Request<UpdateZoneResourcesRequest>,
-    ) -> Result<Response<UpdateZoneResourcesReply>, Status> {
-        let request = request.into_inner();
-        let uuid = Uuid::from_str(&request.zone_id).map_err(|error| ApiError {
-            message: error.to_string(),
-        })?;
-        let Some(mut zone) = self.zones.read(uuid).await.map_err(ApiError::from)? else {
-            return Err(ApiError {
-                message: "zone not found".to_string(),
-            }
-            .into());
-        };
-
-        let Some(ref mut status) = zone.status else {
-            return Err(ApiError {
-                message: "zone state not available".to_string(),
-            }
-            .into());
-        };
-
-        if status.state() != ZoneState::Created {
-            return Err(ApiError {
-                message: "zone is in an invalid state".to_string(),
-            }
-            .into());
-        }
-
-        if status.domid == 0 || status.domid == u32::MAX {
-            return Err(ApiError {
-                message: "zone domid is invalid".to_string(),
-            }
-            .into());
-        }
-
-        let mut resources = request.resources.unwrap_or_default();
-        if resources.target_memory > resources.max_memory {
-            resources.max_memory = resources.target_memory;
-        }
-
-        if resources.target_cpus < 1 {
-            resources.target_cpus = 1;
-        }
-
-        let initial_resources = zone
-            .spec
-            .clone()
-            .unwrap_or_default()
-            .initial_resources
-            .unwrap_or_default();
-        if resources.target_cpus > initial_resources.max_cpus {
-            resources.target_cpus = initial_resources.max_cpus;
-        }
-        resources.max_cpus = initial_resources.max_cpus;
-
-        self.runtime
-            .set_memory_resources(
-                status.domid,
-                resources.target_memory * 1024 * 1024,
-                resources.max_memory * 1024 * 1024,
-            )
-            .await
-            .map_err(|error| ApiError {
-                message: format!("failed to set memory resources: {}", error),
-            })?;
-        self.runtime
-            .set_cpu_resources(status.domid, resources.target_cpus)
-            .await
-            .map_err(|error| ApiError {
-                message: format!("failed to set cpu resources: {}", error),
-            })?;
-        status.resource_status = Some(ZoneResourceStatus {
-            active_resources: Some(resources),
-        });
-
-        self.zones
-            .update(uuid, zone)
-            .await
-            .map_err(ApiError::from)?;
-        Ok(Response::new(UpdateZoneResourcesReply {}))
-    }
-}

crates/daemon/src/control/attach_zone_console.rs

@@ -0,0 +1,84 @@
+use std::pin::Pin;
+use std::str::FromStr;
+
+use anyhow::{anyhow, Result};
+use async_stream::try_stream;
+use tokio::select;
+use tokio::sync::mpsc::channel;
+use tokio_stream::{Stream, StreamExt};
+use tonic::{Status, Streaming};
+use uuid::Uuid;
+
+use krata::v1::control::{ZoneConsoleReply, ZoneConsoleRequest};
+
+use crate::console::DaemonConsoleHandle;
+use crate::control::ApiError;
+
+enum ConsoleDataSelect {
+    Read(Option<Vec<u8>>),
+    Write(Option<Result<ZoneConsoleRequest, Status>>),
+}
+
+pub struct AttachZoneConsoleRpc {
+    console: DaemonConsoleHandle,
+}
+
+impl AttachZoneConsoleRpc {
+    pub fn new(console: DaemonConsoleHandle) -> Self {
+        Self { console }
+    }
+
+    pub async fn process(
+        self,
+        mut input: Streaming<ZoneConsoleRequest>,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<ZoneConsoleReply, Status>> + Send + 'static>>>
+    {
+        let Some(request) = input.next().await else {
+            return Err(anyhow!("expected to have at least one request"));
+        };
+        let request = request?;
+        let uuid = Uuid::from_str(&request.zone_id)?;
+        let (sender, mut receiver) = channel(100);
+        let console = self
+            .console
+            .attach(uuid, sender)
+            .await
+            .map_err(|error| anyhow!("failed to attach to console: {}", error))?;
+
+        let output = try_stream! {
+            if request.replay_history {
+                yield ZoneConsoleReply { data: console.initial.clone(), };
+            }
+            loop {
+                let what = select! {
+                    x = receiver.recv() => ConsoleDataSelect::Read(x),
+                    x = input.next() => ConsoleDataSelect::Write(x),
+                };
+
+                match what {
+                    ConsoleDataSelect::Read(Some(data)) => {
+                        yield ZoneConsoleReply { data, };
+                    },
+
+                    ConsoleDataSelect::Read(None) => {
+                        break;
+                    }
+
+                    ConsoleDataSelect::Write(Some(request)) => {
+                        let request = request?;
+                        if !request.data.is_empty() {
+                            console.send(request.data).await.map_err(|error| ApiError {
+                                message: error.to_string(),
+                            })?;
+                        }
+                    },
+
+                    ConsoleDataSelect::Write(None) => {
+                        break;
+                    }
+                }
+            }
+        };
+        Ok(Box::pin(output))
+    }
+}

crates/daemon/src/control/create_zone.rs

@@ -0,0 +1,56 @@
+use crate::db::zone::ZoneStore;
+use crate::zlt::ZoneLookupTable;
+use anyhow::{anyhow, Result};
+use krata::v1::common::{Zone, ZoneState, ZoneStatus};
+use krata::v1::control::{CreateZoneReply, CreateZoneRequest};
+use tokio::sync::mpsc::Sender;
+use uuid::Uuid;
+
+pub struct CreateZoneRpc {
+    zones: ZoneStore,
+    zlt: ZoneLookupTable,
+    zone_reconciler_notify: Sender<Uuid>,
+}
+
+impl CreateZoneRpc {
+    pub fn new(
+        zones: ZoneStore,
+        zlt: ZoneLookupTable,
+        zone_reconciler_notify: Sender<Uuid>,
+    ) -> Self {
+        Self {
+            zones,
+            zlt,
+            zone_reconciler_notify,
+        }
+    }
+
+    pub async fn process(self, request: CreateZoneRequest) -> Result<CreateZoneReply> {
+        let Some(spec) = request.spec else {
+            return Err(anyhow!("zone spec not provided"));
+        };
+        let uuid = Uuid::new_v4();
+        self.zones
+            .update(
+                uuid,
+                Zone {
+                    id: uuid.to_string(),
+                    status: Some(ZoneStatus {
+                        state: ZoneState::Creating.into(),
+                        network_status: None,
+                        exit_status: None,
+                        error_status: None,
+                        resource_status: None,
+                        host: self.zlt.host_uuid().to_string(),
+                        domid: u32::MAX,
+                    }),
+                    spec: Some(spec),
+                },
+            )
+            .await?;
+        self.zone_reconciler_notify.send(uuid).await?;
+        Ok(CreateZoneReply {
+            zone_id: uuid.to_string(),
+        })
+    }
+}

crates/daemon/src/control/destroy_zone.rs

@@ -0,0 +1,42 @@
+use std::str::FromStr;
+
+use anyhow::{anyhow, Result};
+use tokio::sync::mpsc::Sender;
+use uuid::Uuid;
+
+use krata::v1::common::ZoneState;
+use krata::v1::control::{DestroyZoneReply, DestroyZoneRequest};
+
+use crate::db::zone::ZoneStore;
+
+pub struct DestroyZoneRpc {
+    zones: ZoneStore,
+    zone_reconciler_notify: Sender<Uuid>,
+}
+
+impl DestroyZoneRpc {
+    pub fn new(zones: ZoneStore, zone_reconciler_notify: Sender<Uuid>) -> Self {
+        Self {
+            zones,
+            zone_reconciler_notify,
+        }
+    }
+
+    pub async fn process(self, request: DestroyZoneRequest) -> Result<DestroyZoneReply> {
+        let uuid = Uuid::from_str(&request.zone_id)?;
+        let Some(mut zone) = self.zones.read(uuid).await? else {
+            return Err(anyhow!("zone not found"));
+        };
+
+        zone.status = Some(zone.status.as_mut().cloned().unwrap_or_default());
+
+        if zone.status.as_ref().unwrap().state() == ZoneState::Destroyed {
+            return Err(anyhow!("zone already destroyed"));
+        }
+
+        zone.status.as_mut().unwrap().state = ZoneState::Destroying.into();
+        self.zones.update(uuid, zone).await?;
+        self.zone_reconciler_notify.send(uuid).await?;
+        Ok(DestroyZoneReply {})
+    }
+}

crates/daemon/src/control/exec_inside_zone.rs

@@ -0,0 +1,116 @@
+use std::pin::Pin;
+use std::str::FromStr;
+
+use anyhow::{anyhow, Result};
+use async_stream::try_stream;
+use tokio::select;
+use tokio_stream::{Stream, StreamExt};
+use tonic::{Status, Streaming};
+use uuid::Uuid;
+
+use krata::idm::internal::Request;
+use krata::{
+    idm::internal::{
+        exec_stream_request_update::Update, request::Request as IdmRequestType,
+        response::Response as IdmResponseType, ExecEnvVar, ExecStreamRequestStart,
+        ExecStreamRequestStdin, ExecStreamRequestUpdate, Request as IdmRequest,
+    },
+    v1::control::{ExecInsideZoneReply, ExecInsideZoneRequest},
+};
+
+use crate::control::ApiError;
+use crate::idm::DaemonIdmHandle;
+
+pub struct ExecInsideZoneRpc {
+    idm: DaemonIdmHandle,
+}
+
+impl ExecInsideZoneRpc {
+    pub fn new(idm: DaemonIdmHandle) -> Self {
+        Self { idm }
+    }
+
+    pub async fn process(
+        self,
+        mut input: Streaming<ExecInsideZoneRequest>,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<ExecInsideZoneReply, Status>> + Send + 'static>>>
+    {
+        let Some(request) = input.next().await else {
+            return Err(anyhow!("expected to have at least one request"));
+        };
+        let request = request?;
+
+        let Some(task) = request.task else {
+            return Err(anyhow!("task is missing"));
+        };
+
+        let uuid = Uuid::from_str(&request.zone_id)?;
+        let idm = self.idm.client(uuid).await?;
+
+        let idm_request = Request {
+            request: Some(IdmRequestType::ExecStream(ExecStreamRequestUpdate {
+                update: Some(Update::Start(ExecStreamRequestStart {
+                    environment: task
+                        .environment
+                        .into_iter()
+                        .map(|x| ExecEnvVar {
+                            key: x.key,
+                            value: x.value,
+                        })
+                        .collect(),
+                    command: task.command,
+                    working_directory: task.working_directory,
+                    tty: task.tty,
+                })),
+            })),
+        };
+
+        let output = try_stream! {
+            let mut handle = idm.send_stream(idm_request).await.map_err(|x| ApiError {
+                message: x.to_string(),
+            })?;
+
+            loop {
+                select! {
+                    x = input.next() => if let Some(update) = x {
+                        let update: Result<ExecInsideZoneRequest, Status> = update.map_err(|error| ApiError {
+                            message: error.to_string()
+                        }.into());
+
+                        if let Ok(update) = update {
+                            if !update.stdin.is_empty() {
+                                let _ = handle.update(IdmRequest {
+                                    request: Some(IdmRequestType::ExecStream(ExecStreamRequestUpdate {
+                                        update: Some(Update::Stdin(ExecStreamRequestStdin {
+                                            data: update.stdin,
+                                            closed: update.stdin_closed,
+                                        })),
+                                    }))}).await;
+                            }
+                        }
+                    },
+                    x = handle.receiver.recv() => match x {
+                        Some(response) => {
+                            let Some(IdmResponseType::ExecStream(update)) = response.response else {
+                                break;
+                            };
+                            let reply = ExecInsideZoneReply {
+                                exited: update.exited,
+                                error: update.error,
+                                exit_code: update.exit_code,
+                                stdout: update.stdout,
+                                stderr: update.stderr,
+                            };
+                            yield reply;
+                        },
+                        None => {
+                            break;
+                        }
+                    }
+                }
+            }
+        };
+
+        Ok(Box::pin(output))
+    }
+}

crates/daemon/src/control/get_host_cpu_topology.rs

@@ -0,0 +1,33 @@
+use anyhow::Result;
+use krata::v1::control::{GetHostCpuTopologyReply, GetHostCpuTopologyRequest, HostCpuTopologyInfo};
+use kratart::Runtime;
+
+pub struct GetHostCpuTopologyRpc {
+    runtime: Runtime,
+}
+
+impl GetHostCpuTopologyRpc {
+    pub fn new(runtime: Runtime) -> Self {
+        Self { runtime }
+    }
+
+    pub async fn process(
+        self,
+        _request: GetHostCpuTopologyRequest,
+    ) -> Result<GetHostCpuTopologyReply> {
+        let power = self.runtime.power_management_context().await?;
+        let cpu_topology = power.cpu_topology().await?;
+        let mut cpus = vec![];
+
+        for cpu in cpu_topology {
+            cpus.push(HostCpuTopologyInfo {
+                core: cpu.core,
+                socket: cpu.socket,
+                node: cpu.node,
+                thread: cpu.thread,
+                class: cpu.class as i32,
+            })
+        }
+        Ok(GetHostCpuTopologyReply { cpus })
+    }
+}

crates/daemon/src/control/get_host_status.rs

@@ -0,0 +1,37 @@
+use crate::command::DaemonCommand;
+use crate::ip::assignment::IpAssignment;
+use crate::zlt::ZoneLookupTable;
+use anyhow::Result;
+use krata::v1::control::{GetHostStatusReply, GetHostStatusRequest};
+
+pub struct GetHostStatusRpc {
+    ip: IpAssignment,
+    zlt: ZoneLookupTable,
+}
+
+impl GetHostStatusRpc {
+    pub fn new(ip: IpAssignment, zlt: ZoneLookupTable) -> Self {
+        Self { ip, zlt }
+    }
+
+    pub async fn process(self, _request: GetHostStatusRequest) -> Result<GetHostStatusReply> {
+        let host_reservation = self.ip.retrieve(self.zlt.host_uuid()).await?;
+        Ok(GetHostStatusReply {
+            host_domid: self.zlt.host_domid(),
+            host_uuid: self.zlt.host_uuid().to_string(),
+            krata_version: DaemonCommand::version(),
+            host_ipv4: host_reservation
+                .as_ref()
+                .map(|x| format!("{}/{}", x.ipv4, x.ipv4_prefix))
+                .unwrap_or_default(),
+            host_ipv6: host_reservation
+                .as_ref()
+                .map(|x| format!("{}/{}", x.ipv6, x.ipv6_prefix))
+                .unwrap_or_default(),
+            host_mac: host_reservation
+                .as_ref()
+                .map(|x| x.mac.to_string().to_lowercase().replace('-', ":"))
+                .unwrap_or_default(),
+        })
+    }
+}

crates/daemon/src/control/get_zone.rs

@@ -0,0 +1,24 @@
+use std::str::FromStr;
+
+use anyhow::Result;
+use uuid::Uuid;
+
+use krata::v1::control::{GetZoneReply, GetZoneRequest};
+
+use crate::db::zone::ZoneStore;
+
+pub struct GetZoneRpc {
+    zones: ZoneStore,
+}
+
+impl GetZoneRpc {
+    pub fn new(zones: ZoneStore) -> Self {
+        Self { zones }
+    }
+
+    pub async fn process(self, request: GetZoneRequest) -> Result<GetZoneReply> {
+        let mut zones = self.zones.list().await?;
+        let zone = zones.remove(&Uuid::from_str(&request.zone_id)?);
+        Ok(GetZoneReply { zone })
+    }
+}

crates/daemon/src/control/list_devices.rs

@@ -0,0 +1,28 @@
+use anyhow::Result;
+
+use krata::v1::control::{DeviceInfo, ListDevicesReply, ListDevicesRequest};
+
+use crate::devices::DaemonDeviceManager;
+
+pub struct ListDevicesRpc {
+    devices: DaemonDeviceManager,
+}
+
+impl ListDevicesRpc {
+    pub fn new(devices: DaemonDeviceManager) -> Self {
+        Self { devices }
+    }
+
+    pub async fn process(self, _request: ListDevicesRequest) -> Result<ListDevicesReply> {
+        let mut devices = Vec::new();
+        let state = self.devices.copy().await?;
+        for (name, state) in state {
+            devices.push(DeviceInfo {
+                name,
+                claimed: state.owner.is_some(),
+                owner: state.owner.map(|x| x.to_string()).unwrap_or_default(),
+            });
+        }
+        Ok(ListDevicesReply { devices })
+    }
+}

crates/daemon/src/control/list_zones.rs

@@ -0,0 +1,21 @@
+use anyhow::Result;
+use krata::v1::common::Zone;
+use krata::v1::control::{ListZonesReply, ListZonesRequest};
+
+use crate::db::zone::ZoneStore;
+
+pub struct ListZonesRpc {
+    zones: ZoneStore,
+}
+
+impl ListZonesRpc {
+    pub fn new(zones: ZoneStore) -> Self {
+        Self { zones }
+    }
+
+    pub async fn process(self, _request: ListZonesRequest) -> Result<ListZonesReply> {
+        let zones = self.zones.list().await?;
+        let zones = zones.into_values().collect::<Vec<Zone>>();
+        Ok(ListZonesReply { zones })
+    }
+}

crates/daemon/src/control/mod.rs

@@ -0,0 +1,351 @@
+use std::pin::Pin;
+
+use anyhow::Error;
+use futures::Stream;
+use tokio::sync::mpsc::Sender;
+use tonic::{Request, Response, Status, Streaming};
+use uuid::Uuid;
+
+use krata::v1::control::{
+    control_service_server::ControlService, CreateZoneReply, CreateZoneRequest, DestroyZoneReply,
+    DestroyZoneRequest, ExecInsideZoneReply, ExecInsideZoneRequest, GetHostCpuTopologyReply,
+    GetHostCpuTopologyRequest, GetHostStatusReply, GetHostStatusRequest, ListDevicesReply,
+    ListDevicesRequest, ListZonesReply, ListZonesRequest, PullImageReply, PullImageRequest,
+    ReadHypervisorConsoleReply, ReadHypervisorConsoleRequest, ReadZoneMetricsReply,
+    ReadZoneMetricsRequest, ResolveZoneIdReply, ResolveZoneIdRequest, SnoopIdmReply,
+    SnoopIdmRequest, UpdateZoneResourcesReply, UpdateZoneResourcesRequest, WatchEventsReply,
+    WatchEventsRequest, ZoneConsoleReply, ZoneConsoleRequest,
+};
+use krata::v1::control::{
+    GetZoneReply, GetZoneRequest, SetHostPowerManagementPolicyReply,
+    SetHostPowerManagementPolicyRequest,
+};
+use krataoci::packer::service::OciPackerService;
+use kratart::Runtime;
+
+use crate::control::attach_zone_console::AttachZoneConsoleRpc;
+use crate::control::create_zone::CreateZoneRpc;
+use crate::control::destroy_zone::DestroyZoneRpc;
+use crate::control::exec_inside_zone::ExecInsideZoneRpc;
+use crate::control::get_host_cpu_topology::GetHostCpuTopologyRpc;
+use crate::control::get_host_status::GetHostStatusRpc;
+use crate::control::get_zone::GetZoneRpc;
+use crate::control::list_devices::ListDevicesRpc;
+use crate::control::list_zones::ListZonesRpc;
+use crate::control::pull_image::PullImageRpc;
+use crate::control::read_hypervisor_console::ReadHypervisorConsoleRpc;
+use crate::control::read_zone_metrics::ReadZoneMetricsRpc;
+use crate::control::resolve_zone_id::ResolveZoneIdRpc;
+use crate::control::set_host_power_management_policy::SetHostPowerManagementPolicyRpc;
+use crate::control::snoop_idm::SnoopIdmRpc;
+use crate::control::update_zone_resources::UpdateZoneResourcesRpc;
+use crate::control::watch_events::WatchEventsRpc;
+use crate::db::zone::ZoneStore;
+use crate::ip::assignment::IpAssignment;
+use crate::{
+    console::DaemonConsoleHandle, devices::DaemonDeviceManager, event::DaemonEventContext,
+    idm::DaemonIdmHandle, zlt::ZoneLookupTable,
+};
+
+pub mod attach_zone_console;
+pub mod create_zone;
+pub mod destroy_zone;
+pub mod exec_inside_zone;
+pub mod get_host_cpu_topology;
+pub mod get_host_status;
+pub mod get_zone;
+pub mod list_devices;
+pub mod list_zones;
+pub mod pull_image;
+pub mod read_hypervisor_console;
+pub mod read_zone_metrics;
+pub mod resolve_zone_id;
+pub mod set_host_power_management_policy;
+pub mod snoop_idm;
+pub mod update_zone_resources;
+pub mod watch_events;
+
+pub struct ApiError {
+    message: String,
+}
+
+impl From<Error> for ApiError {
+    fn from(value: Error) -> Self {
+        ApiError {
+            message: value.to_string(),
+        }
+    }
+}
+
+impl From<ApiError> for Status {
+    fn from(value: ApiError) -> Self {
+        Status::unknown(value.message)
+    }
+}
+
+#[derive(Clone)]
+pub struct DaemonControlService {
+    zlt: ZoneLookupTable,
+    devices: DaemonDeviceManager,
+    events: DaemonEventContext,
+    console: DaemonConsoleHandle,
+    idm: DaemonIdmHandle,
+    zones: ZoneStore,
+    ip: IpAssignment,
+    zone_reconciler_notify: Sender<Uuid>,
+    packer: OciPackerService,
+    runtime: Runtime,
+}
+
+impl DaemonControlService {
+    #[allow(clippy::too_many_arguments)]
+    pub fn new(
+        zlt: ZoneLookupTable,
+        devices: DaemonDeviceManager,
+        events: DaemonEventContext,
+        console: DaemonConsoleHandle,
+        idm: DaemonIdmHandle,
+        zones: ZoneStore,
+        ip: IpAssignment,
+        zone_reconciler_notify: Sender<Uuid>,
+        packer: OciPackerService,
+        runtime: Runtime,
+    ) -> Self {
+        Self {
+            zlt,
+            devices,
+            events,
+            console,
+            idm,
+            zones,
+            ip,
+            zone_reconciler_notify,
+            packer,
+            runtime,
+        }
+    }
+}
+
+#[tonic::async_trait]
+impl ControlService for DaemonControlService {
+    async fn get_host_status(
+        &self,
+        request: Request<GetHostStatusRequest>,
+    ) -> Result<Response<GetHostStatusReply>, Status> {
+        let request = request.into_inner();
+        adapt(
+            GetHostStatusRpc::new(self.ip.clone(), self.zlt.clone())
+                .process(request)
+                .await,
+        )
+    }
+
+    type SnoopIdmStream =
+        Pin<Box<dyn Stream<Item = Result<SnoopIdmReply, Status>> + Send + 'static>>;
+
+    async fn snoop_idm(
+        &self,
+        request: Request<SnoopIdmRequest>,
+    ) -> Result<Response<Self::SnoopIdmStream>, Status> {
+        let request = request.into_inner();
+        adapt(
+            SnoopIdmRpc::new(self.idm.clone(), self.zlt.clone())
+                .process(request)
+                .await,
+        )
+    }
+
+    async fn get_host_cpu_topology(
+        &self,
+        request: Request<GetHostCpuTopologyRequest>,
+    ) -> Result<Response<GetHostCpuTopologyReply>, Status> {
+        let request = request.into_inner();
+        adapt(
+            GetHostCpuTopologyRpc::new(self.runtime.clone())
+                .process(request)
+                .await,
+        )
+    }
+
+    async fn set_host_power_management_policy(
+        &self,
+        request: Request<SetHostPowerManagementPolicyRequest>,
+    ) -> Result<Response<SetHostPowerManagementPolicyReply>, Status> {
+        let request = request.into_inner();
+        adapt(
+            SetHostPowerManagementPolicyRpc::new(self.runtime.clone())
+                .process(request)
+                .await,
+        )
+    }
+
+    async fn list_devices(
+        &self,
+        request: Request<ListDevicesRequest>,
+    ) -> Result<Response<ListDevicesReply>, Status> {
+        let request = request.into_inner();
+        adapt(
+            ListDevicesRpc::new(self.devices.clone())
+                .process(request)
+                .await,
+        )
+    }
+
+    type PullImageStream =
+        Pin<Box<dyn Stream<Item = Result<PullImageReply, Status>> + Send + 'static>>;
+
+    async fn pull_image(
+        &self,
+        request: Request<PullImageRequest>,
+    ) -> Result<Response<Self::PullImageStream>, Status> {
+        let request = request.into_inner();
+        adapt(
+            PullImageRpc::new(self.packer.clone())
+                .process(request)
+                .await,
+        )
+    }
+
+    async fn create_zone(
+        &self,
+        request: Request<CreateZoneRequest>,
+    ) -> Result<Response<CreateZoneReply>, Status> {
+        let request = request.into_inner();
+        adapt(
+            CreateZoneRpc::new(
+                self.zones.clone(),
+                self.zlt.clone(),
+                self.zone_reconciler_notify.clone(),
+            )
+            .process(request)
+            .await,
+        )
+    }
+
+    async fn destroy_zone(
+        &self,
+        request: Request<DestroyZoneRequest>,
+    ) -> Result<Response<DestroyZoneReply>, Status> {
+        let request = request.into_inner();
+        adapt(
+            DestroyZoneRpc::new(self.zones.clone(), self.zone_reconciler_notify.clone())
+                .process(request)
+                .await,
+        )
+    }
+
+    async fn resolve_zone_id(
+        &self,
+        request: Request<ResolveZoneIdRequest>,
+    ) -> Result<Response<ResolveZoneIdReply>, Status> {
+        let request = request.into_inner();
+        adapt(
+            ResolveZoneIdRpc::new(self.zones.clone())
+                .process(request)
+                .await,
+        )
+    }
+
+    async fn get_zone(
+        &self,
+        request: Request<GetZoneRequest>,
+    ) -> Result<Response<GetZoneReply>, Status> {
+        let request = request.into_inner();
+        adapt(GetZoneRpc::new(self.zones.clone()).process(request).await)
+    }
+
+    async fn update_zone_resources(
+        &self,
+        request: Request<UpdateZoneResourcesRequest>,
+    ) -> Result<Response<UpdateZoneResourcesReply>, Status> {
+        let request = request.into_inner();
+        adapt(
+            UpdateZoneResourcesRpc::new(self.runtime.clone(), self.zones.clone())
+                .process(request)
+                .await,
+        )
+    }
+
+    async fn list_zones(
+        &self,
+        request: Request<ListZonesRequest>,
+    ) -> Result<Response<ListZonesReply>, Status> {
+        let request = request.into_inner();
+        adapt(ListZonesRpc::new(self.zones.clone()).process(request).await)
+    }
+
+    type AttachZoneConsoleStream =
+        Pin<Box<dyn Stream<Item = Result<ZoneConsoleReply, Status>> + Send + 'static>>;
+
+    async fn attach_zone_console(
+        &self,
+        request: Request<Streaming<ZoneConsoleRequest>>,
+    ) -> Result<Response<Self::AttachZoneConsoleStream>, Status> {
+        let input = request.into_inner();
+        adapt(
+            AttachZoneConsoleRpc::new(self.console.clone())
+                .process(input)
+                .await,
+        )
+    }
+
+    type ExecInsideZoneStream =
+        Pin<Box<dyn Stream<Item = Result<ExecInsideZoneReply, Status>> + Send + 'static>>;
+
+    async fn exec_inside_zone(
+        &self,
+        request: Request<Streaming<ExecInsideZoneRequest>>,
+    ) -> Result<Response<Self::ExecInsideZoneStream>, Status> {
+        let input = request.into_inner();
+        adapt(
+            ExecInsideZoneRpc::new(self.idm.clone())
+                .process(input)
+                .await,
+        )
+    }
+
+    async fn read_zone_metrics(
+        &self,
+        request: Request<ReadZoneMetricsRequest>,
+    ) -> Result<Response<ReadZoneMetricsReply>, Status> {
+        let request = request.into_inner();
+        adapt(
+            ReadZoneMetricsRpc::new(self.idm.clone())
+                .process(request)
+                .await,
+        )
+    }
+
+    type WatchEventsStream =
+        Pin<Box<dyn Stream<Item = Result<WatchEventsReply, Status>> + Send + 'static>>;
+
+    async fn watch_events(
+        &self,
+        request: Request<WatchEventsRequest>,
+    ) -> Result<Response<Self::WatchEventsStream>, Status> {
+        let request = request.into_inner();
+        adapt(
+            WatchEventsRpc::new(self.events.clone())
+                .process(request)
+                .await,
+        )
+    }
+
+    async fn read_hypervisor_console(
+        &self,
+        request: Request<ReadHypervisorConsoleRequest>,
+    ) -> Result<Response<ReadHypervisorConsoleReply>, Status> {
+        let request = request.into_inner();
+        adapt(
+            ReadHypervisorConsoleRpc::new(self.runtime.clone())
+                .process(request)
+                .await,
+        )
+    }
+}
+
+fn adapt<T>(result: anyhow::Result<T>) -> Result<Response<T>, Status> {
+    result
+        .map(Response::new)
+        .map_err(|error| Status::unknown(error.to_string()))
+}

crates/daemon/src/control/pull_image.rs

@@ -0,0 +1,100 @@
+use crate::control::ApiError;
+use crate::oci::convert_oci_progress;
+use anyhow::Result;
+use async_stream::try_stream;
+use krata::v1::common::OciImageFormat;
+use krata::v1::control::{PullImageReply, PullImageRequest};
+use krataoci::name::ImageName;
+use krataoci::packer::service::OciPackerService;
+use krataoci::packer::{OciPackedFormat, OciPackedImage};
+use krataoci::progress::{OciProgress, OciProgressContext};
+use std::pin::Pin;
+use tokio::select;
+use tokio::task::JoinError;
+use tokio_stream::Stream;
+use tonic::Status;
+
+enum PullImageSelect {
+    Progress(Option<OciProgress>),
+    Completed(Result<Result<OciPackedImage, anyhow::Error>, JoinError>),
+}
+
+pub struct PullImageRpc {
+    packer: OciPackerService,
+}
+
+impl PullImageRpc {
+    pub fn new(packer: OciPackerService) -> Self {
+        Self { packer }
+    }
+
+    pub async fn process(
+        self,
+        request: PullImageRequest,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<PullImageReply, Status>> + Send + 'static>>> {
+        let name = ImageName::parse(&request.image)?;
+        let format = match request.format() {
+            OciImageFormat::Unknown => OciPackedFormat::Squashfs,
+            OciImageFormat::Squashfs => OciPackedFormat::Squashfs,
+            OciImageFormat::Erofs => OciPackedFormat::Erofs,
+            OciImageFormat::Tar => OciPackedFormat::Tar,
+        };
+        let (context, mut receiver) = OciProgressContext::create();
+        let our_packer = self.packer;
+
+        let output = try_stream! {
+            let mut task = tokio::task::spawn(async move {
+                our_packer.request(name, format, request.overwrite_cache, request.update, context).await
+            });
+            let abort_handle = task.abort_handle();
+            let _task_cancel_guard = scopeguard::guard(abort_handle, |handle| {
+                handle.abort();
+            });
+
+            loop {
+                let what = select! {
+                    x = receiver.changed() => match x {
+                        Ok(_) => PullImageSelect::Progress(Some(receiver.borrow_and_update().clone())),
+                        Err(_) => PullImageSelect::Progress(None),
+                    },
+                    x = &mut task => PullImageSelect::Completed(x),
+                };
+                match what {
+                    PullImageSelect::Progress(Some(progress)) => {
+                        let reply = PullImageReply {
+                            progress: Some(convert_oci_progress(progress)),
+                            digest: String::new(),
+                            format: OciImageFormat::Unknown.into(),
+                        };
+                        yield reply;
+                    },
+
+                    PullImageSelect::Completed(result) => {
+                        let result = result.map_err(|err| ApiError {
+                            message: err.to_string(),
+                        })?;
+                        let packed = result.map_err(|err| ApiError {
+                            message: err.to_string(),
+                        })?;
+                        let reply = PullImageReply {
+                            progress: None,
+                            digest: packed.digest,
+                            format: match packed.format {
+                                OciPackedFormat::Squashfs => OciImageFormat::Squashfs.into(),
+                                OciPackedFormat::Erofs => OciImageFormat::Erofs.into(),
+                                OciPackedFormat::Tar => OciImageFormat::Tar.into(),
+                            },
+                        };
+                        yield reply;
+                        break;
+                    },
+
+                    _ => {
+                        continue;
+                    }
+                }
+            }
+        };
+        Ok(Box::pin(output))
+    }
+}

crates/daemon/src/control/read_hypervisor_console.rs

@@ -0,0 +1,23 @@
+use anyhow::Result;
+use krata::v1::control::{ReadHypervisorConsoleReply, ReadHypervisorConsoleRequest};
+use kratart::Runtime;
+
+pub struct ReadHypervisorConsoleRpc {
+    runtime: Runtime,
+}
+
+impl ReadHypervisorConsoleRpc {
+    pub fn new(runtime: Runtime) -> Self {
+        Self { runtime }
+    }
+
+    pub async fn process(
+        self,
+        _: ReadHypervisorConsoleRequest,
+    ) -> Result<ReadHypervisorConsoleReply> {
+        let data = self.runtime.read_hypervisor_console(false).await?;
+        Ok(ReadHypervisorConsoleReply {
+            data: data.to_string(),
+        })
+    }
+}

crates/daemon/src/control/read_zone_metrics.rs

@@ -0,0 +1,40 @@
+use std::str::FromStr;
+
+use anyhow::Result;
+use uuid::Uuid;
+
+use krata::idm::internal::MetricsRequest;
+use krata::idm::internal::{
+    request::Request as IdmRequestType, response::Response as IdmResponseType,
+    Request as IdmRequest,
+};
+use krata::v1::control::{ReadZoneMetricsReply, ReadZoneMetricsRequest};
+
+use crate::idm::DaemonIdmHandle;
+use crate::metrics::idm_metric_to_api;
+
+pub struct ReadZoneMetricsRpc {
+    idm: DaemonIdmHandle,
+}
+
+impl ReadZoneMetricsRpc {
+    pub fn new(idm: DaemonIdmHandle) -> Self {
+        Self { idm }
+    }
+
+    pub async fn process(self, request: ReadZoneMetricsRequest) -> Result<ReadZoneMetricsReply> {
+        let uuid = Uuid::from_str(&request.zone_id)?;
+        let client = self.idm.client(uuid).await?;
+        let response = client
+            .send(IdmRequest {
+                request: Some(IdmRequestType::Metrics(MetricsRequest {})),
+            })
+            .await?;
+
+        let mut reply = ReadZoneMetricsReply::default();
+        if let Some(IdmResponseType::Metrics(metrics)) = response.response {
+            reply.root = metrics.root.map(idm_metric_to_api);
+        }
+        Ok(reply)
+    }
+}

crates/daemon/src/control/resolve_zone_id.rs

@@ -0,0 +1,30 @@
+use anyhow::Result;
+use krata::v1::common::Zone;
+use krata::v1::control::{ResolveZoneIdReply, ResolveZoneIdRequest};
+
+use crate::db::zone::ZoneStore;
+
+pub struct ResolveZoneIdRpc {
+    zones: ZoneStore,
+}
+
+impl ResolveZoneIdRpc {
+    pub fn new(zones: ZoneStore) -> Self {
+        Self { zones }
+    }
+
+    pub async fn process(self, request: ResolveZoneIdRequest) -> Result<ResolveZoneIdReply> {
+        let zones = self.zones.list().await?;
+        let zones = zones
+            .into_values()
+            .filter(|x| {
+                let comparison_spec = x.spec.as_ref().cloned().unwrap_or_default();
+                (!request.name.is_empty() && comparison_spec.name == request.name)
+                    || x.id == request.name
+            })
+            .collect::<Vec<Zone>>();
+        Ok(ResolveZoneIdReply {
+            zone_id: zones.first().cloned().map(|x| x.id).unwrap_or_default(),
+        })
+    }
+}

crates/daemon/src/control/set_host_power_management_policy.rs

@@ -0,0 +1,25 @@
+use anyhow::Result;
+use krata::v1::control::{SetHostPowerManagementPolicyReply, SetHostPowerManagementPolicyRequest};
+use kratart::Runtime;
+
+pub struct SetHostPowerManagementPolicyRpc {
+    runtime: Runtime,
+}
+
+impl SetHostPowerManagementPolicyRpc {
+    pub fn new(runtime: Runtime) -> Self {
+        Self { runtime }
+    }
+
+    pub async fn process(
+        self,
+        request: SetHostPowerManagementPolicyRequest,
+    ) -> Result<SetHostPowerManagementPolicyReply> {
+        let power = self.runtime.power_management_context().await?;
+        let scheduler = &request.scheduler;
+
+        power.set_smt_policy(request.smt_awareness).await?;
+        power.set_scheduler_policy(scheduler).await?;
+        Ok(SetHostPowerManagementPolicyReply {})
+    }
+}

crates/daemon/src/control/snoop_idm.rs

@@ -0,0 +1,39 @@
+use crate::idm::DaemonIdmHandle;
+use crate::zlt::ZoneLookupTable;
+use anyhow::Result;
+use async_stream::try_stream;
+use krata::v1::control::{SnoopIdmReply, SnoopIdmRequest};
+use std::pin::Pin;
+use tokio_stream::Stream;
+use tonic::Status;
+
+pub struct SnoopIdmRpc {
+    idm: DaemonIdmHandle,
+    zlt: ZoneLookupTable,
+}
+
+impl SnoopIdmRpc {
+    pub fn new(idm: DaemonIdmHandle, zlt: ZoneLookupTable) -> Self {
+        Self { idm, zlt }
+    }
+
+    pub async fn process(
+        self,
+        _request: SnoopIdmRequest,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<SnoopIdmReply, Status>> + Send + 'static>>> {
+        let mut messages = self.idm.snoop();
+        let zlt = self.zlt.clone();
+        let output = try_stream! {
+            while let Ok(event) = messages.recv().await {
+                let Some(from_uuid) = zlt.lookup_uuid_by_domid(event.from).await else {
+                    continue;
+                };
+                let Some(to_uuid) = zlt.lookup_uuid_by_domid(event.to).await else {
+                    continue;
+                };
+                yield SnoopIdmReply { from: from_uuid.to_string(), to: to_uuid.to_string(), packet: Some(event.packet) };
+            }
+        };
+        Ok(Box::pin(output))
+    }
+}

crates/daemon/src/control/update_zone_resources.rs

@@ -0,0 +1,82 @@
+use std::str::FromStr;
+
+use anyhow::{anyhow, Result};
+use uuid::Uuid;
+
+use krata::v1::common::{ZoneResourceStatus, ZoneState};
+use krata::v1::control::{UpdateZoneResourcesReply, UpdateZoneResourcesRequest};
+use kratart::Runtime;
+
+use crate::db::zone::ZoneStore;
+
+pub struct UpdateZoneResourcesRpc {
+    runtime: Runtime,
+    zones: ZoneStore,
+}
+
+impl UpdateZoneResourcesRpc {
+    pub fn new(runtime: Runtime, zones: ZoneStore) -> Self {
+        Self { runtime, zones }
+    }
+
+    pub async fn process(
+        self,
+        request: UpdateZoneResourcesRequest,
+    ) -> Result<UpdateZoneResourcesReply> {
+        let uuid = Uuid::from_str(&request.zone_id)?;
+        let Some(mut zone) = self.zones.read(uuid).await? else {
+            return Err(anyhow!("zone not found"));
+        };
+
+        let Some(ref mut status) = zone.status else {
+            return Err(anyhow!("zone state not available"));
+        };
+
+        if status.state() != ZoneState::Created {
+            return Err(anyhow!("zone is in an invalid state"));
+        }
+
+        if status.domid == 0 || status.domid == u32::MAX {
+            return Err(anyhow!("zone domid is invalid"));
+        }
+
+        let mut resources = request.resources.unwrap_or_default();
+        if resources.target_memory > resources.max_memory {
+            resources.max_memory = resources.target_memory;
+        }
+
+        if resources.target_cpus < 1 {
+            resources.target_cpus = 1;
+        }
+
+        let initial_resources = zone
+            .spec
+            .clone()
+            .unwrap_or_default()
+            .initial_resources
+            .unwrap_or_default();
+        if resources.target_cpus > initial_resources.max_cpus {
+            resources.target_cpus = initial_resources.max_cpus;
+        }
+        resources.max_cpus = initial_resources.max_cpus;
+
+        self.runtime
+            .set_memory_resources(
+                status.domid,
+                resources.target_memory * 1024 * 1024,
+                resources.max_memory * 1024 * 1024,
+            )
+            .await
+            .map_err(|error| anyhow!("failed to set memory resources: {}", error))?;
+        self.runtime
+            .set_cpu_resources(status.domid, resources.target_cpus)
+            .await
+            .map_err(|error| anyhow!("failed to set cpu resources: {}", error))?;
+        status.resource_status = Some(ZoneResourceStatus {
+            active_resources: Some(resources),
+        });
+
+        self.zones.update(uuid, zone).await?;
+        Ok(UpdateZoneResourcesReply {})
+    }
+}

crates/daemon/src/control/watch_events.rs

@@ -0,0 +1,31 @@
+use crate::event::DaemonEventContext;
+use anyhow::Result;
+use async_stream::try_stream;
+use krata::v1::control::{WatchEventsReply, WatchEventsRequest};
+use std::pin::Pin;
+use tokio_stream::Stream;
+use tonic::Status;
+
+pub struct WatchEventsRpc {
+    events: DaemonEventContext,
+}
+
+impl WatchEventsRpc {
+    pub fn new(events: DaemonEventContext) -> Self {
+        Self { events }
+    }
+
+    pub async fn process(
+        self,
+        _request: WatchEventsRequest,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<WatchEventsReply, Status>> + Send + 'static>>>
+    {
+        let mut events = self.events.subscribe();
+        let output = try_stream! {
+            while let Ok(event) = events.recv().await {
+                yield WatchEventsReply { event: Some(event), };
+            }
+        };
+        Ok(Box::pin(output))
+    }
+}

crates/daemon/src/idm.rs

@@ -31,7 +31,7 @@ type ClientMap = Arc<Mutex<HashMap<u32, IdmInternalClient>>>;
 
 #[derive(Clone)]
 pub struct DaemonIdmHandle {
-    glt: ZoneLookupTable,
+    zlt: ZoneLookupTable,
     clients: ClientMap,
     feeds: BackendFeedMap,
     tx_sender: Sender<(u32, IdmTransportPacket)>,
@@ -45,7 +45,7 @@ impl DaemonIdmHandle {
     }
 
     pub async fn client(&self, uuid: Uuid) -> Result<IdmInternalClient> {
-        let Some(domid) = self.glt.lookup_domid_by_uuid(&uuid).await else {
+        let Some(domid) = self.zlt.lookup_domid_by_uuid(&uuid).await else {
             return Err(anyhow!("unable to find domain {}", uuid));
         };
         self.client_by_domid(domid).await
@@ -72,7 +72,7 @@ pub struct DaemonIdmSnoopPacket {
 }
 
 pub struct DaemonIdm {
-    glt: ZoneLookupTable,
+    zlt: ZoneLookupTable,
     clients: ClientMap,
     feeds: BackendFeedMap,
     tx_sender: Sender<(u32, IdmTransportPacket)>,
@@ -84,7 +84,7 @@ pub struct DaemonIdm {
 }
 
 impl DaemonIdm {
-    pub async fn new(glt: ZoneLookupTable) -> Result<DaemonIdm> {
+    pub async fn new(zlt: ZoneLookupTable) -> Result<DaemonIdm> {
         debug!("allocating channel service for idm");
         let (service, tx_raw_sender, rx_receiver) =
             ChannelService::new("krata-channel".to_string(), None).await?;
@@ -98,7 +98,7 @@ impl DaemonIdm {
         let feeds = Arc::new(Mutex::new(HashMap::new()));
 
         Ok(DaemonIdm {
-            glt,
+            zlt,
             rx_receiver,
             tx_receiver,
             tx_sender,
@@ -111,7 +111,7 @@ impl DaemonIdm {
     }
 
     pub async fn launch(mut self) -> Result<DaemonIdmHandle> {
-        let glt = self.glt.clone();
+        let zlt = self.zlt.clone();
         let clients = self.clients.clone();
         let feeds = self.feeds.clone();
         let tx_sender = self.tx_sender.clone();
@@ -124,7 +124,7 @@ impl DaemonIdm {
             }
         });
         Ok(DaemonIdmHandle {
-            glt,
+            zlt,
             clients,
             feeds,
             tx_sender,

crates/daemon/src/ip/assignment.rs

@@ -36,13 +36,13 @@ impl IpAssignment {
         store: IpReservationStore,
     ) -> Result<Self> {
         let mut state = IpAssignment::fetch_current_state(&store).await?;
-        let reservation = if let Some(reservation) = store.read(host_uuid).await? {
+        let gateway_reservation = if let Some(reservation) = store.read(Uuid::nil()).await? {
             reservation
         } else {
             IpAssignment::allocate(
                 &mut state,
                 &store,
-                host_uuid,
+                Uuid::nil(),
                 ipv4_network,
                 ipv6_network,
                 None,
@@ -51,12 +51,27 @@ impl IpAssignment {
             )
             .await?
         };
+
+        if store.read(host_uuid).await?.is_none() {
+            let _ = IpAssignment::allocate(
+                &mut state,
+                &store,
+                host_uuid,
+                ipv4_network,
+                ipv6_network,
+                Some(gateway_reservation.gateway_ipv4),
+                Some(gateway_reservation.gateway_ipv6),
+                Some(gateway_reservation.gateway_mac),
+            )
+            .await?;
+        }
+
         let assignment = IpAssignment {
             ipv4_network,
             ipv6_network,
-            gateway_ipv4: reservation.ipv4,
+            gateway_ipv4: gateway_reservation.ipv4,
-            gateway_ipv6: reservation.ipv6,
+            gateway_ipv6: gateway_reservation.ipv6,
-            gateway_mac: reservation.gateway_mac,
+            gateway_mac: gateway_reservation.mac,
             store,
             state: Arc::new(RwLock::new(state)),
         };
@@ -92,13 +107,17 @@ impl IpAssignment {
             .filter(|ip| {
                 let last = ip.octets()[3];
                 // filter for IPs ending in .1 to .250 because .250+ can have special meaning
-                last > 0 && last < 250
+                (1..250).contains(&last)
             })
             .find(|ip| !state.ipv4.contains_key(ip));
 
         let found_ipv6: Option<Ipv6Addr> = ipv6_network
             .iter()
             .filter(|ip| !ip.is_loopback() && !ip.is_multicast())
+            .filter(|ip| {
+                let last = ip.octets()[15];
+                last > 0
+            })
             .find(|ip| !state.ipv6.contains_key(ip));
 
         let Some(ipv4) = found_ipv4 else {
@@ -114,7 +133,7 @@ impl IpAssignment {
         };
 
         let mut mac = MacAddr6::random();
-        mac.set_local(false);
+        mac.set_local(true);
         mac.set_multicast(false);
 
         let reservation = IpReservation {

crates/daemon/src/lib.rs

@@ -16,6 +16,7 @@ use kratart::Runtime;
 use log::{debug, info};
 use reconcile::zone::ZoneReconciler;
 use std::path::Path;
+use std::time::Duration;
 use std::{net::SocketAddr, path::PathBuf, str::FromStr, sync::Arc};
 use tokio::{
     fs,
@@ -41,13 +42,13 @@ pub mod metrics;
 pub mod oci;
 pub mod reconcile;
 pub mod zlt;
-
 pub struct Daemon {
     store: String,
     _config: Arc<DaemonConfig>,
-    glt: ZoneLookupTable,
+    zlt: ZoneLookupTable,
     devices: DaemonDeviceManager,
     zones: ZoneStore,
+    ip: IpAssignment,
     events: DaemonEventContext,
     zone_reconciler_task: JoinHandle<()>,
     zone_reconciler_notify: Sender<Uuid>,
@@ -127,7 +128,7 @@ impl Daemon {
         let ipv4_network = Ipv4Network::from_str(&config.network.ipv4.subnet)?;
         let ipv6_network = Ipv6Network::from_str(&config.network.ipv6.subnet)?;
         let ip_reservation_store = IpReservationStore::open(database)?;
-        let ip_assignment =
+        let ip =
             IpAssignment::new(host_uuid, ipv4_network, ipv6_network, ip_reservation_store).await?;
         debug!("initializing zone reconciler");
         let zone_reconciler = ZoneReconciler::new(
@@ -141,7 +142,7 @@ impl Daemon {
             kernel_path,
             initrd_path,
             addons_path,
-            ip_assignment,
+            ip.clone(),
             config.clone(),
         )?;
 
@@ -161,9 +162,10 @@ impl Daemon {
         Ok(Self {
             store,
             _config: config,
-            glt: zlt,
+            zlt,
             devices,
             zones,
+            ip,
             events,
             zone_reconciler_task,
             zone_reconciler_notify,
@@ -178,12 +180,13 @@ impl Daemon {
     pub async fn listen(&mut self, addr: ControlDialAddress) -> Result<()> {
         debug!("starting control service");
         let control_service = DaemonControlService::new(
-            self.glt.clone(),
+            self.zlt.clone(),
             self.devices.clone(),
             self.events.clone(),
             self.console.clone(),
             self.idm.clone(),
             self.zones.clone(),
+            self.ip.clone(),
             self.zone_reconciler_notify.clone(),
             self.packer.clone(),
             self.runtime.clone(),
@@ -206,6 +209,8 @@ impl Daemon {
             server = server.tls_config(tls_config)?;
         }
 
+        server = server.http2_keepalive_interval(Some(Duration::from_secs(10)));
+
         let server = server.add_service(ControlServiceServer::new(control_service));
         info!("listening on address {}", addr);
         match addr {

crates/daemon/src/reconcile/zone/create.rs

@@ -184,10 +184,19 @@ impl ZoneCreator<'_> {
             initial_resources.max_cpus = initial_resources.target_cpus;
         }
         spec.initial_resources = Some(initial_resources);
+        let kernel_options = spec.kernel_options.clone().unwrap_or_default();
         let info = self
             .runtime
             .launch(ZoneLaunchRequest {
-                format: LaunchPackedFormat::Squashfs,
+                format: match image.format {
+                    OciPackedFormat::Squashfs => LaunchPackedFormat::Squashfs,
+                    OciPackedFormat::Erofs => LaunchPackedFormat::Erofs,
+                    _ => {
+                        return Err(anyhow!(
+                            "oci image is in an invalid format, which isn't compatible with launch"
+                        ));
+                    }
+                },
                 uuid: Some(uuid),
                 name: if spec.name.is_empty() {
                     None
@@ -208,7 +217,8 @@ impl ZoneCreator<'_> {
                     .map(|x| (x.key.clone(), x.value.clone()))
                     .collect::<HashMap<_, _>>(),
                 run: empty_vec_optional(task.command.clone()),
-                debug: false,
+                kernel_verbose: kernel_options.verbose,
+                kernel_cmdline_append: kernel_options.cmdline_append,
                 addons_image: Some(self.addons_path.to_path_buf()),
                 network: ZoneLaunchNetwork {
                     ipv4: reservation.ipv4.to_string(),

crates/daemon/src/reconcile/zone/mod.rs

@@ -27,7 +27,7 @@ use tokio::{
     select,
     sync::{
         mpsc::{channel, Receiver, Sender},
-        Mutex, RwLock,
+        RwLock,
     },
     task::JoinHandle,
     time::sleep,
@@ -45,16 +45,9 @@ enum ZoneReconcilerResult {
 }
 
 struct ZoneReconcilerEntry {
-    task: JoinHandle<()>,
     sender: Sender<()>,
 }
 
-impl Drop for ZoneReconcilerEntry {
-    fn drop(&mut self) {
-        self.task.abort();
-    }
-}
-
 #[derive(Clone)]
 pub struct ZoneReconciler {
     devices: DaemonDeviceManager,
@@ -66,7 +59,7 @@ pub struct ZoneReconciler {
     kernel_path: PathBuf,
     initrd_path: PathBuf,
     addons_path: PathBuf,
-    tasks: Arc<Mutex<HashMap<Uuid, ZoneReconcilerEntry>>>,
+    tasks: Arc<RwLock<HashMap<Uuid, ZoneReconcilerEntry>>>,
     zone_reconciler_notify: Sender<Uuid>,
     zone_reconcile_lock: Arc<RwLock<()>>,
     ip_assignment: IpAssignment,
@@ -99,7 +92,7 @@ impl ZoneReconciler {
             kernel_path,
             initrd_path,
             addons_path: modules_path,
-            tasks: Arc::new(Mutex::new(HashMap::new())),
+            tasks: Arc::new(RwLock::new(HashMap::new())),
             zone_reconciler_notify,
             zone_reconcile_lock: Arc::new(RwLock::with_max_readers((), PARALLEL_LIMIT)),
             ip_assignment,
@@ -125,7 +118,7 @@ impl ZoneReconciler {
                                 error!("failed to start zone reconciler task {}: {}", uuid, error);
                             }
 
-                            let map = self.tasks.lock().await;
+                            let map = self.tasks.read().await;
                             if let Some(entry) = map.get(&uuid) {
                                 if let Err(error) = entry.sender.send(()).await {
                                     error!("failed to notify zone reconciler task {}: {}", uuid, error);
@@ -271,7 +264,7 @@ impl ZoneReconciler {
 
             if destroyed {
                 self.zones.remove(uuid).await?;
-                let mut map = self.tasks.lock().await;
+                let mut map = self.tasks.write().await;
                 map.remove(&uuid);
             } else {
                 self.zones.update(uuid, zone.clone()).await?;
@@ -337,7 +330,7 @@ impl ZoneReconciler {
     }
 
     async fn launch_task_if_needed(&self, uuid: Uuid) -> Result<()> {
-        let mut map = self.tasks.lock().await;
+        let mut map = self.tasks.write().await;
         match map.entry(uuid) {
             Entry::Occupied(_) => {}
             Entry::Vacant(entry) => {
@@ -350,7 +343,7 @@ impl ZoneReconciler {
     async fn launch_task(&self, uuid: Uuid) -> Result<ZoneReconcilerEntry> {
         let this = self.clone();
         let (sender, mut receiver) = channel(10);
-        let task = tokio::task::spawn(async move {
+        tokio::task::spawn(async move {
             'notify_loop: loop {
                 if receiver.recv().await.is_none() {
                     break 'notify_loop;
@@ -372,7 +365,7 @@ impl ZoneReconciler {
                 }
             }
         });
-        Ok(ZoneReconcilerEntry { task, sender })
+        Ok(ZoneReconcilerEntry { sender })
     }
 }
 
@@ -380,9 +373,9 @@ pub fn ip_reservation_to_network_status(ip: &IpReservation) -> ZoneNetworkStatus
     ZoneNetworkStatus {
         zone_ipv4: format!("{}/{}", ip.ipv4, ip.ipv4_prefix),
         zone_ipv6: format!("{}/{}", ip.ipv6, ip.ipv6_prefix),
-        zone_mac: ip.mac.to_string().replace('-', ":"),
+        zone_mac: ip.mac.to_string().to_lowercase().replace('-', ":"),
         gateway_ipv4: format!("{}/{}", ip.gateway_ipv4, ip.ipv4_prefix),
         gateway_ipv6: format!("{}/{}", ip.gateway_ipv6, ip.ipv6_prefix),
-        gateway_mac: ip.gateway_mac.to_string().replace('-', ":"),
+        gateway_mac: ip.gateway_mac.to_string().to_lowercase().replace('-', ":"),
     }
 }

crates/krata/proto/krata/v1/common.proto

@@ -25,6 +25,7 @@ message ZoneSpec {
     ZoneTaskSpec task = 6;
     repeated ZoneSpecAnnotation annotations = 7;
     repeated ZoneSpecDevice devices = 8;
+    ZoneKernelOptionsSpec kernel_options = 9;
 }
 
 message ZoneResourceSpec {
@@ -40,6 +41,11 @@ message ZoneImageSpec {
     }
 }
 
+message ZoneKernelOptionsSpec {
+    bool verbose = 1;
+    string cmdline_append = 2;
+}
+
 enum OciImageFormat {
     OCI_IMAGE_FORMAT_UNKNOWN = 0;
     OCI_IMAGE_FORMAT_SQUASHFS = 1;

crates/krata/proto/krata/v1/control.proto

@@ -10,7 +10,7 @@ import "krata/idm/transport.proto";
 import "krata/v1/common.proto";
 
 service ControlService {
-    rpc HostStatus(HostStatusRequest) returns (HostStatusReply);
+    rpc GetHostStatus(GetHostStatusRequest) returns (GetHostStatusReply);
     rpc SnoopIdm(SnoopIdmRequest) returns (stream SnoopIdmReply);
     rpc GetHostCpuTopology(GetHostCpuTopologyRequest) returns (GetHostCpuTopologyReply);
     rpc SetHostPowerManagementPolicy(SetHostPowerManagementPolicyRequest) returns (SetHostPowerManagementPolicyReply);
@@ -35,14 +35,19 @@ service ControlService {
     rpc ReadZoneMetrics(ReadZoneMetricsRequest) returns (ReadZoneMetricsReply);
 
     rpc WatchEvents(WatchEventsRequest) returns (stream WatchEventsReply);
+
+    rpc ReadHypervisorConsole(ReadHypervisorConsoleRequest) returns (ReadHypervisorConsoleReply);
 }
 
-message HostStatusRequest {}
+message GetHostStatusRequest {}
 
-message HostStatusReply {
+message GetHostStatusReply {
     string host_uuid = 1;
     uint32 host_domid = 2;
     string krata_version = 3;
+    string host_ipv4 = 4;
+    string host_ipv6 = 5;
+    string host_mac = 6;
 }
 
 message CreateZoneRequest {
@@ -99,6 +104,7 @@ message ExecInsideZoneReply {
 message ZoneConsoleRequest {
     string zone_id = 1;
     bytes data = 2;
+    bool replay_history = 3;
 }
 
 message ZoneConsoleReply {
@@ -252,3 +258,9 @@ message UpdateZoneResourcesRequest {
 }
 
 message UpdateZoneResourcesReply {}
+
+message ReadHypervisorConsoleRequest {}
+
+message ReadHypervisorConsoleReply {
+    string data = 1;
+}

crates/krata/src/idm/client.rs

@@ -495,6 +495,7 @@ impl<R: IdmRequest, E: IdmSerializable> IdmClient<R, E> {
                             IdmTransportPacketForm::StreamRequestClosed => {
                                 let mut update_streams = request_update_streams.lock().await;
                                 update_streams.remove(&packet.id);
+                                println!("stream request closed: {}", packet.id);
                             }
 
                             IdmTransportPacketForm::StreamResponseUpdate => {

crates/network/Cargo.toml

@@ -16,7 +16,7 @@ clap = { workspace = true }
 env_logger = { workspace = true }
 etherparse = { workspace = true }
 futures = { workspace = true }
-krata = { path = "../krata", version = "^0.0.17" }
+krata = { path = "../krata", version = "^0.0.19" }
 krata-advmac = { workspace = true }
 libc = { workspace = true }
 log = { workspace = true }

crates/network/src/backend.rs

@@ -127,7 +127,8 @@ impl NetworkBackend {
         let (tx_sender, tx_receiver) = channel::<BytesMut>(TX_CHANNEL_BUFFER_LEN);
         let mut udev = ChannelDevice::new(mtu, Medium::Ethernet, tx_sender.clone());
         let mac = self.metadata.gateway.mac;
-        let nat = Nat::new(mtu, proxy, mac, addresses.clone(), tx_sender.clone())?;
+        let local_cidrs = addresses.clone();
+        let nat = Nat::new(mtu, proxy, mac, local_cidrs, tx_sender.clone())?;
         let hardware_addr = HardwareAddress::Ethernet(mac);
         let config = Config::new(hardware_addr);
         let mut iface = Interface::new(config, &mut udev, Instant::now());

crates/network/src/hbridge.rs

@@ -1,21 +1,15 @@
-use std::{
+use std::{io::ErrorKind, net::IpAddr};
-    io::ErrorKind,
-    net::{IpAddr, Ipv4Addr},
-};
 
-use advmac::MacAddr6;
 use anyhow::{anyhow, Result};
 use bytes::BytesMut;
 use futures::TryStreamExt;
 use log::error;
-use smoltcp::wire::EthernetAddress;
+use smoltcp::wire::{EthernetAddress, Ipv4Cidr, Ipv6Cidr};
 use tokio::{select, task::JoinHandle};
 use tokio_tun::Tun;
 
 use crate::vbridge::{BridgeJoinHandle, VirtualBridge};
 
-const HOST_IPV4_ADDR: Ipv4Addr = Ipv4Addr::new(10, 75, 0, 1);
-
 #[derive(Debug)]
 enum HostBridgeProcessSelect {
     Send(Option<BytesMut>),
@@ -27,7 +21,14 @@ pub struct HostBridge {
 }
 
 impl HostBridge {
-    pub async fn new(mtu: usize, interface: String, bridge: &VirtualBridge) -> Result<HostBridge> {
+    pub async fn new(
+        mtu: usize,
+        interface: String,
+        bridge: &VirtualBridge,
+        ipv4: Ipv4Cidr,
+        ipv6: Ipv6Cidr,
+        mac: EthernetAddress,
+    ) -> Result<HostBridge> {
         let tun = Tun::builder()
             .name(&interface)
             .tap(true)
@@ -38,10 +39,6 @@ impl HostBridge {
         let (connection, handle, _) = rtnetlink::new_connection()?;
         tokio::spawn(connection);
 
-        let mut mac = MacAddr6::random();
-        mac.set_local(true);
-        mac.set_multicast(false);
-
         let mut links = handle.link().get().match_name(interface.clone()).execute();
         let link = links.try_next().await?;
         if link.is_none() {
@@ -54,25 +51,32 @@ impl HostBridge {
 
         handle
             .address()
-            .add(link.header.index, IpAddr::V4(HOST_IPV4_ADDR), 16)
+            .add(
+                link.header.index,
+                IpAddr::V4(ipv4.address().into()),
+                ipv4.prefix_len(),
+            )
             .execute()
             .await?;
 
         handle
             .address()
-            .add(link.header.index, IpAddr::V6(mac.to_link_local_ipv6()), 10)
+            .add(
+                link.header.index,
+                IpAddr::V6(ipv6.address().into()),
+                ipv6.prefix_len(),
+            )
             .execute()
             .await?;
 
         handle
             .link()
             .set(link.header.index)
-            .address(mac.to_array().to_vec())
+            .address(mac.0.to_vec())
             .up()
             .execute()
             .await?;
 
-        let mac = EthernetAddress(mac.to_array());
         let bridge_handle = bridge.join(mac).await?;
 
         let task = tokio::task::spawn(async move {

crates/network/src/lib.rs

@@ -1,17 +1,21 @@
-use std::{collections::HashMap, time::Duration};
+use std::{collections::HashMap, str::FromStr, time::Duration};
 
-use anyhow::Result;
+use anyhow::{anyhow, Result};
 use autonet::{AutoNetworkChangeset, AutoNetworkWatcher, NetworkMetadata};
 use futures::{future::join_all, TryFutureExt};
 use hbridge::HostBridge;
 use krata::{
     client::ControlClientProvider,
     dial::ControlDialAddress,
-    v1::{common::Zone, control::control_service_client::ControlServiceClient},
+    v1::{
+        common::Zone,
+        control::{control_service_client::ControlServiceClient, GetHostStatusRequest},
+    },
 };
 use log::warn;
+use smoltcp::wire::{EthernetAddress, Ipv4Cidr, Ipv6Cidr};
 use tokio::{task::JoinHandle, time::sleep};
-use tonic::transport::Channel;
+use tonic::{transport::Channel, Request};
 use uuid::Uuid;
 use vbridge::VirtualBridge;
 
@@ -41,10 +45,27 @@ pub struct NetworkService {
 
 impl NetworkService {
     pub async fn new(control_address: ControlDialAddress) -> Result<NetworkService> {
-        let control = ControlClientProvider::dial(control_address).await?;
+        let mut control = ControlClientProvider::dial(control_address).await?;
+        let host_status = control
+            .get_host_status(Request::new(GetHostStatusRequest {}))
+            .await?
+            .into_inner();
+        let host_ipv4 = Ipv4Cidr::from_str(&host_status.host_ipv4)
+            .map_err(|_| anyhow!("failed to parse host ipv4 cidr"))?;
+        let host_ipv6 = Ipv6Cidr::from_str(&host_status.host_ipv6)
+            .map_err(|_| anyhow!("failed to parse host ipv6 cidr"))?;
+        let host_mac = EthernetAddress::from_str(&host_status.host_mac)
+            .map_err(|_| anyhow!("failed to parse host mac address"))?;
         let bridge = VirtualBridge::new()?;
-        let hbridge =
+        let hbridge = HostBridge::new(
-            HostBridge::new(HOST_BRIDGE_MTU + EXTRA_MTU, "krata0".to_string(), &bridge).await?;
+            HOST_BRIDGE_MTU + EXTRA_MTU,
+            "krata0".to_string(),
+            &bridge,
+            host_ipv4,
+            host_ipv6,
+            host_mac,
+        )
+        .await?;
         Ok(NetworkService {
             control,
             zones: HashMap::new(),

crates/runtime/Cargo.toml

@@ -12,20 +12,20 @@ resolver = "2"
 anyhow = { workspace = true }
 backhand = { workspace = true }
 ipnetwork = { workspace = true }
-krata = { path = "../krata", version = "^0.0.17" }
+krata = { path = "../krata", version = "^0.0.19" }
 krata-advmac = { workspace = true }
-krata-oci = { path = "../oci", version = "^0.0.17" }
+krata-oci = { path = "../oci", version = "^0.0.19" }
 log = { workspace = true }
 serde_json = { workspace = true }
 tokio = { workspace = true }
 uuid = { workspace = true }
-krata-loopdev = { path = "../loopdev", version = "^0.0.17" }
+krata-loopdev = { path = "../loopdev", version = "^0.0.19" }
-krata-xencall = { path = "../xen/xencall", version = "^0.0.17" }
+krata-xencall = { path = "../xen/xencall", version = "^0.0.19" }
-krata-xenclient = { path = "../xen/xenclient", version = "^0.0.17" }
+krata-xenclient = { path = "../xen/xenclient", version = "^0.0.19" }
-krata-xenevtchn = { path = "../xen/xenevtchn", version = "^0.0.17" }
+krata-xenevtchn = { path = "../xen/xenevtchn", version = "^0.0.19" }
-krata-xengnt = { path = "../xen/xengnt", version = "^0.0.17" }
+krata-xengnt = { path = "../xen/xengnt", version = "^0.0.19" }
-krata-xenplatform = { path = "../xen/xenplatform", version = "^0.0.17" }
+krata-xenplatform = { path = "../xen/xenplatform", version = "^0.0.19" }
-krata-xenstore = { path = "../xen/xenstore", version = "^0.0.17" }
+krata-xenstore = { path = "../xen/xenstore", version = "^0.0.19" }
 walkdir = { workspace = true }
 indexmap = { workspace = true }
 

crates/runtime/src/launch.rs

@@ -37,7 +37,8 @@ pub struct ZoneLaunchRequest {
     pub env: HashMap<String, String>,
     pub run: Option<Vec<String>>,
     pub pcis: Vec<PciDevice>,
-    pub debug: bool,
+    pub kernel_verbose: bool,
+    pub kernel_cmdline_append: String,
     pub image: OciPackedImage,
     pub addons_image: Option<PathBuf>,
     pub network: ZoneLaunchNetwork,
@@ -139,9 +140,14 @@ impl ZoneLauncher {
             None
         };
         let mut cmdline_options = ["console=hvc0"].to_vec();
-        if !request.debug {
+        if !request.kernel_verbose {
             cmdline_options.push("quiet");
         }
+
+        if !request.kernel_cmdline_append.is_empty() {
+            cmdline_options.push(&request.kernel_cmdline_append);
+        }
+
         let cmdline = cmdline_options.join(" ");
 
         let zone_mac_string = request.network.zone_mac.to_string().replace('-', ":");

crates/runtime/src/lib.rs

@@ -1,5 +1,6 @@
 use anyhow::{anyhow, Result};
 use krataloopdev::LoopControl;
+use log::debug;
 use std::{fs, path::PathBuf, str::FromStr, sync::Arc};
 use tokio::sync::Semaphore;
 use uuid::Uuid;
@@ -168,6 +169,13 @@ pub struct Runtime {
 impl Runtime {
     pub async fn new() -> Result<Self> {
         let context = RuntimeContext::new().await?;
+        debug!("testing for hypervisor presence");
+        context
+            .xen
+            .call
+            .get_version_capabilities()
+            .await
+            .map_err(|_| anyhow!("hypervisor is not present"))?;
         Ok(Self {
             context,
             launch_semaphore: Arc::new(Semaphore::new(10)),
@@ -297,4 +305,16 @@ impl Runtime {
         let context = RuntimeContext::new().await?;
         Ok(PowerManagementContext { context })
     }
+
+    pub async fn read_hypervisor_console(&self, clear: bool) -> Result<Arc<str>> {
+        let index = 0_u32;
+        let (rawbuf, newindex) = self
+            .context
+            .xen
+            .call
+            .read_console_ring_raw(clear, index)
+            .await?;
+        let buf = std::str::from_utf8(&rawbuf[..newindex as usize])?;
+        Ok(Arc::from(buf))
+    }
 }

crates/xen/xencall/examples/console_read.rs

@@ -0,0 +1,18 @@
+use xencall::error::Result;
+use xencall::XenCall;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    env_logger::init();
+
+    let call = XenCall::open(0)?;
+    let index = 0_u32;
+    let (buf, newindex) = call.read_console_ring_raw(false, index).await?;
+
+    match std::str::from_utf8(&buf[..newindex as usize]) {
+        Ok(v) => print!("{}", v),
+        _ => panic!("unable to decode Xen console messages"),
+    };
+
+    Ok(())
+}

crates/xen/xencall/src/lib.rs

@@ -26,12 +26,12 @@ use std::sync::Arc;
 use std::time::Duration;
 use sys::{
     CpuId, E820Entry, ForeignMemoryMap, PhysdevMapPirq, Sysctl, SysctlCputopo, SysctlCputopoinfo,
-    SysctlPhysinfo, SysctlPmOp, SysctlPmOpValue, SysctlSetCpuFreqGov, SysctlValue,
+    SysctlPhysinfo, SysctlPmOp, SysctlPmOpValue, SysctlReadconsole, SysctlSetCpuFreqGov,
-    VcpuGuestContextAny, HYPERVISOR_PHYSDEV_OP, HYPERVISOR_SYSCTL, PHYSDEVOP_MAP_PIRQ,
+    SysctlValue, VcpuGuestContextAny, HYPERVISOR_PHYSDEV_OP, HYPERVISOR_SYSCTL, PHYSDEVOP_MAP_PIRQ,
     XEN_DOMCTL_MAX_INTERFACE_VERSION, XEN_DOMCTL_MIN_INTERFACE_VERSION, XEN_MEM_SET_MEMORY_MAP,
     XEN_SYSCTL_CPUTOPOINFO, XEN_SYSCTL_MAX_INTERFACE_VERSION, XEN_SYSCTL_MIN_INTERFACE_VERSION,
     XEN_SYSCTL_PHYSINFO, XEN_SYSCTL_PM_OP, XEN_SYSCTL_PM_OP_DISABLE_TURBO,
-    XEN_SYSCTL_PM_OP_ENABLE_TURBO, XEN_SYSCTL_PM_OP_SET_CPUFREQ_GOV,
+    XEN_SYSCTL_PM_OP_ENABLE_TURBO, XEN_SYSCTL_PM_OP_SET_CPUFREQ_GOV, XEN_SYSCTL_READCONSOLE,
 };
 use tokio::sync::Semaphore;
 use tokio::time::sleep;
@@ -1087,4 +1087,33 @@ impl XenCall {
             .await?;
         Ok(())
     }
+
+    pub async fn read_console_ring_raw(
+        &self,
+        clear: bool,
+        index: u32,
+    ) -> Result<([u8; 16384], u32)> {
+        let mut u8buf = [0u8; 16384];
+        let mut sysctl = Sysctl {
+            cmd: XEN_SYSCTL_READCONSOLE,
+            interface_version: self.sysctl_interface_version,
+            value: SysctlValue {
+                console: SysctlReadconsole {
+                    clear: clear as u8,
+                    incremental: 1,
+                    pad: 0,
+                    index,
+                    buffer: addr_of_mut!(u8buf) as u64,
+                    count: 16384,
+                },
+            },
+        };
+        self.hypercall1(HYPERVISOR_SYSCTL, addr_of_mut!(sysctl) as c_ulong)
+            .await?;
+        // Safety: We are passing a SysctlReadconsole struct as part of the hypercall, and
+        // calling the hypercall is known to not change the underlying value outside changing
+        // the values on some SysctlReadconsole fields.
+        let newindex = unsafe { sysctl.value.console.index };
+        Ok((u8buf, newindex))
+    }
 }

crates/xen/xencall/src/sys.rs

@@ -752,6 +752,7 @@ pub struct SysctlCputopoinfo {
 
 #[repr(C)]
 pub union SysctlValue {
+    pub console: SysctlReadconsole,
     pub cputopoinfo: SysctlCputopoinfo,
     pub pm_op: SysctlPmOp,
     pub phys_info: SysctlPhysinfo,
@@ -765,6 +766,7 @@ pub struct Sysctl {
     pub value: SysctlValue,
 }
 
+pub const XEN_SYSCTL_READCONSOLE: u32 = 1;
 pub const XEN_SYSCTL_PHYSINFO: u32 = 3;
 pub const XEN_SYSCTL_PM_OP: u32 = 12;
 pub const XEN_SYSCTL_CPUTOPOINFO: u32 = 16;
@@ -802,3 +804,14 @@ pub struct SysctlPhysinfo {
     pub max_mfn: u64,
     pub hw_cap: [u32; 8],
 }
+
+#[repr(C)]
+#[derive(Clone, Copy, Debug, Default)]
+pub struct SysctlReadconsole {
+    pub clear: u8,
+    pub incremental: u8,
+    pub pad: u16,
+    pub index: u32,
+    pub buffer: u64,
+    pub count: u32,
+}

crates/xen/xenclient/Cargo.toml

@@ -13,9 +13,9 @@ async-trait = { workspace = true }
 indexmap = { workspace = true }
 libc = { workspace = true }
 log = { workspace = true }
-krata-xencall = { path = "../xencall", version = "^0.0.17" }
+krata-xencall = { path = "../xencall", version = "^0.0.19" }
-krata-xenplatform = { path = "../xenplatform", version = "^0.0.17" }
+krata-xenplatform = { path = "../xenplatform", version = "^0.0.19" }
-krata-xenstore = { path = "../xenstore", version = "^0.0.17" }
+krata-xenstore = { path = "../xenstore", version = "^0.0.19" }
 regex = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }

crates/xen/xenplatform/Cargo.toml

@@ -16,7 +16,7 @@ flate2 = { workspace = true }
 indexmap = { workspace = true }
 libc = { workspace = true }
 log = { workspace = true }
-krata-xencall = { path = "../xencall", version = "^0.0.17" }
+krata-xencall = { path = "../xencall", version = "^0.0.19" }
 memchr = { workspace = true }
 nix = { workspace = true }
 regex = { workspace = true }

crates/zone/Cargo.toml

@@ -14,8 +14,8 @@ cgroups-rs = { workspace = true }
 env_logger = { workspace = true }
 futures = { workspace = true }
 ipnetwork = { workspace = true }
-krata = { path = "../krata", version = "^0.0.17" }
+krata = { path = "../krata", version = "^0.0.19" }
-krata-xenstore = { path = "../xen/xenstore", version = "^0.0.17" }
+krata-xenstore = { path = "../xen/xenstore", version = "^0.0.19" }
 libc = { workspace = true }
 log = { workspace = true }
 nix = { workspace = true, features = ["ioctl", "process", "fs"] }
@@ -29,6 +29,7 @@ serde_json = { workspace = true }
 sys-mount = { workspace = true }
 sysinfo = { workspace = true }
 tokio = { workspace = true }
+tokio-util = { workspace = true }
 
 [lib]
 name = "kratazone"

crates/zone/src/exec.rs

@@ -1,5 +1,6 @@
 use std::{collections::HashMap, process::Stdio};
 
+use crate::childwait::ChildWait;
 use anyhow::{anyhow, Result};
 use krata::idm::{
     client::IdmClientStreamResponseHandle,
@@ -16,9 +17,9 @@ use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
     join,
     process::Command,
+    select,
 };
-
+use tokio_util::sync::CancellationToken;
-use crate::childwait::ChildWait;
 
 pub struct ZoneExecTask {
     pub wait: ChildWait,
@@ -70,16 +71,21 @@ impl ZoneExecTask {
         if start.tty {
             let pty = Pty::new().map_err(|error| anyhow!("unable to allocate pty: {}", error))?;
             pty.resize(Size::new(24, 80))?;
-            let mut child = ChildDropGuard {
+            let pts = pty
-                inner: pty_process::Command::new(exe)
+                .pts()
+                .map_err(|error| anyhow!("unable to allocate pts: {}", error))?;
+            let child = std::panic::catch_unwind(move || {
+                let pts = pts;
+                pty_process::Command::new(exe)
                     .args(cmd)
                     .envs(env)
                     .current_dir(dir)
-                    .spawn(
+                    .spawn(&pts)
-                        &pty.pts()
+            })
-                            .map_err(|error| anyhow!("unable to allocate pts: {}", error))?,
+            .map_err(|_| anyhow!("internal error"))
-                    )
+            .map_err(|error| anyhow!("failed to spawn: {}", error))??;
-                    .map_err(|error| anyhow!("failed to spawn: {}", error))?,
+            let mut child = ChildDropGuard {
+                inner: child,
                 kill: true,
             };
             let pid = child
@@ -111,9 +117,12 @@ impl ZoneExecTask {
                 }
             });
 
+            let cancel = CancellationToken::new();
+            let stdin_cancel = cancel.clone();
             let stdin_task = tokio::task::spawn(async move {
                 loop {
                     let Some(request) = receiver.recv().await else {
+                        stdin_cancel.cancel();
                         break;
                     };
 
@@ -136,28 +145,44 @@ impl ZoneExecTask {
             });
 
             code = loop {
-                if let Ok(event) = wait_subscription.recv().await {
+                select! {
-                    if event.pid.as_raw() as u32 == pid {
+                    result = wait_subscription.recv() => match result {
-                        break event.status;
+                        Ok(event) => {
+                            if event.pid.as_raw() as u32 == pid {
+                                child.kill = false;
+                                break event.status;
+                            }
+                        }
+                        _ => {
+                            child.inner.start_kill()?;
+                            child.kill = false;
+                            break -1;
+                        }
+                    },
+                    _ = cancel.cancelled() => {
+                        child.inner.start_kill()?;
+                        child.kill = false;
+                        break -1;
                     }
                 }
             };
 
-            child.kill = false;
-
             let _ = join!(pty_read_task);
             stdin_task.abort();
         } else {
-            let mut child = Command::new(exe)
+            let mut child = std::panic::catch_unwind(|| {
-                .args(cmd)
+                Command::new(exe)
-                .envs(env)
+                    .args(cmd)
-                .current_dir(dir)
+                    .envs(env)
-                .stdin(Stdio::piped())
+                    .current_dir(dir)
-                .stdout(Stdio::piped())
+                    .stdin(Stdio::piped())
-                .stderr(Stdio::piped())
+                    .stdout(Stdio::piped())
-                .kill_on_drop(true)
+                    .stderr(Stdio::piped())
-                .spawn()
+                    .kill_on_drop(true)
-                .map_err(|error| anyhow!("failed to spawn: {}", error))?;
+                    .spawn()
+            })
+            .map_err(|_| anyhow!("internal error"))
+            .map_err(|error| anyhow!("failed to spawn: {}", error))??;
 
             let pid = child.id().ok_or_else(|| anyhow!("pid is not provided"))?;
             let mut stdin = child
@@ -221,9 +246,12 @@ impl ZoneExecTask {
                 }
             });
 
+            let cancel = CancellationToken::new();
+            let stdin_cancel = cancel.clone();
             let stdin_task = tokio::task::spawn(async move {
                 loop {
                     let Some(request) = receiver.recv().await else {
+                        stdin_cancel.cancel();
                         break;
                     };
 
@@ -247,9 +275,21 @@ impl ZoneExecTask {
             });
 
             code = loop {
-                if let Ok(event) = wait_subscription.recv().await {
+                select! {
-                    if event.pid.as_raw() as u32 == pid {
+                    result = wait_subscription.recv() => match result {
-                        break event.status;
+                        Ok(event) => {
+                            if event.pid.as_raw() as u32 == pid {
+                                break event.status;
+                            }
+                        }
+                        _ => {
+                            child.start_kill()?;
+                            break -1;
+                        }
+                    },
+                    _ = cancel.cancelled() => {
+                        child.start_kill()?;
+                        break -1;
                     }
                 }
             };