Contour 中 Envoy 优雅停服的实现与源码分析

apiVersion: extensions/v1beta1
kind: DaemonSet
metadata:
  annotations:
  labels:
    app: envoy
  name: envoy
  namespace: projectcontour
spec:
  revisionHistoryLimit: 10
  selector:
    matchLabels:
      app: envoy
  template:
    metadata:
      annotations:
        prometheus.io/path: /stats/prometheus
        prometheus.io/port: "8002"
        prometheus.io/scrape: "true"
      creationTimestamp: null
      labels:
        app: envoy
    spec:
      automountServiceAccountToken: false
      containers:
      - command:                          # <----- New Pod
        - /bin/contour
        args:
        	- envoy
        	- shutdown-manager
        image: stevesloka/envoyshutdown
        imagePullPolicy: Always
        lifecycle:
          preStop:                       # <----- PreStop Hook
            exec:
            	command:
            		- /bin/contour
            		- envoy
            		- shutdown
         livenessProbe:                  # <------ Liveness probe  
           httpGet:
             path: /healthz
             port: 8090
           initialDelaySeconds: 3
           periodSeconds: 10
        name: shutdown-manager
        terminationMessagePath: /dev/termination-log
        terminationMessagePolicy: File
      - args:
        - -c
        - /config/envoy.json
        - --service-cluster $(CONTOUR_NAMESPACE)
        - --service-node $(ENVOY_POD_NAME)
        - --log-level info
        command:
        - envoy
        env:
        - name: CONTOUR_NAMESPACE
          valueFrom:
            fieldRef:
              apiVersion: v1
              fieldPath: metadata.namespace
        - name: ENVOY_POD_NAME
          valueFrom:
            fieldRef:
              apiVersion: v1
              fieldPath: metadata.name
        image: docker.io/envoyproxy/envoy:v1.13.0
        imagePullPolicy: IfNotPresent
        lifecycle:                                # <----- PreStop Hook
          preStop:
            httpGet:
              path: /shutdown
              port: 8090
              scheme: HTTP
        name: envoy
        ports:
        - containerPort: 80
          hostPort: 80
          name: http
          protocol: TCP
        - containerPort: 443
          hostPort: 443
          name: https
          protocol: TCP
        readinessProbe:
          failureThreshold: 4
          httpGet:
            path: /ready
            port: 8002
            scheme: HTTP
          initialDelaySeconds: 3
          periodSeconds: 3
          successThreshold: 1
          timeoutSeconds: 1
        volumeMounts:
        - mountPath: /config
          name: envoy-config
        - mountPath: /certs
          name: envoycert
        - mountPath: /ca
          name: cacert
      dnsPolicy: ClusterFirst
      initContainers:
      - args:
        - bootstrap
        - /config/envoy.json
        - --xds-address=contour
        - --xds-port=8001
        - --envoy-cafile=/ca/cacert.pem
        - --envoy-cert-file=/certs/tls.crt
        - --envoy-key-file=/certs/tls.key
        command:
        - contour
        env:
        - name: CONTOUR_NAMESPACE
          valueFrom:
            fieldRef:
              apiVersion: v1
              fieldPath: metadata.namespace
        image: docker.io/projectcontour/contour:master
        imagePullPolicy: Always
        name: envoy-initconfig
        resources: {}
        terminationMessagePath: /dev/termination-log
        terminationMessagePolicy: File
        volumeMounts:
        - mountPath: /config
          name: envoy-config
        - mountPath: /certs
          name: envoycert
          readOnly: true
        - mountPath: /ca
          name: cacert
          readOnly: true
      restartPolicy: Always
      schedulerName: default-scheduler
      securityContext: {}
      terminationGracePeriodSeconds: 300
      volumes:
      - emptyDir: {}
        name: envoy-config
      - name: envoycert
        secret:
          defaultMode: 420
          secretName: envoycert
      - name: cacert
        secret:
          defaultMode: 420
          secretName: cacert
  updateStrategy:
    rollingUpdate:
      maxUnavailable: 10%
    type: RollingUpdate

$ ./contour envoy shutdown-manager -h
usage: contour envoy shutdown-manager [<flags>]

Start envoy shutdown-manager.

