Add consumer count to stream queue metrics

[acogoluegnes]

This commit adds the "consumers" metrics to stream queues (consumer count).
It is computed by counting the element of the consumer_created ETS table
for the given stream queue and for each member of the Osiris cluster.

Fixes #4622

[acogoluegnes]

A/C:

Get the code and start a cluster:

cd /tmp
git clone [email protected]:rabbitmq/rabbitmq-server.git rabbitmq-server-4622
cd rabbitmq-server-4622
git checkout rabbitmq-server-4622-stream-queue-consumer-count
make start-cluster NODES=3 PLUGINS='rabbitmq_stream rabbitmq_management'

Download the performance tool and start it:

wget https://github.com/rabbitmq/rabbitmq-java-tools-binaries-dev/releases/download/v-stream-perf-test-latest/stream-perf-test-latest.jar
java -jar stream-perf-test-latest.jar --rate 100 --producers 2 --consumers 9 --stream-count 2

The performance tool creates 2 streams, publishes to them, and balances 9 consumers across them.

Go to the queue list page and make sure the Consumers column shows up (+/- characters on the right of the table, check Consumer count). The 2 streams should be listed, with 5 consumers for the first one, and 4 consumers for the second one.

List the queues with the REST API, the consumer count should be the same as for the management UI:

curl -u guest:guest http://localhost:15672/api/queues | jq .

[
  {
    "arguments": {
      "x-max-length-bytes": 20000000000,
      "x-queue-leader-locator": "least-leaders",
      "x-queue-type": "stream",
      "x-stream-max-segment-size-bytes": 500000000
    },
    "auto_delete": false,
    "consumers": 5,
    "durable": true,
    "effective_policy_definition": {},
    "exclusive": false,
    "members": [
      "rabbit-1@acogoluegnes-nuc",
      "rabbit-3@acogoluegnes-nuc",
      "rabbit-2@acogoluegnes-nuc"
    ],
    "memory": 9248,
    "messages": 56879,
    "messages_details": {
      "rate": 100.4
    },
    "messages_ready": 56879,
    "messages_ready_details": {
      "rate": 100.4
    },
    "messages_unacknowledged": 0,
    "messages_unacknowledged_details": {
      "rate": 0
    },
    "name": "stream-1",
    "node": "rabbit-1@acogoluegnes-nuc",
    "online": [
      "rabbit-3@acogoluegnes-nuc",
      "rabbit-2@acogoluegnes-nuc",
      "rabbit-1@acogoluegnes-nuc"
    ],
    "operator_policy": null,
    "policy": null,
    "readers": {
      "rabbit-1@acogoluegnes-nuc": 2,
      "rabbit-2@acogoluegnes-nuc": 3,
      "rabbit-3@acogoluegnes-nuc": 2
    },
    "reductions": 0,
    "reductions_details": {
      "rate": 0
    },
    "state": "running",
    "type": "stream",
    "vhost": "/"
  },
  {
    "arguments": {
      "x-max-length-bytes": 20000000000,
      "x-queue-leader-locator": "least-leaders",
      "x-queue-type": "stream",
      "x-stream-max-segment-size-bytes": 500000000
    },
    "auto_delete": false,
    "consumers": 4,
    "durable": true,
    "effective_policy_definition": {},
    "exclusive": false,
    "members": [
      "rabbit-2@acogoluegnes-nuc",
      "rabbit-3@acogoluegnes-nuc",
      "rabbit-1@acogoluegnes-nuc"
    ],
    "memory": 9280,
    "messages": 56787,
    "messages_details": {
      "rate": 100.4
    },
    "messages_ready": 56787,
    "messages_ready_details": {
      "rate": 100.4
    },
    "messages_unacknowledged": 0,
    "messages_unacknowledged_details": {
      "rate": 0
    },
    "name": "stream-2",
    "node": "rabbit-2@acogoluegnes-nuc",
    "online": [
      "rabbit-3@acogoluegnes-nuc",
      "rabbit-2@acogoluegnes-nuc",
      "rabbit-1@acogoluegnes-nuc"
    ],
    "operator_policy": null,
    "policy": null,
    "readers": {
      "rabbit-1@acogoluegnes-nuc": 2,
      "rabbit-2@acogoluegnes-nuc": 2,
      "rabbit-3@acogoluegnes-nuc": 2
    },
    "reductions": 0,
    "reductions_details": {
      "rate": 0
    },
    "state": "running",
    "type": "stream",
    "vhost": "/"
  }
]

