k8s探针实现grpc健康检查

这篇文章教你们如何利用k8s实现grpc健康检查

原理跟github上的grpc-ecosystem/grpc-health-probe 同样, 这里是简单实现

一. 配置Liveness和Readiness探针

kubelet 使用 liveness probe（存活探针）来肯定什么时候重启容器。例如，当应用程序处于运行状态但没法作进一步操做，liveness 探针将捕获到 deadlock，重启处于该状态下的容器，使应用程序在存在 bug 的状况下依然可以继续运行下去。

kubelet 使用 readiness probe（就绪探针）来肯定容器是否已经就绪能够接受流量。只有当 Pod 中的容器都处于就绪状态时 kubelet 才会认定该 Pod处于就绪状态。该信号的做用是控制哪些 Pod应该做为service的后端。若是 Pod 处于非就绪状态，那么它们将会被从 service 的 load balancer中移除。

点击这里看k8s官方文档

1.1 k8s设置livenessProbe

livenessProbe:
  exec:
    command:
    - /root/rpc_check
    - -a
    - 127.0.0.1:19000
  initialDelaySeconds: 2
  periodSeconds: 2
复制代码

配置成功以后,k8s会每2秒执行 /root/rpc_check -a 127.0.0.1:19000, 执行成功的话表明存活,不成功的话k8s会重启pod

二. 建立容器

2.1 建立项目 $GOPATH/src/grpc-demo

grpc健康检查客户端

# $GOPATH/src/grpc-demo/cmd/check/main.go
package main
import(
	"os"
	"log"
	"time"
	"errors"
	"context"
	"path/filepath"
	"google.golang.org/grpc"
	cli "gopkg.in/urfave/cli.v1"
	pb "google.golang.org/grpc/health/grpc_health_v1"
)
const (
	VERSION  = "1.0.1"
	USAGE    = "grpc health check client"
)
var app *cli.App
func init(){
	app 	    = cli.NewApp()
	app.Name 	= filepath.Base(os.Args[0])
	app.Version = VERSION
	app.Usage 	= USAGE
	app.Flags = []cli.Flag{
		cli.StringFlag{Name: "address, a", Usage: "请求地址"},
		cli.StringFlag{Name: "service, s", Usage: "请求参数service", Value: "NULL"},
	}
	app.Action 	= func(ctx *cli.Context) error {
		a := ctx.GlobalString("address")
		s := ctx.GlobalString("service")
		if a == "" {
			log.Fatalln("Missing address parameter! see --help")
			return errors.New("Missing address parameter! see --help")
		}
		conn, err := grpc.Dial(a, grpc.WithInsecure())
		if err != nil {
			log.Fatalf("did not connect: %v", err)
			return err
		}
		defer conn.Close()
		f := pb.NewHealthClient(conn)
		c, cancel := context.WithTimeout(context.Background(), time.Second * 30)
		defer cancel()
		r, err := f.Check(c, &pb.HealthCheckRequest{
			Service: s,
		})
		if err != nil {
			log.Fatalf("could not greet: %v", err)
			return err
		}
		log.Println(r)
		return nil
	}
}
func main() {
	if err := app.Run(os.Args); err != nil {
		os.Exit(1)
	}
}
复制代码

grpc健康检查服务端

# $GOPATH/src/grpc-demo/cmd/server/main.go
package main
import (
    "os"
	"net"
	"log"
	"strconv"
	"syscall"
	"errors"
	"context"
	"os/signal"
	"path/filepath"
	"google.golang.org/grpc"
	
	"grpc-demo/app/health"
	cli "gopkg.in/urfave/cli.v1"
	pb "google.golang.org/grpc/health/grpc_health_v1"
)
const (
	VERSION = "1.0.1"
	USAGE   = "grpc health check server"
)
var app *cli.App
func init(){
	app 	    = cli.NewApp()
	app.Name 	= filepath.Base(os.Args[0])
	app.Version = VERSION
	app.Usage 	= USAGE
	app.Flags = []cli.Flag{
		cli.UintFlag{Name: "port, p", 	Usage: "端口"},
	}
	app.Action 	= func(ctx *cli.Context) error {
		p := ctx.GlobalUint("port")
		if p == 0 {
			log.Fatalf("Missing port!")
			return errors.New("Missing port!")
		}
		grpcServer := grpc.NewServer()
		lis, err := net.Listen("tcp", ":"+strconv.Itoa(int(p)))
		if err != nil {
			log.Fatalf("Failed to listen:%+v",err)
			return err
		}
		pb.RegisterHealthServer(grpcServer, health.New())
		go func() {
			sigs := make(chan os.Signal, 1)
			signal.Notify(sigs, syscall.SIGTERM, syscall.SIGINT)
			_ = <-sigs
			grpcServer.GracefulStop()
		}()	
		log.Printf("service started")
		if err := grpcServer.Serve(lis); err != nil {
			log.Fatalf("Failed to serve: %+v", err)
			return err
		}
		return nil
	}
}
func main() {
	if err := app.Run(os.Args); err != nil {
		os.Exit(1)
	}
}
复制代码

健康检查实现方法 health.go

# $GOPATH/src/grpc-demo/app/health.go
package health
import(
	"log"
	"context"
	pb "google.golang.org/grpc/health/grpc_health_v1"
)
type Health struct{}
func New() *Health {
	return &Health{}
}
func (h *Health) Check(ctx context.Context, in *pb.HealthCheckRequest)(*pb.HealthCheckResponse, error){
	log.Printf("checking............%s", in.Service)
	var s pb.HealthCheckResponse_ServingStatus = 1
	return &pb.HealthCheckResponse{
		Status : s,
	}, nil
}
func (h *Health) Watch(in *pb.HealthCheckRequest, w pb.Health_WatchServer)(error){
	log.Printf("watching............%s", in.Service)
	var s pb.HealthCheckResponse_ServingStatus = 1
	r := &pb.HealthCheckResponse{
		Status : s,
	}
	for {
		w.Send(r)
	}
	return nil
}
复制代码

编绎

go build -o rpc_srv  $GOPATH/src/grpc-demo/cmd/server/*.go
go build -o rpc_check  $GOPATH/src/grpc-demo/cmd/check/*.go
复制代码

Dockerfile

FROM ubuntu:16.04
ADD rpc_srv /root/rpc_srv
ADD rpc_check /root/rpc_check
RUN chmod +x /root/rpc_srv && chmod +x /root/rpc_check
EXPOSE 19000
CMD /root/rpc_srv -p 19000
复制代码

总结

k8s实现grpc健康检查的方法跟envoy很是像, 就是调用服务中实现的health_check方法, 只是envoy集成了调用, 而k8s要本身写调用程序或使用grpc-health-probe. k8s+nginx+consul能够组成一个比较成熟的grpc服务发现与服务治理方案