更新时间：2026-06-23 GMT+08:00
配置网络时延和连通率指标以监控ER网络连通质量
您可以参考以下操作，配置ER网络的时延与连通率监控指标，配置完成后可在云监控服务（Cloud Eye）控制台查看实时监控数据，从而来监控ER的网络连通质量。具体监控数据如下：
- 平均时延（avg_latency ）：单次测试中所有请求响应耗时的平均值，反映网络数据传输速度与链路拥堵程度，衡量网络的"流畅度"。
- 连通率（connectivity_rate ）：测试中成功建立连接并完成数据交互的请求占总请求的比例，反映网络链路的接通能力与连接稳定性，衡量网络的"可用度"。
方案架构
本方案通过企业路由器er-x实现同区域内vpc-web（192.168.0.0/16）与vpc-client（172.16.0.0/16）的跨VPC互通，vpc-web中部署云服务器ecs-web模拟服务端，vpc-client中部署云服务器ecs-client模拟客户端。基于该组网，配置网络链路的平均时延和连通率监控指标，实时监控客户端和服务端的网络通信质量。
网络规划说明
本示例中，同区域VPC互通组网规划如图1所示，将2个VPC接入ER中，组网规划说明如表1所示。
| 资源 | 说明 |
|---|---|
| VPC |
|
| ER | 开启“默认路由表关联”和“默认路由表传播”功能，添加完“虚拟私有云（VPC）”连接，系统会自动执行以下配置：
|
| ECS | 2个ECS分别位于不同的VPC内，VPC中的ECS如果位于不同的安全组，需要在安全组中添加规则放通网络。 |
| 目的地址 | 下一跳 | 路由类型 |
|---|---|---|
| 固定网段：10.0.0.0/8 | 企业路由器 er-x | 静态路由：自定义 |
| 固定网段：172.16.0.0/12 | 企业路由器 er-x | 静态路由：自定义 |
| 固定网段：192.168.0.0/16 | 企业路由器 er-x | 静态路由：自定义 |
| 目的地址 | 下一跳 | 路由类型 |
|---|---|---|
| vpc-web网段：192.168.0.0/16 | VPC1连接：er-attach-web | 传播路由 |
| vpc-client网段：172.16.0.0/16 | VPC2连接：er-attach-client | 传播路由 |
资源规划说明
企业路由器ER、虚拟私有云VPC、弹性云服务器ECS只要位于同一个区域内即可，可用区可以任意选择，不用保持一致。 
以下资源规划详情仅为示例，您可以根据需要自行修改。
| 资源类型 | 资源数量 | 说明 |
|---|---|---|
| VPC | 2 | 两个VPC，一个部署服务端ECS，一个部署客户端ECS，需要接入ER中。
|
| ER | 1 |
|
| ECS | 2 | 两台ECS，一台模拟服务端，一台模拟客户端，本示例如下：
本示例仅部署一台客户端ECS用于演示，实际建议在企业路由器所在的每个可用区（AZ）内分别部署一台客户端ECS，以更全面地监控网络状况。 |
步骤一：获取环境信息
请提前准备步骤3.c中配置所需的环境信息，需要获取的环境信息及获取方法请参见表5。
| 环境 | 样例 | 获取方法 |
|---|---|---|
| 资源所在区域对应的项目，{project_name} | cn-east-x | |
| 资源所在区域对应的项目ID，{project_id} | 15289aca74exxxxxx37dea0315d99 | |
| IAM用户所属账号名 | IAM-er-test | |
| IAM用户名 | er-test | |
| 服务端ECS的私有IP地址 | 192.168.0.164 |
|
| 自定义监听端口 | 8000 | 步骤3.c中设置的监听端口，您可以自定义 |
| ER实例ID | cd711600-xxxx-xxxx-938f-e41eb64f5e25 |
|
步骤二：创建资源
本示例的资源规划详情，请参见表4。
- 在区域A内，创建1个企业路由器。
创建企业路由器，具体方法请参见创建企业路由器。
- 在区域A内，创建2个VPC。
创建VPC及子网，具体方法请参见创建虚拟私有云和子网。
- 在区域A内，创建2个ECS。
本示例中，用作服务端的ecs-web需要绑定EIP，用于部署Nginx服务。
创建ECS，具体方法请参见自定义购买ECS。
步骤三：在企业路由器中添加并配置VPC连接
- 将2个VPC分别接入企业路由器中，即在企业路由器中添加VPC连接。
开启该功能后，会在VPC路由表中自动添加指向ER的路由，目的地址固定为10.0.0.0/8，172.16.0.0/12，192.168.0.0/16。
添加“虚拟私有云（VPC）”连接，具体方法请参见在企业路由器中添加VPC连接。
- 检查ER路由表中指向VPC的路由。
本示例中，ER开启了“默认路由表关联”和“默认路由表传播”功能，那么在ER中添加“虚拟私有云（VPC）”连接时，系统会自动添加ER指向VPC的路由，无需手动添加，只需要检查即可。
ER路由规划详情，请参见和表3。
查看ER路由，具体方法请参见查看路由。
- 登录任意一个ECS，执行以下命令，验证网络通信情况。
弹性云服务器有多种登录方法，具体请参见登录弹性云服务器。
本示例是通过管理控制台远程登录（VNC方式）。
以登录ecs-web为例，执行以下命令，验证两个VPC网络的互通情况。
ping ecs-client的私有IP地址
命令示例：
ping 172.16.0.60
回显如下信息，表示两个VPC网络互通。[root@ecs-web ~]# ping 172.16.0.60 PING 172.16.0.60 (172.16.0.60) 56(84) bytes of data. 64 bytes from 172.16.0.60: icmp_seq=1 ttl=63 time=2.02 ms 64 bytes from 172.16.0.60: icmp_seq=2 ttl=63 time=1.86 ms ^C --- 172.16.0.60 ping statistics --- 2 packets transmitted, 2 received, 0% packet loss, time 1000ms rtt min/avg/max/mdev = 1.859/1.941/2.023/0.082 ms
步骤四：在服务端ECS部署Nginx服务
- 使用root用户，登录服务端ECS（ecs-web）。
弹性云服务器有多种登录方法，具体请参见登录弹性云服务器。
本示例是通过管理控制台远程登录（VNC方式）。
- 执行以下命令，安装Nginx。 回显类似如下信息，表示安装完成。
[root@ecs-web ~]# yum install nginx Last metadata expiration check: 0:24:32 ago on Thu 11 Jun 2026 02:01:16 PM CST. Dependencies resolved. ... Complete!
- 执行以下步骤，配置自定义监听端口。
- 执行以下命令，检查Nginx配置文件语法是否正确。 回显类似如下信息，表示配置正确。
[root@ecs-web ~]# nginx -t nginx: the configuration file /etc/nginx/nginx.conf syntax is ok nginx: configuration file /etc/nginx/nginx.conf test is successful
- 执行以下命令，运行Nginx。
步骤五：在客户端ECS部署配置脚本
- 登录客户端ECS（ecs-client）。
弹性云服务器有多种登录方法，具体请参见登录弹性云服务器。
本示例是通过管理控制台远程登录（VNC方式）。
- 执行以下步骤，创建监控指标自动配置脚本。
- 执行以下命令，创建脚本文件ces.py。
vim ces.py
- 按i进入编辑模式。
- 将配置信息添加至脚本文件ces.py。
import asyncio import sys from urllib.parse import urlparse import requests import time import concurrent.futures import yaml def load_parameter(file_path): with open(file_path, 'r', encoding='utf-8') as file: parameter = yaml.safe_load(file) return parameter def check_connectivity_with_curl(url): parsed_url = urlparse(url) host = parsed_url.hostname port = parsed_url.port or (443 if parsed_url.scheme == "https" else 80) path = parsed_url.path or "/" if parsed_url.query: path += f"?{parsed_url.query}" async def _raw_socket_request(): start_time = time.perf_counter() reader, writer = None, None try: if parsed_url.scheme == "https": import ssl ssl_context = ssl._create_unverified_context() reader, writer = await asyncio.wait_for( asyncio.open_connection(host, port, ssl=ssl_context), timeout=5.0 ) else: reader, writer = await asyncio.wait_for( asyncio.open_connection(host, port), timeout=5.0 ) request_header = ( f"GET {path} HTTP/1.1\r\n" f"Host: {host}\r\n" f"User-Agent: curl/7.29.0\r\n" f"Connection: close\r\n\r\n" ) writer.write(request_header.encode('utf-8')) await writer.drain() response_line = await asyncio.wait_for(reader.readline(), timeout=5.0) duration_ms = (time.perf_counter() - start_time) * 1000 status_code = int(response_line.split()[1]) return status_code, duration_ms except Exception: duration_ms = (time.perf_counter() - start_time) * 1000 return -1, duration_ms finally: if writer: writer.close() try: await writer.wait_closed() except Exception: pass try: loop = asyncio.get_event_loop() except RuntimeError: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) return loop.run_until_complete(_raw_socket_request()) def get_connectivity_stats(url, total_requests=100, max_workers=100): latencies = [] success_count = 0 with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: futures = [executor.submit(check_connectivity_with_curl, url) for _ in range(total_requests)] for future in concurrent.futures.as_completed(futures): status_code, latency = future.result() if status_code == 200: success_count += 1 latencies.append(latency) connectivity_rate = (success_count / total_requests) * 100 avg_latency = sum(latencies) / len(latencies) if latencies else 0 return connectivity_rate, avg_latency def get_connectivity_rate(parameter, rate): return { "metric": { "namespace": parameter['namespace'], "dimensions": [{ "name": parameter['dimensions_name'], "value": parameter['dimensions_value'] }], "metric_name": parameter['metric_name_connectivity_rate'] }, "ttl": int(parameter['ttl']), "collect_time": int(time.time() * 1000), "value": rate, "unit": parameter['unit_connectivity_rate'], "type": parameter['type'] } def get_avg_latency(parameter, latency): return { "metric": { "namespace": parameter['namespace'], "dimensions": [{ "name": parameter['dimensions_name'], "value": parameter['dimensions_value'] }], "metric_name": parameter['metric_name_avg_latency'] }, "ttl": int(parameter['ttl']), "collect_time": int(time.time() * 1000), "value": latency, "unit": parameter['unit_avg_latency'], "type": parameter['type'] } class TokenManager: def __init__(self, parameter): self.token = None self.auth_url = parameter['auth_url'] self.auth_body = {"auth": {"identity": {"methods": ["password"], "password": { "user": {"name": parameter['iam_user'], "domain": {"name": parameter['iam_domain']}, "password": parameter['iam_password']}}}, "scope": {"project": {"name": parameter['iam_project']}}}} self.refresh_token() def refresh_token(self): headers = {'Content-Type': 'application/json'} try: response = requests.post(self.auth_url, json=self.auth_body, headers=headers, verify=False) new_token = response.headers.get('X-Subject-Token') if new_token: self.token = new_token else: print("not found X-Subject-Token") except Exception as e: print(f"get Token failed: {e}") def create_metric_data(create_metric_data_url, token, data): headers = {'X-Auth-Token': token} try: res = requests.post(create_metric_data_url, json=data, headers=headers, verify=False, timeout=5) return res.status_code except Exception as e: print(f"error: {e}") def main_job(parameter, token_manager): while True: rate, latency = get_connectivity_stats(parameter['target_url'], int(parameter['total_requests']), int(parameter['max_workers'])) data = [get_connectivity_rate(parameter, rate), get_avg_latency(parameter, latency)] for i in range(3): status = create_metric_data(parameter['create_metric_data_url'], token_manager.token, data) if status == 401: token_manager.refresh_token() continue break time.sleep(10) if __name__ == "__main__": parameter = load_parameter(sys.argv[1].strip()) parameter['iam_password'] = sys.argv[2].strip() manager = TokenManager(parameter) main_job(parameter, manager) - 按ESC退出编辑，并输入:wq!保存配置。
- 执行以下命令，创建脚本文件ces.py。
- 执行以下步骤，添加脚本运行所需的环境信息。
- 执行以下命令，创建参数文件parameter.yml。
vim parameter.yml
- 按i进入编辑模式。
- 将配置信息添加至参数文件parameter.yml。 请仔细约束注释信息，将示例替换成步骤一：获取环境信息中获取的环境信息。
# token获取url，请将{project_name}替换成资源所在区域对应的项目 auth_url: https://iam.{project_name}.myhuaweicloud.com/v3/auth/tokens # 上报监控数据url，请将{project_name}和{project_id}替换成资源所在区域对应的项目及项目ID create_metric_data_url: https://ces.{project_name}.myhuaweicloud.com/V1.0/{project_id}/metric-data # IAM用户所属账号名，以下仅为示例，请替换 iam_domain: IAM-er-test # IAM用户名，以下仅为示例，请替换 iam_user: er-test # IAM用户所属账号的项目，即资源所在区域对应的项目，请替换 iam_project: {project_name} # 服务端url，以下仅为示例，请将IP地址替换成服务端ECS的私有IP地址 ，端口替换成自定义监听端口 target_url: http://172.16.0.245:8000 # 单次测试发送的请求数量，推荐100或者根据实际情况配置 total_requests: 100 # 单次测试的最大并发数，推荐100或者根据实际情况配置 max_workers: 100 # 服务命名空间，格式为service.item，service和item以点号拼接组成。 # service和item必须以字母（大写或小写）开头，后面可以跟零个或多个字母（大写或小写）、数字、下划线（_），长度为[3,32]个字符 # 样例：TEST.ER namespace: "TEST.ER" # 数据的有效期，超出该有效期则自动删除该数据，单位秒。大小为[1,604800]的整数。 ttl: 172800 # 指标资源维度名称，可固定为er_instance_id dimensions_name: er_instance_id # 指标资源维度值，即ER的实例ID，以下仅为示例，请替换 dimensions_value: eb97488a-070c-4db8-a719-ee5e66490e8d # 连通率指标ID，可固定为connectivity_rate metric_name_connectivity_rate: connectivity_rate # 时延指标ID，可固定为avg_latency metric_name_avg_latency: avg_latency # 连通率指标单位，可固定为% unit_connectivity_rate: "%" # 时延指标单位，可固定为ms unit_avg_latency: ms # 指标数据类型，可固定为float type: float - 按ESC退出编辑，并输入:wq!保存配置。
- 执行以下命令，创建参数文件parameter.yml。
- 执行以下命令，运行ces.py和parameter.yml。
nohup python3 ces.py parameter.yml 密码 &
请将命令中的密码替换成当前账号对应的实际密码。
- 脚本运行完成后，执行以下步骤，可在云监控服务控制台查看监控指标。
父主题： 使用CES服务监控ER网络指标
