diff --git a/internal/collector/telemetry/process_telemetry.go b/internal/collector/telemetry/process_telemetry.go index fe0240640b1..2ba633d3c6a 100644 --- a/internal/collector/telemetry/process_telemetry.go +++ b/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,6 +154,10 @@ 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))) @@ -134,8 +165,11 @@ func (pmv *ProcessMetricsViews) updateViews() { 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))) } } } diff --git a/internal/collector/telemetry/process_telemetry_test.go b/internal/collector/telemetry/process_telemetry_test.go index 97b15b91640..b28643706f5 100644 --- a/internal/collector/telemetry/process_telemetry_test.go +++ b/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,12 +52,6 @@ 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) @@ -64,7 +60,7 @@ func TestProcessTelemetry(t *testing.T) { assert.Len(t, row.Tags, 0) lastValue := row.Data.(*view.LastValueData) - if viewName == "process/cpu_seconds" { + 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) continue diff --git a/service/telemetry.go b/service/telemetry.go index 6678102744b..c6df9923cc6 100644 --- a/service/telemetry.go +++ b/service/telemetry.go @@ -60,13 +60,17 @@ 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)...) - processMetricsViews := telemetry.NewProcessMetricsViews(ballastSizeBytes) views = append(views, processMetricsViews.Views()...) views = append(views, fluentobserv.Views(level)...) tel.views = views