k8s编程operator系列：
k8s编程operator——(1) client-go基础部分
k8s编程operator——(2) client-go中的informer
k8s编程operator——(3) 自定义资源CRD
k8s编程operator——(4) kubebuilder & controller-runtime
k8s编程operator实战之云编码平台——①架构设计
k8s编程operator实战之云编码平台——②controller初步实现
 
在上一章中介绍了整个云编码平台的架构设计，本章中将实现架构中的controller。需要使用的技术： k8s二次开发kubebuilder、 grpc

controller用来创建、删除、获取Pod信息以及状态的维护。

 

1、工作空间镜像制作

        首先我们要制作code-server的docker镜像，先来制作一个含有go运行环境的工作空间镜像，工具包括：go sdk、make、git

首先创建一个文件夹用来存放我们制作镜像需要的文件：

mkdir go_template

 

1、下载code-server

下载code-server v4.9.0版本：

下载的网址为：https://github.com/coder/code-server/releases

wget https://github.com/coder/code-server/releases/download/v4.9.0/code-server-4.9.0-linux-amd64.tar.gz

2、下载go sdk，版本为v1.19.4

下载地址：https://golang.google.cn/dl/

wget https://golang.google.cn/dl/go1.19.4.linux-amd64.tar.gz

3、编写Dockerfile

Dockerfile如下：

• 基础镜像使用ubuntu:20.04
• 14、15行为安装必要的软件
• 16行安装git
• 在启动code-server时不使用密码验证，默认打开/root/workspace文件

FROM ubuntu:20.04

WORKDIR /.workspace

COPY code-server-4.9.0-linux-amd64.tar.gz .
COPY go1.19.4.linux-amd64.tar.gz .

RUN tar zxvf code-server-4.9.0-linux-amd64.tar.gz && \
    rm -f code-server-4.9.0-linux-amd64.tar.gz    && \
    tar zxvf go1.19.4.linux-amd64.tar.gz -C /usr/local     && \
    rm -f go1.19.4.linux-amd64.tar.gz             && \
    mkdir -p /root/workspace /go/{src,pkg,bin}    && \
    apt-get -y update                             && \
    apt-get -qq update                            && \
    apt-get -qq install -y --no-install-recommends ca-certificates curl && \
    apt install git

ENV GO111MODULE on
ENV GOPROXY https://goproxy.cn,direct
ENV GOROOT /usr/local/go
ENV PATH /usr/local/go/bin:$PATH
ENV GOPATH /go
ENV PATH $GOPATH/bin:$PATH

WORKDIR /.workspace/code-server-4.9.0-linux-amd64

EXPOSE 9999

CMD ["./bin/code-server", "--port", "9999", "--host", "0.0.0.0", "--auth", "none", "--disable-update-check", "--open", "/root/workspace/"]

4、构建镜像

在go_template中构建镜像，需要将code-server、go sdk和Docker都放在该文件夹下，也不要放多余的其它文件

docker build -t code-server-go1.19:v0.1 .

构建查看镜像（我的镜像名为mangohow/code-server-go1.19）：

5、push到docker hub

在构建完成后，最好将镜像推送到dockerhub或者其它镜像仓库中

首先要给镜像打一个tag，yourUsername为你的镜像仓库的用户名

docker tag code-server-go1.19:v0.1 yourUsername/code-server-go1.19:v0.1

推送到镜像仓库

# 先登录
docker login
docker push yourUsername/code-server-go1.19:v0.1

 

2、controller实现

controller的实现直接使用kubebuilder，它会为我们生成一套模板代码

注意：最好直接在linux上编码，在windows上有很多问题

 

2.1 使用kubebuilder创建工程

1、创建文件夹

mkdir cloud-ide-k8s-controller
cd cloud-ide-k8s-controller/

2、创建工程

# 初始化
go mod init cloud-ide-k8s-controller
kubebuilder init

# 创建api
kubebuilder create api --group cloud-ide  --version v1 --kind Pod

由于我们没有创建自定义资源，所以在Create Resource选择中选择n

到此，工程已经创建完毕，开始编码

 

2.2 代码实现

2.2.1 引入grpc

Pod的创建、删除以及信息查询等工作我们都在controller中实现，然后通过grpc的方式提供给web后端来使用。

使用grpc需要安装protobuf和proto-gen-go，关于这部分的安装在此就不再介绍，网上教程很多。

1、grpc的proto文件定义

首先先实现四个grpc service：

