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"
	v1 "k8s.io/api/core/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/klog/v2"
	"os"
	"strings"
	"time"
)

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")
	time.Sleep(1 * time.Second)
	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")
}

func GetNodeIpByCore(nodeName string) (string, error) {
	ip := os.Getenv("node_ip")
	if ip != "" {
		klog.Info("[kube] get node ip by core:]", ip)
		return ip, nil
	}
	client, err := GetKubeClient()
	if err != nil {
		return "", err
	}
	nodeItem, err := client.InformerSharedfactory.Core().V1().Nodes().Lister().Get(nodeName)
	if err != nil {
		logrus.WithError(err).Error("[kube] get node info error")
		for {
			time.Sleep(1 * time.Second)
			nodeItem, err = client.InformerSharedfactory.Core().V1().Nodes().Lister().Get(nodeName)
			if err != nil {
				logrus.WithError(err).Error("[kube] get node info error")
			} else {
				break
			}
		}
	}
	// 解析 Node IP
	var internalIP, externalIP string
	for _, addr := range nodeItem.Status.Addresses {
		if addr.Type == v1.NodeInternalIP {
			internalIP = addr.Address
		} else if addr.Type == v1.NodeExternalIP {
			externalIP = addr.Address
		}
	}

	if internalIP == "" && externalIP == "" {
		return "", fmt.Errorf("internalIP or externalIP not found")
	}

	if internalIP != "" {
		return internalIP, nil
	}

	return externalIP, nil
}