Refactor backend logic with an Origin struct

I feel like it's going nowhere...
Backend logic is messy and unclear, buffers are needlessly kept alive,
borrowing subsets of the HTTP session is harder and harder, reseting
default answers request is way too hard for what it's worth (it's
probably broken right now), gauges are desynched and we are cloning
again and again String ids...

Signed-off-by: Eloi DEMOLIS <[email protected]>

Author: Wonshtrum

command/assets/custom_200.html

@@ -1,5 +1,4 @@
 HTTP/1.1 200 OK
-%Content-Length: %CONTENT_LENGTH
 Sozu-Id: %REQUEST_ID
 
 <h1>%CLUSTER_ID Custom 200</h1>

command/assets/custom_404.html

@@ -1,6 +1,5 @@
 HTTP/1.1 404 Not Found
 Cache-Control: no-cache
-Connection: close
 Sozu-Id: %REQUEST_ID
 
 <h1>My own 404 error page</h1>

command/assets/custom_503.html

@@ -1,7 +1,6 @@
 HTTP/1.1 503 Service Unavailable
 Cache-Control: no-cache
 Connection: close
-%Content-Length: %CONTENT_LENGTH
 Sozu-Id: %REQUEST_ID
 
 <h1>MyCluster: 503 Service Unavailable</h1>

command/src/state.rs

@@ -1724,6 +1724,7 @@ mod tests {
                 hostname: String::from("test.local"),
                 path: PathRule::prefix(String::from("/abc")),
                 address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
+                required_auth: Some(false),
                 redirect: Some(RedirectPolicy::Forward.into()),
                 redirect_scheme: Some(RedirectScheme::UseSame.into()),
                 ..Default::default()

e2e/src/http_utils/mod.rs

@@ -24,12 +24,17 @@ pub fn http_request<S1: Into<String>, S2: Into<String>, S3: Into<String>, S4: In
     )
 }
 
