wmj-test/src/k8s_client.go

package k8s_manager

import (
	"context"
	"fmt"
	log "github.com/sirupsen/logrus"
	"github.com/thoas/go-funk"
	appv1 "k8s.io/api/apps/v1"
	v1 "k8s.io/api/core/v1"
	netv1 "k8s.io/api/networking/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/selection"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"os"
	"strings"
)

type K8sClient struct {
	*kubernetes.Clientset
	Namespace string
}

func CreateClient() *K8sClient {
	var err error
	var config *rest.Config
	if os.Getenv("KUBE_CONFIG") != "" {
		config, err = clientcmd.BuildConfigFromFlags("", os.Getenv("KUBE_CONFIG"))
	} else {
		config, err = rest.InClusterConfig()
	}
	if err != nil {
		log.WithError(err).Error("InClusterConfig")
		return nil
	}

	client, err := kubernetes.NewForConfig(config)
	if err != nil {
		log.WithError(err).Error("NewForConfig")
	}
	return &K8sClient{client, "apps"}
}

func (c *K8sClient) ListPods() {
	pods, err := c.CoreV1().Pods("").List(context.Background(), metav1.ListOptions{})
	if err != nil {
		log.WithError(err).Error("ListPods")
	} else {
		fmt.Printf("There are %d pods in the cluster\n", len(pods.Items))
	}
}

func (c *K8sClient) GetDeploymentStatus(deploymentName string) (*appv1.DeploymentStatus, error) {
	get, err := c.AppsV1().Deployments(c.Namespace).Get(context.Background(), deploymentName, metav1.GetOptions{})
	// log.WithError(err).WithField("dep", get).WithField("name", app.DeploymentName()).Info("GetAppStatus")
	return &get.Status, err
}

func (c *K8sClient) DeploymentExists(deploymentName string) bool {
	_, e := c.GetDeploymentStatus(deploymentName)
	if e == nil {
		return true
	}
	if !strings.Contains(e.Error(), "not found") {
		log.WithError(e).Error("DeploymentExists")
	}
	return false
}

func (c *K8sClient) UpdateDeployment(app *AppModel) {
	var rep int32 = 1
	labelKey, labelValue := app.GetLabel()
	l := map[string]string{labelKey: labelValue}
	podLabels := map[string]string{
		labelKey:        labelValue,
		"cg-deployment": app.DeploymentName(),
	}
	deployment := &appv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name:   app.DeploymentName(),
			Labels: l,
		},
		Spec: appv1.DeploymentSpec{
			Replicas: &rep,
			Selector: &metav1.LabelSelector{
				MatchLabels: l,
			},
			Template: v1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: podLabels,
				},
				Spec: v1.PodSpec{
					Containers: []v1.Container{{
						Name:         app.ContainerName(),
						Image:        app.Image,
						Command:      app.GetCommandOrBlack(),
						Env:          app.GetEnv(),
						VolumeMounts: nil,
						Ports: []v1.ContainerPort{
							{ContainerPort: 80},
						},
						SecurityContext: &v1.SecurityContext{
							RunAsUser: UnnamedPointer(int64(0)),
						},
					}},
				},
			},
		},
	}
	var err error
	if c.DeploymentExists(app.DeploymentName()) {
		deployment, err = c.AppsV1().Deployments(c.Namespace).Update(context.Background(), deployment, metav1.UpdateOptions{})
	} else {
		deployment, err = c.AppsV1().Deployments(c.Namespace).Create(context.Background(), deployment, metav1.CreateOptions{})
	}
	if err != nil {
		//d, _ := json.Marshal(deployment)
		//fmt.Println(string(d))
		log.WithError(err).Error("Update Deployment error.")
	}
}

func (c *K8sClient) ServiceExists(servername string) bool {
	_, err := c.CoreV1().Services(c.Namespace).Get(context.Background(), servername, metav1.GetOptions{})
	if err == nil {
		return true
	}
	if !strings.Contains(err.Error(), "not found") {
		log.WithError(err).Error("ServiceExists")
	}
	return false
}

func (c *K8sClient) UpdateService(app *AppModel) {
	ports := app.GetPorts()
	if len(ports) == 0 {
		return
	}
	servicePorts := funk.Map(ports, func(a int32) v1.ServicePort {
		return v1.ServicePort{
			Port:       a,
			TargetPort: intstr.FromInt(int(a))}
	}).([]v1.ServicePort)
	labelKey, labelValue := app.GetLabel()
	l := map[string]string{labelKey: labelValue}
	service := &v1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:   app.ServiceName(),
			Labels: l,
		},
		Spec: v1.ServiceSpec{
			Type:      v1.ServiceTypeClusterIP,
			ClusterIP: "",
			Selector:  l,
			Ports:     servicePorts,
		},
	}
	if app.ServiceName() != app.DefaultServiceName() {
		if c.ServiceExists(app.DefaultServiceName()) {
			err := c.CoreV1().Services(c.Namespace).Delete(context.Background(), app.DefaultServiceName(), metav1.DeleteOptions{})
			if err != nil {
				log.WithError(err).Error("UpdateService Delete error")
			}
		}
	}
	var err error
	if c.ServiceExists(app.ServiceName()) {
		_, err = c.CoreV1().Services(c.Namespace).Update(context.Background(), service, metav1.UpdateOptions{})
	} else {
		_, err = c.CoreV1().Services(c.Namespace).Create(context.Background(), service, metav1.CreateOptions{})
	}
	if err != nil {
		log.WithError(err).Error("UpdateService")
	}

}

