Error: The connection was lost from mosquitto_pub when worker runs in Kubernetes cluster

[ninomllr]

I am trying to get a system to work where we have the following setup through RabbitMQ:

MQTT (IoT Devices) --> RabbitMQ <-- AMQP (.NET Worker)

The c# worker looks like this, as we deleted everything to find out what the issue could be. For faking the IoT devices we are testing with mosquitto_pub. When we run this code locally and publish to RabbitMQ with Mosquitto, everything works (even with more code where we subscribe and interpret the messages and write them to a database). We also tried to dockerize the applications and we run everything in Docker locally, with success.

As soon as we install the .NET worker on our Kubernetes cluster though, we receive Error: The connection was lost. messages on publishing through Mosquitto. Every other message works and then we get the error again. This behaviour happens ONLY when we have a Worker running in Kubernetes (microk8s instance).

We have tested the following setups:

• Worker in VSCode debugging --> RabbitMQ (on Microk8s) <-- MQTT Mosquitto = works
• Worker in Docker container run locally --> RabbitMQ (on Microk8s) <-- MQTT Mosquitto = works
• Worker in Kubernetes Pod --> RabbitMQ (on Microk8s) <-- MQTT Mosquitto = doesn't work (it works but Mosquitto gives the error every other message).

I would assume that it's maybe not even an error with RabbitMQ or RabbitMQ .NET client, but I am a bit lost on where to find the logs to actually start finding a solution to this. Can someone help out here?

Worker.cs

using System;
using System.Text;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Npgsql;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Exceptions;
using System.Net;
using System.IO;

namespace RabbitMQWorkerService
{
	public class Worker : BackgroundService
	{
		private readonly ILogger<Worker> _logger;
		private readonly IConfiguration _configuration;
		private ConnectionFactory _connectionFactory;
		private IConnection _connection;
		private IModel _channel;
		private Timer _timer;
		private static object _lock = new object();

		private const string QueueName = "aaaa.smartbin.queue";

		/// <summary>
		/// Constructor
		/// </summary>
		/// <param name="logger"></param>
		public Worker(ILogger<Worker> logger, IConfiguration configuration)
		{
			_logger = logger;
			_configuration = configuration;
		}

		/// <summary>
		/// Create async connection with RabbitMQ broker
		/// It will be create one Channel pro Queue
		/// </summary>
		/// <param name="cancellationToken"></param>
		/// <returns></returns>
		public override Task StartAsync(CancellationToken cancellationToken)
		{
			_logger.LogInformation("START WORKER --------------------------.");

			_logger.LogInformation("AFTER TIMER --------------------------.");

			string rabbitHost = _configuration["RabbitHostName"];

			_logger.LogInformation("rabbitHost--------------------------." + rabbitHost);

			int rabbitPort = int.Parse(_configuration["RabbitPort"]);
			string rabbitUserName = _configuration["RabbitUserName"];
			string rabbitPassword = _configuration["RabbitPassword"];

			_connectionFactory = new ConnectionFactory
			{
				HostName = rabbitHost,
				Port = rabbitPort,
				UserName = rabbitUserName,
				Password = rabbitPassword,
				DispatchConsumersAsync = true
			};
			_connection = _connectionFactory.CreateConnection();
			_channel = _connection.CreateModel();



			string ExchangeName = "amq.topic";
			string RoutingKey = "aaaa.smartbin.*";

			return base.StartAsync(cancellationToken);

		}

		/// <summary>
		/// Subscriber / Consumer      
		/// Wait for request from smart bin 
		/// For now take only Topics ala RoutingKey
		/// Queue has more channels depends on IoT devices
		/// </summary>
		/// <param name="stoppingToken"></param>
		/// <returns></returns>
		protected override async Task ExecuteAsync(CancellationToken stoppingToken)
		{
			_logger.LogInformation("BEFORE CREATES CHannels --------------------------.");

			_logger.LogInformation("AFTER CREATES CHannels --------------------------.");
		}

		/// <summary>
		/// Stop connection
		/// </summary>
		/// <param name="cancellationToken"></param>
		/// <returns></returns>
		public override async Task StopAsync(CancellationToken cancellationToken)
		{
			System.Console.WriteLine("STOP ASYNC ------------------------");
			_logger.LogInformation("RabbitMQ connection is closed.");
		}

	}
}

[michaelklishin]

I will convert this issue to a GitHub discussion. Currently GitHub will automatically close and lock the issue even though your question will be transferred and responded to elsewhere. This is to let you know that we do not intend to ignore this but this is how the current GitHub conversion mechanism makes it seem for the users :(

[michaelklishin]

This client cannot possibly affect connections from mosquitto_pub. RabbitMQ log locations are documented. You haven't mentioned how you provision RabbitMQ on Kubernetes
but nodes on Kubernetes usually log to standard output and error streams and can be inspected using kubectl logs.

We highly recommend using our own Kubernetes cluster Operator instead of DIY solutions.

[ninomllr]

Hm I assume that this is right that the rabbitmq-dotnet-client can not affect mosquitto_pub, but for some reason there is a problem when the first one runs in Kubernetes that affects mosquitto_pub.

As for logs, yes I have the stdout log from RabbitMQ, all successfull calls from mosquitto_pub end up as:

2021-05-04 14:48:09.068 [info] <0.17993.0> MQTT vhost picked using plugin configuration or default
2021-05-04 14:48:09.078 [info] <0.17993.0> accepting MQTT connection <0.17993.0> (172.25.0.244:49788 -> 10.1.4.18:1883, client id: mosq-DVNK3TJ84WoVPSlIRk)
2021-05-04 14:48:09.109 [info] <0.17993.0> closing MQTT connection <0.17993.0> (172.25.0.244:49788 -> 10.1.4.18:1883)
2021-05-04 14:48:09.110 [info] <0.18024.0> Closing all channels from connection '172.25.0.244:49788 -> 10.1.4.18:1883' because it has been closed

Which looks fine to me. What I noticed is that when I have the Error: Connection refused error, there is no log entry at all, which leads me to believe that the message request doesn't even reach the pod that runs RabbitMQ.

I also found this error message which seems to come from .NET AMQP connection, but it doesn't have a correlation between when the message happened and when mosquitto_pub has issues with the connection:

2021-05-04 14:46:45.967 [info] <0.17912.0> supervisor: {<0.17912.0>,amqp_channel_sup_sup}, errorContext: shutdown_error, reason: shutdown, offender: [{nb_children,1},{id,channel_sup},{mfargs,{amqp_channel_sup,start_link,[direct,<0.17909.0>,<<"172.25.0.244:48542 -> 10.1.4.18:1883">>]}},{restart_type,temporary},{shutdown,infinity},{child_type,supervisor}]
2021-05-04 14:46:45.968 [error] <0.17912.0> Supervisor {<0.17912.0>,amqp_channel_sup_sup} had child channel_sup started with amqp_channel_sup:start_link(direct, <0.17909.0>, <<"172.25.0.244:48542 -> 10.1.4.18:1883">>) at undefined exit with reason shutdown in context shutdown_error

[ninomllr]

After debugging some more I found that even with telnet I get Connection refused from time to time: telnet 172.25.0.244 32149 so I am starting to think that probably there is an issue with my Kubernetes cluster and not so much with RabbitMQ or RabbitMQ.NET. It's just super confusing that it only happens when the worker is running in that cluster, but otherwise no issue.

[ninomllr]

I finally figured it out, and it was our deployment yaml file of the worker that was wrong. So this request was definitely placed in the wrong repo, sorry for that!