[system/process] add support for mutlierr (#172)

## What does this PR do?

- Previously, we weren't passing errors to the caller while monitoring
set of processes.
- With the recent introduction of the status reporter for metricsets, it
is impossible to change the status to degraded if such errors are not
passed to the caller.
- Fix this by passing errors to the caller. We also populate the process
related information to our best-effort.

## Checklist

- [x] My code follows the style guidelines of this project
- [x] I have commented my code, particularly in hard-to-understand areas
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] I have added an entry in `CHANGELOG.md`

## Manual testing and general information

- See elastic/beats#40400 for testing it on
`metricbeat`
**NOTE**: 
   - **Only applicable if you're using `system/process` module**
- Non-fatal errors are only received when you have insufficient
privileges.

Steps:
   - While receiving any error, test for nature of error
   - call `errors.Is(err, NonFatalErr{}))` on received error
- If true, error is non-fatal and you can proceed further (metrics will
be partially available, most probably insufficient privileges).
      - Else, log the error and stop execution (metrics will be empty)

Genreal info related to the changes in this PR:
- While getting process related information, you might also receive a
non-nil error.
   - Such errors come in two flavours:
       - Fatal errors: 
- This indicates that the error was fatal (for eg. `no process found`,)
- Caller should stop further execution if they receive fatal errors
       - Non-fatal errors: 
- This indicates that the error was fatal (for eg. `not enough
privileges`)
          - It means that metrics are partially filled.
- Further execution can be continued if non-fatal errors are encountered

- Closes
#164

metric/system/process/helper_test.go

