diff --git a/plugins/inputs/cpu/cpu.go b/plugins/inputs/cpu/cpu.go index 51d4a5423055c..30e98361f9578 100644 --- a/plugins/inputs/cpu/cpu.go +++ b/plugins/inputs/cpu/cpu.go @@ -3,7 +3,7 @@ package cpu import ( _ "embed" - "errors" + "encoding/json" "fmt" "time" @@ -19,7 +19,7 @@ var sampleConfig string type CPU struct { ps psutil.PS - lastStats map[string]cpu.TimesStat + lastStats map[string]stats cpuInfo map[string]cpu.InfoStat coreID bool physicalID bool @@ -47,6 +47,7 @@ func (c *CPU) Init() error { c.cpuInfo = make(map[string]cpu.InfoStat) for _, ci := range cpuInfo { c.cpuInfo[fmt.Sprintf("cpu%d", ci.CPU)] = ci + c.Log.Tracef("CPU #%d: %+v", ci.CPU, ci) } } else { c.Log.Warnf("Failed to gather info about CPUs: %s", err) @@ -57,44 +58,28 @@ func (c *CPU) Init() error { } func (c *CPU) Gather(acc telegraf.Accumulator) error { - times, err := c.ps.CPUTimes(c.PerCPU, c.TotalCPU) + statistics, err := cpuTimes(c.PerCPU, c.TotalCPU) if err != nil { return fmt.Errorf("error getting CPU info: %w", err) } now := time.Now() - for _, cts := range times { + var report bool + for _, s := range statistics { tags := map[string]string{ - "cpu": cts.CPU, + "cpu": s.CPU, } if c.coreID { - tags["core_id"] = c.cpuInfo[cts.CPU].CoreID + tags["core_id"] = c.cpuInfo[s.CPU].CoreID } if c.physicalID { - tags["physical_id"] = c.cpuInfo[cts.CPU].PhysicalID + tags["physical_id"] = c.cpuInfo[s.CPU].PhysicalID } - total := totalCPUTime(cts) - active := activeCPUTime(cts) - + // Add cpu time metrics if c.CollectCPUTime { - // Add cpu time metrics - fieldsC := map[string]interface{}{ - "time_user": cts.User, - "time_system": cts.System, - "time_idle": cts.Idle, - "time_nice": cts.Nice, - "time_iowait": cts.Iowait, - "time_irq": cts.Irq, - "time_softirq": cts.Softirq, - "time_steal": cts.Steal, - "time_guest": cts.Guest, - "time_guest_nice": cts.GuestNice, - } - if c.ReportActive { - fieldsC["time_active"] = activeCPUTime(cts) - } - acc.AddCounter("cpu", fieldsC, tags, now) + fields := s.cycleFields(c.ReportActive) + acc.AddCounter("cpu", fields, tags, now) } // Add in percentage @@ -103,57 +88,63 @@ func (c *CPU) Gather(acc telegraf.Accumulator) error { continue } - lastCts, ok := c.lastStats[cts.CPU] + last, ok := c.lastStats[s.CPU] if !ok { continue } - lastTotal := totalCPUTime(lastCts) - lastActive := activeCPUTime(lastCts) - totalDelta := total - lastTotal - if totalDelta < 0 { - err = errors.New("current total CPU time is less than previous total CPU time") + var fields map[string]interface{} + if fields, err = s.percentageFields(last, c.ReportActive); err != nil { + // Break the loop here to update the last statistics cache for the + // next cycle. The error will be propagated. + report = true break - } - - if totalDelta == 0 { + } else if fields == nil { + // We don't have a delta in time so we can't emit statistics here + report = true continue } + acc.AddGauge("cpu", fields, tags, now) - fieldsG := map[string]interface{}{ - "usage_user": 100 * (cts.User - lastCts.User - (cts.Guest - lastCts.Guest)) / totalDelta, - "usage_system": 100 * (cts.System - lastCts.System) / totalDelta, - "usage_idle": 100 * (cts.Idle - lastCts.Idle) / totalDelta, - "usage_nice": 100 * (cts.Nice - lastCts.Nice - (cts.GuestNice - lastCts.GuestNice)) / totalDelta, - "usage_iowait": 100 * (cts.Iowait - lastCts.Iowait) / totalDelta, - "usage_irq": 100 * (cts.Irq - lastCts.Irq) / totalDelta, - "usage_softirq": 100 * (cts.Softirq - lastCts.Softirq) / totalDelta, - "usage_steal": 100 * (cts.Steal - lastCts.Steal) / totalDelta, - "usage_guest": 100 * (cts.Guest - lastCts.Guest) / totalDelta, - "usage_guest_nice": 100 * (cts.GuestNice - lastCts.GuestNice) / totalDelta, - } - if c.ReportActive { - fieldsG["usage_active"] = 100 * (active - lastActive) / totalDelta + // Set a report marker if any of the fields is invalid + report = report || !valid(fields) + } + + // Debug print the raw data for invalid values + if c.Log.Level().Includes(telegraf.Trace) && report { + c.Log.Trace("Detected invalid field values!") + for _, s := range statistics { + curr, err := json.Marshal(s) + if err == nil { + last, ok := c.lastStats[s.CPU] + if ok { + if prev, err := json.Marshal(last); err == nil { + c.Log.Tracef("Invalid raw values %d CPU %q: %s; %s", now.UnixNano(), s.CPU, string(curr), string(prev)) + } + } else { + c.Log.Tracef("Invalid raw values %d CPU %q: %s; {}", now.UnixNano(), s.CPU, string(curr)) + } + } } - acc.AddGauge("cpu", fieldsG, tags, now) } - c.lastStats = make(map[string]cpu.TimesStat) - for _, cts := range times { - c.lastStats[cts.CPU] = cts + // Update the last-value cache for computing the percentages + c.lastStats = make(map[string]stats, len(statistics)) + for _, s := range statistics { + c.lastStats[s.CPU] = s } return err } -func totalCPUTime(t cpu.TimesStat) float64 { - total := t.User + t.System + t.Nice + t.Iowait + t.Irq + t.Softirq + t.Steal + t.Idle - return total -} - -func activeCPUTime(t cpu.TimesStat) float64 { - active := totalCPUTime(t) - t.Idle - return active +func valid(fields map[string]interface{}) bool { + for _, raw := range fields { + v := raw.(float64) + if v < 0.0 || v > 100.0 { + return false + } + } + return true } func init() { diff --git a/plugins/inputs/cpu/cpu_notwindows.go b/plugins/inputs/cpu/cpu_notwindows.go new file mode 100644 index 0000000000000..f38d43166b602 --- /dev/null +++ b/plugins/inputs/cpu/cpu_notwindows.go @@ -0,0 +1,88 @@ +//go:build !windows + +package cpu + +import ( + "errors" + + "github.com/shirou/gopsutil/v4/cpu" +) + +type stats cpu.TimesStat + +func cpuTimes(perCPU, totalCPU bool) ([]stats, error) { + var cpuTimes []stats + if perCPU { + perCPUTimes, err := cpu.Times(true) + if err != nil { + return nil, err + } + for _, s := range perCPUTimes { + cpuTimes = append(cpuTimes, stats(s)) + } + } + if totalCPU { + totalCPUTimes, err := cpu.Times(false) + if err != nil { + return nil, err + } + for _, s := range totalCPUTimes { + cpuTimes = append(cpuTimes, stats(s)) + } + } + return cpuTimes, nil +} + +func (s *stats) cycleFields(active bool) map[string]interface{} { + fields := map[string]interface{}{ + "time_user": s.User, + "time_system": s.System, + "time_idle": s.Idle, + "time_nice": s.Nice, + "time_iowait": s.Iowait, + "time_irq": s.Irq, + "time_softirq": s.Softirq, + "time_steal": s.Steal, + "time_guest": s.Guest, + "time_guest_nice": s.GuestNice, + } + if active { + fields["time_active"] = s.User + s.System + s.Nice + s.Iowait + s.Irq + s.Softirq + s.Steal + } + return fields +} + +func (s *stats) percentageFields(last stats, active bool) (map[string]interface{}, error) { + total := s.User + s.System + s.Nice + s.Iowait + s.Irq + s.Softirq + s.Steal + s.Idle + lastTotal := last.User + last.System + last.Nice + last.Iowait + last.Irq + last.Softirq + last.Steal + last.Idle + + if total < lastTotal { + return nil, errors.New("current total CPU time is less than previous total CPU time") + } + + if total == lastTotal { + return nil, nil + } + + dT := total - lastTotal + + fields := map[string]interface{}{ + "usage_user": 100.0 * (s.User - last.User - (s.Guest - last.Guest)) / dT, + "usage_system": 100.0 * (s.System - last.System) / dT, + "usage_idle": 100.0 * (s.Idle - last.Idle) / dT, + "usage_nice": 100.0 * (s.Nice - last.Nice - (s.GuestNice - last.GuestNice)) / dT, + "usage_iowait": 100.0 * (s.Iowait - last.Iowait) / dT, + "usage_irq": 100.0 * (s.Irq - last.Irq) / dT, + "usage_softirq": 100.0 * (s.Softirq - last.Softirq) / dT, + "usage_steal": 100.0 * (s.Steal - last.Steal) / dT, + "usage_guest": 100.0 * (s.Guest - last.Guest) / dT, + "usage_guest_nice": 100.0 * (s.GuestNice - last.GuestNice) / dT, + } + if active { + dActive := s.User + s.System + s.Nice + s.Iowait + s.Irq + s.Softirq + s.Steal + dActive -= last.User + last.System + last.Nice + last.Iowait + last.Irq + last.Softirq + last.Steal + fields["usage_active"] = 100.0 * dActive / dT + } + + return fields, nil +} diff --git a/plugins/inputs/cpu/cpu_windows.go b/plugins/inputs/cpu/cpu_windows.go new file mode 100644 index 0000000000000..64266b1c1ddb2 --- /dev/null +++ b/plugins/inputs/cpu/cpu_windows.go @@ -0,0 +1,245 @@ +//go:build windows + +package cpu + +import ( + "errors" + "fmt" + "unsafe" + + "golang.org/x/sys/windows" +) + +var hasGroups = procNtQuerySystemInformationEx.Find() == nil + +type stats struct { + CPU string + User uint64 + System uint64 + Idle uint64 + Irq uint64 +} + +func cpuTimes(perCPU, totalCPU bool) ([]stats, error) { + if !perCPU && !totalCPU { + return nil, nil + } + + // Windows 11 and later splits more than 64 cores into processor groups. + // Most API calls will only return processors of the group the current thread + // is running on which leads to issues when Windows is switching the processor + // group for this plugin as counters are then no longer monotonic! + // Therefore, we need to use the (undocumented) NtQuerySystemInformationEx + // function allowing to specify the processor group. However, on older + // Windows versions this function is not available. The following logic uses + // the extended function whenever available and falls back to the standard + // version. + var perfStats []systemProcessorPerformanceInformation + var err error + if hasGroups { + perfStats, err = queryWithGroups() + } else { + perfStats, err = queryWithoutGroups() + } + if err != nil { + return nil, err + } + + var elements int + if perCPU { + elements += len(perfStats) + } + if totalCPU { + elements++ + } + + // Accumulate the total-CPU stats from the individual ones as the gopsutil'result + // cpu.Times() function is based on GetSystemTimes() which only report data + // for the current processor group running the plugin. This leads to + // non-monotonic counters when Windows switches the processor group for the + // plugin. + result := make([]stats, 0, elements) + total := stats{CPU: "cpu-total"} + for core, s := range perfStats { + if perCPU { + result = append(result, stats{ + CPU: fmt.Sprintf("cpu%d", core), + User: uint64(s.UserTime), + System: uint64(s.KernelTime - s.IdleTime), + Idle: uint64(s.IdleTime), + Irq: uint64(s.InterruptTime), + }) + } + + if totalCPU { + total.User += uint64(s.UserTime) + total.System += uint64(s.KernelTime - s.IdleTime) + total.Idle += uint64(s.IdleTime) + total.Irq += uint64(s.InterruptTime) + } + } + + if totalCPU { + result = append(result, total) + } + + return result, nil +} + +func (s *stats) cycleFields(active bool) map[string]interface{} { + fields := map[string]interface{}{ + "time_user": float64(s.User) / 10_000_000.0, // 100ns to seconds + "time_system": float64(s.System) / 10_000_000.0, // 100ns to seconds + "time_idle": float64(s.Idle) / 10_000_000.0, // 100ns to seconds + "time_nice": float64(0), + "time_iowait": float64(0), + "time_irq": float64(s.Irq) / 10_000_000.0, // 100ns to seconds + "time_softirq": float64(0), + "time_steal": float64(0), + "time_guest": float64(0), + "time_guest_nice": float64(0), + } + if active { + fields["time_active"] = float64(s.User+s.System+s.Irq) / 10_000_000.0 // 100ns to seconds + } + return fields +} + +func (s *stats) percentageFields(last stats, active bool) (map[string]interface{}, error) { + total := s.User + s.System + s.Irq + s.Idle + lastTotal := last.User + last.System + last.Irq + last.Idle + + if total < lastTotal { + return nil, errors.New("current total CPU time is less than previous total CPU time") + } + + if total == lastTotal { + return nil, nil + } + + dT := float64(total - lastTotal) + + fields := map[string]interface{}{ + "usage_user": 100.0 * float64(s.User-last.User) / dT, + "usage_system": 100.0 * float64(s.System-last.System) / dT, + "usage_idle": 100.0 * float64(s.Idle-last.Idle) / dT, + "usage_nice": float64(0), + "usage_iowait": float64(0), + "usage_irq": 100.0 * float64(s.Irq-last.Irq) / dT, + "usage_softirq": float64(0), + "usage_steal": float64(0), + "usage_guest": float64(0), + "usage_guest_nice": float64(0), + } + if active { + dActive := s.User + s.System + s.Irq + dActive -= last.User + last.System + last.Irq + fields["usage_active"] = 100.0 * float64(dActive) / dT + } + + return fields, nil +} + +// Windows API section +// The following code was copied from +// https://github.com/shirou/gopsutil/blob/master/cpu/cpu_windows.go +// under the BSD 3-Clause Clear License + +var ( + modKernel32 = windows.NewLazySystemDLL("kernel32.dll") + modNt = windows.NewLazySystemDLL("ntdll.dll") + procGetActiveProcessorGroupCount = modKernel32.NewProc("GetActiveProcessorGroupCount") + procNtQuerySystemInformation = modNt.NewProc("NtQuerySystemInformation") + procNtQuerySystemInformationEx = modNt.NewProc("NtQuerySystemInformationEx") +) + +type systemProcessorPerformanceInformation struct { + IdleTime int64 // idle time in 100ns (this is not a filetime). + KernelTime int64 // kernel time in 100ns. kernel time includes idle time. (this is not a filetime). + UserTime int64 // usertime in 100ns (this is not a filetime). + DpcTime int64 // dpc time in 100ns (this is not a filetime). + InterruptTime int64 // interrupt time in 100ns + InterruptCount uint64 // ULONG needs to be uint64 +} + +const systemProcessorPerformanceInformationClass = 8 + +var systemProcessorPerformanceInfoSize = uint32(unsafe.Sizeof(systemProcessorPerformanceInformation{})) + +// queryWithoutGroups queries SystemProcessorPerformanceInformation using the +// non-Ex NtQuerySystemInformation call. This is the legacy fallback for +// environments where NtQuerySystemInformationEx is not available. +// NOTE: This API call only returns data for the calling thread's processor group! +func queryWithoutGroups() ([]systemProcessorPerformanceInformation, error) { + maxCores := 2056 + buf := make([]systemProcessorPerformanceInformation, maxCores) + bufSize := uintptr(systemProcessorPerformanceInfoSize) * uintptr(maxCores) + + // Invoke windows API; the returned err from the windows dll proc will + // always be non-nil even when successful. See + // https://godoc.org/golang.org/x/sys/windows#LazyProc.Call for more info + var retSize uint32 + //nolint:gosec // Unsafe calls required when using the Windows API + retCode, _, err := procNtQuerySystemInformation.Call( + uintptr(systemProcessorPerformanceInformationClass), // System Information Class -> SystemProcessorPerformanceInformation + uintptr(unsafe.Pointer(&buf[0])), // pointer to first element in result buffer + bufSize, // size of the buffer in memory + uintptr(unsafe.Pointer(&retSize)), // pointer to the size of the returned results the windows proc will set this + ) + if retCode != 0 { + return nil, fmt.Errorf("calling to NtQuerySystemInformation failed: [0x%x] %w", retCode, err) + } + + // Trim results to the number of returned elements + n := retSize / systemProcessorPerformanceInfoSize + return buf[:n], nil +} + +// queryWithGroups queries SystemProcessorPerformanceInformation for every active +// processor group via NtQuerySystemInformationEx and concatenates the results. The +// group index is passed as the InputBuffer per the Ex calling convention documented at +// https://www.geoffchappell.com/studies/windows/km/ntoskrnl/api/ex/sysinfo/queryex.htm +func queryWithGroups() ([]systemProcessorPerformanceInformation, error) { + // Get the number of processor groups. This should be at least one, a zero + // count indicates an error. + r, _, err := procGetActiveProcessorGroupCount.Call() + if r == 0 { + return nil, fmt.Errorf("calling GetActiveProcessorGroupCount failed: %w", err) + } + groups := uint16(r) + + var result []systemProcessorPerformanceInformation + for g := uint16(0); g < groups; g++ { + processors := windows.GetActiveProcessorCount(g) + if processors == 0 { + return nil, fmt.Errorf("calling GetActiveProcessorCount for processor group %d failed", g) + } + // buffer sized exactly for this group's logical CPU count + buf := make([]systemProcessorPerformanceInformation, processors) + bufSize := uintptr(systemProcessorPerformanceInfoSize) * uintptr(processors) + + // InputBuffer is a USHORT (2 bytes) holding the target processor group index. + var retSize uint32 + //nolint:gosec // Unsafe calls required when using the Windows API + retCode, _, err := procNtQuerySystemInformationEx.Call( + systemProcessorPerformanceInformationClass, // System Information Class -> SystemProcessorPerformanceInformation + uintptr(unsafe.Pointer(&g)), // InputBuffer: pointer to USHORT group index + unsafe.Sizeof(g), // InputBufferLength: sizeof(USHORT) being 2 + uintptr(unsafe.Pointer(&buf[0])), // pointer to first element in result buffer + bufSize, // size of the buffer in memory + uintptr(unsafe.Pointer(&retSize)), // pointer to the size of the returned results the windows proc will set this + ) + if retCode != 0 { + return nil, fmt.Errorf("calling NtQuerySystemInformationEx for processor group %d failed: [0x%x]: %w", g, retCode, err) + } + // Guard against a retSize that is not a whole number of entries or exceeds + // the allocated buffer (e.g. CPU hot-add racing with GetActiveProcessorCount). + if retSize%systemProcessorPerformanceInfoSize != 0 || uintptr(retSize) > bufSize { + return nil, fmt.Errorf("calling NtQuerySystemInformationEx for processor group=%d returned unexpected size %d (bufSize=%d)", g, retSize, bufSize) + } + n := retSize / systemProcessorPerformanceInfoSize + result = append(result, buf[:n]...) + } + + return result, nil +}