Custom Distribution Summary Hi-Res (#168)

* Custom Distribution Summary Hi-Res

* Update var name

---------

Co-authored-by: ecbadeaux <everettc1810@gmail.com>

Author: brharrington

lib/collectors/cgroup/src/cgroup.cpp

@@ -131,7 +131,7 @@ void CGroup::CpuProcessingCapacity(const absl::Time& now, const double cpuCount,
     registry_->CreateCounter("cgroup.cpu.processingCapacity").Increment(delta_t * cpuCount);
 }
 
-void CGroup::CpuUtilizationV2(const absl::Time& now, const double cpuCount, const absl::Duration& interval) noexcept
+void CGroup::CpuUtilizationV2(const absl::Time& now, const double cpuCount, const std::unordered_map<std::string, int64_t>& stats, const absl::Duration& interval) noexcept
 {
     static absl::Time last_updated;
     if (last_updated == absl::UnixEpoch())
@@ -151,24 +151,21 @@ void CGroup::CpuUtilizationV2(const absl::Time& now, const double cpuCount, cons
     registry_->CreateGauge("sys.cpu.numProcessors").Set(cpuCount);
     registry_->CreateGauge("titus.cpu.requested").Set(cpuCount);
 
-    std::unordered_map<std::string, int64_t> stats;
-    parse_kv_from_file(path_prefix_, "cpu.stat", &stats);
-
     static auto prev_system_time = static_cast<int64_t>(-1);
     if (prev_system_time >= 0)
     {
-        auto secs = (stats["system_usec"] - prev_system_time) / MICROS;
+        auto secs = (stats.at("system_usec") - prev_system_time) / MICROS;
         registry_->CreateGauge("sys.cpu.utilization", {{"id", "system"}}).Set((secs / avail_cpu_time) * 100);
     }
-    prev_system_time = stats["system_usec"];
+    prev_system_time = stats.at("system_usec");
 
     static auto prev_user_time = static_cast<int64_t>(-1);
     if (prev_user_time >= 0)
     {
-        auto secs = (stats["user_usec"] - prev_user_time) / MICROS;
+        auto secs = (stats.at("user_usec") - prev_user_time) / MICROS;
         registry_->CreateGauge("sys.cpu.utilization", {{"id", "user"}}).Set((secs / avail_cpu_time) * 100);
     }
-    prev_user_time = stats["user_usec"];
+    prev_user_time = stats.at("user_usec");
 }
 
 void CGroup::CpuPeakUtilizationV2(const absl::Time& now, const std::unordered_map<std::string, int64_t>& stats,
@@ -207,7 +204,7 @@ void CGroup::CpuStats(const bool fiveSecondMetricsEnabled, const bool sixtySecon
     if (sixtySecondMetricsEnabled)
     {
         CpuThrottleV2(stats);
-        CpuUtilizationV2(absl::Now(), cpuCount, absl::Seconds(60));
+        CpuUtilizationV2(absl::Now(), cpuCount, stats, absl::Seconds(60));
     }
 
     // Collect 5 second metrics if enabled

lib/collectors/cgroup/src/cgroup.h

@@ -27,7 +27,7 @@ class CGroup
     double GetNumCpu() noexcept;
     void CpuThrottleV2(const std::unordered_map<std::string, int64_t>& stats) noexcept;
     void CpuTimeV2(const std::unordered_map<std::string, int64_t>& stats) noexcept;
-    void CpuUtilizationV2(const absl::Time& now, const double cpuCount, const absl::Duration& interval) noexcept;
+    void CpuUtilizationV2(const absl::Time& now, const double cpuCount, const std::unordered_map<std::string, int64_t>& stats, const absl::Duration& interval) noexcept;
     void CpuPeakUtilizationV2(const absl::Time& now, const std::unordered_map<std::string, int64_t>& stats,
                               const double cpuCount) noexcept;
     void CpuProcessingCapacity(const absl::Time& now, const double cpuCount, const absl::Duration& interval) noexcept;

lib/collectors/cgroup/test/cgroup_test.cpp

@@ -106,10 +106,13 @@ TEST(CGroup, CpuUtilizationV2)
     CGroupTest cGroup{&registry, "lib/collectors/cgroup/test/resources/sample1"};
     setenv("TITUS_NUM_CPU", "1", 1);
 
+    std::unordered_map<std::string, int64_t> stats;
+    atlasagent::parse_kv_from_file(cGroup.path_prefix_, "cpu.stat", &stats);
+
     // Use a fixed base time for consistent testing
     auto baseTime = absl::FromUnixSeconds(1000000000); // Fixed timestamp
     auto cpuCount = cGroup.GetNumCpu();
-    cGroup.CpuUtilizationV2(baseTime, cpuCount, absl::Seconds(60));
+    cGroup.CpuUtilizationV2(baseTime, cpuCount, stats, absl::Seconds(60));
 
     auto memoryWriter = static_cast<MemoryWriter*>(WriterTestHelper::GetImpl());
     auto messages = memoryWriter->GetMessages();
@@ -121,7 +124,8 @@ TEST(CGroup, CpuUtilizationV2)
 
     // Second call after 60 seconds to compute utilization
     cGroup.SetPrefix("lib/collectors/cgroup/test/resources/sample2");
-    cGroup.CpuUtilizationV2(baseTime + absl::Seconds(60), cpuCount, absl::Seconds(60));
+    atlasagent::parse_kv_from_file(cGroup.path_prefix_, "cpu.stat", &stats);
+    cGroup.CpuUtilizationV2(baseTime + absl::Seconds(60), cpuCount, stats, absl::Seconds(60));
 
     messages = memoryWriter->GetMessages();
     EXPECT_EQ(messages.size(), 5);

lib/collectors/proc/src/proc.cpp

@@ -623,10 +623,28 @@ catch (const std::exception& ex)
     return;
 }
 
-void Proc::UpdateCoreUtilization(const std::vector<std::vector<std::string>>& cpuLines) try
+void Proc::UpdateCoreUtilization(const std::vector<std::vector<std::string>>& cpuLines, const bool sixtySecondMetricsEnabled) try
 {
-    static DistributionSummary coresDistSummary = registry_->CreateDistributionSummary("sys.cpu.coreUtilization");
+    /*
+    These metrics were previously recorded as a distribution summary, which behaved correctly
+    when collected at 60-second intervals. However, when we introduced 5-second collection intervals,
+    internal Netflix users were upset that the max usage values were no longer representative of the
+    max core usage across the minute. The 5-second max values would capture brief CPU spikes that
+    weren't indicative of sustained core utilization over a full minute.
+    
+    To address this issue, we now manually implement the internal representation of a distribution
+    summary by updating its constituent counter fields directly. This approach allows us to compute
+    the maximum average CPU usage across all cores over a 60-second window, rather than reporting
+    the maximum usage over individual 5-second intervals.
+    */
+    
     static std::unordered_map<std::string, CpuStatFields> previousCpuStats;
+    static std::unordered_map<std::string, double> previousCoreUsages;
+
+    static auto counterCount = registry_->CreateCounter("sys.cpu.coreUtilization", {{"statistic", "count"}});
+    static auto counterTotal = registry_->CreateCounter("sys.cpu.coreUtilization", {{"statistic", "totalAmount"}});
+    static auto counterTotalSquares = registry_->CreateCounter("sys.cpu.coreUtilization", {{"statistic", "totalOfSquares"}});
+    static auto gaugeMax = registry_->CreateMaxGauge("sys.cpu.coreUtilization", {{"statistic", "max"}});
 
     for (unsigned int i = ProcStatConstants::FirstProcessorIndex; i < cpuLines.size(); ++i)
     {
@@ -641,12 +659,37 @@ void Proc::UpdateCoreUtilization(const std::vector<std::vector<std::string>>& cp
             auto computedVals = ComputeGaugeValues(prevStats, currentStats);
             auto usage = computedVals.user + computedVals.system + computedVals.stolen + computedVals.nice +
                          computedVals.wait + computedVals.interrupt + computedVals.guest;
-            coresDistSummary.Record(usage);
+            
+            counterCount.Increment();
+            counterTotal.Increment(usage);
+            counterTotalSquares.Increment(usage * usage);
+            previousCoreUsages[key] += usage;
 
             // Update the stored stats for next iteration
             it->second = currentStats;
         }
     }
+
+    // If 60-second metrics are enabled, compute the max average usage across all cores over the minute
+    // previousCoreUsages will never be empty this is just for the unit test
+    if (sixtySecondMetricsEnabled && !previousCoreUsages.empty())
+    {
+        // Find the max usage in previousCoreUsages
+        double maxUsage = 0.0;
+        for (const auto& [key, usage] : previousCoreUsages)
+        {
+            maxUsage = std::max(maxUsage, usage);
+        }
+
+        // Divide the usage by 12 to get average over the minute
+        double avgUsage = maxUsage / 12.0;
+
+        // Set the gauge mean to the max average usage
+        gaugeMax.Set(avgUsage);
+
+        previousCoreUsages.clear();
+    }
+
     return;
 }
 catch (const std::exception& ex)
@@ -708,7 +751,7 @@ void Proc::CpuStats(const bool fiveSecondMetrics, const bool sixtySecondMetricsE
     // If 5-second metrics are enabled, collect additional detailed metrics
     if (fiveSecondMetrics)
     {
-        UpdateCoreUtilization(cpuLines);
+        UpdateCoreUtilization(cpuLines, sixtySecondMetricsEnabled);
     }
 
     // Always collect peak stats (called every 1 second)

lib/collectors/proc/src/proc.h

@@ -32,7 +32,7 @@ class Proc
    private:
     void PeakCpuStats(const std::vector<std::string>& aggregateLine);
     void UpdateUtilizationGauges(const std::vector<std::string>& aggregateLine);
-    void UpdateCoreUtilization(const std::vector<std::vector<std::string>>& cpu_lines);
+    void UpdateCoreUtilization(const std::vector<std::vector<std::string>>& cpu_lines, const bool sixtySecondMetricsEnabled);
     void UpdateNumProcs(const unsigned int numberProcessors);
 
     void handle_line(FILE* fp) noexcept;

lib/collectors/proc/test/proc_test.cpp

@@ -273,7 +273,7 @@ TEST(Proc, CpuStats)
     proc.CpuStats(true, true);
     messages = memoryWriter->GetMessages();
 
-    EXPECT_EQ(16, messages.size());
+    EXPECT_EQ(23, messages.size());
     EXPECT_EQ(messages.at(0), "g:sys.cpu.utilization,id=user:11.314429\n");
     EXPECT_EQ(messages.at(1), "g:sys.cpu.utilization,id=system:1.291190\n");
     EXPECT_EQ(messages.at(2), "g:sys.cpu.utilization,id=stolen:0.006184\n");
@@ -282,16 +282,24 @@ TEST(Proc, CpuStats)
     EXPECT_EQ(messages.at(5), "g:sys.cpu.utilization,id=interrupt:0.002278\n");
 
     EXPECT_EQ(messages.at(6), "g:sys.cpu.numProcessors:3.000000\n");
-    EXPECT_EQ(messages.at(7), "d:sys.cpu.coreUtilization:5.528345\n");
-    EXPECT_EQ(messages.at(8), "d:sys.cpu.coreUtilization:14.059896\n");
-    EXPECT_EQ(messages.at(9), "d:sys.cpu.coreUtilization:7.583136\n");
-
-    EXPECT_EQ(messages.at(10), "m:sys.cpu.peakUtilization,id=user:11.314429\n");
-    EXPECT_EQ(messages.at(11), "m:sys.cpu.peakUtilization,id=system:1.291190\n");
-    EXPECT_EQ(messages.at(12), "m:sys.cpu.peakUtilization,id=stolen:0.006184\n");
-    EXPECT_EQ(messages.at(13), "m:sys.cpu.peakUtilization,id=nice:0.029293\n");
-    EXPECT_EQ(messages.at(14), "m:sys.cpu.peakUtilization,id=wait:0.011066\n");
-    EXPECT_EQ(messages.at(15), "m:sys.cpu.peakUtilization,id=interrupt:0.002278\n");
+    EXPECT_EQ(messages.at(7), "c:sys.cpu.coreUtilization,statistic=count:1.000000\n");
+    EXPECT_EQ(messages.at(8), "c:sys.cpu.coreUtilization,statistic=totalAmount:5.528345\n");
+    EXPECT_EQ(messages.at(9), "c:sys.cpu.coreUtilization,statistic=totalOfSquares:30.562600\n");
+
+    EXPECT_EQ(messages.at(10), "c:sys.cpu.coreUtilization,statistic=count:1.000000\n");
+    EXPECT_EQ(messages.at(11), "c:sys.cpu.coreUtilization,statistic=totalAmount:14.059896\n");
+    EXPECT_EQ(messages.at(12), "c:sys.cpu.coreUtilization,statistic=totalOfSquares:197.680671\n");
+    EXPECT_EQ(messages.at(13), "c:sys.cpu.coreUtilization,statistic=count:1.000000\n");
+    EXPECT_EQ(messages.at(14), "c:sys.cpu.coreUtilization,statistic=totalAmount:7.583136\n");
+    EXPECT_EQ(messages.at(15), "c:sys.cpu.coreUtilization,statistic=totalOfSquares:57.503949\n");
+    EXPECT_EQ(messages.at(16), "m:sys.cpu.coreUtilization,statistic=max:1.171658\n");
+
+    EXPECT_EQ(messages.at(17), "m:sys.cpu.peakUtilization,id=user:11.314429\n");
+    EXPECT_EQ(messages.at(18), "m:sys.cpu.peakUtilization,id=system:1.291190\n");
+    EXPECT_EQ(messages.at(19), "m:sys.cpu.peakUtilization,id=stolen:0.006184\n");
+    EXPECT_EQ(messages.at(20), "m:sys.cpu.peakUtilization,id=nice:0.029293\n");
+    EXPECT_EQ(messages.at(21), "m:sys.cpu.peakUtilization,id=wait:0.011066\n");
+    EXPECT_EQ(messages.at(22), "m:sys.cpu.peakUtilization,id=interrupt:0.002278\n");
 }
 
 TEST(Proc, UptimeStats)