Error: Pipelining of requests forbidden when getting metrics

[Socolin]

Hello,

When a function is running at the same time as the metric collector I get this exception: Pipelining of requests forbidden of type System.NotSupportedException at RabbitMQ.Client.Impl.RpcContinuationQueue.Enqueue

The problem is that the same channel is used in different thread, this is not supported by RabbitMQ librar. Iit should use another IModel to do that. https://github.com/Azure/azure-functions-rabbitmq-extension/blob/09c079a229c5693a959eb3e6680645e0099469a2/src/Trigger/RabbitMQListener.cs#L187-L197

[
    {
        "method": "RabbitMQ.Client.Impl.RpcContinuationQueue.Enqueue",
        "level": 0,
        "assembly": "RabbitMQ.Client, Version=6.0.0.0, Culture=neutral, PublicKeyToken=89e7d7c5feba84ce",
        "line": 0
    },
    {
        "method": "RabbitMQ.Client.Impl.ModelBase.Enqueue",
        "level": 1,
        "assembly": "RabbitMQ.Client, Version=6.0.0.0, Culture=neutral, PublicKeyToken=89e7d7c5feba84ce",
        "line": 0
    },
    {
        "method": "RabbitMQ.Client.Impl.ModelBase.QueueDeclare",
        "level": 2,
        "assembly": "RabbitMQ.Client, Version=6.0.0.0, Culture=neutral, PublicKeyToken=89e7d7c5feba84ce",
        "line": 0
    },
    {
        "method": "RabbitMQ.Client.Impl.AutorecoveringModel.QueueDeclarePassive",
        "level": 3,
        "assembly": "RabbitMQ.Client, Version=6.0.0.0, Culture=neutral, PublicKeyToken=89e7d7c5feba84ce",
        "line": 0
    },
    {
        "method": "Microsoft.Azure.WebJobs.Extensions.RabbitMQ.RabbitMQListener.GetMetricsAsync",
        "level": 4,
        "assembly": "WebJobs.Extensions.RabbitMQ, Version=2021.2.16.1, Culture=neutral, PublicKeyToken=null"
    },
    {
        "method": "Microsoft.Azure.WebJobs.Extensions.RabbitMQ.RabbitMQListener+<Microsoft-Azure-WebJobs-Host-Scale-IScaleMonitor-GetMetricsAsync>d__22.MoveNext",
        "level": 5,
        "assembly": "WebJobs.Extensions.RabbitMQ, Version=2021.2.16.1, Culture=neutral, PublicKeyToken=null",
        "line": 0
    },
    {
        "method": "System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw",
        "level": 6,
        "assembly": "System.Private.CoreLib, Version=4.0.0.0, Culture=neutral, PublicKeyToken=7cec85d7bea7798e",
        "line": 0
    },
    {
        "method": "System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess",
        "level": 7,
        "assembly": "System.Private.CoreLib, Version=4.0.0.0, Culture=neutral, PublicKeyToken=7cec85d7bea7798e",
        "line": 0
    },
    {
        "method": "System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification",
        "level": 8,
        "assembly": "System.Private.CoreLib, Version=4.0.0.0, Culture=neutral, PublicKeyToken=7cec85d7bea7798e",
        "line": 0
    },
    {
        "method": "System.Runtime.CompilerServices.TaskAwaiter`1.GetResult",
        "level": 9,
        "assembly": "System.Private.CoreLib, Version=4.0.0.0, Culture=neutral, PublicKeyToken=7cec85d7bea7798e",
        "line": 0
    },
    {
        "method": "Microsoft.Azure.WebJobs.Script.Scale.FunctionsScaleMonitorService+<TakeMetricsSamplesAsync>d__13.MoveNext",
        "level": 10,
        "assembly": "Microsoft.Azure.WebJobs.Script, Version=3.0.0.0, Culture=neutral, PublicKeyToken=null",
        "line": 95,
        "fileName": "D:\\a\\1\\s\\src\\WebJobs.Script\\Scale\\FunctionsScaleMonitorService.cs"
    }
]

[yojagad]

Can you elaborate on this a little bit more ? Do you get the IModel object(_rabbitMqModel) from a different function using RabbitMQ binding and use the same in GetMetricsAsync() ?

[Socolin]

I don't gent anything, the current code reuse the same IModel for evrything, for getting the metrics, for consuming and for publishing. It should use one IModel / thread and even use 2 connections (one for publish, one for consumemer)

https://www.cloudamqp.com/blog/2017-12-29-part1-rabbitmq-best-practice.html#don%E2%80%99t-share-channels-between-threads

[yojagad]

Thanks for clarifying that. Aside from our current implementation of IScaleMonitor and GetMetricsAsync(), can I get more info on what your function actually does as well, including the signature, etc ? This will help me narrow down on a plausible solution and also run some scenarios on our end.

[ghost]

This issue has been automatically marked as stale because it has been marked as requiring author feedback but has not had any activity for 4 days. It will be closed if no further activity occurs within 3 days of this comment.

[Socolin]

I'm using 2 trigger function, one of them with a dead letter queue