Messages going through federation are not mentioned in statistics or metrics

[mokrinsky]

Hello there!

We have either a problem or misunderstanding with federation plugin. Lemme explain the situation:
I've made two nodes with identical configuration (8gb, 8core), one node has only producers, second one has only consumers. And second node collects messages from the first one over federation. Both nodes are running v3.11.14, but I ran these tests on 3.12.6 and results are the same. I tried running performance testing over these rabbits, here are the logs:

Producer:

podman run --network host -it --rm docker.io/pivotalrabbitmq/perf-test:latest --producers 1 --consumers 0 --uri amqp://localhost/sync --exchange sync.DirectExch --routing-key rk -s 500 --rate 6000 --use-millis
id: test-114035-043, starting producer #0
id: test-114035-043, starting producer #0, channel #0
id: test-114035-043, time 1.001 s, sent: 2079 msg/s
^Ctest stopped (Producer thread interrupted)
id: test-114035-043, sending rate avg: 6073 msg/s
id: test-114035-043, receiving rate avg: 0 msg/s

Consumer:

podman run --network host -it --rm docker.io/pivotalrabbitmq/perf-test:latest --producers 0 --consumers 1 --uri amqp://localhost/sync --exchange sync.DirectExch -u TestQueue -f durable --routing-key rk --use-millis
id: test-114034-228, starting consumer #0
id: test-114034-228, starting consumer #0, channel #0
id: test-114034-228, time 2.000 s, received: 75 msg/s, min/median/75th/95th/99th consumer latency: 43/196/204/210/263 ms
id: test-114034-228, time 197.000 s, received: 405 msg/s, min/median/75th/95th/99th consumer latency: 82140/82159/82172/82181/82182 ms
^Ctest stopped
id: test-114034-228, sending rate avg: 0 msg/s
id: test-114034-228, receiving rate avg: 3129 msg/s
id: test-114034-228, consumer latency min/median/75th/95th/99th 369/72679/78217/81557/82118 ms

So basically consumer reads messages slower than they are published which is not a problem by itself cause there could be several bottle necks.
But from my logic if message is published but not delivered - it should be visible somewhere. However, these messages are basically gone. Screenshots of management pages of both nodes:
Producer: https://imgur.com/a/FoW2LWY
Consumer: https://imgur.com/a/nV2zzU8

And finally, the problem I end up having is growing memory on the consumer node, which is kinda visible on the second screenshot, while there is no visible reason for it to grow. On the screenshot it's about 800mb, but we have it in production and we frequently go out of memory with consumer rabbits with zero messages in unacked state.

So, could you please help me, is there any way to monitor process of the federation? I mean it's obvious memory growth is exactly these messages, but from any kind of monitoring or metrics these messages are basically invisible. From the side view it looks like unreasonable growth of consumed memory - publish and consume rates are equal, no messages in queues, but publish rate on publish node is twice as high as consume rate on consumer node, and looks like messages disappear.

And a side question. I tested both of these nodes locally and they both are able to handle publish&consume rate about 50k/s without federation. Why does federation works THAT slow?

[lukebakken]

Please refrain from pasting large amounts of text into a discussion or comment on GitHub. I have edited your post to use details and summary. Another option is to save output to a file and attach the file. In addition, you can upload those images directly to GitHub. There is no need to use a third-party service.

And a side question. I tested both of these nodes locally and they both are able to handle publish&consume rate about 50k/s without federation. Why does federation works THAT slow?

