第316篇：Telemetry 订阅与流式数据推送

关键词：Telemetry、gRPC Dial-out、流式推送、思科 Telemetry、华为 iTelemetry、指标采集、实时监控

一、引言

第315篇介绍了 gNMI 协议的双向流式能力。在网络运维中，最核心的需求之一是实时采集设备指标——接口流量、CPU 利用率、内存使用、温度、光功率等。传统的 SNMP 轮询（Polling）方式存在固有缺陷：固定的轮询间隔导致要么数据不够及时，要么设备过载。

Telemetry（遥测）是一种"推送式"数据采集模型：设备主动将变更的或周期性采样的数据推送到采集服务器，采集方无需轮询。这带来了更低的延迟、更高的效率和更好的可扩展性。

二、核心概念

2.1 轮询（Pull）vs 推送（Push）模型

2.2 Telemetry 主流实现

2.3 Telemetry 数据流架构

三、技术细节

3.1 华为 iTelemetry 设备侧配置

华为设备的 iTelemetry 需要在设备侧配置 gRPC Dial-out 和 sensor group：

# 华为设备 iTelemetry 配置示例

# 1. 创建传感器组（定义采集路径和频率）
sensor-group name InterfaceTraffic
  sensor-path path /junos:interfaces/interface/administration/admin-status
  sensor-path path /junos:interfaces/interface/oper-admin/oper-status
  sensor-path path /junos:interfaces/interface/if-counters/input-pkts
  sensor-path path /junos:interfaces/interface/if-counters/output-pkts
  sensor-path path /junos:interfaces/interface/if-counters/input-bytes
  sensor-path path /junos:interfaces/interface/if-counters/output-bytes
  sensor-path path /junos:interfaces/interface/if-counters/input-errors
  sensor-path path /junos:interfaces/interface/if-counters/output-errors
  sampling-interval 1000    # 采样间隔 1000ms
  sensor-group-end

# 2. 创建流编码类型
telemetry system
  grpc
    vpn naming-instance public
    destination-ip 10.100.1.100 vrf-name public port 50051
      no tls
    destination-end
  grpc-end
system-end

# 3. 绑定传感器组到流
telemetry sensor-group InterfaceTraffic
  streaming-telemetry destination 10.100.1.100
    encoding-type dpb         # 或使用 json / jp
    sample-interval 1000
  streaming-telemetry-end
sensor-group-end

3.2 Python Telemetry 接收服务器

3.2.1 基于 gRPC 的 gNMI Subscribe 接收器

# scripts/telemetry_gnmi_receiver.py
import grpc
from concurrent import futures
import json
import time
from datetime import datetime

import gnmi.gnmi_pb2 as gnmi_pb2
import gnmi.gnmi_pb2_grpc as gnmi_grpc

class TelemetryReceiver(gnmi_grpc.gNMIServicer):
    """gNMI Telemetry 接收服务器"""

    def __init__(self):
        self.connections = {}
        self.metrics_store = {}
        self.start_time = time.time()

    def Subscribe(self, request_iterator, context):
        """处理 gNMI Subscribe 流"""
        # 获取客户端地址
        client_addr = context.peer()
        conn_id = f"{client_addr}_{int(time.time())}"
        self.connections[conn_id] = {
            "start_time": time.time(),
            "notifications": 0,
            "updates": 0,
        }

        print(f"[{datetime.now()}] 新订阅连接: {conn_id}")

        try:
            for request in request_iterator:
                if request.HasField("subscribe"):
                    sub = request.subscribe
                    print(f"订阅路径:")
                    for s in sub.subscription:
                        path = "/".join(s.path.element)
                        mode_map = {
                            0: "DEFAULT",
                            1: "STREAM",
                            2: "ONCE",
                            3: "ON_CHANGE",
                        }
                        print(f"  {path} (mode={mode_map.get(s.mode, '?')})")

                elif request.HasField("poll"):
                    print("收到 Poll 请求")

                elif request.HasField("suppress_update"):
                    print(f"抑制更新阈值: {request.suppress_update.min_interval_ms}ms")

        except grpc.RpcError as e:
            print(f"连接 {conn_id} 断开: {e}")

        finally:
            stats = self.connections.get(conn_id, {})
            print(f"连接 {conn_id} 统计: "
                  f"通知={stats.get('notifications', 0)}, "
                  f"更新={stats.get('updates', 0)}")
            del self.connections[conn_id]

    def Get(self, request, context):
        """处理 Get 请求"""
        return gnmi_pb2.GetResponse()

    def Set(self, request, context):
        """处理 Set 请求"""
        return gnmi_pb2.SetResponse()