createSpaceAndWaitForRunning：创建Pod并且等待Pod的状态变为Running，当我们在创建Pod后，它需要一段时间来启动，在这段时间之间是无法访问的，因此我们需要等待它启动完成后再返回
deleteSpace：删除Pod
getPodSpaceStatus：获取Pod运行的状态
getPodSpaceInfo：获取Pod的信息

在工程中创建pb/proto文件夹，然后创建proto文件：

service.proto如下：

syntax = "proto3";

package pb;

option go_package = "./;pb";

// 限制用户的工作空间的资源使用，有CPU、内存和存储的大小限制
message ResourceLimit {
  string cpu = 1;
  string Memory = 2;
  string Storage = 3;
}

// 在创建Pod时，需要webserver提供要创建的Pod的信息
// 其中有：Pod的名称、Pod的命名空间、使用的镜像、以及要使用的端口和资源限制
message PodInfo {
  string name = 1;
  string namespace = 2;
  string image = 3;
  uint32 port = 4;
  ResourceLimit resourceLimit = 5;
}

message Response {
  int32 status = 1;
  string message = 2;
}

message QueryOption {
  string name = 1;
  string namespace = 2;
}

message PodStatus {
  int32 status = 1;
  string message = 2;
}

message PodSpaceInfo {
  string nodeName = 1;
  string ip = 2;
  int32 port = 3;
}

service CloudIdeService {
  // 创建云IDE空间并等待Pod状态变为Running
  rpc createSpaceAndWaitForRunning(PodInfo) returns (PodSpaceInfo);
  // 删除云IDE空间
  rpc deleteSpace(QueryOption) returns (Response);
  // 获取Pod运行状态
  rpc getPodSpaceStatus(QueryOption) returns (PodStatus);
  // 获取云IDE空间Pod的信息
  rpc getPodSpaceInfo(QueryOption) returns (PodSpaceInfo);
}

生成代码：

