Mediapipe graph process time metric

docs/metrics.md

@@ -241,6 +241,7 @@ For [MediaPipe Graphs](./mediapipe.md) execution there are 4 generic metrics whi
 | counter      | ovms_responses | Useful to track number of packets generated by MediaPipe graph. Keep in mind that single request may trigger production of multiple (or zero) packets, therefore tracking number of responses is complementary to tracking accepted requests. For example tracking streaming partial responses of LLM text generation graphs. |
 | gauge      | ovms_current_graphs | Number of graphs currently in-process. For unary communication it is equal to number of currently processing requests (each request initializes separate MediaPipe graph). For streaming communication it is equal to number of active client connections. Each connection is able to reuse the graph and decide when to delete it when the connection is closed. |
 | counter      | ovms_graph_error | Counts errors in MediaPipe graph execution phase. For example V3 LLM text generation fails in LLMCalculator due to missing prompt - calculator returns an error and graph cancels. |
+| histogram      | ovms_graph_processing_time_us | Time for which mediapipe graph was opened and has been successfully closed. |
 
 Exposing custom metrics in calculator implementations (MediaPipe graph nodes) is not supported yet.
 

src/mediapipe_internal/mediapipegraphexecutor.hpp

@@ -27,6 +27,7 @@
 #include "../model_metric_reporter.hpp"
 #include "../profiler.hpp"
 #include "../status.hpp"
+#include "../timer.hpp"
 #pragma GCC diagnostic push
 #pragma GCC diagnostic ignored "-Wdeprecated-declarations"
 #include "mediapipe/framework/calculator_graph.h"
