ebpf/kube/client.go

package kube

import (
	"context"
	"fmt"
	"github.com/coroot/coroot-node-agent/kube/transfer"
	"github.com/coroot/coroot-node-agent/utils"
	"github.com/coroot/coroot-node-agent/utils/try"
	"github.com/sirupsen/logrus"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"strings"
)

var CwK8sClient *transfer.CwK8sClient

func GetKubeClient() (*transfer.CwK8sClient, error) {
	if CwK8sClient == nil {
		return CwK8sClient, fmt.Errorf("K8sClient is nil")
	}
	return CwK8sClient, nil
}

func NewKubeClient() (*transfer.CwK8sClient, error) {
	if CwK8sClient != nil {
		return CwK8sClient, nil
	}
	ctx := context.Background()
	config, err := rest.InClusterConfig()
	if err != nil {
		return nil, err
	}
	config.TLSClientConfig.Insecure = false
	// creates the clientset
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		logrus.WithError(err).Error("[kube] connect init error")
		return nil, err
	}
	cwk8sclient := transfer.NewCwK8sClient(ctx, clientset)
	//go cwk8sclient.InformerSharedfactory.Start(make(chan struct{}, 0))
	try.GoParams(cwk8sclient.InformerSharedfactory.Start, utils.CatchFn, make(chan struct{}, 0))
	CwK8sClient = cwk8sclient
	logrus.Info("[kube] connect init success")
	return cwk8sclient, nil
}

func GetWorkload(ns, pod string) (string, error) {
	client, err := GetKubeClient()
	if err != nil {
		return "", err
	}
	podsItems, err := client.InformerSharedfactory.Core().V1().Pods().Lister().Pods(ns).Get(pod)
	if err != nil {
		return "", err
	}
	if len(podsItems.ObjectMeta.OwnerReferences) > 0 {
		ownerReferences := podsItems.ObjectMeta.OwnerReferences[0]
		kind := ownerReferences.Kind
		workload := ownerReferences.Name
		switch kind {
		case "ReplicaSet":
			lastDash := strings.LastIndex(workload, "-")
			if lastDash != -1 {
				workload = workload[:lastDash]
			}
		}
		return workload, nil
	}
	return "", fmt.Errorf("workload not found")
}