Refactor of metricbeat process-gathering metrics and system/process

libbeat/cmd/instance/metrics/metrics.go

@@ -21,11 +21,9 @@
 package metrics
 
 import (
-	"fmt"
 	"os"
 	"runtime"
 
-	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/logp"
 	"github.com/elastic/beats/v7/libbeat/metric/system/cgroup"
 	"github.com/elastic/beats/v7/libbeat/metric/system/cpu"
@@ -118,129 +116,45 @@ func reportMemStats(m monitoring.Mode, V monitoring.Visitor) {
 }
 
 func getRSSSize() (uint64, error) {
-	state, err := getBeatProcessState()
+	state, err := beatProcessStats.GetSelf()
 	if err != nil {
 		return 0, err
 	}
 
-	iRss, err := state.GetValue("memory.rss.bytes")
-	if err != nil {
-		return 0, fmt.Errorf("error getting Resident Set Size: %v", err)
-	}
-
-	rss, ok := iRss.(uint64)
-	if !ok {
-		return 0, fmt.Errorf("error converting Resident Set Size to uint64: %v", iRss)
-	}
-	return rss, nil
-}
-
-func getBeatProcessState() (common.MapStr, error) {
-	pid, err := process.GetSelfPid()
-	if err != nil {
-		return nil, fmt.Errorf("error getting PID for self process: %v", err)
-	}
-
-	state, err := beatProcessStats.GetOne(pid)
-	if err != nil {
-		return nil, fmt.Errorf("error retrieving process stats: %v", err)
-	}
-
-	return state, nil
+	return state.Memory.Rss.Bytes.ValueOr(0), nil
 }
 
 func reportBeatCPU(_ monitoring.Mode, V monitoring.Visitor) {
 	V.OnRegistryStart()
 	defer V.OnRegistryFinished()
 
-	totalCPUUsage, cpuTicks, err := getCPUUsage()
+	state, err := beatProcessStats.GetSelf()
 	if err != nil {
 		logp.Err("Error retrieving CPU percentages: %v", err)
 		return
 	}
 
-	userTime, systemTime, err := process.GetOwnResourceUsageTimeInMillis()
-	if err != nil {
-		logp.Err("Error retrieving CPU usage time: %v", err)
-		return
-	}
-
 	monitoring.ReportNamespace(V, "user", func() {
-		monitoring.ReportInt(V, "ticks", int64(cpuTicks.User))
+		monitoring.ReportInt(V, "ticks", int64(state.CPU.User.Ticks.ValueOr(0)))
 		monitoring.ReportNamespace(V, "time", func() {
-			monitoring.ReportInt(V, "ms", userTime)
+			monitoring.ReportInt(V, "ms", int64(state.CPU.User.Ticks.ValueOr(0)))
 		})
 	})
 	monitoring.ReportNamespace(V, "system", func() {
-		monitoring.ReportInt(V, "ticks", int64(cpuTicks.System))
+		monitoring.ReportInt(V, "ticks", int64(state.CPU.System.Ticks.ValueOr(0)))
 		monitoring.ReportNamespace(V, "time", func() {
-			monitoring.ReportInt(V, "ms", systemTime)
+			monitoring.ReportInt(V, "ms", int64(state.CPU.System.Ticks.ValueOr(0)))
 		})
 	})
 	monitoring.ReportNamespace(V, "total", func() {
-		monitoring.ReportFloat(V, "value", totalCPUUsage)
-		monitoring.ReportInt(V, "ticks", int64(cpuTicks.Total))
+		monitoring.ReportFloat(V, "value", state.CPU.Total.Value.ValueOr(0))
+		monitoring.ReportInt(V, "ticks", int64(state.CPU.Total.Ticks.ValueOr(0)))
 		monitoring.ReportNamespace(V, "time", func() {
-			monitoring.ReportInt(V, "ms", userTime+systemTime)
+			monitoring.ReportInt(V, "ms", int64(state.CPU.Total.Ticks.ValueOr(0)))
 		})
 	})
 }
 
-func getCPUUsage() (float64, *process.Ticks, error) {
-	state, err := getBeatProcessState()
-	if err != nil {
-		return 0.0, nil, err
-	}
-
-	iTotalCPUUsage, err := state.GetValue("cpu.total.value")
-	if err != nil {
-		return 0.0, nil, fmt.Errorf("error getting total CPU since start: %v", err)
-	}
-
-	totalCPUUsage, ok := iTotalCPUUsage.(float64)
-	if !ok {
-		return 0.0, nil, fmt.Errorf("error converting value of CPU usage since start to float64: %v", iTotalCPUUsage)
-	}
-
-	iTotalCPUUserTicks, err := state.GetValue("cpu.user.ticks")
-	if err != nil {
-		return 0.0, nil, fmt.Errorf("error getting number of user CPU ticks since start: %v", err)
-	}
-
-	totalCPUUserTicks, ok := iTotalCPUUserTicks.(uint64)
-	if !ok {
-		return 0.0, nil, fmt.Errorf("error converting value of user CPU ticks since start to uint64: %v", iTotalCPUUserTicks)
-	}
-
-	iTotalCPUSystemTicks, err := state.GetValue("cpu.system.ticks")
-	if err != nil {
-		return 0.0, nil, fmt.Errorf("error getting number of system CPU ticks since start: %v", err)
-	}
-
-	totalCPUSystemTicks, ok := iTotalCPUSystemTicks.(uint64)
-	if !ok {
-		return 0.0, nil, fmt.Errorf("error converting value of system CPU ticks since start to uint64: %v", iTotalCPUSystemTicks)
-	}
-
-	iTotalCPUTicks, err := state.GetValue("cpu.total.ticks")
-	if err != nil {
-		return 0.0, nil, fmt.Errorf("error getting total number of CPU ticks since start: %v", err)
-	}
-
-	totalCPUTicks, ok := iTotalCPUTicks.(uint64)
-	if !ok {
-		return 0.0, nil, fmt.Errorf("error converting total value of CPU ticks since start to uint64: %v", iTotalCPUTicks)
-	}
-
-	p := process.Ticks{
-		User:   totalCPUUserTicks,
-		System: totalCPUSystemTicks,
-		Total:  totalCPUTicks,
-	}
-
-	return totalCPUUsage, &p, nil
-}
-
 func reportSystemLoadAverage(_ monitoring.Mode, V monitoring.Visitor) {
 	V.OnRegistryStart()
 	defer V.OnRegistryFinished()
@@ -281,11 +195,7 @@ func reportBeatCgroups(_ monitoring.Mode, V monitoring.Visitor) {
 	V.OnRegistryStart()
 	defer V.OnRegistryFinished()
 
-	pid, err := process.GetSelfPid()
-	if err != nil {
-		logp.Err("error getting PID for self process: %v", err)
-		return
-	}
+	pid := os.Getpid()
 
 	cgroups, err := cgroup.NewReaderOptions(cgroup.ReaderOptions{
 		RootfsMountpoint:         resolve.NewTestResolver("/"),

libbeat/cmd/instance/metrics/metrics_file_descriptors.go

@@ -21,7 +21,7 @@
 package metrics
 
 import (
-	"fmt"
+	"github.com/pkg/errors"
 
 	"github.com/elastic/beats/v7/libbeat/logp"
 	"github.com/elastic/beats/v7/libbeat/monitoring"
@@ -49,40 +49,17 @@ func reportFDUsage(_ monitoring.Mode, V monitoring.Visitor) {
 }
 
 func getFDUsage() (open, hardLimit, softLimit uint64, err error) {
-	state, err := getBeatProcessState()
-	if err != nil {
-		return 0, 0, 0, err
-	}
 
-	iOpen, err := state.GetValue("fd.open")
+	state, err := beatProcessStats.GetSelf()
 	if err != nil {
-		return 0, 0, 0, fmt.Errorf("error getting number of open FD: %v", err)
+		return 0, 0, 0, errors.Wrap(err, "error fetching self process")
 	}
 
-	open, ok := iOpen.(uint64)
-	if !ok {
-		return 0, 0, 0, fmt.Errorf("error converting value of open FDs to uint64: %v", iOpen)
-	}
+	open = state.FD.Open.ValueOr(0)
 
-	iHardLimit, err := state.GetValue("fd.limit.hard")
-	if err != nil {
-		return 0, 0, 0, fmt.Errorf("error getting FD hard limit: %v", err)
-	}
+	hardLimit = state.FD.Limit.Hard.ValueOr(0)
 
-	hardLimit, ok = iHardLimit.(uint64)
-	if !ok {
-		return 0, 0, 0, fmt.Errorf("error converting values of FD hard limit: %v", iHardLimit)
-	}
-
-	iSoftLimit, err := state.GetValue("fd.limit.soft")
-	if err != nil {
-		return 0, 0, 0, fmt.Errorf("error getting FD hard limit: %v", err)
-	}
-
-	softLimit, ok = iSoftLimit.(uint64)
-	if !ok {
-		return 0, 0, 0, fmt.Errorf("error converting values of FD hard limit: %v", iSoftLimit)
-	}
+	softLimit = state.FD.Limit.Soft.ValueOr(0)
 
 	return open, hardLimit, softLimit, nil
 }

libbeat/metric/system/process/helpers.go

@@ -0,0 +1,94 @@
+// 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 (
+	"time"
+
+	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/libbeat/metric/system/numcpu"
+	"github.com/elastic/beats/v7/libbeat/opt"
+)
+
+// unixTimeMsToTime converts a unix time given in milliseconds since Unix epoch
+// to a common.Time value.
+func unixTimeMsToTime(unixTimeMs uint64) string {
+	return common.Time(time.Unix(0, int64(unixTimeMs*1000000))).String()
+}
+
+func stripNullByte(buf []byte) string {
+	return string(buf[0 : len(buf)-1])
+}
+
+func stripNullByteRaw(buf []byte) []byte {
+	return buf[0 : len(buf)-1]
+}
+
+// GetProcMemPercentage returns process memory usage as a percent of total memory usage
+func GetProcMemPercentage(proc ProcState, totalPhyMem uint64) opt.Float {
+	if totalPhyMem == 0 {
+		return opt.NewFloatNone()
+	}
+
+	perc := (float64(proc.Memory.Rss.Bytes.ValueOr(0)) / float64(totalPhyMem))
+
+	return opt.FloatWith(common.Round(perc, 4))
+}
+
+// isProcessInSlice looks up proc in the processes slice and returns if
+// found or not
+func isProcessInSlice(processes []ProcState, proc *ProcState) bool {
+	for _, p := range processes {
+		if p.Pid == proc.Pid {
+			return true
+		}
+	}
+	return false
+}
+
+// GetProcCPUPercentage returns the percentage of total CPU time consumed by
+// the process during the period between the given samples. Two percentages are
+// returned (these must be multiplied by 100). The first is a normalized based
+// on the number of cores such that the value ranges on [0, 1]. The second is
+// not normalized and the value ranges on [0, number_of_cores].
+//
+// Implementation note: The total system CPU time (including idle) is not
+// provided so this method will resort to using the difference in wall-clock
+// time multiplied by the number of cores as the total amount of CPU time
+// available between samples. This could result in incorrect percentages if the
+// wall-clock is adjusted (prior to Go 1.9) or the machine is suspended.
+func GetProcCPUPercentage(s0, s1 ProcState) ProcState {
+	// Skip if we're missing the total ticks
+	if s0.CPU.Total.Ticks.IsZero() || s1.CPU.Total.Ticks.IsZero() {
+		return s1
+	}
+
+	timeDelta := s1.SampleTime.Sub(s0.SampleTime)
+	timeDeltaMillis := timeDelta / time.Millisecond
+	totalCPUDeltaMillis := int64(s1.CPU.Total.Ticks.ValueOr(0) - s0.CPU.Total.Ticks.ValueOr(0))
+
+	pct := float64(totalCPUDeltaMillis) / float64(timeDeltaMillis)
+	normalizedPct := pct / float64(numcpu.NumCPU())
+
+	s1.CPU.Total.Norm.Pct = opt.FloatWith(common.Round(normalizedPct, common.DefaultDecimalPlacesCount))
+	s1.CPU.Total.Pct = opt.FloatWith(common.Round(pct, common.DefaultDecimalPlacesCount))
+	s1.CPU.Total.Value = opt.FloatWith(common.Round(float64(s1.CPU.Total.Ticks.ValueOr(0)), common.DefaultDecimalPlacesCount))
+
+	return s1
+
+}