集群管理模块 原创

集群管理模块是用来管理多个集群的情况，可以对其进行增删改查，还可以通过该模块来切换集群。

主要涉及的功能有：

• 增加集群
• 删除集群
• 更新集群
• 列出集群
• 初始化集群

用户可以管理多个集群，并且集群是可以切换的。

1、新增 models

在 internal/app/models 目录下创建 k8s_cluster.go，增加 K8s 集群的 model，如下：

package models

// K8sCluster K8s集群结构体
type K8sCluster struct {
	ClusterName    string `json:"cluster_name" description:"集群名"`
	ClusterVersion string `json:"cluster_version" description:"集群版本"`
	KubeConfig     string `json:"kube_config" description:"KubeConfig文本"`
	Status         int    `json:"status" description:"集群状态"`
	*Base
}

func (k *K8sCluster) TableName() string {
	return "k8s_cluster"
}

为 K8s 集群定义了 cluster_name，kube_config，cluster_version，status 几个字段，用它们来记录 K8s 集群信息。

再在该目录下增加如下方法，以便后续调用，如下：

package models

import "gorm.io/gorm"

// K8sCluster K8s集群结构体
type K8sCluster struct {
	ClusterName    string `json:"cluster_name" description:"集群名"`
	ClusterVersion string `json:"cluster_version" description:"集群版本"`
	KubeConfig     string `json:"kube_config" description:"KubeConfig文本"`
	Status         int    `json:"status" description:"集群状态"`
	*Base
}

func (k *K8sCluster) TableName() string {
	return "k8s_cluster"
}

// Create 插入数据
func (k *K8sCluster) Create(db *gorm.DB) error {
	return db.Create(&k).Error
}

// GetByName 根据集群名获取集群信息
func (k *K8sCluster) GetByName(db *gorm.DB) (*K8sCluster, error) {
	var kc *K8sCluster
	if k.ClusterName != "" {
		db.Where("cluster_name =? and is_del=?", k.ClusterName, 0).First(&kc)
	}
	return kc, nil
}

// GetByID 通过ID获取集群信息
func (k *K8sCluster) GetByID(db *gorm.DB) (*K8sCluster, error) {
	var kc *K8sCluster
	db.Where("id =? and is_del=?", k.ID, 0).First(&kc)
	return kc, nil
}

// List 列出集群信息
func (k *K8sCluster) List(db *gorm.DB, page, limit int) ([]*K8sCluster, error) {
	var (
		kc  []*K8sCluster
		err error
	)

	startIndex := (page - 1) * limit
	db = db.Offset(startIndex).Limit(limit)

	if k.ClusterName != "" {
		db = db.Where("cluster_name=?", k.ClusterName)
	}

	if err = db.Where("is_del = ?", 0).Find(&kc).Error; err != nil {
		return nil, err
	}
	return kc, nil
}

// Update 更新数据
func (k *K8sCluster) Update(db *gorm.DB, values interface{}) error {
	if err := db.Model(k).Where("id = ? AND is_del = ?", k.ID, 0).First(&K8sCluster{}).Updates(values).Error; err != nil {
		return err
	}
	return nil
}

// Delete 删除数据
func (k *K8sCluster) Delete(db *gorm.DB) error {
	var kc K8sCluster
	if err := db.Where("id=? AND is_del=?", k.ID, 0).First(&kc).Error; err != nil {
		return err
	}
	kc.IsDel = 1
	if err := db.Updates(&kc).Error; err != nil {
		return err
	}
	return nil
}

// Save 保存数据
func (k *K8sCluster) Save(db *gorm.DB) error {
	return db.Save(k).Error
}

2、新增 dao 方法

定义好数据模型，下面就需要定义操作数据的 dao 方法，在我们的平台中，无非就是增删改查。

在 internal/app/dao 目录中新增 k8s_cluster.go，写入以下内容：

package dao

import (
	"time"

	"github.com/joker-bai/hawkeye/internal/app/models"
)

