Add uptime & rss self-observability metrics, and fix cpu time to work…

… on non-Linux OSs (#1549)

internal/collector/telemetry/process_telemetry.go

@@ -20,17 +20,30 @@ import (
  "runtime"
  "time"
 
- "github.com/prometheus/procfs"
+ "github.com/shirou/gopsutil/process"
  "go.opencensus.io/stats"
  "go.opencensus.io/stats/view"
 )
 
 // ProcessMetricsViews is a struct that contains views related to process metrics (cpu, mem, etc)
 type ProcessMetricsViews struct {
+ prevTimeUnixNano int64
  ballastSizeBytes uint64
  views []*view.View
  done chan struct{}
- proc *procfs.Proc
+ proc *process.Process
+}
+
+var mUptime = stats.Float64(
+ "process/uptime",
+ "Uptime of the process",
+ stats.UnitSeconds)
+var viewProcessUptime = &view.View{
+ Name: mUptime.Name(),
+ Description: mUptime.Description(),
+ Measure: mUptime,
+ Aggregation: view.Sum(),
+ TagKeys: nil,
 }
 
 var mRuntimeAllocMem = stats.Int64(
@@ -81,23 +94,37 @@ var viewCPUSeconds = &view.View{
  TagKeys: nil,
 }
 
+var mRSSMemory = stats.Int64(
+ "process/memory/rss",
+ "Total physical memory (resident set size)",
+ stats.UnitDimensionless)
+var viewRSSMemory = &view.View{
+ Name: mRSSMemory.Name(),
+ Description: mRSSMemory.Description(),
+ Measure: mRSSMemory,
+ Aggregation: view.LastValue(),
+ TagKeys: nil,
+}
+
 // NewProcessMetricsViews creates a new set of ProcessMetrics (mem, cpu) that can be used to measure
 // basic information about this process.
-func NewProcessMetricsViews(ballastSizeBytes uint64) *ProcessMetricsViews {
+func NewProcessMetricsViews(ballastSizeBytes uint64) (*ProcessMetricsViews, error) {
  pmv := &ProcessMetricsViews{
+ prevTimeUnixNano: time.Now().UnixNano(),
  ballastSizeBytes: ballastSizeBytes,
- views: []*view.View{viewAllocMem, viewTotalAllocMem, viewSysMem, viewCPUSeconds},
+ views: []*view.View{viewProcessUptime, viewAllocMem, viewTotalAllocMem, viewSysMem, viewCPUSeconds, viewRSSMemory},
  done: make(chan struct{}),
  }
 
- // procfs.Proc is not available on windows and expected to fail.
  pid := os.Getpid()
- proc, err := procfs.NewProc(pid)
- if err == nil {
- pmv.proc = &proc
+
+ var err error
+ pmv.proc, err = process.NewProcess(int32(pid))
+ if err != nil {
+ return nil, err
  }
 
- return pmv
+ return pmv, nil
 }
 
 // StartCollection starts a ticker'd goroutine that will update the PMV measurements every 5 seconds
@@ -127,15 +154,22 @@ func (pmv *ProcessMetricsViews) StopCollection() {
 }
 
 func (pmv *ProcessMetricsViews) updateViews() {
+ now := time.Now().UnixNano()
+ stats.Record(context.Background(), mUptime.M(float64(now-pmv.prevTimeUnixNano)/1e9))
+ pmv.prevTimeUnixNano = now
+
  ms := &runtime.MemStats{}
  pmv.readMemStats(ms)
  stats.Record(context.Background(), mRuntimeAllocMem.M(int64(ms.Alloc)))
  stats.Record(context.Background(), mRuntimeTotalAllocMem.M(int64(ms.TotalAlloc)))
  stats.Record(context.Background(), mRuntimeSysMem.M(int64(ms.Sys)))
 
  if pmv.proc != nil {
- if procStat, err := pmv.proc.Stat(); err == nil {
- stats.Record(context.Background(), mCPUSeconds.M(int64(procStat.CPUTime())))
+ if times, err := pmv.proc.Times(); err == nil {
+ stats.Record(context.Background(), mCPUSeconds.M(int64(times.Total())))
+ }
+ if mem, err := pmv.proc.MemoryInfo(); err == nil {
+ stats.Record(context.Background(), mRSSMemory.M(int64(mem.RSS)))
  }
  }
 }

internal/collector/telemetry/process_telemetry_test.go

@@ -15,7 +15,6 @@
 package telemetry
 
 import (
- "runtime"
  "testing"
  "time"
 
@@ -27,17 +26,20 @@ import (
 func TestProcessTelemetry(t *testing.T) {
  const ballastSizeBytes uint64 = 0
 
- pmv := NewProcessMetricsViews(ballastSizeBytes)
+ pmv, err := NewProcessMetricsViews(ballastSizeBytes)
+ require.NoError(t, err)
  assert.NotNil(t, pmv)
 
  expectedViews := []string{
  // Changing a metric name is a breaking change.
  // Adding new metrics is ok as long it follows the conventions described at
  // https://pkg.go.dev/go.opentelemetry.io/collector/obsreport?tab=doc#hdr-Naming_Convention_for_New_Metrics
+ "process/uptime",
  "process/runtime/heap_alloc_bytes",
  "process/runtime/total_alloc_bytes",
  "process/runtime/total_sys_memory_bytes",
  "process/cpu_seconds",
+ "process/memory/rss",
  }
  processViews := pmv.Views()
  assert.Len(t, processViews, len(expectedViews))
@@ -50,26 +52,26 @@ func TestProcessTelemetry(t *testing.T) {
  <-time.After(200 * time.Millisecond)
 
  for _, viewName := range expectedViews {
- if (runtime.GOOS == "windows" || runtime.GOOS == "darwin") && viewName == "process/cpu_seconds" {
- // "process/cpu_seconds" is not supported on windows or darwin because there is
- // no procfs which is used for reading that metric.
- continue
- }
-
  rows, err := view.RetrieveData(viewName)
  require.NoError(t, err, viewName)
 
  require.Len(t, rows, 1, viewName)
  row := rows[0]
  assert.Len(t, row.Tags, 0)
 
- lastValue := row.Data.(*view.LastValueData)
- if viewName == "process/cpu_seconds" {
+ var value float64
+ if viewName == "process/uptime" {
+ value = row.Data.(*view.SumData).Value
+ } else {
+ value = row.Data.(*view.LastValueData).Value
+ }
+
+ if viewName == "process/uptime" || viewName == "process/cpu_seconds" {
  // This likely will still be zero when running the test.
- assert.True(t, lastValue.Value >= 0, viewName)
+ assert.True(t, value >= 0, viewName)
  continue
  }
 
- assert.True(t, lastValue.Value > 0, viewName)
+ assert.True(t, value > 0, viewName)
  }
 }

service/telemetry.go

@@ -61,14 +61,18 @@ func (tel *appTelemetry) init(asyncErrorChannel chan<- error, ballastSizeBytes u
  return nil
  }
 
+ processMetricsViews, err := telemetry.NewProcessMetricsViews(ballastSizeBytes)
+ if err != nil {
+ return err
+ }
+
  var views []*view.View
  views = append(views, obsreport.Configure(telemetry.UseLegacyMetrics(), telemetry.UseNewMetrics())...)
  views = append(views, processor.MetricViews(level)...)
  views = append(views, queuedprocessor.MetricViews(level)...)
  views = append(views, batchprocessor.MetricViews(level)...)
  views = append(views, tailsamplingprocessor.SamplingProcessorMetricViews(level)...)
  views = append(views, kafkareceiver.MetricViews()...)
- processMetricsViews := telemetry.NewProcessMetricsViews(ballastSizeBytes)
  views = append(views, processMetricsViews.Views()...)
  views = append(views, fluentobserv.Views(level)...)
  tel.views = views