从源码看istio探针重写

背景

istio注入sidecar会重写pod原有的probe,这里记录一下相关源码

源码

istiod

pkg/kube/inject/webhook.go中

func NewWebhook(p WebhookParameters) (*Webhook, error) {
    ...
    注册注入的http入口方法
    p.Mux.HandleFunc("/inject", wh.serveInject)
    p.Mux.HandleFunc("/inject/", wh.serveInject)
    ...
}

处理注入的http入口方法
func (wh *Webhook) serveInject(w http.ResponseWriter, r *http.Request) {
    ...
    处理注入
            reviewResponse = wh.inject(ar, path)
    ...
}


注入方法
func (wh *Webhook) inject(ar *kube.AdmissionReview, path string) \*kube.AdmissionResponse {
    ...
    注入pod
    patchBytes, err := injectPod(params)
    ...
}

注入pod
func injectPod(req InjectionParameters) ([]byte, error) {
    ...
    pod根据注入模板处理后
    if err := postProcessPod(mergedPod, *injectedPodData, req); err != nil {
        return nil, fmt.Errorf("failed to process pod: %v", err)
    }
    ...
}

pod根据注入模板处理后
func postProcessPod(pod *corev1.Pod, injectedPod corev1.Pod, req InjectionParameters) error {
    ...
    重写
    if err := applyRewrite(pod, req); err != nil {
        return err
    }
    ...
}

重写探针
func applyRewrite(pod *corev1.Pod, req InjectionParameters) error {
    找到sidecar容器
    sidecar := FindSidecar(pod)
    if sidecar == nil {
        return nil
    }
    判断是否需要重写
    rewrite := ShouldRewriteAppHTTPProbers(pod.Annotations, req.valuesConfig.asStruct.GetSidecarInjectorWebhook().GetRewriteAppHTTPProbe().GetValue())
    if rewrite {
        if prober := DumpAppProbers(pod, req.meshConfig.GetDefaultConfig().GetStatusPort()); prober != "" {
            写入环境变量
            _, previouslyInjected := pod.Annotations[annotation.SidecarStatus.Name]
            sidecar.Env = mergeOrAppendProbers(previouslyInjected, sidecar.Env, prober)
        }
        修改探针
        patchRewriteProbe(pod.Annotations, pod, req.meshConfig.GetDefaultConfig().GetStatusPort())
    }
    return nil
}

合并或追加应用探针
func mergeOrAppendProbers(previouslyInjected bool, envVars []corev1.EnvVar, newProbers string) []corev1.EnvVar {
    如果未注入
    if !previouslyInjected {
        return append(envVars, corev1.EnvVar{Name: status.KubeAppProberEnvName, Value: newProbers})
    }
    ...
}

pkg/kube/inject/app_probe.go中

改写探针
func patchRewriteProbe(annotations map[string]string, pod *corev1.Pod, defaultPort int32) {
    ...
    遍历容器
    for i, c := range pod.Spec.Containers {
        if c.Name == ProxyContainerName {
            continue
        }
        转换探针
        convertProbe(&c, statusPort)
        pod.Spec.Containers[i] = c
    }
    ...
}

转换探针
func convertProbe(c *corev1.Container, statusPort int) {
    生成探针URL
    readyz, livez, startupz, prestopz, poststartz := status.FormatProberURL(c.Name)
    ...
    转换探针
    if probePatch := convertAppProber(c.ReadinessProbe, readyz, statusPort); probePatch != nil {
        c.ReadinessProbe = probePatch
    }
    ...
}

转换探针
func convertAppProber(probe *corev1.Probe, newURL string, statusPort int) *corev1.Probe {
    if probe == nil {
        return nil
    }
    if probe.HTTPGet != nil {
        return convertAppProberHTTPGet(probe, newURL, statusPort)
    } else if probe.TCPSocket != nil {
        return convertAppProberTCPSocket(probe, newURL, statusPort)
    } else if probe.GRPC != nil {
        return convertAppProberGRPC(probe, newURL, statusPort)
    }

    return nil
}

转换HTTPGet探针
func convertAppProberHTTPGet(probe *corev1.Probe, newURL string, statusPort int) *corev1.Probe {
    p := probe.DeepCopy()
    rewriteHTTPGetAction(p.HTTPGet, newURL, statusPort)
    return p
}


重写HTTPGetAction
func rewriteHTTPGetAction(action *corev1.HTTPGetAction, url string, port int) {
    action.Port = intstr.FromInt32(int32(port))
    action.Path = url
    // Kubelet -> HTTP -> Pilot Agent -> HTTPS -> Application
    if action.Scheme == corev1.URISchemeHTTPS {
        action.Scheme = corev1.URISchemeHTTP
    }
}

