When JMX MBean attribute values get collected into Activity snapshots (one activity per sampling iteration and one snapshot per MBean), it is possible to aggregate attribute values. Aggregations of individual MBean attributes can be performed into an already Activity-contained snapshot (append some additional attributes for MBean) or make a new snapshot containing these aggregated attribute values (create a kind of new virtual MBean).
Aggregations configuration is defined using JSON format. Default aggregations configuration file path is config/aggregations.json. System property to set custom aggregations configuration file path to use is tnt4j.stream.jmx.aggregations.config.
A high-level aggregations configuration looks like this:
[
{
aggregator1 config
},
{
aggregator2 config
},
...
{
aggregatorN config
}
]In general, it is a list of aggregator configurations. The aggregations manager is able to load that list and initiate an aggregator instance based on the aggregator type class defined in the configuration. The aggregator itself is responsible for loading the rest of the configuration over interface method com.jkoolcloud.tnt4j.stream.jmx.aggregations.ActivityAggregator.configure(Map<String, ?> cfg) throws IllegalArgumentException implementation.
Available aggregator implementations:
com.jkoolcloud.tnt4j.stream.jmx.aggregations.SnapshotAggregator- picks individual MBean attribute values from Activity contained snapshots and puts them into existing or new snapshot.
Aggregator configuration schema is:
aggregatorId- aggregator identifier, to identify the aggregator instance in the logs, e.g. when it fails to load the configuration or perform some aggregation actions. Optional, if not defined, then made of class name and sequence index.type- aggregator implementation class to use. Required.enabled- flag indicating whether the aggregator configuration shall be loaded and used. Optional, default value -true.ignoreEmpty- flag indicating whether to ignore aggregator-created empty snapshots. Optional, default value -false.snapshots- aggregator implementation-specific configuration array defining how values are aggregated into snapshots. Required:name- aggregation snapshot name RegEx string. It can be Activity contained snapshot name (to append properties) or any other (to create a new snapshot and add it to the Activity). To define Activity contained snapshot name use ObjectName notation syntax, see attributebeanIdas a sample. Required.category- aggregation snapshot category RegEx string. Optional, if aggregation snapshotnamehas it defined like thiscategory:beanIdor sets default valuejmx.aggregatedin any other case if omitted.enabled- flag indicating if snapshot aggregation shall be used. Optional, default value -true.properties- list of snapshot properties to aggregate and store MBean attribute values. Required:beanId- property-bound bean identifier, it can have variable expression likevarName=?or object name pattern. To definebeanIduse ObjectName notation syntax. When omitted, the aggregation's target snapshot is used to resolve property values. Optional.attribute- property bound bean attribute name to get value, it can have variable expression like${attrName1}-${attrName2}. Required.
NOTE: when all variables resolve to anullvalue, the aggregated value also becomesnull.default- defines default value(s) mapping forattributedefined variables or unresolved aggregated value in general. Optional:can define a default aggregated value when all variables resolve to
nullvalues:"default": "-"
can define mapping for individual variables and/or unresolved aggregated value:
"attribute": "${attrName1}-${attrName2}-${attrName3}", "default": { "attrName1": 0, "attrName2": "NA", "": "-", ">>": "UNRESOLVED" }- key
""defines fallback default value to be applied for any unmapped variable. - key
">>"defines the default aggregated value (when the attribute-referenced value is unresolved or all variables resolve tonullvalues).
- key
name- property (snapshot property) name, it can have variable expression like${varName}. Required.where- set of property-used variable definitions. Optional:[variable name]- variable name[variable values]- variable values string delimited (if variable has multiple values) by|symbol
transparent- flag indicating whether this property definition produces a transient snapshot property. Optional, default value -false.
A sample aggregator configuration may look like this:
{
"aggregatorId": "MLSnapshotsAggregator",
"type": "com.jkoolcloud.tnt4j.stream.jmx.aggregations.SnapshotAggregator",
"enabled": true,
"snapshots": [
{
"name": "kafka.aggregated:KafkaStatsML",
"enabled": true,
"properties": [
{
"beanId": "kafka.server:name=UnderReplicatedPartitions,type=ReplicaManager",
"attribute": "Value",
"name": "UnderReplicatedPartitions"
},
{
"beanId": "kafka.server:name=IsrShrinksPerSec,type=ReplicaManager",
"attribute": "MeanRate",
"name": "IsrShrinksPerSec"
},
{
"beanId": "kafka.server:name=IsrExpandsPerSec,type=ReplicaManager",
"attribute": "MeanRate",
"name": "IsrExpandsPerSec"
},
{
"beanId": "kafka.controller:name=ActiveControllerCount,type=KafkaController",
"attribute": "Value",
"name": "ActiveControllerCount"
},
{
"beanId": "kafka.controller:name=OfflinePartitionsCount,type=KafkaController",
"attribute": "Value",
"name": "OfflinePartitionsCount"
},
{
"beanId": "kafka.controller:name=LeaderElectionRateAndTimeMs,type=ControllerStats",
"attribute": "Mean",
"name": "LeaderElectionRateAndTimeMs"
},
{
"beanId": "kafka.controller:name=UncleanLeaderElectionsPerSec,type=ControllerStats",
"attribute": "MeanRate",
"name": "UncleanLeaderElectionsPerSec"
},
{
"beanId": "kafka.network:name=TotalTimeMs,request=?,type=RequestMetrics",
"where": {
"request": "Produce|FetchConsumer|FetchFollower"
},
"attribute": "Mean",
"name": "${request}-TotalTimeMs"
},
{
"beanId": "kafka.server:delayedOperation=?,name=PurgatorySize,type=DelayedOperationPurgatory",
"where": {
"delayedOperation": "Produce|Fetch"
},
"attribute": "Value",
"name": "${delayedOperation}-PurgatorySize"
},
{
"beanId": "kafka.server:name=BytesInPerSec,type=BrokerTopicMetrics",
"attribute": "MeanRate",
"name": "BytesInPerSec"
},
{
"beanId": "kafka.server:name=BytesOutPerSec,type=BrokerTopicMetrics",
"attribute": "MeanRate",
"name": "BytesOutPerSec"
},
{
"beanId": "kafka.network:name=RequestsPerSec,request=?,type=RequestMetrics,*",
"where": {
"request": "Produce|FetchConsumer|FetchFollower"
},
"attribute": "Count",
"name": "${request}-RequestsPerSec"
}
]
},
{
"name": ".*",
"category": ".*",
"enabled": true,
"properties": [
{
"beanId": "kafka.server:type=KafkaServer,name=ClusterId",
"attribute": "Value",
"name": "kafkaClusterId",
"transparent": true
},
{
"beanId": "kafka.server:type=app-info,id=*",
"attribute": "id",
"name": "kafkaBrokerId",
"transparent": true
},
{
"attribute": "${kafkaBrokerId}-${kafkaClusterId}",
"name": "kafkaNode"
}
]
}
]
}