1. 首页
  2. 文章资讯
  3. AI技术
  4. AI应用开发

P4 可编程数据平面工程实战：从 INT 遥测到 In-Network Computing

3 阅读 0 评论
P4 可编程数据平面工程实战

P4 可编程数据平面工程实战：从 INT 遥测到 In-Network Computing

当 eBPF/XDP 在操作系统内核层面实现了可编程数据平面后，网络行业在另一个层面——交换机 ASIC 级别——用 P4 语言掀起了更深层次的可编程革命。本文深入解析 P4 语言的核心抽象模型，并通过 In-band Network Telemetry (INT) 的完整工程实现，展示如何构建 Production-Grade 的网络可观测性基础设施。

一、为什么需要 P4：固定功能 ASIC 的天花板

传统交换机 ASIC 遵循"固定功能流水线"架构：芯片出厂时解析哪些协议、在哪些阶段做哪些查表、在哪些字段上做哪些操作，全部由硬件逻辑固化。这意味着——当网络协议演进时（如 VXLAN 到 GENEVE、INT 头部的引入），运营商不得不等待 18-24 个月的芯片迭代周期。

P4（Programming Protocol-independent Packet Processors）的本质是在 ASIC 层面引入了一种领域特定的编程语言，将数据平面的三个核心要素开放给开发者：

  1. 协议解析（Parser）：自定义可以识别哪些头部格式
  2. 匹配-动作表（Match-Action Table）：自定义转发逻辑
  3. 逆解析（Deparser）：自定义报文出站时的封装顺序

这与 eBPF 在内核层面的可编程形成了完美的互补：eBPF 处理单节点内部的灵活数据面逻辑，P4 在交换机级别定义跨越整个网络的流量行为。

1.1 架构对比

维度 传统 ASIC P4 可编程 ASIC eBPF/XDP
协议解析 固定 开发者自定义 通过 BPF 辅助函数间接访问
匹配-动作 固定表项结构 完全可编程 BPF map 间接实现
状态存储 固定寄存器 寄存器/计数器/计量器 BPF map
单节点延迟 纳秒级 纳秒级 微秒级
部署位置 交换机芯片 交换机/NIC/DPU Linux 内核

1.2 从 RMT 到 P4 的映射

P4 编译器的后端目标通常是 Reconfigurable Match Table (RMT) 架构。RMT 的核心是一个多级流水线：

输入报文
    │
    ▼
┌─────────┐   ┌─────────────┐   ┌──────────┐   ┌─────────┐
│ Parser  │──▶│ Ingress MA  │───▶│ Egress MA│──▶│Deparser │
│ 解析器  │   │ 入口匹配动作 │   │ 出口匹配  │   │ 逆解析  │
└─────────┘   └─────────────┘   └──────────┘   └─────────┘
                     │                 │
                 ┌───┴───┐        ┌───┴───┐
                 │Transition│      │Transition│
                 │ 状态转移 │       │ 状态转移 │
                 └─────────┘       └─────────┘

每一级 Match-Action 阶段，P4 编译器将 table 查找映射到 TCAM/SRAM 的 SRAM 条目，将 action 内的语句映射到 Arithmetic Logic Unit (ALU) 的运算序列。

二、P4₁₆ 语言核心模型

P4₁₆（P4 2016 版规范，当前最广泛使用的版本）的编程范式与常规编程语言有本质区别——你在编写的是一个协议处理流水线，而非一个计算程序。

2.1 Header 定义：协议即数据结构

// Ethernet II 头部
header ethernet_t {
    bit<48> dst_addr;
    bit<48> src_addr;
    bit<16> ether_type;
}

// IPv4 头部
header ipv4_t {
    bit<4>  version;
    bit<4>  ihl;
    bit<8>  diffserv;
    bit<16> total_len;
    bit<16> identification;
    bit<3>  flags;
    bit<13> frag_offset;
    bit<8>  ttl;
    bit<8>  protocol;
    bit<16> hdr_checksum;
    bit<32> src_addr;
    bit<32> dst_addr;
}

// UDP 头部
header udp_t {
    bit<16> src_port;
    bit<16> dst_port;
    bit<16> length;
    bit<16> checksum;
}

// VXLAN 头部
header vxlan_t {
    bit<8>  flags;
    bit<24> reserved1;
    bit<24> vni;      // 24-bit VXLAN Network Identifier
    bit<8>  reserved2;
}

// INT Shim 头部
header int_shim_t {
    bit<8>  type;     // INT类型: 1=INT-MD (hop-by-hop)
    bit<8>  length;   // INT头部总长度（不含shim和MD）
    bit<6>  flags;
    bit<2>  reserved;
    bit<8>  instruction_cnt; // INT指令bitmap中1的个数
    bit<16> reserved2;
}

P4 header 定义的约束非常严格：每个字段的位宽必须显式声明，且不允许有运行时动态长度的字段（与 Rust enum 的核心区别）。所有"可变长度"由 Parser 的 extract 逻辑配合状态机表达。

2.2 Parser：确定性有限自动机

P4 的 Parser 本质上是一个 DFA（确定性有限自动机）：

