Golang Kubenetes容器多集群平台开发实践

前言

   Go语言在基础服务开发领域优势

        Go语言在高并发、通信交互复杂、重业务逻辑的分布式系统中非常适用，具有开发体验好、一定量级下服务稳定、性能满足需要等优势。随着kubernetes在企业中快速扩展，席卷云计算浪潮，而这些云原生技术都是用Go编写，Go语法简洁具有非常简单的并发编程支持。Go语言迅速成为云原生时代的编程语言，更成为kubernetes二次开发首选语言。​

今天给大家分享一个Go语言开发运维平台的案例

   LuBan运维平台是一个基于Go语言+Vue开发的Kubernetes多集群管理平台，可以兼容不同云厂商Kubernetes集群，同时集成CMDB资产管理、支持容器分批发布、发布暂停、JAVA应用诊断等功能。

项目地址：​​https://github.com/dnsjia/luban​​

技术框架​

后端框架：Golang 1.16 + Gin前端框架：Vue3 + AntDesign 2.x数据库：MySQL 5.7、Redis 5.0

资产管理

支持同步阿里云ecs到平台，代码实现思路比较简单，通过云平台API接口发起http请求调用查询资产数据

代码示例：

type CmdbInterface interface {
  GetHostMenu(pid, echo int) ([]*host.TreeList, error)
  UpdateHostGroup(tgr *host.TreeGroupRequest) error
  DeleteHostGroup(groupId int) (err error)
  SShNodeTree(pid int, search string) ([]*host.SShNodeTreeList, error)
  ListVirtualMachine(treeId string, page *models.PageResult, keyword string) (result models.PageResult, err error)
  GetServerAssetsUsers(instanceId string, username interface{}) (data *[]host.SSHLoginUser, err error)
  GetGlobalServerSSHUsers(page *models.PageResult) (result models.PageResult, err error)
  DisableServerSSHUsers(sshUser *host.SystemSSHUser) (err error)
  UpdateServerSSHUsers(sshData *host.SSHGlobalReq, hostUserId string) (err error)
  CreateServerSSHUser(sshData *host.SSHGlobalReq) (err error)
  GetServerAssets(page *models.PageResult) (result models.PageResult, err error)
  DeleteServerSSHUsers(uid int) (err error)
  DeleteHost(groupId int) (err error)
  CollectionDeleteServer(s *host.ServerReq) (err error)
  ConnectionSSH(instanceId string, credentialId string) (sshClient *ssh.Client, err error)
  DeleteAssets(id uint) (err error)
  GetVirtualByInstanceId(instanceId string) (h *host.VirtualMachine, err error)
  GetByCredentialId(credentialId string) (h host.SSHGlobalConfig, err error)
  GetVirtualByUUID(uuid string) (h *host.VirtualMachine, err error)
  ListPlatform(page *models.PageResult) (result models.PageResult, err error)
  CreateCloudAccount(platform *host.CloudPlatform) (err error)
  GetCloudAccount(cloudType string) (platform *host.CloudPlatform, err error)
  InsertSSHRecord(sshRecord *host.SSHRecord) error
  ...
  Create([]*host.VirtualMachine) error // cmdb资产同步方法
  Updates(uuid string, value map[string]interface{}) error
  NewAsynqScheduler() (scheduler *asynq.Scheduler, crontab string)
  NewAsynqServer() *asynq.Server
}

// 程序启动后通过调用tasks.TaskBeat启动定时任务，将通过配置文件crontab参数
// 定期同步资产到运维平台
// tasks.TaskBeta()   定时任务
// 配置文件如下# 每隔两小时同步一次
crontab:
  aliyun: "* */2 * * *"

const (
  SyncAliYunCloud             = "cmdb:aliyun"
  SyncTencentCloud            = "cmdb:tencent"
  TypeDeployDeployment        = "deploy:deployment"
  CheckDeployDeploymentStatus = "deploy:checkDeployment"
)

// NewAliCloudTask 同步阿里云资产同步任务
func NewAliCloudTask(conf *cmdb.CloudPlatform) *asynq.Task {
  payload, err := json.Marshal(conf)
  if err != nil {
    log.Logger.Errorf("创建阿里云资产同步任务失败，err=%v", err)
  }
  return asynq.NewTask(SyncAliYunCloud, payload)
}

func HandleAliCloudTask(ctx context.Context, t *asynq.Task) error {
  var a cmdb.CloudPlatform
  if err := json.Unmarshal(t.Payload(), &a); err != nil {
    return err
  }

  _, err := cloudvendor.GetVendorClient(&a)
  if err != nil {
    log.Logger.Warnf("AccountVerify GetVendorClient failed，%v", err)
    return err
  }

  cloudsync.SyncAliYunHost(&a)

  log.Logger.Info("Aliyun Cloud assets are successfully synchronized...")
  return nil
}