def serve(port=50051):
    """启动 gNMI 接收服务器"""
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    receiver = TelemetryReceiver()
    gnmi_grpc.add_gNMIServicer_to_server(receiver, server)
    server.add_insecure_port(f"[::]:{port}")
    server.start()
    print(f"gNMI Telemetry 接收服务器启动，端口: {port}")
    server.wait_for_termination()

3.2.2 华为 iTelemetry 接收器（GPB/JSON 格式）

# scripts/telemetry_huawei_receiver.py
import grpc
from concurrent import futures
import json
import time
from datetime import datetime
import struct

class HuaweiTelemetryReceiver(grpc.Servicer):
    """华为 iTelemetry gRPC Dial-out 接收器"""

    def __init__(self):
        self.metrics_store = {}
        self.stats = {
            "connections": 0,
            "total_messages": 0,
            "start_time": datetime.now().isoformat(),
        }

    def Telemetry(self, request_iterator, context):
        """处理华为 iTelemetry gRPC 流"""
        self.stats["connections"] += 1
        conn_start = time.time()
        msg_count = 0

        print(f"[{datetime.now()}] 华为 Telemetry 连接建立")

        try:
            for message in request_iterator:
                msg_count += 1
                self.stats["total_messages"] += 1

                # 解析消息（华为 Telemetry 使用自定义消息格式）
                try:
                    # 华为 Telemetry 消息结构:
                    # header (4 bytes version + 4 bytes message type + ...)
                    # payload (JSON 或 GPB 编码的指标数据)

                    if isinstance(message, bytes):
                        # 尝试 JSON 解码
                        try:
                            data = json.loads(message.decode("utf-8"))
                        except json.JSONDecodeError:
                            # GPB 编码，需使用华为的 GPB 解析库
                            data = {"raw_bytes": len(message),
                                   "encoding": "GPB"}

                        # 提取指标
                        self._process_telemetry_data(data)

                except Exception as e:
                    print(f"  解析错误: {e}")

                # 定期输出统计
                if msg_count % 100 == 0:
                    elapsed = time.time() - conn_start
                    rate = msg_count / elapsed if elapsed > 0 else 0
                    print(f"  已接收 {msg_count} 条消息 "
                          f"(速率: {rate:.1f} msg/s)")

        except Exception as e:
            print(f"连接断开: {e}")
        finally:
            elapsed = time.time() - conn_start
            print(f"连接结束: {msg_count} 条消息, "
                  f"运行 {elapsed:.0f}s")

    def _process_telemetry_data(self, data):
        """处理解析后的 Telemetry 数据"""
        if "raw_bytes" in data:
            return

        # 提取设备信息
        device_info = data.get("device_info", {})
        timestamp = data.get("timestamp", 0)

        # 提取指标数据
        sensor_data = data.get("sensor_data_list", [])
        for sensor in sensor_data:
            path = sensor.get("sensor_path", "")
            values = sensor.get("value", [])

            # 存储到指标库
            key = f"{path}"
            self.metrics_store[key] = {
                "values": values,
                "timestamp": timestamp,
                "last_update": datetime.now().isoformat(),
            }


