Add count aggregation function for MAL

[peachisai]

• If this pull request closes/resolves/fixes an existing issue, replace the issue number. Closes #.
• Update the CHANGES log.

This is my scenario
In rocketmq different consumer group has a different consumer offset

rocketmq_consumer_offset{cluster="DefaultCluster",broker="broker-a",topic="topic1",group="group3",} 347.0
rocketmq_consumer_offset{cluster="DefaultCluster",broker="broker-a",topic="topic1",group="group2",} 601.0
rocketmq_consumer_offset{cluster="DefaultCluster",broker="broker-a",topic="topic1",group="group1",} 694.0

  - name: consumer_group_count
    exp: rocketmq_consumer_offset.sum(['cluster','topic','group']).count(['group'])

Then I can get the consumer group number to caculate the blacklog messages

[wu-sheng]

Please provide UTs for the new functions.

[wankai123]

  - name: consumer_group_count
    exp: rocketmq_consumer_offset.sum(['cluster','topic','group']).count(['group'])

I think the result of your expression is:

rocketmq_consumer_offset{group="group3"} 1
rocketmq_consumer_offset{group="group2"} 1
rocketmq_consumer_offset{group="group1"} 1

Is that expected?

please refer to the other UT test cases and add yours:
https://github.com/apache/skywalking/blob/master/oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/AggregationTest.java

[peachisai]

  - name: consumer_group_count
    exp: rocketmq_consumer_offset.sum(['cluster','topic','group']).count(['group'])

I think the result of your expression is:

rocketmq_consumer_offset{group="group3"} 1
rocketmq_consumer_offset{group="group2"} 1
rocketmq_consumer_offset{group="group1"} 1

Is that expected?

please refer to the other UT test cases and add yours: https://github.com/apache/skywalking/blob/master/oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/AggregationTest.java

Yes, After the sum aggregation. I will serve UT tests up later

[wankai123]

Yes, After the sum aggregation. I will serve UT tests up later

sum is used to aggregate the value of the metrics, I think is unnecessary to use it here.

[peachisai]

rocketmq_consumer_offset{cluster="DefaultCluster",broker="broker-a",topic="topic1",group="group3",} 347.0

rocketmq_consumer_offset{cluster="DefaultCluster",broker="broker-a",topic="topic1",group="group3",} 347.0
rocketmq_consumer_offset{cluster="DefaultCluster",broker="broker-b",topic="topic1",group="group3",} 347.0

I need to pre-aggregation on the ['cluster','topic','group'] dimension

[wankai123]

I need to pre-aggregation on the ['cluster','topic','group'] dimension

you should aggregate the dimension items in func count directly. Please refer other aggregate func max/min/avg, we don't need aggregate twice.

[wankai123]

From the rocketMQ official exporter doc I found these metrics samples:

# HELP rocketmq_producer_offset TopicOffset
# TYPE rocketmq_producer_offset counter
rocketmq_producer_offset{cluster="MQCluster",broker="broker-b",topic="DEV_TID_tfq",} 1878633.0
rocketmq_producer_offset{cluster="MQCluster",broker="broker-a",topic="DEV_TID_tfq",} 3843787.0
rocketmq_producer_offset{cluster="MQCluster",broker="broker-a",topic="DEV_TID_topic_tfq",} 2798195.0
rocketmq_producer_offset{cluster="MQCluster",broker="broker-b",topic="DEV_TID_topic_tfq",} 145

# HELP rocketmq_consumer_offset GroupOffset
# TYPE rocketmq_consumer_offset counter
rocketmq_consumer_offset{cluster="MQCluster",broker="broker-b",topic="DEV_TID_topic_tfq",group="DEV_CID_consumer_cfq",} 1462030.0
rocketmq_consumer_offset{cluster="MQCluster",broker="broker-a",topic="DEV_TID_tfq",group="DEV_CID_cfq",} 3843787.0
rocketmq_consumer_offset{cluster="MQCluster",broker="broker-a",topic="DEV_TID_topic_tfq",group="DEV_CID_consumer_cfq",} 2800569.0
rocketmq_consumer_offset{cluster="MQCluster",broker="broker-b",topic="DEV_TID_tfq",group="DEV_CID_cfq",} 1878633.0

Could you give more details on how to calculate the backlog messages?

[peachisai]

From the rocketMQ official exporter doc I found these metrics samples:

