constructor/k8s.go

Rama main · réplica de lectura del repositorio original.

package constructor

import (
	"fmt"
	"net"
	"regexp"
	"strconv"
	"strings"

	"github.com/coroot/coroot/db"
	"github.com/coroot/coroot/model"
	"github.com/coroot/coroot/timeseries"
	"github.com/coroot/coroot/utils"
	"inet.af/netaddr"
	"k8s.io/klog"
)

var (
	jobSuffixRe = regexp.MustCompile(`-([a-z0-9]{5}|\d{10,})$`)
)

type serviceId struct {
	name, ns string
}

func (c *Constructor) loadKubernetesMetadata(w *model.World, metrics map[string][]*model.MetricValues, servicesByClusterIP map[string]*model.Service, inferenceGroups map[model.ApplicationId]model.ApplicationId, project *db.Project) {
	jobOwners := c.getJobOwners(metrics["kube_job_owner"], project.ClusterId())
	pods := c.podInfo(w, metrics["kube_pod_info"], metrics["kube_pod_labels"], metrics["kube_pod_owner"], jobOwners, project)
	c.podLabels(metrics["kube_pod_labels"], pods, inferenceGroups, project)
	podAnnotations(metrics["kube_pod_annotations"], pods)

	appsByPodIP := map[string]*model.Application{}
	for _, pod := range pods {
		if pod.Pod.IP != "" && pod.Owner != nil {
			appsByPodIP[pod.Pod.IP] = pod.Owner
		}
	}
	services := loadServices(metrics)
	for _, s := range services {
		if s.ClusterIP != "" {
			servicesByClusterIP[s.ClusterIP] = s
		}
		for ep := range s.Endpoints {
			if app := appsByPodIP[ep.IP().String()]; app != nil {
				s.DestinationApps[app.Id] = app
			}
		}
		for _, app := range s.DestinationApps {
			app.KubernetesServices = append(app.KubernetesServices, s)
		}
	}

	for _, q := range QUERIES {
		switch {
		case strings.HasPrefix(q.Name, "kube_pod_status_"):
			podStatus(q.Name, metrics[q.Name], pods)
		case strings.HasPrefix(q.Name, "kube_pod_init_container_"):
			podContainer(q.Name, metrics[q.Name], pods)
		case strings.HasPrefix(q.Name, "kube_pod_container_"):
			podContainer(q.Name, metrics[q.Name], pods)
		}
	}
	c.loadApplications(w, metrics, project)
}

func loadServices(metrics map[string][]*model.MetricValues) map[serviceId]*model.Service {
	services := map[serviceId]*model.Service{}
	for _, m := range metrics["kube_service_info"] {
		name := m.Labels["service"]
		s := &model.Service{
			Name:            name,
			Namespace:       m.Labels["namespace"],
			ClusterIP:       m.Labels["cluster_ip"],
			Endpoints:       map[netaddr.IPPort]struct{}{},
			LoadBalancerIPs: utils.NewStringSet(),
			Ports:           utils.NewStringSet(),
			NodePorts:       utils.NewStringSet(),
			DestinationApps: map[model.ApplicationId]*model.Application{},
		}
		services[serviceId{name: s.Name, ns: s.Namespace}] = s
	}
	for _, m := range metrics["kube_service_spec_type"] {
		if s := services[serviceId{name: m.Labels["service"], ns: m.Labels["namespace"]}]; s != nil {
			s.Type.Update(m.Values, m.Labels["type"])
		}
	}
	for _, m := range metrics["kube_service_status_load_balancer_ingress"] {
		if s := services[serviceId{name: m.Labels["service"], ns: m.Labels["namespace"]}]; s != nil {
			s.LoadBalancerIPs.Add(m.Labels["ip"])
		}
	}
	for _, m := range metrics["kube_endpoint_address"] {
		if s := services[serviceId{name: m.Labels["endpoint"], ns: m.Labels["namespace"]}]; s != nil {
			if ip, err := netaddr.ParseIP(m.Labels["ip"]); err == nil {
				port, _ := strconv.ParseUint(m.Labels["port_number"], 10, 16)
				s.Endpoints[netaddr.IPPortFrom(ip, uint16(port))] = struct{}{}
			}
		}
	}
	for _, m := range metrics["kube_service_port"] {
		if s := services[serviceId{name: m.Labels["service"], ns: m.Labels["namespace"]}]; s != nil {
			s.Ports.Add(m.Labels["port"])
			if np := m.Labels["node_port"]; np != "" {
				s.NodePorts.Add(np)
			}
		}
	}
	for _, s := range services {
		if s.Name == "kubernetes" {
			s.Name = "kube-apiserver"
		}
	}
	return services
}

