[cdc-connector][mysql]Mysql add numRecordsOutBySnapshot and numRecordsOutByIncremental metrics #3456

ChengJie1053 · 2024-07-06T10:06:11Z

numRecordsOutBySnapshot metrics collects snapshot statistics

numRecordsOutByIncremental metrics collects incremental statistics

numRecordsOutByIncremental has three metrics, respectively is numRecordsOutByIncrementalInsert, numRecordsOutByIncrementalUpdate, numRecordsOutByIncrementalDelete

2024.7.12 change metrics name，change numRecordsOutByIncrementalInsert to numRecordsOutByDataChangeEventInsert

…sOutByIncremental metrics

SML0127 · 2024-07-12T08:46:16Z

...main/java/org/apache/flink/cdc/connectors/mysql/source/metrics/MySqlSourceReaderMetrics.java

+    public static final String IO_NUM_RECORDS_OUT_INCREMENTAL_INSERT =
+            ".numRecordsOutByIncrementalInsert";
+
+    public static final String IO_NUM_RECORDS_OUT_INCREMENTAL_DELETE =
+            ".numRecordsOutByIncrementalDelete";
+
+    public static final String IO_NUM_RECORDS_OUT_INCREMENTAL_UPDATE =
+            ".numRecordsOutByIncrementalUpdate";


Thx for contirbution @ChengJie1053

Nit: How about changing INCREMENTAL to BINLOG or CHANGE_EVENT? INCREMENTAL can be understood as a Incremental Snapshot step.

I think your suggestion is great, and I will CHANGE it to CHANGE EVENT

SML0127 · 2024-07-12T08:53:14Z

...dc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlRecordEmitter.java

+            if (RecordUtils.isDataChangeRecord(element)) {
+                Struct value = (Struct) element.value();
+                if (value != null) {
+                    Envelope.Operation operation =
+                            Envelope.Operation.forCode(
+                                    value.getString(Envelope.FieldName.OPERATION));
+                    switch (operation) {
+                        case CREATE:
+                            sourceReaderMetrics.numRecordsOutInsertIncrease(tableId);
+                            break;
+                        case UPDATE:
+                            sourceReaderMetrics.numRecordsOutUpdateIncrease(tableId);
+                            break;
+                        case DELETE:
+                            sourceReaderMetrics.numRecordsOutDeleteIncrease(tableId);
+                            break;
+                    }
+                }
+            }


We checked whether element is data change record or not in line 101.
How about removing this condition?

Suggested change

if (RecordUtils.isDataChangeRecord(element)) {

Struct value = (Struct) element.value();

if (value != null) {

Envelope.Operation operation =

Envelope.Operation.forCode(

value.getString(Envelope.FieldName.OPERATION));

switch (operation) {

case CREATE:

sourceReaderMetrics.numRecordsOutInsertIncrease(tableId);

break;

case UPDATE:

sourceReaderMetrics.numRecordsOutUpdateIncrease(tableId);

break;

case DELETE:

sourceReaderMetrics.numRecordsOutDeleteIncrease(tableId);

break;

}

}

}

Struct value = (Struct) element.value();

if (value != null) {

Envelope.Operation operation =

Envelope.Operation.forCode(

value.getString(Envelope.FieldName.OPERATION));

switch (operation) {

case CREATE:

sourceReaderMetrics.numRecordsOutInsertIncrease(tableId);

break;

case UPDATE:

sourceReaderMetrics.numRecordsOutUpdateIncrease(tableId);

break;

case DELETE:

sourceReaderMetrics.numRecordsOutDeleteIncrease(tableId);

break;

}

}

Thank you very much for helping me review the code.
I will delete the isDataChangeRecord judgment

SML0127

LGTM!

SML0127

PTAL

SML0127 · 2024-07-15T09:17:56Z

...main/java/org/apache/flink/cdc/connectors/mysql/source/metrics/MySqlSourceReaderMetrics.java

+    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();
+    }
+


I applied this metrics to my code, and left some comments.

Metric names should be general.

In your case, It generate uniqe metrics for each tables. But in monitorig tool such as grafana, it distingush tables by property such as job name in metric, not by metric name
So I suggest to remove tableId.identifier() in metric name.

Suggested change

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();

}

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();

}

Ok, thanks for reviewing the code for me, but the original purpose of this metric is to see each table at a granular level, if you need to see the total, I will add other metrics

SML0127 · 2024-07-15T09:19:05Z

...main/java/org/apache/flink/cdc/connectors/mysql/source/metrics/MySqlSourceReaderMetrics.java

+    public static final String IO_NUM_RECORDS_OUT_SNAPSHOT = ".numRecordsOutBySnapshot";
+
+    public static final String IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_INSERT =
+            ".numRecordsOutByDataChangeEventInsert";
+
+    public static final String IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_DELETE =
+            ".numRecordsOutByDataChangeEventDelete";
+
+    public static final String IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_UPDATE =
+            ".numRecordsOutByDataChangeEventUpdate";


I suggest to remove dot(.) from metric name.

Suggested change

