diff --git a/CHANGES.md b/CHANGES.md index 6b72e40b..63f97d4f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -27,10 +27,16 @@ Release Notes. * Support detect ztunnel environment in the inbound request. * Increase the transmit buffer size in the network profiling and access log module. * Reduce eBPF verifier complexity by converting hot inline helpers to BPF-to-BPF calls, fixing the "program too large" failure when loading access log programs on newer kernels. +* Reduce memory by not retaining ELF symbols for processes that are not being profiled. +* Increase the default access log protocol per-CPU buffer to 1MB to reduce dropped samples. +* Aggregate dropped perf event sample warnings periodically instead of logging every burst. +* Lower the log level for connection query errors of short-lived exited processes. +* Enable pprof by default but bind it to 127.0.0.1 so it is not exposed on the network. #### Bug Fixes * Fix the base image cannot run in the arm64. * Fix process fork tracepoint reporting thread TID instead of process TGID, causing repeated process detected/dead churn. +* Fix panic in the access log module when handling HTTP/2 streams without a body. #### Documentation * Add a dead link checker in the CI. diff --git a/configs/rover_configs.yaml b/configs/rover_configs.yaml index c497c16e..a5a6b0c9 100644 --- a/configs/rover_configs.yaml +++ b/configs/rover_configs.yaml @@ -151,7 +151,7 @@ access_log: queue_size: ${ROVER_ACCESS_LOG_CONNECTION_ANALYZE_QUEUE_SIZE:2000} protocol_analyze: # The size of socket data buffer on each CPU - per_cpu_buffer: ${ROVER_ACCESS_LOG_PROTOCOL_ANALYZE_PER_CPU_BUFFER:400KB} + per_cpu_buffer: ${ROVER_ACCESS_LOG_PROTOCOL_ANALYZE_PER_CPU_BUFFER:1MB} # The count of parallel protocol event parse parse_parallels: ${ROVER_ACCESS_LOG_PROTOCOL_ANALYZE_PARSE_PARALLELS:2} # The count of parallel protocol analyzer @@ -161,6 +161,8 @@ access_log: pprof: # Is active the pprof - active: ${ROVER_PPROF_ACTIVE:false} + active: ${ROVER_PPROF_ACTIVE:true} + # The bind host of the pprof HTTP server, defaults to 127.0.0.1 so it only listens on the local host + host: ${ROVER_PPROF_HOST:127.0.0.1} # The bind port of the pprof HTTP server port: ${ROVER_PPROF_PORT:6060} \ No newline at end of file diff --git a/docs/en/setup/configuration/pprof.md b/docs/en/setup/configuration/pprof.md index 98e5675b..51c5f598 100644 --- a/docs/en/setup/configuration/pprof.md +++ b/docs/en/setup/configuration/pprof.md @@ -4,10 +4,11 @@ Pprof is a feature to collect self runtime profiling data through `pprof` module ## Configuration -| Name | Default | Environment Key | Description | -|-----------|---------|----------------------|-------------------------------------| -| `enabled` | `false` | `ROVER_PPROF_ACTIVE` | Enable pprof module. | -| `port` | `6060` | `ROVER_PPROF_PORT` | The HTTP port to expose pprof data. | +| Name | Default | Environment Key | Description | +|-----------|-------------|----------------------|------------------------------------------------------------------------------------| +| `enabled` | `true` | `ROVER_PPROF_ACTIVE` | Enable pprof module. | +| `host` | `127.0.0.1` | `ROVER_PPROF_HOST` | The bind host of the pprof HTTP server, only listens on the local host by default. | +| `port` | `6060` | `ROVER_PPROF_PORT` | The HTTP port to expose pprof data. | ## Expose Paths diff --git a/pkg/accesslog/collector/protocols/http1.go b/pkg/accesslog/collector/protocols/http1.go index 8a76a2c3..c24b1bc5 100644 --- a/pkg/accesslog/collector/protocols/http1.go +++ b/pkg/accesslog/collector/protocols/http1.go @@ -220,10 +220,11 @@ func (p *HTTP1Protocol) HandleHTTPData(metrics *HTTP1Metrics, _ *PartitionConnec details, idRange, allInclude = AppendSocketDetailsFromBuffer(details, response.HeaderBuffer(), idRange, allInclude) details, idRange, allInclude = AppendSocketDetailsFromBuffer(details, response.BodyBuffer(), idRange, allInclude) - if !allInclude { + if !allInclude || idRange == nil || len(details) == 0 { + from, to := DataIDRangeBounds(idRange) return fmt.Errorf("cannot found full detail events for HTTP/1.x protocol, "+ "data id: %d-%d, current details count: %d", - idRange.From, idRange.To, len(details)) + from, to, len(details)) } http1Log.Debugf("found fully HTTP1 request and response, contains %d detail events, "+ diff --git a/pkg/accesslog/collector/protocols/http2.go b/pkg/accesslog/collector/protocols/http2.go index 0db9741d..3e3aeb37 100644 --- a/pkg/accesslog/collector/protocols/http2.go +++ b/pkg/accesslog/collector/protocols/http2.go @@ -245,17 +245,19 @@ func (r *HTTP2Protocol) validateIsStreamOpenTooLong(connection *PartitionConnect func (r *HTTP2Protocol) HandleWholeStream(_ *PartitionConnection, stream *HTTP2Streaming) error { details := make([]events.SocketDetail, 0) - var allInclude = true + allInclude := true var idRange *buffer.DataIDRange details, idRange, allInclude = AppendSocketDetailsFromBuffer(details, stream.ReqHeaderBuffer, idRange, allInclude) details, idRange, allInclude = AppendSocketDetailsFromBuffer(details, stream.ReqBodyBuffer, idRange, allInclude) details, idRange, allInclude = AppendSocketDetailsFromBuffer(details, stream.RespHeaderBuffer, idRange, allInclude) details, idRange, allInclude = AppendSocketDetailsFromBuffer(details, stream.RespBodyBuffer, idRange, allInclude) - if !allInclude { - return fmt.Errorf("cannot found any detail events for HTTP/2 protocol, data id: %d-%d, current details count: %d", - stream.ReqHeaderBuffer.FirstSocketBuffer().DataID(), stream.RespBodyBuffer.LastSocketBuffer().DataID(), - len(details)) + // idRange is nil / details empty when no buffer contributed any event (e.g. all buffers absent); + // guard it so the error path never dereferences a nil buffer as it used to (which panicked). + if !allInclude || idRange == nil || len(details) == 0 { + from, to := DataIDRangeBounds(idRange) + return fmt.Errorf("cannot found full detail events for HTTP/2 protocol, data id: %d-%d, current details count: %d", + from, to, len(details)) } idRange.DeleteDetails(stream.ReqHeaderBuffer) diff --git a/pkg/accesslog/collector/protocols/http2_test.go b/pkg/accesslog/collector/protocols/http2_test.go new file mode 100644 index 00000000..9d801873 --- /dev/null +++ b/pkg/accesslog/collector/protocols/http2_test.go @@ -0,0 +1,186 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package protocols + +import ( + "testing" + "time" + + "github.com/apache/skywalking-rover/pkg/accesslog/common" + aclevents "github.com/apache/skywalking-rover/pkg/accesslog/events" + analyzeevents "github.com/apache/skywalking-rover/pkg/profiling/task/network/analyze/events" + "github.com/apache/skywalking-rover/pkg/tools/buffer" + + v3 "skywalking.apache.org/repo/goapi/collect/ebpf/accesslog/v3" +) + +// buildDetailBuffer builds a minimal, fully-captured buffer that holds a single detail event with +// the given data id, so BuildDetails() is non-empty and BuildTotalDataIDRange() covers it. +func buildDetailBuffer(dataID, start, end uint64) *buffer.Buffer { + b := buffer.NewBuffer() + b.AppendDataEvent(&analyzeevents.SocketDataUploadEvent{ + DataID0: dataID, DataLen: 1, StartTime0: start, EndTime0: end, Finished: 1, + }) + b.AppendDetailEvent(&aclevents.SocketDetailEvent{DataID0: dataID, StartTime: start, EndTime: end}) + b.PrepareForReading() + return b +} + +type noopQueueConsumer struct{} + +func (noopQueueConsumer) Consume(_ chan common.KernelLog, _ chan common.ProtocolLog) {} + +// TestAppendSocketDetailsFromBuffer covers the buffer-classification logic at the heart of the panic +// fix: a nil (absent) buffer is skipped, a present-but-empty buffer is incomplete, and a +// fully-captured buffer contributes its details. +func TestAppendSocketDetailsFromBuffer(t *testing.T) { + // a nil buffer with allInclude=true must be skipped, not marked incomplete (the core fix). + if _, idRange, include := AppendSocketDetailsFromBuffer(nil, nil, nil, true); !include || idRange != nil { + t.Fatalf("nil buffer should be skipped: include=%v idRange=%v", include, idRange) + } + // once incomplete, a following nil buffer keeps the stream incomplete. + if _, _, include := AppendSocketDetailsFromBuffer(nil, nil, nil, false); include { + t.Fatal("incomplete state must be propagated through a nil buffer") + } + // a present-but-empty buffer is treated as incomplete. + if _, _, include := AppendSocketDetailsFromBuffer(nil, buffer.NewBuffer(), nil, true); include { + t.Fatal("empty buffer should be marked incomplete") + } + // a present buffer whose details fall outside its captured data-id range (e.g. dropped perf + // samples left a gap) is incomplete, not skipped. + incomplete := buffer.NewBuffer() + incomplete.AppendDataEvent(&analyzeevents.SocketDataUploadEvent{DataID0: 1, DataLen: 1, Finished: 1}) + incomplete.AppendDetailEvent(&aclevents.SocketDetailEvent{DataID0: 5}) + incomplete.PrepareForReading() + if _, _, include := AppendSocketDetailsFromBuffer(nil, incomplete, nil, true); include { + t.Fatal("buffer with details outside its data-id range should be incomplete") + } + + // a fully-captured buffer contributes its detail and stays complete. + details, idRange, include := AppendSocketDetailsFromBuffer(nil, buildDetailBuffer(1, 10, 20), nil, true) + if !include || len(details) != 1 || idRange == nil || idRange.From != 1 || idRange.To != 1 { + t.Fatalf("complete buffer: include=%v len=%d idRange=%v", include, len(details), idRange) + } +} + +func TestDataIDRangeBounds(t *testing.T) { + if from, to := DataIDRangeBounds(nil); from != 0 || to != 0 { + t.Fatalf("nil range must be (0,0), got (%d,%d)", from, to) + } + if from, to := DataIDRangeBounds(&buffer.DataIDRange{From: 3, To: 9}); from != 3 || to != 9 { + t.Fatalf("expected (3,9), got (%d,%d)", from, to) + } +} + +// TestHTTP2HandleWholeStreamIncompleteNoPanic guards against the SIGSEGV that used to happen when a +// stream reached HandleWholeStream without a fully-captured set of detail events. The error path +// previously dereferenced possibly-nil buffers (stream.RespBodyBuffer.LastSocketBuffer().DataID()); +// it must now return an error safely instead of panicking. +func TestHTTP2HandleWholeStreamIncompleteNoPanic(t *testing.T) { + r := &HTTP2Protocol{} + tests := []struct { + name string + stream *HTTP2Streaming + }{ + { + name: "all buffers absent", + stream: &HTTP2Streaming{ReqHeader: map[string]string{}, RespHeader: map[string]string{}}, + }, + { + name: "present but empty request header buffer, bodies absent", + stream: &HTTP2Streaming{ + ReqHeader: map[string]string{}, + RespHeader: map[string]string{}, + ReqHeaderBuffer: buffer.NewBuffer(), + }, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + err := r.HandleWholeStream(nil, test.stream) + if err == nil { + t.Fatal("expected an error when detail events are incomplete, got nil") + } + }) + } +} + +// TestHTTP2HandleWholeStreamReportsWhenBodiless proves the root-cause fix: a stream with request and +// response headers but no bodies (e.g. a GET with a 204/304 response that ends on the HEADERS frame) +// must now be reported instead of being dropped/panicking on the nil body buffers. +func TestHTTP2HandleWholeStreamReportsWhenBodiless(t *testing.T) { + r := &HTTP2Protocol{ctx: &common.AccessLogContext{ + Queue: common.NewQueue(100, time.Minute, noopQueueConsumer{}), + }} + stream := &HTTP2Streaming{ + ReqHeader: map[string]string{":path": "/", ":authority": "example"}, + RespHeader: map[string]string{":status": "204"}, + ReqHeaderBuffer: buildDetailBuffer(1, 10, 20), + RespHeaderBuffer: buildDetailBuffer(2, 30, 40), + // ReqBodyBuffer and RespBodyBuffer are intentionally nil (bodiless). + } + if err := r.HandleWholeStream(nil, stream); err != nil { + t.Fatalf("bodiless stream should be reported, got error: %v", err) + } +} + +func TestHTTP2ParseHTTPMethod(t *testing.T) { + r := &HTTP2Protocol{} + tests := []struct { + method string + expect v3.AccessLogHTTPProtocolRequestMethod + }{ + {"", v3.AccessLogHTTPProtocolRequestMethod_Get}, // absent method defaults to GET + {"post", v3.AccessLogHTTPProtocolRequestMethod_Post}, // lower-cased method is upper-cased first + {"patch", v3.AccessLogHTTPProtocolRequestMethod_Patch}, + {"weird", v3.AccessLogHTTPProtocolRequestMethod_Get}, // unknown method falls back to GET + } + for _, test := range tests { + stream := &HTTP2Streaming{ReqHeader: map[string]string{}} + if test.method != "" { + stream.ReqHeader[":method"] = test.method + } + if got := r.ParseHTTPMethod(stream); got != test.expect { + t.Fatalf("method %q: expected %v, got %v", test.method, test.expect, got) + } + } +} + +func TestHTTP2FirstDetail(t *testing.T) { + r := &HTTP2Protocol{} + def := &aclevents.SocketDetailEvent{DataID0: 99} + if got := r.FirstDetail(nil, def); got != def { + t.Fatal("nil buffer should return the default detail") + } + if got := r.FirstDetail(buffer.NewBuffer(), def); got != def { + t.Fatal("empty buffer should return the default detail") + } + if got := r.FirstDetail(buildDetailBuffer(7, 1, 2), def); got == def || got.DataID() != 7 { + t.Fatalf("present buffer should return its own first detail, got DataID=%d", got.DataID()) + } +} + +func TestHTTP2BufferSizeOfZero(t *testing.T) { + r := &HTTP2Protocol{} + if size := r.BufferSizeOfZero(nil); size != 0 { + t.Fatalf("nil buffer size must be 0, got %d", size) + } + if size := r.BufferSizeOfZero(buildDetailBuffer(1, 1, 2)); size == 0 { + t.Fatal("present buffer size must be non-zero") + } +} diff --git a/pkg/accesslog/collector/protocols/protocol.go b/pkg/accesslog/collector/protocols/protocol.go index 673796c9..63d68d8f 100644 --- a/pkg/accesslog/collector/protocols/protocol.go +++ b/pkg/accesslog/collector/protocols/protocol.go @@ -58,9 +58,16 @@ type Protocol interface { func AppendSocketDetailsFromBuffer(result []events.SocketDetail, buf *buffer.Buffer, dataIDRange *buffer.DataIDRange, allDetailInclude bool) ([]events.SocketDetail, *buffer.DataIDRange, bool) { - if buf == nil || !allDetailInclude { + // a previous buffer was already incomplete: keep propagating the incompleteness. + if !allDetailInclude { return result, dataIDRange, false } + // a nil buffer means this part of the message is legitimately absent (a request without body, + // or a response that ends the stream on the HEADERS frame such as 204/304/redirect or a gRPC + // trailers-only response): skip it rather than treating the whole message as incomplete. + if buf == nil { + return result, dataIDRange, true + } details := buf.BuildDetails() if details == nil || details.Len() == 0 { return result, dataIDRange, false @@ -81,6 +88,15 @@ func AppendSocketDetailsFromBuffer(result []events.SocketDetail, buf *buffer.Buf return result, dataIDRange.Append(currentDataIDRange), true } +// DataIDRangeBounds returns the from/to data id bounds of the range, or (0, 0) when it is nil, so +// callers can build an error message without dereferencing a possibly-nil range. +func DataIDRangeBounds(dataIDRange *buffer.DataIDRange) (from, to uint64) { + if dataIDRange != nil { + from, to = dataIDRange.From, dataIDRange.To + } + return +} + func AnalyzeTraceInfo(fetcher func(key string) string, protocolLog *logger.Logger) *v3.AccessLogTraceInfo { context, err := tracing.AnalyzeTracingContext(func(key string) string { return fetcher(key) diff --git a/pkg/pprof/config.go b/pkg/pprof/config.go index 496957ce..9f29e83e 100644 --- a/pkg/pprof/config.go +++ b/pkg/pprof/config.go @@ -22,7 +22,10 @@ import "github.com/apache/skywalking-rover/pkg/module" type Config struct { module.Config `mapstructure:",squash"` - Port int `mapstructure:"port"` + // Host is the bind address of the pprof HTTP server. It defaults to 127.0.0.1 so the debug + // endpoints are only reachable from the local host and are not exposed on the network. + Host string `mapstructure:"host"` + Port int `mapstructure:"port"` } func (c *Config) IsActive() bool { diff --git a/pkg/pprof/module.go b/pkg/pprof/module.go index 8505db9c..39684b67 100644 --- a/pkg/pprof/module.go +++ b/pkg/pprof/module.go @@ -67,7 +67,7 @@ func (m *Module) Start(_ context.Context, mgr *module.Manager) error { defer m.mutex.Unlock() m.server = &http.Server{ - Addr: fmt.Sprintf(":%d", m.config.Port), + Addr: fmt.Sprintf("%s:%d", m.config.Host, m.config.Port), ReadHeaderTimeout: 3 * time.Second, Handler: mux, } diff --git a/pkg/process/api/process.go b/pkg/process/api/process.go index aed6d181..75a09ba2 100644 --- a/pkg/process/api/process.go +++ b/pkg/process/api/process.go @@ -78,6 +78,8 @@ type DetectedProcess interface { DetectType() ProcessDetectType // ProfilingStat of process ProfilingStat() *profiling.Info + // SupportProfiling reports whether the process can be profiled without retaining its symbols + SupportProfiling() bool // ExposePorts define which ports are exposed ExposePorts() []int // ExposeHosts define which hosts are exposed diff --git a/pkg/process/finders/base/tool.go b/pkg/process/finders/base/tool.go index f2413be1..725cf615 100644 --- a/pkg/process/finders/base/tool.go +++ b/pkg/process/finders/base/tool.go @@ -44,6 +44,16 @@ func BuildProfilingStat(ps *process.Process) (*profiling.Info, error) { return process_tool.ProfilingStat(ps.Pid, exePath) } +// SupportProfiling reports whether the process's executable can be profiled without retaining the +// parsed symbol data in memory (see process_tool.SupportProfiling). +func SupportProfiling(ps *process.Process) (bool, error) { + exePath := tryToFindFileExecutePath(ps) + if exePath == "" { + return false, fmt.Errorf("could not found executable file") + } + return process_tool.SupportProfiling(exePath) +} + func tryToFindFileExecutePath(ps *process.Process) string { exe, err := ps.Exe() if err != nil { diff --git a/pkg/process/finders/kubernetes/finder.go b/pkg/process/finders/kubernetes/finder.go index 21eae18d..2edfbbfa 100644 --- a/pkg/process/finders/kubernetes/finder.go +++ b/pkg/process/finders/kubernetes/finder.go @@ -350,7 +350,7 @@ func (f *ProcessFinder) BuildNecessaryProperties(ps api.DetectedProcess) []*comm return []*commonv3.KeyStringValuePair{ { Key: "support_ebpf_profiling", - Value: strconv.FormatBool(ps.ProfilingStat() != nil), + Value: strconv.FormatBool(ps.SupportProfiling()), }, } } diff --git a/pkg/process/finders/kubernetes/process.go b/pkg/process/finders/kubernetes/process.go index bf206837..c96e47a2 100644 --- a/pkg/process/finders/kubernetes/process.go +++ b/pkg/process/finders/kubernetes/process.go @@ -18,6 +18,9 @@ package kubernetes import ( + "errors" + "os" + "strings" "sync" "github.com/shirou/gopsutil/process" @@ -31,11 +34,11 @@ type Process struct { original *process.Process // process data - pid int32 - cmd string - profilingOnce sync.Once - profiling *profiling.Info - podContainer *PodContainer + pid int32 + cmd string + supportProfilingOnce sync.Once + supportProfiling bool + podContainer *PodContainer // entity for the backend entity *api.ProcessEntity @@ -67,12 +70,22 @@ func (p *Process) DetectType() api.ProcessDetectType { return api.Kubernetes } +// ProfilingStat builds the full profiling info (heavy: parses and retains the ELF symbol tables). +// It intentionally does NOT cache the result on the process: the info is only needed while a +// profiling task runs and is held by that task's runner, so it is released once the task ends. func (p *Process) ProfilingStat() *profiling.Info { - p.profilingOnce.Do(func() { - stat, _ := base.BuildProfilingStat(p.original) - p.profiling = stat + stat, _ := base.BuildProfilingStat(p.original) + return stat +} + +// SupportProfiling reports whether the process can be profiled, WITHOUT retaining the symbol data. +// The result is a bool that is cheap to cache, so keep-alive reporting does not re-parse the ELF +// symbol tables of every discovered process on each report. +func (p *Process) SupportProfiling() bool { + p.supportProfilingOnce.Do(func() { + p.supportProfiling, _ = base.SupportProfiling(p.original) }) - return p.profiling + return p.supportProfiling } func (p *Process) PodContainer() *PodContainer { @@ -89,7 +102,15 @@ func (p *Process) ExposePorts() []int { } connections, err := p.original.Connections() if err != nil { - log.Warnf("query the process connection error: pid: %d, error: %v", p.pid, err) + // A short-lived process may exit between discovery and this query, making its + // /proc//net/* files disappear. That is an expected race for ephemeral processes + // (e.g. curl/sleep from load generators), not a real error, so keep it at debug level + // to avoid flooding the log. + if errors.Is(err, os.ErrNotExist) || strings.Contains(err.Error(), "no such file or directory") { + log.Debugf("skip querying connections for the exited process, pid: %d, error: %v", p.pid, err) + } else { + log.Warnf("query the process connection error: pid: %d, error: %v", p.pid, err) + } return result } for _, c := range connections { diff --git a/pkg/tools/btf/linker.go b/pkg/tools/btf/linker.go index 3abb43ba..271a10d3 100644 --- a/pkg/tools/btf/linker.go +++ b/pkg/tools/btf/linker.go @@ -199,7 +199,9 @@ func (m *Linker) asyncReadEvent(rd *perf.Reader, emap *ebpf.Map, recordPool *per } if record.LostSamples != 0 { - log.Warnf("perf event queue(%s) full, dropped %d samples", emap.String(), record.LostSamples) + // accumulate and report in an aggregated, periodic summary instead of logging + // every burst on its own (see lostsamples.go). + recordLostSamples(emap.String(), record.LostSamples) recordPool.PutRecord(record) continue } diff --git a/pkg/tools/btf/lostsamples.go b/pkg/tools/btf/lostsamples.go new file mode 100644 index 00000000..8114fd92 --- /dev/null +++ b/pkg/tools/btf/lostsamples.go @@ -0,0 +1,75 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package btf + +import ( + "sync" + "time" +) + +// lostSampleLogInterval is how often the aggregated dropped-sample summary is logged. The kernel +// drops samples in bursts, so logging every single occurrence floods the log; instead the counts +// are accumulated and reported once per interval. +const lostSampleLogInterval = 30 * time.Second + +var lostSamples = &lostSampleCounter{counts: make(map[string]uint64)} + +// lostSampleCounter accumulates the number of perf event samples the kernel dropped (because the +// per-CPU buffer was full and userspace did not drain it in time), keyed by the perf map name. +type lostSampleCounter struct { + mutex sync.Mutex + counts map[string]uint64 + loggerStart sync.Once +} + +// recordLostSamples accumulates dropped samples for a perf map and lazily starts the periodic +// aggregated logger on first loss. +func recordLostSamples(mapName string, count uint64) { + lostSamples.mutex.Lock() + lostSamples.counts[mapName] += count + lostSamples.mutex.Unlock() + + lostSamples.loggerStart.Do(func() { + go lostSamples.logPeriodically() + }) +} + +// drain atomically returns the samples accumulated since the last drain and resets the counter, so +// each reporting interval directly reads the newly-dropped counts without tracking previous totals, +// and the map does not grow unbounded. +func (c *lostSampleCounter) drain() map[string]uint64 { + c.mutex.Lock() + defer c.mutex.Unlock() + if len(c.counts) == 0 { + return nil + } + drained := c.counts + c.counts = make(map[string]uint64) + return drained +} + +func (c *lostSampleCounter) logPeriodically() { + ticker := time.NewTicker(lostSampleLogInterval) + defer ticker.Stop() + for range ticker.C { + for name, dropped := range c.drain() { + log.Warnf("perf event queue(%s) dropped %d samples in the last %s, "+ + "consider increasing the per_cpu_buffer or parse_parallels", name, dropped, lostSampleLogInterval) + } + } +} diff --git a/pkg/tools/btf/lostsamples_test.go b/pkg/tools/btf/lostsamples_test.go new file mode 100644 index 00000000..35028d1f --- /dev/null +++ b/pkg/tools/btf/lostsamples_test.go @@ -0,0 +1,37 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package btf + +import "testing" + +func TestLostSampleCounterDrain(t *testing.T) { + c := &lostSampleCounter{counts: make(map[string]uint64)} + c.counts["q1"] += 3 + c.counts["q1"] += 2 + c.counts["q2"] += 7 + + drained := c.drain() + if drained["q1"] != 5 || drained["q2"] != 7 { + t.Fatalf("unexpected drained counts: %v", drained) + } + + // after a drain the counter must be reset, so a subsequent drain with no new samples is empty + if again := c.drain(); again != nil { + t.Fatalf("expected nil after reset, got: %v", again) + } +} diff --git a/pkg/tools/process/process.go b/pkg/tools/process/process.go index 87930e21..31dac896 100644 --- a/pkg/tools/process/process.go +++ b/pkg/tools/process/process.go @@ -61,26 +61,40 @@ func KernelFileProfilingStat() (*profiling.Info, error) { return kernelFinder.Analyze(profiling.KernelProcSymbolFilePath) } -// ProfilingStat is validating the exe file could be profiling and get info -func ProfilingStat(pid int32, exePath string) (*profiling.Info, error) { +// SupportProfiling reports whether the executable can be profiled (not excluded and has a usable +// symbol table) WITHOUT retaining the parsed symbol/module data in memory. Use this for the +// discovery-time "support_ebpf_profiling" flag: the heavy symbol data is only needed while an +// actual profiling task runs and is built on demand by ProfilingStat. +func SupportProfiling(exePath string) (bool, error) { stat, err := os.Stat(exePath) if err != nil { - return nil, fmt.Errorf("check file error: %v", err) + return false, fmt.Errorf("check file error: %v", err) } for _, notSupport := range NotSupportProfilingExe { if strings.HasPrefix(stat.Name(), notSupport) { - return nil, fmt.Errorf("not support %s language profiling", notSupport) + return false, fmt.Errorf("not support %s language profiling", notSupport) } } - context := newAnalyzeContext() - // the executable file must have the symbols - symbols, err := context.GetFinder(exePath).AnalyzeSymbols(exePath) - if err != nil || len(symbols) == 0 { - return nil, fmt.Errorf("could not found any symbol in the execute file: %s, error: %v", exePath, err) + // the executable file must have the symbols; the parsed symbols are discarded here so they + // are not retained in memory for every discovered process. + symbols, err := newAnalyzeContext().GetFinder(exePath).AnalyzeSymbols(exePath) + if err != nil { + return false, fmt.Errorf("could not analyze symbols in the executable file %s: %w", exePath, err) + } + if len(symbols) == 0 { + return false, fmt.Errorf("could not find any symbol in the executable file: %s", exePath) + } + return true, nil +} + +// ProfilingStat is validating the exe file could be profiling and get info +func ProfilingStat(pid int32, exePath string) (*profiling.Info, error) { + if support, err := SupportProfiling(exePath); !support { + return nil, err } - return analyzeProfilingInfo(context, pid) + return analyzeProfilingInfo(newAnalyzeContext(), pid) } // Modules Read the profiling info of the process, without the symbol check