def serve_huawei_telemetry(port=57000):
    """启动华为 Telemetry 接收服务器"""
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))

    # 华为 Telemetry 使用通用的 gRPC stream 方法
    receiver = HuaweiTelemetryReceiver()

    # 注册 unary-stream 处理
    # 注意：华为 Telemetry 的具体 service/method 名取决于设备版本
    server.add_insecure_port(f"[::]:{port}")
    server.start()
    print(f"华为 Telemetry 接收服务器启动，端口: {port}")
    server.wait_for_termination()

3.3 思科 Telemetry 接收器

# scripts/telemetry_cisco_receiver.py
import grpc
from concurrent import futures
import json
from datetime import datetime

class CiscoTelemetryReceiver:
    """思科 gRPC Telemetry 接收器"""

    def __init__(self):
        self.metrics = {}
        self.alerts = []

    def receive_stream(self, request_iterator, context):
        """处理思科 Telemetry 流"""
        print(f"[{datetime.now()}] 思科 Telemetry 连接")

        for message in request_iterator:
            # 思科 Telemetry 使用 gRPC 流式传输
            # 消息格式为 GPB 编码
            self._process_cisco_message(message)

    def _process_cisco_message(self, message):
        """解析思科 Telemetry 消息"""
        # 思科 Telemetry GPB 消息结构:
        # telemetry_tailf_CdpVrfHeader (路由上下文)
        #   -> updates[]
        #     -> sensor_path (路径)
        #     -> values[] (指标值)

        # 提取 VRF 信息
        vrf = getattr(message, "vrf_name", "default")

        # 提取更新
        for update in message.updates:
            sensor_path = update.sensor_path
            path_str = "/".join(sensor_path.path) if hasattr(sensor_path, 'path') else str(sensor_path)

            # 提取值
            values = []
            for val in update.values:
                if val.IsInitialized() and hasattr(val, 'dpb64'):
                    values.append(val.dpb64)
                elif hasattr(val, 'dpstr'):
                    values.append(val.dpstr)
                elif hasattr(val, 'dpbool'):
                    values.append(val.dpbool)

            if values:
                self.metrics[path_str] = {
                    "vrf": vrf,
                    "values": values,
                    "timestamp": datetime.now().isoformat(),
                }

                # 阈值告警
                self._check_threshold(path_str, values)

    def _check_threshold(self, path, values):
        """检查阈值告警"""
        # 接口利用率 > 80%
        if "output-bytes" in path and values:
            # 计算利用率（需要与上一次值比较）
            pass


def serve_cisco_telemetry(port=57000):
    """启动思科 Telemetry 接收服务器"""
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))

    # 思科 Telemetry service 定义
    # service telemetry {
    #   rpc TailfEvents (stream telemetry_tnc_CdpVrfStream) returns ();
    # }

    receiver = CiscoTelemetryReceiver()
    server.add_insecure_port(f"[::]:{port}")
    server.start()
    print(f"思科 Telemetry 接收服务器启动，端口: {port}")
    server.wait_for_termination()

3.4 Telemetry 数据接入时序数据库

# scripts/telemetry_to_influxdb.py
from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS
import json

class TelemetryToInfluxDB:
    """Telemetry 数据写入 InfluxDB"""

    def __init__(self, url="http://localhost:8086",
                 token="my-token", org="my-org", bucket="telemetry"):
        self.client = InfluxDBClient(url=url, token=token, org=org)
        self.write_api = self.client.write_api(
            write_options=SYNCHRONOUS
        )
        self.bucket = bucket
        self.org = org

    def write_metrics(self, device_ip, metrics_data):
        """将 Telemetry 指标写入 InfluxDB"""
        points = []

        for path, data in metrics_data.items():
            # 将路径转换为 InfluxDB 的 measurement 和 tag
            parts = path.strip("/").split("/")
            measurement = parts[0] if parts else "telemetry"

            # 构建 tag（设备IP、接口名等）
            tags = {
                "device_ip": device_ip,
                "path": path,
            }

            # 提取接口名作为 tag
            for i, part in enumerate(parts):
                if part == "interface" and i + 1 < len(parts):
                    tags["interface"] = parts[i + 1]

            # 构建 field
            fields = {}
            for val in data.get("values", []):
                if isinstance(val, (int, float)):
                    fields["value"] = val

            if fields:
                point = Point(measurement).tags(**tags).fields(**fields)
                points.append(point)

        if points:
            self.write_api.write(bucket=self.bucket, org=self.org, record=points)
            print(f"写入 {len(points)} 个指标点到 InfluxDB")

    def close(self):
        self.write_api.close()
        self.client.close()


