Repository: pingcap/tiflow
Pull request #4108 (2021-12-28 07:43:33)

pkg/p2p, metrics(cdc): add dashboard panels for Peer Messages
<!--
Thank you for contributing to TiDB-CDC! Please read MD's [CONTRIBUTING](https://github.com/pingcap/tidb-cdc/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve?
- We need some Grafana panels to monitor the status of `pkg/p2p`.

### What is changed and how it works?
- Added Grafana panels.
- Added some new metrics.
- Minor adjustments to the implementation in `pkg/p2p`.
- Modified some default configurations regarding message batching to optimize for performance.


![Screenshot from 2021-12-28 15-51-22](https://user-images.githubusercontent.com/18745285/147542009-bf569330-8fc5-4570-91b6-9df5990dcda3.png)

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test

Code changes

 - Has persistent data change

Related changes

 - Need to update key monitor metrics in both TiCDC document and official document

### Release note <!-- bugfixes or new feature need a release note -->

```release-note
None
```


ti-chi-bot: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: Please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

ti-chi-bot: [REVIEW NOTIFICATION]

This pull request has not been approved.


To complete the [pull request process](https://book.prow.tidb.io/#/en/workflows/pr), please ask the reviewers in the [list](https://prow.tidb.io/tichi/repos/pingcap/tiflow/pulls/4108/owners) to review by filling `/cc @reviewer` in the comment.
After your PR has acquired the required number of LGTMs, you can assign this pull request to the committer in the [list](https://prow.tidb.io/tichi/repos/pingcap/tiflow/pulls/4108/owners) by filling  `/assign @committer` in the comment to help you merge this pull request.

The full list of commands accepted by this bot can be found [here](https://prow.tidb.io/command-help?repo=pingcap%2Ftiflow).

<details>

Reviewer can indicate their review by submitting an approval review.
Reviewer can cancel approval by submitting a request changes review.
</details>

<!--Review Notification Identifier-->
ti-chi-bot: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: Please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

ti-chi-bot: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: Please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

ti-chi-bot: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: Please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

ti-chi-bot: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: Please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

liuzix: /run-leak-tests
liuzix: /run-leak-tests
ti-chi-bot: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: Please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

liuzix: /run-leak-tests
ti-chi-bot: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: Please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

codecov-commenter: # [Codecov](https://codecov.io/gh/pingcap/tiflow/pull/4108?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) Report
> Merging [#4108](https://codecov.io/gh/pingcap/tiflow/pull/4108?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) (35510fe) into [master](https://codecov.io/gh/pingcap/tiflow/commit/3873d3938927e8a801276a5ab4ae889028e5073f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) (3873d39) will **decrease** coverage by `1.8203%`.
> The diff coverage is `68.3620%`.


| Flag | Coverage Δ | |
|---|---|---|
| cdc | `58.8359% <68.3620%> (+0.5994%)` | :arrow_up: |
| dm | `52.1810% <ø> (-3.8536%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap#carryforward-flags-in-the-pull-request-comment) to find out more.

```diff
@@               Coverage Diff                @@
##             master      #4108        +/-   ##
================================================
- Coverage   57.0741%   55.2538%   -1.8204%     
================================================
  Files           478        489        +11     
  Lines         56551      60394      +3843     
================================================
+ Hits          32276      33370      +1094     
- Misses        20978      23662      +2684     
- Partials       3297       3362        +65     
```

ti-chi-bot: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: Please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

liuzix: /merge
ti-chi-bot: This pull request has been accepted and is ready to merge. <details>Commit hash: d399269017bc71a791df4e06548253505e1d3d04</details>
ti-chi-bot: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: Please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

ti-chi-bot: @liuzix: Your PR was out of date, I have automatically updated it for you.

At the same time I will also trigger all tests for you:

/run-all-tests

If the CI test fails, you just re-trigger the test that failed and the bot will merge the PR for you after the CI passes.


<details>

Instructions for interacting with me using PR comments are available [here](https://prow.tidb.io/command-help).  If you have questions or suggestions related to my behavior, please file an issue against the [ti-community-infra/tichi](https://github.com/ti-community-infra/tichi/issues/new?title=Prow%20issue:) repository.
</details>
ti-chi-bot: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: Please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

liuzix: /run-all-tests
liuzix: /run-kafka-integration-test
/run-verify
ti-chi-bot: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: Please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

ti-chi-bot: @liuzix: Your PR was out of date, I have automatically updated it for you.

At the same time I will also trigger all tests for you:

/run-all-tests

If the CI test fails, you just re-trigger the test that failed and the bot will merge the PR for you after the CI passes.


<details>

Instructions for interacting with me using PR comments are available [here](https://prow.tidb.io/command-help).  If you have questions or suggestions related to my behavior, please file an issue against the [ti-community-infra/tichi](https://github.com/ti-community-infra/tichi/issues/new?title=Prow%20issue:) repository.
</details>
ti-chi-bot: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: Please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

ti-chi-bot: @liuzix: Your PR was out of date, I have automatically updated it for you.

At the same time I will also trigger all tests for you:

/run-all-tests

If the CI test fails, you just re-trigger the test that failed and the bot will merge the PR for you after the CI passes.


<details>

Instructions for interacting with me using PR comments are available [here](https://prow.tidb.io/command-help).  If you have questions or suggestions related to my behavior, please file an issue against the [ti-community-infra/tichi](https://github.com/ti-community-infra/tichi/issues/new?title=Prow%20issue:) repository.
</details>
liuzix: /run-dm-integration-test

Patch:
diff --git a/cdc/capture/capture.go b/cdc/capture/capture.go
index 1008c4b71..ed65f53ef 100644
--- a/cdc/capture/capture.go
+++ b/cdc/capture/capture.go
@@ -198,6 +198,13 @@ func (c *Capture) reset(ctx context.Context) error {
 		c.grpcService.Reset(c.MessageServer)
 
 		messageClientConfig := conf.Debug.Messages.ToMessageClientConfig()
+
+		// Puts the advertise-addr of the local node to the client config.
+		// This is for metrics purpose only, so that the receiver knows which
+		// node the connections are from.
+		advertiseAddr := conf.AdvertiseAddr
+		messageClientConfig.AdvertisedAddr = advertiseAddr
+
 		c.MessageRouter = p2p.NewMessageRouter(c.info.ID, conf.Security, messageClientConfig)
 	}
 
diff --git a/cdc/metrics.go b/cdc/metrics.go
index 1a75a316a..1c9985217 100644
--- a/cdc/metrics.go
+++ b/cdc/metrics.go
@@ -30,6 +30,7 @@ import (
 	"github.com/pingcap/tiflow/pkg/db"
 	"github.com/pingcap/tiflow/pkg/etcd"
 	"github.com/pingcap/tiflow/pkg/orchestrator"
+	"github.com/pingcap/tiflow/pkg/p2p"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
@@ -50,6 +51,7 @@ func init() {
 	initServerMetrics(registry)
 	actor.InitMetrics(registry)
 	orchestrator.InitMetrics(registry)
+	p2p.InitMetrics(registry)
 	// Sorter metrics
 	sorter.InitMetrics(registry)
 	memory.InitMetrics(registry)
diff --git a/metrics/grafana/ticdc.json b/metrics/grafana/ticdc.json
index 83e9388f2..60574864a 100644
--- a/metrics/grafana/ticdc.json
+++ b/metrics/grafana/ticdc.json
@@ -9863,6 +9863,778 @@
       "title": "TiKV",
       "type": "row"
     },
+    {
+      "collapsed": true,
+      "gridPos": {
+        "h": 1,
+        "w": 24,
+        "x": 0,
+        "y": 7
+      },
+      "id": 294,
+      "panels": [],
+      "title": "Peer Messages",
+      "type": "row"
+    },
+    {
+      "aliasColors": {},
+      "bars": false,
+      "dashLength": 10,
+      "dashes": false,
+      "datasource": "${DS_CDC-CLUSTER1}",
+      "fieldConfig": {
+        "defaults": {},
+        "overrides": []
+      },
+      "fill": 1,
+      "fillGradient": 0,
+      "gridPos": {
+        "h": 7,
+        "w": 4,
+        "x": 0,
+        "y": 8
+      },
+      "hiddenSeries": false,
+      "id": 321,
+      "legend": {
+        "avg": false,
+        "current": false,
+        "max": false,
+        "min": false,
+        "show": true,
+        "total": false,
+        "values": false
+      },
+      "lines": true,
+      "linewidth": 1,
+      "nullPointMode": "null",
+      "options": {
+        "alertThreshold": true
+      },
+      "percentage": false,
+      "pluginVersion": "7.5.7",
+      "pointradius": 2,
+      "points": false,
+      "renderer": "flot",
+      "seriesOverrides": [],
+      "spaceLength": 10,
+      "stack": false,
+      "steppedLine": false,
+      "targets": [
+        {
+          "exemplar": true,
+          "expr": "sum without (from) (rate(ticdc_message_server_message_count{instance=~\"$capture\"}[30s]))",
+          "interval": "",
+          "legendFormat": "{{instance}}",
+          "queryType": "randomWalk",
+          "refId": "A"
+        }
+      ],
+      "thresholds": [],
+      "timeFrom": null,
+      "timeRegions": [],
+      "timeShift": null,
+      "title": "Message Receive Rate",
+      "tooltip": {
+        "shared": true,
+        "sort": 0,
+        "value_type": "individual"
+      },
+      "type": "graph",
+      "xaxis": {
+        "buckets": null,
+        "mode": "time",
+        "name": null,
+        "show": true,
+        "values": []
+      },
+      "yaxes": [
+        {
+          "format": "short",
+          "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
+      }
+    },
+    {
+      "aliasColors": {},
+      "bars": false,
+      "dashLength": 10,
+      "dashes": false,
+      "datasource": "${DS_CDC-CLUSTER1}",
+      "fieldConfig": {
+        "defaults": {},
+        "overrides": []
+      },
+      "fill": 1,
+      "fillGradient": 0,
+      "gridPos": {
+        "h": 7,
+        "w": 4,
+        "x": 4,
+        "y": 8
+      },
+      "hiddenSeries": false,
+      "id": 323,
+      "legend": {
+        "avg": false,
+        "current": false,
+        "max": false,
+        "min": false,
+        "show": true,
+        "total": false,
+        "values": false
+      },
+      "lines": true,
+      "linewidth": 1,
+      "nullPointMode": "null",
+      "options": {
+        "alertThreshold": true
+      },
+      "percentage": false,
+      "pluginVersion": "7.5.7",
+      "pointradius": 2,
+      "points": false,
+      "renderer": "flot",
+      "seriesOverrides": [],
+      "spaceLength": 10,
+      "stack": false,
+      "steppedLine": false,
+      "targets": [
+        {
+          "exemplar": true,
+          "expr": "sum without (to) (rate(ticdc_message_client_message_count{instance=~\"$capture\"}[30s]))",
+          "interval": "",
+          "legendFormat": "{{instance}}",
+          "queryType": "randomWalk",
+          "refId": "A"
+        }
+      ],
+      "thresholds": [],
+      "timeFrom": null,
+      "timeRegions": [],
+      "timeShift": null,
+      "title": "Message Send Rate",
+      "tooltip": {
+        "shared": true,
+        "sort": 0,
+        "value_type": "individual"
+      },
+      "type": "graph",
+      "xaxis": {
+        "buckets": null,
+        "mode": "time",
+        "name": null,
+        "show": true,
+        "values": []
+      },
+      "yaxes": [
+        {
+          "format": "short",
+          "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
+      }
+    },
+    {
+      "aliasColors": {},
+      "bars": false,
+      "dashLength": 10,
+      "dashes": false,
+      "datasource": "${DS_CDC-CLUSTER1}",
+      "fieldConfig": {
+        "defaults": {},
+        "overrides": []
+      },
+      "fill": 0,
+      "fillGradient": 0,
+      "gridPos": {
+        "h": 7,
+        "w": 4,
+        "x": 8,
+        "y": 8
+      },
+      "hiddenSeries": false,
+      "id": 352,
+      "legend": {
+        "avg": false,
+        "current": false,
+        "hideEmpty": true,
+        "hideZero": true,
+        "max": false,
+        "min": false,
+        "show": true,
+        "total": false,
+        "values": false
+      },
+      "lines": true,
+      "linewidth": 1,
+      "nullPointMode": "null",
+      "options": {
+        "alertThreshold": true
+      },
+      "percentage": false,
+      "pluginVersion": "7.5.7",
+      "pointradius": 2,
+      "points": false,
+      "renderer": "flot",
+      "seriesOverrides": [],
+      "spaceLength": 10,
+      "stack": false,
+      "steppedLine": false,
+      "targets": [
+        {
+          "exemplar": true,
+          "expr": "histogram_quantile(0.5, sum(rate(ticdc_message_server_message_batch_size_bucket[30s])) by (le))",
+          "interval": "",
+          "legendFormat": "p50",
+          "queryType": "randomWalk",
+          "refId": "A"
+        },
+        {
+          "exemplar": true,
+          "expr": "histogram_quantile(0.95, sum(rate(ticdc_message_server_message_batch_size_bucket[30s])) by (le))",
+          "hide": false,
+          "interval": "",
+          "legendFormat": "p90",
+          "refId": "B"
+        },
+        {
+          "exemplar": true,
+          "expr": "histogram_quantile(0.99, sum(rate(ticdc_message_server_message_batch_size_bucket[30s])) by (le))",
+          "hide": false,
+          "interval": "",
+          "legendFormat": "p99",
+          "refId": "C"
+        }
+      ],
+      "thresholds": [],
+      "timeFrom": null,
+      "timeRegions": [],
+      "timeShift": null,
+      "title": "Message Batch Size",
+      "tooltip": {
+        "shared": true,
+        "sort": 0,
+        "value_type": "individual"
+      },
+      "transformations": [],
+      "type": "graph",
+      "xaxis": {
+        "buckets": null,
+        "mode": "time",
+        "name": null,
+        "show": true,
+        "values": []
+      },
+      "yaxes": [
+        {
+          "format": "short",
+          "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
+      }
+    },
+    {
+      "aliasColors": {},
+      "bars": false,
+      "dashLength": 10,
+      "dashes": false,
+      "datasource": "${DS_CDC-CLUSTER1}",
+      "fieldConfig": {
+        "defaults": {
+          "unit": "short"
+        },
+        "overrides": []
+      },
+      "fill": 0,
+      "fillGradient": 0,
+      "gridPos": {
+        "h": 7,
+        "w": 6,
+        "x": 12,
+        "y": 8
+      },
+      "hiddenSeries": false,
+      "id": 354,
+      "legend": {
+        "avg": false,
+        "current": false,
+        "max": false,
+        "min": false,
+        "show": true,
+        "total": false,
+        "values": false
+      },
+      "lines": true,
+      "linewidth": 1,
+      "nullPointMode": "null",
+      "options": {
+        "alertThreshold": true
+      },
+      "percentage": false,
+      "pluginVersion": "7.5.7",
+      "pointradius": 2,
+      "points": false,
+      "renderer": "flot",
+      "seriesOverrides": [],
+      "spaceLength": 10,
+      "stack": false,
+      "steppedLine": false,
+      "targets": [
+        {
+          "exemplar": true,
+          "expr": "histogram_quantile(0.5, sum(rate(ticdc_message_server_message_batch_bytes_bucket [30s])) by (le))",
+          "hide": false,
+          "interval": "",
+          "legendFormat": "p50",
+          "refId": "C"
+        },
+        {
+          "exemplar": true,
+          "expr": "histogram_quantile(0.95, sum(rate(ticdc_message_server_message_batch_bytes_bucket [30s])) by (le))",
+          "interval": "",
+          "legendFormat": "p95",
+          "queryType": "randomWalk",
+          "refId": "A"
+        },
+        {
+          "exemplar": true,
+          "expr": "histogram_quantile(0.99, sum(rate(ticdc_message_server_message_batch_bytes_bucket [30s])) by (le))",
+          "hide": false,
+          "interval": "",
+          "legendFormat": "p99",
+          "refId": "B"
+        }
+      ],
+      "thresholds": [],
+      "timeFrom": null,
+      "timeRegions": [],
+      "timeShift": null,
+      "title": "Receive message batch bytes percentiles",
+      "tooltip": {
+        "shared": true,
+        "sort": 0,
+        "value_type": "individual"
+      },
+      "type": "graph",
+      "xaxis": {
+        "buckets": null,
+        "mode": "time",
+        "name": null,
+        "show": true,
+        "values": []
+      },
+      "yaxes": [
+        {
+          "format": "short",
+          "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
+      }
+    },
+    {
+      "aliasColors": {},
+      "bars": false,
+      "dashLength": 10,
+      "dashes": false,
+      "datasource": "${DS_CDC-CLUSTER1}",
+      "fieldConfig": {
+        "defaults": {},
+        "overrides": []
+      },
+      "fill": 0,
+      "fillGradient": 0,
+      "gridPos": {
+        "h": 7,
+        "w": 6,
+        "x": 18,
+        "y": 8
+      },
+      "hiddenSeries": false,
+      "id": 356,
+      "legend": {
+        "avg": false,
+        "current": false,
+        "max": false,
+        "min": false,
+        "show": true,
+        "total": false,
+        "values": false
+      },
+      "lines": true,
+      "linewidth": 1,
+      "nullPointMode": "null",
+      "options": {
+        "alertThreshold": true
+      },
+      "percentage": false,
+      "pluginVersion": "7.5.7",
+      "pointradius": 2,
+      "points": false,
+      "renderer": "flot",
+      "seriesOverrides": [],
+      "spaceLength": 10,
+      "stack": false,
+      "steppedLine": false,
+      "targets": [
+        {
+          "exemplar": true,
+          "expr": "histogram_quantile(0.5, sum(rate(ticdc_message_server_message_bytes_bucket [30s])) by (le))",
+          "interval": "",
+          "legendFormat": "p50",
+          "queryType": "randomWalk",
+          "refId": "A"
+        },
+        {
+          "exemplar": true,
+          "expr": "histogram_quantile(0.95, sum(rate(ticdc_message_server_message_bytes_bucket [30s])) by (le))",
+          "hide": false,
+          "interval": "",
+          "legendFormat": "p95",
+          "refId": "B"
+        },
+        {
+          "exemplar": true,
+          "expr": "histogram_quantile(0.99, sum(rate(ticdc_message_server_message_bytes_bucket [30s])) by (le))",
+          "hide": false,
+          "interval": "",
+          "legendFormat": "p99",
+          "refId": "C"
+        }
+      ],
+      "thresholds": [],
+      "timeFrom": null,
+      "timeRegions": [],
+      "timeShift": null,
+      "title": "Receive Message Bytes Percentile",
+      "tooltip": {
+        "shared": true,
+        "sort": 0,
+        "value_type": "individual"
+      },
+      "type": "graph",
+      "xaxis": {
+        "buckets": null,
+        "mode": "time",
+        "name": null,
+        "show": true,
+        "values": []
+      },
+      "yaxes": [
+        {
+          "format": "short",
+          "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
+      }
+    },
+    {
+      "columns": [],
+      "datasource": "${DS_CDC-CLUSTER1}",
+      "fieldConfig": {
+        "defaults": {
+          "color": {
+            "mode": "thresholds"
+          },
+          "custom": {
+            "align": "left",
+            "displayMode": "color-background",
+            "filterable": false
+          },
+          "mappings": [],
+          "noValue": "0",
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "rgb(255, 255, 255)",
+                "value": null
+              },
+              {
+                "color": "dark-blue",
+                "value": 1
+              },
+              {
+                "color": "dark-red",
+                "value": 2
+              }
+            ]
+          }
+        },
+        "overrides": [
+          {
+            "matcher": {
+              "id": "byName",
+              "options": "dest\\src"
+            },
+            "properties": [
+              {
+                "id": "custom.width",
+                "value": 137
+              }
+            ]
+          }
+        ]
+      },
+      "fontSize": "100%",
+      "gridPos": {
+        "h": 8,
+        "w": 20,
+        "x": 0,
+        "y": 15
+      },
+      "id": 350,
+      "options": {
+        "showHeader": true,
+        "sortBy": []
+      },
+      "pageSize": null,
+      "pluginVersion": "7.5.7",
+      "scroll": true,
+      "showHeader": true,
+      "sort": {
+        "col": 0,
+        "desc": true
+      },
+      "styles": [
+        {
+          "alias": "Time",
+          "dateFormat": "YYYY-MM-DD HH:mm:ss",
+          "pattern": "Time",
+          "type": "date"
+        },
+        {
+          "alias": "",
+          "colorMode": null,
+          "colors": [
+            "rgba(245, 54, 54, 0.9)",
+            "rgba(237, 129, 40, 0.89)",
+            "rgba(50, 172, 45, 0.97)"
+          ],
+          "decimals": 2,
+          "pattern": "/.*/",
+          "thresholds": [],
+          "type": "number",
+          "unit": "short"
+        }
+      ],
+      "targets": [
+        {
+          "exemplar": true,
+          "expr": "ticdc_message_server_cur_stream_count",
+          "format": "time_series",
+          "instant": true,
+          "interval": "",
+          "legendFormat": "",
+          "queryType": "randomWalk",
+          "refId": "A"
+        }
+      ],
+      "title": "Stream Count Between Nodes",
+      "transform": "timeseries_to_columns",
+      "transformations": [
+        {
+          "id": "labelsToFields",
+          "options": {
+            "valueLabel": "from"
+          }
+        },
+        {
+          "id": "organize",
+          "options": {
+            "excludeByName": {
+              "Time": true,
+              "instance": false,
+              "job": true
+            },
+            "indexByName": {},
+            "renameByName": {
+              "instance": "dest\\src"
+            }
+          }
+        }
+      ],
+      "type": "table"
+    },
+    {
+      "aliasColors": {},
+      "bars": false,
+      "dashLength": 10,
+      "dashes": false,
+      "datasource": "${DS_CDC-CLUSTER1}",
+      "fieldConfig": {
+        "defaults": {},
+        "overrides": []
+      },
+      "fill": 1,
+      "fillGradient": 0,
+      "gridPos": {
+        "h": 8,
+        "w": 4,
+        "x": 20,
+        "y": 15
+      },
+      "hiddenSeries": false,
+      "id": 358,
+      "legend": {
+        "avg": false,
+        "current": false,
+        "max": false,
+        "min": false,
+        "show": true,
+        "total": false,
+        "values": false
+      },
+      "lines": true,
+      "linewidth": 1,
+      "nullPointMode": "null",
+      "options": {
+        "alertThreshold": true
+      },
+      "percentage": false,
+      "pluginVersion": "7.5.7",
+      "pointradius": 2,
+      "points": false,
+      "renderer": "flot",
+      "seriesOverrides": [],
+      "spaceLength": 10,
+      "stack": false,
+      "steppedLine": false,
+      "targets": [
+        {
+          "exemplar": true,
+          "expr": "sum(ticdc_message_server_cur_stream_count)",
+          "interval": "",
+          "legendFormat": "actual",
+          "queryType": "randomWalk",
+          "refId": "A"
+        },
+        {
+          "exemplar": true,
+          "expr": "2 * count(process_start_time_seconds{tidb_cluster=\"$tidb_cluster\", job=\"ticdc\"}) - 1",
+          "hide": false,
+          "interval": "",
+          "legendFormat": "expected",
+          "refId": "B"
+        }
+      ],
+      "thresholds": [],
+      "timeFrom": null,
+      "timeRegions": [],
+      "timeShift": null,
+      "title": "Total Stream Count",
+      "tooltip": {
+        "shared": true,
+        "sort": 0,
+        "value_type": "individual"
+      },
+      "type": "graph",
+      "xaxis": {
+        "buckets": null,
+        "mode": "time",
+        "name": null,
+        "show": true,
+        "values": []
+      },
+      "yaxes": [
+        {
+          "format": "short",
+          "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": true,
       "gridPos": {
@@ -11011,5 +11783,5 @@
   "timezone": "browser",
   "title": "Test-Cluster-TiCDC",
   "uid": "YiGL8hBZ1",
-  "version": 29
-}
\ No newline at end of file
+  "version": 30
+}
diff --git a/pkg/config/messages.go b/pkg/config/messages.go
index 5450c25e9..541c074c4 100644
--- a/pkg/config/messages.go
+++ b/pkg/config/messages.go
@@ -34,8 +34,8 @@ type MessagesConfig struct {
 
 // read only
 var defaultMessageConfig = &MessagesConfig{
-	ClientMaxBatchInterval:       TomlDuration(time.Millisecond * 100),
-	ClientMaxBatchSize:           8 * 1024, // 8MB
+	ClientMaxBatchInterval:       TomlDuration(time.Millisecond * 200),
+	ClientMaxBatchSize:           8 * 1024 * 1024, // 8MB
 	ClientMaxBatchCount:          128,
 	ClientRetryRateLimit:         1.0, // Once per second
 	ServerMaxPendingMessageCount: 102400,
diff --git a/pkg/p2p/metrics.go b/pkg/p2p/metrics.go
index 6a1cf2ce6..d5a53824b 100644
--- a/pkg/p2p/metrics.go
+++ b/pkg/p2p/metrics.go
@@ -39,8 +39,26 @@ var (
 		Namespace: "ticdc",
 		Subsystem: "message_server",
 		Name:      "message_batch_size",
-		Help:      "size of message batches received",
-		Buckets:   prometheus.LinearBuckets(0, 5, 16),
+		Help:      "size in number of messages of message batches received",
+		Buckets:   prometheus.ExponentialBuckets(1, 2, 10),
+	}, []string{"from"})
+
+	// serverMessageBatchBytesHistogram records the wire sizes as reported by protobuf.
+	serverMessageBatchBytesHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "ticdc",
+		Subsystem: "message_server",
+		Name:      "message_batch_bytes",
+		Help:      "size in bytes of message batches received",
+		Buckets:   prometheus.ExponentialBuckets(8.0, 2, 16),
+	}, []string{"from"})
+
+	// serverMessageBytesHistogram records the wire sizes as reported by protobuf.
+	serverMessageBytesHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "ticdc",
+		Subsystem: "message_server",
+		Name:      "message_bytes",
+		Help:      "size in bytes of messages received",
+		Buckets:   prometheus.ExponentialBuckets(8.0, 2, 16),
 	}, []string{"from"})
 
 	serverAckCount = prometheus.NewCounterVec(prometheus.CounterOpts{
@@ -89,6 +107,8 @@ func InitMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(serverStreamCount)
 	registry.MustRegister(serverMessageCount)
 	registry.MustRegister(serverMessageBatchHistogram)
+	registry.MustRegister(serverMessageBytesHistogram)
+	registry.MustRegister(serverMessageBatchBytesHistogram)
 	registry.MustRegister(serverAckCount)
 	registry.MustRegister(serverRepeatedMessageCount)
 	registry.MustRegister(grpcClientMetrics)
diff --git a/pkg/p2p/server.go b/pkg/p2p/server.go
index 4fdbcfb25..a91f03730 100755
--- a/pkg/p2p/server.go
+++ b/pkg/p2p/server.go
@@ -36,6 +36,10 @@ import (
 	"google.golang.org/grpc/status"
 )
 
+const (
+	messageServerReportsIndividualMessageSize = true
+)
+
 // MessageServerConfig stores configurations for the MessageServer
 type MessageServerConfig struct {
 	// The maximum number of entries to be cached for topics with no handler registered
@@ -530,6 +534,7 @@ func (m *MessageServer) registerPeer(
 
 	log.Info("peer connection received",
 		zap.String("sender-id", streamMeta.SenderId),
+		zap.String("sender-advertise-addr", streamMeta.SenderAdvertisedAddr),
 		zap.String("addr", clientIP),
 		zap.Int64("epoch", streamMeta.Epoch))
 
@@ -697,6 +702,12 @@ func (m *MessageServer) receive(stream p2p.CDCPeerToPeer_SendMessageServer, stre
 	metricsServerMessageBatchHistogram := serverMessageBatchHistogram.With(prometheus.Labels{
 		"from": streamHandle.GetStreamMeta().SenderAdvertisedAddr,
 	})
+	metricsServerMessageBatchBytesHistogram := serverMessageBatchBytesHistogram.With(prometheus.Labels{
+		"from": streamHandle.GetStreamMeta().SenderAdvertisedAddr,
+	})
+	metricsServerMessageBytesHistogram := serverMessageBytesHistogram.With(prometheus.Labels{
+		"from": streamHandle.GetStreamMeta().SenderAdvertisedAddr,
+	})
 
 	for {
 		failpoint.Inject("ServerInjectServerRestart", func() {
@@ -714,10 +725,24 @@ func (m *MessageServer) receive(stream p2p.CDCPeerToPeer_SendMessageServer, stre
 		batchSize := len(packet.GetEntries())
 		log.Debug("received packet", zap.String("streamHandle", streamHandle.GetStreamMeta().SenderId),
 			zap.Int("num-entries", batchSize))
+
+		batchBytes := packet.Size()
+		metricsServerMessageBatchBytesHistogram.Observe(float64(batchBytes))
 		metricsServerMessageBatchHistogram.Observe(float64(batchSize))
+		metricsServerMessageCount.Add(float64(batchSize))
+
+		entries := packet.GetEntries()
+		if batchSize > 0 {
+			if messageServerReportsIndividualMessageSize /* true for now */ {
+				// Note that this can be costly if the number of messages is huge.
+				// However, the current usage of this package in TiCDC should not
+				// cause any problem, as the messages are for metadata only.
+				for _, entry := range entries {
+					messageWireSize := entry.Size()
+					metricsServerMessageBytesHistogram.Observe(float64(messageWireSize))
+				}
+			}
 
-		if len(packet.GetEntries()) > 0 {
-			metricsServerMessageCount.Inc()
 			// See the comment above on why use scheduleTaskBlocking.
 			if err := m.scheduleTaskBlocking(stream.Context(), taskOnMessageBatch{
 				streamMeta:     streamHandle.GetStreamMeta(),


Test patch:
diff --git a/pkg/cmd/server/server_test.go b/pkg/cmd/server/server_test.go
index 1a36d1cc0..52507ae32 100644
--- a/pkg/cmd/server/server_test.go
+++ b/pkg/cmd/server/server_test.go
@@ -198,8 +198,8 @@ func TestParseCfg(t *testing.T) {
 			},
 			// We expect the default configuration here.
 			Messages: &config.MessagesConfig{
-				ClientMaxBatchInterval:       config.TomlDuration(time.Millisecond * 100),
-				ClientMaxBatchSize:           8 * 1024,
+				ClientMaxBatchInterval:       config.TomlDuration(time.Millisecond * 200),
+				ClientMaxBatchSize:           8 * 1024 * 1024,
 				ClientMaxBatchCount:          128,
 				ClientRetryRateLimit:         1.0,
 				ServerMaxPendingMessageCount: 102400,
@@ -474,8 +474,8 @@ cert-allowed-cn = ["dd","ee"]
 			},
 			// We expect the default configuration here.
 			Messages: &config.MessagesConfig{
-				ClientMaxBatchInterval:       config.TomlDuration(time.Millisecond * 100),
-				ClientMaxBatchSize:           8 * 1024,
+				ClientMaxBatchInterval:       config.TomlDuration(time.Millisecond * 200),
+				ClientMaxBatchSize:           8 * 1024 * 1024,
 				ClientMaxBatchCount:          128,
 				ClientRetryRateLimit:         1.0,
 				ServerMaxPendingMessageCount: 102400,
diff --git a/pkg/config/config_test_data.go b/pkg/config/config_test_data.go
index 68a2d0459..f254ed076 100644
--- a/pkg/config/config_test_data.go
+++ b/pkg/config/config_test_data.go
@@ -128,8 +128,8 @@ const (
       "cleanup-speed-limit": 10000
     },
     "messages": {
-      "client-max-batch-interval": 100000000,
-      "client-max-batch-size": 8192,
+      "client-max-batch-interval": 200000000,
+      "client-max-batch-size": 8388608,
       "client-max-batch-count": 128,
       "client-retry-rate-limit": 1,
       "server-max-pending-message-count": 102400,