diff --git a/.gitignore b/.gitignore index 761c3a0..3857bee 100644 --- a/.gitignore +++ b/.gitignore @@ -39,3 +39,6 @@ coverage.html # eBPF build outputs (regenerated via ebpftracer/make build) ebpftracer/ebpf/*.o + +# Local build helper (upstream) +/build.sh diff --git a/README.md b/README.md index 35cd2a6..997495f 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,7 @@ gathers container and host metrics, logs, and L7 traffic using eBPF and exposes them in Prometheus format. Minimum Linux kernel: **5.8** (L7 events use a BPF ring buffer). +The kernel must also be built with `CONFIG_BPF_EVENTS=y` (kprobe and tracepoint BPF programs); some embedded and vendor kernels disable it. > This project is a fork of > [coroot/coroot-node-agent](https://github.com/coroot/coroot-node-agent) diff --git a/cgroup/cgroup.go b/cgroup/cgroup.go index 91296e6..fbad1c6 100644 --- a/cgroup/cgroup.go +++ b/cgroup/cgroup.go @@ -23,7 +23,7 @@ var ( crioIdRegexp = regexp.MustCompile(`crio-([a-z0-9]{64})`) containerdIdRegexp = regexp.MustCompile(`cri-containerd[-:]([a-z0-9]{64})`) lxcIdRegexp = regexp.MustCompile(`/lxc/([^/]+)`) - systemSliceIdRegexp = regexp.MustCompile(`(/(system|runtime|reserved|kube|azure)\.slice/([^/]+))`) + systemSliceIdRegexp = regexp.MustCompile(`(/(system|runtime|reserved|kube|azure|podruntime)\.slice/([^/]+))`) talosIdRegexp = regexp.MustCompile(`/(system|podruntime)/([^/]+)`) lxcPayloadRegexp = regexp.MustCompile(`/lxc\.payload\.([^/]+)`) ) @@ -196,7 +196,7 @@ func containerByCgroup(cgroupPath string) (ContainerType, string, error) { return ContainerTypeUnknown, "", fmt.Errorf("invalid talos runtime cgroup %s", cgroupPath) } return ContainerTypeTalosRuntime, path.Join("/talos/", matches[2]), nil - case prefix == "system.slice" || prefix == "runtime.slice" || prefix == "reserved.slice" || prefix == "kube.slice" || prefix == "azure.slice": + case prefix == "system.slice" || prefix == "runtime.slice" || prefix == "reserved.slice" || prefix == "kube.slice" || prefix == "azure.slice" || prefix == "podruntime.slice": if strings.HasSuffix(cgroupPath, ".scope") { return ContainerTypeStandaloneProcess, "", nil } diff --git a/cgroup/cgroup_test.go b/cgroup/cgroup_test.go index 7e8b7af..4ac82cd 100644 --- a/cgroup/cgroup_test.go +++ b/cgroup/cgroup_test.go @@ -171,6 +171,16 @@ func TestContainerByCgroup(t *testing.T) { as.Equal("/azure.slice/walinuxagent.service", id) as.Nil(err) + typ, id, err = containerByCgroup("/podruntime.slice/containerd.service") + as.Equal(typ, ContainerTypeSystemdService) + as.Equal("/podruntime.slice/containerd.service", id) + as.Nil(err) + + typ, id, err = containerByCgroup("/podruntime.slice/kubelet.service") + as.Equal(typ, ContainerTypeSystemdService) + as.Equal("/podruntime.slice/kubelet.service", id) + as.Nil(err) + typ, id, err = containerByCgroup("/system.slice/system-postgresql.slice/postgresql@9.4-main.service") as.Equal(typ, ContainerTypeSystemdService) as.Equal("/system.slice/system-postgresql.slice", id) diff --git a/common/api.go b/common/api.go index 14fde0f..bcf9df7 100644 --- a/common/api.go +++ b/common/api.go @@ -1,7 +1,12 @@ package common import ( + "crypto/tls" + "crypto/x509" + "os" + "github.com/coroot/coroot-node-agent/flags" + "k8s.io/klog/v2" ) func AuthHeaders() map[string]string { @@ -11,3 +16,24 @@ func AuthHeaders() map[string]string { } return res } + +func TlsConfig() *tls.Config { + cfg := &tls.Config{InsecureSkipVerify: *flags.InsecureSkipVerify} + if *flags.CAFile != "" { + ca, err := os.ReadFile(*flags.CAFile) + if err != nil { + klog.Fatalln(err) + return cfg + } + pool, err := x509.SystemCertPool() + if err != nil { + klog.Warningln("failed to load system cert pool, starting with empty pool:", err) + pool = x509.NewCertPool() + } + if !pool.AppendCertsFromPEM(ca) { + klog.Fatalf("failed to parse CA from %s", *flags.CAFile) + } + cfg.RootCAs = pool + } + return cfg +} diff --git a/common/net.go b/common/net.go index 290ae04..4442492 100644 --- a/common/net.go +++ b/common/net.go @@ -36,24 +36,9 @@ func init() { } if r := flags.EphemeralPortRange; r != nil && *r != "" { klog.Infoln("ephemeral-port-range:", *r) - parts := strings.Split(*r, "-") - if len(parts) != 2 { - klog.Exitf("invalid port range: %s", *r) - } - from, err := strconv.ParseUint(parts[0], 10, 16) - if err != nil { - klog.Exitf("invalid port range: %s", *r) - } - to, err := strconv.ParseUint(parts[1], 10, 16) - if err != nil { - klog.Exitf("invalid port range: %s", *r) - } - if from > to { - klog.Exitf("invalid port range: %s", *r) - } - PortFilter = &portFilter{ - from: uint16(from), - to: uint16(to), + var err error + if PortFilter, err = newPortFilter(*r); err != nil { + klog.Exitln(err) } } var err error @@ -121,16 +106,58 @@ func (f connectionFilter) ShouldBeSkipped(dst, actualDst netaddr.IP) bool { return true } -type portFilter struct { +type portRange struct { from uint16 to uint16 } +type portFilter struct { + ranges []portRange +} + +func newPortFilter(s string) (*portFilter, error) { + f := &portFilter{} + for _, r := range strings.Fields(strings.ReplaceAll(s, ",", " ")) { + from, to, ok := strings.Cut(r, "-") + if !ok { + return nil, fmt.Errorf("invalid port range: %s", r) + } + f1, err := strconv.ParseUint(from, 10, 16) + if err != nil { + return nil, fmt.Errorf("invalid port range: %s", r) + } + t1, err := strconv.ParseUint(to, 10, 16) + if err != nil { + return nil, fmt.Errorf("invalid port range: %s", r) + } + if f1 > t1 { + return nil, fmt.Errorf("invalid port range: %s", r) + } + f.ranges = append(f.ranges, portRange{from: uint16(f1), to: uint16(t1)}) + } + if len(f.ranges) == 0 { + return nil, fmt.Errorf("invalid port range: %s", s) + } + return f, nil +} + +var wellKnownPorts = map[uint16]struct{}{ + 50051: {}, // gRPC +} + func (f *portFilter) ShouldBeSkipped(port uint16) bool { if f == nil { return false } - return port >= f.from && port <= f.to + if _, ok := wellKnownPorts[port]; ok { + return false + } + for _, r := range f.ranges { + if port >= r.from && port <= r.to { + return true + } + } + return false } type HostPort struct { diff --git a/common/net_test.go b/common/net_test.go index de3f2d9..6b798a8 100644 --- a/common/net_test.go +++ b/common/net_test.go @@ -157,3 +157,39 @@ func BenchmarkNormalizeFQDN(b *testing.B) { NormalizeFQDN("example.io.svc.default.cluster.local", "TypeA") } } + +func TestPortFilter(t *testing.T) { + var nilFilter *portFilter + assert.False(t, nilFilter.ShouldBeSkipped(40000)) + + f, err := newPortFilter("32768-60999") + assert.NoError(t, err) + assert.False(t, f.ShouldBeSkipped(32767)) + assert.True(t, f.ShouldBeSkipped(32768)) + assert.True(t, f.ShouldBeSkipped(60999)) + assert.False(t, f.ShouldBeSkipped(61000)) + assert.False(t, f.ShouldBeSkipped(50051)) // well-known port within the range + + for _, s := range []string{"1024-23768 30000-65535", "1024-23768,30000-65535", " 1024-23768 , 30000-65535 "} { + f, err = newPortFilter(s) + assert.NoError(t, err, s) + assert.False(t, f.ShouldBeSkipped(1023), s) + assert.True(t, f.ShouldBeSkipped(1024), s) + assert.True(t, f.ShouldBeSkipped(23768), s) + assert.False(t, f.ShouldBeSkipped(23769), s) // between the ranges + assert.False(t, f.ShouldBeSkipped(29999), s) + assert.True(t, f.ShouldBeSkipped(30000), s) + assert.True(t, f.ShouldBeSkipped(65535), s) + assert.False(t, f.ShouldBeSkipped(50051), s) + } + + f, err = newPortFilter("8080-8080") + assert.NoError(t, err) + assert.True(t, f.ShouldBeSkipped(8080)) + assert.False(t, f.ShouldBeSkipped(8081)) + + for _, s := range []string{"", " ", "32768", "-", "a-b", "32768-", "-60999", "60999-32768", "32768-65536", "-1-10"} { + _, err = newPortFilter(s) + assert.Error(t, err, s) + } +} diff --git a/containers/app.go b/containers/app.go index bec1869..0e8b56a 100644 --- a/containers/app.go +++ b/containers/app.go @@ -6,7 +6,7 @@ import ( ) var ( - phpCmd = regexp.MustCompile(`.*php\d*\.?\d*$`) + phpCmd = regexp.MustCompile(`.*php(-fpm)?\d*\.?\d*$`) pythonCmd = regexp.MustCompile(`.*python\d*\.?\d*$`) rubyCmd = regexp.MustCompile(`.*ruby\d*\.?\d*$`) nodejsCmd = regexp.MustCompile(`.*node(js)?\d*\.?\d*$`) diff --git a/containers/container.go b/containers/container.go index b79a55d..29d2f3f 100644 --- a/containers/container.go +++ b/containers/container.go @@ -361,12 +361,17 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { if disks, err := node.GetDisks(); err == nil { ioStat := c.cgroup.IOStat() + seenVolumes := map[string]struct{}{} for majorMinor, mounts := range c.getMounts() { var device string if dev := disks.GetParentBlockDevice(majorMinor); dev != nil { device = dev.Name } for mountPoint, fsStat := range mounts { + if _, ok := seenVolumes[mountPoint+":"+device]; ok { + continue + } + seenVolumes[mountPoint+":"+device] = struct{}{} dls := []string{mountPoint, device, c.metadata.volumes[mountPoint]} ch <- c.gauge(metrics.DiskSize, float64(fsStat.CapacityBytes), dls...) ch <- c.gauge(metrics.DiskUsed, float64(fsStat.UsedBytes), dls...) @@ -1749,19 +1754,45 @@ func (c *Container) updatePythonStats(s PythonStatsUpdate) { } func (c *Container) getMounts() map[string]map[string]*proc.FSStat { + // c.processes and c.mounts are mutated by the handleEvents goroutine + // (onFileOpen), so both are read and updated under c.lock. + c.lock.RLock() if len(c.mounts) == 0 { + c.lock.RUnlock() return nil } - // Copy pids under read lock — c.processes is mutated by handleEvents goroutine - c.lock.RLock() pids := make([]uint32, 0, len(c.processes)) for pid := range c.processes { pids = append(pids, pid) } c.lock.RUnlock() - res := map[string]map[string]*proc.FSStat{} + // Drop mounts that are gone from the container's mount namespace, so a + // remounted volume is not reported twice (stale + current entry). + var current map[string]proc.MountInfo + for _, pid := range pids { + if current = proc.GetMountInfo(pid); current != nil { + break + } + } + c.lock.Lock() + if len(current) > 0 { + for mntId := range c.mounts { + if mi, ok := current[mntId]; ok { + c.mounts[mntId] = mi + } else { + delete(c.mounts, mntId) + } + } + } + mounts := make([]proc.MountInfo, 0, len(c.mounts)) for _, mi := range c.mounts { + mounts = append(mounts, mi) + } + c.lock.Unlock() + + res := map[string]map[string]*proc.FSStat{} + for _, mi := range mounts { var stat *proc.FSStat for _, pid := range pids { s, err := proc.StatFS(proc.Path(pid, "root", mi.MountPoint)) @@ -2180,7 +2211,10 @@ const tlsExeRecheckInterval = 10 * time.Second // connection (onConnectionOpen ignores unknown pids) and left a short-lived // process's first TLS calls, often all of them, unprobed. func (c *Container) ensureProcess(pid uint32) *Process { - if p := c.processes[pid]; p != nil { + c.lock.RLock() + p := c.processes[pid] + c.lock.RUnlock() + if p != nil { return p } return c.onProcessStart(pid) diff --git a/containers/registry.go b/containers/registry.go index 2a0b87f..a873797 100644 --- a/containers/registry.go +++ b/containers/registry.go @@ -331,6 +331,8 @@ func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) { } } if c := r.getOrCreateContainer(e.Pid); c != nil { + // onProcessStart, not ensureProcess: it also replaces a + // registered process whose pid was reused. p := c.onProcessStart(e.Pid) if r.processInfoCh != nil && p != nil { r.processInfoCh <- ProcessInfo{Pid: p.Pid, ContainerId: c.id, StartedAt: p.StartedAt, Flags: p.Flags} @@ -613,6 +615,10 @@ func (r *Registry) getOrCreateContainer(pid uint32) *Container { r.containerLock.Lock() r.containersByPid[pid] = c r.containerLock.Unlock() + // The pid may have been mapped without its start event (e.g. a unit's + // new main process after a restart); register it so the container + // does not go zombie while the process is alive. + c.ensureProcess(pid) return c } r.containerLock.RUnlock() @@ -688,6 +694,7 @@ func (r *Registry) getOrCreateContainer(pid uint32) *Container { } if c := r.containersByCgroupId[cg.Id]; c != nil { r.containersByPid[pid] = c + c.ensureProcess(pid) return c } if c := r.containersById[id]; c != nil { @@ -718,6 +725,7 @@ func (r *Registry) getOrCreateContainer(pid uint32) *Container { r.containersByPid[pid] = c r.containersByCgroupId[cg.Id] = c r.containersById[id] = c + c.ensureProcess(pid) return c } diff --git a/ebpftracer/tracer.go b/ebpftracer/tracer.go index 9b84de2..959e95b 100644 --- a/ebpftracer/tracer.go +++ b/ebpftracer/tracer.go @@ -19,6 +19,7 @@ import ( "time" "github.com/cilium/ebpf" + "github.com/cilium/ebpf/features" "github.com/cilium/ebpf/link" "github.com/cilium/ebpf/perf" "github.com/cilium/ebpf/ringbuf" @@ -589,6 +590,11 @@ func (t *Tracer) ebpf(ch chan<- Event) error { } t.programVariant = variant _ = unix.Setrlimit(unix.RLIMIT_MEMLOCK, &unix.Rlimit{Cur: unix.RLIM_INFINITY, Max: unix.RLIM_INFINITY}) + for _, pt := range []ebpf.ProgramType{ebpf.TracePoint, ebpf.Kprobe} { + if err := features.HaveProgramType(pt); errors.Is(err, ebpf.ErrNotSupported) { + return fmt.Errorf("kernel does not support BPF %s programs (CONFIG_BPF_EVENTS is not set?): %w", pt, ebpf.ErrNotSupported) + } + } c, err := ebpf.NewCollectionWithOptions(collectionSpec, ebpf.CollectionOptions{ //Programs: ebpf.ProgramOptions{LogLevel: 2, LogSize: 20 * 1024 * 1024}, }) diff --git a/flags/flags.go b/flags/flags.go index 2edbb08..3edb2d5 100644 --- a/flags/flags.go +++ b/flags/flags.go @@ -40,7 +40,7 @@ var ( Envar("TRACK_PUBLIC_NETWORK"). Default("0.0.0.0/0"). Strings() - EphemeralPortRange = kingpin.Flag("ephemeral-port-range", "Destination and Listen TCP ports from this range will be skipped").Default("32768-60999").Envar("EPHEMERAL_PORT_RANGE").String() + EphemeralPortRange = kingpin.Flag("ephemeral-port-range", `Destination and Listen TCP ports from these ranges will be skipped, e.g. "32768-60999" or "1024-23768 30000-65535"`).Default("32768-60999").Envar("EPHEMERAL_PORT_RANGE").String() Provider = kingpin.Flag("provider", "`provider` label for `node_cloud_info` metric").Envar("PROVIDER").String() Region = kingpin.Flag("region", "`region` label for `node_cloud_info` metric").Envar("REGION").String() @@ -62,6 +62,7 @@ var ( LogsEndpoint = kingpin.Flag("logs-endpoint", "The URL of the endpoint to send logs to").Envar("LOGS_ENDPOINT").URL() ProfilesEndpoint = kingpin.Flag("profiles-endpoint", "The URL of the endpoint to send profiles to").Envar("PROFILES_ENDPOINT").URL() InsecureSkipVerify = kingpin.Flag("insecure-skip-verify", "whether to skip verifying the certificate or not").Envar("INSECURE_SKIP_VERIFY").Default("false").Bool() + CAFile = kingpin.Flag("ca-file", "Path to the custom CA certificate file").Envar("CA_FILE").String() ScrapeInterval = kingpin.Flag("scrape-interval", "How often to gather metrics from the agent").Default("15s").Envar("SCRAPE_INTERVAL").Duration() WalDir = kingpin.Flag("wal-dir", "Path to where the agent stores data (e.g. the metrics Write-Ahead Log)").Default("/tmp/nudgebee-node-agent").Envar("WAL_DIR").String() diff --git a/logs/otel.go b/logs/otel.go index 80abce1..9aeb2be 100644 --- a/logs/otel.go +++ b/logs/otel.go @@ -2,7 +2,6 @@ package logs import ( "context" - "crypto/tls" "time" otel "github.com/agoda-com/opentelemetry-logs-go" @@ -41,7 +40,7 @@ func Init(machineId, hostname, version string) { if endpointUrl.Scheme != "https" { opts = append(opts, otlplogshttp.WithInsecure()) } else { - opts = append(opts, otlplogshttp.WithTLSClientConfig(&tls.Config{InsecureSkipVerify: *flags.InsecureSkipVerify})) + opts = append(opts, otlplogshttp.WithTLSClientConfig(common.TlsConfig())) } client := otlplogshttp.NewClient(opts...) exporter, _ := otlplogs.NewExporter(context.Background(), otlplogs.WithClient(client)) diff --git a/profiling/profiling.go b/profiling/profiling.go index c5cde04..e698b3f 100644 --- a/profiling/profiling.go +++ b/profiling/profiling.go @@ -2,7 +2,6 @@ package profiling import ( "bytes" - "crypto/tls" "fmt" "io" "net/http" @@ -38,7 +37,7 @@ var ( httpClient = http.Client{ Timeout: UploadTimeout, Transport: &http.Transport{ - TLSClientConfig: &tls.Config{InsecureSkipVerify: *flags.InsecureSkipVerify}, + TLSClientConfig: common.TlsConfig(), }, } endpointUrl *url.URL diff --git a/prom/remote_writer.go b/prom/remote_writer.go index 4141ad8..127363c 100644 --- a/prom/remote_writer.go +++ b/prom/remote_writer.go @@ -1,8 +1,8 @@ package prom import ( + "bytes" "crypto/md5" - "crypto/tls" "encoding/hex" "errors" "fmt" @@ -67,7 +67,7 @@ func StartAgent(reg *prometheus.Registry, machineId, systemUuid string) error { httpClient: http.Client{ Timeout: RemoteWriteTimeout, Transport: &http.Transport{ - TLSClientConfig: &tls.Config{InsecureSkipVerify: *flags.InsecureSkipVerify}, + TLSClientConfig: common.TlsConfig(), }, }, spoolDir: path.Join(*flags.WalDir, "spool"), @@ -111,32 +111,53 @@ func (a *Agent) sendLoop() { time.Sleep(5 * time.Second) continue } - err = func() error { - if err := a.send(fName); err != nil { - return err + if fi, statErr := os.Stat(fName); statErr == nil && fi.Size() == 0 { + klog.Warningln("discarding empty spool file:", fName) + _ = os.Remove(fName) + continue + } + + err = a.send(fName) + if err == nil { + if rmErr := os.Remove(fName); rmErr != nil && !errors.Is(rmErr, os.ErrNotExist) { + klog.Warningln("failed to remove sent spool file:", rmErr) } - return os.Remove(fName) - }() - if err != nil { - dur := b.Duration() - klog.Warningf( - "failed to send metrics to %s, next attempt in %s: %s", - a.url, dur.String(), err, - ) - time.Sleep(dur) + b.Reset() + continue + } + + // truncateSpoolIfNeeded (scrape goroutine) removed the file after it + // was listed: move on to the next one instead of backing off. + if errors.Is(err, os.ErrNotExist) { + continue + } + + if errors.Is(err, errRejectedByCollector) { + klog.Warningf("dropping spool file %s: %s", fName, err) + _ = os.Remove(fName) + b.Reset() continue } - b.Reset() + + dur := b.Duration() + klog.Warningf( + "failed to send metrics to %s, next attempt in %s: %s", + a.url, dur.String(), err, + ) + time.Sleep(dur) } } +var errRejectedByCollector = errors.New("rejected by the collector") + func (a *Agent) send(fPath string) error { - f, err := os.Open(fPath) + // Read into memory so the request has an explicit Content-Length; streaming an *os.File + // makes net/http use chunked encoding, which some middleboxes reset (seen as EOF). + payload, err := os.ReadFile(fPath) if err != nil { return err } - defer f.Close() - req, err := http.NewRequest(http.MethodPost, a.url.String(), f) + req, err := http.NewRequest(http.MethodPost, a.url.String(), bytes.NewReader(payload)) if err != nil { return err } @@ -154,6 +175,9 @@ func (a *Agent) send(fPath string) error { } defer resp.Body.Close() if resp.StatusCode >= 300 { + if resp.StatusCode == http.StatusBadRequest || resp.StatusCode == http.StatusRequestEntityTooLarge { + return fmt.Errorf("%w: %s", errRejectedByCollector, resp.Status) + } return errors.New(resp.Status) } klog.Infof("sent metrics in %s", time.Since(t).Truncate(time.Millisecond)) @@ -183,6 +207,9 @@ func (a *Agent) scrape() error { } func (a *Agent) writeToSpool(timestamp int64, payload []byte) error { + if len(payload) == 0 { + return nil + } if err := a.truncateSpoolIfNeeded(); err != nil { return err } @@ -198,6 +225,9 @@ func (a *Agent) writeToSpool(timestamp int64, payload []byte) error { if _, err = f.Write(payload); err != nil { return err } + if err = f.Sync(); err != nil { + return err + } if err = f.Close(); err != nil { return err } @@ -219,13 +249,18 @@ func (a *Agent) truncateSpoolIfNeeded() error { for _, f := range files { st, err := os.Stat(f) if err != nil { + // sendLoop removed it after sending: failing here would drop + // the payload being spooled. + if errors.Is(err, os.ErrNotExist) { + continue + } return err } totalSize += st.Size() } if totalSize > a.maxSpoolSize { klog.Warningln("spool size exceeded, removing the oldest file:", files[0]) - if err = os.Remove(files[0]); err != nil { + if err = os.Remove(files[0]); err != nil && !errors.Is(err, os.ErrNotExist) { return err } } diff --git a/tracing/tracing.go b/tracing/tracing.go index 151fb60..3a276a1 100644 --- a/tracing/tracing.go +++ b/tracing/tracing.go @@ -2,7 +2,6 @@ package tracing import ( "context" - "crypto/tls" "fmt" "math/rand" "net" @@ -70,7 +69,7 @@ func Init(machineId, hostname, version string) { if endpointUrl.Scheme != "https" { opts = append(opts, otlptracehttp.WithInsecure()) } else { - opts = append(opts, otlptracehttp.WithTLSClientConfig(&tls.Config{InsecureSkipVerify: *flags.InsecureSkipVerify})) + opts = append(opts, otlptracehttp.WithTLSClientConfig(common.TlsConfig())) } client := otlptracehttp.NewClient(opts...) exporter, err := otlptrace.New(context.Background(), client)