Expose REST API to fetch runtime metrics in prometheus format

Co-authored-by:jaystarshot <jay.narale@uber.com>

presto-native-execution/etc/config.properties

@@ -2,4 +2,6 @@ discovery.uri=http://127.0.0.1:58215
 presto.version=testversion
 http-server.http.port=7777
 shutdown-onset-sec=1
-register-test-functions=true
+register-test-functions=true
+runtime-metrics-collection-enabled=true
+runtime-metrics-reporter=prometheus

presto-native-execution/presto_cpp/main/CMakeLists.txt

@@ -47,6 +47,7 @@ target_link_libraries(
   presto_common
   presto_exception
   presto_http
+  presto_runtime_metrics
   presto_operators
   velox_aggregates
   velox_caching

presto-native-execution/presto_cpp/main/PrestoMain.cpp

@@ -17,7 +17,6 @@
 #include <glog/logging.h>
 #include "presto_cpp/main/PrestoServer.h"
 #include "presto_cpp/main/common/Utils.h"
-#include "velox/common/base/StatsReporter.h"
 
 DEFINE_string(etc_dir, ".", "etc directory for presto configuration");
 
@@ -30,8 +29,3 @@ int main(int argc, char* argv[]) {
   presto.run();
   PRESTO_SHUTDOWN_LOG(INFO) << "Exiting main()";
 }
-
-// Initialize singleton for the reporter.
-folly::Singleton<facebook::velox::BaseStatsReporter> reporter([]() {
-  return new facebook::velox::DummyStatsReporter();
-});

presto-native-execution/presto_cpp/main/PrestoServer.cpp

@@ -26,6 +26,7 @@
 #include "presto_cpp/main/common/ConfigReader.h"
 #include "presto_cpp/main/common/Counters.h"
 #include "presto_cpp/main/common/Utils.h"
+#include "presto_cpp/main/common/runtime-metrics/PrestoRuntimeMetricsReporter.h"
 #include "presto_cpp/main/http/filters/AccessLogFilter.h"
 #include "presto_cpp/main/http/filters/HttpEndpointLatencyFilter.h"
 #include "presto_cpp/main/http/filters/InternalAuthenticationFilter.h"
@@ -36,7 +37,6 @@
 #include "presto_cpp/main/operators/PartitionAndSerialize.h"
 #include "presto_cpp/main/operators/ShuffleRead.h"
 #include "presto_cpp/main/operators/UnsafeRowExchangeSource.h"
-#include "presto_cpp/main/types/PrestoToVeloxConnector.h"
 #include "presto_cpp/main/types/PrestoToVeloxQueryPlan.h"
 #include "velox/common/base/Counters.h"
 #include "velox/common/base/StatsReporter.h"
@@ -335,6 +335,18 @@ void PrestoServer::run() {
           proxygen::ResponseHandler* downstream) {
         server->reportServerInfo(downstream);
       });