// K8sClusterCreate 创建集群
func (d *Dao) K8sClusterCreate(cluster_name, cluster_version, kube_config string) error {
	nowTime := uint32(time.Now().Unix())
	kc := models.K8sCluster{
		ClusterName:    cluster_name,
		ClusterVersion: cluster_version,
		KubeConfig:     kube_config,
		Status:         0,
		Base: &models.Base{
			CreatedAt:  nowTime,
			ModifiedAt: nowTime,
			IsDel:      0,
		},
	}
	return kc.Create(d.engine)
}

// K8sClusterGetByName 通过集群名称获取集群
func (d *Dao) K8sClusterGetByName(name string) (*models.K8sCluster, error) {
	kc := models.K8sCluster{
		ClusterName: name,
	}
	return kc.GetByName(d.engine)
}

func (d *Dao) K8sClusterGetById(id uint32) (*models.K8sCluster, error) {
	kc := models.K8sCluster{
		Base: &models.Base{
			ID: id,
		},
	}
	return kc.GetByID(d.engine)
}

// K8sClusterUpdate 更新集群
func (d *Dao) K8sClusterUpdate(id uint32, cluster_name, cluster_version, kube_config string, status int) error {
	nowTime := uint32(time.Now().Unix())
	kc := models.K8sCluster{
		Base: &models.Base{
			ID: id,
		},
	}
	values := map[string]interface{}{
		"cluster_name":    cluster_name,
		"cluster_version": cluster_version,
		"kube_config":     kube_config,
		"status":          status,
		"modified_at":     nowTime,
	}
	return kc.Update(d.engine, values)
}

// K8sClusterList 列出集群信息
func (d *Dao) K8sClusterList(cluster_name string, page, limit int) ([]*models.K8sCluster, error) {
	kc := models.K8sCluster{
		ClusterName: cluster_name,
	}
	return kc.List(d.engine, page, limit)
}

// K8sClusterDelete 删除集群信息
func (d *Dao) K8sClusterDelete(id uint32) error {
	kc := models.K8sCluster{
		Base: &models.Base{
			ID: id,
		},
	}
	return kc.Delete(d.engine)
}

3、新增 services 方法

3.1、新增入参校验

在 internal/app/requests 目录中新建 k8s_cluster.go 文件，新增如下内容：

package requests

import (
	"github.com/gin-gonic/gin"
	"github.com/joker-bai/hawkeye/pkg/app"
	"github.com/thedevsaddam/govalidator"
)

type K8sClusterCreateRequest struct {
	ClusterName    string `json:"cluster_name" form:"cluster_name" valid:"cluster_name"`
	ClusterVersion string `json:"cluster_version" form:"cluster_version" valid:"cluster_version"`
	KubeConfig     string `json:"kube_config" form:"kube_config" valid:"kube_config"`
}

func ValidK8sClusterCreateRequest(data interface{}, ctx *gin.Context) map[string][]string {
	rules := govalidator.MapData{
		"cluster_name":    []string{"required"},
		"cluster_version": []string{"required"},
		"kube_config":     []string{"required"},
	}
	messages := govalidator.MapData{
		"cluster_name": []string{
			"required: 用户名为必填字段,字段为 cluster_name",
		},
		"cluster_version": []string{
			"required: 集群版本为必填项,字段为 cluster_version",
		},
		"kube_config": []string{
			"required: kubeconfig为必填字段,字段为 kube_config",
		},
	}

	// 校验入参

	return app.ValidateOptions(data, rules, messages)
}

type K8sClusterUpdateRequest struct {
	ID             uint32 `json:"id,omitempty" form:"id" valid:"id"`
	ClusterName    string `json:"cluster_name" form:"cluster_name" valid:"cluster_name"`
	ClusterVersion string `json:"cluster_version" form:"cluster_version" valid:"cluster_version"`
	KubeConfig     string `json:"kube_config" form:"kube_config" valid:"kube_config"`
	Status         int    `json:"status" form:"status" valid:"status"`
}