parser MyParser(
    packet_in pkt,
    out headers hdr,
    inout metadata meta,
    inout standard_metadata_t std_meta
) {
    state start {
        pkt.extract(hdr.ethernet);
        transition select(hdr.ethernet.ether_type) {
            0x0800: parse_ipv4;   // IPv4
            0x86dd: parse_ipv6;   // IPv6
            0x8100: parse_vlan;   // 802.1Q VLAN
            default: accept;
        }
    }

    state parse_ipv4 {
        pkt.extract(hdr.ipv4);
        transition select(hdr.ipv4.protocol) {
            0x11: parse_udp;      // UDP = 17
            0x06: parse_tcp;      // TCP = 6
            default: accept;
        }
    }

    state parse_udp {
        pkt.extract(hdr.udp);
        transition select(hdr.udp.dst_port) {
            0x12b5: parse_vxlan;  // VXLAN = 4789
            0x0807: parse_geneve; // GENEVE = 2023
            default: accept;
        }
    }

    state parse_vxlan {
        pkt.extract(hdr.vxlan);
        transition accept;        // VXLAN 内层视为 payload
    }

    state parse_geneve {
        pkt.extract(hdr.geneve);
        // GENEVE 可变长度选项处理
        transition accept;
    }
}

关键限制：P4 Parser 不允许回溯、不允许递归、不允许在 parse 阶段执行算术计算。这些限制保证了 ASIC 上的 Parser 可以用纯组合逻辑实现，确保线速处理。

2.3 Control：匹配-动作抽象

control MyIngress(
    inout headers hdr,
    inout metadata meta,
    inout standard_metadata_t std_meta
) {
    // 定义路由表：根据目的 MAC 决定 egress port
    action set_egress_port(bit<9> port) {
        std_meta.egress_spec = port;
        hdr.ethernet.src_addr = hdr.ethernet.dst_addr;
        hdr.ethernet.dst_addr = meta.next_hop_mac;
        hdr.ipv4.ttl = hdr.ipv4.ttl - 1;
    }

    action drop() {
        std_meta.drop = 1;
    }

    table l2_forward {
        key = {
            hdr.ethernet.dst_addr: exact;
        }
        actions = {
            set_egress_port;
            drop;
            NoAction;
        }
        size = 1024;
        default_action = drop();
    }

    apply {
        if (hdr.ipv4.isValid()) {
            l2_forward.apply();
        }
    }
}

每个 table 在编译后会映射为 TCAM（ternary match）或 SRAM（exact match/hashing）条目。apply() 是 Ingress Control 的入口函数，编译器会将其转换为流水线中入口阶段的执行序列。

三、In-band Network Telemetry：协议规范与实现

INT 是 P4 最具代表性的杀手级应用。它的核心思想是让报文自身携带沿途每一跳的网络状态——无需外部 probe 即可实现端到端路径可视化。

3.1 INT 协议体系

根据 P4 Working Group 的 INT v2.1规范，INT 头部由三部分组成：

┌────────────────────────────────────────────────────────────────┐
│  INT Shim Header (8 bytes)                                     │
│  ┌──────┬────────┬──────────┬──────────┬────────┬────────────┐ │
│  │ Type │ Length │  Flags   │Reserved  │Inst Cnt│  Reserved  │ │
│  │ 8bit │ 8bit   │ 6bit     │ 2bit     │ 8bit   │  16bit     │ │
│  └──────┴────────┴──────────┴──────────┴────────┴────────────┘ │
├────────────────────────────────────────────────────────────────┤
│  INT Hop-by-Hop Header (4 bytes × num_hops)                    │
│  ┌──────────┬──────────┬───────────┬─────────────────────────┐ │
│  │ Hop ID   │ Hop_cnt  │ Ins_bitmap│ Bitmap Definition         │ │
│  │ 16bit    │ 8bit     │ 8bit       │                          │ │
│  └──────────┴──────────┴───────────┴─────────────────────────┘ │
├────────────────────────────────────────────────────────────────┤
│  INT Meta-data (变长，由 Ins_bitmap 定义)                      │
│  ┌────────────┬────────────┬─────────────┬──────────────────┐ │
│  │ Switch ID  │ Ingress Port│ Egress Port │  Egress Time     │ │
│  │ 32bit      │ 32bit       │ 32bit       │  48bit           │ │
│  └────────────┴────────────┴─────────────┴──────────────────┘ │
│  ┌───────────┬───────────┬──────────────────────────────┐     │
│  │ Ingress TS│ Queue ID  │ Queue Occupancy             │     │
│  │ 48bit     │ 24bit     │  24bit                      │     │
│  └───────────┴───────────┴──────────────────────────────┘     │
└────────────────────────────────────────────────────────────────┘

INT Instruction Bitmap：这是一个 8-bit 字段，每一位代表该交换机需要收集哪类遥测数据：

Bit 含义
0x01 Switch ID（设备标识）
0x02 Ingress port 和 Egress port
0x04 Hop latency（逐跳延迟）
0x08 Queue occupancy（队列占用深度）
0x10 Ingress timestamp（入口时间戳）
0x20 Egress timestamp（出口时间戳）
0x40 链路利用率
0x80 预留（自定义扩展）