通过调用cloudsync.SyncAliYunHost(&account)方法传入阿里云Ak，创建aliClient同步资产

// SyncAliYunHost 同步阿里云主机
func SyncAliYunHost(task *cmdb.CloudPlatform) {
  defer func() {
    if err := recover(); err != nil {
      log.Logger.Error(fmt.Sprintf("sync panic err: %v", err))
    }
  }()

  // 获取cloud账户
  conf := cmdb.CloudPlatform{
    Type:      cmdb.AliYun,
    AccessKey: task.AccessKey,
    SecretKey: task.SecretKey,
  }

  aliClient, _ = cloudvendor.GetVendorClient(&conf)

  // 获取所有可用区
  regionSet, _ := aliClient.GetRegions()
  for _, region := range regionSet {
    // 获取所有区域下的ecs主机
    instancesInfo, err := aliClient.GetInstances(region.RegionId)
    if err != nil {
      log.Logger.Error(fmt.Sprintf("同步资产发生错误, err: %v", err))
    }
    // 判断区域下是否有ecs
    if len(instancesInfo) != 0 {
      for _, i := range instancesInfo {
        // 根据主机实例id获取db中的主机信息,并获取有差异的主机
        diffHosts, _ := getDiffHosts(&i)
        if len(diffHosts) != 0 {
          // 有差异的更新任务状态为同步中
          //updateTaskState(diffHosts)

          // 同步有差异的主机数据
          syncDiffHosts(diffHosts)
        }
      }
    }
  }
  return
}

同步逻辑

/Users/micheng/code/luban/pkg/cloud/cloudsync/hostsyncor.go
// addHost 添加云主机
func addHost(hostResource []*cmdb.VirtualMachine) error {
  if err := luban.CoreV1.Cmdb().Create(hostResource); err != nil {
    return err
  }

  return nil
}

// syncDiffHosts 更新变化的主机
func syncDiffHosts(diff map[string][]*cmdb.VirtualMachine) {
  for k, v := range diff {
    switch k {
    case "update":
      for _, host := range v {
        // HostName、PublicAddr、PrivateAddr、VmExpiredTime、Status、Mem、CPU、BandWidth
        if err := luban.CoreV1.Cmdb().Updates(host.UUID, map[string]interface{}{
          "hostname":        &host.HostName,
          "public_addr":     &host.PublicAddr,
          "private_addr":    &host.PrivateAddr,
          "vm_expired_time": &host.VmExpiredTime,
          "status":          &host.Status,
          "mem":             &host.Mem,
          "cpu":             &host.CPU,
          "bandwidth":       &host.BandWidth,
        }); err != nil {
          log.Logger.Error("更新主机资源失败", zap.Any("err", err.Error()))
        }
      }
    case "add":
      if err := addHost(v); err != nil {
        log.Logger.Error("同步新增主机失败", zap.Any("err", err.Error()))
      }
    }
  }
}

资产管理支持为用户授权

WebSSH支持屏幕录像、文件管理

鲁班运维平台Web终端集成了文件管理， 使用户更加方便上传下载文件。

接入集群

获取集群凭证, 可在k8s控制节点执行 cat $HOME/.kube/config。将集群文件内容粘贴到集群管理内。

工作负载

计算Pod内多个容器代码逻辑

// GetPodListFromChannels returns a list of all Pods in the cluster
// reading required resource list once from the channels.
func GetPodListFromChannels(channels *k8scommon.ResourceChannels, dsQuery *dataselect.DataSelectQuery) (*PodList, error) {

  pods := <-channels.PodList.List
  err := <-channels.PodList.Error
  if err != nil {
    return nil, err
  }

  eventList := <-channels.EventList.List
  err = <-channels.EventList.Error
  if err != nil {
    return nil, err
  }

  podList := ToPodList(pods.Items, eventList.Items, dsQuery)
  podList.Status = getStatus(pods, eventList.Items)
  return &podList, nil
}