func (c *K8sClient) CreateApp(app *AppModel) {
	c.UpdateDeployment(app)
	c.UpdateService(app)
	c.UpdateIngress(app)
}

func (c *K8sClient) DeleteApp(app *AppModel) {
	c.DeleteDeployment(app.DeploymentName())
	c.DeleteAppService(app)
	c.DeleteIngress(app.IngressName())
}

func (c *K8sClient) DeleteAppService(app *AppModel) {
	k, v := app.GetLabel()
	label := labels.FormatLabels(map[string]string{k: v})
	svcs, err := c.CoreV1().Services(c.Namespace).List(context.Background(), metav1.ListOptions{
		LabelSelector: label,
	})
	if err != nil {
		log.WithError(err).Error("DeleteAppService Get")
		return
	}
	for _, svc := range svcs.Items {
		err := c.CoreV1().Services(c.Namespace).Delete(context.Background(), svc.Name, metav1.DeleteOptions{})
		if err != nil {
			log.WithError(err).WithField("name", svc.Name).Error("Delete svc")
		}
	}
}

func (c *K8sClient) DeleteDeployment(depName string) {
	if !c.DeploymentExists(depName) {
		return
	}
	err := c.AppsV1().Deployments(c.Namespace).Delete(context.Background(), depName, metav1.DeleteOptions{})
	if err != nil {
		log.WithError(err).Error("Delete Deployment")
	}
}

func (c *K8sClient) DeleteService(svcName string) {
	if !c.ServiceExists(svcName) {
		return
	}
	err := c.CoreV1().Services(c.Namespace).Delete(context.Background(), svcName, metav1.DeleteOptions{})
	if err != nil {
		log.WithError(err).Error("Delete Service")
	}
}

func (c *K8sClient) GetAllServices() *v1.ServiceList {
	req, _ := labels.NewRequirement("cg-label", selection.Exists, []string{})
	list, err := c.CoreV1().Services(c.Namespace).List(context.Background(), metav1.ListOptions{LabelSelector: req.String()})
	if err != nil {
		log.WithError(err).Error("GetAllServices")
	}
	return list
}

func (c *K8sClient) GetPodFromDeployment(deploymentName string) *v1.Pod {
	label := labels.FormatLabels(map[string]string{"cg-deployment": deploymentName})
	pods, err := c.CoreV1().Pods(c.Namespace).List(context.Background(), metav1.ListOptions{LabelSelector: label})
	if err != nil {
		log.WithError(err).Error("GetPodFromDeployment")
		return nil
	}
	if len(pods.Items) == 0 {
		log.WithField("label", label).Error("GetPodFromDeployment no such label pod.")
		return nil
	}
	return &pods.Items[0]
}

func (c *K8sClient) IngressExists(name string) bool {
	_, err := c.NetworkingV1().Ingresses(c.Namespace).Get(context.Background(), name, metav1.GetOptions{})
	if err == nil {
		return true
	}
	if !strings.Contains(err.Error(), "not found") {
		log.WithError(err).Error("IngressExists")
	}
	return false
}

func (c *K8sClient) UpdateIngress(a *AppModel) {
	ports := a.GetPorts()
	if len(ports) == 0 {
		return
	}
	exportPort := ports[0]
	if exportPort == 0 {
		return
	}
	pt := netv1.PathTypePrefix
	//nginx := "nginx"
	k, v := a.GetLabel()
	label := map[string]string{k: v}
	ingress := &netv1.Ingress{
		ObjectMeta: metav1.ObjectMeta{
			Name:        a.IngressName(),
			Labels:      label,
			Annotations: map[string]string{
				//"traefik.ingress.kubernetes.io/router.middlewares": "default-strip-prefix-1@kubernetescrd",
			},
		},
		Spec: netv1.IngressSpec{
			Rules: []netv1.IngressRule{
				{
					//Host: a.ServiceName(),
					IngressRuleValue: netv1.IngressRuleValue{
						HTTP: &netv1.HTTPIngressRuleValue{
							Paths: []netv1.HTTPIngressPath{
								{
									Path:     "/" + a.ServiceName(),
									PathType: &pt,
									Backend: netv1.IngressBackend{
										Service: &netv1.IngressServiceBackend{
											Name: a.ServiceName(),
											Port: netv1.ServiceBackendPort{
												Number: exportPort,
											},
										},
									},
								},
							},
						},
					},
				},
			},
			//IngressClassName: &nginx,
		},
	}
	var err error
	if c.IngressExists(a.IngressName()) {
		_, err = c.NetworkingV1().Ingresses(c.Namespace).Update(context.Background(), ingress, metav1.UpdateOptions{})
	} else {
		_, err = c.NetworkingV1().Ingresses(c.Namespace).Create(context.Background(), ingress, metav1.CreateOptions{})
	}
	if err != nil {
		log.WithError(err).Error("Create Ingress")
	}
}

func (c *K8sClient) DeleteIngress(name string) {
	if !c.IngressExists(name) {
		return
	}
	err := c.NetworkingV1().Ingresses(c.Namespace).Delete(context.Background(), name, metav1.DeleteOptions{})
	if err != nil {
		log.WithError(err).Error("Delete Ingress")
	}
}