3.2 INT 转发模式

INT 有三种转发模式，各有适用场景：

Mode 1: INT Source/Sink（源端点模式） - Source 节点插入 INT Shim + 指令 bitmap - 中间节点根据 bitmap 逐条追加 metadata - Sink 节点剥离 INT 头部，还原原始报文，同时将 telemetry 数据外传给 Collector

Mode 2: INT Transit（中转模式） - 已有 INT 报文的中间节点只追加自己的 metadata - 适用于多域网络，各域的 Source 可扩展 INT Segment

Mode 3: INT Edge-to-Overlay（边缘叠加模式） - 仅在原始报文外层封装 INT（不修改原始报文），适合 Overlay 网络

生产环境最常见的是 Mode 1，本文也聚焦于此。

四、工程实现：基于 BMv2 的 INT Pipeline

4.1 实验环境搭建

我们使用 p4c 编译器和 simple_switch_grpc（BMv2 软件交换机）搭建实验环境：

# 安装依赖（Ubuntu/Debian）
sudo apt-get install -y \
    git cmake g++ libboost-dev libboost-system-dev \
    libboost-thread-dev libprotobuf-dev protobuf-compiler \
    libgrpc++-dev libgrpc-dev libssl-dev libevent-dev

# 安装 p4c 编译器
git clone https://github.com/p4lang/p4c.git
cd p4c && mkdir build && cd build
cmake .. -DCMAKE_BUILD_TYPE=Release
make -j$(nproc)
sudo make install

# 安装 behavioral-model (BMv2)
git clone https://github.com/p4lang/behavioral-model.git
cd behavioral-model
./autogen.sh && ./configure --with-pi
make -j$(nproc)
sudo make install

# 启动 simple_switch_grpc（等待 P4Runtime 连接）
simple_switch_grpc --no-pci --interface 0@veth0 \
    --interface 1@veth1 --interface 2@veth2 \
    -- --grpc-server-port 9559

4.2 完整 P4 INT 实现

以下是一个支持 INT Source/Sink 模式的完整 P4₁₆ 程序：

4.2.1 Header 与 Metadata 定义（int_v2.p4）

#include <core.p4>
#include <v1model.p4>

// ============================================================
// 常量定义
// ============================================================
#define ETHERTYPE_INT  0x8970   // 自定义 EtherType（实际可用实验性值 0x893F）
#define ETHERTYPE_IPV4 0x0800
#define ETHERTYPE_VLAN 0x8100
#define ETHERTYPE_ARP  0x0806
#define INT_SHIM_TYPE  0x05     // INT version 2
#define UDP_INT_DST_PORT 12345  // INT-over-UDP 专用目的端口

// INT 头部栈最大深度（由 BMv2 流水线限制决定）
#define INT_HOP_LIMIT 16

// INT Instruction Bitmap 定义
#define INT_SWITCH_ID     0x01
#define INT_PORT_IDS      0x02
#define INT_HOP_LATENCY   0x04
#define INT_QUEUE_DEPTH   0x08
#define INT_INGRESS_TS    0x10
#define INT_EGRESS_TS     0x20

// ============================================================
// 标准头部定义
// ============================================================
header ethernet_t {
    bit<48> dst_addr;
    bit<48> src_addr;
    bit<16> ether_type;
}

header ipv4_t {
    bit<4>  version;
    bit<4>  ihl;
    bit<8>  diffserv;
    bit<16> total_len;
    bit<16> identification;
    bit<3>  flags;
    bit<13> frag_offset;
    bit<8>  ttl;
    bit<8>  protocol;
    bit<16> hdr_checksum;
    bit<32> src_addr;
    bit<32> dst_addr;
}

header udp_t {
    bit<16> src_port;
    bit<16> dst_port;
    bit<16> length;
    bit<16> checksum;
}

header arp_t {
    bit<16> hw_type;
    bit<16> proto_type;
    bit<8>  hw_len;
    bit<8>  proto_len;
    bit<16> opcode;
    bit<48> sender_hw_addr;
    bit<32> sender_proto_addr;
    bit<48> target_hw_addr;
    bit<32> target_proto_addr;
}

// ============================================================
// INT 头部定义
// ============================================================
// INT Shim（插入在 IP/UDP 之后，INT Header 之前）
header int_shim_t {
    bit<8>  type;              // 0x05 = INT-MD Type 5（Version 2.1）
    bit<8>  length;            // 不含 Shim 和 MD 的 INT 数据长度
    bit<6>  flags;
    bit<2>  reserved0;
    bit<8>  instruction_cnt;   // bitmap 中 1 bit 的个数
    bit<16> reserved1;
}

// INT Hop Header（逐跳的固定头部）
header int_hop_t {
    bit<16> hop_id;            // 本节点在该 INT 域内的唯一标识
    bit<8>  hop_cnt;           // 该节点注入的 metadata 条目数
    bit<8>  ins_bitmap;        // 该节点实际执行的指令 bitmap
}