func (c *Constructor) loadApplications(w *model.World, metrics map[string][]*model.MetricValues, project *db.Project) {
	for queryName := range metrics {
		var (
			kind      model.ApplicationKind
			nameLabel string
		)
		switch {
		case strings.HasPrefix(queryName, "kube_deployment_"):
			kind = model.ApplicationKindDeployment
			nameLabel = "deployment"
		case strings.HasPrefix(queryName, "kube_statefulset_"):
			kind = model.ApplicationKindStatefulSet
			nameLabel = "statefulset"
		case strings.HasPrefix(queryName, "kube_daemonset_"):
			kind = model.ApplicationKindDaemonSet
			nameLabel = "daemonset"
		case strings.HasPrefix(queryName, "kube_cronjob_"):
			kind = model.ApplicationKindCronJob
			nameLabel = "cronjob"
		default:
			continue
		}
		for _, m := range metrics[queryName] {
			app := w.GetApplication(c.newApplicationId(project.ClusterId(), m.Labels["namespace"], kind, m.Labels[nameLabel]))
			if app == nil {
				continue
			}
			switch queryName {
			case "kube_deployment_spec_replicas", "kube_statefulset_replicas", "kube_daemonset_status_desired_number_scheduled":
				app.DesiredInstances = merge(app.DesiredInstances, m.Values, timeseries.Any)
			case "kube_deployment_annotations", "kube_statefulset_annotations", "kube_daemonset_annotations", "kube_cronjob_annotations":
				app.Annotations.UpdateFromLabels(m.Labels, m.Values)
			}
		}
	}
}

func (c *Constructor) getJobOwners(metrics []*model.MetricValues, clusterId string) map[podId]model.ApplicationId {
	owners := make(map[podId]model.ApplicationId, len(metrics))
	for _, m := range metrics {
		if m.Labels["owner_is_controller"] == "false" {
			continue
		}
		kind := model.ApplicationKind(m.Labels["owner_kind"])
		name := m.Labels["owner_name"]
		jobName := m.Labels["job_name"]
		ns := m.Labels["namespace"]
		if jobName == "" || name == "" || kind == "" || kind == "<none>" {
			continue
		}
		owners[podId{name: jobName, ns: ns}] = c.newApplicationId(clusterId, ns, kind, name)
	}
	return owners
}

