From c4d47dc99a3ebf14cffed694ae692004320a8b64 Mon Sep 17 00:00:00 2001 From: Alexandre Proulx Date: Tue, 3 Mar 2026 00:40:07 +0000 Subject: [PATCH 01/10] cgroup: add podruntime.slice to recognized systemd slices (cherry picked from commit 5b2cb5f9f880c89ab3ac7983c0d6afa3643e47ad) --- cgroup/cgroup.go | 4 ++-- cgroup/cgroup_test.go | 10 ++++++++++ 2 files changed, 12 insertions(+), 2 deletions(-) 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) From 56503f677dfce10e267a6dc471712d9dd382112d Mon Sep 17 00:00:00 2001 From: Nikolay Sivko Date: Tue, 17 Mar 2026 14:53:02 -0300 Subject: [PATCH 02/10] add php-fpm detection to application type matching (cherry picked from commit 91fdf8d97d177b2b380ba258a95869f00bf0b0ef) --- containers/app.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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*$`) From 5cbecc340aa201c6069305492c9599f4f838ae7f Mon Sep 17 00:00:00 2001 From: Vishnu Kvs <116954249+vishnukumarkvs@users.noreply.github.com> Date: Tue, 14 Apr 2026 22:34:02 +0530 Subject: [PATCH 03/10] feat: add custom CA file support for TLS cert verification (cherry picked from commit fa609dac7bce1a06b224dcf8aaeff03f8525e040) --- common/api.go | 26 ++++++++++++++++++++++++++ flags/flags.go | 1 + logs/otel.go | 3 +-- profiling/profiling.go | 3 +-- prom/remote_writer.go | 3 +-- tracing/tracing.go | 3 +-- 6 files changed, 31 insertions(+), 8 deletions(-) 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/flags/flags.go b/flags/flags.go index 2edbb08..5ea4fa2 100644 --- a/flags/flags.go +++ b/flags/flags.go @@ -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..7d408d3 100644 --- a/prom/remote_writer.go +++ b/prom/remote_writer.go @@ -2,7 +2,6 @@ package prom import ( "crypto/md5" - "crypto/tls" "encoding/hex" "errors" "fmt" @@ -67,7 +66,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"), 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) From cc74b291dab6482a757ae8d8a924b6b4579857a4 Mon Sep 17 00:00:00 2001 From: Nikolay Sivko Date: Mon, 25 May 2026 11:41:21 -0300 Subject: [PATCH 04/10] send remote-write requests with an explicit Content-Length (cherry picked from commit a5999aff838846b8f5a460b141e534e16c89e204) --- prom/remote_writer.go | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/prom/remote_writer.go b/prom/remote_writer.go index 7d408d3..b7050c8 100644 --- a/prom/remote_writer.go +++ b/prom/remote_writer.go @@ -1,6 +1,7 @@ package prom import ( + "bytes" "crypto/md5" "encoding/hex" "errors" @@ -130,12 +131,13 @@ func (a *Agent) sendLoop() { } 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 } From affba50add06921392134e76fead00ec79a1e9c3 Mon Sep 17 00:00:00 2001 From: Nikolay Sivko Date: Mon, 25 May 2026 16:37:58 -0300 Subject: [PATCH 05/10] clean up stale container mounts to prevent duplicate disk metrics (cherry picked from commit 3d898f9b37b5972bc76447224d4d850fe888c135) Fork adaptation: upstream updates c.mounts in getMounts without a lock; here onFileOpen writes c.mounts under c.lock from the event goroutine, so the reconcile runs under c.lock and the statfs loop uses a snapshot. --- .gitignore | 3 +++ containers/container.go | 37 ++++++++++++++++++++++++++++++++++--- 2 files changed, 37 insertions(+), 3 deletions(-) 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/containers/container.go b/containers/container.go index b79a55d..9378846 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)) From a0735952e1044e1602ba3c3cb502686fdb7964e2 Mon Sep 17 00:00:00 2001 From: Nikolay Sivko Date: Tue, 4 Aug 2026 11:22:38 -0300 Subject: [PATCH 06/10] fix: systemd services disappearing after a restart (cherry picked from commit 75d6656426ff5139443cfb2c9c06c3622ce013c4) Fork adaptation: only the process-registration half is taken here. The fork already had ensureProcess (#353); its lookup now holds c.lock. Start events keep calling onProcessStart, which replaces a reused pid. The createdAt fallback for --min-container-age comes with that flag (B2). --- containers/container.go | 5 ++++- containers/registry.go | 8 ++++++++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/containers/container.go b/containers/container.go index 9378846..29d2f3f 100644 --- a/containers/container.go +++ b/containers/container.go @@ -2211,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 } From 3aeaf927c5e955206ee223832a5a209ac23b89d8 Mon Sep 17 00:00:00 2001 From: Nikolay Sivko Date: Wed, 5 Aug 2026 14:49:24 -0300 Subject: [PATCH 07/10] support multiple ephemeral port ranges (cherry picked from commit 2d90d4abb42df1dd2a8b8a8b8b060e04fcb87a30) --- common/net.go | 67 ++++++++++++++++++++++++++++++++-------------- common/net_test.go | 36 +++++++++++++++++++++++++ flags/flags.go | 2 +- 3 files changed, 84 insertions(+), 21 deletions(-) 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/flags/flags.go b/flags/flags.go index 5ea4fa2..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() From f07ba52d24838510a83b8c3967d75ca31d7cfc75 Mon Sep 17 00:00:00 2001 From: Nikolay Sivko Date: Mon, 10 Aug 2026 13:15:51 -0300 Subject: [PATCH 08/10] fix: don't get stuck on a bad spool chunk (cherry picked from commit 273c780a4eea756467de15c6b3d6fc4e3fe56cd6) --- prom/remote_writer.go | 49 +++++++++++++++++++++++++++++++------------ 1 file changed, 36 insertions(+), 13 deletions(-) diff --git a/prom/remote_writer.go b/prom/remote_writer.go index b7050c8..c42b558 100644 --- a/prom/remote_writer.go +++ b/prom/remote_writer.go @@ -111,25 +111,39 @@ 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 { + 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 + } + + 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 { // 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). @@ -155,6 +169,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)) @@ -184,6 +201,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 } @@ -199,6 +219,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 } From a40127a645c02b97f4c173d144e235a5200a37ce Mon Sep 17 00:00:00 2001 From: KR Ravindra <42912207+KR-Ravindra@users.noreply.github.com> Date: Tue, 8 Sep 2026 06:35:10 +0000 Subject: [PATCH 09/10] ebpftracer: report missing BPF tracing program types clearly On kernels built without CONFIG_BPF_EVENTS (e.g. NVIDIA JetPack 5 / L4T 5.10 with CONFIG_KPROBES=n), BPF_PROG_TYPE_TRACEPOINT and BPF_PROG_TYPE_KPROBE are compiled out and bpf(BPF_PROG_LOAD) fails with a bare EINVAL. The agent then exits with failed to load collection: program sched_process_exit: load program: invalid argument where the program name is whichever program happened to load first, and nothing points at the kernel configuration. Probe TracePoint and Kprobe support with cilium/ebpf/features before loading the collection and fail with an error naming the missing kernel option. Only a conclusive ebpf.ErrNotSupported is reported; any other probe result falls through to the real collection load so existing error paths are unchanged. Document the CONFIG_BPF_EVENTS requirement in the README. (cherry picked from commit f76895d48f7ee98da2224376140351e057c6c336) --- README.md | 1 + ebpftracer/tracer.go | 6 ++++++ 2 files changed, 7 insertions(+) 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/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}, }) From 2beb65ff8bc82e97dad60d7ea55f4fd2937a5052 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Wed, 7 Oct 2026 17:01:21 +0530 Subject: [PATCH 10/10] fix(prom): tolerate spool files removed by the other loop sendLoop and truncateSpoolIfNeeded (scrape goroutine) both remove spool files. A file removed between listing and use made sendLoop back off (5s to 1m) instead of moving to the next file, and made truncateSpoolIfNeeded return an error, so writeToSpool dropped the payload it was about to spool. Both now skip a file that is gone. --- prom/remote_writer.go | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/prom/remote_writer.go b/prom/remote_writer.go index c42b558..127363c 100644 --- a/prom/remote_writer.go +++ b/prom/remote_writer.go @@ -119,13 +119,19 @@ func (a *Agent) sendLoop() { err = a.send(fName) if err == nil { - if rmErr := os.Remove(fName); rmErr != nil { + if rmErr := os.Remove(fName); rmErr != nil && !errors.Is(rmErr, os.ErrNotExist) { klog.Warningln("failed to remove sent spool file:", rmErr) } 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) @@ -243,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 } }