From 1d279f8b2bed65b5a57f34d936fe5c43688a333e Mon Sep 17 00:00:00 2001
From: karteekmurthys <karteek.murthy@ibm.com>
Date: Fri, 22 Dec 2023 11:19:18 -0800
Subject: [PATCH 1/6] Expose Rest API to return metrics in prometheus format

---
 presto-native-execution/etc/config.properties |   6 +-
 .../presto_cpp/main/PrestoMain.cpp            |   6 +-
 .../presto_cpp/main/PrestoServer.cpp          |  14 ++
 .../presto_cpp/main/PrestoServer.h            |   2 +
 .../presto_cpp/main/common/CMakeLists.txt     |   3 +-
 .../main/common/PrometheusStatsReporter.cpp   | 111 +++++++++++++++
 .../main/common/PrometheusStatsReporter.h     | 129 ++++++++++++++++++
 .../presto_cpp/main/tests/CMakeLists.txt      |   5 +
 .../tests/PrometheusStatsReporterTest.cpp     |  68 +++++++++
 9 files changed, 336 insertions(+), 8 deletions(-)
 create mode 100644 presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.cpp
 create mode 100644 presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.h
 create mode 100644 presto-native-execution/presto_cpp/main/tests/PrometheusStatsReporterTest.cpp

diff --git a/presto-native-execution/etc/config.properties b/presto-native-execution/etc/config.properties
index 73654217fa154..8f98a65ae9115 100644
--- a/presto-native-execution/etc/config.properties
+++ b/presto-native-execution/etc/config.properties
@@ -1,5 +1,7 @@
-discovery.uri=http://127.0.0.1:58215
+discovery.uri=http://127.0.0.1:8080
 presto.version=testversion
 http-server.http.port=7777
 shutdown-onset-sec=1
-register-test-functions=true
\ No newline at end of file
+register-test-functions=true
+query.max-memory-per-node=30GB
+task.max-drivers-per-task=1
\ No newline at end of file
diff --git a/presto-native-execution/presto_cpp/main/PrestoMain.cpp b/presto-native-execution/presto_cpp/main/PrestoMain.cpp
index 9e74b5b1dacfd..554677a83801d 100644
--- a/presto-native-execution/presto_cpp/main/PrestoMain.cpp
+++ b/presto-native-execution/presto_cpp/main/PrestoMain.cpp
@@ -27,11 +27,7 @@ int main(int argc, char* argv[]) {
   google::InstallFailureSignalHandler();
   PRESTO_STARTUP_LOG(INFO) << "Entering main()";
   facebook::presto::PrestoServer presto(FLAGS_etc_dir);
+  facebook::velox::BaseStatsReporter::registered = true;
   presto.run();
   PRESTO_SHUTDOWN_LOG(INFO) << "Exiting main()";
 }
-
-// Initialize singleton for the reporter.
-folly::Singleton<facebook::velox::BaseStatsReporter> reporter([]() {
-  return new facebook::velox::DummyStatsReporter();
-});
diff --git a/presto-native-execution/presto_cpp/main/PrestoServer.cpp b/presto-native-execution/presto_cpp/main/PrestoServer.cpp
index 16eb451b24673..6b9982e5de292 100644
--- a/presto-native-execution/presto_cpp/main/PrestoServer.cpp
+++ b/presto-native-execution/presto_cpp/main/PrestoServer.cpp
@@ -24,6 +24,7 @@
 #include "presto_cpp/main/TaskResource.h"
 #include "presto_cpp/main/common/ConfigReader.h"
 #include "presto_cpp/main/common/Counters.h"
+#include "presto_cpp/main/common/PrometheusStatsReporter.h"
 #include "presto_cpp/main/common/Utils.h"
 #include "presto_cpp/main/http/filters/AccessLogFilter.h"
 #include "presto_cpp/main/http/filters/HttpEndpointLatencyFilter.h"
@@ -319,6 +320,14 @@ void PrestoServer::run() {
           proxygen::ResponseHandler* downstream) {
         server->reportServerInfo(downstream);
       });