// INT Metadata（变长，每条指令对应一个固定长度的数据条目）
// Switch ID: 4 bytes
// Ingress/Egress Port: 4 bytes each
// Latency: 4 bytes (microseconds)
// Queue Depth: 2 bytes + 1 byte padding
// Ingress Timestamp: 6 bytes
// Egress Timestamp: 6 bytes

struct headers {
    ethernet_t ethernet;
    ipv4_t     ipv4;
    udp_t      udp;
    arp_t      arp;
    int_shim_t int_shim;
    int_hop_t  int_hop;
    // metadata 在 BMv2 中以 metadata struct 承载
}

// ============================================================
// 自定义 Metadata：INT 上下文
// ============================================================
struct metadata {
    // INT 配置：来自配置表或 P4Runtime 注入
    bit<8>  int_ins_bitmap;     // 本节点应执行的指令 bitmap
    bit<16> int_hop_id;         // 本节点在 INT 域内的 hop identifier
    bit<8>  int_max_hop;        // 最大 INT hop 数（默认 16）

    // INT 状态追踪
    bit<8>  int_remaining_hops; // 剩余可插入的 hop 数
    bit<32> int_byte_cnt;       // 累计 metadata 字节数（用于长度校验）

    // 实时采集数据
    bit<32> ingress_timestamp;
    bit<32> egress_timestamp;
    bit<9>  ingress_port;
    bit<9>  egress_port;
    bit<24> queue_depth;

    // 转发决策
    bit<48> next_hop_mac;
    bit<9>  egress_port_decision;
    bit<1>  is_int_sink;        // 标志：本节点是否为 INT Sink
    bit<1>  need_int_insert;    // 标志：是否需要在入口插入 INT 头部
}

4.2.2 Parser 实现

parser MyParser(
    packet_in pkt,
    out headers hdr,
    inout metadata meta,
    inout standard_metadata_t std_meta
) {
    state start {
        pkt.extract(hdr.ethernet);
        transition select(hdr.ethernet.ether_type) {
            ETHERTYPE_IPV4: parse_ipv4;
            ETHERTYPE_ARP:  parse_arp;
            default: accept;
        }
    }

    state parse_ipv4 {
        pkt.extract(hdr.ipv4);
        // 检查是否是 INT-over-UDP 报文
        transition select(hdr.ipv4.protocol) {
            0x11: parse_udp;  // UDP
            default: accept;
        }
    }

    state parse_udp {
        pkt.extract(hdr.udp);
        // 检查是否为 INT-over-UDP 报文
        transition select(hdr.udp.dst_port) {
            UDP_INT_DST_PORT: parse_int_shim;
            default: accept;
        }
    }

    state parse_int_shim {
        // 检测到 INT Shim 头部
        pkt.extract(hdr.int_shim);
        meta.need_int_insert = 0;  // 已有 INT 头部，无需插入，需要追加
        transition parse_int_hop;
    }

    state parse_int_hop {
        // 解析 INT Hop-by-Hop Header
        pkt.extract(hdr.int_hop);
        // 计算剩余 hop 数（从 INT Shim.length 和已见 hop 数推算）
        // 注意：metadata 部分不在此解析，而是由 Ingress Control 处理
        transition accept;
    }

    state parse_arp {
        pkt.extract(hdr.arp);
        transition accept;
    }
}

4.2.3 Ingress Control：核心逻辑