func ValidK8sClusterUpdateRequest(data interface{}, ctx *gin.Context) map[string][]string {
	rules := govalidator.MapData{
		"id":              []string{"required"},
		"cluster_name":    []string{"required"},
		"cluster_version": []string{"required"},
		"kube_config":     []string{"required"},
		"status":          []string{"required"},
	}
	messages := govalidator.MapData{
		"id": []string{
			"required: 用户名ID不能为空",
		},
		"cluster_name": []string{
			"required: 用户名为必填字段,字段为 cluster_name",
		},
		"cluster_version": []string{
			"required: 集群版本为必填项,字段为 cluster_version",
		},
		"kube_config": []string{
			"required: kubeconfig为必填字段,字段为 kube_config",
		},
		"status": []string{
			"required: status为必填字段,字段为 status",
		},
	}

	// 校验入参

	return app.ValidateOptions(data, rules, messages)
}

type K8sClusterListRequest struct {
	ClusterName string `json:"cluster_name,omitempty" form:"cluster_name"`
	Page        int    `json:"page,omitempty" form:"page" valid:"page"`
	Limit       int    `json:"limit,omitempty" form:"limit" valid:"limit"`
}

func ValidK8sClusterListRequest(data interface{}, ctx *gin.Context) map[string][]string {
	rules := govalidator.MapData{
		"page":  []string{"required"},
		"limit": []string{"required"},
	}
	messages := govalidator.MapData{
		"page": []string{
			"required: 页数不能为空",
		},
		"limit": []string{
			"required: 每页条数不能为空",
		},
	}

	// 校验入参

	return app.ValidateOptions(data, rules, messages)
}

3.2、新增 services 方法

在 internal/app/services 目录中新增 k8s_cluster.go 文件，写入以下内容：

package services

import (
	"github.com/joker-bai/kubemana/internal/app/models"
	"github.com/joker-bai/kubemana/internal/app/requests"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
)

func (s *Services) K8sClusterCreate(param *requests.K8sClusterCreateRequest) error {
	return s.dao.K8sClusterCreate(param.ClusterName, param.ClusterVersion, param.KubeConfig)
}

func (s *Services) K8sClusterUpdate(param *requests.K8sClusterUpdateRequest) error {
	return s.dao.K8sClusterUpdate(param.ID, param.ClusterName, param.ClusterVersion, param.KubeConfig, param.Status)
}

func (s *Services) K8sClusterDelete(param *requests.K8sClusterDeleteRequest) error {
	return s.dao.K8sClusterDelete(param.ID)
}

func (s *Services) K8sClusterList(param *requests.K8sClusterListRequest) ([]*models.K8sCluster, error) {
	return s.dao.K8sClusterList(param.ClusterName, param.Page, param.Limit)
}

// 初始化K8s集群
func (s *Services) K8sClusterInit(param *requests.K8sClusterInitRequest) (*kubernetes.Clientset, error) {
	// 通过ID从数据库获取kubeconfig
	kc, err := s.dao.K8sClusterGetById(param.ID)
	if err != nil {
		return nil, err
	}

	cf, err := clientcmd.RESTConfigFromKubeConfig([]byte(kc.KubeConfig))
	if err != nil {
		return nil, err
	}

	clientset, err := kubernetes.NewForConfig(cf)
	if err != nil {
		return nil, err
	}
	return clientset, nil
}

4、新增 controllers 方法

在 internal/app/controllers/api/v1 目录下新增 k8s_cluster.go 文件，输入以下内容：

package v1

import (
	"github.com/gin-gonic/gin"
	"github.com/joker-bai/kubemana/global"
	"github.com/joker-bai/kubemana/internal/app/requests"
	"github.com/joker-bai/kubemana/internal/app/services"
	"github.com/joker-bai/kubemana/pkg/app"
	"github.com/joker-bai/kubemana/pkg/errorcode"
	"go.uber.org/zap"
)

type K8sClusterController struct{}

