ebpf/containers/registry.go

package containers

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

	"github.com/coroot/coroot-node-agent/kube"
	. "github.com/coroot/coroot-node-agent/utils"
	"github.com/coroot/coroot-node-agent/utils/enums"
	. "github.com/coroot/coroot-node-agent/utils/modelse"
	"github.com/coroot/coroot-node-agent/utils/try"
	. "github.com/coroot/coroot-node-agent/utils/worker"

	"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/ebpftracer/tracer"
	"github.com/coroot/coroot-node-agent/flags"
	"github.com/coroot/coroot-node-agent/proc"
	"github.com/prometheus/client_golang/prometheus"
	klog "github.com/sirupsen/logrus"
	"github.com/vishvananda/netns"
	"inet.af/netaddr"
)

const MinTrafficStatsUpdateInterval = 5 * time.Second

var (
	selfNetNs                = netns.None()
	hostNetNsId              = netns.None().UniqueId()
	agentPid                 = uint32(os.Getpid())
	containerIdRegexp        = regexp.MustCompile(`[a-z0-9]{64}`)
	cronjobPodName           = regexp.MustCompile(`([a-z0-9-]+)-([0-9]{8})-[bcdfghjklmnpqrstvwxz2456789]{5}`)
	cronjobPodScheduleWindow = 7 * 24 * time.Hour
)

const OPEN_STACK = 1

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 // calcId
	containersByCgroupId map[string]*Container      // fmt.Sprintf("%s/%d", cg.Id, pid)
	containersByPid      map[uint32]*Container
	ip2fqdn              map[netaddr.IP]string
	ip2fqdnLock          sync.Mutex

	processInfoCh chan<- ProcessInfo

	whiteListRules          WhiteListMap
	whiteLastUpdatedTime    int
	connServer              ServerWorker
	trafficStatsLastUpdated time.Time
	trafficStatsLock        sync.Mutex
	trafficStatsUpdateCh    chan *TrafficStatsUpdate
	nodeInfo                *NodeInfoT
	isFusing                bool
	IsFuseException         bool
	RegistryApps            map[uint32]AppStatusInfo
}

var (
	uprobes    []tracer.Uprobe
	uprobesMap map[string]tracer.Uprobe
)

func NewRegistry(reg prometheus.Registerer, kernelVersion string, nodeInfo *NodeInfoT, 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),

		hostConntrack: ct,

		containersById:       map[ContainerID]*Container{},
		containersByCgroupId: map[string]*Container{},
		containersByPid:      map[uint32]*Container{},
		ip2fqdn:              map[netaddr.IP]string{},

		processInfoCh: processInfoCh,

		tracer:               ebpftracer.NewTracer(kernelVersion, *flags.DisableL7Tracing, *flags.DisableE2ETracing, *flags.DisableStackTracing),
		whiteListRules:       make(WhiteListMap),
		trafficStatsUpdateCh: make(chan *TrafficStatsUpdate),
		nodeInfo:             nodeInfo,
		RegistryApps:         make(map[uint32]AppStatusInfo),
	}
	// 初始化软负载集群节点
	proxyClient, clientErr := NewProxyClient(*flags.ConfigServer, false)
	if clientErr == nil {
		// 负载健康检测
		try.Go(proxyClient.CheckEndpoints, CatchFn)
		klog.Infof("New Proxy Client success.config_server is [%s]", *flags.ConfigServer)
	} else {
		klog.WithError(clientErr).Errorf("NewProxyClient error, Please check [export CONFIG_ENDPOINT=ip:port]")
		return nil, clientErr
	}

	r.connServer, err = NewServerHTTPWorker()
	if err != nil {
		klog.Errorf("init connServer error:%s.", err)
		return nil, err
	}
	if !*flags.DisableRegisterHost {
		// Register Host
		err = r.RegisterHost()
		if err != nil {
			klog.WithError(err).Errorf("Failed Register Host.")
			return nil, err
		}
		try.Go(r.TaskRegisterHost, CatchFn)
	}

	if err = reg.Register(r); err != nil {
		return nil, err
	}

	//_, err = r.getWhiteList()
	//if err != nil {
	//	return nil, err
	//}
	try.Go(r.PullAllAppInfo, CatchFn)
	//go r.handleEvents(r.events)
	try.GoParams(r.handleEvents, CatchFn, r.events)

	if err = r.tracer.Run(r.events); err != nil {
		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.CloseContainers()
	r.tracer.Close()
	close(r.events)
}