func ToPod(pod *coreV1.Pod, warnings []k8scommon.Event) *Pod {
  if pod == nil {
    return nil
  }
  var restarts = int32(0)
  var containers []*ExtContainer
  for _, container := range pod.Spec.Containers {
    bc := ExContainerStatus(pod.Status.ContainerStatuses, &container)
    containers = append(containers, bc)
    if bc.Restarts > restarts {
      restarts = bc.Restarts
    }
  }
  var intContainers []*ExtContainer
  cn := len(containers)
  for _, container := range pod.Spec.InitContainers {
    bc := ExContainerStatus(pod.Status.InitContainerStatuses, &container)
    intContainers = append(intContainers, bc)
    if bc.Restarts > restarts {
      restarts = bc.Restarts
    }
  }
  cn += len(intContainers)
  podDetail := Pod{
    ObjectMeta:      k8s.NewObjectMeta(pod.ObjectMeta),
    TypeMeta:        k8s.NewTypeMeta(k8s.ResourceKindPod),
    Warnings:        warnings,
    Status:          getPodStatus(*pod),
    InitContainers:  intContainers,
    RestartCount:    getRestartCount(*pod),
    NodeName:        pod.Spec.NodeName,
    ContainerImages: k8scommon.GetContainerImages(&pod.Spec),
    PodIP:           pod.Status.PodIP,
    ContainerNum:    cn,
    Containers:      containers,
  }
  return &podDetail
}

容器组

支持动态打印、日志下载

1. 容器终端连接支持屏幕录像、操作审计。
2. 容器终端连接文件管理支持上传、下载

容器监控

容器监控依赖于Prometheus，需要在导入集群时填写prometheus地址即可完成监控展示

应用发布

应用详情支持展示应用CPU、内存使用率。同时支持配置弹性伸缩策略

HPA弹性伸缩​

为应用配置弹性伸缩（HPA），同时支持两种指标cpu、memory

应用发布支持分批发布，基于Openkruise实现

在发布的过程中支持随时暂停发布， 这里使用的是原生操作。控制器还是会做 ​​​replicas​​​ 数量管理

发布暂停代码逻辑如下

  // 根据环境id, 获取kubeConfig
  cs, err := luban.CoreV1.App().GetClusterConfig(context.TODO(), pause.EnvId, pause.Namespace)
  if err != nil {
    response.FailWithMessage(response.AppEnvIdNotFound, response.AppEnvIdNotFoundMsg, c)
    return
  }
  ...
  // 调用kubernetes api
  client, err := Init.GetK8sClient(cs.KubeConfig)
  if err != nil {
    response.FailWithMessage(http.StatusInternalServerError, response.NotOkMsg, c)
    return
  }

  dClient := client.AppsV1().Deployments(pause.Namespace)
  // RetryOnConflict 对资源进行更新 防止其他代码同时对资源进行不相关更新引起的冲突
  // RetryOnConflict 将等待一段时间，如退避，然后重试。出现非“冲突”错误，或者重试次数过多并放弃，RetryOnConflict 将向调用者返回错误。
  retryErr := retry.RetryOnConflict(retry.DefaultRetry, func() error {
    result, getErr := dClient.Get(context.TODO(), pause.AppName, metav1.GetOptions{})
    if getErr != nil {
      log.Logger.Error(fmt.Sprintf("获取deployment 发生错误， err=%s", getErr))
      return getErr
    }

    if pause.IsPause {
      result.Spec.Paused = true
      msg = "发布已暂停"
    } else {
      result.Spec.Paused = false
      msg = "已恢复发布"
    }

    _, updateErr := dClient.Update(context.TODO(), result, metav1.UpdateOptions{})
    return updateErr
  })

....

应用诊断

基于阿里arthas对JAVA应用进行性能诊断，快速定位Java线上问题、可以在不重启JVM进程的情况下，查看程序的运行情况。线程耗时分析、方法执行分析和性能分析。

对需要诊断的应用Pod Annotation增加​​diagnosis: true​​字段

面对线上服务器cpu使用率一直处于100% ，cpu使用率居高不下，某些线程一直占用着cpu资源，那又如何查看占用cpu较高的线程？按照传统方法我们需通过top命令定位到cpu占用率较高的线程之后，使用jstack pid命令查看当前java进程的堆栈状态。隔段时间再执行一次stack命令获取thread dump，通过thread dump分析线程状态

在dump中，线程一般存在如下几种状态：
1、RUNNABLE，线程处于执行中
2、BLOCKED，线程被阻塞
3、WAITING，线程正在等待

针对以上问题，可以使用应用诊断快速定位线程是否存在死锁，线程CPU使用率过高问题。很大程度上节约了线上问题排查故障时间

源码查看

通过jad反编译class文件

线程堆栈

JVM信息

项目地址：​https://github.com/dnsjia/luban​​

文档地址：​​https://docs.dnsjia.com/application/diagnosis/​​

公众号： 自记小屋

另开设了容器开发实战培训班，想学习提升技能的同学欢迎入群交流学习。

​