// Create godoc
// @Summary 列出K8s集群
// @Description 列出K8s集群
// @Tags K8s集群管理
// @Produce json
// @Security ApiKeyAuth
// @Param cluster_name query string false "K8s集群名" maxlength(100)
// @Param page query int true "页码"
// @Param limit query int true "每页数量"
// @Success 200 {object} string "成功"
// @Failure 400 {object} errorcode.Error "请求错误"
// @Failure 500 {object} errorcode.Error "内部错误"
// @Router /api/v1/k8s/cluster/list [get]
func (u *K8sClusterController) List(ctx *gin.Context) {
	param := requests.K8sClusterListRequest{}
	response := app.NewResponse(ctx)

	if ok := app.Validate(ctx, &param, requests.ValidK8sClusterListRequest); !ok {
		return
	}

	svc := services.New(ctx)
	k8s, err := svc.K8sClusterList(&param)
	if err != nil {
		global.Logger.Error("获取K8s集群列表失败", zap.String("error", err.Error()))
		response.ToErrorResponse(errorcode.ErrorK8sClusterListFail)
		return
	}

	response.ToResponseList(k8s, len(k8s))
}

// Create godoc
// @Summary 创建K8s集群
// @Description 创建K8s集群
// @Tags K8s集群管理
// @Produce json
// @Security ApiKeyAuth
// @Param body body requests.K8sClusterCreateRequest true "body"
// @Success 200 {object} string "成功"
// @Failure 400 {object} errorcode.Error "请求错误"
// @Failure 500 {object} errorcode.Error "内部错误"
// @Router /api/v1/k8s/cluster/create [post]
func (u *K8sClusterController) Create(ctx *gin.Context) {
	param := requests.K8sClusterCreateRequest{}
	response := app.NewResponse(ctx)

	if ok := app.Validate(ctx, &param, requests.ValidK8sClusterCreateRequest); !ok {
		return
	}

	svc := services.New(ctx)
	err := svc.K8sClusterCreate(&param)
	if err != nil {
		global.Logger.Error("创建K8s集群失败", zap.String("error", err.Error()))
		response.ToErrorResponse(errorcode.ErrorK8sClusterCreateFail)
		return
	}

	response.ToResponse(gin.H{
		"data": "创建K8s集群成功",
	})
}

// @Summary 修改K8s集群
// @Description 修改K8s集群
// @Tags K8s集群管理
// @Produce json
// @Security ApiKeyAuth
// @Param body body requests.K8sClusterUpdateRequest true "body"
// @Success 200 {object} string "成功"
// @Failure 400 {object} errorcode.Error "请求错误"
// @Failure 500 {object} errorcode.Error "内部错误"
// @Router /api/v1/k8s/cluster/update [post]
func (u *K8sClusterController) Update(ctx *gin.Context) {
	param := requests.K8sClusterUpdateRequest{}
	response := app.NewResponse(ctx)

	if ok := app.Validate(ctx, &param, requests.ValidK8sClusterUpdateRequest); !ok {
		return
	}

	svc := services.New(ctx)
	if err := svc.K8sClusterUpdate(&param); err != nil {
		global.Logger.Error("K8s集群修改失败", zap.String("error", err.Error()))
		response.ToErrorResponse(errorcode.ErrorK8sClusterUpdateFail)
		return
	}

	response.ToResponse(gin.H{
		"data": "K8s集群修改成功",
	})
}

// @Summary 删除K8s集群
// @Description 删除K8s集群
// @Tags K8s集群管理
// @Produce json
// @Security ApiKeyAuth
// @Param body body requests.K8sClusterDeleteRequest true "body"
// @Success 200 {object} string "成功"
// @Failure 400 {object} errorcode.Error "请求错误"
// @Failure 500 {object} errorcode.Error "内部错误"
// @Router /api/v1/k8s/cluster/delete [post]
func (u *K8sClusterController) Delete(ctx *gin.Context) {
	param := requests.K8sClusterDeleteRequest{}
	response := app.NewResponse(ctx)

	if ok := app.Validate(ctx, &param, requests.ValidK8sClusterDeleteRequest); !ok {
		return
	}

	svc := services.New(ctx)
	if err := svc.K8sClusterDelete(&param); err != nil {
		global.Logger.Error("K8s集群删除失败", zap.String("error", err.Error()))
		response.ToErrorResponse(errorcode.ErrorK8sClusterDeleteFail)
		return
	}

	response.ToResponse(gin.H{
		"data": "K8s集群删除成功",
	})
}