Change the prefetch-count value for your Federation link to 1000 and re-test (https://www.rabbitmq.com/federation-reference.html#upstreams)

Please export the definitions from both nodes so we know how you're setting up Federation. Pretty-print the json then attach the files to your response. Thank you.

[mokrinsky]

Removed all the sensitive information, obviously.
There's no need to re-test cause I already had prefetch at 1000.

Producer

{
  "users": [
    {
      "name": "federator",
      "tags": [],
      "limits": {}
    },
    {
      "name": "guest",
      "tags": [
        "administrator"
      ],
      "limits": {}
    }
  ],
  "vhosts": [
    {
      "name": "sync"
    },
    {
      "name": "/"
    }
  ],
  "permissions": [
    {
      "user": "guest",
      "vhost": "sync",
      "configure": ".*",
      "write": ".*",
      "read": ".*"
    },
    {
      "user": "federator",
      "vhost": "sync",
      "configure": ".*",
      "write": ".*",
      "read": ".*"
    },
    {
      "user": "guest",
      "vhost": "/",
      "configure": ".*",
      "write": ".*",
      "read": ".*"
    }
  ],
  "topic_permissions": [],
  "parameters": [],
  "global_parameters": [
    {
      "name": "internal_cluster_id",
      "value": "rabbitmq-cluster-id-9GrWlwieDj67T7j00ezm4A"
    }
  ],
  "policies": [
    {
      "vhost": "/",
      "name": "limit_logging_queue",
      "pattern": "internal_logging",
      "apply-to": "queues",
      "definition": {
        "max-length": 100
      },
      "priority": 0
    },
    {
      "vhost": "sync",
      "name": "SyncDefault",
      "pattern": "^(?:(?:command\\.)?sync\\.)(.*)default$",
      "apply-to": "exchanges",
      "definition": {
        "federation-upstream-set": "all"
      },
      "priority": 95
    },
    {
      "vhost": "sync",
      "name": "sync",
      "pattern": "^((command\\.)?sync\\.)(.*)(?<!ault)$",
      "apply-to": "exchanges",
      "definition": {
        "federation-upstream-set": "all"
      },
      "priority": 100
    }
  ],
  "queues": [
    {
      "name": "federation: sync.DirectExch -> [email protected]",
      "vhost": "sync",
      "durable": true,
      "auto_delete": false,
      "arguments": {
        "x-internal-purpose": "federation"
      }
    },
    {
      "name": "internal_logging",
      "vhost": "/",
      "durable": true,
      "auto_delete": false,
      "arguments": {}
    }
  ],
  "exchanges": [
    {
      "name": "federation: sync.DirectExch -> [email protected] B",
      "vhost": "sync",
      "type": "x-federation-upstream",
      "durable": true,
      "auto_delete": true,
      "internal": true,
      "arguments": {
        "x-downstream-name": "[email protected]",
        "x-downstream-vhost": "sync",
        "x-internal-purpose": "federation",
        "x-max-hops": 1
      }
    },
    {
      "name": "sync.DirectExch",
      "vhost": "sync",
      "type": "direct",
      "durable": true,
      "auto_delete": false,
      "internal": false,
      "arguments": {}
    },
    {
      "name": "localPerf",
      "vhost": "sync",
      "type": "direct",
      "durable": true,
      "auto_delete": false,
      "internal": false,
      "arguments": {}
    }
  ],
  "bindings": [
    {
      "source": "federation: sync.DirectExch -> [email protected] B",
      "vhost": "sync",
      "destination": "federation: sync.DirectExch -> [email protected]",
      "destination_type": "queue",
      "routing_key": "",
      "arguments": {}
    }
  ]
}

Consumer

{
  "users": [
    {
      "name": "federator",
      "tags": [],
      "limits": {}
    },
    {
      "name": "guest",
      "tags": [
        "administrator"
      ],
      "limits": {}
    }
  ],
  "vhosts": [
    {
      "name": "sync"
    },
    {
      "name": "/"
    }
  ],
  "permissions": [
    {
      "user": "guest",
      "vhost": "sync",
      "configure": ".*",
      "write": ".*",
      "read": ".*"
    },
    {
      "user": "federator",
      "vhost": "sync",
      "configure": ".*",
      "write": ".*",
      "read": ".*"
    },
    {
      "user": "guest",
      "vhost": "/",
      "configure": ".*",
      "write": ".*",
      "read": ".*"
    }
  ],
  "topic_permissions": [],
  "parameters": [
    {
      "value": {
        "ack-mode": "no-ack",
        "prefetch-count": 1000,
        "trust-user-id": false,
        "uri": "amqp://federator:[email protected]"
      },
      "vhost": "sync",
      "component": "federation-upstream",
      "name": "facility1"
    }
  ],
  "global_parameters": [
    {
      "name": "internal_cluster_id",
      "value": "rabbitmq-cluster-id-mjq4YMpNpCNn-WLSq0mQLQ"
    }
  ],
  "policies": [
    {
      "vhost": "/",
      "name": "limit_logging_queue",
      "pattern": "internal_logging",
      "apply-to": "queues",
      "definition": {
        "max-length": 100
      },
      "priority": 0
    },
    {
      "vhost": "sync",
      "name": "SyncDefault",
      "pattern": "^(?:(?:command\\.)?sync\\.)(.*)default$",
      "apply-to": "exchanges",
      "definition": {
        "federation-upstream-set": "all"
      },
      "priority": 95
    },
    {
      "vhost": "sync",
      "name": "sync",
      "pattern": "^((command\\.)?sync\\.)(.*)(?<!ault)$",
      "apply-to": "exchanges",
      "definition": {
        "federation-upstream-set": "all"
      },
      "priority": 100
    }
  ],
  "queues": [
    {
      "name": "internal_logging",
      "vhost": "/",
      "durable": true,
      "auto_delete": false,
      "arguments": {}
    }
  ],
  "exchanges": [
    {
      "name": "sync.DirectExch",
      "vhost": "sync",
      "type": "direct",
      "durable": true,
      "auto_delete": false,
      "internal": false,
      "arguments": {}
    },
    {
      "name": "localPerf",
      "vhost": "sync",
      "type": "direct",
      "durable": true,
      "auto_delete": false,
      "internal": false,
      "arguments": {}
    }
  ],
  "bindings": []
}

[lukebakken]

But from my logic if message is published but not delivered - it should be visible somewhere

This is what publisher confirmations are for - https://www.rabbitmq.com/tutorials/tutorial-seven-java.html

If your publisher is not using them, or using the mandatory flag, it will not know if a message was routed to a queue.

[mokrinsky]

And, if I get it correctly, it can cause "memory leaks" cause rmq doesn't know if message is in queue, right?

[lukebakken]

No, this does not explain the memory issue you report.

I will return to this issue when time allows.