func (r *Registry) CloseContainers() {
	for pid, c := range r.containersByPid {
		if c == nil {
			klog.Warnf("container for pid %d is nil, skipping", pid)
			continue
		}
		if c.Isl7AttachSuccess() {
			c.Detach(r.tracer, pid, APP_UNINSTALL)
		}
	}
}

func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) {
	gcTicker := time.NewTicker(gcInterval)
	defer gcTicker.Stop()
	var fuseOnce bool
	for {
		select {
		case now := <-gcTicker.C:
			//_, err1 := os.Stat("/tmp/fuse")
			//if err1 == nil {
			//	r.isFusing = true
			//} else {
			//	r.isFusing = false
			//}

			//_, err := r.pullWhiteList()
			_, err := r.pullWhiteListV2()
			if err != nil {
				klog.WithError(err).Errorf("connWhiteList error")
			}
			runtimeApps := make(map[uint32]AppStatusInfo)

			for pid, appInfo := range r.RegistryApps {
				c, ok := r.containersByPid[pid]
				if !ok {
					klog.WithFields(map[string]interface{}{
						"pid":     pid,
						"appname": appInfo.AppName,
					}).Infof("[handle] proc or container already deleted.")
					continue
				}
				if c == nil {
					klog.WithFields(map[string]interface{}{
						"pid":     pid,
						"appname": appInfo.AppName,
					}).Infof("[handle] container exists but c.data not ready.")
				}
			}

			for pid, c := range r.containersByPid {
				if c == nil {
					klog.WithField("pid", pid).Warningln("[handle] container is nil.")
					app, ok := r.RegistryApps[pid]
					if ok {
						klog.WithFields(map[string]interface{}{"pid": pid, "appname": app.AppName}).
							Infof("[handle] proc or container is deleted.")
					}
				}
				if !common.IsOpenFilter() && !fuseOnce {
					c.AgentCtrl(r, pid)
				}
				c.BuildActiveApps(runtimeApps, pid)

				cg, err := proc.ReadCgroup(pid)
				if err != nil {
					delete(r.containersByPid, pid)
					_, ok := r.RegistryApps[pid]
					if ok {
						klog.Infof("[handle] cgroup not find %d", pid)
					}
					if c != nil {
						c.onProcessExit(pid, false)
					}
					continue
				}
				if c != nil && cg.Id != c.cgroup.Id {
					delete(r.containersByPid, pid)
					_, ok := r.RegistryApps[pid]
					if ok {
						klog.Infof("[handle] cgroup id changed %d", pid)
					}
					c.onProcessExit(pid, false)
					// 重新验证cgroup，可能需要重新创建容器
					if newC := r.getOrCreateContainer(pid); newC != nil {
						// 更新容器引用
						r.containersByPid[pid] = newC
					}
				}
			}
			r.RegistryApps = runtimeApps
			saveAppInfo(r.RegistryApps)
			if r.isFusing {
				fuseOnce = true
			} else {
				fuseOnce = false
			}

			activeIPs := map[netaddr.IP]struct{}{}
			for id, c := range r.containersById {
				for dst := range c.connectLastAttempt {
					activeIPs[dst.IP()] = struct{}{}
				}
				if !c.Dead(now) {
					continue
				}
				klog.Debugln("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)
						app, ok := r.RegistryApps[pid]
						if ok {
							klog.WithFields(map[string]interface{}{"pid": pid, "appname": app.AppName}).
								Infof("[handle] Dead proc or container is deleted from containers.")
						}
					}
				}
				if ok := prometheus.WrapRegistererWith(setLabels(string(id),
					c.K8sContainer.ns,
					c.K8sContainer.workload,
					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()
		case u := <-r.trafficStatsUpdateCh:
			if u == nil {
				continue
			}
			if c := r.containersByPid[u.Pid]; c != nil {
				c.updateTrafficStats(u)
			}
		case e, more := <-ch:
			if e.Pid == uint32(os.Getpid()) {
				continue
			}
			if !more {
				return
			}
			switch e.Type {
			case ebpftracer.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)
					app, ok := r.RegistryApps[e.Pid]
					if ok {
						klog.WithFields(map[string]interface{}{"pid": e.Pid, "appname": app.AppName}).
							Infof("[handle] [EventTypeProcessStart nil] proc or container is deleted from containers.")
					}
				case c != nil: // revalidating by cgroup
					cg, err := proc.ReadCgroup(e.Pid)
					if err != nil || cg.Id != c.cgroup.Id {
						delete(r.containersByPid, e.Pid)
						app, ok := r.RegistryApps[e.Pid]
						if ok {
							klog.WithFields(map[string]interface{}{"pid": e.Pid, "appname": app.AppName}).
								Infof("[handle] [EventTypeProcessStart cgroup] proc or container is deleted from containers.")
						}
						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:
				c := r.containersByPid[e.Pid]
				cid := ""
				if c != nil {
					c.onProcessExit(e.Pid, e.Reason == ebpftracer.EventReasonOOMKill)
					cid = string(c.id)
				}
				delete(r.containersByPid, e.Pid)
				if app, ok := r.RegistryApps[e.Pid]; ok {
					klog.WithFields(map[string]interface{}{
						"pid":     e.Pid,
						"appname": app.AppName,
						"cid":     cid,
					}).Info("[handle] EventTypeProcessExit proc or container is deleted from containers.")
				}

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

			case ebpftracer.EventTypeListenOpen:
				//fmt.Println("ebpftracer.EventTypeListenOpen==================", e.Pid)
				if c := r.getOrCreateContainer(e.Pid); c != nil {
					c.onListenOpen(e.Pid, e.SrcAddr, false)
					// cmdline InstanceID agentID
					if !common.IsOpenFilter() || common.IsFilterPid(e.Pid) {
						if c.buildIDs(e.Pid) {
							c.eventReady()
						}
					}
					if common.IsOpenFilter() && common.IsFilterPid(e.Pid) {
						c.WhiteSettingInfo.AppName = enums.TestApp
						c.WhiteSettingInfo.WhiteStackSettingInfo.WhiteList = os.Getenv("WHITE_LIST")
						err := c.RegisterAppInfo(r, e.Pid)
						if err != nil {
							klog.WithError(err).Errorf("[registry] Failed registerAppInfo. pid is %d", e.Pid)
							continue
						}
						err = c.AttachUprobes(r.tracer, e.Pid, "Listen Open Event")
						if err != nil {
							klog.WithField("pid", e.Pid).WithError(err).Errorf("[AttachUprobes] [end] Failed attach stack trace!")
						}
						if !r.tracer.DisableStackTracing() {
							err = c.AttachStack(r.tracer, e.Pid)
							if err != nil {
								klog.WithField("pid", e.Pid).WithError(err).Errorf("[registry] [end] Failed attach stack trace!")
							}
						}
					}
				} else {
					klog.Infoln("TCP listen open from unknown container", e)
				}
			case ebpftracer.EventTypeAcceptOpen:
				//klog.Infoln("ebpftracer.EventTypeAcceptOpen==================", e.Pid)
				if c := r.getOrCreateContainer(e.Pid); c != nil {
					c.onAcceptOpen(e.Pid, e.Fd, e.SrcAddr, e.DstAddr, e.Timestamp, false, e.Duration)
					c.eventReady()
				} else {
					klog.Infoln("TCP connection from unknown container", e)
				}
			case ebpftracer.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, e.Duration)
					if !common.IsOpenFilter() || common.IsFilterPid(e.Pid) {
						if !c.checkEventReady() && c.buildIDs(e.Pid) {
							c.eventReady()
						}
					}
					if common.IsOpenFilter() && common.IsFilterPid(e.Pid) {
						c.WhiteSettingInfo.AppName = enums.TestApp
						if os.Getenv("APP_NAME") != "" {
							c.WhiteSettingInfo.AppName = os.Getenv("APP_NAME")
						}
						err := c.RegisterAppInfo(r, e.Pid)
						if err != nil {
							klog.WithError(err).Errorf("[registry] Failed registerAppInfo. pid is %d", e.Pid)
							continue
						}
						err = c.AttachUprobes(r.tracer, e.Pid, "Connection Open Event")
						if err != nil {
							klog.WithField("pid", e.Pid).WithError(err).Errorf("[AttachUprobes] [end] Failed attach stack trace!")
						}
						// 禁用stack
						if !r.tracer.DisableStackTracing() {
							err = c.AttachStack(r.tracer, e.Pid)
							if err != nil {
								klog.WithField("pid", e.Pid).WithError(err).Errorf("[registry] [end] Failed attach stack trace!")
							}
						} else {
							klog.Warnf("StackTrace tracing is disabled")
						}
					}
				} else {
					klog.Debugln("TCP connection from unknown container", e)
				}
			case ebpftracer.EventTypeListenClose:
				if c := r.containersByPid[e.Pid]; c != nil {
					c.onListenClose(e.Pid, e.SrcAddr)
				}
			case ebpftracer.EventTypeConnectionError:
				if c := r.getOrCreateContainer(e.Pid); c != nil {
					c.onConnectionOpen(e.Pid, e.Fd, e.SrcAddr, e.DstAddr, 0, true, e.Duration)
				} else {
					klog.Infoln("TCP connection error from unknown container", e)
				}
			case ebpftracer.EventTypeConnectionClose:
				if c := r.containersByPid[e.Pid]; c != nil {
					c.onConnectionClose(e)
				}
			case ebpftracer.EventTypeAcceptClose:
				if c := r.containersByPid[e.Pid]; c != nil {
					c.onAcceptClose(e)
				}
			case ebpftracer.EventTypeTCPRetransmit:
				srcDst := AddrPair{src: e.SrcAddr, dst: e.DstAddr}
				for _, c := range r.containersById {
					if c.onRetransmission(srcDst) {
						break
					}
				}
			case ebpftracer.EventTypeL7Request:

				klog.Debugln("e.L7Request Payload:", string(e.L7Request.Payload))
				if e.L7Request == nil {
					continue
				}

				if c := r.containersByPid[e.Pid]; c != nil {
					//fmt.Println("EventTypeL7Request", e.Pid, c.Isl7AttachSuccess())
					//a, _ := json.Marshal(e.L7Request)
					//fmt.Println("EventTypeL7Request", e.Pid, string(a))
					//klog.Debugln("Payload:", string(e.L7Request.Payload))
					ip2fqdn := c.onL7RequestApm(e.Pid, e.Fd, e.Timestamp, e.L7Request)
					r.ip2fqdnLock.Lock()
					for ip, fqdn := range ip2fqdn {
						r.ip2fqdn[ip] = fqdn
					}
					r.ip2fqdnLock.Unlock()
				}
			case ebpftracer.EventTypeFunEnt:
				if e.StackEvent == nil {
					continue
				}

				if c := r.containersByPid[uint32(e.StackEvent.Pid)]; c != nil {
					/*if e.StackEvent.Type == uint64(CodeTypeJava) {
						klog.Debugf("e.EventTypeFunEnt: TraceId:%d, Pid:%d, Location:%d, Goid:%d, TimeNs:%d, Ip:%X, CallerIp:%d, Bp:%d, CallerBp:%d\n", e.StackEvent.TraceId, e.StackEvent.Pid, e.StackEvent.Location, e.StackEvent.Goid, e.StackEvent.TimeNsStart, e.StackEvent.Ip, e.StackEvent.CallerIp, e.StackEvent.Bp, e.StackEvent.CallerBp)
						klog.Debugf("e.EventTypeFunEnt: TraceId: MethedName: %d -- %s -- %s", e.StackEvent.Type, e.StackEvent.MethedName, e.StackEvent.ClassName)
					} else {
						klog.Debugf("e.EventTypeFunEnt: TraceId:%d, Pid:%d, Location:%d, Goid:%d, TimeNs:%d, Ip:%X, CallerIp:%x, Bp:%x, CallerBp:%x\n", e.StackEvent.TraceId, e.StackEvent.Pid, e.StackEvent.Location, e.StackEvent.Goid, e.StackEvent.TimeNsStart, e.StackEvent.Ip, e.StackEvent.CallerIp, e.StackEvent.Bp, e.StackEvent.CallerBp)
					}*/
					c.StackProcess2(*e.StackEvent, r.tracer)
				} else {
					// fmt.Printf("e.EventTypeFunEnt ErrorError: TraceId:%d, Pid:%d, Location:%d, Goid:%d, TimeNs:%d, Ip:%X, CallerIp:%x, Bp:%x, CallerBp:%x", e.StackEvent.TraceId, e.StackEvent.Pid, e.StackEvent.Location, e.StackEvent.Goid, e.StackEvent.TimeNsStart, e.StackEvent.Ip, e.StackEvent.CallerIp, e.StackEvent.Bp, e.StackEvent.CallerBp)
					// fmt.Printf("e.EventTypeFunEnt ErrorError: TraceId:%x, FPid:%x, Nid:%x, Level:%d\n", e.StackEvent.Fpid, e.StackEvent.Nid, e.StackEvent.Level)
				}
			}
		}
	}
}

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.Infoln("ignoring without persisting", "cg", cg.Id, "pid", pid)
		} else {
			klog.Infoln("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, r)
	if err != nil {
		klog.Warningf("failed to create container pid=%d cg=%s id=%s: %s", pid, cg.Id, id, err)
		return nil
	}

	//klog.Infoln("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[Workload],
		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 (r *Registry) updateTrafficStatsIfNecessary() {
	r.trafficStatsLock.Lock()
	defer r.trafficStatsLock.Unlock()

	if time.Now().Sub(r.trafficStatsLastUpdated) < MinTrafficStatsUpdateInterval {
		return
	}
	iter := r.tracer.ActiveConnectionsIterator()
	cid := ConnectionId{}
	stats := Connection{}
	for iter.Next(&cid, &stats) {
		r.trafficStatsUpdateCh <- &TrafficStatsUpdate{
			Pid:           cid.PID,
			FD:            cid.FD,
			BytesSent:     stats.BytesSent,
			BytesReceived: stats.BytesReceived,
		}
	}
	if err := iter.Err(); err != nil {
		klog.Warningln(err)
	}
	r.trafficStatsUpdateCh <- nil
	r.trafficStatsLastUpdated = time.Now()
}

func calcId(cg *cgroup.Cgroup, md *ContainerMetadata, pid uint32) (ContainerID, map[string]string) {
	// 卡一下防止概率性获取为bash
	time.Sleep(1 * time.Millisecond)
	procName := proc.GetProcName(pid)
	extensionTag := map[string]string{Namespace: "", Workload: "", PodName: "", ProcessName: procName}
	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 {
		//extensionTag[ProcessName] = procName
		return ContainerID(fmt.Sprintf("/%s/%s/%d", "standalone", procName, 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
		if *flags.RunInContainer {
			extensionTag[Workload], _ = kube.GetWorkload(namespace, pod)
		}
		extensionTag[PodName] = pod
		//extensionTag[ProcessName] = name
		if g := cronjobPodName.FindStringSubmatch(pod); len(g) == 3 {
			now := time.Now()
			tsMiniutes, _ := strconv.ParseUint(g[2], 10, 64)
			scheduledAt := time.Unix(int64(tsMiniutes)*60, 0)
			if scheduledAt.After(now.Add(-cronjobPodScheduleWindow)) && scheduledAt.Before(now.Add(cronjobPodScheduleWindow)) {
				return ContainerID(fmt.Sprintf("/k8s-cronjob/%s/%s/%s", namespace, g[1], name)), extensionTag
			}
		}
		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.env != nil {
		allocId := md.env["NOMAD_ALLOC_ID"]
		group := md.env["NOMAD_GROUP_NAME"]
		job := md.env["NOMAD_JOB_NAME"]
		namespace := md.env["NOMAD_NAMESPACE"]
		task := md.env["NOMAD_TASK_NAME"]
		if allocId != "" && group != "" && job != "" && namespace != "" && task != "" {
			return ContainerID(fmt.Sprintf("/nomad/%s/%s/%s/%s/%s", namespace, job, group, allocId, task)), 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.ContainerTypeSystemdService:
		md := &ContainerMetadata{}
		md.systemdTriggeredBy = SystemdTriggeredBy(cg.ContainerId)
		return md, nil
	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)
}

type TrafficStatsUpdate struct {
	Pid           uint32
	FD            uint64
	BytesSent     uint64
	BytesReceived uint64
}

func (r *Registry) IsFusing() bool {
	return r.isFusing
}