Skip to content

[receiver/hostmetrics] Cheaper parent PID and number of threads retrieval on Windows #38589

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Show file tree
Hide file tree
Changes from 9 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
# Use this changelog template to create an entry for release notes.

# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
change_type: 'enhancement'

# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
component: hostmetricsreceiver

# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
note: >
Reduced the cost of retrieving number of threads and parent process ID on Windows.
Disable the featuregate `hostmetrics.process.OnWindowsUseNewGetProcesses` to fallback to the previous implementation.

# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
issues: [32947, 38589]

# (Optional) One or more lines of additional information to render under the primary note.
# These lines will be padded with 2 spaces and then inserted directly into the document.
# Use pipe (|) for multiline entries.
subtext:

# If your change doesn't affect end users or the exported elements of any package,
# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
# Optional: The change log or logs in which this entry should be included.
# e.g. '[user]' or '[user, api]'
# Include 'user' if the change is relevant to end users.
# Include 'api' if there is a change to a library API.
# Default: '[user]'
change_logs: [user]
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

//go:build !windows

package processscraper // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/hostmetricsreceiver/internal/scraper/processscraper"

import (
"context"

"github.com/shirou/gopsutil/v4/process"
)

func getGopsutilProcessHandles(ctx context.Context) (processHandles, error) {
processes, err := process.ProcessesWithContext(ctx)
if err != nil {
return nil, err
}
wrapped := make([]wrappedProcessHandle, len(processes))
for i, p := range processes {
wrapped[i] = wrappedProcessHandle{
Process: p,
}
}

return &gopsProcessHandles{handles: wrapped}, nil
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

//go:build windows

package processscraper // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/hostmetricsreceiver/internal/scraper/processscraper"

import (
"context"
"fmt"
"unsafe"

"github.com/shirou/gopsutil/v4/process"
"go.opentelemetry.io/collector/featuregate"
"golang.org/x/sys/windows"
)

var useNewGetProcessHandles = featuregate.GlobalRegistry().MustRegister(
"hostmetrics.process.OnWindowsUseNewGetProcesses",
featuregate.StageBeta,
featuregate.WithRegisterDescription("If disabled, the scraper will use the legacy implementation to retrieve process handles."),
)

func getGopsutilProcessHandles(ctx context.Context) (processHandles, error) {
if !useNewGetProcessHandles.IsEnabled() {
return getGopsutilProcessHandlesLegacy(ctx)
}

snap, err := windows.CreateToolhelp32Snapshot(windows.TH32CS_SNAPPROCESS, 0)
if err != nil {
return nil, fmt.Errorf("could not create snapshot: %w", err)
}
defer func() {
_ = windows.CloseHandle(snap)
}()

var pe32 windows.ProcessEntry32
pe32.Size = uint32(unsafe.Sizeof(pe32))
if err = windows.Process32First(snap, &pe32); err != nil {
return nil, fmt.Errorf("could not get first process: %w", err)
}

wrappedProcesses := make([]wrappedProcessHandle, 0, 64)
for {
select {
case <-ctx.Done():
return nil, ctx.Err()
default:
// Ignoring any errors here to keep same behavior as the legacy implementation
// based on the `process.ProcessesWithContext` from the `gopsutil` package.
p, _ := process.NewProcess(int32(pe32.ProcessID))
if p != nil {
wrappedProcess := wrappedProcessHandle{
Process: p,
parentPid: int32(pe32.ParentProcessID),
initialNumThreads: int32(pe32.Threads),
flags: flagParentPidSet | flagUseInitialNumThreadsOnce,
}
wrappedProcesses = append(wrappedProcesses, wrappedProcess)
}
}

if err = windows.Process32Next(snap, &pe32); err != nil {
break
}
}

return &gopsProcessHandles{handles: wrappedProcesses}, nil
}

func getGopsutilProcessHandlesLegacy(ctx context.Context) (processHandles, error) {
processes, err := process.ProcessesWithContext(ctx)
if err != nil {
return nil, err
}
wrapped := make([]wrappedProcessHandle, len(processes))
for i, p := range processes {
wrapped[i] = wrappedProcessHandle{
Process: p,
}
}

return &gopsProcessHandles{handles: wrapped}, nil
}
Original file line number Diff line number Diff line change
Expand Up @@ -109,18 +109,26 @@ func (p *gopsProcessHandles) Pid(index int) int32 {
}

func (p *gopsProcessHandles) At(index int) processHandle {
return p.handles[index]
return &(p.handles[index])
}

func (p *gopsProcessHandles) Len() int {
return len(p.handles)
}

const (
flagParentPidSet = 1 << 0
flagUseInitialNumThreadsOnce = 1 << 1
)

type wrappedProcessHandle struct {
*process.Process
parentPid int32
initialNumThreads int32
flags uint8 // bitfield to track if fields are set
}

func (p wrappedProcessHandle) CgroupWithContext(ctx context.Context) (string, error) {
func (p *wrappedProcessHandle) CgroupWithContext(ctx context.Context) (string, error) {
pid := p.Process.Pid
statPath := getEnvWithContext(ctx, string(common.HostProcEnvKey), "/proc", strconv.Itoa(int(pid)), "cgroup")
contents, err := os.ReadFile(statPath)
Expand All @@ -131,6 +139,36 @@ func (p wrappedProcessHandle) CgroupWithContext(ctx context.Context) (string, er
return strings.TrimSuffix(string(contents), "\n"), nil
}

func (p *wrappedProcessHandle) PpidWithContext(ctx context.Context) (int32, error) {
if p.flags&flagParentPidSet != 0 {
return p.parentPid, nil
}

parentPid, err := p.Process.PpidWithContext(ctx)
if err != nil {
return 0, err
}

p.parentPid = parentPid
p.flags |= flagParentPidSet
return parentPid, nil
}

func (p *wrappedProcessHandle) NumThreadsWithContext(ctx context.Context) (int32, error) {
if p.flags&flagUseInitialNumThreadsOnce != 0 {
// The number of threads can fluctuate so use the initially cached value only the first time.
p.flags &^= flagUseInitialNumThreadsOnce
return p.initialNumThreads, nil
}

numThreads, err := p.Process.NumThreadsWithContext(ctx)
if err != nil {
return 0, err
}

return numThreads, nil
}

// copied from gopsutil:
// GetEnvWithContext retrieves the environment variable key. If it does not exist it returns the default.
// The context may optionally contain a map superseding os.EnvKey.
Expand All @@ -150,19 +188,6 @@ func getEnvWithContext(ctx context.Context, key string, dfault string, combineWi
return filepath.Join(segments...)
}

func getProcessHandlesInternal(ctx context.Context) (processHandles, error) {
processes, err := process.ProcessesWithContext(ctx)
if err != nil {
return nil, err
}
wrapped := make([]wrappedProcessHandle, len(processes))
for i, p := range processes {
wrapped[i] = wrappedProcessHandle{Process: p}
}

return &gopsProcessHandles{handles: wrapped}, nil
}

func parentPid(ctx context.Context, handle processHandle, pid int32) (int32, error) {
// special case for pid 0 and pid 1 in darwin
if pid == 0 || (pid == 1 && runtime.GOOS == "darwin") {
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,71 @@
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

//go:build windows

package processscraper
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This file could be removed when we remove the featureflag to fallback to the old behavior.


import (
"context"
"testing"

"github.com/stretchr/testify/require"
"go.opentelemetry.io/collector/featuregate"
"go.opentelemetry.io/collector/scraper"
)

func BenchmarkGetProcessMetadata(b *testing.B) {
ctx := context.Background()
config := &Config{
MuteProcessExeError: true,
MuteProcessNameError: true,
MuteProcessAllErrors: true, // Only way to pass the benchmark
}

scraper, err := newProcessScraper(scraper.Settings{}, config)
if err != nil {
b.Fatalf("Failed to create process scraper: %v", err)
}

benchmarks := []struct {
name string
useLegacy bool
parentPidEnabled bool
}{
{
name: "New-ExcludeParentPid",
},
{
name: "Old-ExcludeParentPid",
useLegacy: true,
},
{
name: "New-IncludeParentPid",
parentPidEnabled: true,
},
{
name: "Old-IncludeParentPid",
parentPidEnabled: true,
useLegacy: true,
},
}

for _, bm := range benchmarks {
b.Run(bm.name, func(b *testing.B) {
// Set feature gate value
previousValue := useNewGetProcessHandles.IsEnabled()
require.NoError(b, featuregate.GlobalRegistry().Set(useNewGetProcessHandles.ID(), !bm.useLegacy))
defer func() {
require.NoError(b, featuregate.GlobalRegistry().Set(useNewGetProcessHandles.ID(), previousValue))
}()
scraper.config.MetricsBuilderConfig.ResourceAttributes.ProcessParentPid.Enabled = bm.parentPidEnabled

for i := 0; i < b.N; i++ {
_, err := scraper.getProcessMetadata(ctx)
if err != nil {
b.Fatalf("Failed to get process metadata: %v", err)
}
}
})
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ func newProcessScraper(settings scraper.Settings, cfg *Config) (*processScraper,
settings: settings,
config: cfg,
getProcessCreateTime: processHandle.CreateTimeWithContext,
getProcessHandles: getProcessHandlesInternal,
getProcessHandles: getGopsutilProcessHandles,
scrapeProcessDelay: cfg.ScrapeProcessDelay,
ucals: make(map[int32]*ucal.CPUUtilizationCalculator),
handleCountManager: handlecount.NewManager(),
Expand Down Expand Up @@ -250,7 +250,7 @@ func (s *processScraper) getProcessMetadata(ctx context.Context) ([]*processMeta
}

command, err := getProcessCommand(ctx, handle)
if err != nil {
if err != nil && !s.config.MuteProcessAllErrors {
Copy link
Member

@dmitryax dmitryax Mar 21, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we add muting here but the related change on line 277 still unmuted? Maybe remove this from this PR to fix consistently as a separate bug_fix if needed? Just filing an issue is fine as well

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point that is a leftover from benchmarking...

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

errs.AddPartial(0, fmt.Errorf("error reading command for process %q (pid %v): %w", executable.name, pid, err))
}

Expand All @@ -271,14 +271,17 @@ func (s *processScraper) getProcessMetadata(ctx context.Context) ([]*processMeta
continue
}

parentPid, err := parentPid(ctx, handle, pid)
if err != nil {
errs.AddPartial(0, fmt.Errorf("error reading parent pid for process %q (pid %v): %w", executable.name, pid, err))
parentProcessID := int32(0)
if s.config.MetricsBuilderConfig.ResourceAttributes.ProcessParentPid.Enabled {
parentProcessID, err = parentPid(ctx, handle, pid)
if err != nil {
errs.AddPartial(0, fmt.Errorf("error reading parent pid for process %q (pid %v): %w", executable.name, pid, err))
}
}

md := &processMetadata{
pid: pid,
parentPid: parentPid,
parentPid: parentProcessID,
executable: executable,
command: command,
username: username,
Expand Down