# 使用示例
influx = TelemetryToInfluxDB()

# 模拟接收到的 Telemetry 数据
simulated_metrics = {
    "/interfaces/interface[name=GE1/0/1]/state/counters/in-octets": {
        "values": [1234567890],
        "timestamp": 1234567890000000000,
    },
    "/interfaces/interface[name=GE1/0/1]/state/counters/out-octets": {
        "values": [9876543210],
        "timestamp": 1234567890000000000,
    },
}

influx.write_metrics("10.1.1.1", simulated_metrics)
influx.close()

3.5 Telemetry 数据可视化：Grafana Dashboard

# scripts/telemetry_dashboard_gen.py
import json

def generate_grafana_dashboard():
    """生成 Grafana Dashboard JSON"""
    dashboard = {
        "dashboard": {
            "title": "Network Telemetry Dashboard",
            "timezone": "browser",
            "panels": [
                {
                    "title": "Interface Traffic (In/Out)",
                    "type": "graph",
                    "gridPos": {"h": 8, "w": 12, "x": 0, "y": 0},
                    "targets": [
                        {
                            "datasource": "InfluxDB",
                            "query": (
                                'from(bucket: "telemetry")'
                                '  |> range(start: -1h)'
                                '  |> filter(fn: (r) => r._measurement == "in-octets")'
                                '  |> filter(fn: (r) => r.interface == "GE1/0/1")'
                                '  |> aggregateWindow(every: 1m, fn: sum)'
                            ),
                            "legendFormat": "In Octets",
                        },
                        {
                            "datasource": "InfluxDB",
                            "query": (
                                'from(bucket: "telemetry")'
                                '  |> range(start: -1h)'
                                '  |> filter(fn: (r) => r._measurement == "out-octets")'
                                '  |> filter(fn: (r) => r.interface == "GE1/0/1")'
                                '  |> aggregateWindow(every: 1m, fn: sum)'
                            ),
                            "legendFormat": "Out Octets",
                        },
                    ],
                    "overrides": [],
                },
                {
                    "title": "Interface Error Rate",
                    "type": "gauge",
                    "gridPos": {"h": 4, "w": 6, "x": 12, "y": 0},
                    "targets": [
                        {
                            "datasource": "InfluxDB",
                            "query": (
                                'from(bucket: "telemetry")'
                                '  |> range(start: -5m)'
                                '  |> filter(fn: (r) => r._measurement == "in-errors")'
                                '  |> sum()'
                            ),
                        },
                    ],
                },
                {
                    "title": "CPU Utilization",
                    "type": "gauge",
                    "gridPos": {"h": 4, "w": 6, "x": 12, "y": 4},
                    "targets": [
                        {
                            "datasource": "InfluxDB",
                            "query": (
                                'from(bucket: "telemetry")'
                                '  |> range(start: -1m)'
                                '  |> filter(fn: (r) => r._measurement == "cpu")'
                                '  |> last()'
                            ),
                        },
                    ],
                    "fieldConfig": {
                        "defaults": {
                            "max": 100,
                            "thresholds": {
                                "steps": [
                                    {"color": "green", "value": None},
                                    {"color": "yellow", "value": 70},
                                    {"color": "red", "value": 90},
                                ]
                            },
                        }
                    },
                },
            ],
            "templating": {
                "list": [
                    {
                        "name": "device_ip",
                        "type": "query",
                        "query": (
                            'import "influxdata/influxdb/schema"'
                            'schema.tagValues(bucket: "telemetry", tag: "device_ip")'
                        ),
                        "datasource": "InfluxDB",
                    }
                ]
            },
        }
    }

    with open("telemetry_dashboard.json", "w", encoding="utf-8") as f:
        json.dump(dashboard, f, indent=2, ensure_ascii=False)

    print("Grafana Dashboard JSON 已生成")

