Merge pull request #947 from lcybo/master

修复group模式报错
This commit is contained in:
agapple
2018-09-17 19:51:12 +08:00
committed by GitHub
4 changed files with 338 additions and 226 deletions

View File

@@ -1,4 +1,34 @@
{ {
"__inputs": [
{
"name": "DS_PROMETHEUS",
"label": "prometheus",
"description": "",
"type": "datasource",
"pluginId": "prometheus",
"pluginName": "Prometheus"
}
],
"__requires": [
{
"type": "grafana",
"id": "grafana",
"name": "Grafana",
"version": "5.2.2"
},
{
"type": "panel",
"id": "graph",
"name": "Graph",
"version": "5.0.0"
},
{
"type": "datasource",
"id": "prometheus",
"name": "Prometheus",
"version": "5.0.0"
}
],
"annotations": { "annotations": {
"list": [ "list": [
{ {
@@ -15,8 +45,8 @@
"editable": true, "editable": true,
"gnetId": null, "gnetId": null,
"graphTooltip": 0, "graphTooltip": 0,
"id": 8, "id": null,
"iteration": 1534740332233, "iteration": 1536989235272,
"links": [], "links": [],
"panels": [ "panels": [
{ {
@@ -189,7 +219,7 @@
"steppedLine": false, "steppedLine": false,
"targets": [ "targets": [
{ {
"expr": "rate(canal_instance_received_binlog_bytes{destination=~\"$destination\"}[2m]) / 1024", "expr": "rate(canal_instance_received_binlog_bytes{destination=~\"$destination\", parser=\"0\"}[2m]) / 1024",
"format": "time_series", "format": "time_series",
"interval": "15s", "interval": "15s",
"intervalFactor": 2, "intervalFactor": 2,
@@ -203,6 +233,22 @@
"intervalFactor": 2, "intervalFactor": 2,
"legendFormat": "outbound", "legendFormat": "outbound",
"refId": "B" "refId": "B"
},
{
"expr": "rate(canal_instance_received_binlog_bytes{destination=~\"$destination\", parser=\"1\"}[2m]) / 1024",
"format": "time_series",
"interval": "15s",
"intervalFactor": 2,
"legendFormat": "inbound-1",
"refId": "C"
},
{
"expr": "rate(canal_instance_received_binlog_bytes{destination=~\"$destination\", parser=\"2\"}[2m]) / 1024",
"format": "time_series",
"interval": "15s",
"intervalFactor": 2,
"legendFormat": "inbound-2",
"refId": "D"
} }
], ],
"thresholds": [], "thresholds": [],
@@ -399,20 +445,36 @@
"steppedLine": false, "steppedLine": false,
"targets": [ "targets": [
{ {
"expr": "clamp_max(rate(canal_instance_sink_blocking_time{destination=~\"$destination\"}[2m]), 1000) / 10", "expr": "clamp_max(rate(canal_instance_publish_blocking_time{destination=~\"$destination\", parser=\"0\"}[2m]), 1000) / 10",
"format": "time_series", "format": "time_series",
"interval": "15s", "interval": "15s",
"intervalFactor": 2, "intervalFactor": 2,
"legendFormat": "dump", "legendFormat": "dump",
"refId": "A" "refId": "B"
}, },
{ {
"expr": "clamp_max(rate(canal_instance_publish_blocking_time{destination=~\"$destination\"}[2m]), 1000) / 10", "expr": "clamp_max(rate(canal_instance_sink_blocking_time{destination=~\"$destination\"}[2m]), 1000) / 10",
"format": "time_series", "format": "time_series",
"interval": "15s", "interval": "15s",
"intervalFactor": 2, "intervalFactor": 2,
"legendFormat": "sink", "legendFormat": "sink",
"refId": "B" "refId": "A"
},
{
"expr": "clamp_max(rate(canal_instance_publish_blocking_time{destination=~\"$destination\", parser=\"1\"}[2m]), 1000) / 10",
"format": "time_series",
"interval": "15s",
"intervalFactor": 2,
"legendFormat": "dump-1",
"refId": "C"
},
{
"expr": "clamp_max(rate(canal_instance_publish_blocking_time{destination=~\"$destination\", parser=\"2\"}[2m]), 1000) / 10",
"format": "time_series",
"interval": "15s",
"intervalFactor": 2,
"legendFormat": "dump-2",
"refId": "D"
} }
], ],
"thresholds": [], "thresholds": [],
@@ -465,7 +527,7 @@
}, },
"id": 32, "id": 32,
"panels": [], "panels": [],
"title": "Server", "title": "Throughput",
"type": "row" "type": "row"
}, },
{ {
@@ -660,180 +722,6 @@
"alignLevel": null "alignLevel": null
} }
}, },
{
"aliasColors": {},
"bars": false,
"dashLength": 10,
"dashes": false,
"datasource": "$datasource",
"description": "Canal instance ringbuffer内未释放的events数量。",
"fill": 1,
"gridPos": {
"h": 5,
"w": 6,
"x": 12,
"y": 7
},
"id": 20,
"legend": {
"avg": false,
"current": false,
"max": false,
"min": false,
"show": true,
"total": false,
"values": false
},
"lines": true,
"linewidth": 1,
"links": [],
"nullPointMode": "null",
"percentage": false,
"pointradius": 5,
"points": false,
"renderer": "flot",
"seriesOverrides": [],
"spaceLength": 10,
"stack": false,
"steppedLine": false,
"targets": [
{
"expr": "canal_instance_store_produce_seq{destination=~\"$destination\"} - canal_instance_store_consume_seq{destination=~\"$destination\"}",
"format": "time_series",
"interval": "15s",
"intervalFactor": 2,
"legendFormat": "events",
"refId": "A"
}
],
"thresholds": [],
"timeFrom": null,
"timeShift": null,
"title": "Store remain events",
"tooltip": {
"shared": true,
"sort": 0,
"value_type": "individual"
},
"type": "graph",
"xaxis": {
"buckets": null,
"mode": "time",
"name": null,
"show": true,
"values": []
},
"yaxes": [
{
"decimals": null,
"format": "none",
"label": "",
"logBase": 1,
"max": null,
"min": null,
"show": true
},
{
"decimals": null,
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
}
],
"yaxis": {
"align": false,
"alignLevel": null
}
},
{
"aliasColors": {},
"bars": false,
"dashLength": 10,
"dashes": false,
"datasource": "$datasource",
"description": "Canal instance ringbuffer 内未释放events占用内存。",
"fill": 1,
"gridPos": {
"h": 5,
"w": 6,
"x": 18,
"y": 7
},
"id": 22,
"legend": {
"avg": false,
"current": false,
"max": false,
"min": false,
"show": true,
"total": false,
"values": false
},
"lines": true,
"linewidth": 1,
"links": [],
"nullPointMode": "null",
"percentage": false,
"pointradius": 5,
"points": false,
"renderer": "flot",
"seriesOverrides": [],
"spaceLength": 10,
"stack": false,
"steppedLine": false,
"targets": [
{
"expr": "(canal_instance_store_produce_mem{destination=~\"$destination\"} - canal_instance_store_consume_mem{destination=~\"$destination\"}) / 1024",
"format": "time_series",
"interval": "15s",
"intervalFactor": 2,
"legendFormat": "memsize",
"refId": "A"
}
],
"thresholds": [],
"timeFrom": null,
"timeShift": null,
"title": "Store remain mem",
"tooltip": {
"shared": true,
"sort": 0,
"value_type": "individual"
},
"type": "graph",
"xaxis": {
"buckets": null,
"mode": "time",
"name": null,
"show": true,
"values": []
},
"yaxes": [
{
"format": "deckbytes",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
},
{
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
}
],
"yaxis": {
"align": false,
"alignLevel": null
}
},
{ {
"collapsed": false, "collapsed": false,
"gridPos": { "gridPos": {
@@ -938,7 +826,7 @@
"bars": false, "bars": false,
"dashLength": 10, "dashLength": 10,
"dashes": false, "dashes": false,
"datasource": "$datasource", "datasource": "${DS_PROMETHEUS}",
"description": "client 请求的GET与ACK包的QPS。", "description": "client 请求的GET与ACK包的QPS。",
"fill": 1, "fill": 1,
"gridPos": { "gridPos": {
@@ -1218,6 +1106,193 @@
"align": false, "align": false,
"alignLevel": null "alignLevel": null
} }
},
{
"collapsed": false,
"gridPos": {
"h": 1,
"w": 24,
"x": 0,
"y": 18
},
"id": 36,
"panels": [],
"title": "Store",
"type": "row"
},
{
"aliasColors": {},
"bars": false,
"dashLength": 10,
"dashes": false,
"datasource": "$datasource",
"description": "Canal instance ringbuffer内未释放的events数量。",
"fill": 1,
"gridPos": {
"h": 5,
"w": 6,
"x": 0,
"y": 19
},
"id": 20,
"legend": {
"avg": false,
"current": false,
"max": false,
"min": false,
"show": true,
"total": false,
"values": false
},
"lines": true,
"linewidth": 1,
"links": [],
"nullPointMode": "null",
"percentage": false,
"pointradius": 5,
"points": false,
"renderer": "flot",
"seriesOverrides": [],
"spaceLength": 10,
"stack": false,
"steppedLine": false,
"targets": [
{
"expr": "canal_instance_store_produce_seq{destination=~\"$destination\"} - canal_instance_store_consume_seq{destination=~\"$destination\"}",
"format": "time_series",
"interval": "15s",
"intervalFactor": 2,
"legendFormat": "events",
"refId": "A"
}
],
"thresholds": [],
"timeFrom": null,
"timeShift": null,
"title": "Store remain events",
"tooltip": {
"shared": true,
"sort": 0,
"value_type": "individual"
},
"type": "graph",
"xaxis": {
"buckets": null,
"mode": "time",
"name": null,
"show": true,
"values": []
},
"yaxes": [
{
"decimals": null,
"format": "none",
"label": "",
"logBase": 1,
"max": null,
"min": null,
"show": true
},
{
"decimals": null,
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
}
],
"yaxis": {
"align": false,
"alignLevel": null
}
},
{
"aliasColors": {},
"bars": false,
"dashLength": 10,
"dashes": false,
"datasource": "$datasource",
"description": "Canal instance ringbuffer 内未释放events占用内存。",
"fill": 1,
"gridPos": {
"h": 5,
"w": 6,
"x": 6,
"y": 19
},
"id": 22,
"legend": {
"avg": false,
"current": false,
"max": false,
"min": false,
"show": true,
"total": false,
"values": false
},
"lines": true,
"linewidth": 1,
"links": [],
"nullPointMode": "null",
"percentage": false,
"pointradius": 5,
"points": false,
"renderer": "flot",
"seriesOverrides": [],
"spaceLength": 10,
"stack": false,
"steppedLine": false,
"targets": [
{
"expr": "(canal_instance_store_produce_mem{destination=~\"$destination\"} - canal_instance_store_consume_mem{destination=~\"$destination\"}) / 1024",
"format": "time_series",
"interval": "15s",
"intervalFactor": 2,
"legendFormat": "memsize",
"refId": "A"
}
],
"thresholds": [],
"timeFrom": null,
"timeShift": null,
"title": "Store remain mem",
"tooltip": {
"shared": true,
"sort": 0,
"value_type": "individual"
},
"type": "graph",
"xaxis": {
"buckets": null,
"mode": "time",
"name": null,
"show": true,
"values": []
},
"yaxes": [
{
"format": "deckbytes",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
},
{
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
}
],
"yaxis": {
"align": false,
"alignLevel": null
}
} }
], ],
"refresh": false, "refresh": false,
@@ -1230,9 +1305,8 @@
"list": [ "list": [
{ {
"current": { "current": {
"tags": [], "text": "prometheus",
"text": "Prometheus", "value": "prometheus"
"value": "Prometheus"
}, },
"hide": 0, "hide": 0,
"label": "datasource", "label": "datasource",
@@ -1245,10 +1319,7 @@
}, },
{ {
"allValue": null, "allValue": null,
"current": { "current": {},
"text": "example",
"value": "example"
},
"datasource": "$datasource", "datasource": "$datasource",
"hide": 0, "hide": 0,
"includeAll": false, "includeAll": false,
@@ -1269,7 +1340,7 @@
] ]
}, },
"time": { "time": {
"from": "now-1h", "from": "now-6h",
"to": "now" "to": "now"
}, },
"timepicker": { "timepicker": {
@@ -1300,5 +1371,5 @@
"timezone": "", "timezone": "",
"title": "Canal instances", "title": "Canal instances",
"uid": "8vh8NGpiz", "uid": "8vh8NGpiz",
"version": 1 "version": 103
} }

View File

@@ -38,6 +38,8 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
protected boolean filterRows = false; protected boolean filterRows = false;
protected boolean filterTableError = false; protected boolean filterTableError = false;
protected boolean useDruidDdlFilter = true; protected boolean useDruidDdlFilter = true;
// instance received binlog bytes
protected final AtomicLong receivedBinlogBytes = new AtomicLong(0L);
private final AtomicLong eventsPublishBlockingTime = new AtomicLong(0L); private final AtomicLong eventsPublishBlockingTime = new AtomicLong(0L);
protected BinlogParser buildParser() { protected BinlogParser buildParser() {
@@ -204,4 +206,8 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
return this.eventsPublishBlockingTime; return this.eventsPublishBlockingTime;
} }
public AtomicLong getReceivedBinlogBytes() {
return this.receivedBinlogBytes;
}
} }

View File

@@ -71,9 +71,6 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
private int dumpErrorCount = 0; // binlogDump失败异常计数 private int dumpErrorCount = 0; // binlogDump失败异常计数
private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值 private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值
// instance received binlog bytes
private final AtomicLong receivedBinlogBytes = new AtomicLong(0L);
protected ErosaConnection buildErosaConnection() { protected ErosaConnection buildErosaConnection() {
return buildMysqlConnection(this.runningInfo); return buildMysqlConnection(this.runningInfo);
} }
@@ -907,8 +904,4 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
this.dumpErrorCountThreshold = dumpErrorCountThreshold; this.dumpErrorCountThreshold = dumpErrorCountThreshold;
} }
public AtomicLong getReceivedBinlogBytes() {
return this.receivedBinlogBytes;
}
} }