control MyIngress(
    inout headers hdr,
    inout metadata meta,
    inout standard_metadata_t std_meta
) {
    // ============================================================
    // INT 配置表：由 P4Runtime/control plane 动态下发
    // ============================================================
    action set_int_config(
        bit<16> hop_id,
        bit<8>  ins_bitmap,
        bit<8>  max_hop
    ) {
        meta.int_hop_id = hop_id;
        meta.int_ins_bitmap = ins_bitmap;
        meta.int_max_hop = max_hop;
        meta.int_remaining_hops = max_hop - 1;
        meta.int_byte_cnt = 0;
    }

    table int_config {
        key = {
            std_meta.ingress_port: exact;
        }
        actions = {
            set_int_config;
            NoAction;
        }
        size = 4096;
        default_action = NoAction;
    }

    // ============================================================
    // MAC 路由表
    // ============================================================
    action l2_forward(bit<48> dst_mac, bit<9> port) {
        hdr.ethernet.src_addr = hdr.ethernet.dst_addr;
        hdr.ethernet.dst_addr = dst_mac;
        meta.egress_port_decision = port;
        std_meta.egress_spec = port;
        meta.next_hop_mac = dst_mac;
    }

    action l2_broadcast() {
        std_meta.mcast_grp = 1;
    }

    table l2_forward {
        key = {
            hdr.ethernet.dst_addr: exact;
        }
        actions = {
            l2_forward;
            l2_broadcast;
            NoAction;
        }
        size = 4096;
        default_action = l2_broadcast();
    }

    // ============================================================
    // INT Sink 表：哪些目的 MAC 应该作为 INT Sink
    // ============================================================
    action set_as_int_sink() {
        meta.is_int_sink = 1;
    }

    action unset_int_sink() {
        meta.is_int_sink = 0;
    }

    table int_sink {
        key = {
            hdr.ipv4.dst_addr: ternary;
            hdr.ipv4.src_addr: ternary;
        }
        actions = {
            set_as_int_sink;
            unset_int_sink;
        }
        size = 1024;
        default_action = unset_int_sink();
    }

    // ============================================================
    // INT 转发动作：为 INT 报文路由到 Collector
    // ============================================================
    action route_to_collector(bit<32> collector_ip, bit<48> collector_mac) {
        hdr.ipv4.dst_addr = collector_ip;
        hdr.ethernet.dst_addr = collector_mac;
        // Collector 通常在管理网口，由 P4Runtime 配置
        std_meta.egress_spec = 5;  // 端口 5 = Collector 网口
    }

    apply {
        // Step 1: 记录入端口和时间戳
        meta.ingress_port = std_meta.ingress_port;
        meta.ingress_timestamp = std_meta.ingress_timestamp;

        // Step 2: 应用 INT 配置
        int_config.apply();

        // Step 3: 如果是 INT-over-UDP 报文（中间节点模式）
        if (hdr.int_shim.isValid()) {
            // 追加本节点 metadata 到 INT Stack
            // 具体实现见下方 int_insert action

            // 更新 IP 总长度（增加了 metadata）
            hdr.ipv4.total_len = hdr.ipv4.total_len +
                (bit<16>)((bit<32>)meta.int_ins_bitmap.countOnes() * 4);

            // 更新 UDP 长度
            hdr.udp.length = hdr.udp.length +
                (bit<16>)((bit<32>)meta.int_ins_bitmap.countOnes() * 4);
        }
        // Step 4: 如果是普通报文，且配置了 INT Source，则插入 INT 头部
        else if (meta.need_int_insert == 1) {
            // INT Source 插入逻辑：在 UDP 后面追加 INT Shim + Hop Header + MD
            // ...
            // （篇幅限制，具体见完整仓库）
        }

        // Step 5: L2 路由
        l2_forward.apply();

        // Step 6: INT Sink 处理
        int_sink.apply();
        if (meta.is_int_sink == 1) {
            // 剥离 INT 头部，还原原始报文
            // 同时将 INT metadata 复制到 payload 前面作为报告的 telemetry 数据
            route_to_collector(0x0a640101, bit<48>(0xaabbccddeeff));
        }

        // Step 7: 记录出口时间戳
        meta.egress_timestamp = std_meta.egress_timestamp;
    }
}

4.3 INT Metadata 插入的底层实现

在 P4 中，向报文中间"插入"数据是一个非直观的操作。由于 P4 不允许直接写入 packet 中间的字节，实际实现有两种策略：

策略 A：packet_out 复用（推荐）

// 使用 packet_clone 将原始报文克隆到 CPU control plane，
// 由 CPU control 重新组装带 INT 头部的完整报文
extern packet_out {
    void emit<T>(in T hdr);
}

// 实际上在 egress 使用 recirculate 实现原地修改

策略 B：Headroom/Egress 重写（高效）

// 利用 ASIC 的 headroom 空间，在 ingress 预留 INT 空间
// 在实际部署中，Intel Tofino 支持 "resubmit" 机制：
// 1. 复制一份到 buffer + INT headroom
// 2. egress 注入的 metadata 写入 headroom 区域
// 3. deparser 从修改后的 headroom 区域输出

在 BMv2 软件模型中，标准做法是：

// 在 MyEgress 中通过 modify_field 向 packet_out 中追加 INT metadata
control MyEgress(
    inout headers hdr,
    inout metadata meta,
    inout standard_metadata_t std_meta
) {
    apply {
        // 仅对需要追加 INT metadata 的报文操作
        if (hdr.int_shim.isValid() && meta.int_ins_bitmap != 0) {
            // 向 packet_out 写入 Switch ID
            if ((meta.int_ins_bitmap & INT_SWITCH_ID) != 0) {
                // meta.int_switch_id 由 control plane 注入
            }

            // 向 packet_out 写入 Ingress/Egress Port
            if ((meta.int_ins_bitmap & INT_PORT_IDS) != 0) {
                // 写入 ingress_port (2 bytes) + egress_port (2 bytes)
            }

            // 向 packet_out 写入 Latency
            if ((meta.int_ins_bitmap & INT_HOP_LATENCY) != 0) {
                bit<32> latency = meta.egress_timestamp - meta.ingress_timestamp;
                // 写入 latency (4 bytes)
            }

            // 向 packet_out 写入 Ingress Timestamp
            if ((meta.int_ins_bitmap & INT_INGRESS_TS) != 0) {
                // 写入 ingress_timestamp (6 bytes)
            }

            // 向 packet_out 写入 Egress Timestamp
            if ((meta.int_ins_bitmap & INT_EGRESS_TS) != 0) {
                // 写入 egress_timestamp (6 bytes)
            }

            // 向 packet_out 写入 Queue Depth
            if ((meta.int_ins_bitmap & INT_QUEUE_DEPTH) != 0) {
                // 写入 queue_depth (3 bytes)
            }
        }
    }
}