generate_grafana_dashboard()

四、实战配置/案例分析

4.1 案例一：全网接口流量实时监控平台

场景：构建一个能实时采集全网 500 台设备接口流量并可视化的平台。

# scripts/full_telemetry_platform.py
import grpc
from concurrent import futures
import json
import threading
from datetime import datetime
from influxdb_client import InfluxDBClient, Point

class TelemetryPlatform:
    """完整的 Telemetry 监控平台"""

    def __init__(self, grpc_port=50051, influx_url="http://localhost:8086"):
        self.grpc_port = grpc_port
        self.connections = {}
        self.alert_rules = {
            "error_rate_threshold": 100,   # 错误包 > 100 触发告警
            "utilization_threshold": 80,    # 利用率 > 80% 触发告警
        }
        self.alerts = []
        self.lock = threading.Lock()

        # InfluxDB
        self.influx = InfluxDBClient(url=influx_url, token="my-token", org="my-org")
        self.write_api = self.influx.write_api()

    def handle_telemetry_stream(self, request_iterator, context):
        """处理 Telemetry 流"""
        client_addr = context.peer()
        conn_id = f"{client_addr}_{datetime.now().timestamp()}"
        self.connections[conn_id] = {"start": datetime.now(), "count": 0}

        for message in request_iterator:
            self.connections[conn_id]["count"] += 1
            self._process_and_store(message, conn_id)

    def _process_and_store(self, message, conn_id):
        """处理并存储指标"""
        # 解析指标（具体格式取决于设备厂商）
        # 写入时序数据库
        # 触发告警检查
        pass

    def start(self):
        """启动平台"""
        server = grpc.server(futures.ThreadPoolExecutor(max_workers=50))
        # 注册 service
        server.add_insecure_port(f"[::]:{self.grpc_port}")
        server.start()

        print(f"Telemetry 平台启动")
        print(f"  gRPC 端口: {self.grpc_port}")
        print(f"  告警规则: {self.alert_rules}")
        server.wait_for_termination()

# 启动
platform = TelemetryPlatform()
platform.start()

4.2 案例二：故障案例——Telemetry 数据乱序与丢失

故障现象：Telemetry 推送的接口流量指标出现乱序，部分数据点丢失，导致 Grafana 面板显示断线。

排查过程：

# scripts/telemetry_quality_check.py
import time

def check_telemetry_quality(metrics_stream):
    """检查 Telemetry 数据质量"""
    last_timestamp = 0
    out_of_order = 0
    gaps = []
    duplicates = 0
    last_value = None

    for metric in metrics_stream:
        ts = metric.get("timestamp", 0)
        value = metric.get("value")

        # 检查乱序
        if ts < last_timestamp:
            out_of_order += 1
            print(f"乱序: {ts} < {last_timestamp}")

        # 检查时间间隔（正常应为 1000ms）
        if last_timestamp > 0:
            interval = ts - last_timestamp
            if abs(interval - 1000) > 500:  # 偏差 > 500ms
                gaps.append({
                    "expected": 1000,
                    "actual": interval,
                    "timestamp": ts,
                })

        # 检查重复
        if value == last_value and ts == last_timestamp:
            duplicates += 1

        last_timestamp = ts
        last_value = value

    return {
        "out_of_order_count": out_of_order,
        "gap_count": len(gaps),
        "duplicates": duplicates,
        "gaps": gaps[:10],  # 前10个
    }

根因分析：

四（续）、高级主题

4.3 Telemetry 订阅管理与生命周期

