ebpf/containers/registry.go

package containers

import (
	"bytes"
	"fmt"
	"os"
	"regexp"
	"strings"
	"sync"
	"time"

	"github.com/coroot/coroot-node-agent/cgroup"
	"github.com/coroot/coroot-node-agent/common"
	"github.com/coroot/coroot-node-agent/ebpftracer"
	"github.com/coroot/coroot-node-agent/flags"
	"github.com/coroot/coroot-node-agent/proc"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/vishvananda/netns"
	"inet.af/netaddr"
	"k8s.io/klog/v2"
)

var (
	selfNetNs         = netns.None()
	hostNetNsId       = netns.None().UniqueId()
	agentPid          = uint32(os.Getpid())
	containerIdRegexp = regexp.MustCompile(`[a-z0-9]{64}`)
)

type ProcessInfo struct {
	Pid         uint32
	ContainerId ContainerID
	StartedAt   time.Time
}

type Registry struct {
	reg prometheus.Registerer

	tracer *ebpftracer.Tracer
	events chan ebpftracer.Event

	hostConntrack *Conntrack

	containersById       map[ContainerID]*Container
	containersByCgroupId map[string]*Container
	containersByPid      map[uint32]*Container
	ip2fqdn              map[netaddr.IP]string
	ip2fqdnLock          sync.Mutex

	processInfoCh chan<- ProcessInfo
}

