同一StatefulSet内Pod间通信的推荐方案及Go语言K8s客户端实现

一、StatefulSet Pod间通信的推荐方式 #

• Headless Service（无头服务）是官方首选方案：StatefulSet的Pod拥有稳定的网络标识，绑定Headless Service（clusterIP: None）后，每个Pod会获得固定的DNS记录，格式为 <pod-name>.<headless-service-name>.<namespace>.svc.cluster.local。同命名空间内可简化为 <pod-name>.<headless-service-name> 访问，Pod重启或重建后名称不变，DNS记录也保持稳定，完全适配StatefulSet的场景。
• 不推荐直接使用Pod IP：Pod IP是临时的，重建后会变更，无法保证通信的稳定性。
• 可选DNS SRV查询：Headless Service会生成SRV记录，可通过DNS查询获取所有Pod的网络信息，但直接通过Kubernetes API查询Pod列表更可靠、可控。

二、Go语言Kubernetes客户端实现请求转发（排除自身Pod） #

核心思路 #

1. 获取当前Pod的名称（通过Kubernetes注入的环境变量）；
2. 初始化Kubernetes客户端，查询同StatefulSet下的所有Pod；
3. 过滤掉自身Pod，得到需要转发的Peer Pod列表；
4. 基于Headless Service的DNS地址，将原始POST请求转发给所有Peer Pod。

代码实现 #

1. 依赖准备

确保项目引入client-go相关依赖：

require (
    k8s.io/client-go v0.28.4
    k8s.io/api v0.28.4
    k8s.io/apimachinery v0.28.4
)

2. 完整示例代码

package main

import (
    "context"
    "fmt"
    "io"
    "os"
    "net/http"
    "strings"

    v1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/rest"
)

// 获取当前Pod的名称（从环境变量读取，需在StatefulSet中配置注入）
func getOwnPodName() (string, error) {
    podName := os.Getenv("POD_NAME")
    if podName == "" {
        return "", fmt.Errorf("POD_NAME环境变量未设置")
    }
    return podName, nil
}

// 初始化集群内Kubernetes客户端
func initK8sClient() (*kubernetes.Clientset, error) {
    config, err := rest.InClusterConfig()
    if err != nil {
        return nil, fmt.Errorf("获取集群内配置失败: %v", err)
    }
    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        return nil, fmt.Errorf("创建K8s客户端失败: %v", err)
    }
    return clientset, nil
}

// 查询同StatefulSet下的其他Peer Pod
func getPeerPods(clientset *kubernetes.Clientset, namespace, ownPodName, labelSelector string) ([]v1.Pod, error) {
    pods, err := clientset.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{
        LabelSelector: labelSelector,
    })
    if err != nil {
        return nil, fmt.Errorf("查询Pod列表失败: %v", err)
    }

    var peerPods []v1.Pod
    for _, pod := range pods.Items {
        if pod.Name != ownPodName {
            peerPods = append(peerPods, pod)
        }
    }
    return peerPods, nil
}

// 转发原始请求到所有Peer Pod
func forwardRequestToPeers(peerPods []v1.Pod, headlessSvcName string, originalReq *http.Request) error {
    // 获取当前Pod所在命名空间
    namespace := os.Getenv("POD_NAMESPACE")
    if namespace == "" {
        namespace = "default"
    }

    // 读取原始请求Body，避免只能读取一次的问题
    bodyBytes, err := io.ReadAll(originalReq.Body)
    if err != nil {
        return fmt.Errorf("读取原始请求Body失败: %v", err)
    }
    defer originalReq.Body.Close()

    client := &http.Client{}
    for _, pod := range peerPods {
        // 构建Peer Pod的访问地址
        podAddr := fmt.Sprintf("%s.%s.%s.svc.cluster.local", pod.Name, headlessSvcName, namespace)
        targetURL := fmt.Sprintf("http://%s%s", podAddr, originalReq.URL.Path)

        // 复制请求
        req, err := http.NewRequest(originalReq.Method, targetURL, strings.NewReader(string(bodyBytes)))
        if err != nil {
            fmt.Printf("为Pod %s复制请求失败: %v\n", pod.Name, err)
            continue
        }
        // 复制请求头
        req.Header = originalReq.Header.Clone()

        // 发送请求
        resp, err := client.Do(req)
        if err != nil {
            fmt.Printf("向Pod %s发送请求失败: %v\n", pod.Name, err)
            continue
        }
        defer resp.Body.Close()

        fmt.Printf("转发请求到Pod %s成功，响应状态: %s\n", pod.Name, resp.Status)
    }
    return nil
}