-pub fn immutable_answer(status: u16) -> String {
+pub fn immutable_answer(status: u16, content_length: bool) -> String {
+    let content_length = if content_length {
+        "\r\nContent-Length: 0"
+    } else {
+        ""
+    };
     match status {
-        400 => String::from("HTTP/1.1 400 Bad Request\r\nCache-Control: no-cache\r\nConnection: close\r\n\r\n"),
-        404 => String::from("HTTP/1.1 404 Not Found\r\nCache-Control: no-cache\r\nConnection: close\r\n\r\n"),
-        502 => String::from("HTTP/1.1 502 Bad Gateway\r\nCache-Control: no-cache\r\nConnection: close\r\n\r\n"),
-        503 => String::from("HTTP/1.1 503 Service Unavailable\r\nCache-Control: no-cache\r\nConnection: close\r\n\r\n"),
+        400 => format!("HTTP/1.1 400 Bad Request\r\nCache-Control: no-cache\r\nConnection: close{content_length}\r\n\r\n"),
+        404 => format!("HTTP/1.1 404 Not Found\r\nCache-Control: no-cache\r\nConnection: close{content_length}\r\n\r\n"),
+        502 => format!("HTTP/1.1 502 Bad Gateway\r\nCache-Control: no-cache\r\nConnection: close{content_length}\r\n\r\n"),
+        503 => format!("HTTP/1.1 503 Service Unavailable\r\nCache-Control: no-cache\r\nConnection: close{content_length}\r\n\r\n"),
         _ => unimplemented!()
     }
 }

e2e/src/tests/tests.rs

@@ -645,10 +645,10 @@ fn try_http_behaviors() -> State {
         .to_http(None)
         .unwrap();
     http_config.answers = BTreeMap::from([
-        ("400".to_string(), immutable_answer(400)),
-        ("404".to_string(), immutable_answer(404)),
-        ("502".to_string(), immutable_answer(502)),
-        ("503".to_string(), immutable_answer(503)),
+        ("400".to_string(), immutable_answer(400, false)),
+        ("404".to_string(), immutable_answer(404, false)),
+        ("502".to_string(), immutable_answer(502, false)),
+        ("503".to_string(), immutable_answer(503, false)),
     ]);
 
     worker.send_proxy_request_type(RequestType::AddHttpListener(http_config));
@@ -671,7 +671,7 @@ fn try_http_behaviors() -> State {
 
     let response = client.receive();
     println!("response: {response:?}");
-    assert_eq!(response, Some(immutable_answer(404)));
+    assert_eq!(response, Some(immutable_answer(404, true)));
     assert_eq!(client.receive(), None);
 
     worker.send_proxy_request_type(RequestType::AddHttpFrontend(RequestHttpFrontend {
@@ -686,7 +686,7 @@ fn try_http_behaviors() -> State {
 
     let response = client.receive();
     println!("response: {response:?}");
-    assert_eq!(response, Some(immutable_answer(503)));
+    assert_eq!(response, Some(immutable_answer(503, true)));
     assert_eq!(client.receive(), None);
 
     let back_address = create_local_address();
@@ -706,7 +706,7 @@ fn try_http_behaviors() -> State {
 
     let response = client.receive();
     println!("response: {response:?}");
-    assert_eq!(response, Some(immutable_answer(400)));
+    assert_eq!(response, Some(immutable_answer(400, true)));
     assert_eq!(client.receive(), None);
 
     let mut backend = SyncBackend::new("backend", back_address, "TEST\r\n\r\n");
@@ -723,7 +723,7 @@ fn try_http_behaviors() -> State {
     let response = client.receive();
     println!("request: {request:?}");
     println!("response: {response:?}");
-    assert_eq!(response, Some(immutable_answer(502)));
+    assert_eq!(response, Some(immutable_answer(502, true)));
     assert_eq!(client.receive(), None);
 
     info!("expecting 200");
@@ -786,7 +786,7 @@ fn try_http_behaviors() -> State {
     let response = client.receive();
     println!("request: {request:?}");
     println!("response: {response:?}");
-    assert_eq!(response, Some(immutable_answer(503)));
+    assert_eq!(response, Some(immutable_answer(503, true)));
     assert_eq!(client.receive(), None);
 
     worker.send_proxy_request_type(RequestType::RemoveBackend(RemoveBackend {
@@ -984,7 +984,7 @@ fn try_https_redirect() -> State {
     client.connect();
     client.send();
     let answer = client.receive();
-    let expected_answer = format!("{answer_301_prefix}https://example.com/redirected?true\r\n\r\n");
+    let expected_answer = format!("{answer_301_prefix}https://example.com/redirected?true\r\nContent-Length: 0\r\n\r\n");
     assert_eq!(answer, Some(expected_answer));
 
     State::Success

lib/src/http.rs

@@ -209,11 +209,14 @@ impl HttpSession {
         }
     }
 
-    fn upgrade_http(&mut self, http: Http<TcpStream, HttpListener>) -> Option<HttpStateMachine> {
+    fn upgrade_http(
+        &mut self,
+        mut http: Http<TcpStream, HttpListener>,
+    ) -> Option<HttpStateMachine> {
         debug!("http switching to ws");
-        let front_token = self.frontend_token;
-        let back_token = match http.backend_token {
-            Some(back_token) => back_token,
+        let frontend_token = self.frontend_token;
+        let origin = match http.origin.take() {
+            Some(origin) => origin,
             None => {
                 warn!(
                     "Could not upgrade http request on cluster '{:?}' ({:?}) using backend '{:?}' into websocket for request '{}'",
@@ -223,7 +226,7 @@ impl HttpSession {
             }
         };
 
-        let ws_context = http.websocket_context();
+        let websocket_context = http.websocket_context();
         let mut container_frontend_timeout = http.container_frontend_timeout;
         let mut container_backend_timeout = http.container_backend_timeout;
         container_frontend_timeout.reset();
@@ -237,25 +240,25 @@ impl HttpSession {
 
         let mut pipe = Pipe::new(
             backend_buffer,
-            http.context.backend_id,
-            http.backend_socket,
-            http.backend,
+            Some(origin.backend_id),
+            Some(origin.socket),
+            Some(origin.backend),
             Some(container_backend_timeout),
             Some(container_frontend_timeout),
             http.context.cluster_id,
             http.request_stream.storage.buffer,
-            front_token,
+            frontend_token,
             http.frontend_socket,
             self.listener.clone(),
             Protocol::HTTP,
             http.context.id,
             http.context.session_address,
-            ws_context,
+            websocket_context,
         );
 
         pipe.frontend_readiness.event = http.frontend_readiness.event;
         pipe.backend_readiness.event = http.backend_readiness.event;
-        pipe.set_back_token(back_token);
+        pipe.set_back_token(origin.token);
 
         gauge_add!("protocol.http", -1);
         gauge_add!("protocol.ws", 1);

lib/src/https.rs

@@ -329,11 +329,14 @@ impl HttpsSession {
         }
     }
 
-    fn upgrade_http(&self, http: Http<FrontRustls, HttpsListener>) -> Option<HttpsStateMachine> {
+    fn upgrade_http(
+        &self,
+        mut http: Http<FrontRustls, HttpsListener>,
+    ) -> Option<HttpsStateMachine> {
         debug!("https switching to wss");
         let front_token = self.frontend_token;
-        let back_token = match http.backend_token {
-            Some(back_token) => back_token,
+        let origin = match http.origin.take() {
+            Some(origin) => origin,
             None => {
                 warn!(
                     "Could not upgrade https request on cluster '{:?}' ({:?}) using backend '{:?}' into secure websocket for request '{}'",
@@ -343,7 +346,7 @@ impl HttpsSession {
             }
         };
 
-        let ws_context = http.websocket_context();
+        let websocket_context = http.websocket_context();
         let mut container_frontend_timeout = http.container_frontend_timeout;
         let mut container_backend_timeout = http.container_backend_timeout;
         container_frontend_timeout.reset();
@@ -357,25 +360,25 @@ impl HttpsSession {
 
         let mut pipe = Pipe::new(
             backend_buffer,
-            http.context.backend_id,
-            http.backend_socket,
-            http.backend,
+            Some(origin.backend_id),
+            Some(origin.socket),
+            Some(origin.backend),
             Some(container_backend_timeout),
             Some(container_frontend_timeout),
             http.context.cluster_id,
             http.request_stream.storage.buffer,
             front_token,
             http.frontend_socket,
             self.listener.clone(),
-            Protocol::HTTP,
+            Protocol::HTTPS,
             http.context.id,
             http.context.session_address,
-            ws_context,
+            websocket_context,
         );
 
         pipe.frontend_readiness.event = http.frontend_readiness.event;
         pipe.backend_readiness.event = http.backend_readiness.event;
-        pipe.set_back_token(back_token);
+        pipe.set_back_token(origin.token);
 
         gauge_add!("protocol.https", -1);
         gauge_add!("protocol.wss", 1);

lib/src/lib.rs

@@ -953,7 +953,6 @@ pub struct SessionMetrics {
     pub service_start: Option<Instant>,
     pub wait_start: Instant,
 
-    pub backend_id: Option<String>,
     pub backend_start: Option<Instant>,
     pub backend_connected: Option<Instant>,
     pub backend_stop: Option<Instant>,
@@ -971,7 +970,6 @@ impl SessionMetrics {
             bout: 0,
             service_start: None,
             wait_start: Instant::now(),
-            backend_id: None,
             backend_start: None,
             backend_connected: None,
             backend_stop: None,
@@ -1072,7 +1070,7 @@ impl SessionMetrics {
         time!("request_time", request_time.as_millis());
         time!("service_time", service_time.as_millis());
 
-        if let Some(backend_id) = self.backend_id.as_ref() {
+        if let Some(backend_id) = context.backend_id {
             if let Some(backend_response_time) = self.backend_response_time() {
                 record_backend_metrics!(
                     context.cluster_id.as_str_or("-"),