public static final String IO_NUM_RECORDS_OUT_SNAPSHOT = ".numRecordsOutBySnapshot";

public static final String IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_INSERT =

".numRecordsOutByDataChangeEventInsert";

public static final String IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_DELETE =

".numRecordsOutByDataChangeEventDelete";

public static final String IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_UPDATE =

".numRecordsOutByDataChangeEventUpdate";

public static final String IO_NUM_RECORDS_OUT_SNAPSHOT = "numRecordsOutBySnapshot";

public static final String IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_INSERT =

"numRecordsOutByDataChangeEventInsert";

public static final String IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_DELETE =

"numRecordsOutByDataChangeEventDelete";

public static final String IO_NUM_RECORDS_OUT_DATA_CHANGE_EVENT_UPDATE =

"numRecordsOutByDataChangeEventUpdate";

Ok, thank you for helping me review the code, I will remove (.)

yuxiqian · 2024-07-16T11:18:16Z

Seems we can't come up with a "best" metric design for everyone: Some may want a task-level unified record counter, while some may want fine-grained counters for each table. Or one may want to disable it entirely to avoid any extra cost.

I wonder if we should add an option to control metric behavior? cc @ruanhang1993

SML0127 · 2024-08-10T07:30:57Z

@ChengJie1053 cc. @yuxiqian
Is thera any changes or progress? If not, I want to create new PR with my suggestion at next week.
(I need this feature.)

github-actions · 2024-10-10T00:03:57Z

This pull request has been automatically marked as stale because it has not had recent activity for 60 days. It will be closed in 30 days if no further activity occurs.

github-actions · 2024-11-09T00:04:12Z

This pull request has been closed because it has not had recent activity. You could reopen it if you try to continue your work, and anyone who are interested in it are encouraged to continue work on this pull request.

[cdc-connector][mysql]Mysql add numRecordsOutBySnapshot and numRecord…

5b9f9a1

…sOutByIncremental metrics

github-actions bot added the mysql-cdc-connector label Jul 6, 2024

SML0127 reviewed Jul 12, 2024

View reviewed changes

Modifying the metrics name

dee79f5

SML0127 approved these changes Jul 12, 2024

View reviewed changes

optimized code

a7b23b8

SML0127 suggested changes Jul 15, 2024

View reviewed changes

SML0127 mentioned this pull request Aug 10, 2024

[FLINK-36027][cdc-connector][mysql] numRecordsOut/numRecordsOutRate metrics for each OperationType #3529

Open

github-actions bot added the Stale label Oct 10, 2024

github-actions bot closed this Nov 9, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[cdc-connector][mysql]Mysql add numRecordsOutBySnapshot and numRecordsOutByIncremental metrics #3456

[cdc-connector][mysql]Mysql add numRecordsOutBySnapshot and numRecordsOutByIncremental metrics #3456

ChengJie1053 commented Jul 6, 2024 •

edited

Loading

SML0127 Jul 12, 2024 •

edited

Loading

ChengJie1053 Jul 12, 2024

SML0127 Jul 12, 2024

ChengJie1053 Jul 12, 2024

SML0127 left a comment

SML0127 left a comment

SML0127 Jul 15, 2024 •

edited

Loading

ChengJie1053 Jul 16, 2024

SML0127 Jul 15, 2024

ChengJie1053 Jul 16, 2024

yuxiqian commented Jul 16, 2024 •

edited

Loading

SML0127 commented Aug 10, 2024

github-actions bot commented Oct 10, 2024

github-actions bot commented Nov 9, 2024

[cdc-connector][mysql]Mysql add numRecordsOutBySnapshot and numRecordsOutByIncremental metrics #3456

[cdc-connector][mysql]Mysql add numRecordsOutBySnapshot and numRecordsOutByIncremental metrics #3456

Conversation

ChengJie1053 commented Jul 6, 2024 • edited Loading

SML0127 Jul 12, 2024 • edited Loading

Choose a reason for hiding this comment

ChengJie1053 Jul 12, 2024

Choose a reason for hiding this comment

SML0127 Jul 12, 2024

Choose a reason for hiding this comment

ChengJie1053 Jul 12, 2024

Choose a reason for hiding this comment

SML0127 left a comment

Choose a reason for hiding this comment

SML0127 left a comment

Choose a reason for hiding this comment

SML0127 Jul 15, 2024 • edited Loading

Choose a reason for hiding this comment

ChengJie1053 Jul 16, 2024

Choose a reason for hiding this comment

SML0127 Jul 15, 2024

Choose a reason for hiding this comment

ChengJie1053 Jul 16, 2024

Choose a reason for hiding this comment

yuxiqian commented Jul 16, 2024 • edited Loading

SML0127 commented Aug 10, 2024

github-actions bot commented Oct 10, 2024

github-actions bot commented Nov 9, 2024

ChengJie1053 commented Jul 6, 2024 •

edited

Loading

SML0127 Jul 12, 2024 •

edited

Loading

SML0127 Jul 15, 2024 •

edited

Loading

yuxiqian commented Jul 16, 2024 •

edited

Loading