Or just extract the consumers field values:

curl -u guest:guest http://localhost:15672/api/queues | jq '.[].consumers'
5
4

Stop the performance tool (Ctrl+C). Refresh the queue list page, the number of consumers should be 0 for both streams.

Check the REST API:

curl -u guest:guest http://localhost:15672/api/queues | jq .

[
  {
    "arguments": {
      "x-max-length-bytes": 20000000000,
      "x-queue-leader-locator": "least-leaders",
      "x-queue-type": "stream",
      "x-stream-max-segment-size-bytes": 500000000
    },
    "auto_delete": false,
    "consumers": 0,
    "durable": true,
    "effective_policy_definition": {},
    "exclusive": false,
    "members": [
      "rabbit-1@acogoluegnes-nuc",
      "rabbit-3@acogoluegnes-nuc",
      "rabbit-2@acogoluegnes-nuc"
    ],
    "memory": 9184,
    "messages": 96958,
    "messages_details": {
      "rate": 0
    },
    "messages_ready": 96958,
    "messages_ready_details": {
      "rate": 0
    },
    "messages_unacknowledged": 0,
    "messages_unacknowledged_details": {
      "rate": 0
    },
    "name": "stream-1",
    "node": "rabbit-1@acogoluegnes-nuc",
    "online": [
      "rabbit-3@acogoluegnes-nuc",
      "rabbit-2@acogoluegnes-nuc",
      "rabbit-1@acogoluegnes-nuc"
    ],
    "operator_policy": null,
    "policy": null,
    "readers": {
      "rabbit-1@acogoluegnes-nuc": 2,
      "rabbit-2@acogoluegnes-nuc": 3,
      "rabbit-3@acogoluegnes-nuc": 2
    },
    "reductions": 0,
    "reductions_details": {
      "rate": 0
    },
    "state": "running",
    "type": "stream",
    "vhost": "/"
  },
  {
    "arguments": {
      "x-max-length-bytes": 20000000000,
      "x-queue-leader-locator": "least-leaders",
      "x-queue-type": "stream",
      "x-stream-max-segment-size-bytes": 500000000
    },
    "auto_delete": false,
    "consumers": 0,
    "durable": true,
    "effective_policy_definition": {},
    "exclusive": false,
    "members": [
      "rabbit-2@acogoluegnes-nuc",
      "rabbit-3@acogoluegnes-nuc",
      "rabbit-1@acogoluegnes-nuc"
    ],
    "memory": 9216,
    "messages": 96957,
    "messages_details": {
      "rate": 0
    },
    "messages_ready": 96957,
    "messages_ready_details": {
      "rate": 0
    },
    "messages_unacknowledged": 0,
    "messages_unacknowledged_details": {
      "rate": 0
    },
    "name": "stream-2",
    "node": "rabbit-2@acogoluegnes-nuc",
    "online": [
      "rabbit-3@acogoluegnes-nuc",
      "rabbit-2@acogoluegnes-nuc",
      "rabbit-1@acogoluegnes-nuc"
    ],
    "operator_policy": null,
    "policy": null,
    "readers": {
      "rabbit-1@acogoluegnes-nuc": 2,
      "rabbit-2@acogoluegnes-nuc": 2,
      "rabbit-3@acogoluegnes-nuc": 2
    },
    "reductions": 0,
    "reductions_details": {
      "rate": 0
    },
    "state": "running",
    "type": "stream",
    "vhost": "/"
  }
]

curl -u guest:guest http://localhost:15672/api/queues | jq '.[].consumers'
0
0

Stop the cluster:

make stop-cluster NODES=3

Delete the sources for the branch:

cd /tmp
rm -rf /tmp/rabbitmq-server-4622