pilot/cmd/pilot-agent/status/server.go中

生成探针URL
func FormatProberURL(container string) (string, string, string, string, string) {
    return fmt.Sprintf("/app-health/%v/readyz", container),
        fmt.Sprintf("/app-health/%v/livez", container),
        fmt.Sprintf("/app-health/%v/startupz", container),
        fmt.Sprintf("/app-lifecycle/%v/prestopz", container),
        fmt.Sprintf("/app-lifecycle/%v/poststartz", container)
}

pilot-agent

pilot/cmd/pilot-agent/status/server.go中

var(
    ...
    KubeAppProberEnvName环境变量名
    KubeAppProberEnvName = "ISTIO_KUBE_APP_PROBERS"
    ...
)

pilot/cmd/pilot-agent/options/statusserver.go中

构建pilot-agent status server options
func NewStatusServerOptions(ipv6 bool, t model.NodeType, proxyConfig *meshconfig.ProxyConfig, agent *istioagent.Agent) *status.Options {
    return &status.Options{
        ...
        PodIP:          InstanceIPVar.Get(),
        ...
        KubeAppProbers: kubeAppProberNameVar.Get(),
        ...
    }
}


var (
    ...
    KubeAppProberEnvName环境变量
    kubeAppProberNameVar = env.Register(status.KubeAppProberEnvName, "", "")
    ...
    pod ip环境变量
    InstanceIPVar        = env.Register("INSTANCE_IP", "", "")
)

pilot/cmd/pilot-agent/status/server.go中

构建pilot-agent status server
func NewServer(config Options) (*Server, error) {
    ...
    解析KubeAppProbers
    if err := json.Unmarshal([]byte(config.KubeAppProbers), &s.appKubeProbers); err != nil {
        return nil, fmt.Errorf("failed to decode app prober err = %v, json string = %v", err, config.KubeAppProbers)
    }
    ...
    初始化appProbeClient
    s.appProbeClient = make(map[string]*http.Client, len(s.appKubeProbers))
    ...
    构建appProbeClient
    for path, prober := range s.appKubeProbers {
        ...
        if prober.HTTPGet != nil {
            ...
            s.appProbeClient[path] = &http.Client{
                Timeout: time.Duration(prober.TimeoutSeconds) * time.Second,
                Transport:     transport,
                CheckRedirect: redirectChecker(),
            }
        }
    }
}


运行pilot-agent status server
func (s *Server) Run(ctx context.Context) {
    ...
    注册app-health和app-lifecycle的handler
    mux.HandleFunc("/app-health/", s.handleAppProbe)
    mux.HandleFunc("/app-lifecycle/", s.handleAppProbe)
    ...
}


处理应用探针请求
func (s *Server) handleAppProbe(w http.ResponseWriter, req *http.Request) {
    ...
    判断探针类型
    switch {
    case prober.HTTPGet != nil:
        s.handleAppProbeHTTPGet(w, req, prober, path)
    case prober.TCPSocket != nil:
        s.handleAppProbeTCPSocket(w, prober)
    case prober.GRPC != nil:
        s.handleAppProbeGRPC(w, req, prober)
    }
    ...
}


处理HTTPGet探针
func (s *Server) handleAppProbeHTTPGet(w http.ResponseWriter, req *http.Request, prober *Prober, path string) {
    ...
    proberPath := prober.HTTPGet.Path

    ...
    hostPort := net.JoinHostPort(s.appProbersDestination, strconv.Itoa(prober.HTTPGet.Port.IntValue()))
    ...
    构建应用探针HTTP请求
    appReq, err := http.NewRequest(http.MethodGet, url, nil)
    if err != nil {
        log.Errorf("Failed to create request to probe app %v, original url %v", err, path)
        w.WriteHeader(http.StatusInternalServerError)
        return
    }
    ...
    获取应用探针HTTP客户端
    httpClient := s.appProbeClient[path]
    发送应用探针HTTP请求
    response, err := httpClient.Do(appReq)
    ...
}

配置

全局

修改istio-system/istio-sidecar-injector这个configmap(默认为true)

  values: |-
    {
      ...
      "sidecarInjectorWebhook": {
        ...
        "rewriteAppHTTPProbe": true,
        ...
      }
      ...
    }

pod级

pod添加注解

metadata:
  annotations:
    sidecar.istio.io/rewriteAppHTTPProbers: "true"