+  if (systemConfig->enableRuntimeMetricsCollection()) {
+    httpServer_->registerGet(
+        "/v1/info/metrics",
+        [](proxygen::HTTPMessage* /*message*/,
+           const std::vector<std::unique_ptr<folly::IOBuf>>& /*body*/,
+           proxygen::ResponseHandler* downstream) {
+          auto reporter = std::dynamic_pointer_cast<
+              facebook::presto::prometheus::PrometheusReporter>(
+              folly::Singleton<facebook::velox::BaseStatsReporter>::try_get());
+          http::sendOkResponse(downstream, reporter->getSerializedMetrics());
+        });
+  }
   httpServer_->registerGet(
       "/v1/info/state",
       [server = this](
@@ -1104,8 +1116,7 @@ std::string PrestoServer::getBaseSpillDirectory() const {
 }
 
 void PrestoServer::enableRuntimeMetricReporting() {
-  // This flag must be set to register the counters.
-  facebook::velox::BaseStatsReporter::registered = true;
+  registerRuntimeMetricsReporters();
   registerStatsCounters();
 }
 
@@ -1134,7 +1145,6 @@ void PrestoServer::populateMemAndCPUInfo() {
     poolInfo.queryMemoryAllocations.insert(
         {queryId, {protocol::MemoryAllocation{"total", bytes}}});
     ++numContexts;
-    poolInfo.reservedBytes += bytes;
   });
   RECORD_METRIC_VALUE(kCounterNumQueryContexts, numContexts);
   cpuMon_.update();
@@ -1173,6 +1183,7 @@ void PrestoServer::reportServerInfo(proxygen::ResponseHandler* downstream) {
   http::sendOkResponse(downstream, json(serverInfo));
 }
 
+void PrestoServer::reportWorkerMetrics(proxygen::ResponseHandler* downstream) {}
 void PrestoServer::reportNodeStatus(proxygen::ResponseHandler* downstream) {
   http::sendOkResponse(downstream, json(fetchNodeStatus()));
 }

presto-native-execution/presto_cpp/main/PrestoServer.h

@@ -193,6 +193,8 @@ class PrestoServer {
 
   void reportNodeStatus(proxygen::ResponseHandler* downstream);
 
+  void reportWorkerMetrics(proxygen::ResponseHandler* downstream);
+
   protocol::NodeStatus fetchNodeStatus();
 
   void populateMemAndCPUInfo();

presto-native-execution/presto_cpp/main/common/CMakeLists.txt

@@ -20,6 +20,6 @@ set_property(TARGET presto_exception PROPERTY JOB_POOL_LINK
 target_link_libraries(presto_common velox_config velox_core velox_exception)
 set_property(TARGET presto_common PROPERTY JOB_POOL_LINK presto_link_job_pool)
 
-if(PRESTO_ENABLE_TESTING)
-  add_subdirectory(tests)
-endif()
+add_subdirectory(runtime-metrics)
+
+add_subdirectory(tests)

presto-native-execution/presto_cpp/main/common/Configs.cpp

@@ -224,7 +224,7 @@ SystemConfig::SystemConfig() {
           STR_PROP(kCacheVeloxTtlThreshold, "2d"),
           STR_PROP(kCacheVeloxTtlCheckInterval, "1h"),
           BOOL_PROP(kEnableRuntimeMetricsCollection, false),
-      };
+          STR_PROP(kRuntimeMetricReporter, "dummy")};
 }
 
 SystemConfig* SystemConfig::instance() {
@@ -627,6 +627,10 @@ bool SystemConfig::enableRuntimeMetricsCollection() const {
   return optionalProperty<bool>(kEnableRuntimeMetricsCollection).value();
 }
 
+const std::string& SystemConfig::getRuntimeMetricReporter() const {
+  return optionalProperty<std::string>(kRuntimeMetricReporter).value();
+}
+
 NodeConfig::NodeConfig() {
   registeredProps_ =
       std::unordered_map<std::string, folly::Optional<std::string>>{

presto-native-execution/presto_cpp/main/common/Configs.h

@@ -338,6 +338,8 @@ class SystemConfig : public ConfigBase {
   static constexpr std::string_view kEnableRuntimeMetricsCollection{
       "runtime-metrics-collection-enabled"};
 
+  static constexpr std::string_view kRuntimeMetricReporter{
+      "runtime-metrics-reporter"};
   /// Specifies the memory arbitrator kind. If it is empty, then there is no
   /// memory arbitration.
   static constexpr std::string_view kMemoryArbitratorKind{
@@ -710,6 +712,8 @@ class SystemConfig : public ConfigBase {
   std::chrono::duration<double> cacheVeloxTtlCheckInterval() const;
 
   bool enableRuntimeMetricsCollection() const;
+
+  const std::string& getRuntimeMetricReporter() const;
 };
 
 /// Provides access to node properties defined in node.properties file.

presto-native-execution/presto_cpp/main/common/runtime-metrics/CMakeLists.txt

@@ -0,0 +1,20 @@
+# 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.
+
+add_library(presto_runtime_metrics PrestoRuntimeMetricsReporter.cpp)
+
+target_link_libraries(presto_runtime_metrics prometheus_reporter)
+
+set_property(TARGET presto_runtime_metrics PROPERTY JOB_POOL_LINK
+                                                    presto_link_job_pool)
+
+add_subdirectory(prometheus-reporter)

presto-native-execution/presto_cpp/main/common/runtime-metrics/PrestoRuntimeMetricsReporter.cpp

@@ -0,0 +1,50 @@
+//
+// Created by Karteek Murthy Samba Murthy on 5/14/24.
+//
+
+#include "presto_cpp/main/common/runtime-metrics/PrestoRuntimeMetricsReporter.h"
+
+namespace facebook::presto {
+namespace {
+
+std::unordered_map<
+    std::string,
+    std::unique_ptr<facebook::velox::BaseStatsReporter>>&
+metricsReporters() {
+  static std::unordered_map<
+      std::string,
+      std::unique_ptr<facebook::velox::BaseStatsReporter>>
+      metricsReporters;
+  return metricsReporters;
+}
+
+void initLabelsForPrometheusReporter(prometheus::Labels& labels) {
+  facebook::velox::BaseStatsReporter::registered = true;
+  auto nodeConfig = facebook::presto::NodeConfig::instance();
+  const std::string cluster = nodeConfig->nodeEnvironment();
+  const char* hostName = std::getenv("HOSTNAME");
+  const std::string worker = !hostName ? "" : hostName;
+  labels.emplace("cluster", cluster);
+  labels.emplace("worker", worker);
+}
+} // namespace
+
+void registerRuntimeMetricsReporters() {
+  prometheus::Labels labels;
+  initLabelsForPrometheusReporter(labels);
+  auto prometheusReporter =
+      std::make_unique<prometheus::PrometheusReporter>(labels);
+  metricsReporters().insert(
+      {kPrometheusReporterName.data(), std::move(prometheusReporter)});
+  metricsReporters().insert(
+      {kDummyReporterName.data(),
+       std::make_unique<facebook::velox::DummyStatsReporter>()});
+}
+
+// Initialize singleton for the reporter
+folly::Singleton<facebook::velox::BaseStatsReporter> reporter([]() {
+  auto systemConfig = facebook::presto::SystemConfig::instance();
+  auto reporterName = systemConfig->getRuntimeMetricReporter();
+  return metricsReporters()[reporterName].get();
+});
+}; // namespace facebook::presto.

presto-native-execution/presto_cpp/main/common/runtime-metrics/PrestoRuntimeMetricsReporter.h

@@ -0,0 +1,24 @@
+/*
+ * 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/Configs.h"
+#include "presto_cpp/main/common/runtime-metrics/prometheus-reporter/PrometheusReporter.h"
+
+namespace facebook::presto {
+
+constexpr std::string_view kPrometheusReporterName = "prometheus";
+constexpr std::string_view kDummyReporterName = "dummy";
+
+void registerRuntimeMetricsReporters();
+}; // namespace facebook::presto

presto-native-execution/presto_cpp/main/common/runtime-metrics/prometheus-reporter/CMakeLists.txt

@@ -0,0 +1,9 @@
+find_package(prometheus-cpp CONFIG REQUIRED)
+
+add_library(prometheus_reporter PrometheusReporter.cpp)
+
+target_link_libraries(prometheus_reporter presto_common prometheus-cpp::core)
+
+set_property(TARGET prometheus_reporter PROPERTY JOB_POOL_LINK
+                                                 presto_link_job_pool)
+add_subdirectory(tests)