# HELP rocketmq_producer_offset TopicOffset
# TYPE rocketmq_producer_offset counter
rocketmq_producer_offset{cluster="MQCluster",broker="broker-b",topic="DEV_TID_tfq",} 1878633.0
rocketmq_producer_offset{cluster="MQCluster",broker="broker-a",topic="DEV_TID_tfq",} 3843787.0
rocketmq_producer_offset{cluster="MQCluster",broker="broker-a",topic="DEV_TID_topic_tfq",} 2798195.0
rocketmq_producer_offset{cluster="MQCluster",broker="broker-b",topic="DEV_TID_topic_tfq",} 145

# HELP rocketmq_consumer_offset GroupOffset
# TYPE rocketmq_consumer_offset counter
rocketmq_consumer_offset{cluster="MQCluster",broker="broker-b",topic="DEV_TID_topic_tfq",group="DEV_CID_consumer_cfq",} 1462030.0
rocketmq_consumer_offset{cluster="MQCluster",broker="broker-a",topic="DEV_TID_tfq",group="DEV_CID_cfq",} 3843787.0
rocketmq_consumer_offset{cluster="MQCluster",broker="broker-a",topic="DEV_TID_topic_tfq",group="DEV_CID_consumer_cfq",} 2800569.0
rocketmq_consumer_offset{cluster="MQCluster",broker="broker-b",topic="DEV_TID_tfq",group="DEV_CID_cfq",} 1878633.0

Could you give more details on how to calculate the backlog messages?

  - name: producer_offset
    exp: rocketmq_producer_offset.sum(['cluster','topic']).downsampling(MAX)

  - name: consumer_group_count
    exp: rocketmq_consumer_offset.count(['cluster','topic','group'])

  - name: consumer_offset
    exp: rocketmq_consumer_offset.sum(['cluster','topic']).downsampling(MAX)

producer_offset*consumer_group_count-consumer_offset

[wu-sheng]

Could you explain (producer_offset*consumer_group_count-consumer_offset)?
Producer is somehow related to group but consuming doesn't?

[peachisai]

Could you explain (producer_offset*consumer_group_count-consumer_offset)? Producer is somehow related to group but consuming doesn't?

This is original metrics

rocketmq_producer_offset{cluster="DefaultCluster",broker="broker-a",topic="topic1",} 2738.0
rocketmq_consumer_offset{cluster="DefaultCluster",broker="broker-a",topic="topic1",group="group3",} 2250.0
rocketmq_consumer_offset{cluster="DefaultCluster",broker="broker-a",topic="topic1",group="group2",} 701.0
rocketmq_consumer_offset{cluster="DefaultCluster",broker="broker-a",topic="topic1",group="group1",} 2738.0

The concept of a consumer group is different from that of a producer group. Each consumer group has its own independent offset, and the group of producer is no longer in use in 5.x version.

[wu-sheng]

I was talking with Kai, I am wondering whether number of the remaining messages should be good to measure by group? Rather all for all grouping?
If something goes wrong, operation team should care more about the remaining per group.

[wankai123]

The count function implementation looks good, I think we can reserve it. What we are concerned about is the calculation of producer_offset*consumer_group_count-consumer_offse

[peachisai]

I was talking with Kai, I am wondering whether number of the remaining messages should be good to measure by group? Rather all for all grouping? If something goes wrong, operation team should care more about the remaining per group.

Do you mean we should caculate the remaining msg in each group?

The topic dimenssion provides each group offset

  - name: consumer_group_offset
    exp: rocketmq_consumer_offset.sum(['cluster','topic','group']).downsampling(MAX)

Or should I use this?

producer_offset-consumer_group_offset

[wu-sheng]

Do you mean we should caculate the remaining msg in each group?

For MQ case, yes, that should be more useful.

producer_offset-consumer_group_offset

Does this work? I think minus only works when labels are same.

[wankai123]

The count function LGTM. Although it's not suitable for offset calculation.
We also can use it to calculate the number of topics or groups. etc.

[peachisai]

Do you mean we should caculate the remaining msg in each group?

For MQ case, yes, that should be more useful.

producer_offset-consumer_group_offset

Does this work? I think minus only works when labels are same.

I am not on my computer now, I will test it later, maybe MQE could works?

[wankai123]

I am not on my computer now, I will test it later, maybe MQE could works?

MQE also requires they have the same labels or one side without any labels