SML0127 commented on code in PR #3456:
URL: https://github.com/apache/flink-cdc/pull/3456#discussion_r1677533767
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/metrics/MySqlSourceReaderMetrics.java:
##########
@@ -44,6 +69,49 @@ public void registerMetrics() {
MetricNames.CURRENT_FETCH_EVENT_TIME_LAG, (Gauge<Long>)
this::getFetchDelay);
}
+ public void numRecordsOutSnapshotIncrease(TableId tableId) {
+ Counter counter =
+ snapshotNumRecordsOutMap.computeIfAbsent(
+ tableId,
+ k ->
+ metricGroup.counter(
+ tableId.identifier() +
IO_NUM_RECORDS_OUT_SNAPSHOT));
+ counter.inc();
+ }
+
+ public void numRecordsOutInsertIncrease(TableId tableId) {
+ Counter counter =
+ insertNumRecordsOutMap.computeIfAbsent(
+ tableId,
+ k ->
+ metricGroup.counter(
+ tableId.identifier()
+ +
IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_INSERT));
+ counter.inc();
+ }
+
+ public void numRecordsOutUpdateIncrease(TableId tableId) {
+ Counter counter =
+ updateNumRecordsOutMap.computeIfAbsent(
+ tableId,
+ k ->
+ metricGroup.counter(
+ tableId.identifier()
+ +
IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_UPDATE));
+ counter.inc();
+ }
+
+ public void numRecordsOutDeleteIncrease(TableId tableId) {
+ Counter counter =
+ deleteNumRecordsOutMap.computeIfAbsent(
+ tableId,
+ k ->
+ metricGroup.counter(
+ tableId.identifier()
+ +
IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_DELETE));
+ counter.inc();
+ }
+
Review Comment:
I applied this metrics to my code, and left some comments.
- Metric names should be general.
- In your case, It generate uniqe metric for each tables. But in
monitorig tool such as grafana, we distingush tables by property in metric, not
by mtric name
So I suggest to remove tableId.
```suggestion
public void numRecordsOutSnapshotIncrease(TableId tableId) {
Counter counter =
snapshotNumRecordsOutMap.computeIfAbsent(
tableId,
k ->
metricGroup.counter(IO_NUM_RECORDS_OUT_SNAPSHOT));
counter.inc();
}
public void numRecordsOutInsertIncrease(TableId tableId) {
Counter counter =
insertNumRecordsOutMap.computeIfAbsent(
tableId,
k ->
metricGroup.counter(IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_INSERT));
counter.inc();
}
public void numRecordsOutUpdateIncrease(TableId tableId) {
Counter counter =
updateNumRecordsOutMap.computeIfAbsent(
tableId,
k ->
metricGroup.counter(IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_UPDATE));
counter.inc();
}
public void numRecordsOutDeleteIncrease(TableId tableId) {
Counter counter =
deleteNumRecordsOutMap.computeIfAbsent(
tableId,
k ->
metricGroup.counter(IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_DELETE));
counter.inc();
}
```
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/metrics/MySqlSourceReaderMetrics.java:
##########
@@ -44,6 +69,49 @@ public void registerMetrics() {
MetricNames.CURRENT_FETCH_EVENT_TIME_LAG, (Gauge<Long>)
this::getFetchDelay);
}
+ public void numRecordsOutSnapshotIncrease(TableId tableId) {
+ Counter counter =
+ snapshotNumRecordsOutMap.computeIfAbsent(
+ tableId,
+ k ->
+ metricGroup.counter(
+ tableId.identifier() +
IO_NUM_RECORDS_OUT_SNAPSHOT));
+ counter.inc();
+ }
+
+ public void numRecordsOutInsertIncrease(TableId tableId) {
+ Counter counter =
+ insertNumRecordsOutMap.computeIfAbsent(
+ tableId,
+ k ->
+ metricGroup.counter(
+ tableId.identifier()
+ +
IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_INSERT));
+ counter.inc();
+ }
+
+ public void numRecordsOutUpdateIncrease(TableId tableId) {
+ Counter counter =
+ updateNumRecordsOutMap.computeIfAbsent(
+ tableId,
+ k ->
+ metricGroup.counter(
+ tableId.identifier()
+ +
IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_UPDATE));
+ counter.inc();
+ }
+
+ public void numRecordsOutDeleteIncrease(TableId tableId) {
+ Counter counter =
+ deleteNumRecordsOutMap.computeIfAbsent(
+ tableId,
+ k ->
+ metricGroup.counter(
+ tableId.identifier()
+ +
IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_DELETE));
+ counter.inc();
+ }
+
Review Comment:
I applied this metrics to my code, and left some comments.
- Metric names should be general.
- In your case, It generate uniqe metric for each tables. But in monitorig
tool such as grafana, we distingush tables by property in metric, not by mtric
name
So I suggest to remove tableId.
```suggestion
public void numRecordsOutSnapshotIncrease(TableId tableId) {
Counter counter =
snapshotNumRecordsOutMap.computeIfAbsent(
tableId,
k ->
metricGroup.counter(IO_NUM_RECORDS_OUT_SNAPSHOT));
counter.inc();
}
public void numRecordsOutInsertIncrease(TableId tableId) {
Counter counter =
insertNumRecordsOutMap.computeIfAbsent(
tableId,
k ->
metricGroup.counter(IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_INSERT));
counter.inc();
}
public void numRecordsOutUpdateIncrease(TableId tableId) {
Counter counter =
updateNumRecordsOutMap.computeIfAbsent(
tableId,
k ->
metricGroup.counter(IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_UPDATE));
counter.inc();
}
public void numRecordsOutDeleteIncrease(TableId tableId) {
Counter counter =
deleteNumRecordsOutMap.computeIfAbsent(
tableId,
k ->
metricGroup.counter(IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_DELETE));
counter.inc();
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