@@ -104,6 +105,12 @@ class MediapipeGraphExecutor {
         MetricGaugeGuard currentGraphsGuard(this->mediapipeServableMetricReporter->currentGraphs.get());
         ::mediapipe::CalculatorGraph graph;
         MP_RETURN_ON_FAIL(graph.Initialize(this->config), std::string("failed initialization of MediaPipe graph: ") + this->name, StatusCode::MEDIAPIPE_GRAPH_INITIALIZATION_ERROR);
+        enum : unsigned int {
+            PROCESS,
+            TIMER_END2
+        };
+        Timer<TIMER_END2> timer;
+        timer.start(PROCESS);
         std::unordered_map<std::string, ::mediapipe::OutputStreamPoller> outputPollers;
         for (auto& name : this->outputNames) {
             if (name.empty()) {
@@ -200,6 +207,9 @@ class MediapipeGraphExecutor {
             SPDLOG_DEBUG("Mediapipe failed to execute. Failed to receive all output packets");
             return Status(StatusCode::MEDIAPIPE_EXECUTION_ERROR, "Unknown error during mediapipe execution");
         }
+        timer.stop(PROCESS);
+        double processTime = timer.template elapsed<std::chrono::microseconds>(PROCESS);
+        OBSERVE_IF_ENABLED(this->mediapipeServableMetricReporter->getProcessingTimeMetric(executionContext), processTime);
         INCREMENT_IF_ENABLED(this->mediapipeServableMetricReporter->getResponsesMetric(executionContext));
         SPDLOG_DEBUG("Received all output stream packets for graph: {}", this->name);
         return StatusCode::OK;
@@ -218,6 +228,12 @@ class MediapipeGraphExecutor {
                 // Init
                 MP_RETURN_ON_FAIL(graph.Initialize(this->config), "graph initialization", StatusCode::MEDIAPIPE_GRAPH_INITIALIZATION_ERROR);
             }
+            enum : unsigned int {
+                PROCESS,
+                TIMER_END2
+            };
+            Timer<TIMER_END2> timer;
+            timer.start(PROCESS);
             {
                 OVMS_PROFILE_SCOPE("Mediapipe graph installing packet observers");
                 // Installing observers
@@ -334,6 +350,9 @@ class MediapipeGraphExecutor {
                 MP_RETURN_ON_FAIL(status, "graph wait until done", mediapipeAbslToOvmsStatus(status.code()));
                 SPDLOG_DEBUG("Graph {}: Done execution", this->name);
             }
+            timer.stop(PROCESS);
+            double processTime = timer.template elapsed<std::chrono::microseconds>(PROCESS);
+            OBSERVE_IF_ENABLED(this->mediapipeServableMetricReporter->getProcessingTimeMetric(executionContext), processTime);
             return StatusCode::OK;
         } catch (...) {
             SPDLOG_DEBUG("Graph {}: Exception while processing MediaPipe graph", this->name);

src/metric_config.cpp

@@ -59,6 +59,7 @@ const std::string METRIC_NAME_REQUESTS_ACCEPTED = "ovms_requests_accepted";
 const std::string METRIC_NAME_REQUESTS_REJECTED = "ovms_requests_rejected";
 
 const std::string METRIC_NAME_GRAPH_ERROR = "ovms_graph_error";
+const std::string METRIC_NAME_PROCESSING_TIME = "ovms_graph_processing_time_us";
 
 bool MetricConfig::validateEndpointPath(const std::string& endpoint) {
     std::regex valid_endpoint_regex("^/[a-zA-Z0-9]*$");

src/metric_config.hpp

@@ -45,6 +45,7 @@ extern const std::string METRIC_NAME_REQUESTS_ACCEPTED;
 extern const std::string METRIC_NAME_REQUESTS_REJECTED;
 
 extern const std::string METRIC_NAME_GRAPH_ERROR;
+extern const std::string METRIC_NAME_PROCESSING_TIME;
 
 class Status;
 /**
@@ -98,6 +99,7 @@ class MetricConfig {
         {METRIC_NAME_REQUESTS_ACCEPTED},
         {METRIC_NAME_REQUESTS_REJECTED},
         {METRIC_NAME_GRAPH_ERROR},
+        {METRIC_NAME_PROCESSING_TIME},
         {METRIC_NAME_RESPONSES}};
 };
 }  // namespace ovms

src/model_metric_reporter.cpp

@@ -333,6 +333,10 @@ MediapipeServableMetricReporter::MediapipeServableMetricReporter(const MetricCon
         return;
     }
 
+    for (int i = 0; i < NUMBER_OF_BUCKETS; i++) {
+        this->buckets.emplace_back(floor(BUCKET_MULTIPLIER * pow(BUCKET_POWER_BASE, i)));
+    }
+
     auto familyName = METRIC_NAME_CURRENT_GRAPHS;
     if (metricConfig->isFamilyEnabled(familyName)) {
         auto family = registry->createFamily<MetricGauge>(familyName,
@@ -552,6 +556,34 @@ MediapipeServableMetricReporter::MediapipeServableMetricReporter(const MetricCon
             {"interface", "REST"}});
         THROW_IF_NULL(this->requestSuccessRestModelReady, "cannot create metric");
     }
+    familyName = METRIC_NAME_PROCESSING_TIME;
+    if (metricConfig->isFamilyEnabled(familyName)) {
+        auto family = registry->createFamily<MetricHistogram>(familyName,
+            "Time packet spent in the MediaPipe graph.");
+        THROW_IF_NULL(family, "cannot create family");
+
+        // KFS
+        this->processingTimeGrpcModelInfer = family->addMetric({{"name", graphName},
+                                                                   {"method", "ModelInfer"}},
+            this->buckets);
+        THROW_IF_NULL(this->processingTimeGrpcModelInfer, "cannot create metric");
+
+        this->processingTimeGrpcModelInferStream = family->addMetric({{"name", graphName},
+                                                                         {"method", "ModelInferStream"}},
+            this->buckets);
+        THROW_IF_NULL(this->processingTimeGrpcModelInfer, "cannot create metric");
+
+        // V3
+        this->processingTimeRestV3Unary = family->addMetric({{"name", graphName},
+                                                                {"method", "Unary"}},
+            this->buckets);
+        THROW_IF_NULL(this->processingTimeRestV3Unary, "cannot create metric");
+
+        this->processingTimeRestV3Stream = family->addMetric({{"name", graphName},
+                                                                 {"method", "Stream"}},
+            this->buckets);
+        THROW_IF_NULL(this->processingTimeRestV3Stream, "cannot create metric");
+    }
 }
 
 }  // namespace ovms

src/model_metric_reporter.hpp

@@ -156,6 +156,9 @@ class ModelMetricReporter : public ServableMetricReporter {
 class MediapipeServableMetricReporter {
     MetricRegistry* registry;
 
+protected:
+    std::vector<double> buckets;
+
 public:
     std::unique_ptr<MetricGauge> currentGraphs;
 
@@ -205,6 +208,24 @@ class MediapipeServableMetricReporter {
     std::unique_ptr<MetricCounter> requestErrorRestV3Unary;
     std::unique_ptr<MetricCounter> requestErrorRestV3Stream;
 
+    std::unique_ptr<MetricHistogram> processingTimeGrpcModelInfer;
+    std::unique_ptr<MetricHistogram> processingTimeGrpcModelInferStream;
+    std::unique_ptr<MetricHistogram> processingTimeRestModelInfer;
+    std::unique_ptr<MetricHistogram> processingTimeRestV3Unary;
+    std::unique_ptr<MetricHistogram> processingTimeRestV3Stream;
+
+    inline MetricHistogram* getProcessingTimeMetric(const ExecutionContext& context) {
+        if (context.method == ExecutionContext::Method::ModelInfer)
+            return this->processingTimeGrpcModelInfer.get();
+        if (context.method == ExecutionContext::Method::ModelInferStream)
+            return this->processingTimeGrpcModelInferStream.get();
+        if (context.method == ExecutionContext::Method::V3Unary)
+            return this->processingTimeRestV3Unary.get();
+        if (context.method == ExecutionContext::Method::V3Stream)
+            return this->processingTimeRestV3Stream.get();
+        return nullptr;
+    }
+
     inline MetricCounter* getRequestsMetric(const ExecutionContext& context, bool success = true) {
         if (context.interface == ExecutionContext::Interface::GRPC) {
             if (context.method == ExecutionContext::Method::ModelInfer)

src/test/metrics_flow_test.cpp

@@ -395,6 +395,8 @@ TEST_F(MetricFlowTest, GrpcModelInfer) {
     checkMediapipeRequestsCounter(server.collect(), METRIC_NAME_REQUESTS_REJECTED, mpName, "gRPC", "ModelInfer", "KServe", numberOfRejectedRequests);
 
     checkMediapipeRequestsCounter(server.collect(), METRIC_NAME_RESPONSES, mpName, "gRPC", "ModelInfer", "KServe", numberOfAcceptedRequests);
+
+    EXPECT_THAT(server.collect(), HasSubstr(METRIC_NAME_PROCESSING_TIME + std::string{"_count{method=\"ModelInfer\",name=\""} + mpName + std::string{"\"} "} + std::to_string(numberOfAcceptedRequests)));
 #endif
 
     EXPECT_THAT(server.collect(), HasSubstr(METRIC_NAME_REQUEST_TIME + std::string{"_count{interface=\"gRPC\",name=\""} + modelName + std::string{"\",version=\"1\"} "} + std::to_string(numberOfSuccessRequests)));
@@ -714,6 +716,8 @@ TEST_F(MetricFlowTest, RestModelInfer) {
     checkMediapipeRequestsCounter(server.collect(), METRIC_NAME_REQUESTS_REJECTED, mpName, "REST", "ModelInfer", "KServe", numberOfRejectedRequests);
 
     checkMediapipeRequestsCounter(server.collect(), METRIC_NAME_RESPONSES, mpName, "REST", "ModelInfer", "KServe", numberOfAcceptedRequests);
+
+    EXPECT_THAT(server.collect(), HasSubstr(METRIC_NAME_PROCESSING_TIME + std::string{"_count{method=\"ModelInfer\",name=\""} + mpName + std::string{"\"} "} + std::to_string(numberOfAcceptedRequests)));
 #endif
 
     EXPECT_THAT(server.collect(), HasSubstr(METRIC_NAME_REQUEST_TIME + std::string{"_count{interface=\"gRPC\",name=\""} + modelName + std::string{"\",version=\"1\"} "} + std::to_string(0)));
@@ -835,6 +839,8 @@ TEST_F(MetricFlowTest, RestV3Unary) {
     checkMediapipeRequestsCounter(server.collect(), METRIC_NAME_REQUESTS_ACCEPTED, "dummy_gpt", "REST", "Unary", "V3", numberOfAcceptedRequests * 2);
     // checkMediapipeRequestsCounter(server.collect(), METRIC_NAME_REQUESTS_REJECTED, "dummy_gpt", "REST", "Unary", "V3", numberOfRejectedRequests);
     checkMediapipeRequestsCounter(server.collect(), METRIC_NAME_RESPONSES, "dummy_gpt", "REST", "Unary", "V3", numberOfAcceptedRequests * 2);
+
+    EXPECT_THAT(server.collect(), HasSubstr(METRIC_NAME_PROCESSING_TIME + std::string{"_count{method=\"Unary\",name=\""} + "dummy_gpt" + std::string{"\"} "} + std::to_string(numberOfAcceptedRequests * 2)));
 }
 #endif
 
@@ -887,6 +893,9 @@ TEST_F(MetricFlowTest, RestV3Stream) {
     // checkMediapipeRequestsCounter(server.collect(), METRIC_NAME_REQUESTS_REJECTED, "dummy_gpt", "REST", "Stream", "V3", numberOfRejectedRequests);
     const int numberOfMockedChunksPerRequest = 9;  // Defined in openai_chat_completions_mock_calculator.cpp
     checkMediapipeRequestsCounter(server.collect(), METRIC_NAME_RESPONSES, "dummy_gpt", "REST", "Stream", "V3", numberOfAcceptedRequests * numberOfMockedChunksPerRequest * 2);
+
+    EXPECT_THAT(server.collect(), HasSubstr(METRIC_NAME_PROCESSING_TIME + std::string{"_count{method=\"Stream\",name=\""} + "dummy_gpt" + std::string{"\"} "} + std::to_string(numberOfAcceptedRequests * 2)));
+
     SPDLOG_ERROR(server.collect());
 }
 #endif
@@ -993,6 +1002,7 @@ std::string MetricFlowTest::prepareConfigContent() {
                          R"(",")" + METRIC_NAME_REQUESTS_REJECTED +
                          R"(",")" + METRIC_NAME_RESPONSES +
                          R"(",")" + METRIC_NAME_GRAPH_ERROR +
+                         R"(",")" + METRIC_NAME_PROCESSING_TIME +
                          R"("]
             }
         },