+  httpServer_->registerGet(
+      "/v1/info/health/metrics",
+      [server = this](
+          proxygen::HTTPMessage* /*message*/,
+          const std::vector<std::unique_ptr<folly::IOBuf>>& /*body*/,
+          proxygen::ResponseHandler* downstream) {
+        server->reportHealthMetrics(downstream);
+      });
   httpServer_->registerGet(
       "/v1/info/state",
       [server = this](
@@ -1093,6 +1102,11 @@ void PrestoServer::reportServerInfo(proxygen::ResponseHandler* downstream) {
   http::sendOkResponse(downstream, json(serverInfo));
 }
 
+void PrestoServer::reportHealthMetrics(proxygen::ResponseHandler* downstream) {
+  auto reporter = std::dynamic_pointer_cast<PrometheusStatsReporter>(
+      folly::Singleton<facebook::velox::BaseStatsReporter>::try_get());
+  http::sendOkResponse(downstream, reporter->getMetricsForPrometheus());
+}
 void PrestoServer::reportNodeStatus(proxygen::ResponseHandler* downstream) {
   http::sendOkResponse(downstream, json(fetchNodeStatus()));
 }
diff --git a/presto-native-execution/presto_cpp/main/PrestoServer.h b/presto-native-execution/presto_cpp/main/PrestoServer.h
index 1fe61f52a9847..e70ca07040e4d 100644
--- a/presto-native-execution/presto_cpp/main/PrestoServer.h
+++ b/presto-native-execution/presto_cpp/main/PrestoServer.h
@@ -169,6 +169,8 @@ class PrestoServer {
 
   void reportNodeStatus(proxygen::ResponseHandler* downstream);
 
+  void reportHealthMetrics(proxygen::ResponseHandler* downstream);
+
   protocol::NodeStatus fetchNodeStatus();
 
   void populateMemAndCPUInfo();
diff --git a/presto-native-execution/presto_cpp/main/common/CMakeLists.txt b/presto-native-execution/presto_cpp/main/common/CMakeLists.txt
index 153883e23d531..1c511121f3adb 100644
--- a/presto-native-execution/presto_cpp/main/common/CMakeLists.txt
+++ b/presto-native-execution/presto_cpp/main/common/CMakeLists.txt
@@ -11,7 +11,8 @@
 # limitations under the License.
 
 add_library(presto_exception Exception.cpp)
-add_library(presto_common Counters.cpp Utils.cpp ConfigReader.cpp Configs.cpp)
+add_library(presto_common Counters.cpp Utils.cpp ConfigReader.cpp Configs.cpp
+                          PrometheusStatsReporter.cpp)
 
 target_link_libraries(presto_exception velox_exception)
 set_property(TARGET presto_exception PROPERTY JOB_POOL_LINK
diff --git a/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.cpp b/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.cpp
new file mode 100644
index 0000000000000..c88bf65d166e6
--- /dev/null
+++ b/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.cpp
@@ -0,0 +1,111 @@
+/*
+ * Licensed 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.
+ */
+
+#include "PrometheusStatsReporter.h"
+
+namespace facebook::presto {
+
+void PrometheusStatsReporter::registerMetricExportType(
+    folly::StringPiece key,
+    facebook::velox::StatType statType) const {
+  registerMetricExportType(key.start(), statType);
+}
+
+void PrometheusStatsReporter::registerMetricExportType(
+    const char* key,
+    facebook::velox::StatType statType) const {
+  std::lock_guard<std::mutex> lock(mutex_);
+  registeredStats_.emplace(key, statType);
+  if (statType == velox::StatType::COUNT) {
+    counterMetrics.emplace(key, 0);
+  }
+  if (statType == velox::StatType::SUM) {
+    gaugeMetrics.emplace(key, std::vector<std::shared_ptr<Gauge>>());
+  }
+}
+
+const uint64_t getCurrentEpochTimestamp() {
+  auto p1 = std::chrono::system_clock::now();
+  return std::chrono::duration_cast<std::chrono::seconds>(p1.time_since_epoch())
+      .count();
+}
+
+void PrometheusStatsReporter::addMetricValue(const char* key, size_t value)
+    const {
+  auto it = registeredStats_.find(key);
+  if (it == registeredStats_.end()) {
+    VLOG(2) << "addMetricValue() for unregistered stat";
+    return;
+  }
+  if (it->second == facebook::velox::StatType::COUNT) {
+    std::lock_guard<std::mutex> lock(mutex_);
+    auto countItr = counterMetrics.find(key);
+    if (countItr == counterMetrics.end()) {
+      counterMetrics.emplace(key, 1);
+      return;
+    }
+    countItr->second++;
+    return;
+  }
+  std::lock_guard<std::mutex> lock(mutex_);
+  std::shared_ptr<Gauge> metric = std::make_shared<Gauge>();
+  metric->timestamp = getCurrentEpochTimestamp();
+  metric->value = value;
+  gaugeMetrics[key].push_back(std::move(metric));
+}
+
+void PrometheusStatsReporter::addMetricValue(
+    const std::string& key,
+    size_t value) const {
+  addMetricValue(key.c_str(), value);
+}
+
+void PrometheusStatsReporter::addMetricValue(
+    folly::StringPiece key,
+    size_t value) const {
+  addMetricValue(key.start(), value);
+}
+
+const std::string PrometheusStatsReporter::getMetricsForPrometheus() {
+  std::lock_guard<std::mutex> lock(mutex_);
+  std::stringstream ss;
+  for (const auto gauge : gaugeMetrics) {
+    auto metricName = gauge.first.substr(gauge.first.find(".") + 1);
+    ss << "# HELP " << metricName << std::endl;
+    ss << "# TYPE " << metricName << " gauge" << std::endl;
+    for (auto metric : gauge.second) {
+      ss << metricName << "{cluster=\"" << cluster_ << "\""
+         << ",worker=\"" << workerPod_ << "\"} " << metric->value << " "
+         << metric->timestamp * 1000 << std::endl;
+    }
+    std::vector<std::shared_ptr<Gauge>> metricsVector = gauge.second;
+    metricsVector.clear();
+  }
+
+  for (const auto counter : counterMetrics) {
+    auto metricName = counter.first.substr(counter.first.find(".") + 1);
+    ss << "# HELP " << metricName << std::endl;
+    ss << "# TYPE " << metricName << " "
+       << "counter" << std::endl;
+    ss << metricName << "{cluster=\"" << cluster_ << "\",worker=\""
+       << workerPod_ << "\"} " << counter.second << std::endl;
+  }
+  return ss.str();
+}
+
+// Initialize singleton for the reporter
+folly::Singleton<facebook::velox::BaseStatsReporter> reporter([]() {
+  return new PrometheusStatsReporter();
+});
+} // namespace facebook::presto
diff --git a/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.h b/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.h
new file mode 100644
index 0000000000000..be6e44b2de478
--- /dev/null
+++ b/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.h
@@ -0,0 +1,129 @@
+/*
+ * Licensed 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.
+ */
+
+#include <folly/dynamic.h>
+#include <fstream>
+#include <iostream>
+#include "presto_cpp/main/common/Configs.h"
+#include "presto_cpp/main/common/Counters.h"
+#include "velox/common/base/StatsReporter.h"
+#include "velox/velox/common/base/Exceptions.h"
+
+namespace facebook::presto {
+
+class PrometheusStatsReporter : public facebook::velox::BaseStatsReporter {
+ public:
+  PrometheusStatsReporter(
+      const std::string cluster = "",
+      const std::string worker = "") {
+    if (cluster.empty()) {
+      auto nodeConfig = facebook::presto::NodeConfig::instance();
+      cluster_ = nodeConfig->nodeEnvironment();
+    } else {
+      cluster_ = cluster;
+    }
+    char* hostName = std::getenv("HOSTNAME");
+    workerPod_ = !hostName ? worker : hostName;
+  }
+
+  struct Gauge {
+    uint64_t timestamp; // Epoch timestamp.
+    size_t value; // Metric value.
+  };
+
+  /// Register a stat of the given stat type.
+  /// @param key The key to identify the stat.
+  /// @param statType How the stat is aggregated.
+  void registerMetricExportType(
+      const char* key,
+      facebook::velox::StatType statType) const override;
+
+  void registerMetricExportType(
+      folly::StringPiece key,
+      facebook::velox::StatType statType) const override;
+
+  void registerHistogramMetricExportType(
+      const char* /*key*/,
+      int64_t /* bucketWidth */,
+      int64_t /* min */,
+      int64_t /* max */,
+      const std::vector<int32_t>& /* pcts */) const override {}
+
+  void registerHistogramMetricExportType(
+      folly::StringPiece /* key */,
+      int64_t /* bucketWidth */,
+      int64_t /* min */,
+      int64_t /* max */,
+      const std::vector<int32_t>& /* pcts */) const override {}
+
+  void addMetricValue(const std::string& key, size_t value = 1) const override;
+
+  void addMetricValue(const char* key, size_t value = 1) const override;
+
+  void addMetricValue(folly::StringPiece key, size_t value = 1) const override;
+
+  virtual void addHistogramMetricValue(const std::string& key, size_t value)
+      const override {}
+
+  virtual void addHistogramMetricValue(const char* key, size_t value)
+      const override {}
+
+  virtual void addHistogramMetricValue(folly::StringPiece key, size_t value)
+      const override {}
+
+  const facebook::velox::StatType getRegisteredStatType(
+      const std::string& metricName) {
+    std::lock_guard<std::mutex> lock(mutex_);
+    return registeredStats_[metricName];
+  }
+
+  /*
+   * Serializes the metrics collected so far in the format suitable for
+   * back filling Prometheus server.
+   *
+   * Given a metric name and a set of labels, time series are frequently
+   * identified using this notation:
+   *
+   *       <metric name>{<label name>=<label value>, ...}
+   *
+   * For example, a time series with the metric name num_tasks_aborted
+   * and the labels cluster="<cluster_id>" and worker="worker-id"
+   * could be written like this:
+   *  # HELP num_tasks_aborted
+   *  # TYPE num_tasks_aborted gauge*
+   *  num_tasks_aborted{cluster="<cluster_id>", worker="worker-id"} value
+   * timestamp
+   *
+   * Above info is from:
+   * https://prometheus.io/docs/concepts/data_model/#metric-names-and-labels
+   */
+  const std::string getMetricsForPrometheus();
+
+ private:
+  /// Mapping of registered stats key to StatType.
+  mutable std::unordered_map<std::string, facebook::velox::StatType>
+      registeredStats_;
+  /// A mapping from stats key of type COUNT to value.
+  mutable std::unordered_map<std::string, int64_t> counterMetrics;
+  /// A mapping from stats key to a vector of Gauge objects. Example AVG type
+  /// is a gauge. After every succesfull query of metrics, the values are
+  /// cleaned up.
+  mutable std::unordered_map<std::string, std::vector<std::shared_ptr<Gauge>>>
+      gaugeMetrics;
+  // Mutex to control access to registeredStats_ and metricMap_ members.
+  mutable std::mutex mutex_;
+  std::string cluster_;
+  std::string workerPod_;
+}; // class PrometheusStatsReporter
+} // namespace facebook::presto
diff --git a/presto-native-execution/presto_cpp/main/tests/CMakeLists.txt b/presto-native-execution/presto_cpp/main/tests/CMakeLists.txt
index 82719e5d1f95a..ad92949dfc142 100644
--- a/presto-native-execution/presto_cpp/main/tests/CMakeLists.txt
+++ b/presto-native-execution/presto_cpp/main/tests/CMakeLists.txt
@@ -27,6 +27,11 @@ add_test(
   COMMAND presto_server_test
   WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR})
 
+add_executable(ahana_stats_test PrometheusStatsReporterTest.cpp)
+
+target_link_libraries(ahana_stats_test presto_server_lib velox_exec_test_lib
+                      gtest gtest_main)
+
 target_link_libraries(
   presto_server_test
   velox_exec_test_lib
diff --git a/presto-native-execution/presto_cpp/main/tests/PrometheusStatsReporterTest.cpp b/presto-native-execution/presto_cpp/main/tests/PrometheusStatsReporterTest.cpp
new file mode 100644
index 0000000000000..78b2e96b21ac8
--- /dev/null
+++ b/presto-native-execution/presto_cpp/main/tests/PrometheusStatsReporterTest.cpp
@@ -0,0 +1,68 @@
+/*
+ * Licensed 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.
+ */
+
+#include "presto_cpp/main/common/PrometheusStatsReporter.h"
+#include <gtest/gtest.h>
+#include <cstring>
+
+namespace facebook::presto {
+
+class PrometheusStatsReporterTest : public testing::Test {
+  void SetUp() override {}
+
+  void TearDown() override {}
+};
+
+/// Tests addStatType and addStats functions.
+TEST_F(PrometheusStatsReporterTest, addStats) {
+  auto reporter = std::make_shared<PrometheusStatsReporter>(
+      "test_cluster", "test_worker_pod");
+
+  reporter->registerMetricExportType("key1", facebook::velox::StatType::COUNT);
+  reporter->registerMetricExportType("key2", facebook::velox::StatType::AVG);
+
+  EXPECT_EQ(
+      facebook::velox::StatType::COUNT,
+      reporter->getRegisteredStatType("key1"));
+  EXPECT_EQ(
+      facebook::velox::StatType::AVG, reporter->getRegisteredStatType("key2"));
+
+  std::unordered_set<size_t> testData = {10, 11, 15};
+  for (auto i : testData) {
+    reporter->addMetricValue("key1", i);
+    reporter->addMetricValue("key2", i + 1000);
+  }
+  // Uses default value of 1 for second parameter.
+  reporter->addMetricValue("key1");
+  auto prometheusFormat = reporter->getMetricsForPrometheus();
+  const std::string expected[] = {
+      "# HELP key2",
+      "# TYPE key2 gauge",
+      "key2{cluster=\"test_cluster\",worker=\"test_worker_pod\"} 1011",
+      "key2{cluster=\"test_cluster\",worker=\"test_worker_pod\"} 1015",
+      "key2{cluster=\"test_cluster\",worker=\"test_worker_pod\"} 1010",
+      "# HELP key1",
+      "# TYPE key1 counter",
+      "key1{cluster=\"test_cluster\",worker=\"test_worker_pod\"} 4"};
+
+  auto i = 0;
+  auto pos = prometheusFormat.find("\n");
+  while (pos != std::string::npos) {
+    auto line = prometheusFormat.substr(0, pos);
+    EXPECT_NE(line.find(expected[i++]), std::string::npos);
+    prometheusFormat = prometheusFormat.substr(pos + 1);
+    pos = prometheusFormat.find("\n");
+  }
+};
+} // namespace facebook::presto

From 49992f352bb5b30cdbb565bc0070cb48acdb7cb2 Mon Sep 17 00:00:00 2001
From: Karteekmurthys <karteek@ahana.io>
Date: Fri, 12 Jan 2024 16:46:45 -0800
Subject: [PATCH 2/6] Overwrite gauge metrics instead of batching them

---
 .../presto_cpp/main/PeriodicTaskManager.cpp   | 11 ++++
 .../main/common/PrometheusStatsReporter.cpp   | 54 ++++++-------------
 .../main/common/PrometheusStatsReporter.h     | 12 +----
 3 files changed, 29 insertions(+), 48 deletions(-)

diff --git a/presto-native-execution/presto_cpp/main/PeriodicTaskManager.cpp b/presto-native-execution/presto_cpp/main/PeriodicTaskManager.cpp
index 4f7bdb24961fd..2b87118796e06 100644
--- a/presto-native-execution/presto_cpp/main/PeriodicTaskManager.cpp
+++ b/presto-native-execution/presto_cpp/main/PeriodicTaskManager.cpp
@@ -284,6 +284,17 @@ void PeriodicTaskManager::updateCacheStats() {
   RECORD_METRIC_VALUE(
       kCounterMemoryCacheSumEvictScore, memoryCacheStats.sumEvictScore);
 
+  if (memoryCacheStats.ssdStats) {
+    RECORD_METRIC_VALUE(
+        kCounterSsdCacheCachedEntries,
+        memoryCacheStats.ssdStats->entriesCached);
+    RECORD_METRIC_VALUE(
+        kCounterSsdCacheCachedRegions,
+        memoryCacheStats.ssdStats->regionsCached);
+    RECORD_METRIC_VALUE(
+        kCounterSsdCacheCachedBytes, memoryCacheStats.ssdStats->bytesCached);
+  }
+
   // Interval cumulatives.
   RECORD_METRIC_VALUE(
       kCounterMemoryCacheNumHit,
diff --git a/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.cpp b/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.cpp
index c88bf65d166e6..f34ae84219545 100644
--- a/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.cpp
+++ b/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.cpp
@@ -27,12 +27,7 @@ void PrometheusStatsReporter::registerMetricExportType(
     facebook::velox::StatType statType) const {
   std::lock_guard<std::mutex> lock(mutex_);
   registeredStats_.emplace(key, statType);
-  if (statType == velox::StatType::COUNT) {
-    counterMetrics.emplace(key, 0);
-  }
-  if (statType == velox::StatType::SUM) {
-    gaugeMetrics.emplace(key, std::vector<std::shared_ptr<Gauge>>());
-  }
+  metricsMap_.emplace(key, 0);
 }
 
 const uint64_t getCurrentEpochTimestamp() {
@@ -43,26 +38,19 @@ const uint64_t getCurrentEpochTimestamp() {
 
 void PrometheusStatsReporter::addMetricValue(const char* key, size_t value)
     const {
+  std::lock_guard<std::mutex> lock(mutex_);
   auto it = registeredStats_.find(key);
   if (it == registeredStats_.end()) {
-    VLOG(2) << "addMetricValue() for unregistered stat";
+    VLOG(1) << "addMetricValue() for unregistered stat " << key;
     return;
   }
   if (it->second == facebook::velox::StatType::COUNT) {
-    std::lock_guard<std::mutex> lock(mutex_);
-    auto countItr = counterMetrics.find(key);
-    if (countItr == counterMetrics.end()) {
-      counterMetrics.emplace(key, 1);
-      return;
-    }
-    countItr->second++;
+    // increment the counter.
+    metricsMap_[key]++;
     return;
   }
-  std::lock_guard<std::mutex> lock(mutex_);
-  std::shared_ptr<Gauge> metric = std::make_shared<Gauge>();
-  metric->timestamp = getCurrentEpochTimestamp();
-  metric->value = value;
-  gaugeMetrics[key].push_back(std::move(metric));
+  // Gauge type metric value must be reset.
+  metricsMap_[key] = value;
 }
 
 void PrometheusStatsReporter::addMetricValue(
@@ -80,26 +68,18 @@ void PrometheusStatsReporter::addMetricValue(
 const std::string PrometheusStatsReporter::getMetricsForPrometheus() {
   std::lock_guard<std::mutex> lock(mutex_);
   std::stringstream ss;
-  for (const auto gauge : gaugeMetrics) {
-    auto metricName = gauge.first.substr(gauge.first.find(".") + 1);
+  for (const auto metric : metricsMap_) {
+    auto metricName = metric.first;
+    std::replace(metricName.begin(), metricName.end(), '.', '_');
+    auto statType = registeredStats_[metric.first];
     ss << "# HELP " << metricName << std::endl;
-    ss << "# TYPE " << metricName << " gauge" << std::endl;
-    for (auto metric : gauge.second) {
-      ss << metricName << "{cluster=\"" << cluster_ << "\""
-         << ",worker=\"" << workerPod_ << "\"} " << metric->value << " "
-         << metric->timestamp * 1000 << std::endl;
+    std::string statTypeStr = "gauge";
+    if (statType == facebook::velox::StatType::COUNT) {
+      statTypeStr = "counter";
     }
-    std::vector<std::shared_ptr<Gauge>> metricsVector = gauge.second;
-    metricsVector.clear();
-  }
-
-  for (const auto counter : counterMetrics) {
-    auto metricName = counter.first.substr(counter.first.find(".") + 1);
-    ss << "# HELP " << metricName << std::endl;
-    ss << "# TYPE " << metricName << " "
-       << "counter" << std::endl;
-    ss << metricName << "{cluster=\"" << cluster_ << "\",worker=\""
-       << workerPod_ << "\"} " << counter.second << std::endl;
+    ss << "# TYPE " << metricName << " " << statTypeStr << std::endl;
+    ss << metricName << "{cluster=\"" << cluster_ << "\""
+       << ",worker=\"" << workerPod_ << "\"} " << metric.second << std::endl;
   }
   return ss.str();
 }
diff --git a/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.h b/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.h
index be6e44b2de478..9bf91e7a37f9c 100644
--- a/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.h
+++ b/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.h
@@ -37,11 +37,6 @@ class PrometheusStatsReporter : public facebook::velox::BaseStatsReporter {
     workerPod_ = !hostName ? worker : hostName;
   }
 
-  struct Gauge {
-    uint64_t timestamp; // Epoch timestamp.
-    size_t value; // Metric value.
-  };
-
   /// Register a stat of the given stat type.
   /// @param key The key to identify the stat.
   /// @param statType How the stat is aggregated.
@@ -115,12 +110,7 @@ class PrometheusStatsReporter : public facebook::velox::BaseStatsReporter {
   mutable std::unordered_map<std::string, facebook::velox::StatType>
       registeredStats_;
   /// A mapping from stats key of type COUNT to value.
-  mutable std::unordered_map<std::string, int64_t> counterMetrics;
-  /// A mapping from stats key to a vector of Gauge objects. Example AVG type
-  /// is a gauge. After every succesfull query of metrics, the values are
-  /// cleaned up.
-  mutable std::unordered_map<std::string, std::vector<std::shared_ptr<Gauge>>>
-      gaugeMetrics;
+  mutable std::unordered_map<std::string, int64_t> metricsMap_;
   // Mutex to control access to registeredStats_ and metricMap_ members.
   mutable std::mutex mutex_;
   std::string cluster_;

From 4d489a495aeae2180ac4e5d7548e190f83df7a74 Mon Sep 17 00:00:00 2001
From: Karteekmurthys <karteek@ahana.io>
Date: Thu, 18 Jan 2024 10:39:00 -0800
Subject: [PATCH 3/6] Refactor PrometheusStatsReporter to StatsReporterImpl"

---
 presto-native-execution/etc/config.properties |  6 ++---
 .../presto_cpp/main/PrestoServer.cpp          |  4 +--
 .../presto_cpp/main/common/CMakeLists.txt     |  2 +-
 ...tatsReporter.cpp => StatsReporterImpl.cpp} | 25 ++++++++-----------
 ...eusStatsReporter.h => StatsReporterImpl.h} |  6 ++---
 .../presto_cpp/main/tests/CMakeLists.txt      |  2 +-
 ...rterTest.cpp => StatsReporterImplTest.cpp} | 16 ++++++------
 7 files changed, 27 insertions(+), 34 deletions(-)
 rename presto-native-execution/presto_cpp/main/common/{PrometheusStatsReporter.cpp => StatsReporterImpl.cpp} (81%)
 rename presto-native-execution/presto_cpp/main/common/{PrometheusStatsReporter.h => StatsReporterImpl.h} (96%)
 rename presto-native-execution/presto_cpp/main/tests/{PrometheusStatsReporterTest.cpp => StatsReporterImplTest.cpp} (80%)

diff --git a/presto-native-execution/etc/config.properties b/presto-native-execution/etc/config.properties
index 8f98a65ae9115..73654217fa154 100644
--- a/presto-native-execution/etc/config.properties
+++ b/presto-native-execution/etc/config.properties
@@ -1,7 +1,5 @@
-discovery.uri=http://127.0.0.1:8080
+discovery.uri=http://127.0.0.1:58215
 presto.version=testversion
 http-server.http.port=7777
 shutdown-onset-sec=1
-register-test-functions=true
-query.max-memory-per-node=30GB
-task.max-drivers-per-task=1
\ No newline at end of file
+register-test-functions=true
\ No newline at end of file
diff --git a/presto-native-execution/presto_cpp/main/PrestoServer.cpp b/presto-native-execution/presto_cpp/main/PrestoServer.cpp
index 6b9982e5de292..c5cbbc66b3ab6 100644
--- a/presto-native-execution/presto_cpp/main/PrestoServer.cpp
+++ b/presto-native-execution/presto_cpp/main/PrestoServer.cpp
@@ -24,7 +24,7 @@
 #include "presto_cpp/main/TaskResource.h"
 #include "presto_cpp/main/common/ConfigReader.h"
 #include "presto_cpp/main/common/Counters.h"
-#include "presto_cpp/main/common/PrometheusStatsReporter.h"
+#include "presto_cpp/main/common/StatsReporterImpl.h"
 #include "presto_cpp/main/common/Utils.h"
 #include "presto_cpp/main/http/filters/AccessLogFilter.h"
 #include "presto_cpp/main/http/filters/HttpEndpointLatencyFilter.h"
@@ -1103,7 +1103,7 @@ void PrestoServer::reportServerInfo(proxygen::ResponseHandler* downstream) {
 }
 
 void PrestoServer::reportHealthMetrics(proxygen::ResponseHandler* downstream) {
-  auto reporter = std::dynamic_pointer_cast<PrometheusStatsReporter>(
+  auto reporter = std::dynamic_pointer_cast<StatsReporterImpl>(
       folly::Singleton<facebook::velox::BaseStatsReporter>::try_get());
   http::sendOkResponse(downstream, reporter->getMetricsForPrometheus());
 }
diff --git a/presto-native-execution/presto_cpp/main/common/CMakeLists.txt b/presto-native-execution/presto_cpp/main/common/CMakeLists.txt
index 1c511121f3adb..af616052ab84e 100644
--- a/presto-native-execution/presto_cpp/main/common/CMakeLists.txt
+++ b/presto-native-execution/presto_cpp/main/common/CMakeLists.txt
@@ -12,7 +12,7 @@
 
 add_library(presto_exception Exception.cpp)
 add_library(presto_common Counters.cpp Utils.cpp ConfigReader.cpp Configs.cpp
-                          PrometheusStatsReporter.cpp)
+                          StatsReporterImpl.cpp)
 
 target_link_libraries(presto_exception velox_exception)
 set_property(TARGET presto_exception PROPERTY JOB_POOL_LINK
diff --git a/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.cpp b/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.cpp
similarity index 81%
rename from presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.cpp
rename to presto-native-execution/presto_cpp/main/common/StatsReporterImpl.cpp
index f34ae84219545..aaa39ce8447ba 100644
--- a/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.cpp
+++ b/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.cpp
@@ -12,17 +12,17 @@
  * limitations under the License.
  */
 
-#include "PrometheusStatsReporter.h"
+#include "StatsReporterImpl.h"
 
 namespace facebook::presto {
 
-void PrometheusStatsReporter::registerMetricExportType(
+void StatsReporterImpl::registerMetricExportType(
     folly::StringPiece key,
     facebook::velox::StatType statType) const {
   registerMetricExportType(key.start(), statType);
 }
 
-void PrometheusStatsReporter::registerMetricExportType(
+void StatsReporterImpl::registerMetricExportType(
     const char* key,
     facebook::velox::StatType statType) const {
   std::lock_guard<std::mutex> lock(mutex_);
@@ -36,8 +36,7 @@ const uint64_t getCurrentEpochTimestamp() {
       .count();
 }
 
-void PrometheusStatsReporter::addMetricValue(const char* key, size_t value)
-    const {
+void StatsReporterImpl::addMetricValue(const char* key, size_t value) const {
   std::lock_guard<std::mutex> lock(mutex_);
   auto it = registeredStats_.find(key);
   if (it == registeredStats_.end()) {
@@ -46,26 +45,24 @@ void PrometheusStatsReporter::addMetricValue(const char* key, size_t value)
   }
   if (it->second == facebook::velox::StatType::COUNT) {
     // increment the counter.
-    metricsMap_[key]++;
+    metricsMap_[key] += value;
     return;
   }
   // Gauge type metric value must be reset.
   metricsMap_[key] = value;
 }
 
-void PrometheusStatsReporter::addMetricValue(
-    const std::string& key,
-    size_t value) const {
+void StatsReporterImpl::addMetricValue(const std::string& key, size_t value)
+    const {
   addMetricValue(key.c_str(), value);
 }
 
-void PrometheusStatsReporter::addMetricValue(
-    folly::StringPiece key,
-    size_t value) const {
+void StatsReporterImpl::addMetricValue(folly::StringPiece key, size_t value)
+    const {
   addMetricValue(key.start(), value);
 }
 
-const std::string PrometheusStatsReporter::getMetricsForPrometheus() {
+const std::string StatsReporterImpl::getMetricsForPrometheus() {
   std::lock_guard<std::mutex> lock(mutex_);
   std::stringstream ss;
   for (const auto metric : metricsMap_) {
@@ -86,6 +83,6 @@ const std::string PrometheusStatsReporter::getMetricsForPrometheus() {
 
 // Initialize singleton for the reporter
 folly::Singleton<facebook::velox::BaseStatsReporter> reporter([]() {
-  return new PrometheusStatsReporter();
+  return new StatsReporterImpl();
 });
 } // namespace facebook::presto
diff --git a/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.h b/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.h
similarity index 96%
rename from presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.h
rename to presto-native-execution/presto_cpp/main/common/StatsReporterImpl.h
index 9bf91e7a37f9c..042f85f215e20 100644
--- a/presto-native-execution/presto_cpp/main/common/PrometheusStatsReporter.h
+++ b/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.h
@@ -22,9 +22,9 @@
 
 namespace facebook::presto {
 
-class PrometheusStatsReporter : public facebook::velox::BaseStatsReporter {
+class StatsReporterImpl : public facebook::velox::BaseStatsReporter {
  public:
-  PrometheusStatsReporter(
+  StatsReporterImpl(
       const std::string cluster = "",
       const std::string worker = "") {
     if (cluster.empty()) {
@@ -115,5 +115,5 @@ class PrometheusStatsReporter : public facebook::velox::BaseStatsReporter {
   mutable std::mutex mutex_;
   std::string cluster_;
   std::string workerPod_;
-}; // class PrometheusStatsReporter
+}; // class StatsReporterImpl
 } // namespace facebook::presto
diff --git a/presto-native-execution/presto_cpp/main/tests/CMakeLists.txt b/presto-native-execution/presto_cpp/main/tests/CMakeLists.txt
index ad92949dfc142..c2b9f81deacc1 100644
--- a/presto-native-execution/presto_cpp/main/tests/CMakeLists.txt
+++ b/presto-native-execution/presto_cpp/main/tests/CMakeLists.txt
@@ -27,7 +27,7 @@ add_test(
   COMMAND presto_server_test
   WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR})
 
-add_executable(ahana_stats_test PrometheusStatsReporterTest.cpp)
+add_executable(ahana_stats_test StatsReporterImplTest.cpp)
 
 target_link_libraries(ahana_stats_test presto_server_lib velox_exec_test_lib
                       gtest gtest_main)
diff --git a/presto-native-execution/presto_cpp/main/tests/PrometheusStatsReporterTest.cpp b/presto-native-execution/presto_cpp/main/tests/StatsReporterImplTest.cpp
similarity index 80%
rename from presto-native-execution/presto_cpp/main/tests/PrometheusStatsReporterTest.cpp
rename to presto-native-execution/presto_cpp/main/tests/StatsReporterImplTest.cpp
index 78b2e96b21ac8..03735f9358aed 100644
--- a/presto-native-execution/presto_cpp/main/tests/PrometheusStatsReporterTest.cpp
+++ b/presto-native-execution/presto_cpp/main/tests/StatsReporterImplTest.cpp
@@ -12,22 +12,22 @@
  * limitations under the License.
  */
 
-#include "presto_cpp/main/common/PrometheusStatsReporter.h"
+#include "presto_cpp/main/common/StatsReporterImpl.h"
 #include <gtest/gtest.h>
 #include <cstring>
 
 namespace facebook::presto {
 
-class PrometheusStatsReporterTest : public testing::Test {
+class StatsReporterImplTest : public testing::Test {
   void SetUp() override {}
 
   void TearDown() override {}
 };
 
 /// Tests addStatType and addStats functions.
-TEST_F(PrometheusStatsReporterTest, addStats) {
-  auto reporter = std::make_shared<PrometheusStatsReporter>(
-      "test_cluster", "test_worker_pod");
+TEST_F(StatsReporterImplTest, addStats) {
+  auto reporter =
+      std::make_shared<StatsReporterImpl>("test_cluster", "test_worker_pod");
 
   reporter->registerMetricExportType("key1", facebook::velox::StatType::COUNT);
   reporter->registerMetricExportType("key2", facebook::velox::StatType::AVG);
@@ -38,7 +38,7 @@ TEST_F(PrometheusStatsReporterTest, addStats) {
   EXPECT_EQ(
       facebook::velox::StatType::AVG, reporter->getRegisteredStatType("key2"));
 
-  std::unordered_set<size_t> testData = {10, 11, 15};
+  std::vector<size_t> testData = {10, 11, 15};
   for (auto i : testData) {
     reporter->addMetricValue("key1", i);
     reporter->addMetricValue("key2", i + 1000);
@@ -49,12 +49,10 @@ TEST_F(PrometheusStatsReporterTest, addStats) {
   const std::string expected[] = {
       "# HELP key2",
       "# TYPE key2 gauge",
-      "key2{cluster=\"test_cluster\",worker=\"test_worker_pod\"} 1011",
       "key2{cluster=\"test_cluster\",worker=\"test_worker_pod\"} 1015",
-      "key2{cluster=\"test_cluster\",worker=\"test_worker_pod\"} 1010",
       "# HELP key1",
       "# TYPE key1 counter",
-      "key1{cluster=\"test_cluster\",worker=\"test_worker_pod\"} 4"};
+      "key1{cluster=\"test_cluster\",worker=\"test_worker_pod\"} 37"};
 
   auto i = 0;
   auto pos = prometheusFormat.find("\n");

From c739d83941c486421fd298510b0c4d567592708c Mon Sep 17 00:00:00 2001
From: Karteekmurthys <karteek@ahana.io>
Date: Thu, 18 Jan 2024 16:20:30 -0800
Subject: [PATCH 4/6] Add new Serializer interface

---
 .../presto_cpp/main/PeriodicTaskManager.cpp           | 11 -----------
 .../presto_cpp/main/PrestoMain.cpp                    |  1 -
 .../presto_cpp/main/PrestoServer.cpp                  |  4 ++++
 .../presto_cpp/main/common/StatsReporterImpl.h        |  3 +++
 4 files changed, 7 insertions(+), 12 deletions(-)

diff --git a/presto-native-execution/presto_cpp/main/PeriodicTaskManager.cpp b/presto-native-execution/presto_cpp/main/PeriodicTaskManager.cpp
index 2b87118796e06..4f7bdb24961fd 100644
--- a/presto-native-execution/presto_cpp/main/PeriodicTaskManager.cpp
+++ b/presto-native-execution/presto_cpp/main/PeriodicTaskManager.cpp
@@ -284,17 +284,6 @@ void PeriodicTaskManager::updateCacheStats() {
   RECORD_METRIC_VALUE(
       kCounterMemoryCacheSumEvictScore, memoryCacheStats.sumEvictScore);
 
-  if (memoryCacheStats.ssdStats) {
-    RECORD_METRIC_VALUE(
-        kCounterSsdCacheCachedEntries,
-        memoryCacheStats.ssdStats->entriesCached);
-    RECORD_METRIC_VALUE(
-        kCounterSsdCacheCachedRegions,
-        memoryCacheStats.ssdStats->regionsCached);
-    RECORD_METRIC_VALUE(
-        kCounterSsdCacheCachedBytes, memoryCacheStats.ssdStats->bytesCached);
-  }
-
   // Interval cumulatives.
   RECORD_METRIC_VALUE(
       kCounterMemoryCacheNumHit,
diff --git a/presto-native-execution/presto_cpp/main/PrestoMain.cpp b/presto-native-execution/presto_cpp/main/PrestoMain.cpp
index 554677a83801d..1d70c4936b713 100644
--- a/presto-native-execution/presto_cpp/main/PrestoMain.cpp
+++ b/presto-native-execution/presto_cpp/main/PrestoMain.cpp
@@ -27,7 +27,6 @@ int main(int argc, char* argv[]) {
   google::InstallFailureSignalHandler();
   PRESTO_STARTUP_LOG(INFO) << "Entering main()";
   facebook::presto::PrestoServer presto(FLAGS_etc_dir);
-  facebook::velox::BaseStatsReporter::registered = true;
   presto.run();
   PRESTO_SHUTDOWN_LOG(INFO) << "Exiting main()";
 }
diff --git a/presto-native-execution/presto_cpp/main/PrestoServer.cpp b/presto-native-execution/presto_cpp/main/PrestoServer.cpp
index c5cbbc66b3ab6..18dbfbd846e4b 100644
--- a/presto-native-execution/presto_cpp/main/PrestoServer.cpp
+++ b/presto-native-execution/presto_cpp/main/PrestoServer.cpp
@@ -227,6 +227,10 @@ void PrestoServer::run() {
     exit(EXIT_FAILURE);
   }
 
+  if(systemConfig->enableRuntimeStatsCollection()) {
+    // This flag must be set to register the counters.
+    facebook::velox::BaseStatsReporter::registered = true;
+  }
   registerStatsCounters();
   registerFileSinks();
   registerFileSystems();
diff --git a/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.h b/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.h
index 042f85f215e20..df9160d12b898 100644
--- a/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.h
+++ b/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.h
@@ -22,6 +22,9 @@
 
 namespace facebook::presto {
 
+class MetricsSerializer {
+  facebook::velox::StringView getMetrics();
+};
 class StatsReporterImpl : public facebook::velox::BaseStatsReporter {
  public:
   StatsReporterImpl(

From 5c510e78a6e7146f9001b65ebe727f23dd2755b2 Mon Sep 17 00:00:00 2001
From: Karteekmurthys <karteek@ahana.io>
Date: Fri, 19 Jan 2024 11:38:34 -0800
Subject: [PATCH 5/6] Prometheus serializer

---
 .../presto_cpp/main/PrestoServer.cpp          | 10 +++-
 .../presto_cpp/main/common/Configs.cpp        |  3 +-
 .../main/common/StatsReporterImpl.cpp         | 25 +-------
 .../main/common/StatsReporterImpl.h           | 58 ++++++++++++++++++-
 .../main/tests/StatsReporterImplTest.cpp      |  6 +-
 5 files changed, 72 insertions(+), 30 deletions(-)

diff --git a/presto-native-execution/presto_cpp/main/PrestoServer.cpp b/presto-native-execution/presto_cpp/main/PrestoServer.cpp
index 18dbfbd846e4b..93dc89d40e784 100644
--- a/presto-native-execution/presto_cpp/main/PrestoServer.cpp
+++ b/presto-native-execution/presto_cpp/main/PrestoServer.cpp
@@ -227,7 +227,7 @@ void PrestoServer::run() {
     exit(EXIT_FAILURE);
   }
 
-  if(systemConfig->enableRuntimeStatsCollection()) {
+  if (systemConfig->enableRuntimeMetricsCollection()) {
     // This flag must be set to register the counters.
     facebook::velox::BaseStatsReporter::registered = true;
   }
@@ -1109,7 +1109,13 @@ void PrestoServer::reportServerInfo(proxygen::ResponseHandler* downstream) {
 void PrestoServer::reportHealthMetrics(proxygen::ResponseHandler* downstream) {
   auto reporter = std::dynamic_pointer_cast<StatsReporterImpl>(
       folly::Singleton<facebook::velox::BaseStatsReporter>::try_get());
-  http::sendOkResponse(downstream, reporter->getMetricsForPrometheus());
+  auto nodeConfig = facebook::presto::NodeConfig::instance();
+  std::string cluster = nodeConfig->nodeEnvironment();
+  char* hostName = std::getenv("HOSTNAME");
+  std::string worker = !hostName ? "" : hostName;
+  prometheus::PrometheusSerializer serializer(
+      prometheus::Labels{{"cluster", cluster}, {"worker", worker}});
+  http::sendOkResponse(downstream, reporter->getMetrics(serializer));
 }
 void PrestoServer::reportNodeStatus(proxygen::ResponseHandler* downstream) {
   http::sendOkResponse(downstream, json(fetchNodeStatus()));
diff --git a/presto-native-execution/presto_cpp/main/common/Configs.cpp b/presto-native-execution/presto_cpp/main/common/Configs.cpp
index 2ea962e840402..57ccb8b49cd42 100644
--- a/presto-native-execution/presto_cpp/main/common/Configs.cpp
+++ b/presto-native-execution/presto_cpp/main/common/Configs.cpp
@@ -602,7 +602,8 @@ std::chrono::duration<double> SystemConfig::cacheVeloxTtlCheckInterval() const {
 }
 
 bool SystemConfig::enableRuntimeMetricsCollection() const {
-  return optionalProperty<bool>(kEnableRuntimeMetricsCollection).value();
+  return optionalProperty<bool>(kEnableRuntimeMetricsCollection)
+      .value_or(false);
 }
 
 NodeConfig::NodeConfig() {
diff --git a/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.cpp b/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.cpp
index aaa39ce8447ba..f87b4f233559e 100644
--- a/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.cpp
+++ b/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.cpp
@@ -30,12 +30,6 @@ void StatsReporterImpl::registerMetricExportType(
   metricsMap_.emplace(key, 0);
 }
 
-const uint64_t getCurrentEpochTimestamp() {
-  auto p1 = std::chrono::system_clock::now();
-  return std::chrono::duration_cast<std::chrono::seconds>(p1.time_since_epoch())
-      .count();
-}
-
 void StatsReporterImpl::addMetricValue(const char* key, size_t value) const {
   std::lock_guard<std::mutex> lock(mutex_);
   auto it = registeredStats_.find(key);
@@ -62,23 +56,10 @@ void StatsReporterImpl::addMetricValue(folly::StringPiece key, size_t value)
   addMetricValue(key.start(), value);
 }
 
-const std::string StatsReporterImpl::getMetricsForPrometheus() {
+const std::string StatsReporterImpl::getMetrics(
+    const MetricsSerializer& serializer) {
   std::lock_guard<std::mutex> lock(mutex_);
-  std::stringstream ss;
-  for (const auto metric : metricsMap_) {
-    auto metricName = metric.first;
-    std::replace(metricName.begin(), metricName.end(), '.', '_');
-    auto statType = registeredStats_[metric.first];
-    ss << "# HELP " << metricName << std::endl;
-    std::string statTypeStr = "gauge";
-    if (statType == facebook::velox::StatType::COUNT) {
-      statTypeStr = "counter";
-    }
-    ss << "# TYPE " << metricName << " " << statTypeStr << std::endl;
-    ss << metricName << "{cluster=\"" << cluster_ << "\""
-       << ",worker=\"" << workerPod_ << "\"} " << metric.second << std::endl;
-  }
-  return ss.str();
+  return serializer.serialize(registeredStats_, metricsMap_);
 }
 
 // Initialize singleton for the reporter
diff --git a/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.h b/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.h
index df9160d12b898..c864fd38816e3 100644
--- a/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.h
+++ b/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.h
@@ -23,8 +23,60 @@
 namespace facebook::presto {
 
 class MetricsSerializer {
-  facebook::velox::StringView getMetrics();
+ public:
+  virtual std::string serialize(
+      const std::unordered_map<std::string, facebook::velox::StatType>&
+          metricStatTypes,
+      const std::unordered_map<std::string, size_t>& metricValues) const = 0;
 };
+
+namespace prometheus {
+using Labels = std::unordered_map<std::string, std::string>;
+class PrometheusSerializer : public MetricsSerializer {
+ public:
+  explicit PrometheusSerializer(const Labels& labels) : labels_(labels) {}
+
+  std::string serialize(
+      const std::unordered_map<std::string, facebook::velox::StatType>&
+          metricStatTypes,
+      const std::unordered_map<std::string, size_t>& metricValues) const {
+    std::stringstream ss;
+    for (const auto metric : metricValues) {
+      auto metricName = metric.first;
+      std::replace(metricName.begin(), metricName.end(), '.', '_');
+      auto statType = metricStatTypes.find(metric.first)->second;
+      ss << "# HELP " << metricName << std::endl;
+      std::string statTypeStr = "gauge";
+      if (statType == facebook::velox::StatType::COUNT) {
+        statTypeStr = "counter";
+      }
+      ss << "# TYPE " << metricName << " " << statTypeStr << std::endl;
+      int i = 0;
+      ss << metricName << "{";
+      for (auto& label : labels_) {
+        ss << label.first << "=\"" << label.second << "\"";
+        if (i < labels_.size() - 1) {
+          // Comma separated labels.
+          ss << ",";
+        }
+        ++i;
+      }
+      ss << "} " << metric.second << std::endl;
+    }
+    return ss.str();
+  }
+
+ private:
+  // A map of labels assigned to each metric which helps in filtering at client
+  // end.
+  const Labels labels_;
+};
+} // namespace prometheus.
+
+/// An implementation of BaseStatsReporter which gathers runtime metrics and
+/// and maintains them in-memory. Users can call
+/// StatsReporterImpl::getMetrics(MetricSerializer) to get metrics in custom
+/// formatted string.
 class StatsReporterImpl : public facebook::velox::BaseStatsReporter {
  public:
   StatsReporterImpl(
@@ -106,14 +158,14 @@ class StatsReporterImpl : public facebook::velox::BaseStatsReporter {
    * Above info is from:
    * https://prometheus.io/docs/concepts/data_model/#metric-names-and-labels
    */
-  const std::string getMetricsForPrometheus();
+  const std::string getMetrics(const MetricsSerializer& serializer);
 
  private:
   /// Mapping of registered stats key to StatType.
   mutable std::unordered_map<std::string, facebook::velox::StatType>
       registeredStats_;
   /// A mapping from stats key of type COUNT to value.
-  mutable std::unordered_map<std::string, int64_t> metricsMap_;
+  mutable std::unordered_map<std::string, size_t> metricsMap_;
   // Mutex to control access to registeredStats_ and metricMap_ members.
   mutable std::mutex mutex_;
   std::string cluster_;
diff --git a/presto-native-execution/presto_cpp/main/tests/StatsReporterImplTest.cpp b/presto-native-execution/presto_cpp/main/tests/StatsReporterImplTest.cpp
index 03735f9358aed..02ebb6f0a3547 100644
--- a/presto-native-execution/presto_cpp/main/tests/StatsReporterImplTest.cpp
+++ b/presto-native-execution/presto_cpp/main/tests/StatsReporterImplTest.cpp
@@ -25,7 +25,7 @@ class StatsReporterImplTest : public testing::Test {
 };
 
 /// Tests addStatType and addStats functions.
-TEST_F(StatsReporterImplTest, addStats) {
+TEST_F(StatsReporterImplTest, prometheusSerializer) {
   auto reporter =
       std::make_shared<StatsReporterImpl>("test_cluster", "test_worker_pod");
 
@@ -45,7 +45,9 @@ TEST_F(StatsReporterImplTest, addStats) {
   }
   // Uses default value of 1 for second parameter.
   reporter->addMetricValue("key1");
-  auto prometheusFormat = reporter->getMetricsForPrometheus();
+  prometheus::PrometheusSerializer serializer(prometheus::Labels{
+      {"cluster", "test_cluster"}, {"worker", "test_worker_pod"}});
+  auto prometheusFormat = reporter->getMetrics(serializer);
   const std::string expected[] = {
       "# HELP key2",
       "# TYPE key2 gauge",

From 3ce8d6cb85368d443391d4f7bda5343950fc3d02 Mon Sep 17 00:00:00 2001
From: Karteekmurthys <karteek@ahana.io>
Date: Mon, 29 Jan 2024 14:04:24 -0800
Subject: [PATCH 6/6] [Native] Prometheus exporter using prometheus-cpp client

---
 presto-native-execution/CMakeLists.txt        |   7 +
 .../presto_cpp/main/CMakeLists.txt            |   4 +
 .../presto_cpp/main/PrestoServer.cpp          |  31 +++-
 .../presto_cpp/main/common/CMakeLists.txt     |   4 +
 .../main/common/StatsReporterImpl.cpp         |   5 -
 .../main/common/StatsReporterImpl.h           |   9 +-
 .../common/prometheus-metrics/CMakeLists.txt  |   8 +
 .../prometheus-metrics/PrometheusReporter.cpp | 171 ++++++++++++++++++
 .../prometheus-metrics/PrometheusReporter.h   |  94 ++++++++++
 .../PrometheusReporterTest.cpp                | 154 ++++++++++++++++
 .../presto_cpp/main/tests/CMakeLists.txt      |   4 +-
 .../main/tests/StatsReporterImplTest.cpp      |   1 -
 .../scripts/setup-macos.sh                    |   8 +
 13 files changed, 484 insertions(+), 16 deletions(-)
 create mode 100644 presto-native-execution/presto_cpp/main/common/prometheus-metrics/CMakeLists.txt
 create mode 100644 presto-native-execution/presto_cpp/main/common/prometheus-metrics/PrometheusReporter.cpp
 create mode 100644 presto-native-execution/presto_cpp/main/common/prometheus-metrics/PrometheusReporter.h
 create mode 100644 presto-native-execution/presto_cpp/main/common/prometheus-metrics/PrometheusReporterTest.cpp

diff --git a/presto-native-execution/CMakeLists.txt b/presto-native-execution/CMakeLists.txt
index 15e5ed229b760..8caaa06c589f3 100644
--- a/presto-native-execution/CMakeLists.txt
+++ b/presto-native-execution/CMakeLists.txt
@@ -60,6 +60,9 @@ option(PRESTO_ENABLE_TESTING "Enable tests" ON)
 
 option(PRESTO_ENABLE_JWT "Enable JWT (JSON Web Token) authentication" OFF)
 
+option(PRESTO_ENABLE_PROMETHEUS_REPORTER
+       "Enables capturing of runtime metrics using prometheus client" OFF)
+
 # Set all Velox options below
 add_compile_definitions(FOLLY_HAVE_INT128_T=1)
 
@@ -201,6 +204,10 @@ if(PRESTO_ENABLE_JWT)
   add_compile_definitions(PRESTO_ENABLE_JWT)
 endif()
 
+if(PRESTO_ENABLE_PROMETHEUS_REPORTER)
+  add_compile_definitions(PRESTO_ENABLE_PROMETHEUS_REPORTER)
+endif()
+
 if("${MAX_LINK_JOBS}")
   set_property(GLOBAL APPEND PROPERTY JOB_POOLS
                                       "presto_link_job_pool=${MAX_LINK_JOBS}")
diff --git a/presto-native-execution/presto_cpp/main/CMakeLists.txt b/presto-native-execution/presto_cpp/main/CMakeLists.txt
index 0a19fc67e0510..f4caf2ac69500 100644
--- a/presto-native-execution/presto_cpp/main/CMakeLists.txt
+++ b/presto-native-execution/presto_cpp/main/CMakeLists.txt
@@ -103,3 +103,7 @@ set_property(TARGET presto_server PROPERTY JOB_POOL_LINK presto_link_job_pool)
 if(PRESTO_ENABLE_TESTING)
   add_subdirectory(tests)
 endif()
+if(PRESTO_ENABLE_PROMETHEUS_REPORTER)
+  message(STATUS "Linking prometheus metrics reporter")
+  target_link_libraries(presto_server_lib prometheus_reporter)
+endif()
diff --git a/presto-native-execution/presto_cpp/main/PrestoServer.cpp b/presto-native-execution/presto_cpp/main/PrestoServer.cpp
index 93dc89d40e784..5f6840d478adb 100644
--- a/presto-native-execution/presto_cpp/main/PrestoServer.cpp
+++ b/presto-native-execution/presto_cpp/main/PrestoServer.cpp
@@ -24,7 +24,6 @@
 #include "presto_cpp/main/TaskResource.h"
 #include "presto_cpp/main/common/ConfigReader.h"
 #include "presto_cpp/main/common/Counters.h"
-#include "presto_cpp/main/common/StatsReporterImpl.h"
 #include "presto_cpp/main/common/Utils.h"
 #include "presto_cpp/main/http/filters/AccessLogFilter.h"
 #include "presto_cpp/main/http/filters/HttpEndpointLatencyFilter.h"
@@ -53,6 +52,25 @@
 #include "velox/functions/prestosql/window/WindowFunctionsRegistration.h"
 #include "velox/serializers/PrestoSerializer.h"
 
+#ifdef PRESTO_ENABLE_PROMETHEUS_REPORTER
+#include "presto_cpp/main/common/prometheus-metrics/PrometheusReporter.h"
+// Initialize singleton for the reporter
+folly::Singleton<facebook::velox::BaseStatsReporter> reporter([]() {
+  auto nodeConfig = facebook::presto::NodeConfig::instance();
+  std::string cluster = nodeConfig->nodeEnvironment();
+  char* hostName = std::getenv("HOSTNAME");
+  std::string worker = !hostName ? "" : hostName;
+  return new facebook::presto::prometheus::PrometheusReporter(
+      ::prometheus::Labels{{"cluster", cluster}, {"worker", worker}});
+});
+#else
+#include "presto_cpp/main/common/StatsReporterImpl.h"
+// Initialize singleton for the reporter
+folly::Singleton<facebook::velox::BaseStatsReporter> reporter([]() {
+  return new facebook::presto::StatsReporterImpl();
+});
+#endif
+
 #ifdef PRESTO_ENABLE_REMOTE_FUNCTIONS
 #include "presto_cpp/main/RemoteFunctionRegisterer.h"
 #endif
@@ -1107,15 +1125,22 @@ void PrestoServer::reportServerInfo(proxygen::ResponseHandler* downstream) {
 }
 
 void PrestoServer::reportHealthMetrics(proxygen::ResponseHandler* downstream) {
-  auto reporter = std::dynamic_pointer_cast<StatsReporterImpl>(
-      folly::Singleton<facebook::velox::BaseStatsReporter>::try_get());
   auto nodeConfig = facebook::presto::NodeConfig::instance();
   std::string cluster = nodeConfig->nodeEnvironment();
   char* hostName = std::getenv("HOSTNAME");
   std::string worker = !hostName ? "" : hostName;
+#ifdef PRESTO_ENABLE_PROMETHEUS_REPORTER
+  auto reporter = std::dynamic_pointer_cast<
+      facebook::presto::prometheus::PrometheusReporter>(
+      folly::Singleton<facebook::velox::BaseStatsReporter>::try_get());
+  http::sendOkResponse(downstream, reporter->getSerializedMetrics());
+#else
+  auto reporter = std::dynamic_pointer_cast<StatsReporterImpl>(
+      folly::Singleton<facebook::velox::BaseStatsReporter>::try_get());
   prometheus::PrometheusSerializer serializer(
       prometheus::Labels{{"cluster", cluster}, {"worker", worker}});
   http::sendOkResponse(downstream, reporter->getMetrics(serializer));
+#endif
 }
 void PrestoServer::reportNodeStatus(proxygen::ResponseHandler* downstream) {
   http::sendOkResponse(downstream, json(fetchNodeStatus()));
diff --git a/presto-native-execution/presto_cpp/main/common/CMakeLists.txt b/presto-native-execution/presto_cpp/main/common/CMakeLists.txt
index af616052ab84e..05d2422594948 100644
--- a/presto-native-execution/presto_cpp/main/common/CMakeLists.txt
+++ b/presto-native-execution/presto_cpp/main/common/CMakeLists.txt
@@ -24,3 +24,7 @@ set_property(TARGET presto_common PROPERTY JOB_POOL_LINK presto_link_job_pool)
 if(PRESTO_ENABLE_TESTING)
   add_subdirectory(tests)
 endif()
+
+if(PRESTO_ENABLE_PROMETHEUS_REPORTER)
+  add_subdirectory(prometheus-metrics)
+endif()
diff --git a/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.cpp b/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.cpp
index f87b4f233559e..4acd897f7daa6 100644
--- a/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.cpp
+++ b/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.cpp
@@ -61,9 +61,4 @@ const std::string StatsReporterImpl::getMetrics(
   std::lock_guard<std::mutex> lock(mutex_);
   return serializer.serialize(registeredStats_, metricsMap_);
 }
-
-// Initialize singleton for the reporter
-folly::Singleton<facebook::velox::BaseStatsReporter> reporter([]() {
-  return new StatsReporterImpl();
-});
 } // namespace facebook::presto
diff --git a/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.h b/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.h
index c864fd38816e3..64b4f2d016928 100644
--- a/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.h
+++ b/presto-native-execution/presto_cpp/main/common/StatsReporterImpl.h
@@ -17,8 +17,8 @@
 #include <iostream>
 #include "presto_cpp/main/common/Configs.h"
 #include "presto_cpp/main/common/Counters.h"
+#include "velox/common/base/Exceptions.h"
 #include "velox/common/base/StatsReporter.h"
-#include "velox/velox/common/base/Exceptions.h"
 
 namespace facebook::presto {
 
@@ -123,13 +123,12 @@ class StatsReporterImpl : public facebook::velox::BaseStatsReporter {
 
   void addMetricValue(folly::StringPiece key, size_t value = 1) const override;
 
-  virtual void addHistogramMetricValue(const std::string& key, size_t value)
+  void addHistogramMetricValue(const std::string& key, size_t value)
       const override {}
 
-  virtual void addHistogramMetricValue(const char* key, size_t value)
-      const override {}
+  void addHistogramMetricValue(const char* key, size_t value) const override {}
 
-  virtual void addHistogramMetricValue(folly::StringPiece key, size_t value)
+  void addHistogramMetricValue(folly::StringPiece key, size_t value)
       const override {}
 
   const facebook::velox::StatType getRegisteredStatType(
diff --git a/presto-native-execution/presto_cpp/main/common/prometheus-metrics/CMakeLists.txt b/presto-native-execution/presto_cpp/main/common/prometheus-metrics/CMakeLists.txt
new file mode 100644
index 0000000000000..e56a4932dcc98
--- /dev/null
+++ b/presto-native-execution/presto_cpp/main/common/prometheus-metrics/CMakeLists.txt
@@ -0,0 +1,8 @@
+find_package(prometheus-cpp CONFIG REQUIRED)
+add_library(prometheus_reporter PrometheusReporter.cpp PrometheusReporter.h)
+target_link_libraries(prometheus_reporter presto_common prometheus-cpp::core)
+set_property(TARGET prometheus_reporter PROPERTY JOB_POOL_LINK
+                                                 presto_link_job_pool)
+add_executable(prometheus_reporter_test PrometheusReporterTest.cpp)
+target_link_libraries(prometheus_reporter_test presto_server_lib
+                      velox_exec_test_lib prometheus_reporter gtest gtest_main)
diff --git a/presto-native-execution/presto_cpp/main/common/prometheus-metrics/PrometheusReporter.cpp b/presto-native-execution/presto_cpp/main/common/prometheus-metrics/PrometheusReporter.cpp
new file mode 100644
index 0000000000000..03ee50784f9d5
--- /dev/null
+++ b/presto-native-execution/presto_cpp/main/common/prometheus-metrics/PrometheusReporter.cpp
@@ -0,0 +1,171 @@
+/*
+ * Licensed 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.
+ */
+
+#include "PrometheusReporter.h"
+namespace facebook::presto::prometheus {
+void PrometheusReporter::registerMetricExportType(
+    const char* key,
+    facebook::velox::StatType statType) const {
+  if (registeredMetrics_.count(key)) {
+    // Already registered;
+    VLOG(1) << "Trying to register already registered metric " << key;
+    return;
+  }
+  // Prometheus format requires '.' to be replaced.
+  std::string keyStr = std::string(key);
+  std::replace(keyStr.begin(), keyStr.end(), '.', '_');
+  switch (statType) {
+    case facebook::velox::StatType::COUNT: {
+      auto& counterFamily =
+          ::prometheus::BuildCounter().Name(keyStr).Register(*registry_);
+      auto& counter = counterFamily.Add(labels_);
+      countersMap_.emplace(std::string(key), counter);
+      break;
+    }
+    case facebook::velox::StatType::SUM:
+    case facebook::velox::StatType::AVG:
+    case facebook::velox::StatType::RATE: {
+      auto& gaugeFamily =
+          ::prometheus::BuildGauge().Name(keyStr).Register(*registry_);
+      auto& gauge = gaugeFamily.Add(labels_);
+      gaugeMap_.emplace(std::string(key), gauge);
+      break;
+    }
+    default:
+      VELOX_UNSUPPORTED(
+          "Unsupported metric type {}", std::to_string((int)statType));
+  }
+  registeredMetrics_.emplace(key, statType);
+}
+
+void PrometheusReporter::registerMetricExportType(
+    folly::StringPiece key,
+    facebook::velox::StatType statType) const {
+  registerMetricExportType(key.start(), statType);
+}
+
+void PrometheusReporter::registerHistogramMetricExportType(
+    const char* key,
+    int64_t bucketWidth,
+    int64_t min,
+    int64_t max,
+    const std::vector<int32_t>& pcts) const {
+  std::lock_guard<std::mutex> l(mutex_);
+  if (registeredHistogramMetrics_.count(key)) {
+    // Already registered;
+    VLOG(1) << "Trying to register already registered metric " << key;
+    return;
+  }
+  int numBuckets = (max - min) / bucketWidth;
+  auto bound = min + bucketWidth;
+  std::string keyStr = std::string(key);
+  std::replace(keyStr.begin(), keyStr.end(), '.', '_');
+  auto& histogramFamily =
+      ::prometheus::BuildHistogram().Name(keyStr).Register(*registry_);
+  ::prometheus::Histogram::BucketBoundaries bucketBoundaries;
+  while (numBuckets) {
+    bucketBoundaries.push_back(bound);
+    bound += bucketWidth;
+    numBuckets--;
+  }
+  VELOX_CHECK_GE(bucketBoundaries.size(), 1);
+  auto& histogramMetric = histogramFamily.Add(labels_, bucketBoundaries);
+  histogramMap_.emplace(key, histogramMetric);
+  // If percentiles are provided, create a Summary type metric and register.
+  if (pcts.size() > 0) {
+    auto& summaryFamily = ::prometheus::BuildSummary()
+                              .Name(keyStr.append(kSummarySuffix))
+                              .Register(*registry_);
+    ::prometheus::Summary::Quantiles quantiles;
+    for (auto pct : pcts) {
+      quantiles.push_back(
+          ::prometheus::detail::CKMSQuantiles::Quantile(pct / (double)100, 0));
+    }
+    auto& summaryMetric = summaryFamily.Add({labels_}, quantiles);
+    summaryMap_.emplace(std::string(key), summaryMetric);
+  }
+  registeredHistogramMetrics_.insert(key);
+}
+
+void PrometheusReporter::registerHistogramMetricExportType(
+    folly::StringPiece key,
+    int64_t bucketWidth,
+    int64_t min,
+    int64_t max,
+    const std::vector<int32_t>& pcts) const {
+  registerHistogramMetricExportType(key.begin(), bucketWidth, min, max, pcts);
+}
+
+void PrometheusReporter::addMetricValue(const std::string& key, size_t value)
+    const {
+  addMetricValue(key.c_str(), value);
+}
+
+void PrometheusReporter::addMetricValue(const char* key, size_t value) const {
+  std::lock_guard<std::mutex> l(mutex_);
+  if (!registeredMetrics_.count(key)) {
+    VLOG(1) << "addMetricValue for unregistred metric " << key;
+    return;
+  }
+  auto statType = registeredMetrics_.find(key)->second;
+  switch (statType) {
+    case velox::StatType::COUNT:
+      countersMap_.find(key)->second.Increment(value);
+      break;
+    case velox::StatType::SUM:
+    case velox::StatType::AVG:
+    case velox::StatType::RATE:
+      gaugeMap_.find(key)->second.Set(value);
+      break;
+  };
+}
+
+void PrometheusReporter::addMetricValue(folly::StringPiece key, size_t value)
+    const {
+  addMetricValue(key.begin(), value);
+}
+
+void PrometheusReporter::addHistogramMetricValue(
+    const std::string& key,
+    size_t value) const {
+  addHistogramMetricValue(key.c_str(), value);
+}
+
+void PrometheusReporter::addHistogramMetricValue(const char* key, size_t value)
+    const {
+  std::lock_guard<std::mutex> l(mutex_);
+  if (!registeredHistogramMetrics_.count(key)) {
+    VLOG(1) << "addMetricValue for unregistered metric " << key;
+    return;
+  }
+  histogramMap_.find(key)->second.Observe(value);
+  if (summaryMap_.count(key)) {
+    summaryMap_.find(key)->second.Observe(value);
+  }
+}
+
+void PrometheusReporter::addHistogramMetricValue(
+    folly::StringPiece key,
+    size_t value) const {
+  addHistogramMetricValue(key.begin(), value);
+}
+
+std::string PrometheusReporter::getSerializedMetrics() const {
+  if (registeredMetrics_.empty() && registeredHistogramMetrics_.empty()) {
+    return "";
+  }
+  ::prometheus::TextSerializer serializer;
+  return serializer.Serialize(registry_->Collect());
+}
+}; // namespace facebook::presto::prometheus
\ No newline at end of file
diff --git a/presto-native-execution/presto_cpp/main/common/prometheus-metrics/PrometheusReporter.h b/presto-native-execution/presto_cpp/main/common/prometheus-metrics/PrometheusReporter.h
new file mode 100644
index 0000000000000..e26e325613f3b
--- /dev/null
+++ b/presto-native-execution/presto_cpp/main/common/prometheus-metrics/PrometheusReporter.h
@@ -0,0 +1,94 @@
+/*
+ * Licensed 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.
+ */
+#include <prometheus/registry.h>
+#include "presto_cpp/main/common/StatsReporterImpl.h"
+#include "prometheus/collectable.h"
+#include "prometheus/counter.h"
+#include "prometheus/gauge.h"
+#include "prometheus/histogram.h"
+#include "prometheus/summary.h"
+#include "prometheus/text_serializer.h"
+#include "velox/common/base/Exceptions.h"
+#include "velox/velox/common/base/GTestMacros.h"
+
+namespace facebook::presto::prometheus {
+using CounterMap = std::unordered_map<std::string, ::prometheus::Counter&>;
+using GaugeMap = std::unordered_map<std::string, ::prometheus::Gauge&>;
+using HistogramMap = std::unordered_map<std::string, ::prometheus::Histogram&>;
+using SummaryMap = std::unordered_map<std::string, ::prometheus::Summary&>;
+
+static constexpr std::string_view kSummarySuffix("_summary");
+
+class PrometheusReporter : public facebook::velox::BaseStatsReporter {
+ public:
+  explicit PrometheusReporter(const ::prometheus::Labels& labels)
+      : labels_(labels) {
+    registry_ = std::make_shared<::prometheus::Registry>();
+  }
+  void registerMetricExportType(const char* key, velox::StatType)
+      const override;
+
+  void registerMetricExportType(folly::StringPiece key, velox::StatType)
+      const override;
+
+  void registerHistogramMetricExportType(
+      const char* key,
+      int64_t bucketWidth,
+      int64_t min,
+      int64_t max,
+      const std::vector<int32_t>& pcts) const override;
+
+  void registerHistogramMetricExportType(
+      folly::StringPiece key,
+      int64_t bucketWidth,
+      int64_t min,
+      int64_t max,
+      const std::vector<int32_t>& pcts) const override;
+
+  void addMetricValue(const std::string& key, size_t value = 1) const override;
+
+  void addMetricValue(const char* key, size_t value = 1) const override;
+
+  void addMetricValue(folly::StringPiece key, size_t value = 1) const override;
+
+  void addHistogramMetricValue(const std::string& key, size_t value)
+      const override;
+
+  void addHistogramMetricValue(const char* key, size_t value) const override;
+
+  void addHistogramMetricValue(folly::StringPiece key, size_t value)
+      const override;
+
+  std::string getSerializedMetrics() const;
+
+ private:
+  mutable std::mutex mutex_;
+  // A map of labels assigned to each metric which helps in filtering at client
+  // end.
+  const ::prometheus::Labels labels_;
+  std::shared_ptr<::prometheus::Registry> registry_;
+  // Maintaining maps for each metric type is for convenience,
+  // the ::prometheus::Registry allows us to Add new metric object each time
+  // which overwrites the existing metric object in its internal mapping.
+  mutable CounterMap countersMap_;
+  mutable GaugeMap gaugeMap_;
+  mutable HistogramMap histogramMap_;
+  mutable SummaryMap summaryMap_;
+  mutable std::unordered_map<std::string, ::facebook::velox::StatType>
+      registeredMetrics_;
+  // @TODO: may be add StatType::HISTOGRAM to avoid this member.
+  mutable std::unordered_set<std::string> registeredHistogramMetrics_;
+  VELOX_FRIEND_TEST(PrometheusReporterTest, testAllMetrics);
+}; // class PrometheusReporter
+}; // namespace facebook::presto::prometheus
\ No newline at end of file
diff --git a/presto-native-execution/presto_cpp/main/common/prometheus-metrics/PrometheusReporterTest.cpp b/presto-native-execution/presto_cpp/main/common/prometheus-metrics/PrometheusReporterTest.cpp
new file mode 100644
index 0000000000000..072c67093ce25
--- /dev/null
+++ b/presto-native-execution/presto_cpp/main/common/prometheus-metrics/PrometheusReporterTest.cpp
@@ -0,0 +1,154 @@
+/*
+ * Licensed 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.
+ */
+
+#include "PrometheusReporter.h"
+#include <gtest/gtest.h>
+
+namespace facebook::presto::prometheus {
+class PrometheusReporterTest : public testing::Test {
+  void SetUp() override {}
+
+  void TearDown() override {}
+};
+
+TEST_F(PrometheusReporterTest, testAllMetrics) {
+  ::prometheus::Labels testLabels(
+      {{"cluster", "test_cluster"}, {"worker", "test_worker_pod"}});
+  std::string labelsSerialized =
+      ("cluster=\"test_cluster\",worker=\"test_worker_pod\"");
+  auto reporter = std::make_shared<PrometheusReporter>(testLabels);
+  reporter->registerMetricExportType(
+      "test.key1", facebook::velox::StatType::COUNT);
+  reporter->registerMetricExportType(
+      "test.key2", facebook::velox::StatType::AVG);
+  reporter->registerMetricExportType(
+      "test.key3", facebook::velox::StatType::SUM);
+  reporter->registerMetricExportType(
+      "test.key4", facebook::velox::StatType::RATE);
+  EXPECT_EQ(
+      facebook::velox::StatType::COUNT,
+      reporter->registeredMetrics_.find("test.key1")->second);
+  EXPECT_EQ(
+      facebook::velox::StatType::AVG,
+      reporter->registeredMetrics_.find("test.key2")->second);
+  EXPECT_EQ(
+      facebook::velox::StatType::SUM,
+      reporter->registeredMetrics_.find("test.key3")->second);
+  EXPECT_EQ(
+      facebook::velox::StatType::RATE,
+      reporter->registeredMetrics_.find("test.key4")->second);
+
+  std::vector<size_t> testData = {10, 11, 15};
+  for (auto i : testData) {
+    reporter->addMetricValue("test.key1", i);
+    reporter->addMetricValue("test.key2", i + 1000);
+  }
+  // Uses default value of 1 for second parameter.
+  reporter->addMetricValue("test.key1");
+  prometheus::PrometheusSerializer serializer(prometheus::Labels{
+      {"cluster", "test_cluster"}, {"worker", "test_worker_pod"}});
+  auto fullSerializedResult = reporter->getSerializedMetrics();
+  std::vector<std::string> expected = {
+      "# TYPE test_key1 counter",
+      "test_key1{" + labelsSerialized + "} 37",
+      "# TYPE test_key2 gauge",
+      "test_key2{" + labelsSerialized + "} 1015",
+      "# TYPE test_key3 gauge",
+      "test_key3{" + labelsSerialized + "} 0",
+      "# TYPE test_key4 gauge",
+      "test_key4{" + labelsSerialized + "} 0"};
+
+  auto verifySerializedResult = [&](std::string& fullSerializedResult,
+                                    std::vector<std::string>& expected) {
+    auto i = 0;
+    auto pos = fullSerializedResult.find("\n");
+    while (pos != std::string::npos) {
+      std::string line = fullSerializedResult.substr(0, pos);
+      EXPECT_EQ(line, expected[i++]);
+      fullSerializedResult = fullSerializedResult.substr(pos + 1);
+      pos = fullSerializedResult.find("\n");
+    }
+  };
+  verifySerializedResult(fullSerializedResult, expected);
+  std::string histSummaryKey = "test.histogram.key1";
+  std::string histogramKey = "test.histogram.key2";
+  // Register Histograms and Summaries.
+  reporter->registerHistogramMetricExportType(
+      histSummaryKey, 10, 0, 100, {50, 99, 100});
+  // Only histogram.
+  reporter->registerHistogramMetricExportType(histogramKey, 10, 0, 100, {});
+  int recordCount = 100;
+  int sum = 0;
+  for (int i = 0; i < recordCount; ++i) {
+    if (i < 20) {
+      reporter->addHistogramMetricValue(histSummaryKey, 20);
+      sum += 20;
+    } else if (i >= 20 && i < 50) {
+      reporter->addHistogramMetricValue(histSummaryKey, 50);
+      sum += 50;
+    } else {
+      reporter->addHistogramMetricValue(histSummaryKey, 85);
+      sum += 85;
+    }
+  }
+  reporter->addHistogramMetricValue(histogramKey, 10);
+  fullSerializedResult = reporter->getSerializedMetrics();
+  std::replace(histSummaryKey.begin(), histSummaryKey.end(), '.', '_');
+  std::replace(histogramKey.begin(), histogramKey.end(), '.', '_');
+  std::vector<std::string> histogramMetricsFormatted = {
+      "# TYPE " + histSummaryKey + " histogram",
+      histSummaryKey + "_count{" + labelsSerialized + "} " +
+          std::to_string(recordCount),
+      histSummaryKey + "_sum{" + labelsSerialized + "} " + std::to_string(sum),
+      histSummaryKey + "_bucket{" + labelsSerialized + ",le=\"10\"} 0",
+      histSummaryKey + "_bucket{" + labelsSerialized + ",le=\"20\"} 20",
+      histSummaryKey + "_bucket{" + labelsSerialized + ",le=\"30\"} 20",
+      histSummaryKey + "_bucket{" + labelsSerialized + ",le=\"40\"} 20",
+      histSummaryKey + "_bucket{" + labelsSerialized + ",le=\"50\"} 50",
+      histSummaryKey + "_bucket{" + labelsSerialized + ",le=\"60\"} 50",
+      histSummaryKey + "_bucket{" + labelsSerialized + ",le=\"70\"} 50",
+      histSummaryKey + "_bucket{" + labelsSerialized + ",le=\"80\"} 50",
+      histSummaryKey + "_bucket{" + labelsSerialized + ",le=\"90\"} 100",
+      histSummaryKey + "_bucket{" + labelsSerialized + ",le=\"100\"} 100",
+      histSummaryKey + "_bucket{" + labelsSerialized + ",le=\"+Inf\"} 100",
+      "# TYPE test_histogram_key2 histogram",
+      histogramKey + "_count{" + labelsSerialized + "} 1",
+      histogramKey + "_sum{" + labelsSerialized + "} 10",
+      histogramKey + "_bucket{" + labelsSerialized + ",le=\"10\"} 1",
+      histogramKey + "_bucket{" + labelsSerialized + ",le=\"20\"} 1",
+      histogramKey + "_bucket{" + labelsSerialized + ",le=\"30\"} 1",
+      histogramKey + "_bucket{" + labelsSerialized + ",le=\"40\"} 1",
+      histogramKey + "_bucket{" + labelsSerialized + ",le=\"50\"} 1",
+      histogramKey + "_bucket{" + labelsSerialized + ",le=\"60\"} 1",
+      histogramKey + "_bucket{" + labelsSerialized + ",le=\"70\"} 1",
+      histogramKey + "_bucket{" + labelsSerialized + ",le=\"80\"} 1",
+      histogramKey + "_bucket{" + labelsSerialized + ",le=\"90\"} 1",
+      histogramKey + "_bucket{" + labelsSerialized + ",le=\"100\"} 1",
+      histogramKey + "_bucket{" + labelsSerialized + ",le=\"+Inf\"} 1",
+      "# TYPE test_histogram_key1_summary summary",
+      histSummaryKey + "_summary_count{" + labelsSerialized + "} " +
+          std::to_string(recordCount),
+      histSummaryKey + "_summary_sum{" + labelsSerialized + "} " +
+          std::to_string(sum),
+      histSummaryKey + "_summary{" + labelsSerialized + ",quantile=\"0.5\"} 50",
+      histSummaryKey + "_summary{" + labelsSerialized +
+          ",quantile=\"0.99\"} 85",
+      histSummaryKey + "_summary{" + labelsSerialized + ",quantile=\"1\"} 85"};
+  expected.insert(
+      expected.end(),
+      histogramMetricsFormatted.begin(),
+      histogramMetricsFormatted.end());
+  verifySerializedResult(fullSerializedResult, expected);
+}
+}; // namespace facebook::presto::prometheus
diff --git a/presto-native-execution/presto_cpp/main/tests/CMakeLists.txt b/presto-native-execution/presto_cpp/main/tests/CMakeLists.txt
index c2b9f81deacc1..db562930f651f 100644
--- a/presto-native-execution/presto_cpp/main/tests/CMakeLists.txt
+++ b/presto-native-execution/presto_cpp/main/tests/CMakeLists.txt
@@ -27,9 +27,9 @@ add_test(
   COMMAND presto_server_test
   WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR})
 
-add_executable(ahana_stats_test StatsReporterImplTest.cpp)
+add_executable(stats_reporter_test StatsReporterImplTest.cpp)
 
-target_link_libraries(ahana_stats_test presto_server_lib velox_exec_test_lib
+target_link_libraries(stats_reporter_test presto_server_lib velox_exec_test_lib
                       gtest gtest_main)
 
 target_link_libraries(
diff --git a/presto-native-execution/presto_cpp/main/tests/StatsReporterImplTest.cpp b/presto-native-execution/presto_cpp/main/tests/StatsReporterImplTest.cpp
index 02ebb6f0a3547..916cf6740a5f9 100644
--- a/presto-native-execution/presto_cpp/main/tests/StatsReporterImplTest.cpp
+++ b/presto-native-execution/presto_cpp/main/tests/StatsReporterImplTest.cpp
@@ -14,7 +14,6 @@
 
 #include "presto_cpp/main/common/StatsReporterImpl.h"
 #include <gtest/gtest.h>
-#include <cstring>
 
 namespace facebook::presto {
 
diff --git a/presto-native-execution/scripts/setup-macos.sh b/presto-native-execution/scripts/setup-macos.sh
index 278835d6cb832..0e95674866364 100755
--- a/presto-native-execution/scripts/setup-macos.sh
+++ b/presto-native-execution/scripts/setup-macos.sh
@@ -26,6 +26,13 @@ function install_six {
 
 export PATH=$(brew --prefix bison)/bin:$PATH
 
+function install_prometheus {
+  git clone https://github.com/jupp0r/prometheus-cpp.git
+  cd prometheus-cpp
+  git submodule init
+  git submodule update
+  cmake_install -DBUILD_SHARED_LIBS=ON -DENABLE_PUSH=OFF -DENABLE_COMPRESSION=OFF
+}
 
 function install_proxygen {
   github_checkout facebook/proxygen "${FB_OS_VERSION}"
@@ -36,6 +43,7 @@ function install_presto_deps {
   install_velox_deps
   run_and_time install_six
   run_and_time install_proxygen
+  run_and_time install_prometheus
 }
 
 if [[ $# -ne 0 ]]; then