func (c *Constructor) podInfo(w *model.World, metrics []*model.MetricValues, labelMetrics []*model.MetricValues, ownerMetrics []*model.MetricValues, jobOwners map[podId]model.ApplicationId, project *db.Project) map[string]*model.Instance {
	pods := map[string]*model.Instance{}
	gitlabProjects := map[string]string{}
	for _, m := range labelMetrics {
		if name := m.Labels["label_project_runner_gitlab_com_name"]; name != "" {
			gitlabProjects[m.Labels["uid"]] = name
		}
	}
	nonControllerOwners := map[string]*model.MetricValues{}
	for _, m := range ownerMetrics {
		nonControllerOwners[m.Labels["uid"]] = m
	}
	podOwners := map[podId]model.ApplicationId{}
	var podsOwnedByPods []*model.Instance

	k8sNodes := make(map[string]*model.Node, len(w.Nodes))
	for _, n := range w.Nodes {
		k8sNodes[n.K8sName.Value()] = n
	}

	for _, m := range metrics {
		w.IntegrationStatus.KubeStateMetrics.Installed = true
		pod := m.Labels["pod"]
		ns := m.Labels["namespace"]
		ownerName := m.Labels["created_by_name"]
		ownerKind := model.ApplicationKind(m.Labels["created_by_kind"])
		nodeName := m.Labels["node"]
		uid := m.Labels["uid"]
		if uid == "" {
			klog.Errorln("invalid 'kube_pod_info' metric: 'uid' label is empty")
			continue
		}
		node := k8sNodes[nodeName]
		var appId model.ApplicationId

		gitlabProject := gitlabProjects[uid]
		if ownerKind == "" || ownerKind == "<none>" {
			if o := nonControllerOwners[uid]; o != nil {
				ownerKind = model.ApplicationKind(o.Labels["owner_kind"])
				ownerName = o.Labels["owner_name"]
			}
		}

		switch {
		case (ownerKind == "" || ownerKind == "<none>") && gitlabProject != "":
			appId = c.newApplicationId(project.ClusterId(), ns, model.ApplicationKindGitLabJob, gitlabProject)
		case ownerKind == "" || ownerKind == "<none>" || ownerKind == "Node":
			appId = c.newApplicationId(project.ClusterId(), ns, model.ApplicationKindStaticPods, strings.TrimSuffix(pod, "-"+nodeName))
		case ownerKind == model.ApplicationKindSparkApplication || ownerKind == model.ApplicationKindArgoWorkflow:
			appId = c.newApplicationId(project.ClusterId(), ns, ownerKind, jobSuffixRe.ReplaceAllString(ownerName, ""))
		case ownerKind == model.ApplicationKindJob:
			if id, ok := jobOwners[podId{name: ownerName, ns: ns}]; ok {
				appId = id
			} else {
				appId = c.newApplicationId(project.ClusterId(), ns, ownerKind, ownerName)
			}
		case ownerName != "":
			appId = c.newApplicationId(project.ClusterId(), ns, ownerKind, ownerName)
		default:
			continue
		}
		podOwners[podId{name: pod, ns: ns}] = appId
		instance := pods[uid]
		if instance == nil {
			app := w.GetOrCreateApplication(appId, false)
			if appId.Kind == model.ApplicationKindCronJob {
				continue
			}
			instance = app.GetOrCreateInstance(pod, node)
			if instance.Pod == nil {
				instance.Pod = &model.Pod{}
			}
			instance.Pod.OwnerKind = ownerKind
			if ownerKind == model.ApplicationKindReplicaSet {
				instance.Pod.ReplicaSet = ownerName
			}
			pods[uid] = instance
		}
		if node != nil && instance.Node == nil {
			instance.Node = node
			node.Instances = append(node.Instances, instance)
		}

		podIp := m.Labels["pod_ip"]
		hostIp := m.Labels["host_ip"]

		if podIp != "" && (podIp != hostIp || (node != nil && node.Fargate)) {
			if ip := net.ParseIP(podIp); ip != nil {
				isActive := m.Values.Last() == 1
				instance.TcpListens[model.Listen{IP: podIp, Port: "0", Proxied: false}] = isActive
				instance.Pod.IP = podIp
			}
		}
		if appId.Kind == model.ApplicationKindPod {
			podsOwnedByPods = append(podsOwnedByPods, instance)
		}
	}
	for _, instance := range podsOwnedByPods {
		id := podId{name: instance.Owner.Id.Name, ns: instance.Owner.Id.Namespace}
		if ownerOfOwner, ok := podOwners[id]; ok {
			if app := w.GetApplication(ownerOfOwner); app != nil {
				delete(w.Applications, instance.Owner.Id)
				instance.Owner = app
				app.Instances = append(app.Instances, instance)
			}
		}
	}
	return pods
}