# scripts/telemetry_subscription_manager.py
import grpc
from gnmi.gnmi import gNMI
from gnmi.proto import gnmi_pb2
import time

class TelemetrySubscriptionManager:
    """Telemetry 订阅管理器"""

    def __init__(self, target_host, target_port=57400, username="admin", password="Pass123"):
        self.target = f"{target_host}:{target_port}"
        self.creds = grpc.ssl_channel_credentials()
        self.channel = grpc.secure_channel(self.target, self.creds)
        self.stub = gNMI.GNMISTub(self.channel)

    def subscribe_once(self, paths, prefix=None):
        """一次性订阅：获取当前值后立即断开"""
        subscribe_request = gnmi_pb2.SubscribeRequest()
        subscribe_request.subscribe.prefix = prefix or gnmi_pb2.Path()
        for p in paths:
            elem = subscribe_request.subscribe.subscribed_paths.add()
            elem.path.element.extend(p.split("/"))
            elem.mode = gnmi_pb2.SubscriptionMode.ONCE

        results = []
        for response in self.stub.Subscribe(subscribe_request):
            if response.HasField("subscription"):
                for notification in response.subscription.notifications:
                    for update in notification.updates:
                        path_str = "/".join(update.path.element)
                        values = []
                        for val in update.val.val:
                            if val.HasField("string_val"):
                                values.append(val.string_val)
                            elif val.HasField("int_val"):
                                values.append(val.int_val)
                        results.append({
                            "path": path_str,
                            "timestamp": notification.timestamp,
                            "values": values
                        })
        return results

    def subscribe_stream(self, paths, sample_interval_ms=1000,
                          suppress_redundant=True):
        """流式订阅：持续接收指标推送"""
        subscribe_request = gnmi_pb2.SubscribeRequest()
        subscribe_request.subscribe.mode = gnmi_pb2.SubscriptionMode.STREAM

        for p in paths:
            elem = subscribe_request.subscribe.subscribed_paths.add()
            elem.path.element.extend(p.split("/"))
            elem.mode = gnmi_pb2.SubscriptionMode.SAMPLE
            elem.sample_interval = sample_interval_ms

        if suppress_redundant:
            subscribe_request.suppress_redundant = True
            subscribe_request.heartbeat_interval = 30000  # 30s 心跳

        # 返回生成器，调用方迭代消费
        for response in self.stub.Subscribe(subscribe_request):
            if response.HasField("subscription"):
                yield response.subscription.notifications

    def poll_once(self, paths):
        """Poll 模式：按需拉取"""
        subscribe_request = gnmi_pb2.SubscribeRequest()
        subscribe_request.poll.SetInParent()  # 声明支持 Poll

        # 先建立订阅
        subscribe_request.subscribe.mode = gnmi_pb2.SubscriptionMode.STREAM
        for p in paths:
            elem = subscribe_request.subscribe.subscribed_paths.add()
            elem.path.element.extend(p.split("/"))
            elem.mode = gnmi_pb2.SubscriptionMode.SAMPLE

        for response in self.stub.Subscribe(subscribe_request):
            if response.HasField("poll"):
                # 收到 poll 就绪，发送 poll 请求
                poll_request = gnmi_pb2.SubscribeRequest()
                poll_request.poll.SetInParent()
                yield poll_request
                break
            elif response.HasField("subscription"):
                yield response.subscription.notifications

4.4 Telemetry 指标计算与告警引擎

# scripts/telemetry_alert_engine.py
import time
from collections import deque
from dataclasses import dataclass, field