Flags:
  -h, --help                   Show context-sensitive help (also try --help-long and --help-man).
      --serve-port=SERVE-PORT  Port to serve the http server on.

// contour/cmd/contour/contour.go
func main() {
  ...
  // 一级子命令：envoy
  envoyCmd := app.Command("envoy", "Sub-command for envoy actions.")
  sdm, shutdownManagerCtx := registerShutdownManager(envoyCmd, log)
  ...
  switch kingpin.MustParse(app.Parse(args)) {
	case sdm.FullCommand():
		doShutdownManager(shutdownManagerCtx)
    ...
  }
}

// contour/cmd/contour/shutdownmanager.go
func registerShutdownManager(cmd *kingpin.CmdClause, log logrus.FieldLogger) (*kingpin.CmdClause, *shutdownmanagerContext) {
	ctx := newShutdownManagerContext()
	ctx.FieldLogger = log.WithField("context", "shutdown-manager")

  // 二级子命令：shutdown-manager
	shutdownmgr := cmd.Command("shutdown-manager", "Start envoy shutdown-manager.")
	shutdownmgr.Flag("serve-port", "Port to serve the http server on.").IntVar(&ctx.httpServePort)

	return shutdownmgr, ctx
}

// 初始化默认参数
func newShutdownManagerContext() *shutdownmanagerContext {
	// Set defaults for parameters which are then overridden via flags, ENV, or ConfigFile
	return &shutdownmanagerContext{
		httpServePort:              8090,
    // const shutdownReadyFile = "/ok"
		shutdownReadyFile:          shutdownReadyFile,
		shutdownReadyCheckInterval: shutdownReadyCheckInterval,
	}
}

// contour/cmd/contour/shutdownmanager.go
func doShutdownManager(config *shutdownmanagerContext) {

	config.Info("started envoy shutdown manager")
	defer config.Info("stopped")

  // 暴露两个接口
	http.HandleFunc("/healthz", config.healthzHandler)
	http.HandleFunc("/shutdown", config.shutdownReadyHandler)
  // 默认监听 8090 端口
	log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", config.httpServePort), nil))
}

func (s *shutdownmanagerContext) shutdownReadyHandler(w http.ResponseWriter, r *http.Request) {
	l := s.WithField("context", "shutdownReadyHandler")
	ctx := r.Context()
	for {
    // 判断 /ok 文件是否存在
		_, err := os.Stat(s.shutdownReadyFile)
		if os.IsNotExist(err) {
      // 不存在就跳过，等待下次执行。说明这时候连接数还没有将到阈值，还不能安全的退出
			l.Infof("file %s does not exist; checking again in %v", s.shutdownReadyFile,
				s.shutdownReadyCheckInterval)
		} else if err == nil {
			l.Infof("detected file %s; sending HTTP response", s.shutdownReadyFile)
			http.StatusText(http.StatusOK)
			if _, err := w.Write([]byte("OK")); err != nil {
				l.Error(err)
			}
			return
		} else {
			l.Errorf("error checking for file: %v", err)
		}

		select {
    // 休眠一段时间
		case <-time.After(s.shutdownReadyCheckInterval):
		case <-ctx.Done():
			l.Infof("client request cancelled")
			return
		}
	}
}

$ ./contour envoy shutdown -h        
usage: contour envoy shutdown [<flags>]

Initiate an shutdown sequence which configures Envoy to begin draining connections.

Flags:
  -h, --help                   Show context-sensitive help (also try --help-long and --help-man).
      --admin-port=ADMIN-PORT  Envoy admin interface port.
      --check-interval=CHECK-INTERVAL  
                               Time to poll Envoy for open connections.
      --check-delay=60s        Time to wait before polling Envoy for open connections.
      --drain-delay=0s         Time to wait before draining Envoy connections.
      --min-open-connections=MIN-OPEN-CONNECTIONS  
                               Min number of open connections when polling Envoy.

func main() {
	...
  // 一级子命令：envoy
  envoyCmd := app.Command("envoy", "Sub-command for envoy actions.")
	sdmShutdown, sdmShutdownCtx := registerShutdown(envoyCmd, log)
  ...
  switch kingpin.MustParse(app.Parse(args)) {
    ...
  	case sdmShutdown.FullCommand():
			sdmShutdownCtx.shutdownHandler()
    ...
  }
}