func (c *Constructor) podLabels(metrics []*model.MetricValues, pods map[string]*model.Instance, inferenceGroups map[model.ApplicationId]model.ApplicationId, project *db.Project) {
	for _, m := range metrics {
		uid := m.Labels["uid"]
		if uid == "" {
			continue
		}
		instance := pods[uid]
		if instance == nil {
			continue
		}
		cluster, role := "", ""
		var manager model.ClusterManager
		switch {
		case m.Labels["label_postgres_operator_crunchydata_com_cluster"] != "":
			cluster = m.Labels["label_postgres_operator_crunchydata_com_cluster"]
			role = m.Labels["label_postgres_operator_crunchydata_com_role"]
			manager = model.ClusterManagerCrunchy
		case m.Labels["label_cluster_name"] != "" && m.Labels["label_team"] != "": // zalando pg operator
			cluster = m.Labels["label_cluster_name"]
			if m.Labels["label_application"] == "spilo" { // not a pooler (pgbouncer)
				role = m.Labels["label_spilo_role"]
			}
			manager = model.ClusterManagerZalando
		case m.Labels["label_cnpg_io_cluster"] != "":
			cluster = m.Labels["label_cnpg_io_cluster"]
			role = m.Labels["label_role"]
			manager = model.ClusterManagerCNPG
		case m.Labels["label_stackgres_io_cluster_name"] != "":
			cluster = m.Labels["label_stackgres_io_cluster_name"]
			role = m.Labels["label_role"]
			manager = model.ClusterManagerStackGres
		case m.Labels["label_app_kubernetes_io_managed_by"] == "percona-server-mongodb-operator":
			cluster = m.Labels["label_app_kubernetes_io_instance"]
			manager = model.ClusterManagerPerconaMongoDB
		case m.Labels["label_app_kubernetes_io_managed_by"] == "percona-xtradb-cluster-operator":
			cluster = m.Labels["label_app_kubernetes_io_instance"]
			manager = model.ClusterManagerPerconaXtraDB
		case m.Labels["label_app_kubernetes_io_managed_by"] == "percona-server-mysql-operator":
			cluster = m.Labels["label_app_kubernetes_io_instance"]
			manager = model.ClusterManagerPerconaServer
		case m.Labels["label_app_kubernetes_io_managed_by"] == "valkey-operator":
			cluster = m.Labels["label_valkey_io_cluster"]
		case strings.HasPrefix(m.Labels["label_helm_sh_chart"], "mongodb"):
			if m.Labels["label_app_kubernetes_io_name"] != "" && m.Labels["label_app_kubernetes_io_instance"] != "" {
				cluster = m.Labels["label_app_kubernetes_io_instance"] + "-" + m.Labels["label_app_kubernetes_io_name"]
			}
		case strings.HasPrefix(m.Labels["label_helm_sh_chart"], "redis") || strings.HasPrefix(m.Labels["label_helm_sh_chart"], "valkey"):
			if m.Labels["label_app_kubernetes_io_name"] != "" && m.Labels["label_app_kubernetes_io_instance"] != "" {
				cluster = m.Labels["label_app_kubernetes_io_instance"] + "-" + m.Labels["label_app_kubernetes_io_name"]
			}
		case strings.HasPrefix(m.Labels["label_helm_sh_chart"], "mysql"):
			if m.Labels["label_app_kubernetes_io_name"] != "" && m.Labels["label_app_kubernetes_io_instance"] != "" {
				cluster = m.Labels["label_app_kubernetes_io_instance"] + "-" + m.Labels["label_app_kubernetes_io_name"]
			}
		case strings.HasPrefix(m.Labels["label_helm_sh_chart"], "mariadb"):
			if m.Labels["label_app_kubernetes_io_name"] != "" && m.Labels["label_app_kubernetes_io_instance"] != "" {
				cluster = m.Labels["label_app_kubernetes_io_instance"] + "-" + m.Labels["label_app_kubernetes_io_name"]
			}
		case strings.HasPrefix(m.Labels["label_helm_sh_chart"], "clickhouse"):
			if m.Labels["label_app_kubernetes_io_name"] != "" && m.Labels["label_app_kubernetes_io_instance"] != "" {
				cluster = m.Labels["label_app_kubernetes_io_instance"] + "-" + m.Labels["label_app_kubernetes_io_name"]
			}
		case m.Labels["label_app_kubernetes_io_managed_by"] == "coroot-operator" && m.Labels["label_app_kubernetes_io_component"] == "clickhouse":
			cluster = m.Labels["label_app_kubernetes_io_part_of"] + "-" + m.Labels["label_app_kubernetes_io_component"]
		case m.Labels["label_serving_kserve_io_inferenceservice"] != "":
			if instance.Owner.Id.Kind == model.ApplicationKindDeployment {
				inferenceGroups[instance.Owner.Id] = c.newApplicationId(project.ClusterId(), instance.Owner.Id.Namespace, model.ApplicationKindInferenceService, m.Labels["label_serving_kserve_io_inferenceservice"])
			}
			role = m.Labels["label_component"]
		case m.Labels["label_nvidia_com_dynamo_graph_deployment_name"] != "":
			if instance.Owner.Id.Kind == model.ApplicationKindDeployment {
				inferenceGroups[instance.Owner.Id] = c.newApplicationId(project.ClusterId(), instance.Owner.Id.Namespace, model.ApplicationKindDynamoGraphDeployment, m.Labels["label_nvidia_com_dynamo_graph_deployment_name"])
			}
			role = m.Labels["label_nvidia_com_dynamo_component_type"]
		case m.Labels["label_disaggregatedset_x_k8s_io_name"] != "":
			if instance.Owner.Id.Kind == model.ApplicationKindDeployment || instance.Owner.Id.Kind == model.ApplicationKindStatefulSet {
				inferenceGroups[instance.Owner.Id] = c.newApplicationId(project.ClusterId(), instance.Owner.Id.Namespace, model.ApplicationKindDisaggregatedSet, m.Labels["label_disaggregatedset_x_k8s_io_name"])
			}
			role = m.Labels["label_disaggregatedset_x_k8s_io_role"]
		case m.Labels["label_leaderworkerset_sigs_k8s_io_name"] != "":
			if instance.Owner.Id.Kind == model.ApplicationKindStatefulSet {
				inferenceGroups[instance.Owner.Id] = c.newApplicationId(project.ClusterId(), instance.Owner.Id.Namespace, model.ApplicationKindLeaderWorkerSet, m.Labels["label_leaderworkerset_sigs_k8s_io_name"])
			}
			if m.Labels["label_leaderworkerset_sigs_k8s_io_worker_index"] == "0" {
				role = "leader"
			} else {
				role = "worker"
			}
		default:
			continue
		}
		if cluster != "" {
			instance.ClusterName.Update(m.Values, cluster)
			if manager != "" && instance.Owner != nil {
				instance.Owner.Cluster.Manager = manager
			}
		}
		if role == "master" {
			role = "primary"
		}
		instance.UpdateClusterRole(role, m.Values)
	}
}