@dataclass
class AlertRule:
    """告警规则"""
    name: str
    metric_path: str
    condition: str  # "gt", "lt", "gte", "lte"
    threshold: float
    duration: int  # 持续 N 秒才触发（去抖动）
    severity: str  # "warning", "critical"
    window: deque = field(default_factory=lambda: deque(maxlen=60))

    def evaluate(self, value, timestamp):
        """评估当前值是否触发告警"""
        self.window.append((timestamp, value))

        # 检查持续时间内的所有值
        now = time.time()
        triggered_count = 0
        for ts, v in self.window:
            if now - ts <= self.duration:
                if self._check_condition(v):
                    triggered_count += 1

        # 只有持续超过 duration 秒才触发
        return triggered_count >= (self.duration / max(self.window_interval, 1))

    def _check_condition(self, value):
        conditions = {
            "gt": lambda v, t: v > t,
            "lt": lambda v, t: v < t,
            "gte": lambda v, t: v >= t,
            "lte": lambda v, t: v <= t,
        }
        return conditions.get(self.condition, lambda v, t: False)(value, self.threshold)


class TelemetryAlertEngine:
    """Telemetry 告警引擎"""

    def __init__(self):
        self.rules = {}
        self.active_alerts = {}

    def add_rule(self, rule: AlertRule):
        self.rules[rule.name] = rule

    def process_metric(self, device_ip, path, value, timestamp):
        """处理单个指标"""
        for name, rule in self.rules.items():
            if rule.metric_path == path:
                is_alert = rule.evaluate(value, timestamp)
                alert_key = f"{device_ip}:{name}"

                if is_alert and alert_key not in self.active_alerts:
                    # 新告警
                    alert = {
                        "device": device_ip,
                        "rule": name,
                        "metric": path,
                        "value": value,
                        "threshold": rule.threshold,
                        "severity": rule.severity,
                        "timestamp": timestamp,
                        "message": f"{name}: {path} = {value} ({rule.condition} {rule.threshold})"
                    }
                    self.active_alerts[alert_key] = alert
                    print(f"🔴 告警触发: {alert['message']}")
                elif not is_alert and alert_key in self.active_alerts:
                    # 告警恢复
                    del self.active_alerts[alert_key]
                    print(f"🟢 告警恢复: {name} on {device_ip}")

    def get_active_alerts(self):
        return list(self.active_alerts.values())

# 使用示例
engine = TelemetryAlertEngine()

# 添加告警规则
engine.add_rule(AlertRule(
    name="HighTraffic",
    metric_path="/interfaces/interface/state/counters/out-octets",
    condition="gt",
    threshold=800000000,  # 800 Mbps
    duration=30,  # 持续 30 秒
    severity="warning"
))

engine.add_rule(AlertRule(
    name="HighErrorRate",
    metric_path="/interfaces/interface/state/counters/in-errors",
    condition="gt",
    threshold=100,
    duration=10,
    severity="critical"
))

# 模拟指标推送
engine.process_metric("10.1.1.1", "/interfaces/interface/state/counters/out-octets", 900000000, time.time())
active = engine.get_active_alerts()
print(f"当前活动告警: {len(active)}")

4.5 Telemetry 设备侧配置速查

华为设备完整配置示例：

# 华为 iTelemetry 设备侧配置
telemetry iTelemetry
 sensor-group group1
  sensor-path /junos-interfaces/interfaces/interface/admin-state
  sensor-path /junos-interfaces/interfaces/interface/oper-state
  sensor-path /junos-interfaces/interfaces/interface/statistics
  interval 1000    # 1秒采样
  !
!
 subscription sub1
  sensor-group group1
  stream sample
  !
  destination-group dest1
   !
  !
!
 destination-group dest1
  grpc
   destination 10.255.100.10
    vrf management
    no verify-ssl
    port 50051
   !
   encoding gpbc
  !
 !

五、总结

思考题

1. Telemetry 推送模型中，如果设备到采集器的网络连接中断，已采集但未推送的数据会如何处理？
2. 当 500 台设备同时以 1 秒间隔推送接口流量数据时，采集器每秒需要处理多少数据？如何设计接收器的性能？
3. ON_CHANGE 模式和固定间隔采样模式各适合什么场景？能否组合使用？

下篇预告：第317篇：模型驱动编程（MDA）在华为设备上的实现