func NewRegistry(reg prometheus.Registerer, kernelVersion string, processInfoCh chan<- ProcessInfo) (*Registry, error) {
	ns, err := proc.GetSelfNetNs()
	if err != nil {
		return nil, err
	}
	selfNetNs = ns
	hostNetNs, err := proc.GetHostNetNs()
	if err != nil {
		return nil, err
	}
	defer hostNetNs.Close()
	hostNetNsId = hostNetNs.UniqueId()

	err = proc.ExecuteInNetNs(hostNetNs, selfNetNs, func() error {
		if err := TaskstatsInit(); err != nil {
			return err
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	if err := cgroup.Init(); err != nil {
		return nil, err
	}
	if err := DockerdInit(); err != nil {
		klog.Warningln(err)
	}
	if err := ContainerdInit(); err != nil {
		klog.Warningln(err)
	}
	if err := CrioInit(); err != nil {
		klog.Warningln(err)
	}
	if err := JournaldInit(); err != nil {
		klog.Warningln(err)
	}
	ct, err := NewConntrack(hostNetNs)
	if err != nil {
		return nil, err
	}

	r := &Registry{
		reg:    reg,
		events: make(chan ebpftracer.Event, 10000), // TODO: 队列

		hostConntrack: ct,

		containersById:       map[ContainerID]*Container{}, // TODO: 缓存的一些关系映射
		containersByCgroupId: map[string]*Container{},
		containersByPid:      map[uint32]*Container{},
		ip2fqdn:              map[netaddr.IP]string{},

		processInfoCh: processInfoCh, // TODO: 进程信息

		tracer: ebpftracer.NewTracer(kernelVersion, *flags.DisableL7Tracing), // TODO: 核心代码
	}
	if err = reg.Register(r); err != nil {
		return nil, err
	}
	go r.handleEvents(r.events)                   // TODO: 核心代码
	if err = r.tracer.Run(r.events); err != nil { // TODO: 核心代码
		close(r.events)
		return nil, err
	}

	return r, nil
}

func (r *Registry) Describe(ch chan<- *prometheus.Desc) {
	ch <- metrics.Ip2Fqdn
}

func (r *Registry) Collect(ch chan<- prometheus.Metric) {
	r.ip2fqdnLock.Lock()
	defer r.ip2fqdnLock.Unlock()
	for ip, fqdn := range r.ip2fqdn {
		ch <- gauge(metrics.Ip2Fqdn, 1, ip.String(), fqdn)
	}
}

func (r *Registry) Close() {
	r.tracer.Close()
	close(r.events)
}

func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) {
	gcTicker := time.NewTicker(gcInterval)
	defer gcTicker.Stop()
	for {
		select {
		case now := <-gcTicker.C: // TODO: 周期性做一些操作, 暂时忽略......
			for pid, c := range r.containersByPid {
				cg, err := proc.ReadCgroup(pid)
				if err != nil {
					delete(r.containersByPid, pid)
					if c != nil {
						c.onProcessExit(pid, false)
					}
					continue
				}
				if c != nil && cg.Id != c.cgroup.Id {
					delete(r.containersByPid, pid)
					c.onProcessExit(pid, false)
				}
			}
			activeIPs := map[netaddr.IP]struct{}{}
			for id, c := range r.containersById {
				if !c.Dead(now) {
					continue
				}
				for dst := range c.connectLastAttempt {
					activeIPs[dst.IP()] = struct{}{}
				}
				klog.Infoln("deleting dead container:", id)
				for cg, cc := range r.containersByCgroupId {
					if cc == c {
						delete(r.containersByCgroupId, cg)
					}
				}
				for pid, cc := range r.containersByPid {
					if cc == c {
						delete(r.containersByPid, pid)
					}
				}
				if ok := prometheus.WrapRegistererWith(setLabels(string(id),
					c.K8sContainer.ns,
					c.K8sContainer.podName,
					c.K8sContainer.containerName,
					c.K8sContainer.pid), r.reg).Unregister(c); !ok {
					klog.Warningln("failed to unregister container:", id)
				}
				delete(r.containersById, id)
				c.Close()
			}
			r.ip2fqdnLock.Lock()
			for ip := range r.ip2fqdn {
				if _, ok := activeIPs[ip]; !ok {
					delete(r.ip2fqdn, ip)
				}
			}
			r.ip2fqdnLock.Unlock() // T
		case e, more := <-ch: // TODO: 核心代码, 读取chan中的数据.... 来自tracer.init方法...
			if e.Pid == uint32(os.Getpid()) {
				continue
			}
			//if e.Pid != 362943 { // TODO: TEST_DEBUG........只拦截一个进程.....
			//	continue
			//}

			if !more {
				return
			}
			// fmt.Println("--------- <- ch -------------pid = ", e.Pid)
			switch e.Type {
			case ebpftracer.EventTypeProcessStart: // TODO: ??????
				// fmt.Println("----EventTypeProcessStart----")
				c, seen := r.containersByPid[e.Pid]
				switch { // possible pids wraparound + missed `process-exit` event
				case c == nil && seen: // ignored
					delete(r.containersByPid, e.Pid)
				case c != nil: // revalidating by cgroup
					cg, err := proc.ReadCgroup(e.Pid) // TODO: 没有什么权限？
					if err != nil || cg.Id != c.cgroup.Id {
						delete(r.containersByPid, e.Pid)
						c.onProcessExit(e.Pid, false)
					}
				}
				if c := r.getOrCreateContainer(e.Pid); c != nil {
					p := c.onProcessStart(e.Pid)
					if r.processInfoCh != nil && p != nil {
						r.processInfoCh <- ProcessInfo{Pid: p.Pid, ContainerId: c.id, StartedAt: p.StartedAt}
					}
				}
			case ebpftracer.EventTypeProcessExit:
				if c := r.containersByPid[e.Pid]; c != nil {
					c.onProcessExit(e.Pid, e.Reason == ebpftracer.EventReasonOOMKill)
				}
				delete(r.containersByPid, e.Pid)

			case ebpftracer.EventTypeFileOpen:
				if c := r.getOrCreateContainer(e.Pid); c != nil {
					c.onFileOpen(e.Pid, e.Fd)
				}

			case ebpftracer.EventTypeListenOpen: // TODO: 核心逻辑：attachTlsUprobes, , 为什么放在这里？？？根据事件绑定uprobe？？？
				fmt.Println("ebpftracer.EventTypeConnectionOpen==================", e.Pid) // ebpftracer.EventTypeConnectionOpen================== 362943
				// fmt.Println("----EventTypeListenOpen----")
				if c := r.getOrCreateContainer(e.Pid); c != nil {
					c.onListenOpen(e.Pid, e.SrcAddr, false)
					c.attachTlsUprobes(r.tracer, e.Pid)
				} else {
					klog.Infoln("TCP listen open from unknown container", e)
				}
			case ebpftracer.EventTypeListenClose:
				// fmt.Println("----EventTypeListenClose----")
				if c := r.containersByPid[e.Pid]; c != nil {
					c.onListenClose(e.Pid, e.SrcAddr)
				}

			case ebpftracer.EventTypeConnectionOpen: // TODO: 核心逻辑：attachTlsUprobes, 为什么放在这里？？？ 根据事件绑定uprobe？？？
				// fmt.Println("----EventTypeConnectionOpen----")
				fmt.Println("ebpftracer.EventTypeConnectionOpen==================", e.Pid)
				if c := r.getOrCreateContainer(e.Pid); c != nil {
					c.onConnectionOpen(e.Pid, e.Fd, e.SrcAddr, e.DstAddr, e.Timestamp, false)
					c.attachTlsUprobes(r.tracer, e.Pid)
				} else {
					klog.Infoln("TCP connection from unknown container", e)
				}
			case ebpftracer.EventTypeConnectionError:
				// fmt.Println("----EventTypeConnectionError----")
				if c := r.getOrCreateContainer(e.Pid); c != nil {
					c.onConnectionOpen(e.Pid, e.Fd, e.SrcAddr, e.DstAddr, 0, true)
				} else {
					klog.Infoln("TCP connection error from unknown container", e)
				}
			case ebpftracer.EventTypeConnectionClose:
				// fmt.Println("----EventTypeConnectionClose----")
				srcDst := AddrPair{src: e.SrcAddr, dst: e.DstAddr}
				for _, c := range r.containersById {
					if c.onConnectionClose(srcDst) {
						break
					}
				}
			case ebpftracer.EventTypeTCPRetransmit:
				// fmt.Println("----EventTypeTCPRetransmit----")
				srcDst := AddrPair{src: e.SrcAddr, dst: e.DstAddr}
				for _, c := range r.containersById {
					if c.onRetransmit(srcDst) {
						break
					}
				}
			case ebpftracer.EventTypeL7Request: // TODO: 核心代码, 处理7层协议请求, 来自tracer.runEventsReader方法......
				// fmt.Println("-----EventTypeL7Request------1", e) // TODO: 需要区分是client还是server吗?
				fmt.Println("e.L7Request Payload:", string(e.L7Request.Payload))
				if e.L7Request == nil {
					continue
				}
				if c := r.containersByPid[e.Pid]; c != nil {
					fmt.Println("-----EventTypeL7Request------2")
					ip2fqdn := c.onL7RequestApm(e.Pid, e.Fd, e.Timestamp, e.L7Request) // TODO: 核心代码
					r.ip2fqdnLock.Lock()
					for ip, fqdn := range ip2fqdn {
						r.ip2fqdn[ip] = fqdn
					}
					r.ip2fqdnLock.Unlock()
				}
			}
		}
	}
}

func (r *Registry) getOrCreateContainer(pid uint32) *Container {
	if c, seen := r.containersByPid[pid]; c != nil {
		return c
	} else if seen { // ignored
		return nil
	}
	cg, err := proc.ReadCgroup(pid)
	if err != nil {
		if !common.IsNotExist(err) {
			klog.Warningln("failed to read proc cgroup:", err)
		}
		return nil
	}
	cgId := fmt.Sprintf("%s/%d", cg.Id, pid)
	if c := r.containersByCgroupId[cgId]; c != nil {
		r.containersByPid[pid] = c
		return c
	}
	if cg.ContainerType == cgroup.ContainerTypeSandbox {
		cmdline := proc.GetCmdline(pid)
		parts := bytes.Split(cmdline, []byte{0})
		if len(parts) > 0 {
			cmd := parts[0]
			lastArg := parts[len(parts)-1]
			if (bytes.HasSuffix(cmd, []byte("runsc-sandbox")) || bytes.HasSuffix(cmd, []byte("runsc"))) && containerIdRegexp.Match(lastArg) {
				cg.ContainerId = string(lastArg)
			}
		}
	}
	md, err := getContainerMetadata(cg)
	if err != nil {
		klog.Warningf("failed to get container metadata for pid %d -> %s: %s", pid, cg.Id, err)
		return nil
	}
	// add ns/workload/podname
	id, extensionTag := calcId(cg, md, pid)

	//klog.Infof("calculated container id %d -> %s -> %s", pid, cg.Id, id)
	if id == "" {
		if cg.Id == "/init.scope" && pid != 1 {
			klog.InfoS("ignoring without persisting", "cg", cg.Id, "pid", pid)
		} else {
			klog.InfoS("ignoring", "cg", cg.Id, "pid", pid)
			r.containersByPid[pid] = nil
		}
		return nil
	}
	if c := r.containersById[id]; c != nil {
		klog.Warningln("id conflict:", id)
		if cg.CreatedAt().After(c.cgroup.CreatedAt()) {
			c.cgroup = cg
			c.metadata = md
			c.runLogParser("")
			if c.nsConntrack != nil {
				_ = c.nsConntrack.Close()
				c.nsConntrack = nil
			}
		}
		setK8sTag(c, extensionTag, pid)
		r.containersByPid[pid] = c
		r.containersByCgroupId[cgId] = c
		return c
	}

	c, err := NewContainer(id, cg, md, r.hostConntrack, pid)
	if err != nil {
		klog.Warningf("failed to create container pid=%d cg=%s id=%s: %s", pid, cg.Id, id, err)
		return nil
	}

	//klog.InfoS("detected a new container", "pid", pid, "cg", cg.Id, "id", id)
	// add ns/workload/podname/pid/ctype
	//sType := fmt.Sprintf("%d", cg.ContainerType)

	setK8sTag(c, extensionTag, pid)
	if err := prometheus.WrapRegistererWith(setLabels(string(id),
		extensionTag[Namespace],
		extensionTag[PodName],
		extensionTag[ProcessName],
		fmt.Sprintf("%d", pid)), r.reg).Register(c); err != nil {
		klog.Warningln("failed to register container:", err)
		return nil
	}
	r.containersByPid[pid] = c
	r.containersByCgroupId[cgId] = c
	r.containersById[id] = c
	return c
}

func calcId(cg *cgroup.Cgroup, md *ContainerMetadata, pid uint32) (ContainerID, map[string]string) {
	extensionTag := map[string]string{Namespace: "", Workload: "", PodName: "", ProcessName: ""}
	if cg.ContainerType == cgroup.ContainerTypeSystemdService {
		if strings.HasPrefix(cg.ContainerId, "/system.slice/crio-conmon-") {
			return "", extensionTag
		}
		return ContainerID(cg.ContainerId), extensionTag
	}
	if cg.ContainerType == cgroup.ContainerTypeStandaloneProcess {
		procName := proc.GetProcName(pid)
		extensionTag[ProcessName] = procName
		return ContainerID(fmt.Sprintf("/%s/%s/%d", "standalone", proc.GetProcName(pid), pid)), extensionTag
	}
	switch cg.ContainerType {
	case cgroup.ContainerTypeDocker, cgroup.ContainerTypeContainerd, cgroup.ContainerTypeSandbox, cgroup.ContainerTypeCrio:
	default:
		return "", extensionTag
	}
	if cg.ContainerId == "" {
		return "", extensionTag
	}
	if md.labels["io.kubernetes.pod.name"] != "" {
		pod := md.labels["io.kubernetes.pod.name"]
		namespace := md.labels["io.kubernetes.pod.namespace"]
		name := md.labels["io.kubernetes.container.name"]
		if cg.ContainerType == cgroup.ContainerTypeSandbox {
			name = "sandbox"
		}
		if name == "" || name == "POD" { // skip pause containers
			return "", extensionTag
		}
		extensionTag[Namespace] = namespace
		extensionTag[Workload] = ""
		extensionTag[PodName] = pod
		extensionTag[ProcessName] = name
		return ContainerID(fmt.Sprintf("/k8s/%s/%s/%s", namespace, pod, name)), extensionTag
	}
	if taskNameParts := strings.SplitN(md.labels["com.docker.swarm.task.name"], ".", 3); len(taskNameParts) == 3 {
		namespace := md.labels["com.docker.stack.namespace"]
		service := md.labels["com.docker.swarm.service.name"]
		if namespace != "" {
			service = strings.TrimPrefix(service, namespace+"_")
		}
		if namespace == "" {
			namespace = "_"
		}
		return ContainerID(fmt.Sprintf("/swarm/%s/%s/%s", namespace, service, taskNameParts[1])), extensionTag
	}
	if md.name == "" { // should be "pure" dockerd container here
		klog.Warningln("empty dockerd container name for:", cg.ContainerId)
		return "", extensionTag
	}
	return ContainerID("/docker/" + md.name), extensionTag
}

func getContainerMetadata(cg *cgroup.Cgroup) (*ContainerMetadata, error) {
	switch cg.ContainerType {
	case cgroup.ContainerTypeDocker, cgroup.ContainerTypeContainerd, cgroup.ContainerTypeSandbox, cgroup.ContainerTypeCrio:
	default:
		return &ContainerMetadata{}, nil
	}
	if cg.ContainerId == "" {
		return &ContainerMetadata{}, nil
	}
	if cg.ContainerType == cgroup.ContainerTypeCrio {
		return CrioInspect(cg.ContainerId)
	}
	var dockerdErr error
	if dockerdClient != nil {
		md, err := DockerdInspect(cg.ContainerId)
		if err == nil {
			return md, nil
		}
		dockerdErr = err
	}
	var containerdErr error
	if containerdClient != nil {
		md, err := ContainerdInspect(cg.ContainerId)
		if err == nil {
			return md, nil
		}
		containerdErr = err
	}
	return nil, fmt.Errorf("failed to interact with dockerd (%s) or with containerd (%s)", dockerdErr, containerdErr)
}