[tunnelbroker] Add exponential backoff to AMQP connection
ClosedPublic
Actions

Authored by bartek on Oct 3 2024, 8:42 AM.

Details

Reviewers

kamil
will
varun

Commits

rCOMM826944b480a4: [tunnelbroker] Add exponential backoff to AMQP connection

Summary

Added connection retries for AMQP. This only works on Tunnelbroker startup, it doesn't fix our logic yet for disconnects while TB is already running

Depends on D13593

Test Plan

Started Tunnelbroker, but waited with starting RabbitMQ. After running Rabbit, the connection was established after a few attempts

Diff Detail

Repository

rCOMM Comm

Lint

Lint Not Applicable

Unit

Tests Not Applicable

Event Timeline

bartek created this revision.Oct 3 2024, 8:42 AM

bartek held this revision as a draft.

Herald added subscribers: tomek, ashoat. · View Herald TranscriptOct 3 2024, 8:42 AM

bartek added a child revision: D13595: [tunnelbroker] Rewrite AMQP handling.Oct 3 2024, 8:57 AM

Harbormaster completed remote builds in B32045: Diff 44872.Oct 3 2024, 8:59 AM

bartek published this revision for review.Oct 3 2024, 9:01 AM

bartek added inline comments.

services/tunnelbroker/src/amqp.rs
32–36 ↗	(On Diff #44872)	I chose this experimentally basing on how long local rabbitmq gets up, but we might want to add more retries.
56 ↗	(On Diff #44872)	I made it this way because this function is going to be split in the next diffs

bartek added a child revision: D13600: [tunnelbroker] Split AMQP connect function.Oct 3 2024, 1:34 PM

kamil accepted this revision.Oct 7 2024, 1:34 AM

This revision is now accepted and ready to land.Oct 7 2024, 1:34 AM

Closed by commit rCOMM826944b480a4: [tunnelbroker] Add exponential backoff to AMQP connection (authored by bartek). · Explain WhyOct 7 2024, 7:28 AM

This revision was automatically updated to reflect the committed changes.

bartek added a commit: rCOMM826944b480a4: [tunnelbroker] Add exponential backoff to AMQP connection.

Revision Contents
Changeset List

Path

Size

services/

tunnelbroker/

src/

amqp.rs

42 lines

shared/

comm-lib/

src/

database.rs

5 lines

Diff 44930

View Options

services/tunnelbroker/src/amqp.rs

Show First 20 Lines • Show All 90 Lines • ▼ Show 20 Lines	Ok(Self {
.try_into()		.try_into()
.expect("Channels vec size doesn't match array size"),		.expect("Channels vec size doesn't match array size"),
})		})
}		}

fn get_channel(		fn get_channel(
&self,		&self,
id_hash: impl std::hash::Hash,		id_hash: impl std::hash::Hash,
) -> Result<lapin::Channel, lapin::Error> {		) -> Option<lapin::Channel> {
let channel_idx = Self::channel_idx_for_hash(id_hash);		let channel_idx = Self::channel_idx_for_hash(id_hash);
let channel = self.channels[channel_idx].clone();		let channel = self.channels[channel_idx].clone();
let channel_id = channel.id();		let channel_id = channel.id();
tracing::trace!(channel_id, channel_idx, "Retrieving AMQP Channel");		tracing::trace!(channel_id, channel_idx, "Retrieving AMQP Channel");
Ok(channel)
		if channel.status().connected() {
		return Some(channel);
		}
		warn!(channel_id, channel_idx, "Channel is dead!");
		None
		}

		async fn reset_channel(
		&mut self,
		id_hash: impl std::hash::Hash,
		) -> Result<lapin::Channel, lapin::Error> {
		let channel_idx = Self::channel_idx_for_hash(id_hash);

		let existing_channel = &self.channels[channel_idx];
		if existing_channel.status().connected() {
		return Ok(existing_channel.clone());
		}

		let new_channel = self.conn.create_channel().await?;
		debug!(
		"Channel for idx={channel_idx} was recreated, new Id={}",
		new_channel.id()
		);
		self.channels[channel_idx] = new_channel.clone();
		Ok(new_channel)
}		}

fn is_connected(&self) -> bool {		fn is_connected(&self) -> bool {
self.conn.status().connected()		self.conn.status().connected()
}		}

fn channel_idx_for_hash(id_hash: impl std::hash::Hash) -> usize {		fn channel_idx_for_hash(id_hash: impl std::hash::Hash) -> usize {
// We have channel pool and want to distribute them between connected		// We have channel pool and want to distribute them between connected
Show All 22 Lines	pub async fn connect() -> Result<Self, lapin::Error> {
let inner = Arc::new(RwLock::new(conn));		let inner = Arc::new(RwLock::new(conn));
info!("Connected to AMQP endpoint: {}", &CONFIG.amqp_uri);		info!("Connected to AMQP endpoint: {}", &CONFIG.amqp_uri);

Ok(Self { inner })		Ok(Self { inner })
}		}

pub async fn channel(		pub async fn channel(
&self,		&self,
id_hash: impl std::hash::Hash,		id_hash: impl std::hash::Hash + Clone,
) -> Result<lapin::Channel, lapin::Error> {		) -> Result<lapin::Channel, lapin::Error> {
if !self.is_connected().await {		if !self.is_connected().await {
warn!("AMQP disconnected while retrieving channel");		warn!("AMQP disconnected while retrieving channel");
self.reset_conn().await?;		self.reset_conn().await?;
}		}
self.inner.read().await.get_channel(id_hash)
		{
		let inner = self.inner.read().await;
		if let Some(channel) = inner.get_channel(id_hash.clone()) {
		return Ok(channel);
		}
		}

		self.inner.write().await.reset_channel(id_hash).await
}		}

async fn reset_conn(&self) -> Result<(), lapin::Error> {		async fn reset_conn(&self) -> Result<(), lapin::Error> {
debug!("Resetting connection...");		debug!("Resetting connection...");
let mut inner = self.inner.write().await;		let mut inner = self.inner.write().await;
if !inner.is_connected() {		if !inner.is_connected() {
let new_conn = ConnectionInner::new().await?;		let new_conn = ConnectionInner::new().await?;
*inner = new_conn;		*inner = new_conn;
▲ Show 20 Lines • Show All 41 Lines • Show Last 20 Lines