protoc --go_out=plugins=grpc:./pb ./pb/proto/*.proto

可以将这个命令加到makefile中，以后在使用就更方便了，可以直接使用make proto

 

2.2.2 实现CloudIdeService

创建cloudspaceservice.go实现CloudIdeService接口

要实现的方法总共有四个：

type CloudIdeServiceClient interface {
	// 创建云IDE空间并等待Pod状态变为Running
	CreateSpaceAndWaitForRunning(ctx context.Context, in *PodInfo, opts ...grpc.CallOption) (*PodSpaceInfo, error)
	// 删除云IDE空间
	DeleteSpace(ctx context.Context, in *QueryOption, opts ...grpc.CallOption) (*Response, error)
	// 获取Pod运行状态
	GetPodSpaceStatus(ctx context.Context, in *QueryOption, opts ...grpc.CallOption) (*PodStatus, error)
	// 获取云IDE空间Pod的信息
	GetPodSpaceInfo(ctx context.Context, in *QueryOption, opts ...grpc.CallOption) (*PodSpaceInfo, error)
}

CreateSpaceAndWaitForRunning：给ApiServer发送请求创建Pod，并且等待Pod处于运行状态后在返回。

DeleteSpace：删除指定的Pod

GetPodSpaceStatus：获取Pod的运行状态，Pod是否存在

GetPodSpaceInfo：获取Pod的ip和port信息

等待Pod处于运行状态

Pod在刚创建时处于Pending状态，当Pod启动完毕后，会处于Running状态，因为我们要监视Pod的状态，当它的状态变为Runnging时，就通知我们的CreateSpaceAndWaitForRunning方法Pod已经就绪。

要监视Pod的状态可以通过client go的informer来实现，使用kubebuilder生成的工程中已经有现成的模板代码，也就是我们只需要在controller中实现Reconcile方法即可。

我们可以实现一个通知器，其中含有map[string]chan struct{}，当我们在CreateSpaceAndWaitForRunning方法中请求ApiServer创建Pod后，就向map中添加一个键值对 key:podname val: chan，然后从chan中读取数据，由于Pod还未准备就绪，因此读取空的chan将会导致阻塞。在Reconcile中当监视到Pod的状态变为Running后，就向chan中发送消息。另一端就可以收到Pod已经就绪的通知。

StatusInformer的实现

package statussync

import (
	"errors"
	"sync"
)

var (
	ErrNotFound = errors.New("not Found")
)

// StatusInformer 状态同步通知器，当Pod状态处于Running时，通知对端
type StatusInformer struct {
	sync.Mutex
	m map[string]chan struct{}
}

func NewManager() *StatusInformer {
	return &StatusInformer{
		m: make(map[string]chan struct{}),
	}
}

// 向map中添加一个chan
func (m *StatusInformer) Add(name string) <-chan struct{} {
	m.Lock()
	defer m.Unlock()
	ch := make(chan struct{}, 1)
	m.m[name] = ch

	return ch
}

func (m *StatusInformer) Delete(name string) {
	m.Lock()
	defer m.Unlock()
	delete(m.m, name)
}

// 同步消息
func (m *StatusInformer) Sync(name string) error {
	m.Lock()
	defer m.Unlock()
	ch, ok := m.m[name]
	if !ok {
		return ErrNotFound
	}
	ch <- struct{}{}

	return nil
}

 

CloudIdeService接口实现：

CloudSpaceService定义

在CloudSpaceService的字段有三个，分别是：client、logger和statusInformer

client可以对Pod进行CRUD，在查询时会查询本地的缓存

package service

import (
	"context"
	"fmt"
	"github.com/go-logr/logr"
	"github.com/mangohow/cloud-ide-k8s-controller/pb"
	"github.com/mangohow/cloud-ide-k8s-controller/tools/statussync"
	v1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

var (
	ResponseSuccess = &pb.Response{Status: 200, Message: "success"}
	ResponseFailed  = &pb.Response{Status: 400, Message: "failed"}
)

const (
	PodNotExist int32 = iota
	PodExist
)

type CloudSpaceService struct {
	client         client.Client            // client可用于pod的CRUD
	logger         *logr.Logger
	statusInformer *statussync.StatusInformer
}

func NewCloudSpaceService(client client.Client, logger *logr.Logger, manager *statussync.StatusInformer) *CloudSpaceService {
	return &CloudSpaceService{
		client:         client,
		logger:         logger,
		statusInformer: manager,
	}
}

var podTpl = &v1.Pod{
	TypeMeta: metav1.TypeMeta{
		Kind:       "Pod",
		APIVersion: "v1",
	},
	ObjectMeta: metav1.ObjectMeta{
		Labels: map[string]string{
			"kind": "cloud-ide",
		},
	},
}

方法CreateSpaceAndWaitForRunning

在Pod的定义中一定要将镜像的拉取策略改为IfNotPresent，之前这个字段没有填。每次在创建工作空间的时候，都要等30~40s，后来我才发现，默认的策略是Always，也就是每次创建工作空间的时候都会从dockerhub拉取镜像，所以才导致这么慢。修改后，启动工作空间就只需要不到5s。

在Resources中，我将代码注释了，因为我的虚拟机配置太低，如果不注释，那么在创建Pod时就可能会因为资源不足而导致Pod创建失败。

// CreateSpaceAndWaitForRunning 创建一个云IDE空间, 并等待Pod的状态变为Running
func (s *CloudSpaceService) CreateSpaceAndWaitForRunning(ctx context.Context, info *pb.PodInfo) (*pb.PodSpaceInfo, error) {
	// 1、获取一个Pod的深拷贝
    pod := podTpl.DeepCopy()
    // 2、填充参数
	s.fillPod(info, pod)
    // 3、创建Pod
	err := s.client.Create(context.Background(), pod)
	if err != nil {
		fmt.Printf("create pod:%s, info:%v\n", err.Error(), info)
		return nil, err
	}
	// 4、向informer中添加chan，当Pod准备就绪时就会收到通知
	ch := s.statusInformer.Add(pod.Name)
	// 从informer中删除
	defer s.statusInformer.Delete(pod.Name)
	// 等待pod状态处于Running
	<-ch
	
    // 返回Pod的信息
	return s.GetPodSpaceInfo(context.Background(), &pb.QueryOption{
		Name:      info.Name,
		Namespace: info.Namespace,
	})
}

func (s *CloudSpaceService) fillPod(info *pb.PodInfo, pod *v1.Pod) {
	pod.Name = info.Name           // 指定Pod名称
	pod.Namespace = info.Namespace    // 指定Pod的命名空间
	pod.Spec.Containers = []v1.Container{
		{
			Name:            info.Name,     // 容器名称和Pod名称相同
			Image:           info.Image,    // 容器的镜像
			ImagePullPolicy: v1.PullIfNotPresent,      // 镜像拉取策略
			Ports: []v1.ContainerPort{
				{
					ContainerPort: int32(info.Port),
				},
			},
			Resources: v1.ResourceRequirements{
				//Limits: map[v1.ResourceName]resource.Quantity{
				//	v1.ResourceCPU:    resource.MustParse(info.ResourceLimit.Cpu),
				//	v1.ResourceMemory: resource.MustParse(info.ResourceLimit.Memory),
				//},
				// 最小需求CPU2核、内存4Gi == 4 * 2^10
				//Requests: map[v1.ResourceName]resource.Quantity{
				//	v1.ResourceCPU:    resource.MustParse("2"),
				//	v1.ResourceMemory: resource.MustParse("4Gi"),
				//},
			},
		},
	}

}

方法GetPodSpaceInfo

根据Pod名称以及命名空间查询Pod的Ip和port

func (s *CloudSpaceService) GetPodSpaceInfo(ctx context.Context, option *pb.QueryOption) (*pb.PodSpaceInfo, error) {
	pod := v1.Pod{}
	err := s.client.Get(context.Background(), client.ObjectKey{Name: option.Name, Namespace: option.Namespace}, &pod)
	if err != nil {
		s.logger.Error(err, "get pod space info")
		return &pb.PodSpaceInfo{}, err
	}

	return &pb.PodSpaceInfo{NodeName: pod.Spec.NodeName,
		Ip:   pod.Status.PodIP,
		Port: pod.Spec.Containers[0].Ports[0].ContainerPort}, nil
}

方法DeleteSpace

根据Pod的名称和命名空间删除Pod

// DeleteSpace 删除一个云空间
func (s *CloudSpaceService) DeleteSpace(ctx context.Context, option *pb.QueryOption) (*pb.Response, error) {
	pod := v1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      option.Name,
			Namespace: option.Namespace,
		},
	}
	err := s.client.Delete(context.Background(), &pod)
	if err != nil {
		s.logger.Error(err, "delete space")
		return ResponseFailed, err
	}

	return ResponseSuccess, nil
}

方法GetPodSpaceStatus

根据Pod的名称和命名空间获取Pod的运行状态

func (s *CloudSpaceService) GetPodSpaceStatus(ctx context.Context, option *pb.QueryOption) (*pb.PodStatus, error) {
	pod := v1.Pod{}
	err := s.client.Get(context.Background(), client.ObjectKey{Name: option.Name, Namespace: option.Namespace}, &pod)
	if err != nil {
		s.logger.Error(err, "get pod space status")
		return &pb.PodStatus{Status: PodNotExist, Message: "NotExist"}, err
	}

	return &pb.PodStatus{Status: PodExist, Message: string(pod.Status.Phase)}, nil
}

 

2.2.3 controller实现

在pod_controller文件中，需要实现的方法主要有两个：Reconcile和SetupWithManager

type PodReconciler struct {
	client.Client
	Scheme         *runtime.Scheme
	statusInformer *statussync.StatusInformer   // 添加statusInformer
}

SetupWithManager

在SetupWithManager方法中需要指定我们要监视的资源的类型，也就是Pod：

func (r *PodReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		// Uncomment the following line adding a pointer to an instance of the controlled resource as an argument
		For(&v1.Pod{}).
		Complete(r)
}

Reconcile

在Reconciler中监视Pod的状态，当Pod处于running状态时，通知对端

func (r *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	// TODO(user): your logic here
	pod := &v1.Pod{}
    // 获取Pod
	err := r.Client.Get(context.Background(), req.NamespacedName, pod)
	if err != nil {
		if !errors.IsNotFound(err) {
			logger.Error(err, "get pod")
			return ctrl.Result{Requeue: true}, err
		}
	}
	fmt.Printf("name:%s, status:%s\n", pod.Name, pod.Status.Phase)
	// 如果Pod的状态处于Running，通知对端Pod已经处于Running状态
	if pod.Status.Phase == v1.PodRunning {
		r.statusInformer.Sync(pod.Name)
	}

	return ctrl.Result{}, nil
}

 

2.2.4 启动grpc server以及优雅退出服务器

我们需要在main函数中启动我们的grpc服务器并注册服务。处理信号，当接收到退出的信号时退出grpc server和controller

StartGrpcServer

func StartGrpcServer(client client.Client, logger *logr.Logger, manager *statussync.StatusInformer) *grpc.Server {
	// 创建listener
    listener, err := net.Listen("tcp", ":6387")
	if err != nil {
		panic(fmt.Errorf("create grpc service: %v", err))
	}
    // 创建grpc server
	server := grpc.NewServer()
    // 注册我们的服务
	pb.RegisterCloudIdeServiceServer(server, service.NewCloudSpaceService(client, logger, manager))
	
    // 启动grpc server
	go func() {
		err := server.Serve(listener)
		if err != nil && err == grpc.ErrServerStopped {
			fmt.Printf("server stopped")
		} else if err != nil {
			panic(fmt.Errorf("start grpc server: %v", err))
		}
	}()

	return server
}

SetupSignal

signal.go：

package signal

import (
	"context"
	"os"
	"os/signal"
	"syscall"
)

var onlyOneSignalHandler = make(chan struct{})

func SetupSignal(fns ...func()) context.Context {
	// 当函数被调用两次，就会panic
	close(onlyOneSignalHandler)
	sigCh := make(chan os.Signal, 2)
   
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		<-sigCh
		for _, fn := range fns {
			fn()
		}
		cancel()
        // 第二次接收到信号，直接退出
		<-sigCh
		os.Exit(1)
	}()

	return ctx
}

main函数如下：

const WatchedNamespace = "cloud-ide"

func main() {
	var metricsAddr string
	var enableLeaderElection bool
	var probeAddr string
	flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.")
	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
	flag.BoolVar(&enableLeaderElection, "leader-elect", false,
		"Enable leader election for controller manager. "+
			"Enabling this will ensure there is only one active controller manager.")
	opts := zap.Options{
		Development: true,
	}
	opts.BindFlags(flag.CommandLine)
	flag.Parse()

	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))

	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
		Scheme:                 scheme,
		MetricsBindAddress:     metricsAddr,
		Port:                   9443,
		HealthProbeBindAddress: probeAddr,
		LeaderElection:         enableLeaderElection,
		LeaderElectionID:       "81275557.my.domain",
		// LeaderElectionReleaseOnCancel defines if the leader should step down voluntarily
		// when the StatusInformer ends. This requires the binary to immediately end when the
		// StatusInformer is stopped, otherwise, this setting is unsafe. Setting this significantly
		// speeds up voluntary leader transitions as the new leader don't have to wait
		// LeaseDuration time first.
		//
		// In the default scaffold provided, the program ends immediately after
		// the manager stops, so would be fine to enable this option. However,
		// if you are doing or is intended to do any operation such as perform cleanups
		// after the manager stops then its usage might be unsafe.
		// LeaderElectionReleaseOnCancel: true,
		Namespace: WatchedNamespace,  // 指定要监视的Pod的namespace，我们的工作空间都创建在同一个命名空间下，只需监视这个命名空间下的Pod即可
	}) 
	if err != nil {
		setupLog.Error(err, "unable to start manager")
		os.Exit(1)
	}
	
    // 创建statusInfomer
	manager := statussync.NewManager()
	if err = controllers.NewPodReconciler(mgr.GetClient(), mgr.GetScheme(), manager).SetupWithManager(mgr); err != nil {
		setupLog.Error(err, "unable to create controller", "controller", "Pod")
		os.Exit(1)
	}
	//+kubebuilder:scaffold:builder

	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
		setupLog.Error(err, "unable to set up health check")
		os.Exit(1)
	}
	if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
		setupLog.Error(err, "unable to set up ready check")
		os.Exit(1)
	}

	// 启动grpc服务
	grpcServer := StartGrpcServer(mgr.GetClient(), &ctrl.Log, manager)
	// 安装信号处理
	ctx := signal.SetupSignal(func() {
		ctrl.Log.Info("receive signal, is going to shutdown")
		grpcServer.GracefulStop()
	})

	setupLog.Info("starting manager")

	if err := mgr.Start(ctx); err != nil {
		setupLog.Error(err, "problem running manager")
		os.Exit(1)
	}
}

 

2.2.5 目录结构

目录结构如下：

 

3、测试

接下来测试各个功能，由于我们的服务是通过grpc的方式提供的。可以使用ApiPost7来发送请求，ApiPost7支持grpc接口的调用，免费的。

官网：https://www.apipost.cn/

3.1 创建工作空间并且等待运行

启动我们的controller，可以使用make来启动：

make run
# 或者编译后再启动,编译后的文件为bin/manager
make build

使用ApiPost创建一个grpc，将service.proto导入

调用CreateSpaceAndWaitForRunning接口创建pod

controller的输出如下：

查看pod：

 

3.2 获取Pod状态

 

3.3 获取pod信息

 

3.4 删除pod

 

总结

至此，controller已经初步实现了功能，可以实现Pod的创建、删除、Pod状态获取、Pod信息获取

github地址：https://github.com/mangohow/cloud-ide-k8s-controller