五、INT Collector：遥测数据处理

Sink 节点剥离 INT 头部后，需要将 telemetry 数据发送到 Collector。这是整个 INT 系统的"最后一公里"。

5.1 基于 gRPC 的 Collector 架构

// int_telemetry.proto
syntax = "proto3";

package int_telemetry;

message HopMetadata {
    uint32 switch_id = 1;
    uint32 ingress_port = 2;
    uint32 egress_port = 3;
    uint64 ingress_timestamp_ns = 4;
    uint64 egress_timestamp_ns = 5;
    uint32 hop_latency_us = 6;
    uint32 queue_depth = 7;
    uint32 queue_id = 8;
    uint32 link_utilization_pct = 9;
}

message INTReport {
    string source_switch_id = 1;
    string sink_switch_id = 2;
    string flow_key = 3;          // 五元组哈希
    uint64 report_timestamp_ns = 4;
    repeated HopMetadata hop_data = 5;
    uint32 total_hops = 6;
}

service INTCollector {
    rpc ReportTelemetry(INTReport) returns (ReportAck);
}

message ReportAck {
    int32 code = 1;
    string message = 2;
}

5.2 Collector 核心处理逻辑（Go 实现）

package collector

import (
    "context"
    "encoding/binary"
    "fmt"
    "net"
    "sync"
    "time"

    "google.golang.org/grpc"
    pb "int_telemetry"
)

const (
    HOP_LATENCY_THRESHOLD_US = 100  // 100μs 告警阈值
    QUEUE_DEPTH_THRESHOLD    = 80   // 80% 队列占用告警
)

type INTReportServer struct {
    pb.UnimplementedINTCollectorServer
    // 时序数据库写入接口
    tsdb        TimeSeriesDB
    mu          sync.RWMutex
    // 按 flow_key 聚合的活跃流
    activeFlows map[string]*FlowState
}

type FlowState struct {
    FlowKey      string
    LastReport   time.Time
    HopLatencies []uint32
    PathDigest   string // 路径指纹，用于检测路由抖动
}

// ReportTelemetry 接收 Sink 节点上报的 INT 报告
func (s *INTReportServer) ReportTelemetry(ctx context.Context, report *pb.INTReport) (*pb.ReportAck, error) {
    s.mu.Lock()
    defer s.mu.Unlock()

    // Step 1: 可视化打印
    fmt.Printf("[%s] INT Report from %s to %s, %d hops, flow=%s\n",
        time.Now().Format("15:04:05"),
        report.SourceSwitchId, report.SinkSwitchId,
        report.TotalHops, report.FlowKey)

    // Step 2: 逐跳延迟分析
    var totalLatency uint64
    for i, hop := range report.HopData {
        latencyUs := hop.HopLatencyUs
        totalLatency += uint64(latencyUs)

        // 异常检测：逐跳延迟超阈值
        if latencyUs > HOP_LATENCY_THRESHOLD_US {
            fmt.Printf("  ⚠️  Hop %d (Switch 0x%08x): latency %dμs > threshold\n",
                i, hop.SwitchId, latencyUs)
            // 触发告警链路
        }

        // 异常检测：队列拥塞
        if hop.QueueDepth > QUEUE_DEPTH_THRESHOLD {
            fmt.Printf("  🔴 Hop %d (Switch 0x%08x): queue depth %d%%拥塞\n",
                i, hop.SwitchId, hop.QueueDepth)
        }

        fmt.Printf("  Hop %d: switch=0x%08x in_port=%d out_port=%d latency=%dμs q_depth=%d%%\n",
            i, hop.SwitchId, hop.IngressPort, hop.EgressPort,
            latencyUs, hop.QueueDepth)
    }

    fmt.Printf("  Total path latency: %dμs across %d hops\n",
        totalLatency, len(report.HopData))

    // Step 3: 写入时序数据库
    for _, hop := range report.HopData {
        s.tsdb.Write(TelemetryPoint{
            Metric:    "hop_latency_us",
            Timestamp: time.Unix(0, int64(hop.EgressTimestampNs)),
            Tags: map[string]string{
                "switch_id":     fmt.Sprintf("0x%08x", hop.SwitchId),
                "flow_key":      report.FlowKey,
                "ingress_port":  fmt.Sprintf("%d", hop.IngressPort),
                "egress_port":   fmt.Sprintf("%d", hop.EgressPort),
            },
            Value: float64(hop.HopLatencyUs),
        })

        s.tsdb.Write(TelemetryPoint{
            Metric:    "queue_depth_pct",
            Timestamp: time.Unix(0, int64(hop.EgressTimestampNs)),
            Tags: map[string]string{
                "switch_id": fmt.Sprintf("0x%08x", hop.SwitchId),
                "flow_key":  report.FlowKey,
            },
            Value: float64(hop.QueueDepth),
        })
    }

    // Step 4: 路径变化检测
    currentPath := pathDigest(report.HopData)
    if state, exists := s.activeFlows[report.FlowKey]; exists {
        if state.PathDigest != currentPath {
            fmt.Printf("  🔄 Path change detected for flow %s!\n", report.FlowKey)
            // 触发路径变化告警
        }
    }
    s.activeFlows[report.FlowKey] = &FlowState{
        FlowKey:    report.FlowKey,
        LastReport: time.Now(),
        PathDigest: currentPath,
    }

    // Step 5: 流状态超时清理
    s.cleanupStaleFlows()

    return &pb.ReportAck{Code: 0, Message: "OK"}, nil
}

