Skip to content

Commit 25abac0

Browse files
WonshtrumKeksoj
Wonshtrum
authored and
Keksoj
committed
Update metrics, backend status and events
Signed-off-by: Eloi DEMOLIS <[email protected]>
1 parent 4ce1404 commit 25abac0

File tree

8 files changed

+402
-154
lines changed

8 files changed

+402
-154
lines changed

‎command/src/proto/display.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -152,7 +152,7 @@ impl Response {
152152
}
153153

154154
impl ResponseContent {
155-
fn display(&self, json: bool) -> Result<(), DisplayError> {
155+
pub fn display(&self, json: bool) -> Result<(), DisplayError> {
156156
let content_type = match &self.content_type {
157157
Some(content_type) => content_type,
158158
None => return Ok(println!("No content")),

‎lib/src/backends.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -297,7 +297,8 @@ impl BackendMap {
297297
})?;
298298
self.available = true;
299299

300-
Ok((next_backend.clone(), tcp_stream))
300+
drop(borrowed_backend);
301+
Ok((next_backend, tcp_stream))
301302
}
302303

303304
pub fn backend_from_sticky_session(

‎lib/src/http.rs‎

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -259,15 +259,16 @@ impl HttpSession {
259259
return None;
260260
};
261261
let backend = mux.router.backends.remove(&back_token).unwrap();
262-
let (cluster_id, backend_readiness, backend_socket, mut container_backend_timeout) =
262+
let (cluster_id, backend, backend_readiness, backend_socket, mut container_backend_timeout) =
263263
match backend {
264264
mux::Connection::H1(mux::ConnectionH1 {
265-
position: mux::Position::Client(mux::BackendStatus::Connected(cluster_id)),
265+
position:
266+
mux::Position::Client(cluster_id, backend, mux::BackendStatus::Connected),
266267
readiness,
267268
socket,
268269
timeout_container,
269270
..
270-
}) => (cluster_id, readiness, socket, timeout_container),
271+
}) => (cluster_id, backend, readiness, socket, timeout_container),
271272
mux::Connection::H1(_) => {
272273
error!("The backend disconnected just after upgrade, abort");
273274
return None;
@@ -283,11 +284,12 @@ impl HttpSession {
283284
container_frontend_timeout.reset();
284285
container_backend_timeout.reset();
285286

287+
let backend_id = backend.borrow().backend_id.clone();
286288
let mut pipe = Pipe::new(
287289
stream.back.storage.buffer,
288-
None,
290+
Some(backend_id),
289291
Some(backend_socket),
290-
None,
292+
Some(backend),
291293
Some(container_backend_timeout),
292294
Some(container_frontend_timeout),
293295
Some(cluster_id),
@@ -307,7 +309,6 @@ impl HttpSession {
307309

308310
gauge_add!("protocol.http", -1);
309311
gauge_add!("protocol.ws", 1);
310-
gauge_add!("http.active_requests", -1);
311312
gauge_add!("websocket.active_requests", 1);
312313
Some(HttpStateMachine::WebSocket(pipe))
313314
}

‎lib/src/https.rs‎

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -343,15 +343,16 @@ impl HttpsSession {
343343
return None;
344344
};
345345
let backend = mux.router.backends.remove(&back_token).unwrap();
346-
let (cluster_id, backend_readiness, backend_socket, mut container_backend_timeout) =
346+
let (cluster_id, backend, backend_readiness, backend_socket, mut container_backend_timeout) =
347347
match backend {
348348
mux::Connection::H1(mux::ConnectionH1 {
349-
position: mux::Position::Client(mux::BackendStatus::Connected(cluster_id)),
349+
position:
350+
mux::Position::Client(cluster_id, backend, mux::BackendStatus::Connected),
350351
readiness,
351352
socket,
352353
timeout_container,
353354
..
354-
}) => (cluster_id, readiness, socket, timeout_container),
355+
}) => (cluster_id, backend, readiness, socket, timeout_container),
355356
mux::Connection::H1(_) => {
356357
error!("The backend disconnected just after upgrade, abort");
357358
return None;
@@ -367,11 +368,12 @@ impl HttpsSession {
367368
container_frontend_timeout.reset();
368369
container_backend_timeout.reset();
369370

371+
let backend_id = backend.borrow().backend_id.clone();
370372
let mut pipe = Pipe::new(
371373
stream.back.storage.buffer,
372-
None,
374+
Some(backend_id),
373375
Some(backend_socket),
374-
None,
376+
Some(backend),
375377
Some(container_backend_timeout),
376378
Some(container_frontend_timeout),
377379
Some(cluster_id),
@@ -391,7 +393,6 @@ impl HttpsSession {
391393

392394
gauge_add!("protocol.https", -1);
393395
gauge_add!("protocol.wss", 1);
394-
gauge_add!("http.active_requests", -1);
395396
gauge_add!("websocket.active_requests", 1);
396397
Some(HttpsStateMachine::WebSocket(pipe))
397398
}

‎lib/src/lib.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -583,7 +583,7 @@ pub enum BackendConnectAction {
583583
pub enum BackendConnectionError {
584584
#[error("Not found: {0:?}")]
585585
NotFound(ObjectKind),
586-
#[error("Too many connections on cluster {0:?}")]
586+
#[error("Too many failed attemps on cluster {0:?}")]
587587
MaxConnectionRetries(Option<String>),
588588
#[error("the sessions slab has reached maximum capacity")]
589589
MaxSessionsMemory,

‎lib/src/protocol/mux/h1.rs‎

Lines changed: 62 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,9 @@ use sozu_command::ready::Ready;
33
use crate::{
44
println_,
55
protocol::mux::{
6-
debug_kawa, forcefully_terminate_answer, parser::H2Error, set_default_answer, update_readiness_after_read, update_readiness_after_write, BackendStatus, Context, Endpoint, GlobalStreamId, MuxResult, Position, StreamState
6+
debug_kawa, forcefully_terminate_answer, parser::H2Error, set_default_answer,
7+
update_readiness_after_read, update_readiness_after_write, BackendStatus, Context,
8+
Endpoint, GlobalStreamId, MuxResult, Position, StreamState,
79
},
810
socket::SocketHandler,
911
timer::TimeoutContainer,
@@ -44,6 +46,16 @@ impl<Front: SocketHandler> ConnectionH1<Front> {
4446
let kawa = parts.rbuffer;
4547
let (size, status) = self.socket.socket_read(kawa.storage.space());
4648
kawa.storage.fill(size);
49+
match self.position {
50+
Position::Client(..) => {
51+
count!("back_bytes_in", size as i64);
52+
parts.metrics.backend_bin += size;
53+
}
54+
Position::Server => {
55+
count!("bytes_in", size as i64);
56+
parts.metrics.bin += size;
57+
}
58+
}
4759
if update_readiness_after_read(size, status, &mut self.readiness) {
4860
return MuxResult::Continue;
4961
}
@@ -53,7 +65,7 @@ impl<Front: SocketHandler> ConnectionH1<Front> {
5365
debug_kawa(kawa);
5466
if kawa.is_error() {
5567
match self.position {
56-
Position::Client(_) => {
68+
Position::Client(..) => {
5769
let StreamState::Linked(token) = stream.state else {
5870
unreachable!()
5971
};
@@ -79,15 +91,11 @@ impl<Front: SocketHandler> ConnectionH1<Front> {
7991
.interest
8092
.insert(Ready::WRITABLE)
8193
}
82-
match self.position {
83-
Position::Server => {
84-
if !was_main_phase {
85-
self.requests += 1;
86-
println_!("REQUESTS: {}", self.requests);
87-
stream.state = StreamState::Link
88-
}
89-
}
90-
Position::Client(_) => {}
94+
if !was_main_phase && self.position.is_server() {
95+
self.requests += 1;
96+
println_!("REQUESTS: {}", self.requests);
97+
gauge_add!("http.active_requests", 1);
98+
stream.state = StreamState::Link
9199
}
92100
};
93101
MuxResult::Continue
@@ -101,7 +109,8 @@ impl<Front: SocketHandler> ConnectionH1<Front> {
101109
println_!("======= MUX H1 WRITABLE {:?}", self.position);
102110
self.timeout_container.reset();
103111
let stream = &mut context.streams[self.stream];
104-
let kawa = stream.wbuffer(&self.position);
112+
let parts = stream.split(&self.position);
113+
let kawa = parts.wbuffer;
105114
kawa.prepare(&mut kawa::h1::BlockConverter);
106115
debug_kawa(kawa);
107116
let bufs = kawa.as_io_slice();
@@ -111,13 +120,23 @@ impl<Front: SocketHandler> ConnectionH1<Front> {
111120
}
112121
let (size, status) = self.socket.socket_write_vectored(&bufs);
113122
kawa.consume(size);
123+
match self.position {
124+
Position::Client(..) => {
125+
count!("back_bytes_out", size as i64);
126+
parts.metrics.backend_bout += size;
127+
}
128+
Position::Server => {
129+
count!("bytes_out", size as i64);
130+
parts.metrics.bout += size;
131+
}
132+
}
114133
if update_readiness_after_write(size, status, &mut self.readiness) {
115134
return MuxResult::Continue;
116135
}
117136

118137
if kawa.is_terminated() && kawa.is_completed() {
119138
match self.position {
120-
Position::Client(_) => self.readiness.interest.insert(Ready::READABLE),
139+
Position::Client(..) => self.readiness.interest.insert(Ready::READABLE),
121140
Position::Server => {
122141
if stream.context.closing {
123142
return MuxResult::CloseSession;
@@ -153,7 +172,12 @@ impl<Front: SocketHandler> ConnectionH1<Front> {
153172
_ => {}
154173
}
155174
// ACCESS LOG
156-
stream.generate_access_log(false, Some(String::from("H1")), context.listener.clone());
175+
stream.generate_access_log(
176+
false,
177+
Some(String::from("H1")),
178+
context.listener.clone(),
179+
);
180+
stream.metrics.reset();
157181
let old_state = std::mem::replace(&mut stream.state, StreamState::Unlinked);
158182
if stream.context.keep_alive_frontend {
159183
self.timeout_container.reset();
@@ -180,9 +204,9 @@ impl<Front: SocketHandler> ConnectionH1<Front> {
180204
}
181205

182206
pub fn force_disconnect(&mut self) -> MuxResult {
183-
match self.position {
184-
Position::Client(_) => {
185-
self.position = Position::Client(BackendStatus::Disconnecting);
207+
match &mut self.position {
208+
Position::Client(_, _, status) => {
209+
*status = BackendStatus::Disconnecting;
186210
self.readiness.event = Ready::HUP;
187211
MuxResult::Continue
188212
}
@@ -196,13 +220,13 @@ impl<Front: SocketHandler> ConnectionH1<Front> {
196220
L: ListenerHandler + L7ListenerHandler,
197221
{
198222
match self.position {
199-
Position::Client(BackendStatus::KeepAlive(_))
200-
| Position::Client(BackendStatus::Disconnecting) => {
223+
Position::Client(_, _, BackendStatus::KeepAlive)
224+
| Position::Client(_, _, BackendStatus::Disconnecting) => {
201225
println_!("close detached client ConnectionH1");
202226
return;
203227
}
204-
Position::Client(BackendStatus::Connecting(_))
205-
| Position::Client(BackendStatus::Connected(_)) => {}
228+
Position::Client(_, _, BackendStatus::Connecting(_))
229+
| Position::Client(_, _, BackendStatus::Connected) => {}
206230
Position::Server => unreachable!(),
207231
}
208232
// reconnection is handled by the server
@@ -221,28 +245,34 @@ impl<Front: SocketHandler> ConnectionH1<Front> {
221245
let stream_context = &mut stream.context;
222246
println_!("end H1 stream {}: {stream_context:#?}", self.stream);
223247
match &mut self.position {
224-
Position::Client(BackendStatus::Connected(cluster_id))
225-
| Position::Client(BackendStatus::Connecting(cluster_id)) => {
248+
Position::Client(_, _, BackendStatus::Connecting(_)) => {
249+
self.stream = usize::MAX;
250+
self.force_disconnect();
251+
}
252+
Position::Client(_, _, status @ BackendStatus::Connected) => {
226253
self.stream = usize::MAX;
227254
// keep alive should probably be used only if the http context is fully reset
228255
// in case end_stream occurs due to an error the connection state is probably
229256
// unrecoverable and should be terminated
230257
if stream_context.keep_alive_backend {
231-
self.position =
232-
Position::Client(BackendStatus::KeepAlive(std::mem::take(cluster_id)))
258+
*status = BackendStatus::KeepAlive;
233259
} else {
234260
self.force_disconnect();
235261
}
236262
}
237-
Position::Client(BackendStatus::KeepAlive(_))
238-
| Position::Client(BackendStatus::Disconnecting) => unreachable!(),
263+
Position::Client(_, _, BackendStatus::KeepAlive)
264+
| Position::Client(_, _, BackendStatus::Disconnecting) => unreachable!(),
239265
Position::Server => match (stream.front.consumed, stream.back.is_main_phase()) {
240266
(true, true) => {
241267
// we have a "forwardable" answer from the back
242268
// if the answer is not terminated we send an RstStream to properly clean the stream
243269
// if it is terminated, we finish the transfer, the backend is not necessary anymore
244270
if !stream.back.is_terminated() {
245-
forcefully_terminate_answer(stream, &mut self.readiness, H2Error::InternalError);
271+
forcefully_terminate_answer(
272+
stream,
273+
&mut self.readiness,
274+
H2Error::InternalError,
275+
);
246276
} else {
247277
stream.state = StreamState::Unlinked;
248278
self.readiness.interest.insert(Ready::WRITABLE);
@@ -271,11 +301,11 @@ impl<Front: SocketHandler> ConnectionH1<Front> {
271301
self.readiness.interest.insert(Ready::ALL);
272302
self.stream = stream;
273303
match &mut self.position {
274-
Position::Client(BackendStatus::KeepAlive(cluster_id)) => {
275-
self.position =
276-
Position::Client(BackendStatus::Connecting(std::mem::take(cluster_id)))
304+
Position::Client(_, _, status @ BackendStatus::KeepAlive) => {
305+
*status = BackendStatus::Connected;
277306
}
278-
Position::Client(_) => {}
307+
Position::Client(_, _, BackendStatus::Disconnecting) => unreachable!(),
308+
Position::Client(_, _, _) => {}
279309
Position::Server => unreachable!(),
280310
}
281311
}

0 commit comments

Comments
 (0)