如何通过gopacket TCP流重组获取连接各阶段延迟？

要实现你需要的三类延迟统计，核心是利用gopacket的TCP流重组能力，把同一条TCP连接的所有数据包关联起来，跟踪每个关键阶段的时间戳并计算差值。下面是具体实现步骤：

1. 定义流状态跟踪结构体 #

首先为每个TCP流维护一个状态对象，用来存储关键事件的时间戳：

import (
    "net"
    "time"
    "github.com/google/gopacket"
    "github.com/google/gopacket/tcpassembly"
    "github.com/google/gopacket/layers"
)

type StreamLatencyTracker struct {
    // 四元组标识唯一TCP流
    srcIP   net.IP
    srcPort uint16
    dstIP   net.IP
    dstPort uint16

    // 关键事件时间戳
    synTime     time.Time
    synAckTime  time.Time
    ackTime     time.Time
    httpReqTime time.Time
}

2. 实现流工厂与流处理逻辑 #

需要实现gopacket的tcpassembly.StreamFactory和tcpassembly.Stream接口，负责创建新流的跟踪对象，以及处理每个流的数据包：

流工厂实现 #

流工厂用于为每个新TCP流生成对应的跟踪器：

type LatencyStreamFactory struct{}

func (f *LatencyStreamFactory) NewStream(srcFlow, dstFlow gopacket.Flow, _ *layers.TCP, _ tcpassembly.AssemblerContext) tcpassembly.Stream {
    tracker := &StreamLatencyTracker{
        srcIP:   srcFlow.Endpoint().IP(),
        srcPort: srcFlow.Endpoint().Port(),
        dstIP:   dstFlow.Endpoint().IP(),
        dstPort: dstFlow.Endpoint().Port(),
    }
    return tracker
}

流处理逻辑实现 #

在ProcessPacket方法中，逐个分析数据包，识别关键事件并记录时间戳，计算延迟：

func (s *StreamLatencyTracker) ProcessPacket(p gopacket.Packet, _ tcpassembly.AssemblerContext) {
    tcpLayer := p.Layer(layers.LayerTypeTCP).(*layers.TCP)
    packetTime := p.Metadata().Timestamp

    // 1. 识别SYN包（客户端发起连接）
    if tcpLayer.SYN && !tcpLayer.ACK {
        s.synTime = packetTime
        return
    }

    // 2. 识别SYN+ACK包（服务器响应连接）
    if tcpLayer.SYN && tcpLayer.ACK {
        s.synAckTime = packetTime
        if !s.synTime.IsZero() {
            serverLatency := s.synAckTime.Sub(s.synTime)
            // 可在此输出或存储延迟数据，示例：
            // fmt.Printf("Server latency %s:%d->%s:%d: %v\n", s.srcIP, s.srcPort, s.dstIP, s.dstPort, serverLatency)
        }
        return
    }

    // 3. 识别ACK包（客户端确认连接）
    if tcpLayer.ACK && !s.synAckTime.IsZero() && s.ackTime.IsZero() {
        s.ackTime = packetTime
        clientLatency := s.ackTime.Sub(s.synAckTime)
        // fmt.Printf("Client latency %s:%d->%s:%d: %v\n", s.srcIP, s.srcPort, s.dstIP, s.dstPort, clientLatency)
        return
    }

    // 4. 解析HTTP请求与响应
    httpLayer := p.Layer(layers.LayerTypeHTTP)
    if httpLayer == nil {
        return
    }

    httpData, _ := httpLayer.(*layers.HTTP)
    // 识别HTTP请求
    if httpData.Request != nil {
        s.httpReqTime = packetTime
    }
    // 识别HTTP响应并计算延迟
    if httpData.Response != nil && !s.httpReqTime.IsZero() {
        httpLatency := packetTime.Sub(s.httpReqTime)
        // fmt.Printf("HTTP latency %s:%d->%s:%d: %v\n", s.srcIP, s.srcPort, s.dstIP, s.dstPort, httpLatency)
        s.httpReqTime = time.Time{} // 重置请求时间，避免重复计算同流后续响应
    }
}

// 实现Stream接口的ReassemblyComplete方法，清理流资源
func (s *StreamLatencyTracker) ReassemblyComplete() {
    // 可在此将跟踪器从全局映射移除，避免内存泄漏
}

3. 初始化流重组器并开始捕获 #

最后初始化gopacket的捕获器和流重组器，将捕获到的数据包送入重组器处理：

func main() {
    // 初始化网络接口捕获（替换为实际需要监听的接口）
    handle, err := pcap.OpenLive("eth0", 1600, true, pcap.BlockForever)
    if err != nil {
        panic(err)
    }
    defer handle.Close()

    // 设置BPF过滤器，只捕获TCP和HTTP流量（可选，减少处理量）
    err = handle.SetBPFFilter("tcp or tcp port 80 or tcp port 443")
    if err != nil {
        panic(err)
    }

    // 初始化流重组器
    factory := &LatencyStreamFactory{}
    assembler := tcpassembly.NewAssembler(tcpassembly.NewStreamPool(factory))

    // 开始捕获并处理数据包
    packetSource := gopacket.NewPacketSource(handle, handle.LinkType())
    for packet := range packetSource.Packets() {
        if tcpLayer := packet.Layer(layers.LayerTypeTCP); tcpLayer != nil {
            tcp, _ := tcpLayer.(*layers.TCP)
            assembler.AssembleWithTimestamp(packet.NetworkLayer().NetworkFlow(), tcp, packet.Metadata().Timestamp)
        }
    }
}

注意事项 #

• 四元组匹配：gopacket的流重组会自动根据四元组（源IP/端口、目标IP/端口）分组数据包，无需手动关联。
• 时间精度：必须使用数据包的Metadata().Timestamp（捕获时的原始时间），不要用本地系统时间，避免误差。
• 流清理：通过ReassemblyComplete方法在流结束时清理跟踪对象，防止内存泄漏。
• HTTPS处理：上述代码仅支持明文HTTP，若需统计HTTPS延迟，需额外处理TLS握手后的应用层解密数据。

内容的提问来源于stack exchange，提问作者world hello