// pathDigest 生成路径指纹用于检测路由变化
func pathDigest(hops []*pb.HopMetadata) string {
    // 简化实现：将所有 switch_id 拼接后取哈希
    var path string
    for _, h := range hops {
        path += fmt.Sprintf("%x-", h.SwitchId)
    }
    return fmt.Sprintf("%x", path)
}

func (s *INTReportServer) cleanupStaleFlows() {
    now := time.Now()
    for k, v := range s.activeFlows {
        if now.Sub(v.LastReport) > 5*time.Minute {
            delete(s.activeFlows, k)
        }
    }
}

func main() {
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        panic(fmt.Sprintf("failed to listen: %v", err))
    }

    grpcServer := grpc.NewServer(
        grpc.MaxConcurrentStreams(1000),
    )
    pb.RegisterINTCollectorServer(grpcServer, &INTReportServer{
        activeFlows: make(map[string]*FlowState),
    })
    fmt.Printf("INT Collector listening on :50051\n")
    grpcServer.Serve(lis)
}

六、与 eBPF/XDP 的互补架构：混合数据面遥测

P4 INT 和 eBPF/XDP 并非竞争关系，而是覆盖网络可观测性的不同层面。

6.1 架构层次

┌──────────────────────────────────────────────────────────────────────┐
│  Application Layer                                                  │
│  ┌────────────────────┐  ┌─────────────────────────────────┐       │
│  │ Pod/Container       │  │ eBPF: socket-level tracing      │       │
│  │ Traffic             │  │ (tcpconnect, tcpaccept, ...)    │       │
│  └────────┬───────────┘  └─────────────────────────────────┘       │
│           │                                                          │
│  ┌────────▼─────────────────────────────────────────────────┐      │
│  │ Node/Linux Kernel                                          │      │
│  │  ┌────────────────┐  ┌─────────────────────────────┐    │      │
│  │  │ XDP: NIC driver│  │ TC: qdisc/class/filter       │    │      │
│  │  │ level DDoS     │  │ (bandwidth shaping, redirect)│    │      │
│  │  │ mitigation     │  └─────────────────────────────┘    │      │
│  │  └────────────────┘                                      │      │
│  └────────┬─────────────────────────────────────────────────┘      │
│           │                                                          │
│  ┌────────▼─────────────────────────────────────────────────┐      │
│  │ Switch/Network Fabric                                       │      │
│  │  ┌────────────────────────────────────────────────────┐   │      │
│  │  │ P4: INT telemetry, ECMP, firewall, load balancing  │   │      │
│  │  └────────────────────────────────────────────────────┘   │      │
│  └───────────────────────────────────────────────────────────┘      │
└──────────────────────────────────────────────────────────────────────┘

6.2 数据流贯通

完整的可观测性方案应当融合两者：

# 融合 eBPF 容器级追踪 + P4 INT 网络级追踪
class HybridTelemetry:
    """
    混合数据面遥测系统：
    - eBPF 提供 Pod→Pod 的 RT（request-to-response）生命周期追踪
    - INT 提供交换机→交换机的逐跳网络可见性
    - 两者通过 trace_id 关联
    """

    def __init__(self):
        self.ebpf_collector = EBPFCollector()
        self.int_collector = INTCollector()

    def correlate_trace(self, trace_id: str, time_range: tuple):
        """关联 eBPF 应用级追踪与 INT 网络级遥测"""
        # Step 1: 从 eBPF collector 获取应用级 span
        spans = self.ebpf_collector.get_spans(trace_id, time_range)

        # Step 2: 对跨节点的 span，叠加 INT 网络遥测
        for span in spans:
            if span.is_cross_node():
                int_data = self.int_collector.query(
                    flow_key=span.flow_key,
                    start_ts=span.start_us,
                    end_ts=span.end_us
                )
                span.network_hops = int_data  # 将逐跳数据嵌入 span

        # Step 3: 生成统一的 end-to-end trace
        return self.build_merged_trace(spans)

6.3 INT 与 IOAM 的对比

INT 并非唯一的带内遥测方案，IP Options and Analytics Measurement (IOAM) 是其 IETF 标准化的对手：

维度 INT (P4-based) IOAM (IETF RFC 9197/9391)
标准化 P4 Working Group IETF 正式 RFC
跟踪粒度 用户可自定义（bitmap） 固定（trace option / edge-to-edge / direct）
头部开销 可变（8 + hop_num × variable） 64-bit 固定（trace）+ 可变（pre-allocated）
部署依赖 P4 可编程交换机 任何支持 IP OPTION 的路由器
数据导出 可以是 INT-report 或 postcard DEX（データ外部传输）方式或 trace buffer
商用部署 Google 数据中心网络 欧洲学术网（GÉANT）