@@ -0,0 +1,76 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. 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 process
+
+import (
+	"errors"
+	"fmt"
+	"syscall"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestErrors(t *testing.T) {
+	cases := []struct {
+		name  string
+		check func(t *testing.T)
+	}{
+		{
+			name: "non fatal error",
+			check: func(t *testing.T) {
+				err := fmt.Errorf("Faced non-fatal error: %w", NonFatalErr{Err: syscall.EPERM})
+				require.True(t, isNonFatal(err), "Should be a non fatal error")
+			},
+		},
+		{
+			name: "non fatal error - unwrapped",
+			check: func(t *testing.T) {
+				err := fmt.Errorf("Faced non-fatal error: %w", syscall.EPERM)
+				require.True(t, isNonFatal(err), "Should be a non fatal error")
+			},
+		},
+		{
+			name: "non fatal error - hierarchy",
+			check: func(t *testing.T) {
+				err := fmt.Errorf("Faced non-fatal error: %w", syscall.EPERM)
+				err2 := errors.Join(toNonFatal(err))
+				require.True(t, isNonFatal(err2), "Should be a non fatal error")
+			},
+		},
+		{
+			name: "fatal error",
+			check: func(t *testing.T) {
+				err := fmt.Errorf("Faced fatal error: %w", errors.New("FATAL"))
+				err = toNonFatal(err) // shouldn't have any effect as it's a fatal error
+				require.Falsef(t, isNonFatal(err), "Should be a fatal error")
+			},
+		},
+		{
+			name: "fatal error - hierarchy",
+			check: func(t *testing.T) {
+				err := fmt.Errorf("Faced fatal error: %w", errors.New("FATAL"))
+				err2 := errors.Join(err)
+				require.Falsef(t, isNonFatal(err2), "Should be a fatal error")
+			},
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.name, c.check)
+	}
+}

metric/system/process/helpers.go

@@ -98,3 +98,40 @@ func GetProcCPUPercentage(s0, s1 ProcState) ProcState {
 	return s1
 
 }
+
+// NonFatalErr indicates an error occurred during metrics
+// collection, however the metrics already
+// gathered and returned are still valid.
+// This error can be safely ignored, this will result
+// in having partial metrics for a process rather than
+// no metrics at all.
+//
+// It was introduced to allow for partial metrics collection
+// on privileged process on Windows.
+type NonFatalErr struct {
+	Err error
+}
+
+func (c NonFatalErr) Error() string {
+	return "Not enough privileges to fetch information: " + c.Err.Error()
+}
+
+func (c NonFatalErr) Is(other error) bool {
+	_, is := other.(NonFatalErr)
+	return is
+}
+
+func (c NonFatalErr) Unwrap() error {
+	return c.Err
+}
+
+// Wraps a NonFatalError around a generic error, if given error is non-fatal in nature
+func toNonFatal(err error) error {
+	if err == nil {
+		return nil
+	}
+	if !isNonFatal(err) {
+		return err
+	}
+	return NonFatalErr{Err: err}
+}

metric/system/process/helpers_others.go

@@ -0,0 +1,35 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. 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.
+
+//go:build !windows
+
+package process
+
+import (
+	"errors"
+	"syscall"
+)
+
+func isNonFatal(err error) bool {
+	if err == nil {
+		return true
+	}
+	return (errors.Is(err, syscall.EACCES) ||
+		errors.Is(err, syscall.EPERM) ||
+		errors.Is(err, syscall.EINVAL) ||
+		errors.Is(err, NonFatalErr{}))
+}

metric/system/process/helpers_windows.go

@@ -0,0 +1,37 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. 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.
+
+//go:build windows
+
+package process
+
+import (
+	"errors"
+	"syscall"
+
+	"golang.org/x/sys/windows"
+)
+
+func isNonFatal(err error) bool {
+	if err == nil {
+		return true
+	}
+	return errors.Is(err, windows.ERROR_ACCESS_DENIED) ||
+		errors.Is(err, syscall.EPERM) ||
+		errors.Is(err, syscall.EINVAL) ||
+		errors.Is(err, windows.ERROR_INVALID_PARAMETER) || errors.Is(err, NonFatalErr{})
+}

metric/system/process/process.go

@@ -25,6 +25,7 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+	"syscall"
 	"time"
 
 	psutil "github.com/shirou/gopsutil/v3/process"
@@ -54,11 +55,11 @@ func ListStates(hostfs resolve.Resolver) ([]ProcState, error) {
 
 	// actually fetch the PIDs from the OS-specific code
 	_, plist, err := init.FetchPids()
-	if err != nil {
+	if err != nil && !isNonFatal(err) {
 		return nil, fmt.Errorf("error gathering PIDs: %w", err)
 	}
 
-	return plist, nil
+	return plist, toNonFatal(err)
 }
 
 // GetPIDState returns the state of a given PID
@@ -90,10 +91,10 @@ func (procStats *Stats) Get() ([]mapstr.M, []mapstr.M, error) {
 	}
 
 	// actually fetch the PIDs from the OS-specific code
-	pidMap, plist, err := procStats.FetchPids()
+	pidMap, plist, wrappedErr := procStats.FetchPids()
 
-	if err != nil {
-		return nil, nil, fmt.Errorf("error gathering PIDs: %w", err)
+	if wrappedErr != nil && !isNonFatal(wrappedErr) {
+		return nil, nil, fmt.Errorf("error gathering PIDs: %w", wrappedErr)
 	}
 	// We use this to track processes over time.
 	procStats.ProcsMap.SetMap(pidMap)
@@ -133,13 +134,13 @@ func (procStats *Stats) Get() ([]mapstr.M, []mapstr.M, error) {
 		rootEvents = append(rootEvents, rootMap)
 	}
 
-	return procs, rootEvents, nil
+	return procs, rootEvents, toNonFatal(wrappedErr)
 }
 
 // GetOne fetches process data for a given PID if its name matches the regexes provided from the host.
 func (procStats *Stats) GetOne(pid int) (mapstr.M, error) {
 	pidStat, _, err := procStats.pidFill(pid, false)
-	if err != nil {
+	if err != nil && !isNonFatal(err) {
 		return nil, fmt.Errorf("error fetching PID %d: %w", pid, err)
 	}
 
@@ -151,9 +152,9 @@ func (procStats *Stats) GetOne(pid int) (mapstr.M, error) {
 // GetOneRootEvent is the same as `GetOne()` but it returns an
 // event formatted as expected by ECS
 func (procStats *Stats) GetOneRootEvent(pid int) (mapstr.M, mapstr.M, error) {
-	pidStat, _, err := procStats.pidFill(pid, false)
-	if err != nil {
-		return nil, nil, fmt.Errorf("error fetching PID %d: %w", pid, err)
+	pidStat, _, wrappedErr := procStats.pidFill(pid, false)
+	if wrappedErr != nil && !isNonFatal(wrappedErr) {
+		return nil, nil, fmt.Errorf("error fetching PID %d: %w", pid, wrappedErr)
 	}
 
 	procStats.ProcsMap.SetPid(pid, pidStat)
@@ -165,7 +166,7 @@ func (procStats *Stats) GetOneRootEvent(pid int) (mapstr.M, mapstr.M, error) {
 
 	rootMap := processRootEvent(&pidStat)
 
-	return procMap, rootMap, err
+	return procMap, rootMap, toNonFatal(wrappedErr)
 }
 
 // GetSelf gets process info for the beat itself
@@ -180,56 +181,41 @@ func (procStats *Stats) GetSelf() (ProcState, error) {
 	}
 
 	pidStat, _, err := procStats.pidFill(self, false)
-	if err != nil {
+	if err != nil && !isNonFatal(err) {
 		return ProcState{}, fmt.Errorf("error fetching PID %d: %w", self, err)
 	}
 
 	procStats.ProcsMap.SetPid(self, pidStat)
 
-	return pidStat, nil
+	return pidStat, toNonFatal(err)
 }
 
 // pidIter wraps a few lines of generic code that all OS-specific FetchPids() functions must call.
 // this also handles the process of adding to the maps/lists in order to limit the code duplication in all the OS implementations
-func (procStats *Stats) pidIter(pid int, procMap ProcsMap, proclist []ProcState) (ProcsMap, []ProcState) {
+func (procStats *Stats) pidIter(pid int, procMap ProcsMap, proclist []ProcState) (ProcsMap, []ProcState, error) {
 	status, saved, err := procStats.pidFill(pid, true)
+	var nonFatalErr error
 	if err != nil {
 		if !errors.Is(err, NonFatalErr{}) {
 			procStats.logger.Debugf("Error fetching PID info for %d, skipping: %s", pid, err)
-			return procMap, proclist
+			// While monitoring a set of processes, some processes might get killed after we get all the PIDs
+			// So, there's no need to capture "process not found" error.
+			if errors.Is(err, syscall.ESRCH) {
+				return procMap, proclist, nil
+			}
+			return procMap, proclist, err
 		}
-		procStats.logger.Debugf("Non fatal error fetching PID some info for %d, metrics are valid, but partial: %s", pid, err)
+		nonFatalErr = fmt.Errorf("non fatal error fetching PID some info for %d, metrics are valid, but partial: %w", pid, err)
+		procStats.logger.Debugf(err.Error())
 	}
 	if !saved {
 		procStats.logger.Debugf("Process name does not match the provided regex; PID=%d; name=%s", pid, status.Name)
-		return procMap, proclist
+		return procMap, proclist, nonFatalErr
 	}
 	procMap[pid] = status
 	proclist = append(proclist, status)
 
-	return procMap, proclist
-}
-
-// NonFatalErr is returned when there was an error
-// collecting metrics, however the metrics already
-// gathered and returned are still valid.
-// This error can be safely ignored, this will result
-// in having partial metrics for a process rather than
-// no metrics at all.
-//
-// It was introduced to allow for partial metrics collection
-// on privileged process on Windows.
-type NonFatalErr struct {
-	Err error
-}
-
-func (c NonFatalErr) Error() string {
-	return "Not enough privileges to fetch information: " + c.Err.Error()
-}
-
-func (c NonFatalErr) Is(other error) bool {
-	_, is := other.(NonFatalErr)
-	return is
+	return procMap, proclist, nonFatalErr
 }
 
 // pidFill is an entrypoint used by OS-specific code to fill out a pid.
@@ -238,7 +224,7 @@ func (c NonFatalErr) Is(other error) bool {
 // The second return value will only be false if an event has been filtered out.
 func (procStats *Stats) pidFill(pid int, filter bool) (ProcState, bool, error) {
 	// Fetch proc state so we can get the name for filtering based on user's filter.
-
+	var wrappedErr error
 	// OS-specific entrypoint, get basic info so we can at least run matchProcess
 	status, err := GetInfoForPid(procStats.Hostfs, pid)
 	if err != nil {
@@ -265,7 +251,8 @@ func (procStats *Stats) pidFill(pid int, filter bool) (ProcState, bool, error) {
 		if !errors.Is(err, NonFatalErr{}) {
 			return status, true, fmt.Errorf("FillPidMetrics: %w", err)
 		}
-		procStats.logger.Debugf("Non-fatal error fetching PID metrics for %d, metrics are valid, but partial: %s", pid, err)
+		wrappedErr = errors.Join(wrappedErr, fmt.Errorf("non-fatal error fetching PID metrics for %d, metrics are valid, but partial: %w", pid, err))
+		procStats.logger.Debugf(wrappedErr.Error())
 	}
 
 	if status.CPU.Total.Ticks.Exists() {
@@ -320,7 +307,7 @@ func (procStats *Stats) pidFill(pid int, filter bool) (ProcState, bool, error) {
 		}
 	}
 
-	return status, true, nil
+	return status, true, wrappedErr
 }
 
 // cacheCmdLine fills out Env and arg metrics from any stored previous metrics for the pid

metric/system/process/process_aix.go

@@ -46,20 +46,22 @@ func (procStats *Stats) FetchPids() (ProcsMap, []ProcState, error) {
 	pid := C.pid_t(0)
 
 	procMap := make(ProcsMap, 0)
+	var wrappedErr err
 	var plist []ProcState
 	for {
 		// getprocs first argument is a void*
 		num, err := C.getprocs(unsafe.Pointer(&info), C.sizeof_struct_procsinfo64, nil, 0, &pid, 1)
 		if err != nil {
 			return nil, nil, fmt.Errorf("error fetching PIDs: %w", err)
 		}
-		procMap, plist = procStats.pidIter(int(info.pi_pid), procMap, plist)
+		procMap, plist, err = procStats.pidIter(int(pid), procMap, plist)
+		wrappedErr = errors.Join(wrappedErr, err)
 
 		if num == 0 {
 			break
 		}
 	}
-	return procMap, plist, nil
+	return procMap, plist, toNonFatal(wrappedErr)
 }
 
 // GetInfoForPid returns basic info for the process