func registerShutdown(cmd *kingpin.CmdClause, log logrus.FieldLogger) (*kingpin.CmdClause, *shutdownContext) {
	ctx := newShutdownContext()
	ctx.FieldLogger = log.WithField("context", "shutdown")

  // 二级命令：shutdown
	shutdown := cmd.Command("shutdown", "Initiate an shutdown sequence which configures Envoy to begin draining connections.")
	shutdown.Flag("admin-port", "Envoy admin interface port.").IntVar(&ctx.adminPort)
	shutdown.Flag("check-interval", "Time to poll Envoy for open connections.").DurationVar(&ctx.checkInterval)
	shutdown.Flag("check-delay", "Time to wait before polling Envoy for open connections.").Default("60s").DurationVar(&ctx.checkDelay)
	shutdown.Flag("drain-delay", "Time to wait before draining Envoy connections.").Default("0s").DurationVar(&ctx.drainDelay)
	shutdown.Flag("min-open-connections", "Min number of open connections when polling Envoy.").IntVar(&ctx.minOpenConnections)

	return shutdown, ctx
}

func (s *shutdownContext) shutdownHandler() {

  // 重试去关闭 envoy
	err := retry.OnError(wait.Backoff{
		Steps:    4,
		Duration: 200 * time.Millisecond,
		Factor:   5.0,
		Jitter:   0.1,
	}, func(err error) bool {
		// Always retry any error.
		return true
	}, func() error {
		s.Infof("attempting to shutdown")
    // 尝试 shutdown
		return shutdownEnvoy(s.adminPort)
	})
	...
	time.Sleep(s.checkDelay)

	for {
    // 连接 envoy 的管理端口，获取指标信息
    // 地址：http://localhost:9001/stats/prometheus
    // 然后获取连接数
    // envoy_http_downstream_cx_active 指标，label 取 ingress_http
		openConnections, err := getOpenConnections(s.adminPort)
		if err != nil {
			s.Error(err)
		} else {
      // 如果连接数 <= 配置的最小连接数
			if openConnections <= s.minOpenConnections {
				...
        // 创建 /ok 文件，表明可以安全的终止服务
				file, err := os.Create(shutdownReadyFile)
				...
				return
			}
		}
    // 连接数没有降下来，需要再 sleep 一段时间
		time.Sleep(s.checkInterval)
	}
}

func shutdownEnvoy(adminPort int) error {
  // 向管理端口发送一个将康检查失败的请求
  // http://localhost:9001/healthcheck/fail
	healthcheckFailURL := fmt.Sprintf(healthcheckFailURLFormat, adminPort)
	// 发送 POST 请求
	resp, err := http.Post(healthcheckFailURL, "", nil)
	if err != nil {
		return fmt.Errorf("creating healthcheck fail POST request failed: %s", err)
	}

	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("POST for %q returned HTTP status %s", healthcheckFailURL, resp.Status)
	}
	return nil
}

func getOpenConnections(adminPort int) (int, error) {
  // 指标地址：http://localhost:9001/stats/prometheus
	prometheusURL := fmt.Sprintf(prometheusURLFormat, adminPort)
	// 发送 Get 请求获取指标
	resp, err := http.Get(prometheusURL)
  ...
	// 提取指标中的连接数
	return parseOpenConnections(resp.Body)
}

// parseOpenConnections returns the sum of open connections from a Prometheus HTTP request
func parseOpenConnections(stats io.Reader) (int, error) {
	...
  // 将指标文本转换为对象
	metricFamilies, err := parser.TextToMetricFamilies(stats)
	
  // 获取 envoy_http_downstream_cx_active 指标
	if _, ok := metricFamilies[prometheusStat]; !ok {
		return -1, fmt.Errorf("error finding Prometheus stat %q in the request result", prometheusStat)
	}

	// 查找 ingress_http label的数据并累加
	for _, metrics := range metricFamilies[prometheusStat].Metric {
		for _, labels := range metrics.Label {
			for _, item := range prometheusLabels() {
				if item == labels.GetValue() {
					openConnections += int(metrics.Gauge.GetValue())
				}
			}
		}
	}
	return openConnections, nil
}