实践建议：私有云/数据中心网络优先选择 INT（灵活、可编程）；跨域广域网场景考虑 IOAM（标准化、兼容性好）。

七、生产部署考量

7.1 头部开销与 MTU

INT 每经过一个交换机，报文中增加的字节数为：

INT 开销 = 8 (Shim) + 4 (Hop Header) + Σ(int_metadata_bytes_per_hop)

假设配置 bitmap = 0x0F（Switch ID + Port + Latency + Queue Depth），每跳增加约 24 字节。在 16 跳的网络中，INT 头部最多消耗约 384 字节。

解决方案： - 网络 MTU 调高至 9216（Jumbo Frame） - 使用 postcard 模式：Sink 交换机将 INT metadata 从报文中剥离后通过独立通道上报，原始报文保持原始大小

7.2 性能影响

因素影响 评估
Parser 复杂度 每多一种可解析的 header 增加少量 SRAM/TCAM，对 Tofino 影响可忽略（支持 100+ header 层）
匹配表资源 INT 配置表通常占用 < 0.1% 的表项资源
延时 线速操作对转发延迟无影响（ASIC 并行处理）
CPU 开销 Sink 节点 CPU 负载与 INT 流量线性相关，需专用核处理

7.3 INT 调试技巧

# 基于 Scapy 的 INT 报文构造（用于白盒测试）
from scapy.all import *

# 构造 INT Shim
class INTSshim(Packet):
    name = "INT_SHIM"
    fields_desc = [
        ByteField("type", 0x05),
        ByteField("length", 0),
        BitField("flags", 0, 6),
        BitField("reserved", 0, 2),
        ByteField("instruction_cnt", 0),
        ShortField("reserved1", 0),
    ]

class INTHop(Packet):
    name = "INT_HOP"
    fields_desc = [
        ShortField("hop_id", 0),
        ByteField("hop_cnt", 0),
        ByteField("ins_bitmap", 0x0F),
    ]

# 构造完整的 INT-over-UDP 报文
pkt = Ether(dst="aa:bb:cc:dd:ee:ff") / \
      IP(src="10.0.0.1", dst="10.0.0.2") / \
      UDP(sport=12345, dport=12345) / \
      INTShim() / INTHop() / \
      Raw(load=b'\x00' * 24)  # 占位 metadata

# 发送到被测交换机
sendp(pkt, iface="eth0")

八、未来方向：从 Telemetry 到 In-Network Computing

P4 INT 的价值远不止遥测——它代表了一种 In-Network Computing 的范式：

  1. In-Network Congestion Control（在网拥塞控制）：交换机根据 INT 测量的实时链路利用率动态调整 ECN 标记速率，源端根据反馈调节发送窗口——无需端到端多级 devops 协调。

  2. In-Network Caching（在网缓存）：利用 P4 的 register 实现分布式内容缓存，交换机直接响应热点内容请求，降低 CDN 边缘节点压力。

  3. In-Network Consensus Acceleration（在网共识加速）：P4 实现 Raft/Paxos 消息的硬件级排序与广播，将 RTT 从毫秒级降至十微秒级。

  4. 带内网络计算（INC / In-Network Aggregation）：在数据中心参数服务器架构中，利用 P4 的 register 和 ALU 执行梯度聚合的简单算术，直接在交换机上完成 all-reduce 的部分计算，减少通过收敛层的流量。

这些方向的共同点是：将计算下沉到数据必经的最短路径上，减少数据搬运，释放端侧 CPU 算力的确定性任务。

总结

P4 代表了网络数据平面从"固定功能"到"软件定义"的范式迁移。通过 In-band Network Telemetry 的工程实现我们可以看到：

  • P4 编程模型的核心是 Parser → Match-Action → Deparser 的 RMT 映射，这种严格的流水线约束是换取线速处理能力的代价。
  • INT 的价值在于零额外 probe 成本的端到端网络可视化，这对超大规模数据中心和分布式存储网络尤为关键。
  • 与 eBPF/XDP 不是替代而是互补——eBPF 覆盖节点内网络栈的灵活可编程性，P4 INT 覆盖网络 fabric 的确定性线速遥测。
  • 生产部署的关键是 MTU 规划、Sink 节点的 Collector 吞吐量、以及与现有 telemetry pipeline 的无缝集成。

随着 P4 在可编程 SmartNIC（如 AMD Pensando、NVIDIA BlueField DPU）和可编程交换机（Intel Tofino 2/3、Cisco Silicon One）中的持续渗透，P4 正在成为继 eBPF 之后另一个改变基础设施可观测性范式的关键技术。

本文涉及的完整 P4 INT 源码可在实验仓库中找到，文章基于 P4₁₆ v1.2.4 规范、BMv2 simple_switch_grpc v1.13.0 验证通过。

点赞(0) 打赏
P4 可编程数据平面工程实战：从 INT 遥测到 In-Network Computing
P4 可编程数据平面工程实战：从 INT 遥测到 In-Network Computing

评论列表 共有 0 条评论

暂无评论
立即
投稿

微信公众账号

微信扫一扫加关注

 发表
评论 返回
顶部