mirrord-protocol update (part of passthrough mirroring prep)

changelog.d/+incoming-protocol-extended.internal.md

@@ -0,0 +1 @@
+Adjusted mirrord-protocol in preparation for passthrough mirroring.

mirrord/agent/src/sniffer.rs

@@ -408,7 +408,7 @@ mod test {
 
     use api::TcpSnifferApi;
     use mirrord_protocol::{
-        tcp::{DaemonTcp, LayerTcp, NewTcpConnection, TcpClose, TcpData},
+        tcp::{DaemonTcp, LayerTcp, NewTcpConnectionV1, TcpClose, TcpData},
         ConnectionId, LogLevel,
     };
     use rstest::rstest;
@@ -525,7 +525,7 @@ mod test {
         let (message, log) = api.recv().await.unwrap();
         assert_eq!(
             message,
-            DaemonTcp::NewConnection(NewTcpConnection {
+            DaemonTcp::NewConnectionV1(NewTcpConnectionV1 {
                 connection_id: 0,
                 remote_address: "1.1.1.1".parse().unwrap(),
                 destination_port: 80,
@@ -681,7 +681,7 @@ mod test {
         let (message, log) = api.recv().await.unwrap();
         assert_eq!(
             message,
-            DaemonTcp::NewConnection(NewTcpConnection {
+            DaemonTcp::NewConnectionV1(NewTcpConnectionV1 {
                 connection_id: 0,
                 remote_address: session_id.source_addr.into(),
                 destination_port: session_id.dest_port,
@@ -784,7 +784,7 @@ mod test {
             assert_eq!(log, None);
             assert_eq!(
                 msg,
-                DaemonTcp::NewConnection(NewTcpConnection {
+                DaemonTcp::NewConnectionV1(NewTcpConnectionV1 {
                     connection_id: i as ConnectionId,
                     remote_address: source_addr.into(),
                     destination_port: 80,
@@ -817,7 +817,7 @@ mod test {
         assert_eq!(log, None);
         assert_eq!(
             msg,
-            DaemonTcp::NewConnection(NewTcpConnection {
+            DaemonTcp::NewConnectionV1(NewTcpConnectionV1 {
                 connection_id: TcpSnifferApi::CONNECTION_CHANNEL_SIZE as ConnectionId,
                 remote_address: source_addr.into(),
                 destination_port: 80,

mirrord/agent/src/sniffer/api.rs

@@ -2,7 +2,7 @@ use std::ops::RangeInclusive;
 
 use futures::{stream::FuturesUnordered, StreamExt};
 use mirrord_protocol::{
-    tcp::{DaemonTcp, LayerTcp, NewTcpConnection, TcpClose, TcpData},
+    tcp::{DaemonTcp, LayerTcp, NewTcpConnectionV1, TcpClose, TcpData},
     ConnectionId, LogMessage, Port,
 };
 use tokio::sync::{
@@ -111,7 +111,7 @@ impl TcpSnifferApi {
                     self.connections.insert(id, StreamNotifyClose::new(BroadcastStream::new(conn.data)));
 
                     Ok((
-                        DaemonTcp::NewConnection(NewTcpConnection {
+                        DaemonTcp::NewConnectionV1(NewTcpConnectionV1 {
                             connection_id: id,
                             remote_address: conn.session_id.source_addr.into(),
                             local_address: conn.session_id.dest_addr.into(),

mirrord/agent/src/sniffer/messages.rs

@@ -38,7 +38,7 @@ pub(crate) struct SnifferCommand {
 /// New TCP connection picked up by [`TcpConnectionSniffer`](super::TcpConnectionSniffer).
 pub(crate) struct SniffedConnection {
     /// Parameters of this connection's TCP session.
-    /// Can be used to create [`NewTcpConnection`](mirrord_protocol::tcp::NewTcpConnection).
+    /// Can be used to create [`NewTcpConnectionV1`](mirrord_protocol::tcp::NewTcpConnectionV1).
     pub session_id: TcpSessionDirectionId,
     /// For receiving data from this connection.
     pub data: broadcast::Receiver<Vec<u8>>,

mirrord/agent/src/steal/connection.rs

@@ -10,7 +10,7 @@ use mirrord_protocol::{
     tcp::{
         ChunkedRequest, ChunkedRequestBodyV1, ChunkedRequestErrorV1, ChunkedRequestErrorV2,
         ChunkedRequestStartV1, ChunkedRequestStartV2, DaemonTcp, HttpRequest, HttpRequestMetadata,
-        HttpRequestTransportType, InternalHttpBody, InternalHttpBodyFrame, InternalHttpBodyNew,
+        IncomingTrafficTransportType, InternalHttpBody, InternalHttpBodyFrame, InternalHttpBodyNew,
         InternalHttpRequest, StealType, TcpClose, TcpData, HTTP_CHUNKED_REQUEST_V2_VERSION,
         HTTP_CHUNKED_REQUEST_VERSION, HTTP_FRAMED_VERSION,
     },
@@ -49,7 +49,7 @@ struct MatchedHttpRequest {
     request_id: RequestId,
     request: Request<Incoming>,
     metadata: HttpRequestMetadata,
-    transport: HttpRequestTransportType,
+    transport: IncomingTrafficTransportType,
 }
 
 impl MatchedHttpRequest {
@@ -58,7 +58,7 @@ impl MatchedHttpRequest {
         request_id: RequestId,
         request: Request<Incoming>,
         metadata: HttpRequestMetadata,
-        transport: HttpRequestTransportType,
+        transport: IncomingTrafficTransportType,
     ) -> Self {
         HTTP_REQUEST_IN_PROGRESS_COUNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
 
@@ -517,7 +517,7 @@ impl TcpConnectionStealer {
 
                 let _ = client
                     .tx
-                    .send(StealerMessage::TcpSteal(DaemonTcp::NewConnection(
+                    .send(StealerMessage::TcpSteal(DaemonTcp::NewConnectionV1(
                         connection,
                     )))
                     .await;
@@ -785,7 +785,7 @@ mod test {
     use hyper_util::rt::TokioIo;
     use mirrord_protocol::tcp::{
         ChunkedRequest, DaemonTcp, Filter, HttpFilter, HttpRequestMetadata,
-        HttpRequestTransportType, InternalHttpBodyFrame, StealType,
+        IncomingTrafficTransportType, InternalHttpBodyFrame, StealType,
     };
     use rstest::rstest;
     use tokio::{
@@ -896,7 +896,7 @@ mod test {
                 source: "1.3.3.7:1337".parse().unwrap(),
                 destination: "2.1.3.7:80".parse().unwrap(),
             },
-            transport: HttpRequestTransportType::Tcp,
+            transport: IncomingTrafficTransportType::Tcp,
             request_id: 0,
             request,
         });
@@ -972,7 +972,7 @@ mod test {
                 source: "1.3.3.7:1337".parse().unwrap(),
                 destination: "2.1.3.7:80".parse().unwrap(),
             },
-            transport: HttpRequestTransportType::Tcp,
+            transport: IncomingTrafficTransportType::Tcp,
             request_id: 0,
             request,
         });

mirrord/agent/src/steal/connections.rs

@@ -10,7 +10,7 @@ use std::{
 
 use hyper::{body::Incoming, Request, Response};
 use mirrord_protocol::{
-    tcp::{HttpRequestMetadata, HttpRequestTransportType, NewTcpConnection},
+    tcp::{HttpRequestMetadata, IncomingTrafficTransportType, NewTcpConnectionV1},
     ConnectionId, LogMessage, RequestId,
 };
 use original_destination::OriginalDestination;
@@ -133,12 +133,12 @@ pub enum ConnectionMessageOut {
         request: Request<Incoming>,
         id: RequestId,
         metadata: HttpRequestMetadata,
-        transport: HttpRequestTransportType,
+        transport: IncomingTrafficTransportType,
     },
     /// Subscribed the client to a new TCP connection.
     ///
     /// This variant translates to
-    /// [`DaemonTcp::NewConnection`](mirrord_protocol::tcp::DaemonTcp::NewConnection).
+    /// [`DaemonTcp::NewConnectionV1`](mirrord_protocol::tcp::DaemonTcp::NewConnectionV1).
     ///
     /// # Note
     ///
@@ -147,7 +147,7 @@ pub enum ConnectionMessageOut {
     /// should follow.
     SubscribedTcp {
         client_id: ClientId,
-        connection: NewTcpConnection,
+        connection: NewTcpConnectionV1,
     },
     /// Subscribed the client to a new filtered HTTP connection.
     ///
@@ -489,7 +489,7 @@ impl ConnectionTask {
                 self.tx
                     .send(ConnectionMessageOut::SubscribedTcp {
                         client_id,
-                        connection: NewTcpConnection {
+                        connection: NewTcpConnectionV1 {
                             connection_id: self.connection_id,
                             remote_address: source.ip(),
                             source_port: source.port(),

mirrord/agent/src/steal/connections/filtered.rs

@@ -17,7 +17,7 @@ use hyper::{
 use hyper_util::rt::{TokioExecutor, TokioIo};
 use mirrord_protocol::{
     tcp::{
-        HttpRequestMetadata, HttpRequestTransportType, HTTP_CHUNKED_REQUEST_V2_VERSION,
+        HttpRequestMetadata, IncomingTrafficTransportType, HTTP_CHUNKED_REQUEST_V2_VERSION,
         HTTP_FILTERED_UPGRADE_VERSION,
     },
     ConnectionId, LogMessage, RequestId,
@@ -712,11 +712,11 @@ where
         self.next_request_id += 1;
 
         let transport = match self.original_destination.connector() {
-            Some(connector) => HttpRequestTransportType::Tls {
+            Some(connector) => IncomingTrafficTransportType::Tls {
                 alpn_protocol: connector.alpn_protocol().map(Vec::from),
                 server_name: connector.server_name().map(|name| name.to_str().into()),
             },
-            None => HttpRequestTransportType::Tcp,
+            None => IncomingTrafficTransportType::Tcp,
         };
 
         tx.send(ConnectionMessageOut::Request {
@@ -1327,7 +1327,7 @@ mod test {
                             connection_id: TestSetup::CONNECTION_ID,
                             id,
                             metadata: HttpRequestMetadata::V1 { destination, .. },
-                            transport: HttpRequestTransportType::Tcp,
+                            transport: IncomingTrafficTransportType::Tcp,
                             ..
                         } => {
                             assert_eq!(received_client_id, client_id);
@@ -1424,7 +1424,7 @@ mod test {
                         connection_id: TestSetup::CONNECTION_ID,
                         id,
                         metadata: HttpRequestMetadata::V1 { destination, .. },
-                        transport: HttpRequestTransportType::Tcp,
+                        transport: IncomingTrafficTransportType::Tcp,
                         ..
                     } => {
                         assert_eq!(received_client_id, subscribed_client_id);
@@ -1511,7 +1511,7 @@ mod test {
                         connection_id: TestSetup::CONNECTION_ID,
                         id,
                         metadata: HttpRequestMetadata::V1 { destination, .. },
-                        transport: HttpRequestTransportType::Tcp,
+                        transport: IncomingTrafficTransportType::Tcp,
                         ..
                     } => {
                         assert_eq!(destination, setup.original_address);
@@ -1551,7 +1551,7 @@ mod test {
                         connection_id: TestSetup::CONNECTION_ID,
                         id,
                         metadata: HttpRequestMetadata::V1 { destination, .. },
-                        transport: HttpRequestTransportType::Tcp,
+                        transport: IncomingTrafficTransportType::Tcp,
                         ..
                     } => {
                         assert_eq!(destination, setup.original_address);
@@ -1617,7 +1617,7 @@ mod test {
                         connection_id: TestSetup::CONNECTION_ID,
                         id,
                         metadata: HttpRequestMetadata::V1 { destination, .. },
-                        transport: HttpRequestTransportType::Tcp,
+                        transport: IncomingTrafficTransportType::Tcp,
                         ..
                     } => {
                         assert_eq!(destination, setup.original_address);
@@ -1702,7 +1702,7 @@ mod test {
                         connection_id: TestSetup::CONNECTION_ID,
                         id,
                         metadata: HttpRequestMetadata::V1 { destination, .. },
-                        transport: HttpRequestTransportType::Tcp,
+                        transport: IncomingTrafficTransportType::Tcp,
                         ..
                     } => {
                         assert_eq!(destination, setup.original_address);
@@ -1755,7 +1755,7 @@ mod test {
                         id,
                         metadata: HttpRequestMetadata::V1 { destination, .. },
                         request,
-                        transport: HttpRequestTransportType::Tcp,
+                        transport: IncomingTrafficTransportType::Tcp,
                         ..
                     } => {
                         assert_eq!(destination, setup.original_address);
@@ -1888,7 +1888,7 @@ mod test {
                         connection_id: TestSetup::CONNECTION_ID,
                         id,
                         metadata: HttpRequestMetadata::V1 { destination, .. },
-                        transport: HttpRequestTransportType::Tcp,
+                        transport: IncomingTrafficTransportType::Tcp,
                         ..
                     } => {
                         assert_eq!(destination, setup.original_address);
@@ -2136,7 +2136,7 @@ mod test {
                 connection_id: 0,
                 id: 0,
                 transport:
-                    HttpRequestTransportType::Tls {
+                    IncomingTrafficTransportType::Tls {
                         alpn_protocol,
                         server_name,
                     },
@@ -2221,7 +2221,7 @@ mod test {
                         connection_id: TestSetup::CONNECTION_ID,
                         id,
                         metadata: HttpRequestMetadata::V1 { destination, .. },
-                        transport: HttpRequestTransportType::Tcp,
+                        transport: IncomingTrafficTransportType::Tcp,
                         ..
                     } => {
                         assert_eq!(received_client_id, 0);

mirrord/cli/src/port_forward.rs

@@ -1039,7 +1039,7 @@ mod test {
         },
         tcp::{
             DaemonTcp, Filter, HttpRequest, HttpResponse, InternalHttpBody, InternalHttpBodyFrame,
-            InternalHttpRequest, InternalHttpResponse, LayerTcp, LayerTcpSteal, NewTcpConnection,
+            InternalHttpRequest, InternalHttpResponse, LayerTcp, LayerTcpSteal, NewTcpConnectionV1,
             StealType, TcpClose, TcpData,
         },
         ClientMessage, DaemonMessage,
@@ -1354,8 +1354,8 @@ mod test {
 
         // send new connection from agent and some data
         test_connection
-            .send(DaemonMessage::Tcp(DaemonTcp::NewConnection(
-                NewTcpConnection {
+            .send(DaemonMessage::Tcp(DaemonTcp::NewConnectionV1(
+                NewTcpConnectionV1 {
                     connection_id: 1,
                     remote_address,
                     destination_port,
@@ -1429,8 +1429,8 @@ mod test {
 
         // send new connection from agent and some data
         test_connection
-            .send(DaemonMessage::TcpSteal(DaemonTcp::NewConnection(
-                NewTcpConnection {
+            .send(DaemonMessage::TcpSteal(DaemonTcp::NewConnectionV1(
+                NewTcpConnectionV1 {
                     connection_id: 1,
                     remote_address,
                     destination_port,
@@ -1526,8 +1526,8 @@ mod test {
 
         // send new connections from agent and some data
         test_connection
-            .send(DaemonMessage::Tcp(DaemonTcp::NewConnection(
-                NewTcpConnection {
+            .send(DaemonMessage::Tcp(DaemonTcp::NewConnectionV1(
+                NewTcpConnectionV1 {
                     connection_id: 1,
                     remote_address,
                     destination_port: destination_port_1,
@@ -1539,8 +1539,8 @@ mod test {
         let mut stream_1 = listener_1.accept().await.unwrap().0;
 
         test_connection
-            .send(DaemonMessage::Tcp(DaemonTcp::NewConnection(
-                NewTcpConnection {
+            .send(DaemonMessage::Tcp(DaemonTcp::NewConnectionV1(
+                NewTcpConnectionV1 {
                     connection_id: 2,
                     remote_address,
                     destination_port: destination_port_2,