FROM golang:1.23.6 AS builder

WORKDIR /go/src/github.com/vearne/grpcreplay/
COPY . /go/src/github.com/vearne/grpcreplay/

# 安装依赖
RUN apt-get update && apt-get install -y libpcap-dev
RUN go get

# 构建应用
RUN CGO_ENABLED=1 go build -ldflags "-s -w" -o grpcr

# 使用多阶段构建减小镜像大小
FROM debian:bookworm-slim

# 安装运行时依赖
RUN apt-get update && \
    apt-get install -y libpcap0.8 && \
    apt-get clean && \
    rm -rf /var/lib/apt/lists/*

WORKDIR /app

# 从构建阶段复制二进制文件
COPY --from=builder /go/src/github.com/vearne/grpcreplay/grpcr /app/

# 设置容器入口点
ENTRYPOINT ["/app/grpcr"]

# 构建镜像
podman build --platform linux/amd64 --rm -t grpcr -f Dockerfile .

# 修改镜像标签
podman tag grpcr yeqown/grpcr:v0.2.6

# 推送到 hub.docker.com 镜像仓库 https://hub.docker.com/repository/docker/yeqown/grpcr
podman push yeqown/grpcr:v0.2.6

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: grpc-capture-pvc
spec
  accessModes:
    - ReadWriteMany  # 多个Pod需要同时访问
  resources:
    requests:
      storage: 10Gi
  storageClassName: YOUR_STORAGE_CLASS_NAME  # 替换为你的存储类名称

apiVersion: apps/v1
kind: Deployment
metadata:
  name: old-grpc
  labels:
    app: old-grpc
spec:
  replicas: 1
  selector:
    matchLabels:
      app: old-grpc
  template:
    metadata:
      labels:
        app: old-grpc
    spec:
      containers:
      - name: old-grpc
        image: your-registry/old-grpc:latest
        ports:
        - containerPort: 50051  # gRPC服务端口
+      # grpcreplay sidecar容器
+      - name: grpcreplay
+        image: docker.io/yeqown/grpcr:v0.2.6
+        securityContext:
+          capabilities:
+            add: ["NET_ADMIN", "NET_RAW"]  # 需要这些权限来捕获网络流量
+        args:
+        - "--input-raw=0.0.0.0:50051"
+        - "--output-grpc=grpc://new-grpc:50051"
+        - "--output-file-directory=/capture/server"
+        - "--record-response"
+        - "--codec=json"
+        volumeMounts:
+        - name: capture-volume
+          mountPath: /capture/server
+      volumes:
+      - name: capture-volume
+        persistentVolumeClaim:
+          claimName: grpc-capture-pvc

apiVersion: apps/v1
kind: Deployment
metadata:
  name: new-grpc
  labels:
    app: new-grpc
spec:
  replicas: 1
  selector:
    matchLabels:
      app: new-grpc
  template:
    metadata:
      labels:
        app: new-grpc
    spec:
      containers:
      - name: new-grpc
        image: your-registry/new-grpc:latest
        ports:
        - containerPort: 50051  # gRPC服务端口    
+      # grpcreplay sidecar容器（可选，用于记录请求和响应）
+      - name: grpcreplay
+        image: docker.io/yeqown/grpcr:v0.2.6
+        securityContext:
+          capabilities:
+            add: ["NET_ADMIN", "NET_RAW"]
+        args:
+        - "--input-raw=0.0.0.0:50051"
+        - "--output-stdout"
+        - "--output-file-directory=/capture/mirror"
+        - "--record-response"
+        - "--codec=json"
+        volumeMounts:
+        - name: capture-volume
+          mountPath: /capture/mirror
+      
+      volumes:
+      - name: capture-volume
+        persistentVolumeClaim:
+          claimName: grpc-capture-pvc

apiVersion: networking.istio.io/v1beta1
kind: Gateway
metadata:
  name: istio-gateway
  namespace: xxx
spec:
  selector:
    istio: ingressgateway
  servers:
    - hosts:
        - old-grpc.xxx.com
      port:
        name: grpc
        number: 443
        protocol: GRPC
---
apiVersion: networking.istio.io/v1
kind: VirtualService
metadata:
  name: old-grpc
spec:
  hosts:
  - old-grpc.default.svc.cluster.local
  - old-grpc.xxx.com
  gateways:
  - xxx/istio-gateway
  http:
  - route:
    - destination:
        host: old-grpc.default.svc.cluster.local
        port:
          number: 50051
      weight: 100
    mirror:
      host: new-grpc.default.svc.cluster.local
    mirrorPercentage:
      value: 100

apiVersion: v1
kind: Service
metadata:
  name: old-grpc
  namespace: xxx
spec:
  selector:
    app: old-grpc
  ports:
  - name: grpc-xxx # 显示的配置为 gRPC 协议
    appProtocol: grpc  # 显式的配置为 gRPC 协议，这个配置的优先级更高
    port: 50051
    protocol: TCP

from grpc_interceptor import ServerInterceptor

class RecordReqRespInterceptor(ServerInterceptor):
    """
    Interceptor to record request and response information.
    Usage:
        server = grpc.server(
            ThreadPoolExecutor(max_workers=5),
            interceptors=[RecordReqRespInterceptor(store_directory)])
    """

    fd = None
    store_directory = None  # /capture/server/capture.log
    _lock = threading.Lock()  # 添加线程锁

    def __init__(self, store_directory=None):
        if not store_directory:
            __LOGGER__.warning(
                "Warning: store_directory is not specified, request and response information will not be recorded.")
            return

        filename = store_directory + "capture.log"
        try:
            self.store_directory = filename
            self.fd = open(filename, 'a')
        except Exception as e:
            __LOGGER__.error(f"Failed to open file {store_directory}: {e}")

    def record(self, call_detail: CallDetails, error: str = None):
        if self.fd is None or self.store_directory is None:
            return

        # 从handler_call_details中提取基本信息
        method = call_detail.method

        if method in [
            "/grpc.health.v1.Health/Check",
            "/grpc.reflection.v1.ServerReflection/ServerReflectionInfo",
            "/grpc.reflection.v1alpha.ServerReflection/ServerReflectionInfo",
        ]:
            return

        metadata = dict(call_detail.invocation_metadata)
        timestamp_nano = time.time_ns()
        timestamp_seconds = int(timestamp_nano / 1000000000)

        req_body = json.dumps(MessageToDict(call_detail.request), separators=(',', ':'))
        resp_body = json.dumps(MessageToDict(call_detail.response), separators=(',', ':'))

        # 构建记录
        record = {
            "method": method,
            "request": {
                "headers": metadata,
                "body": req_body,
            },
            "response": {
                "headers": {
                    "failed": error is not None,
                    "error": error if error else "",
                },
                "body": resp_body,
            }
        }

        try:
            with self._lock:  # 使用锁保证写入原子性
                self.fd.write(json.dumps(record) + '\n')
                self.fd.flush()
        except Exception as e:
            __LOGGER__.error(f"Failed to write to file {self.store_directory}: {e}")

    # 实现 ServerInterceptor 抽象方法
    def intercept(
            self,
            method: Callable,
            request_or_iterator: Any,
            context: grpc.ServicerContext,
            method_name: str,
    ):
        response = method(request_or_iterator, context)
        self.record(CallDetails(method_name, context, request_or_iterator, response))

        return response

{
    "method": "/pbpkg.Service/Method",
    "request": {
        "headers": {
            ":authority": "10.90.74.23:50051",
            ":method": "POST",
            ":path": "/pbpkg.Service/Method",
            ":scheme": "http",
            "content-type": "application/grpc",
            "grpc-accept-encoding": "gzip",
            "te": "trailers",
            "user-agent": "grpc-go/1.61.0"
        },
        "body": "<ignored data string>"
    },
    "response": {
        "headers": {
            ":status": "200",
            "content-type": "application/grpc",
            "grpc-message": "",
            "grpc-status": "0"
        },
        "body": "<ignored data string>"
    }
}

def parse_capture_log(log_file) -> Record:
    """
    解析 grpcreplay 生成的 capture.log 文件，返回一个 Record 对象
    """

    records = {}
    with open(log_file, "r") as f:
        for line in f:
            record = Record.parse_raw(line)
            records[record.uuid] = record

def main():
    source_records = parse_capture_log("capture.source.log")
    mirror_records = parse_capture_log("capture.mirror.log")

    for source_record in source_records:
        mirror_record = mirror_records.get(source_record.uuid)

        diff = source_record.diff(mirror_record)
        if diff:
            # 展示差异
            print(f"Request {source_record.uuid} is different:")
            print(diff)

    # 统计对比结果：总共有多少条记录，有多少条记录不同；多少个 method 不同；method 各有多少条记录不同；
    # 统计 method 不同的原因：请求头不同、请求体不同、响应头不同、响应体不同；
    print_summary()

def normalize_json(obj: Any) -> Any:
  """递归地规范化 JSON 对象，使其可以进行顺序无关的比较
    Args:
        obj: 要规范化的对象，可以是字典、列表或其他 JSON 支持的类型
  Returns:
      规范化后的对象
  """
  if isinstance(obj, dict):
      # 如果字典为空，返回 None
      if not obj:
          return None
      normalized = [(k, normalize_json(v)) for k, v in obj.items()]
      normalized = [(k, v) for k, v in normalized if v is not None]
      return sorted(normalized) if normalized else None
  elif isinstance(obj, list):
      if not obj:
          return None
      normalized = [normalize_json(x) for x in obj]
      normalized = [x for x in normalized if x is not None]
      return sorted(normalized) if normalized else None
  else:
      return obj

def stringify_json(obj: Any) -> Any:
    """将规范化的对象转换回 JSON 对象

    Args:
        obj: 规范化后的对象，可能是元组列表（字典）或排序后的列表

    Returns:
        转换回原始 JSON 结构的对象，如果输入为 None，返回空字典或空列表
    """
    if obj is None:
        return {}
    elif isinstance(obj, list):
        if obj and isinstance(obj[0], tuple):
            return {k: stringify_json(v) for k, v in obj}
        return [stringify_json(x) for x in obj]
    elif isinstance(obj, tuple):
        return {obj[0]: stringify_json(obj[1])}
    else:
        return obj

# 测试
json1 = '{"a": [2, 1], "b": {"x": 1, "y": 2}}'
json2 = '{"a": [1, 2], "b": {"y": 2, "x": 1}, "c": []}'
normalized1 = normalize_json(json.loads(json1))
normalized2 = normalize_json(json.loads(json2))
print(normalized1)  # [('a', [1, 2]), ('b', [('x', 1), ('y', 2)])]
print(normalized2)  # [('a', [1, 2]), ('b', [('x', 1), ('y', 2)])]
restored1 = stringify_json(normalized1)
restored2 = stringify_json(normalized2)
print(json.dumps(restored1))  # {"a": [1, 2], "b": {"x": 1, "y": 2}}
print(json.dumps(restored2))  # {"a": [1, 2], "b": {"x": 1, "y": 2}}