func main() {
    // 配置参数，根据实际环境修改
    const (
        namespace           = "default"               // StatefulSet所在命名空间
        statefulSetLabel    = "app=my-statefulset"    // StatefulSet的标签选择器
        headlessServiceName = "MyService"             // Headless Service名称
        listenPort          = ":8080"
    )

    // 获取自身Pod名称
    ownPodName, err := getOwnPodName()
    if err != nil {
        panic(fmt.Sprintf("获取自身Pod名称失败: %v", err))
    }

    // 初始化K8s客户端
    clientset, err := initK8sClient()
    if err != nil {
        panic(fmt.Sprintf("初始化K8s客户端失败: %v", err))
    }

    // 处理外部POST请求的接口
    http.HandleFunc("/api/forward", func(w http.ResponseWriter, r *http.Request) {
        if r.Method != http.MethodPost {
            w.WriteHeader(http.StatusMethodNotAllowed)
            fmt.Fprintf(w, "仅支持POST请求")
            return
        }

        // 获取Peer Pod列表
        peerPods, err := getPeerPods(clientset, namespace, ownPodName, statefulSetLabel)
        if err != nil {
            w.WriteHeader(http.StatusInternalServerError)
            fmt.Fprintf(w, "获取Peer Pod失败: %v", err)
            return
        }

        if len(peerPods) == 0 {
            w.WriteHeader(http.StatusOK)
            fmt.Fprintf(w, "无Peer Pod可转发")
            return
        }

        // 转发请求
        err = forwardRequestToPeers(peerPods, headlessServiceName, r)
        if err != nil {
            w.WriteHeader(http.StatusInternalServerError)
            fmt.Fprintf(w, "转发请求失败: %v", err)
            return
        }

        w.WriteHeader(http.StatusOK)
        fmt.Fprintf(w, "请求已成功转发到所有Peer Pod")
    })

    // 启动HTTP服务
    fmt.Printf("服务启动，监听端口: %s\n", listenPort)
    if err := http.ListenAndServe(listenPort, nil); err != nil {
        panic(fmt.Sprintf("启动HTTP服务失败: %v", err))
    }
}

关键配置说明 #

1. StatefulSet注入环境变量

在StatefulSet的Pod模板中添加环境变量，注入当前Pod的名称和命名空间：

spec:
  template:
    spec:
      containers:
      - name: your-container-name
        image: your-image
        env:
        - name: POD_NAME
          valueFrom:
            fieldRef:
              fieldPath: metadata.name
        - name: POD_NAMESPACE
          valueFrom:
            fieldRef:
              fieldPath: metadata.namespace

2. RBAC权限配置

Pod需要具备查询Pod列表的权限，创建对应的Role和RoleBinding：

apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  namespace: default
  name: pod-reader
rules:
- apiGroups: [""]
  resources: ["pods"]
  verbs: ["list"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  namespace: default
  name: read-pods
subjects:
- kind: ServiceAccount
  name: default # 使用默认ServiceAccount，或自定义ServiceAccount
roleRef:
  kind: Role
  name: pod-reader
  apiGroup: rbac.authorization.k8s.io

内容的提问来源于stack exchange，提问作者Sai Satwik Kuppili