// @Summary 初始化K8s集群
// @Description 初始化K8s集群
// @Tags K8s集群管理
// @Produce json
// @Security ApiKeyAuth
// @Param body body requests.K8sClusterDeleteRequest true "body"
// @Success 200 {object} string "成功"
// @Failure 400 {object} errorcode.Error "请求错误"
// @Failure 500 {object} errorcode.Error "内部错误"
// @Router /api/v1/k8s/cluster/init [post]
func (u *K8sClusterController) Init(ctx *gin.Context) {
	param := requests.K8sClusterInitRequest{}
	response := app.NewResponse(ctx)

	if ok := app.Validate(ctx, &param, requests.ValidK8sClusterInitRequest); !ok {
		return
	}

	var err error
	svc := services.New(ctx)
	global.K8S, err = svc.K8sClusterInit(&param)
	if err != nil {
		global.Logger.Error("K8s集群初始化失败", zap.String("error", err.Error()))
		response.ToErrorResponse(errorcode.ErrorK8sClusterInitFail)
		return
	}

	response.ToResponse(gin.H{
		"data": "K8s集群初始化成功",
	})
}

再在 global目录下新增 k8s.go，新增一个全局变量，如下：

package global

import "k8s.io/client-go/kubernetes"

var (
	K8S *kubernetes.Clientset
)

5、新增路由

5.1、新增路由方法

在 interna/app/routers 目录下新增 k8s.go 文件，写入以下内容：

package routers

import (
	"github.com/gin-gonic/gin"
	v1 "github.com/joker-bai/kubemana/internal/app/controllers/api/v1"
)

type K8sRouter struct{}

func (r *K8sRouter) Inject(router *gin.RouterGroup) {

	k8s := router.Group("/k8s")
	{
		// K8s集群管理
		kc := new(v1.K8sClusterController)
		k8s.POST("/cluster/create", kc.Create)
		k8s.POST("/cluster/update", kc.Update)
		k8s.POST("/cluster/delete", kc.Delete)
		k8s.GET("/cluster/list", kc.List)
		k8s.POST("/cluster/init", kc.Init)
	}
}

5.2、加入路由组

在 initialize/router.go 文件中把 k8s 的路由加入 /api/v1 组中，如下：

package initialize

import (
	"github.com/gin-gonic/gin"
	_ "github.com/joker-bai/kubemana/docs"
	"github.com/joker-bai/kubemana/internal/app/middlewares"
	"github.com/joker-bai/kubemana/internal/app/routers"
	swaggerFiles "github.com/swaggo/files"
	ginSwagger "github.com/swaggo/gin-swagger"
)

type injector interface {
	Inject(router *gin.RouterGroup)
}

func (s *Engine) injectRouterGroup(router *gin.RouterGroup) {
	......
	// auth
	for _, r := range []injector{
		new(routers.UserRouter),
		new(routers.K8sRouter),
	} {
		r.Inject(router.Group("/api/v1", middlewares.AuthJWT()))
	}

	......
}

6、测试一下

首先创建一张 k8s_cluster 表，命令如下：

CREATE TABLE `k8s_cluster` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  `cluster_name` varchar(191) NOT NULL DEFAULT '' COMMENT '集群名称',
  `kube_config` varchar(12800) NOT NULL DEFAULT '' COMMENT 'kubeconfig内容',
  `cluster_version` varchar(191) NOT NULL DEFAULT '' COMMENT '集群版本',
  `status` tinyint(3) unsigned DEFAULT '0' COMMENT '集群状态,0表示正常,1表示不正常',
  `created_at` int(10) unsigned DEFAULT '0' COMMENT '创建时间',
  `modified_at` int(10) unsigned DEFAULT '0' COMMENT '修改时间',
  `deleted_at` int(10) unsigned DEFAULT '0' COMMENT '删除时间',
  `is_del` tinyint(3) unsigned DEFAULT '0' COMMENT '是否删除，0表示未删除，1表示删除',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8 COMMENT='K8s集群表';

然后在 apifox 上创建接口进行测试，我这里只测试一个创建集群（记得配置认证 token），如下：

到数据库里查看数据情况，如下：

7、代码版本

本节开发完，记得做代码标记，如下：

$ swag init
$ go mod tidy
$ git add .
$ git commit -m "新增K8s集群管理模块"