Improve intproxy deps (metalbear-co#2057)

* IntProxy protocol extracted to a separate crate

* mirrord-intproxy-protocol deps updated

* async support feature-gated

* mirrord-intproxy deps updated

* mirrord-console deps updated

* layer deps updated

* Changelog entry

.vscode/settings.json

@@ -1,5 +1,5 @@
 {
     "rust-analyzer.cargo.features": [
-        "binary"
+        "binary",
     ]
 }

changelog.d/2039.internal.md

@@ -0,0 +1 @@
+Improved crates structure around internal proxy and mirrord console.

mirrord/cli/Cargo.toml

@@ -22,7 +22,7 @@ mirrord-progress = { path = "../progress" }
 mirrord-kube = { path = "../kube" }
 mirrord-config = { path = "../config" }
 mirrord-protocol = { path = "../protocol" }
-mirrord-console = { path = "../console" }
+mirrord-console = { path = "../console", features = ["async-logger"] }
 mirrord-analytics = { path = "../analytics" }
 mirrord-intproxy = { path = "../intproxy" }
 

mirrord/console/Cargo.toml

@@ -19,16 +19,19 @@ required-features = ["binary"]
 
 [features]
 default = []
-binary = ["dep:tracing", "dep:tracing-subscriber"]
+binary = ["dep:tracing", "dep:tracing-subscriber", "dep:tokio", "mirrord-intproxy-protocol/codec-async"]
+async-logger = ["mirrord-intproxy-protocol/codec-async", "dep:tokio", "dep:drain", "dep:tokio-util"]
 
 [dependencies]
-mirrord-intproxy = { path = "../intproxy" }
-tokio.workspace = true
-tracing = { workspace = true, optional = true }
-tracing-subscriber = { workspace = true, optional = true }
-log = { version = "0.4", features = ["std"] }
+mirrord-intproxy-protocol = { path = "../intproxy/protocol", features = ["codec"] }
+
 bincode.workspace = true
+log = { version = "0.4", features = ["std"] }
 miette = "5"
 thiserror.workspace = true
-drain.workspace = true
-tokio-util.workspace = true
+
+tokio = { workspace = true, optional = true }
+tracing = { workspace = true, optional = true }
+tracing-subscriber = { workspace = true, optional = true }
+drain = { workspace = true, optional = true }
+tokio-util = { workspace = true, optional = true }

mirrord/console/src/async_logger.rs

@@ -1,6 +1,6 @@
 use drain::Watch;
 use log::LevelFilter;
-use mirrord_intproxy::codec::AsyncEncoder;
+use mirrord_intproxy_protocol::codec::AsyncEncoder;
 use tokio::{
     io::BufWriter,
     net::TcpStream,

mirrord/console/src/error.rs

@@ -1,7 +1,7 @@
 use std::io;
 
 use miette::Diagnostic;
-use mirrord_intproxy::codec::CodecError;
+use mirrord_intproxy_protocol::codec::CodecError;
 use thiserror::Error;
 
 pub type Result<T, E = ConsoleError> = std::result::Result<T, E>;

mirrord/console/src/lib.rs

@@ -1,9 +1,11 @@
 #![warn(clippy::indexing_slicing)]
 
+#[cfg(feature = "async-logger")]
 pub mod async_logger;
 pub mod error;
 pub mod logger;
 pub mod protocol;
 
+#[cfg(feature = "async-logger")]
 pub use async_logger::init_async_logger;
 pub use logger::init_logger;

mirrord/console/src/logger.rs

@@ -1,7 +1,7 @@
 use std::{io::BufWriter, net::TcpStream, sync::Mutex};
 
 use log::LevelFilter;
-use mirrord_intproxy::codec::SyncEncoder;
+use mirrord_intproxy_protocol::codec::SyncEncoder;
 
 use crate::{
     error::Result,

mirrord/console/src/main.rs

@@ -1,6 +1,6 @@
 use bincode::Decode;
 use mirrord_console::protocol::{Hello, Record};
-use mirrord_intproxy::codec::AsyncDecoder;
+use mirrord_intproxy_protocol::codec::AsyncDecoder;
 use tokio::{
     io::BufReader,
     net::{TcpListener, TcpStream},

mirrord/intproxy/Cargo.toml

@@ -20,20 +20,16 @@ mirrord-config = { path = "../config" }
 mirrord-kube = { path = "../kube" }
 mirrord-operator = { path = "../operator", features = ["client"] }
 mirrord-protocol = { path = "../protocol" }
+mirrord-intproxy-protocol = { path = "./protocol", features = ["codec-async"] }
 
-bincode.workspace = true
-futures.workspace = true
 serde.workspace = true
-serde_json.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
 tracing.workspace = true
-tokio-util.workspace = true
 tokio-stream.workspace = true
 hyper = { workspace = true, features = ["client", "http1", "http2"] }
 hyper-util.workspace = true
 http-body-util.workspace = true
-fancy-regex.workspace = true
 bytes.workspace = true
 
 rand = "0.8"

mirrord/intproxy/protocol/Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "mirrord-intproxy-protocol"
+version.workspace = true
+authors.workspace = true
+description.workspace = true
+documentation.workspace = true
+readme.workspace = true
+homepage.workspace = true
+repository.workspace = true
+license.workspace = true
+keywords.workspace = true
+categories.workspace = true
+publish.workspace = true
+edition.workspace = true
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+mirrord-protocol = { path = "../../protocol" }
+
+bincode.workspace = true
+thiserror = { workspace = true, optional = true }
+tokio = { workspace = true, optional = true }
+
+[features]
+codec = ["dep:thiserror"]
+codec-async = ["codec", "dep:tokio"]

mirrord/intproxy/src/codec.rs → mirrord/intproxy/protocol/src/codec.rs

@@ -1,6 +1,6 @@
 //! Custom codec used in `layer <-> proxy` communication.
 //! Supports both synchronous (required by the layer) and asynchronous (convenient for the proxy)
-//! IO.
+//! IO. Asynchronous IO is feature-gated with the `codec-async` feature.
 //!
 //! An encoded message consists of two parts:
 //! * prefix: 4 bytes containing payload length in bytes (big-endian [`u32`])
@@ -16,10 +16,11 @@ use bincode::{
     Decode, Encode,
 };
 use thiserror::Error;
-use tokio::{
-    io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt},
-    net::tcp::{OwnedReadHalf, OwnedWriteHalf},
-};
+
+#[cfg(feature = "codec-async")]
+mod codec_async;
+#[cfg(feature = "codec-async")]
+pub use codec_async::*;
 
 /// Errors that can occur when using this codec.
 #[derive(Error, Debug)]
@@ -164,119 +165,3 @@ pub fn make_sync_framed<T1: Encode, T2: Decode>(
 
     Ok((sender, receiver))
 }
-
-/// Handles sending messages of type `T` through the underlying [AsyncWrite] of type `W`.
-#[derive(Debug)]
-pub struct AsyncEncoder<T, W> {
-    buffer: Vec<u8>,
-    writer: W,
-    _phantom: std::marker::PhantomData<fn() -> T>,
-}
-
-impl<T, W> AsyncEncoder<T, W> {
-    /// Wraps the underlying IO handler.
-    pub fn new(writer: W) -> Self {
-        Self {
-            buffer: Vec::with_capacity(BUFFER_SIZE),
-            writer,
-            _phantom: Default::default(),
-        }
-    }
-
-    /// Unwraps the underlying IO handler.
-    pub fn into_inner(self) -> W {
-        self.writer
-    }
-}
-
-impl<T, W> AsyncEncoder<T, W>
-where
-    T: Encode,
-    W: AsyncWrite + Unpin,
-{
-    /// Encodes the given value into the inner IO handler.
-    pub async fn send(&mut self, value: &T) -> Result<()> {
-        self.buffer.resize(PREFIX_BYTES, 0);
-        let bytes: u32 =
-            bincode::encode_into_std_write(value, &mut self.buffer, bincode::config::standard())?
-                .try_into()?;
-        self.buffer
-            .get_mut(..PREFIX_BYTES)
-            .expect("buffer to short")
-            .copy_from_slice(&bytes.to_be_bytes());
-
-        self.writer.write_all(&self.buffer).await?;
-
-        Ok(())
-    }
-
-    /// Flushes the inner IO handler.
-    pub async fn flush(&mut self) -> Result<()> {
-        self.writer.flush().await.map_err(Into::into)
-    }
-}
-
-/// Handles receiving messages of type `T` from the underlying [AsyncRead] of type `W`.
-#[derive(Debug)]
-pub struct AsyncDecoder<T, R> {
-    buffer: Vec<u8>,
-    reader: R,
-    _phantom: std::marker::PhantomData<fn() -> T>,
-}
-
-impl<T, R> AsyncDecoder<T, R> {
-    /// Wraps the underlying IO handler.
-    pub fn new(reader: R) -> Self {
-        Self {
-            buffer: Vec::with_capacity(BUFFER_SIZE),
-            reader,
-            _phantom: Default::default(),
-        }
-    }
-
-    /// Unwraps the underlying IO handler.
-    pub fn into_inner(self) -> R {
-        self.reader
-    }
-}
-
-impl<T, R> AsyncDecoder<T, R>
-where
-    T: Decode,
-    R: AsyncRead + Unpin,
-{
-    /// Decodes the next message from the underlying IO handler.
-    /// Does not read any excessive bytes.
-    pub async fn receive(&mut self) -> Result<Option<T>> {
-        let mut len_buffer = [0; 4];
-        match self.reader.read_exact(&mut len_buffer).await {
-            Ok(..) => {}
-            Err(e) if e.kind() == ErrorKind::UnexpectedEof => return Ok(None),
-            Err(e) => Err(e)?,
-        }
-        let len = u32::from_be_bytes(len_buffer);
-
-        self.buffer.resize(len as usize, 0);
-        self.reader.read_exact(&mut self.buffer).await?;
-
-        let value = bincode::decode_from_slice(&self.buffer, bincode::config::standard())?.0;
-
-        Ok(Some(value))
-    }
-}
-
-/// Creates a new pair of [`AsyncEncoder`] and [`AsyncDecoder`], using the given asynchronous
-/// [`TcpStream`](tokio::net::TcpStream).
-pub fn make_async_framed<T1: Encode, T2: Decode>(
-    stream: tokio::net::TcpStream,
-) -> (
-    AsyncEncoder<T1, OwnedWriteHalf>,
-    AsyncDecoder<T2, OwnedReadHalf>,
-) {
-    let (reader, writer) = stream.into_split();
-
-    let sender = AsyncEncoder::new(writer);
-    let receiver = AsyncDecoder::new(reader);
-
-    (sender, receiver)
-}