View File

@@ -1,14 +1,15 @@
package com.alibaba.otter.canal.prometheus.impl; package com.alibaba.otter.canal.prometheus.impl;
import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST; import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST;
import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABELS_LIST;
import com.alibaba.otter.canal.parse.inbound.group.GroupEventParser;
import com.alibaba.otter.canal.parse.inbound.mysql.AbstractMysqlEventParser;
import io.prometheus.client.Collector; import io.prometheus.client.Collector;
import io.prometheus.client.CounterMetricFamily; import io.prometheus.client.CounterMetricFamily;
import io.prometheus.client.GaugeMetricFamily; import io.prometheus.client.GaugeMetricFamily;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Arrays; import java.util.Arrays;
import java.util.Collections;
import java.util.List; import java.util.List;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentMap;
@@ -19,7 +20,6 @@ import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.instance.core.CanalInstance; import com.alibaba.otter.canal.instance.core.CanalInstance;
import com.alibaba.otter.canal.parse.CanalEventParser; import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
import com.alibaba.otter.canal.prometheus.InstanceRegistry; import com.alibaba.otter.canal.prometheus.InstanceRegistry;
import com.google.common.base.Preconditions; import com.google.common.base.Preconditions;
@@ -34,11 +34,13 @@ public class ParserCollector extends Collector implements InstanceRegistry {
private static final String RECEIVED_BINLOG = "canal_instance_received_binlog_bytes"; private static final String RECEIVED_BINLOG = "canal_instance_received_binlog_bytes";
private static final String PARSER_MODE = "canal_instance_parser_mode"; private static final String PARSER_MODE = "canal_instance_parser_mode";
private static final String MODE_LABEL = "parallel"; private static final String MODE_LABEL = "parallel";
private static final String PARSER_LABEL = "parser";
private static final String PUBLISH_BLOCKING_HELP = "Publish blocking time of dump thread in milliseconds"; private static final String PUBLISH_BLOCKING_HELP = "Publish blocking time of dump thread in milliseconds";
private static final String RECEIVED_BINLOG_HELP = "Received binlog bytes"; private static final String RECEIVED_BINLOG_HELP = "Received binlog bytes";
private static final String MODE_HELP = "Parser mode(parallel/serial) of instance"; private static final String MODE_HELP = "Parser mode(parallel/serial) of instance";
private final List<String> modeLabels = Arrays.asList(DEST, MODE_LABEL); private final List<String> modeLabels = Arrays.asList(DEST, MODE_LABEL);
private final ConcurrentMap<String, ParserMetricsHolder> instances = new ConcurrentHashMap<String, ParserMetricsHolder>(); private final List<String> parserLabels = Arrays.asList(DEST, PARSER_LABEL);
private final ConcurrentMap<String, ParserMetricsHolder> instances = new ConcurrentHashMap<>();
private ParserCollector() {} private ParserCollector() {}
@@ -52,53 +54,87 @@ public class ParserCollector extends Collector implements InstanceRegistry {
@Override @Override
public List<MetricFamilySamples> collect() { public List<MetricFamilySamples> collect() {
List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>(); List<MetricFamilySamples> mfs = new ArrayList<>();
boolean hasParallel = false;
CounterMetricFamily bytesCounter = new CounterMetricFamily(RECEIVED_BINLOG, CounterMetricFamily bytesCounter = new CounterMetricFamily(RECEIVED_BINLOG,
RECEIVED_BINLOG_HELP, DEST_LABELS_LIST); RECEIVED_BINLOG_HELP, parserLabels);
GaugeMetricFamily modeGauge = new GaugeMetricFamily(PARSER_MODE, GaugeMetricFamily modeGauge = new GaugeMetricFamily(PARSER_MODE,
MODE_HELP, modeLabels); MODE_HELP, modeLabels);
CounterMetricFamily blockingCounter = new CounterMetricFamily(PUBLISH_BLOCKING, CounterMetricFamily blockingCounter = new CounterMetricFamily(PUBLISH_BLOCKING,
PUBLISH_BLOCKING_HELP, DEST_LABELS_LIST); PUBLISH_BLOCKING_HELP, parserLabels);
for (ParserMetricsHolder emh : instances.values()) { for (ParserMetricsHolder emh : instances.values()) {
if (emh.isParallel) { if (emh instanceof GroupParserMetricsHolder) {
blockingCounter.addMetric(emh.destLabelValues, (emh.eventsPublishBlockingTime.doubleValue() / NANO_PER_MILLI)); GroupParserMetricsHolder group = (GroupParserMetricsHolder) emh;
hasParallel = true; for (ParserMetricsHolder semh : group.holders) {
singleCollect(bytesCounter, blockingCounter, modeGauge, semh);
}
}
else {
singleCollect(bytesCounter, blockingCounter, modeGauge, emh);
} }
modeGauge.addMetric(emh.modeLabelValues, 1);
bytesCounter.addMetric(emh.destLabelValues, emh.receivedBinlogBytes.doubleValue());
} }
mfs.add(bytesCounter); mfs.add(bytesCounter);
mfs.add(modeGauge); mfs.add(modeGauge);
if (hasParallel) { if (!blockingCounter.samples.isEmpty()) {
mfs.add(blockingCounter); mfs.add(blockingCounter);
} }
return mfs; return mfs;
} }
private void singleCollect(CounterMetricFamily bytesCounter, CounterMetricFamily blockingCounter, GaugeMetricFamily modeGauge, ParserMetricsHolder holder) {
if (holder.isParallel) {
blockingCounter.addMetric(holder.parserLabelValues, (holder.eventsPublishBlockingTime.doubleValue() / NANO_PER_MILLI));
}
modeGauge.addMetric(holder.modeLabelValues, 1);
bytesCounter.addMetric(holder.parserLabelValues, holder.receivedBinlogBytes.doubleValue());
}
@Override @Override
public void register(CanalInstance instance) { public void register(CanalInstance instance) {
final String destination = instance.getDestination(); final String destination = instance.getDestination();
ParserMetricsHolder holder = new ParserMetricsHolder(); ParserMetricsHolder holder;
CanalEventParser parser = instance.getEventParser(); CanalEventParser parser = instance.getEventParser();
if (!(parser instanceof MysqlEventParser)) { if (parser instanceof AbstractMysqlEventParser) {
throw new IllegalArgumentException("CanalEventParser must be MysqlEventParser"); holder = singleHolder(destination, (AbstractMysqlEventParser)parser, "0");
} else if (parser instanceof GroupEventParser) {
holder = groupHolder(destination, (GroupEventParser)parser);
} else {
throw new IllegalArgumentException("CanalEventParser must be either AbstractMysqlEventParser or GroupEventParser.");
} }
MysqlEventParser mysqlParser = (MysqlEventParser) parser; Preconditions.checkNotNull(holder);
holder.destLabelValues = Collections.singletonList(destination);
holder.modeLabelValues = Arrays.asList(destination, Boolean.toString(mysqlParser.isParallel()));
holder.eventsPublishBlockingTime = mysqlParser.getEventsPublishBlockingTime();
holder.receivedBinlogBytes = mysqlParser.getReceivedBinlogBytes();
holder.isParallel = mysqlParser.isParallel();
Preconditions.checkNotNull(holder.eventsPublishBlockingTime);
Preconditions.checkNotNull(holder.receivedBinlogBytes);
ParserMetricsHolder old = instances.put(destination, holder); ParserMetricsHolder old = instances.put(destination, holder);
if (old != null) { if (old != null) {
logger.warn("Remove stale ParserCollector for instance {}.", destination); logger.warn("Remove stale ParserCollector for instance {}.", destination);
} }
} }
private ParserMetricsHolder singleHolder(String destination, AbstractMysqlEventParser parser, String id) {
ParserMetricsHolder holder = new ParserMetricsHolder();
holder.parserLabelValues = Arrays.asList(destination, id);
holder.modeLabelValues = Arrays.asList(destination, Boolean.toString(parser.isParallel()));
holder.eventsPublishBlockingTime = parser.getEventsPublishBlockingTime();
holder.receivedBinlogBytes = parser.getReceivedBinlogBytes();
holder.isParallel = parser.isParallel();
Preconditions.checkNotNull(holder.eventsPublishBlockingTime);
Preconditions.checkNotNull(holder.receivedBinlogBytes);
return holder;
}
private GroupParserMetricsHolder groupHolder(String destination, GroupEventParser group) {
List<CanalEventParser> parsers = group.getEventParsers();
GroupParserMetricsHolder groupHolder = new GroupParserMetricsHolder();
int num = parsers.size();
for (int i = 0; i < num; i ++) {
CanalEventParser parser = parsers.get(i);
if (parser instanceof AbstractMysqlEventParser) {
ParserMetricsHolder single = singleHolder(destination, (AbstractMysqlEventParser)parser, Integer.toString(i + 1));
groupHolder.holders.add(single);
} else {
logger.warn("Null or non AbstractMysqlEventParser, ignore.");
}
}
return groupHolder;
}
@Override @Override
public void unregister(CanalInstance instance) { public void unregister(CanalInstance instance) {
final String destination = instance.getDestination(); final String destination = instance.getDestination();
@@ -106,11 +142,17 @@ public class ParserCollector extends Collector implements InstanceRegistry {
} }
private class ParserMetricsHolder { private class ParserMetricsHolder {
private List<String> destLabelValues; private List<String> parserLabelValues;
private List<String> modeLabelValues; private List<String> modeLabelValues;
// metrics for single parser
private AtomicLong receivedBinlogBytes; private AtomicLong receivedBinlogBytes;
private AtomicLong eventsPublishBlockingTime; private AtomicLong eventsPublishBlockingTime;
// parser mode
private boolean isParallel; private boolean isParallel;
} }
private class GroupParserMetricsHolder extends ParserMetricsHolder {
private final List<ParserMetricsHolder> holders = new ArrayList<>();
}
} }