func podAnnotations(metrics []*model.MetricValues, pods map[string]*model.Instance) {
	for _, m := range metrics {
		uid := m.Labels["uid"]
		if uid == "" {
			continue
		}
		instance := pods[uid]
		if instance == nil {
			continue
		}
		instance.Annotations.UpdateFromLabels(m.Labels, m.Values)
	}
}

func podStatus(queryName string, metrics []*model.MetricValues, pods map[string]*model.Instance) {
	for _, m := range metrics {
		uid := m.Labels["uid"]
		if uid == "" {
			continue
		}
		instance := pods[uid]
		if instance == nil {
			//klog.Warningln("unknown pod:", uid, m.Labels["pod"], m.Labels["namespace"])
			continue
		}
		switch queryName {
		case "kube_pod_status_phase":
			instance.Pod.LifeSpan = merge(instance.Pod.LifeSpan, m.Values, timeseries.NanSum)
			if m.Values.Last() > 0 {
				instance.Pod.Phase = m.Labels["phase"]
			}
			if m.Labels["phase"] == "Running" {
				instance.Pod.Running = merge(instance.Pod.Running, m.Values, timeseries.Any)
			}
		case "kube_pod_status_ready":
			instance.Pod.Ready = merge(instance.Pod.Ready, m.Values, timeseries.Any)
		case "kube_pod_status_scheduled":
			instance.Pod.Scheduled = m.Values.Last() > 0
		}
	}
}

func podContainer(queryName string, metrics []*model.MetricValues, pods map[string]*model.Instance) {
	for _, m := range metrics {
		uid := m.Labels["uid"]
		if uid == "" {
			continue
		}
		instance := pods[uid]
		if instance == nil {
			//klog.Warningln("unknown pod:", uid, m.Labels["pod"], m.Labels["namespace"])
			continue
		}
		containerId := fmt.Sprintf("/k8s/%s/%s/%s", m.Labels["namespace"], m.Labels["pod"], m.Labels["container"])
		container := instance.GetOrCreateContainer(containerId, m.Labels["container"])

		switch queryName {
		case "kube_pod_init_container_info":
			container.InitContainer = true
		case "kube_pod_container_resource_requests":
			switch m.Labels["resource"] {
			case "cpu":
				container.CpuRequest = merge(container.CpuRequest, m.Values, timeseries.Max)
			case "memory":
				container.MemoryRequest = merge(container.MemoryRequest, m.Values, timeseries.Max)
			}
		case "kube_pod_container_status_ready":
			container.Ready = m.Values.Last() > 0
		case "kube_pod_container_status_waiting":
			if m.Values.Last() > 0 {
				container.Status = model.ContainerStatusWaiting
			}
		case "kube_pod_container_status_running":
			if m.Values.Last() > 0 {
				container.Status = model.ContainerStatusRunning
				container.Reason = ""
			}
		case "kube_pod_container_status_terminated":
			if m.Values.Last() > 0 {
				container.Status = model.ContainerStatusTerminated
			}
		case "kube_pod_container_status_waiting_reason":
			if m.Values.Last() > 0 {
				container.Status = model.ContainerStatusWaiting
				container.Reason = m.Labels["reason"]
			}
		case "kube_pod_container_status_terminated_reason":
			if m.Values.Last() > 0 {
				container.Status = model.ContainerStatusTerminated
				container.Reason = m.Labels["reason"]
			}
		case "kube_pod_container_status_last_terminated_reason":
			if m.Values.Last() > 0 {
				container.LastTerminatedReason = m.Labels["reason"]
			}
		}
	}
}