diff --git a/docs/source/design/tent/metrics.md b/docs/source/design/tent/metrics.md new file mode 100644 index 00000000..463756b9 --- /dev/null +++ b/docs/source/design/tent/metrics.md @@ -0,0 +1,388 @@ +# TENT Metrics System + +TENT provides a built-in metrics system based on yalantinglibs, compatible with Prometheus for monitoring data transfer performance and system health. + +## Overview + +The metrics system supports two metric types: + +- **Counter**: Monotonically increasing values (e.g., total bytes transferred, total requests) +- **Histogram**: Distribution of values with configurable buckets (e.g., latency) + +All metrics are thread-safe and designed for high-performance data paths. + +## Performance Optimization + +The metrics system provides two levels of control for performance optimization: + +### Compile-time Disable (Zero Overhead) + +By default, metrics are **disabled** at compile time for maximum performance. To enable metrics, build with: + +```bash +cmake -DTENT_METRICS_ENABLED=ON .. +``` + +When disabled at compile time (`TENT_METRICS_ENABLED=OFF`, the default), all metrics macros expand to `((void)0)`, resulting in **zero runtime overhead**. + +### Runtime Disable (Minimal Overhead) + +When metrics are enabled at compile time, you can still disable them at runtime: + +```cpp +// Disable metrics collection at runtime +TentMetrics::setEnabled(false); + +// Re-enable metrics collection +TentMetrics::setEnabled(true); + +// Check current state +bool enabled = TentMetrics::isEnabled(); +``` + +When disabled at runtime, record functions return immediately after a single atomic load (~1ns overhead). + +## Configuration + +### Configuration Sources (Priority Order) + +1. **Config File** (highest priority) +2. **Environment Variables** (medium priority) +3. **Default Values** (lowest priority) + +### Config File Format + +TENT metrics configuration is integrated into the main `transfer-engine.json` configuration file: + +```json +{ + "local_segment_name": "", + "metadata_type": "p2p", + "metadata_servers": "127.0.0.1:2379", + "log_level": "warning", + "metrics": { + "enabled": true, + "http_port": 9100, + "http_host": "0.0.0.0", + "http_server_threads": 2, + "report_interval_seconds": 30, + "enable_prometheus": true, + "enable_json": true, + "latency_buckets": [0.0001, 0.0005, 0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0], + "size_buckets": [1024, 4096, 16384, 65536, 262144, 1048576, 4194304, 16777216, 67108864, 268435456, 1073741824] + }, + "transports": { + // ... transport configuration + } +} +``` + +**Note**: +- `report_interval_seconds`: Set to 0 to disable periodic logging +- `latency_buckets`: Values are in **seconds** (e.g., 0.001 = 1ms). The system internally converts to microseconds for histogram storage. +- `size_buckets`: Values are in bytes + +### Environment Variables + +```bash +# Basic settings +TENT_METRICS_ENABLED=true +TENT_METRICS_HTTP_PORT=9100 +TENT_METRICS_HTTP_HOST=0.0.0.0 +TENT_METRICS_HTTP_SERVER_THREADS=2 +TENT_METRICS_REPORT_INTERVAL=30 # Set to 0 to disable periodic logging + +# Output formats +TENT_METRICS_ENABLE_PROMETHEUS=true +TENT_METRICS_ENABLE_JSON=true + +# Custom buckets (comma-separated, latency in seconds, size in bytes) +TENT_METRICS_LATENCY_BUCKETS="0.0001,0.0005,0.001,0.005,0.01,0.05,0.1,0.5,1.0" +TENT_METRICS_SIZE_BUCKETS="1024,4096,16384,65536,262144,1048576" +``` + +## Quick Start + +### Build with Metrics Enabled + +```bash +# Enable metrics at compile time (disabled by default) +cmake -DTENT_METRICS_ENABLED=ON .. +make +``` + +### Basic Usage + +```cpp +#include "tent/metrics/tent_metrics.h" +#include "tent/metrics/config_loader.h" + +// Load configuration from transfer-engine.json +auto config = MetricsConfigLoader::loadWithDefaults(); + +// Initialize TENT metrics system +auto& tent_metrics = TentMetrics::instance(); +tent_metrics.initialize(config); + +// HTTP server starts automatically +``` + +### Recording Transfer Metrics + +```cpp +// Using convenience macros (recommended) +TENT_RECORD_READ_COMPLETED(1024*1024, 0.025); // 1MB read in 25ms +TENT_RECORD_WRITE_COMPLETED(512*1024, 0.015); // 512KB write in 15ms +TENT_RECORD_READ_FAILED(1024*1024); // 1MB read failed +TENT_RECORD_WRITE_FAILED(512*1024); // 512KB write failed + +// Direct API usage +auto& tent_metrics = TentMetrics::instance(); +tent_metrics.recordReadCompleted(1024*1024, 0.025); +tent_metrics.recordWriteCompleted(512*1024, 0.015); +tent_metrics.recordReadFailed(1024*1024); +tent_metrics.recordWriteFailed(512*1024); +``` + +### RAII Latency Measurement + +```cpp +// Automatic latency measurement using RAII +{ + TENT_SCOPED_READ_LATENCY(1024 * 1024); // e.g. 1MB + // ... perform read operation ... +} // latency automatically recorded when scope exits + +{ + TENT_SCOPED_WRITE_LATENCY(512 * 1024); // e.g. 512KB + // ... perform write operation ... +} +``` + +## HTTP Server Endpoints + +The HTTP server provides multiple endpoints: + +- **`/metrics`**: Prometheus format +- **`/metrics/summary`**: Human-readable summary +- **`/metrics/json`**: JSON format +- **`/health`**: Health check endpoint + +### Example Responses + +**Prometheus Format (`/metrics`)**: +``` +# HELP tent_read_bytes_total Total bytes read via TENT +# TYPE tent_read_bytes_total counter +tent_read_bytes_total 1048576 + +# HELP tent_write_bytes_total Total bytes written via TENT +# TYPE tent_write_bytes_total counter +tent_write_bytes_total 524288 + +# HELP tent_read_requests_total Total read requests via TENT +# TYPE tent_read_requests_total counter +tent_read_requests_total 100 + +# HELP tent_write_requests_total Total write requests via TENT +# TYPE tent_write_requests_total counter +tent_write_requests_total 50 + +# HELP tent_read_failures_total Total read failures via TENT +# TYPE tent_read_failures_total counter +tent_read_failures_total 2 + +# HELP tent_write_failures_total Total write failures via TENT +# TYPE tent_write_failures_total counter +tent_write_failures_total 1 + +# HELP tent_read_latency_us Read latency distribution in microseconds +# TYPE tent_read_latency_us histogram +tent_read_latency_us_bucket{le="100"} 10 +tent_read_latency_us_bucket{le="500"} 50 +... +``` + +**JSON Format (`/metrics/json`)**: +```json +{ + "tent_read_bytes_total": 1048576, + "tent_write_bytes_total": 524288, + "tent_read_requests_total": 100, + "tent_write_requests_total": 50, + "tent_read_failures_total": 2, + "tent_write_failures_total": 1 +} +``` + +**Summary Format (`/metrics/summary`)**: +``` +Read: 1.00 MB (100 reqs, 2 fails) | Write: 512.00 KB (50 reqs, 1 fails) +``` + +## Available Metrics + +| Metric Name | Type | Description | +|-------------|------|-------------| +| `tent_read_bytes_total` | Counter | Total bytes read via TENT | +| `tent_write_bytes_total` | Counter | Total bytes written via TENT | +| `tent_read_requests_total` | Counter | Total read requests via TENT | +| `tent_write_requests_total` | Counter | Total write requests via TENT | +| `tent_read_failures_total` | Counter | Total read failures via TENT | +| `tent_write_failures_total` | Counter | Total write failures via TENT | +| `tent_read_latency_us` | Histogram | Read latency distribution in microseconds | +| `tent_write_latency_us` | Histogram | Write latency distribution in microseconds | +| `tent_read_size_bytes` | Histogram | Read request size distribution in bytes | +| `tent_write_size_bytes` | Histogram | Write request size distribution in bytes | + +## Integration with TransferEngine + +The metrics system is automatically integrated with TransferEngine. When TransferEngine starts, it initializes the metrics system: + +```cpp +#include "tent/metrics/tent_metrics.h" +#include "tent/metrics/config_loader.h" + +// Load configuration +auto metrics_config = MetricsConfigLoader::loadWithDefaults(); +if (metrics_config.enabled) { + TentMetrics::instance().initialize(metrics_config); +} +``` + +Metrics are automatically recorded at the TENT layer: + +- **Latency tracking**: Start time is recorded when `submitTransfer` is called +- **Metrics recording**: When `getTransferStatus` detects task completion, latency is calculated and metrics are recorded + +This provides end-to-end latency measurement across all transport types (RDMA, TCP, NVLink, etc.). + +**Note**: Remember to build with `-DTENT_METRICS_ENABLED=ON` to enable metrics collection. + +## Adding New Metrics + +To add new metrics to the TENT metrics system, follow these steps: + +### Step 1: Declare the Metric + +Add the metric member variable in `tent_metrics.h`: + +```cpp +// In TentMetrics class private section: + +// For a new counter: +ylt::metric::counter_t new_counter_{"tent_new_counter", "Description of the counter"}; + +// For a new histogram: +ylt::metric::histogram_t new_histogram_{"tent_new_histogram", "Description", + std::vector{/* bucket boundaries */}}; +``` + +### Step 2: Register the Metric + +Add the metric pointer to `registerMetrics()` in `tent_metrics.cpp`: + +```cpp +void TentMetrics::registerMetrics() { + counters_ = { + &read_bytes_total_, + // ... existing counters ... + &new_counter_, // Add new counter here + }; + + histograms_ = { + &read_latency_, + // ... existing histograms ... + &new_histogram_, // Add new histogram here + }; +} +``` + +### Step 3: Add Recording Methods (Optional) + +If needed, add public methods to record the metric: + +```cpp +// In tent_metrics.h: +void recordNewMetric(int64_t value); + +// In tent_metrics.cpp: +void TentMetrics::recordNewMetric(int64_t value) { + if (!initialized_ || !runtime_enabled_.load(std::memory_order_relaxed)) return; + new_counter_.inc(value); + // or for histogram: + // new_histogram_.observe(value); +} +``` + +### Automatic Serialization + +Once registered in `registerMetrics()`, the new metric will be **automatically included** in: +- `/metrics` (Prometheus format) +- `/metrics/json` (JSON format) + +No changes to `getPrometheusMetrics()` or `getJsonMetrics()` are required. + +## Advanced Configuration + +### Custom Buckets + +Define custom histogram buckets for specific use cases: + +```cpp +// Latency buckets (in seconds, converted to microseconds internally) +std::vector rdma_latency_buckets = { + 0.000001, 0.000005, 0.00001, 0.00005, 0.0001, // 1-100μs + 0.0005, 0.001, 0.005, 0.01, 0.05, 0.1 // 0.5-100ms +}; + +// Size buckets for different data patterns (in bytes) +std::vector message_size_buckets = { + 64, 256, 1024, 4096, 16384, 65536, 262144 // 64B to 256KB +}; +``` + +### Validation + +```cpp +MetricsConfig config = MetricsConfigLoader::loadWithDefaults(); +std::string error_msg; +if (!MetricsConfigLoader::validateConfig(config, &error_msg)) { + LOG(ERROR) << "Invalid metrics config: " << error_msg; + return; +} +``` + +## Prometheus Integration + +### Prometheus Configuration + +```yaml +# prometheus.yml +scrape_configs: + - job_name: 'tent-metrics' + static_configs: + - targets: ['localhost:9100'] + scrape_interval: 15s + metrics_path: /metrics +``` + +### Grafana Queries + +```promql +# Transfer throughput (MB/s) +rate(tent_read_bytes_total[5m]) / 1024 / 1024 +rate(tent_write_bytes_total[5m]) / 1024 / 1024 + +# Request rate +rate(tent_read_requests_total[5m]) +rate(tent_write_requests_total[5m]) + +# Failure rate +rate(tent_read_failures_total[5m]) / rate(tent_read_requests_total[5m]) + +# P99 latency (note: latency is in microseconds, convert to seconds for display) +histogram_quantile(0.99, rate(tent_read_latency_us_bucket[5m])) / 1000000 +histogram_quantile(0.99, rate(tent_write_latency_us_bucket[5m])) / 1000000 +``` + diff --git a/mooncake-transfer-engine/tent/CMakeLists.txt b/mooncake-transfer-engine/tent/CMakeLists.txt index 9d6ee0fe..82bced27 100644 --- a/mooncake-transfer-engine/tent/CMakeLists.txt +++ b/mooncake-transfer-engine/tent/CMakeLists.txt @@ -7,4 +7,5 @@ endif() set(CMAKE_POSITION_INDEPENDENT_CODE ON) add_subdirectory(src) -add_subdirectory(plugins) \ No newline at end of file +add_subdirectory(plugins) +add_subdirectory(tests) diff --git a/mooncake-transfer-engine/tent/config/transfer-engine.json b/mooncake-transfer-engine/tent/config/transfer-engine.json index 7a3cd78b..adc2e0b8 100644 --- a/mooncake-transfer-engine/tent/config/transfer-engine.json +++ b/mooncake-transfer-engine/tent/config/transfer-engine.json @@ -10,6 +10,24 @@ "rdma_blacklist": [] }, "log_level": "warning", + "metrics": { + "enabled": true, + "http_port": 9100, + "http_host": "0.0.0.0", + "http_server_threads": 2, + "report_interval_seconds": 30, + "enable_prometheus": true, + "enable_json": true, + "latency_buckets": [ + 0.000125, 0.00015, 0.0002, 0.00025, 0.0003, 0.0004, 0.0005, + 0.00075, 0.001, 0.0015, 0.002, 0.003, 0.005, 0.007, + 0.015, 0.02, 0.05, 0.1, 0.2, 0.5, 1.0 + ], + "size_buckets": [ + 1024, 4096, 16384, 65536, 262144, 1048576, 4194304, + 16777216, 67108864, 268435456, 1073741824 + ] + }, "transports": { "rdma": { "enable" : true, diff --git a/mooncake-transfer-engine/tent/include/tent/common/config.h b/mooncake-transfer-engine/tent/include/tent/common/config.h index 841be71a..4074dc0d 100644 --- a/mooncake-transfer-engine/tent/include/tent/common/config.h +++ b/mooncake-transfer-engine/tent/include/tent/common/config.h @@ -19,10 +19,12 @@ #include #include +#include #include #include #include #include +#include namespace mooncake { namespace tent { @@ -72,6 +74,13 @@ class Config { struct ConfigHelper { Status loadFromEnv(Config& config); + + // Common parsing utilities for environment variable values + static bool parseBool(const std::string& str, bool default_value = false); + static int parseInt(const std::string& str, int default_value = 0); + static uint16_t parsePort(const std::string& str, + uint16_t default_value = 0); + static std::vector parseDoubleArray(const std::string& str); }; } // namespace tent diff --git a/mooncake-transfer-engine/tent/include/tent/metrics/config_loader.h b/mooncake-transfer-engine/tent/include/tent/metrics/config_loader.h new file mode 100644 index 00000000..a61a7c10 --- /dev/null +++ b/mooncake-transfer-engine/tent/include/tent/metrics/config_loader.h @@ -0,0 +1,100 @@ +// Copyright 2025 KVCache.AI +// +// 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. + +#ifndef TENT_METRICS_CONFIG_LOADER_H +#define TENT_METRICS_CONFIG_LOADER_H + +#include "tent/common/config.h" + +namespace mooncake { +namespace tent { + +/** + * @brief Configuration structure for TENT metrics system + */ +struct MetricsConfig { + bool enabled = true; + std::string http_host = "0.0.0.0"; + uint16_t http_port = 9100; + uint16_t http_server_threads = 2; // HTTP server thread count + uint32_t report_interval_seconds = 30; // 0 means disabled + bool enable_prometheus = true; + bool enable_json = true; + std::vector latency_buckets; + std::vector size_buckets; +}; + +// Helper class to load metrics configuration from various sources +// Uses ConfigHelper for common parsing utilities +class MetricsConfigLoader { + public: + // Load configuration from TENT Config object + static MetricsConfig loadFromConfig(const Config& config); + + // Load configuration from environment variables + static MetricsConfig loadFromEnvironment(); + + // Load configuration with defaults and overrides + // Priority: Config file > Environment variables > Defaults + static MetricsConfig loadWithDefaults(const Config* config = nullptr); + + // Validate configuration + static bool validateConfig(const MetricsConfig& config, + std::string* error_msg = nullptr); + + // Get default configuration + static MetricsConfig getDefaultConfig(); + + private: + // Apply environment variable overrides to config + static void applyEnvironmentOverrides(MetricsConfig& config); +}; + +// Configuration keys used in config files and environment variables +namespace config_keys { +// Main metrics configuration +constexpr const char* METRICS_ENABLED = "metrics/enabled"; +constexpr const char* METRICS_HTTP_PORT = "metrics/http_port"; +constexpr const char* METRICS_HTTP_HOST = "metrics/http_host"; +constexpr const char* METRICS_HTTP_SERVER_THREADS = + "metrics/http_server_threads"; +constexpr const char* METRICS_REPORT_INTERVAL = + "metrics/report_interval_seconds"; +constexpr const char* METRICS_ENABLE_PROMETHEUS = "metrics/enable_prometheus"; +constexpr const char* METRICS_ENABLE_JSON = "metrics/enable_json"; + +// Bucket configurations +constexpr const char* METRICS_LATENCY_BUCKETS = "metrics/latency_buckets"; +constexpr const char* METRICS_SIZE_BUCKETS = "metrics/size_buckets"; + +// Environment variable names (with TENT_ prefix) +constexpr const char* ENV_METRICS_ENABLED = "TENT_METRICS_ENABLED"; +constexpr const char* ENV_METRICS_HTTP_PORT = "TENT_METRICS_HTTP_PORT"; +constexpr const char* ENV_METRICS_HTTP_HOST = "TENT_METRICS_HTTP_HOST"; +constexpr const char* ENV_METRICS_HTTP_SERVER_THREADS = + "TENT_METRICS_HTTP_SERVER_THREADS"; +constexpr const char* ENV_METRICS_REPORT_INTERVAL = + "TENT_METRICS_REPORT_INTERVAL"; +constexpr const char* ENV_METRICS_ENABLE_PROMETHEUS = + "TENT_METRICS_ENABLE_PROMETHEUS"; +constexpr const char* ENV_METRICS_ENABLE_JSON = "TENT_METRICS_ENABLE_JSON"; +constexpr const char* ENV_METRICS_LATENCY_BUCKETS = + "TENT_METRICS_LATENCY_BUCKETS"; +constexpr const char* ENV_METRICS_SIZE_BUCKETS = "TENT_METRICS_SIZE_BUCKETS"; +} // namespace config_keys + +} // namespace tent +} // namespace mooncake + +#endif // TENT_METRICS_CONFIG_LOADER_H \ No newline at end of file diff --git a/mooncake-transfer-engine/tent/include/tent/metrics/tent_metrics.h b/mooncake-transfer-engine/tent/include/tent/metrics/tent_metrics.h new file mode 100644 index 00000000..bfbfe0eb --- /dev/null +++ b/mooncake-transfer-engine/tent/include/tent/metrics/tent_metrics.h @@ -0,0 +1,276 @@ +// Copyright 2025 KVCache.AI +// +// 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. + +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +#include "tent/common/status.h" +#include "tent/metrics/config_loader.h" + +// Compile-time metrics enable/disable switch +// Can be set via CMake option TENT_METRICS_ENABLED or define +// TENT_METRICS_ENABLED=0/1 Default is OFF (0) to match CMake default: +// option(TENT_METRICS_ENABLED ... OFF) +#ifndef TENT_METRICS_ENABLED +#define TENT_METRICS_ENABLED 0 +#endif + +#if TENT_METRICS_ENABLED +#include +#include +#endif + +namespace mooncake::tent { + +// Estimated buffer size for Prometheus metrics serialization (pre-allocation +// optimization) +constexpr size_t kPrometheusBufferSize = 4096; + +/** + * @brief TENT metrics system with HTTP server for Prometheus scraping + * + * This class provides: + * - Metrics collection using yalantinglibs + * - HTTP server for /metrics endpoint (Prometheus format) + * - Optional periodic logging of metrics summary + * - Compile-time disable option (TENT_METRICS_ENABLED=0) for zero overhead + * - Runtime disable option via setEnabled(false) for minimal overhead + */ +class TentMetrics { + public: + static TentMetrics& instance(); + + // Initialize with configuration and start HTTP server + Status initialize(const MetricsConfig& config); + + // Cleanup and stop HTTP server + void shutdown(); + + // Runtime enable/disable switch + // When disabled, record* functions return immediately with minimal overhead + static void setEnabled(bool enabled) { + runtime_enabled_.store(enabled, std::memory_order_relaxed); + } + static bool isEnabled() { + return runtime_enabled_.load(std::memory_order_relaxed); + } + + // Record transfer operations + void recordReadCompleted(size_t bytes, double latency_seconds = 0.0); + void recordWriteCompleted(size_t bytes, double latency_seconds = 0.0); + void recordReadFailed(size_t bytes); + void recordWriteFailed(size_t bytes); + + // Get metrics for HTTP server + std::string getPrometheusMetrics(); + std::string getJsonMetrics(); + std::string getSummaryString(); + + // Check if initialized + bool isInitialized() const { return initialized_; } + + private: + TentMetrics() = default; + ~TentMetrics(); + TentMetrics(const TentMetrics&) = delete; + TentMetrics& operator=(const TentMetrics&) = delete; + + // Runtime enable flag (atomic for thread-safe access with minimal overhead) + static inline std::atomic runtime_enabled_{true}; + + std::atomic initialized_{false}; + MetricsConfig config_; + +#if TENT_METRICS_ENABLED + // Initialize HTTP server with endpoints + void initHttpServer(); + + // HTTP server for metrics endpoint + std::unique_ptr http_server_; + + // Periodic metric reporting thread + std::thread metric_report_thread_; + std::atomic metric_report_running_{false}; + std::mutex metric_report_mutex_; + std::condition_variable metric_report_cv_; + + // Counters - stored as pointers for unified management + std::vector counters_; + ylt::metric::counter_t read_bytes_total_{"tent_read_bytes_total", + "Total bytes read via TENT"}; + ylt::metric::counter_t write_bytes_total_{"tent_write_bytes_total", + "Total bytes written via TENT"}; + ylt::metric::counter_t read_requests_total_{"tent_read_requests_total", + "Total read requests via TENT"}; + ylt::metric::counter_t write_requests_total_{ + "tent_write_requests_total", "Total write requests via TENT"}; + ylt::metric::counter_t read_failures_total_{"tent_read_failures_total", + "Total read failures via TENT"}; + ylt::metric::counter_t write_failures_total_{ + "tent_write_failures_total", "Total write failures via TENT"}; + + // Histograms - stored as pointers for unified management + std::vector histograms_; + // Store bucket boundaries separately since ylt histogram doesn't expose + // them publicly + std::vector> histogram_boundaries_; + + // Latency histograms use microseconds (us) as unit + // Default buckets: 100us, 500us, 1ms, 5ms, 10ms, 50ms, 100ms, 500ms, 1s + static inline const std::vector kLatencyBuckets{ + 100, 500, 1000, 5000, 10000, 50000, 100000, 500000, 1000000}; + ylt::metric::histogram_t read_latency_{ + "tent_read_latency_us", "Read latency distribution in microseconds", + kLatencyBuckets}; + ylt::metric::histogram_t write_latency_{ + "tent_write_latency_us", "Write latency distribution in microseconds", + kLatencyBuckets}; + // Size histograms for request size distribution (in bytes) + // Default buckets: 1KB, 4KB, 16KB, 64KB, 256KB, 1MB, 4MB, 16MB, 64MB, + // 256MB, 1GB + static inline const std::vector kSizeBuckets{ + 1024, 4096, 16384, 65536, 262144, 1048576, + 4194304, 16777216, 67108864, 268435456, 1073741824}; + ylt::metric::histogram_t read_size_{ + "tent_read_size_bytes", "Read request size distribution in bytes", + kSizeBuckets}; + ylt::metric::histogram_t write_size_{ + "tent_write_size_bytes", "Write request size distribution in bytes", + kSizeBuckets}; + + // Helper to register all metrics to the vectors + void registerMetrics(); +#endif // TENT_METRICS_ENABLED +}; + +#if TENT_METRICS_ENABLED + +/** + * @brief RAII helper for automatic latency measurement + * + * When runtime metrics are disabled (TentMetrics::isEnabled() == false), + * this class skips time recording to minimize overhead. + */ +class ScopedLatencyRecorder { + public: + enum class OperationType { Read, Write }; + + ScopedLatencyRecorder(OperationType type, size_t bytes) + : type_(type), bytes_(bytes), enabled_(TentMetrics::isEnabled()) { + // Only record start time if metrics are enabled (avoid clock overhead + // when disabled) + if (enabled_) { + start_ = std::chrono::steady_clock::now(); + } + } + + ~ScopedLatencyRecorder() { + if (!enabled_ || failed_) + return; // Skip if disabled or already marked as failed + auto end = std::chrono::steady_clock::now(); + double latency = std::chrono::duration(end - start_).count(); + if (type_ == OperationType::Read) { + TentMetrics::instance().recordReadCompleted(bytes_, latency); + } else { + TentMetrics::instance().recordWriteCompleted(bytes_, latency); + } + } + + void markFailed() { + if (!enabled_) return; // Skip if disabled + failed_ = true; + if (type_ == OperationType::Read) { + TentMetrics::instance().recordReadFailed(bytes_); + } else { + TentMetrics::instance().recordWriteFailed(bytes_); + } + } + + private: + OperationType type_; + size_t bytes_; + std::chrono::steady_clock::time_point start_; + bool enabled_; // Captured at construction time for consistent behavior + bool failed_ = false; +}; + +// Convenience macros for recording metrics (enabled version) +#define TENT_RECORD_READ_COMPLETED(bytes, latency) \ + do { \ + if (::mooncake::tent::TentMetrics::isEnabled()) { \ + ::mooncake::tent::TentMetrics::instance().recordReadCompleted( \ + bytes, latency); \ + } \ + } while (0) + +#define TENT_RECORD_WRITE_COMPLETED(bytes, latency) \ + do { \ + if (::mooncake::tent::TentMetrics::isEnabled()) { \ + ::mooncake::tent::TentMetrics::instance().recordWriteCompleted( \ + bytes, latency); \ + } \ + } while (0) + +#define TENT_RECORD_READ_FAILED(bytes) \ + do { \ + if (::mooncake::tent::TentMetrics::isEnabled()) { \ + ::mooncake::tent::TentMetrics::instance().recordReadFailed(bytes); \ + } \ + } while (0) + +#define TENT_RECORD_WRITE_FAILED(bytes) \ + do { \ + if (::mooncake::tent::TentMetrics::isEnabled()) { \ + ::mooncake::tent::TentMetrics::instance().recordWriteFailed( \ + bytes); \ + } \ + } while (0) + +// RAII macro for automatic latency measurement +#define TENT_SCOPED_READ_LATENCY(bytes) \ + ::mooncake::tent::ScopedLatencyRecorder _tent_latency_recorder_( \ + ::mooncake::tent::ScopedLatencyRecorder::OperationType::Read, bytes) + +#define TENT_SCOPED_WRITE_LATENCY(bytes) \ + ::mooncake::tent::ScopedLatencyRecorder _tent_latency_recorder_( \ + ::mooncake::tent::ScopedLatencyRecorder::OperationType::Write, bytes) + +#else // !TENT_METRICS_ENABLED + +// No-op stub class for ScopedLatencyRecorder when metrics are disabled +class ScopedLatencyRecorder { + public: + enum class OperationType { Read, Write }; + ScopedLatencyRecorder(OperationType, size_t) {} + void markFailed() {} +}; + +// Zero-overhead macros when metrics are disabled at compile time +#define TENT_RECORD_READ_COMPLETED(bytes, latency) ((void)0) +#define TENT_RECORD_WRITE_COMPLETED(bytes, latency) ((void)0) +#define TENT_RECORD_READ_FAILED(bytes) ((void)0) +#define TENT_RECORD_WRITE_FAILED(bytes) ((void)0) +#define TENT_SCOPED_READ_LATENCY(bytes) ((void)0) +#define TENT_SCOPED_WRITE_LATENCY(bytes) ((void)0) + +#endif // TENT_METRICS_ENABLED + +} // namespace mooncake::tent diff --git a/mooncake-transfer-engine/tent/include/tent/runtime/transfer_engine_impl.h b/mooncake-transfer-engine/tent/include/tent/runtime/transfer_engine_impl.h index ede628e6..194bbf42 100644 --- a/mooncake-transfer-engine/tent/include/tent/runtime/transfer_engine_impl.h +++ b/mooncake-transfer-engine/tent/include/tent/runtime/transfer_engine_impl.h @@ -17,6 +17,7 @@ #include #include +#include #include #include #include @@ -54,6 +55,7 @@ struct TaskInfo { bool staging{false}; TransferStatusEnum status{TransferStatusEnum::PENDING}; volatile TransferStatusEnum staging_status{TransferStatusEnum::PENDING}; + std::chrono::steady_clock::time_point start_time{}; // For latency tracking }; class TransferEngineImpl { @@ -179,6 +181,10 @@ class TransferEngineImpl { Status maybeFireSubmitHooks(Batch* batch, bool check = true); + void recordTaskCompletionMetrics(TaskInfo& task, + TransferStatusEnum prev_status, + TransferStatusEnum new_status); + private: struct AllocatedMemory { void* addr; diff --git a/mooncake-transfer-engine/tent/src/CMakeLists.txt b/mooncake-transfer-engine/tent/src/CMakeLists.txt index 54e017f2..8055f525 100644 --- a/mooncake-transfer-engine/tent/src/CMakeLists.txt +++ b/mooncake-transfer-engine/tent/src/CMakeLists.txt @@ -70,6 +70,7 @@ add_subdirectory(metastore) add_subdirectory(runtime) add_subdirectory(platform) add_subdirectory(transport) +add_subdirectory(metrics) add_subdirectory(python) file(GLOB TENT_ENGINE_SOURCES "${CMAKE_CURRENT_SOURCE_DIR}/*.cpp") diff --git a/mooncake-transfer-engine/tent/src/common/config.cpp b/mooncake-transfer-engine/tent/src/common/config.cpp index 36bebca0..3fc01f22 100644 --- a/mooncake-transfer-engine/tent/src/common/config.cpp +++ b/mooncake-transfer-engine/tent/src/common/config.cpp @@ -14,6 +14,10 @@ #include "tent/common/config.h" +#include + +#include + namespace mooncake { namespace tent { Status Config::load(const std::string& content) { @@ -53,5 +57,81 @@ Status ConfigHelper::loadFromEnv(Config& config) { return Status::OK(); } +bool ConfigHelper::parseBool(const std::string& str, bool default_value) { + std::string lower_str = str; + std::transform(lower_str.begin(), lower_str.end(), lower_str.begin(), + ::tolower); + + if (lower_str == "true" || lower_str == "1" || lower_str == "yes" || + lower_str == "on") { + return true; + } else if (lower_str == "false" || lower_str == "0" || lower_str == "no" || + lower_str == "off") { + return false; + } else { + LOG(WARNING) << "Invalid boolean value '" << str + << "', using default: " << default_value; + return default_value; + } +} + +int ConfigHelper::parseInt(const std::string& str, int default_value) { + try { + return std::stoi(str); + } catch (const std::exception& e) { + LOG(WARNING) << "Failed to parse integer '" << str << "': " << e.what() + << ", using default: " << default_value; + return default_value; + } +} + +uint16_t ConfigHelper::parsePort(const std::string& str, + uint16_t default_value) { + try { + int port = std::stoi(str); + if (port > 0 && port <= 65535) { + return static_cast(port); + } else { + LOG(WARNING) << "Port " << port + << " out of range (1-65535), using default: " + << default_value; + return default_value; + } + } catch (const std::exception& e) { + LOG(WARNING) << "Failed to parse port '" << str << "': " << e.what() + << ", using default: " << default_value; + return default_value; + } +} + +std::vector ConfigHelper::parseDoubleArray(const std::string& str) { + std::vector result; + std::stringstream ss(str); + std::string item; + + while (std::getline(ss, item, ',')) { + try { + // Trim whitespace + item.erase(0, item.find_first_not_of(" \t")); + item.erase(item.find_last_not_of(" \t") + 1); + + if (!item.empty()) { + double value = std::stod(item); + if (value > 0) { + result.push_back(value); + } + } + } catch (const std::exception& e) { + LOG(WARNING) << "Failed to parse double value '" << item + << "': " << e.what(); + } + } + + // Sort the result + std::sort(result.begin(), result.end()); + + return result; +} + } // namespace tent } // namespace mooncake \ No newline at end of file diff --git a/mooncake-transfer-engine/tent/src/metrics/CMakeLists.txt b/mooncake-transfer-engine/tent/src/metrics/CMakeLists.txt new file mode 100644 index 00000000..109840d5 --- /dev/null +++ b/mooncake-transfer-engine/tent/src/metrics/CMakeLists.txt @@ -0,0 +1,29 @@ +# Option to enable/disable TENT metrics at compile time +option(TENT_METRICS_ENABLED "Enable TENT metrics collection" OFF) + +file(GLOB TENT_METRICS_SOURCES "*.cpp") +add_library(tent_metrics STATIC ${TENT_METRICS_SOURCES}) +target_link_libraries(tent_metrics PUBLIC tent_common tent_interface glog pthread) + +# Pass compile definition based on option +if(TENT_METRICS_ENABLED) + target_compile_definitions(tent_metrics PUBLIC TENT_METRICS_ENABLED=1) + message(STATUS "TENT metrics: ENABLED") +else() + target_compile_definitions(tent_metrics PUBLIC TENT_METRICS_ENABLED=0) + message(STATUS "TENT metrics: DISABLED (zero overhead)") +endif() + +# Add yalantinglibs dependency for metrics and HTTP server +# Note: yalantinglibs bundles Asio in include/ylt/thirdparty/, no external Asio required +find_path(YLT_INCLUDE_DIR ylt/metric.hpp PATHS ${CMAKE_SOURCE_DIR}/thirdparties/yalantinglibs/include) +if(YLT_INCLUDE_DIR) + target_include_directories(tent_metrics PUBLIC ${YLT_INCLUDE_DIR}) + message(STATUS "Found yalantinglibs at: ${YLT_INCLUDE_DIR}") +else() + if(TENT_METRICS_ENABLED) + message(FATAL_ERROR "yalantinglibs not found, but it is required when TENT_METRICS_ENABLED is ON.") + else() + message(WARNING "yalantinglibs not found, metrics will be disabled.") + endif() +endif() \ No newline at end of file diff --git a/mooncake-transfer-engine/tent/src/metrics/config_loader.cpp b/mooncake-transfer-engine/tent/src/metrics/config_loader.cpp new file mode 100644 index 00000000..e948a7c5 --- /dev/null +++ b/mooncake-transfer-engine/tent/src/metrics/config_loader.cpp @@ -0,0 +1,239 @@ +// Copyright 2025 KVCache.AI +// +// 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 "tent/metrics/config_loader.h" + +#include + +#include + +namespace mooncake { +namespace tent { + +void MetricsConfigLoader::applyEnvironmentOverrides(MetricsConfig& config) { + if (const char* env_val = std::getenv(config_keys::ENV_METRICS_ENABLED)) { + config.enabled = ConfigHelper::parseBool(env_val, config.enabled); + } + + if (const char* env_val = std::getenv(config_keys::ENV_METRICS_HTTP_PORT)) { + config.http_port = ConfigHelper::parsePort(env_val, config.http_port); + } + + if (const char* env_val = std::getenv(config_keys::ENV_METRICS_HTTP_HOST)) { + config.http_host = env_val; + } + + if (const char* env_val = + std::getenv(config_keys::ENV_METRICS_REPORT_INTERVAL)) { + config.report_interval_seconds = + ConfigHelper::parseInt(env_val, config.report_interval_seconds); + } + + if (const char* env_val = + std::getenv(config_keys::ENV_METRICS_HTTP_SERVER_THREADS)) { + int threads = + ConfigHelper::parseInt(env_val, config.http_server_threads); + if (threads > 0 && threads <= 65535) { + config.http_server_threads = static_cast(threads); + } + } + + if (const char* env_val = + std::getenv(config_keys::ENV_METRICS_ENABLE_PROMETHEUS)) { + config.enable_prometheus = + ConfigHelper::parseBool(env_val, config.enable_prometheus); + } + + if (const char* env_val = + std::getenv(config_keys::ENV_METRICS_ENABLE_JSON)) { + config.enable_json = + ConfigHelper::parseBool(env_val, config.enable_json); + } + + if (const char* env_val = + std::getenv(config_keys::ENV_METRICS_LATENCY_BUCKETS)) { + auto buckets = ConfigHelper::parseDoubleArray(env_val); + if (!buckets.empty()) { + config.latency_buckets = buckets; + } + } + + if (const char* env_val = + std::getenv(config_keys::ENV_METRICS_SIZE_BUCKETS)) { + auto buckets = ConfigHelper::parseDoubleArray(env_val); + if (!buckets.empty()) { + config.size_buckets = buckets; + } + } +} + +MetricsConfig MetricsConfigLoader::loadFromConfig(const Config& config) { + MetricsConfig metrics_config = getDefaultConfig(); + + // Load basic settings from Config object + metrics_config.enabled = + config.get(config_keys::METRICS_ENABLED, metrics_config.enabled); + metrics_config.http_port = static_cast( + config.get(config_keys::METRICS_HTTP_PORT, + static_cast(metrics_config.http_port))); + metrics_config.http_host = + config.get(config_keys::METRICS_HTTP_HOST, metrics_config.http_host); + metrics_config.http_server_threads = static_cast( + config.get(config_keys::METRICS_HTTP_SERVER_THREADS, + static_cast(metrics_config.http_server_threads))); + metrics_config.report_interval_seconds = + config.get(config_keys::METRICS_REPORT_INTERVAL, + metrics_config.report_interval_seconds); + metrics_config.enable_prometheus = + config.get(config_keys::METRICS_ENABLE_PROMETHEUS, + metrics_config.enable_prometheus); + metrics_config.enable_json = config.get(config_keys::METRICS_ENABLE_JSON, + metrics_config.enable_json); + + // Load bucket configurations + auto latency_buckets_array = + config.getArray(config_keys::METRICS_LATENCY_BUCKETS); + if (!latency_buckets_array.empty()) { + metrics_config.latency_buckets = latency_buckets_array; + } + + auto size_buckets_array = + config.getArray(config_keys::METRICS_SIZE_BUCKETS); + if (!size_buckets_array.empty()) { + metrics_config.size_buckets = size_buckets_array; + } + + LOG(INFO) << "Loaded metrics config from Config object: enabled=" + << metrics_config.enabled + << ", port=" << metrics_config.http_port; + + return metrics_config; +} + +MetricsConfig MetricsConfigLoader::loadFromEnvironment() { + MetricsConfig metrics_config = getDefaultConfig(); + applyEnvironmentOverrides(metrics_config); + + LOG(INFO) << "Loaded metrics config from environment: enabled=" + << metrics_config.enabled + << ", port=" << metrics_config.http_port; + + return metrics_config; +} + +MetricsConfig MetricsConfigLoader::loadWithDefaults(const Config* config) { + // Priority: Config file > Environment variables > Defaults + + // 1. Start with defaults + MetricsConfig metrics_config = getDefaultConfig(); + + // 2. Override with environment variables + applyEnvironmentOverrides(metrics_config); + + // 3. Override with file config (highest priority) + if (config) { + metrics_config.enabled = + config->get(config_keys::METRICS_ENABLED, metrics_config.enabled); + metrics_config.http_port = static_cast( + config->get(config_keys::METRICS_HTTP_PORT, + static_cast(metrics_config.http_port))); + metrics_config.http_host = config->get(config_keys::METRICS_HTTP_HOST, + metrics_config.http_host); + metrics_config.http_server_threads = static_cast( + config->get(config_keys::METRICS_HTTP_SERVER_THREADS, + static_cast(metrics_config.http_server_threads))); + metrics_config.report_interval_seconds = + config->get(config_keys::METRICS_REPORT_INTERVAL, + metrics_config.report_interval_seconds); + metrics_config.enable_prometheus = + config->get(config_keys::METRICS_ENABLE_PROMETHEUS, + metrics_config.enable_prometheus); + metrics_config.enable_json = config->get( + config_keys::METRICS_ENABLE_JSON, metrics_config.enable_json); + + auto latency_buckets_array = + config->getArray(config_keys::METRICS_LATENCY_BUCKETS); + if (!latency_buckets_array.empty()) { + metrics_config.latency_buckets = latency_buckets_array; + } + + auto size_buckets_array = + config->getArray(config_keys::METRICS_SIZE_BUCKETS); + if (!size_buckets_array.empty()) { + metrics_config.size_buckets = size_buckets_array; + } + } + + return metrics_config; +} + +bool MetricsConfigLoader::validateConfig(const MetricsConfig& config, + std::string* error_msg) { + // Validate port range (http_port is uint16_t, so max is 65535) + if (config.http_port == 0) { + if (error_msg) { + *error_msg = "Invalid HTTP port: 0 (must be 1-65535)"; + } + return false; + } + + // Validate HTTP server threads + if (config.http_server_threads == 0) { + if (error_msg) { + *error_msg = "Invalid HTTP server threads: must be > 0"; + } + return false; + } + + // Validate at least one output format is enabled + if (!config.enable_prometheus && !config.enable_json) { + if (error_msg) { + *error_msg = + "At least one output format (Prometheus or JSON) must be " + "enabled"; + } + return false; + } + + // Validate buckets are sorted and positive + for (size_t i = 1; i < config.latency_buckets.size(); ++i) { + if (config.latency_buckets[i] <= config.latency_buckets[i - 1]) { + if (error_msg) { + *error_msg = + "Latency buckets must be sorted in ascending order"; + } + return false; + } + } + + for (size_t i = 1; i < config.size_buckets.size(); ++i) { + if (config.size_buckets[i] <= config.size_buckets[i - 1]) { + if (error_msg) { + *error_msg = "Size buckets must be sorted in ascending order"; + } + return false; + } + } + + return true; +} + +MetricsConfig MetricsConfigLoader::getDefaultConfig() { + MetricsConfig config; + // Default values are already set in the struct definition + return config; +} + +} // namespace tent +} // namespace mooncake diff --git a/mooncake-transfer-engine/tent/src/metrics/tent_metrics.cpp b/mooncake-transfer-engine/tent/src/metrics/tent_metrics.cpp new file mode 100644 index 00000000..0f4a2923 --- /dev/null +++ b/mooncake-transfer-engine/tent/src/metrics/tent_metrics.cpp @@ -0,0 +1,391 @@ +// Copyright 2025 KVCache.AI +// +// 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 "tent/metrics/tent_metrics.h" + +#include +#include +#include +#include + +namespace mooncake::tent { + +TentMetrics& TentMetrics::instance() { + static TentMetrics instance; + return instance; +} + +TentMetrics::~TentMetrics() { shutdown(); } + +#if TENT_METRICS_ENABLED + +Status TentMetrics::initialize(const MetricsConfig& config) { + // Use compare_exchange to prevent race condition during initialization + bool expected = false; + if (!initialized_.compare_exchange_strong(expected, true)) { + return Status::OK(); // Already initialized by another thread + } + + config_ = config; + + // Set runtime enabled state from config + runtime_enabled_.store(config_.enabled, std::memory_order_relaxed); + + // Configure histogram buckets if provided (recreate histograms) + // Note: config latency_buckets are in seconds, convert to microseconds for + // histogram + if (!config_.latency_buckets.empty()) { + // Convert seconds to microseconds for histogram buckets + std::vector latency_buckets_us; + latency_buckets_us.reserve(config_.latency_buckets.size()); + for (double bucket_sec : config_.latency_buckets) { + latency_buckets_us.push_back(bucket_sec * + 1000000.0); // seconds -> microseconds + } + read_latency_ = ylt::metric::histogram_t( + "tent_read_latency_us", "Read latency distribution in microseconds", + latency_buckets_us); + write_latency_ = ylt::metric::histogram_t( + "tent_write_latency_us", + "Write latency distribution in microseconds", latency_buckets_us); + } + + // Configure size histogram buckets if provided + if (!config_.size_buckets.empty()) { + read_size_ = ylt::metric::histogram_t( + "tent_read_size_bytes", "Read request size distribution in bytes", + config_.size_buckets); + write_size_ = ylt::metric::histogram_t( + "tent_write_size_bytes", "Write request size distribution in bytes", + config_.size_buckets); + } + + // Register all metrics to vectors for unified serialization + registerMetrics(); + + // Initialize and start HTTP server + initHttpServer(); + + // Start periodic metric reporting thread if interval > 0 + if (config_.report_interval_seconds > 0) { + metric_report_running_ = true; + metric_report_thread_ = std::thread([this]() { + while (metric_report_running_) { + std::string summary = getSummaryString(); + LOG(INFO) << "TENT Metrics: " << summary; + + // Use condition variable for interruptible sleep + std::unique_lock lock(metric_report_mutex_); + metric_report_cv_.wait_for( + lock, std::chrono::seconds(config_.report_interval_seconds), + [this]() { return !metric_report_running_.load(); }); + } + }); + } + + LOG(INFO) + << "TENT metrics initialized successfully, HTTP server listening on " + << config_.http_host << ":" << config_.http_port + << ", runtime_enabled=" << (runtime_enabled_.load() ? "true" : "false"); + return Status::OK(); +} + +void TentMetrics::initHttpServer() { + using namespace coro_http; + + // Create HTTP server with configurable threads + http_server_ = std::make_unique( + config_.http_server_threads, config_.http_port); + + // Register /metrics endpoint for Prometheus + http_server_->set_http_handler( + "/metrics", [this](coro_http_request& req, coro_http_response& resp) { + std::string metrics = getPrometheusMetrics(); + resp.add_header("Content-Type", "text/plain; version=0.0.4"); + resp.set_status_and_content(status_type::ok, std::move(metrics)); + }); + + // Register /metrics/summary endpoint for human-readable summary + http_server_->set_http_handler( + "/metrics/summary", + [this](coro_http_request& req, coro_http_response& resp) { + std::string summary = getSummaryString(); + resp.add_header("Content-Type", "text/plain"); + resp.set_status_and_content(status_type::ok, std::move(summary)); + }); + + // Register /metrics/json endpoint for JSON format + http_server_->set_http_handler( + "/metrics/json", + [this](coro_http_request& req, coro_http_response& resp) { + std::string json = getJsonMetrics(); + resp.add_header("Content-Type", "application/json"); + resp.set_status_and_content(status_type::ok, std::move(json)); + }); + + // Register /health endpoint for health check + http_server_->set_http_handler( + "/health", [](coro_http_request& req, coro_http_response& resp) { + resp.add_header("Content-Type", "text/plain"); + resp.set_status_and_content(status_type::ok, "OK"); + }); + + // Start the HTTP server asynchronously + http_server_->async_start(); +} + +void TentMetrics::shutdown() { + if (!initialized_) return; + + // Stop metric reporting thread + metric_report_running_ = false; + metric_report_cv_.notify_all(); // Wake up the sleeping thread immediately + if (metric_report_thread_.joinable()) { + metric_report_thread_.join(); + } + + // Stop HTTP server + if (http_server_) { + http_server_->stop(); + http_server_.reset(); + } + + // Clear metric vectors + counters_.clear(); + histograms_.clear(); + histogram_boundaries_.clear(); + + initialized_ = false; + LOG(INFO) << "TENT metrics shutdown complete"; +} + +void TentMetrics::registerMetrics() { + // Pre-allocate vectors to avoid reallocation + counters_.reserve(6); + histograms_.reserve(4); + histogram_boundaries_.reserve(4); + + // Register all counters - add new counters here + counters_ = { + &read_bytes_total_, &write_bytes_total_, &read_requests_total_, + &write_requests_total_, &read_failures_total_, &write_failures_total_, + }; + + // Register all histograms - add new histograms here + // Note: histogram_boundaries_ must match the order of histograms_ + histograms_ = { + &read_latency_, + &write_latency_, + &read_size_, + &write_size_, + }; + histogram_boundaries_ = { + kLatencyBuckets, + kLatencyBuckets, + kSizeBuckets, + kSizeBuckets, + }; +} + +void TentMetrics::recordReadCompleted(size_t bytes, double latency_seconds) { + // Fast path: check runtime switch first + if (!initialized_ || !runtime_enabled_.load(std::memory_order_relaxed)) + return; + + read_bytes_total_.inc(static_cast(bytes)); + read_requests_total_.inc(); + read_size_.observe(static_cast(bytes)); + if (latency_seconds > 0.0) { + // Convert seconds to microseconds for histogram (int64_t internally) + int64_t latency_us = static_cast(latency_seconds * 1000000.0); + read_latency_.observe(latency_us); + } +} + +void TentMetrics::recordWriteCompleted(size_t bytes, double latency_seconds) { + // Fast path: check runtime switch first + if (!initialized_ || !runtime_enabled_.load(std::memory_order_relaxed)) + return; + + write_bytes_total_.inc(static_cast(bytes)); + write_requests_total_.inc(); + write_size_.observe(static_cast(bytes)); + if (latency_seconds > 0.0) { + // Convert seconds to microseconds for histogram (int64_t internally) + int64_t latency_us = static_cast(latency_seconds * 1000000.0); + write_latency_.observe(latency_us); + } +} + +void TentMetrics::recordReadFailed(size_t bytes) { + // Fast path: check runtime switch first + if (!initialized_ || !runtime_enabled_.load(std::memory_order_relaxed)) + return; + + read_failures_total_.inc(); + read_requests_total_.inc(); // Count failed requests too +} + +void TentMetrics::recordWriteFailed(size_t bytes) { + // Fast path: check runtime switch first + if (!initialized_ || !runtime_enabled_.load(std::memory_order_relaxed)) + return; + + write_failures_total_.inc(); + write_requests_total_.inc(); // Count failed requests too +} + +std::string TentMetrics::getPrometheusMetrics() { + if (!initialized_) return ""; + + try { + std::string result; + // Pre-allocate buffer to avoid reallocation during serialization + result.reserve(kPrometheusBufferSize); + + // Serialize all counters + for (auto* counter : counters_) { + counter->serialize(result); + } + + // Serialize all histograms + for (auto* histogram : histograms_) { + histogram->serialize(result); + } + + return result; + } catch (const std::exception& e) { + LOG(ERROR) << "Failed to serialize Prometheus metrics: " << e.what(); + return ""; + } +} + +std::string TentMetrics::getJsonMetrics() { + if (!initialized_) return "{}"; + + try { + nlohmann::json root; + + // Serialize all counters + for (auto* counter : counters_) { + root[counter->str_name()] = counter->value(); + } + + // Serialize all histograms + for (size_t h = 0; h < histograms_.size(); ++h) { + auto* histogram = histograms_[h]; + const auto& boundaries = histogram_boundaries_[h]; + + auto bucket_counts = histogram->get_bucket_counts(); + + // Calculate total count + int64_t total_count = 0; + for (auto& bucket : bucket_counts) { + total_count += bucket->value(); + } + + nlohmann::json hist_obj; + hist_obj["count"] = total_count; + + nlohmann::json buckets_obj; + for (size_t i = 0; + i < boundaries.size() && i < bucket_counts.size(); ++i) { + buckets_obj[std::to_string(static_cast( + boundaries[i]))] = bucket_counts[i]->value(); + } + hist_obj["buckets"] = buckets_obj; + + root[histogram->str_name()] = hist_obj; + } + + return root.dump(2); // Pretty print with 2-space indent + } catch (const std::exception& e) { + LOG(ERROR) << "Failed to serialize JSON metrics: " << e.what(); + return R"({"error": "Failed to serialize metrics"})"; + } +} + +std::string TentMetrics::getSummaryString() { + if (!initialized_) return "Metrics not initialized"; + + std::ostringstream oss; + oss << std::fixed << std::setprecision(2); + + double read_bytes = read_bytes_total_.value(); + double write_bytes = write_bytes_total_.value(); + double read_reqs = read_requests_total_.value(); + double write_reqs = write_requests_total_.value(); + double read_fails = read_failures_total_.value(); + double write_fails = write_failures_total_.value(); + + // Format bytes in human-readable form + auto formatBytes = [](double bytes) -> std::string { + std::ostringstream s; + s << std::fixed << std::setprecision(2); + if (bytes >= 1e12) + s << bytes / 1e12 << " TB"; + else if (bytes >= 1e9) + s << bytes / 1e9 << " GB"; + else if (bytes >= 1e6) + s << bytes / 1e6 << " MB"; + else if (bytes >= 1e3) + s << bytes / 1e3 << " KB"; + else + s << bytes << " B"; + return s.str(); + }; + + oss << "Read: " << formatBytes(read_bytes) << " (" + << static_cast(read_reqs) << " reqs, " + << static_cast(read_fails) << " fails) | " + << "Write: " << formatBytes(write_bytes) << " (" + << static_cast(write_reqs) << " reqs, " + << static_cast(write_fails) << " fails)"; + + return oss.str(); +} + +#else // !TENT_METRICS_ENABLED + +// Stub implementations when metrics are disabled at compile time +Status TentMetrics::initialize(const MetricsConfig& config) { + config_ = config; + initialized_ = true; + LOG(INFO) + << "TENT metrics disabled at compile time (TENT_METRICS_ENABLED=0)"; + return Status::OK(); +} + +void TentMetrics::shutdown() { initialized_ = false; } + +void TentMetrics::recordReadCompleted(size_t, double) {} +void TentMetrics::recordWriteCompleted(size_t, double) {} +void TentMetrics::recordReadFailed(size_t) {} +void TentMetrics::recordWriteFailed(size_t) {} + +std::string TentMetrics::getPrometheusMetrics() { + return "# TENT metrics disabled at compile time\n"; +} + +std::string TentMetrics::getJsonMetrics() { + return R"({"status": "disabled", "message": "TENT metrics disabled at compile time"})"; +} + +std::string TentMetrics::getSummaryString() { + return "TENT metrics disabled at compile time"; +} + +#endif // TENT_METRICS_ENABLED + +} // namespace mooncake::tent diff --git a/mooncake-transfer-engine/tent/src/runtime/CMakeLists.txt b/mooncake-transfer-engine/tent/src/runtime/CMakeLists.txt index e58a9563..53f2dbc2 100644 --- a/mooncake-transfer-engine/tent/src/runtime/CMakeLists.txt +++ b/mooncake-transfer-engine/tent/src/runtime/CMakeLists.txt @@ -16,4 +16,4 @@ if (TARGET metastore_etcd) target_link_libraries(tent_runtime PUBLIC metastore_etcd) endif() -target_link_libraries(tent_runtime PUBLIC tent_common tent_rpc tent_platform_all tent_transport_all) +target_link_libraries(tent_runtime PUBLIC tent_common tent_rpc tent_platform_all tent_transport_all tent_metrics) diff --git a/mooncake-transfer-engine/tent/src/runtime/transfer_engine_impl.cpp b/mooncake-transfer-engine/tent/src/runtime/transfer_engine_impl.cpp index 70382690..d0c69fb5 100644 --- a/mooncake-transfer-engine/tent/src/runtime/transfer_engine_impl.cpp +++ b/mooncake-transfer-engine/tent/src/runtime/transfer_engine_impl.cpp @@ -30,6 +30,8 @@ #include "tent/runtime/slab.h" #include "tent/common/utils/ip.h" #include "tent/common/utils/random.h" +#include "tent/metrics/tent_metrics.h" +#include "tent/metrics/config_loader.h" namespace mooncake { namespace tent { @@ -200,6 +202,28 @@ Status TransferEngineImpl::construct() { staging_proxy_ = std::make_unique(this); + // Initialize and start Metrics system + auto metrics_config = MetricsConfigLoader::loadWithDefaults(conf_.get()); + if (metrics_config.enabled) { + std::string validation_error; + if (!MetricsConfigLoader::validateConfig(metrics_config, + &validation_error)) { + LOG(WARNING) << "Invalid metrics configuration: " + << validation_error << ", Metrics system disabled"; + } else { + // Initialize metrics + auto status = TentMetrics::instance().initialize(metrics_config); + if (!status.ok()) { + LOG(WARNING) << "Failed to initialize TENT metrics: " + << status.ToString(); + } else { + LOG(INFO) << "TENT Metrics system initialized"; + } + } + } else { + LOG(INFO) << "Metrics system disabled by configuration"; + } + if (conf_->get("verbose", false)) { LOG(INFO) << "========== Transfer Engine Parameters =========="; LOG(INFO) << " - Segment Name: " << local_segment_name_; @@ -218,6 +242,8 @@ Status TransferEngineImpl::construct() { } Status TransferEngineImpl::deconstruct() { + // Metrics cleanup is handled automatically by TentMetrics destructor + local_segment_tracker_->forEach([&](BufferDesc& desc) -> Status { for (size_t type = 0; type < kSupportedTransportTypes; ++type) { if (transport_list_[type]) @@ -717,6 +743,9 @@ Status TransferEngineImpl::submitTransfer( batch->task_list.insert(batch->task_list.end(), request_list.size(), TaskInfo{}); + // Record start time for metrics tracking + auto submit_time = std::chrono::steady_clock::now(); + auto merged = mergeRequests(request_list, merge_requests_); std::unordered_map next_sub_task_id; for (auto& kv : merged.task_lookup) { @@ -735,6 +764,8 @@ Status TransferEngineImpl::submitTransfer( task.status = PENDING; task.request = merged_request; task.staging = false; + task.start_time = + submit_time; // Record start time for latency tracking task.type = resolveTransport(merged_request, 0); if (task.type == UNSPEC) { LOG(WARNING) << "Unable to find registered buffer for request: " @@ -892,6 +923,7 @@ Status TransferEngineImpl::getTransferStatus(BatchID batch_id, size_t task_id, if (task_id >= batch->task_list.size()) return Status::InvalidArgument("Invalid task ID" LOC_MARK); auto& task = batch->task_list[task_id]; + auto prev_status = task.status; if (task.staging) { CHECK_STATUS(staging_proxy_->getStatus(&task, task_status)); } else { @@ -916,6 +948,11 @@ Status TransferEngineImpl::getTransferStatus(BatchID batch_id, size_t task_id, task_status.transferred_bytes = 0; } batch->task_list[task_id].status = task_status.s; + + // Record metrics when task transitions to terminal state + recordTaskCompletionMetrics(batch->task_list[task_id], prev_status, + task_status.s); + if (task_status.s == COMPLETED) CHECK_STATUS(maybeFireSubmitHooks(batch)); return Status::OK(); } @@ -955,6 +992,7 @@ Status TransferEngineImpl::getTransferStatus(BatchID batch_id, } continue; } + auto prev_status = task.status; if (task.staging) { CHECK_STATUS(staging_proxy_->getStatus(&task, task_status)); } else { @@ -985,6 +1023,10 @@ Status TransferEngineImpl::getTransferStatus(BatchID batch_id, } // memorize task result task.status = task_status.s; + + // Record metrics when task transitions to terminal state + recordTaskCompletionMetrics(batch->task_list[task_id], prev_status, + task_status.s); } if (success_tasks == total_tasks) overall_status.s = COMPLETED; CHECK_STATUS(maybeFireSubmitHooks(batch, overall_status.s == COMPLETED)); @@ -1035,5 +1077,37 @@ Status TransferEngineImpl::unlockStageBuffer(uint64_t addr) { return staging_proxy_->unpinStageBuffer(addr); } +void TransferEngineImpl::recordTaskCompletionMetrics( + TaskInfo& task, TransferStatusEnum prev_status, + TransferStatusEnum new_status) { + if (prev_status == PENDING && new_status != PENDING && !task.derived) { + auto start_time = task.start_time; + if (start_time.time_since_epoch().count() > 0) { + auto end_time = std::chrono::steady_clock::now(); + double latency_seconds = + std::chrono::duration(end_time - start_time).count(); + if (new_status == COMPLETED) { + if (task.request.opcode == Request::READ) { + TentMetrics::instance().recordReadCompleted( + task.request.length, latency_seconds); + } else { + TentMetrics::instance().recordWriteCompleted( + task.request.length, latency_seconds); + } + } else if (new_status == FAILED) { + if (task.request.opcode == Request::READ) { + TentMetrics::instance().recordReadFailed( + task.request.length); + } else { + TentMetrics::instance().recordWriteFailed( + task.request.length); + } + } + // Reset start_time to prevent duplicate recording + task.start_time = std::chrono::steady_clock::time_point{}; + } + } +} + } // namespace tent } // namespace mooncake diff --git a/mooncake-transfer-engine/tent/tests/CMakeLists.txt b/mooncake-transfer-engine/tent/tests/CMakeLists.txt new file mode 100644 index 00000000..99748330 --- /dev/null +++ b/mooncake-transfer-engine/tent/tests/CMakeLists.txt @@ -0,0 +1,12 @@ +# TENT Tests and Examples + +# TENT Metrics Example +add_executable(tent_metrics_example tent_metrics_example.cpp) +target_link_libraries(tent_metrics_example PRIVATE tent_metrics tent_common glog ibverbs) +target_include_directories(tent_metrics_example PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/../include) + +# TENT Metrics Config Loader Unit Test +add_executable(metrics_config_loader_test metrics_config_loader_test.cpp) +target_link_libraries(metrics_config_loader_test PRIVATE tent_metrics tent_common gtest gtest_main glog) +target_include_directories(metrics_config_loader_test PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/../include) +add_test(NAME metrics_config_loader_test COMMAND metrics_config_loader_test) diff --git a/mooncake-transfer-engine/tent/tests/metrics_config_loader_test.cpp b/mooncake-transfer-engine/tent/tests/metrics_config_loader_test.cpp new file mode 100644 index 00000000..64f0f545 --- /dev/null +++ b/mooncake-transfer-engine/tent/tests/metrics_config_loader_test.cpp @@ -0,0 +1,433 @@ +// Copyright 2025 KVCache.AI +// +// 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 + +#include +#include + +#include "tent/metrics/config_loader.h" + +namespace mooncake { +namespace tent { +namespace { + +//------------------------------------------------------------------------------ +// Helper class for managing environment variables in tests +//------------------------------------------------------------------------------ + +class EnvVarGuard { + public: + EnvVarGuard(const char* name, const char* value) : name_(name) { + const char* old = std::getenv(name); + if (old) { + old_value_ = old; + had_value_ = true; + } + setenv(name, value, 1); + } + + ~EnvVarGuard() { + if (had_value_) { + setenv(name_.c_str(), old_value_.c_str(), 1); + } else { + unsetenv(name_.c_str()); + } + } + + private: + std::string name_; + std::string old_value_; + bool had_value_ = false; +}; + +//------------------------------------------------------------------------------ +// MetricsConfig Default Values Tests +//------------------------------------------------------------------------------ + +TEST(MetricsConfigLoaderTest, GetDefaultConfigReturnsExpectedDefaults) { + MetricsConfig config = MetricsConfigLoader::getDefaultConfig(); + + EXPECT_TRUE(config.enabled); + EXPECT_EQ(config.http_host, "0.0.0.0"); + EXPECT_EQ(config.http_port, 9100); + EXPECT_EQ(config.http_server_threads, 2); + EXPECT_EQ(config.report_interval_seconds, 30); + EXPECT_TRUE(config.enable_prometheus); + EXPECT_TRUE(config.enable_json); + EXPECT_TRUE(config.latency_buckets.empty()); + EXPECT_TRUE(config.size_buckets.empty()); +} + +//------------------------------------------------------------------------------ +// MetricsConfigLoader::loadFromConfig Tests +//------------------------------------------------------------------------------ + +TEST(MetricsConfigLoaderTest, LoadFromConfigWithAllValues) { + Config config; + std::string json_content = R"({ + "metrics/enabled": false, + "metrics/http_port": 8080, + "metrics/http_host": "127.0.0.1", + "metrics/http_server_threads": 4, + "metrics/report_interval_seconds": 60, + "metrics/enable_prometheus": false, + "metrics/enable_json": true, + "metrics/latency_buckets": [0.001, 0.01, 0.1, 1.0], + "metrics/size_buckets": [1024, 10240, 102400] + })"; + ASSERT_TRUE(config.load(json_content).ok()); + + MetricsConfig metrics_config = MetricsConfigLoader::loadFromConfig(config); + + EXPECT_FALSE(metrics_config.enabled); + EXPECT_EQ(metrics_config.http_port, 8080); + EXPECT_EQ(metrics_config.http_host, "127.0.0.1"); + EXPECT_EQ(metrics_config.http_server_threads, 4); + EXPECT_EQ(metrics_config.report_interval_seconds, 60); + EXPECT_FALSE(metrics_config.enable_prometheus); + EXPECT_TRUE(metrics_config.enable_json); + ASSERT_EQ(metrics_config.latency_buckets.size(), 4); + EXPECT_DOUBLE_EQ(metrics_config.latency_buckets[0], 0.001); + EXPECT_DOUBLE_EQ(metrics_config.latency_buckets[3], 1.0); + ASSERT_EQ(metrics_config.size_buckets.size(), 3); + EXPECT_DOUBLE_EQ(metrics_config.size_buckets[0], 1024); +} + +TEST(MetricsConfigLoaderTest, LoadFromConfigWithPartialValues) { + Config config; + std::string json_content = R"({ + "metrics/http_port": 9200 + })"; + ASSERT_TRUE(config.load(json_content).ok()); + + MetricsConfig metrics_config = MetricsConfigLoader::loadFromConfig(config); + + // Only http_port should be overridden, others should be defaults + EXPECT_TRUE(metrics_config.enabled); + EXPECT_EQ(metrics_config.http_port, 9200); + EXPECT_EQ(metrics_config.http_host, "0.0.0.0"); + EXPECT_EQ(metrics_config.http_server_threads, 2); +} + +TEST(MetricsConfigLoaderTest, LoadFromConfigWithEmptyConfig) { + Config config; + std::string json_content = "{}"; + ASSERT_TRUE(config.load(json_content).ok()); + + MetricsConfig metrics_config = MetricsConfigLoader::loadFromConfig(config); + + // All values should be defaults + MetricsConfig default_config = MetricsConfigLoader::getDefaultConfig(); + EXPECT_EQ(metrics_config.enabled, default_config.enabled); + EXPECT_EQ(metrics_config.http_port, default_config.http_port); + EXPECT_EQ(metrics_config.http_host, default_config.http_host); +} + +//------------------------------------------------------------------------------ +// MetricsConfigLoader::loadFromEnvironment Tests +//------------------------------------------------------------------------------ + +TEST(MetricsConfigLoaderTest, LoadFromEnvironmentWithAllVars) { + EnvVarGuard g1(config_keys::ENV_METRICS_ENABLED, "false"); + EnvVarGuard g2(config_keys::ENV_METRICS_HTTP_PORT, "9999"); + EnvVarGuard g3(config_keys::ENV_METRICS_HTTP_HOST, "192.168.1.1"); + EnvVarGuard g4(config_keys::ENV_METRICS_HTTP_SERVER_THREADS, "8"); + EnvVarGuard g5(config_keys::ENV_METRICS_REPORT_INTERVAL, "120"); + EnvVarGuard g6(config_keys::ENV_METRICS_ENABLE_PROMETHEUS, "true"); + EnvVarGuard g7(config_keys::ENV_METRICS_ENABLE_JSON, "false"); + + MetricsConfig config = MetricsConfigLoader::loadFromEnvironment(); + + EXPECT_FALSE(config.enabled); + EXPECT_EQ(config.http_port, 9999); + EXPECT_EQ(config.http_host, "192.168.1.1"); + EXPECT_EQ(config.http_server_threads, 8); + EXPECT_EQ(config.report_interval_seconds, 120); + EXPECT_TRUE(config.enable_prometheus); + EXPECT_FALSE(config.enable_json); +} + +TEST(MetricsConfigLoaderTest, LoadFromEnvironmentWithPartialVars) { + EnvVarGuard g1(config_keys::ENV_METRICS_HTTP_PORT, "8888"); + + MetricsConfig config = MetricsConfigLoader::loadFromEnvironment(); + + EXPECT_EQ(config.http_port, 8888); + // Other values should be defaults + EXPECT_TRUE(config.enabled); + EXPECT_EQ(config.http_host, "0.0.0.0"); +} + +TEST(MetricsConfigLoaderTest, LoadFromEnvironmentWithLatencyBuckets) { + EnvVarGuard g1(config_keys::ENV_METRICS_LATENCY_BUCKETS, + "0.001,0.005,0.01"); + + MetricsConfig config = MetricsConfigLoader::loadFromEnvironment(); + + ASSERT_EQ(config.latency_buckets.size(), 3); + EXPECT_DOUBLE_EQ(config.latency_buckets[0], 0.001); + EXPECT_DOUBLE_EQ(config.latency_buckets[1], 0.005); + EXPECT_DOUBLE_EQ(config.latency_buckets[2], 0.01); +} + +TEST(MetricsConfigLoaderTest, LoadFromEnvironmentWithSizeBuckets) { + EnvVarGuard g1(config_keys::ENV_METRICS_SIZE_BUCKETS, "1024,2048,4096"); + + MetricsConfig config = MetricsConfigLoader::loadFromEnvironment(); + + ASSERT_EQ(config.size_buckets.size(), 3); + EXPECT_DOUBLE_EQ(config.size_buckets[0], 1024); + EXPECT_DOUBLE_EQ(config.size_buckets[1], 2048); + EXPECT_DOUBLE_EQ(config.size_buckets[2], 4096); +} + +//------------------------------------------------------------------------------ +// MetricsConfigLoader::loadWithDefaults Tests +//------------------------------------------------------------------------------ + +TEST(MetricsConfigLoaderTest, LoadWithDefaultsNoConfigNoEnv) { + MetricsConfig config = MetricsConfigLoader::loadWithDefaults(nullptr); + + MetricsConfig default_config = MetricsConfigLoader::getDefaultConfig(); + EXPECT_EQ(config.enabled, default_config.enabled); + EXPECT_EQ(config.http_port, default_config.http_port); +} + +TEST(MetricsConfigLoaderTest, LoadWithDefaultsConfigOverridesEnv) { + // Set environment variable + EnvVarGuard g1(config_keys::ENV_METRICS_HTTP_PORT, "7777"); + + // Create config with different value + Config file_config; + std::string json_content = R"({ + "metrics/http_port": 6666 + })"; + ASSERT_TRUE(file_config.load(json_content).ok()); + + // Config file should take priority over environment + MetricsConfig config = MetricsConfigLoader::loadWithDefaults(&file_config); + + EXPECT_EQ(config.http_port, 6666); +} + +TEST(MetricsConfigLoaderTest, LoadWithDefaultsEnvOverridesDefault) { + EnvVarGuard g1(config_keys::ENV_METRICS_HTTP_PORT, "5555"); + + MetricsConfig config = MetricsConfigLoader::loadWithDefaults(nullptr); + + EXPECT_EQ(config.http_port, 5555); +} + +//------------------------------------------------------------------------------ +// MetricsConfigLoader::validateConfig Tests +//------------------------------------------------------------------------------ + +TEST(MetricsConfigLoaderTest, ValidateConfigValidConfig) { + MetricsConfig config = MetricsConfigLoader::getDefaultConfig(); + std::string error_msg; + + EXPECT_TRUE(MetricsConfigLoader::validateConfig(config, &error_msg)); + EXPECT_TRUE(error_msg.empty()); +} + +TEST(MetricsConfigLoaderTest, ValidateConfigInvalidPortZero) { + MetricsConfig config = MetricsConfigLoader::getDefaultConfig(); + config.http_port = 0; + std::string error_msg; + + EXPECT_FALSE(MetricsConfigLoader::validateConfig(config, &error_msg)); + EXPECT_FALSE(error_msg.empty()); + EXPECT_NE(error_msg.find("port"), std::string::npos); +} + +TEST(MetricsConfigLoaderTest, ValidateConfigInvalidThreadsZero) { + MetricsConfig config = MetricsConfigLoader::getDefaultConfig(); + config.http_server_threads = 0; + std::string error_msg; + + EXPECT_FALSE(MetricsConfigLoader::validateConfig(config, &error_msg)); + EXPECT_FALSE(error_msg.empty()); + EXPECT_NE(error_msg.find("threads"), std::string::npos); +} + +TEST(MetricsConfigLoaderTest, ValidateConfigNoOutputFormatEnabled) { + MetricsConfig config = MetricsConfigLoader::getDefaultConfig(); + config.enable_prometheus = false; + config.enable_json = false; + std::string error_msg; + + EXPECT_FALSE(MetricsConfigLoader::validateConfig(config, &error_msg)); + EXPECT_FALSE(error_msg.empty()); + EXPECT_NE(error_msg.find("format"), std::string::npos); +} + +TEST(MetricsConfigLoaderTest, ValidateConfigUnsortedLatencyBuckets) { + MetricsConfig config = MetricsConfigLoader::getDefaultConfig(); + config.latency_buckets = {0.1, 0.05, 0.2}; // Not sorted + std::string error_msg; + + EXPECT_FALSE(MetricsConfigLoader::validateConfig(config, &error_msg)); + EXPECT_FALSE(error_msg.empty()); + EXPECT_NE(error_msg.find("Latency"), std::string::npos); +} + +TEST(MetricsConfigLoaderTest, ValidateConfigUnsortedSizeBuckets) { + MetricsConfig config = MetricsConfigLoader::getDefaultConfig(); + config.size_buckets = {1024, 512, 2048}; // Not sorted + std::string error_msg; + + EXPECT_FALSE(MetricsConfigLoader::validateConfig(config, &error_msg)); + EXPECT_FALSE(error_msg.empty()); + EXPECT_NE(error_msg.find("Size"), std::string::npos); +} + +TEST(MetricsConfigLoaderTest, ValidateConfigDuplicateBucketValues) { + MetricsConfig config = MetricsConfigLoader::getDefaultConfig(); + config.latency_buckets = {0.1, 0.1, 0.2}; // Duplicate values + std::string error_msg; + + EXPECT_FALSE(MetricsConfigLoader::validateConfig(config, &error_msg)); + EXPECT_FALSE(error_msg.empty()); +} + +TEST(MetricsConfigLoaderTest, ValidateConfigValidSortedBuckets) { + MetricsConfig config = MetricsConfigLoader::getDefaultConfig(); + config.latency_buckets = {0.001, 0.01, 0.1, 1.0}; + config.size_buckets = {1024, 10240, 102400}; + std::string error_msg; + + EXPECT_TRUE(MetricsConfigLoader::validateConfig(config, &error_msg)); + EXPECT_TRUE(error_msg.empty()); +} + +TEST(MetricsConfigLoaderTest, ValidateConfigNullErrorMsg) { + MetricsConfig config = MetricsConfigLoader::getDefaultConfig(); + config.http_port = 0; + + // Should not crash when error_msg is nullptr + EXPECT_FALSE(MetricsConfigLoader::validateConfig(config, nullptr)); +} + +//------------------------------------------------------------------------------ +// ConfigHelper Parsing Tests +//------------------------------------------------------------------------------ + +TEST(ConfigHelperTest, ParseBoolTrue) { + EXPECT_TRUE(ConfigHelper::parseBool("true", false)); + EXPECT_TRUE(ConfigHelper::parseBool("TRUE", false)); + EXPECT_TRUE(ConfigHelper::parseBool("True", false)); + EXPECT_TRUE(ConfigHelper::parseBool("1", false)); + EXPECT_TRUE(ConfigHelper::parseBool("yes", false)); + EXPECT_TRUE(ConfigHelper::parseBool("YES", false)); +} + +TEST(ConfigHelperTest, ParseBoolFalse) { + EXPECT_FALSE(ConfigHelper::parseBool("false", true)); + EXPECT_FALSE(ConfigHelper::parseBool("FALSE", true)); + EXPECT_FALSE(ConfigHelper::parseBool("False", true)); + EXPECT_FALSE(ConfigHelper::parseBool("0", true)); + EXPECT_FALSE(ConfigHelper::parseBool("no", true)); + EXPECT_FALSE(ConfigHelper::parseBool("NO", true)); +} + +TEST(ConfigHelperTest, ParseBoolInvalid) { + EXPECT_TRUE(ConfigHelper::parseBool("invalid", true)); + EXPECT_FALSE(ConfigHelper::parseBool("invalid", false)); +} + +TEST(ConfigHelperTest, ParseIntValid) { + EXPECT_EQ(ConfigHelper::parseInt("42", 0), 42); + EXPECT_EQ(ConfigHelper::parseInt("-10", 0), -10); + EXPECT_EQ(ConfigHelper::parseInt("0", 100), 0); +} + +TEST(ConfigHelperTest, ParseIntInvalid) { + EXPECT_EQ(ConfigHelper::parseInt("not-a-number", 99), 99); + EXPECT_EQ(ConfigHelper::parseInt("", 50), 50); +} + +TEST(ConfigHelperTest, ParsePortValid) { + EXPECT_EQ(ConfigHelper::parsePort("8080", 0), 8080); + EXPECT_EQ(ConfigHelper::parsePort("65535", 0), 65535); + EXPECT_EQ(ConfigHelper::parsePort("1", 0), 1); +} + +TEST(ConfigHelperTest, ParsePortInvalid) { + EXPECT_EQ(ConfigHelper::parsePort("not-a-port", 9100), 9100); + EXPECT_EQ(ConfigHelper::parsePort("", 9100), 9100); +} + +TEST(ConfigHelperTest, ParseDoubleArrayValid) { + auto result = ConfigHelper::parseDoubleArray("1.0,2.5,3.14"); + ASSERT_EQ(result.size(), 3); + EXPECT_DOUBLE_EQ(result[0], 1.0); + EXPECT_DOUBLE_EQ(result[1], 2.5); + EXPECT_DOUBLE_EQ(result[2], 3.14); +} + +TEST(ConfigHelperTest, ParseDoubleArrayEmpty) { + auto result = ConfigHelper::parseDoubleArray(""); + EXPECT_TRUE(result.empty()); +} + +TEST(ConfigHelperTest, ParseDoubleArraySingleValue) { + auto result = ConfigHelper::parseDoubleArray("42.0"); + ASSERT_EQ(result.size(), 1); + EXPECT_DOUBLE_EQ(result[0], 42.0); +} + +//------------------------------------------------------------------------------ +// Config Keys Constants Tests +//------------------------------------------------------------------------------ + +TEST(ConfigKeysTest, ConfigKeyConstants) { + // Verify config key constants are correctly defined + EXPECT_STREQ(config_keys::METRICS_ENABLED, "metrics/enabled"); + EXPECT_STREQ(config_keys::METRICS_HTTP_PORT, "metrics/http_port"); + EXPECT_STREQ(config_keys::METRICS_HTTP_HOST, "metrics/http_host"); + EXPECT_STREQ(config_keys::METRICS_HTTP_SERVER_THREADS, + "metrics/http_server_threads"); + EXPECT_STREQ(config_keys::METRICS_REPORT_INTERVAL, + "metrics/report_interval_seconds"); + EXPECT_STREQ(config_keys::METRICS_ENABLE_PROMETHEUS, + "metrics/enable_prometheus"); + EXPECT_STREQ(config_keys::METRICS_ENABLE_JSON, "metrics/enable_json"); + EXPECT_STREQ(config_keys::METRICS_LATENCY_BUCKETS, + "metrics/latency_buckets"); + EXPECT_STREQ(config_keys::METRICS_SIZE_BUCKETS, "metrics/size_buckets"); +} + +TEST(ConfigKeysTest, EnvVarConstants) { + // Verify environment variable constants are correctly defined + EXPECT_STREQ(config_keys::ENV_METRICS_ENABLED, "TENT_METRICS_ENABLED"); + EXPECT_STREQ(config_keys::ENV_METRICS_HTTP_PORT, "TENT_METRICS_HTTP_PORT"); + EXPECT_STREQ(config_keys::ENV_METRICS_HTTP_HOST, "TENT_METRICS_HTTP_HOST"); + EXPECT_STREQ(config_keys::ENV_METRICS_HTTP_SERVER_THREADS, + "TENT_METRICS_HTTP_SERVER_THREADS"); + EXPECT_STREQ(config_keys::ENV_METRICS_REPORT_INTERVAL, + "TENT_METRICS_REPORT_INTERVAL"); + EXPECT_STREQ(config_keys::ENV_METRICS_ENABLE_PROMETHEUS, + "TENT_METRICS_ENABLE_PROMETHEUS"); + EXPECT_STREQ(config_keys::ENV_METRICS_ENABLE_JSON, + "TENT_METRICS_ENABLE_JSON"); + EXPECT_STREQ(config_keys::ENV_METRICS_LATENCY_BUCKETS, + "TENT_METRICS_LATENCY_BUCKETS"); + EXPECT_STREQ(config_keys::ENV_METRICS_SIZE_BUCKETS, + "TENT_METRICS_SIZE_BUCKETS"); +} + +} // namespace +} // namespace tent +} // namespace mooncake diff --git a/mooncake-transfer-engine/tent/tests/tent_metrics_example.cpp b/mooncake-transfer-engine/tent/tests/tent_metrics_example.cpp new file mode 100644 index 00000000..05d3b747 --- /dev/null +++ b/mooncake-transfer-engine/tent/tests/tent_metrics_example.cpp @@ -0,0 +1,260 @@ +/** + * @file tent_metrics_example.cpp + * @brief Example demonstrating TENT metrics usage with HTTP server + * + * This example shows how to: + * 1. Initialize the TENT metrics system with HTTP server + * 2. Record transfer metrics with latency tracking + * 3. Use ScopedLatencyRecorder for automatic latency measurement + * 4. Use runtime enable/disable switch for dynamic control + * 5. Access metrics via HTTP endpoints: + * - GET /metrics - Prometheus format + * - GET /metrics/summary - Human readable summary + * - GET /metrics/json - JSON format + * - GET /health - Health check + * + * Performance optimization features: + * - Compile-time disable: Build with -DTENT_METRICS_ENABLED=OFF for zero + * overhead + * - Runtime disable: Use TentMetrics::setEnabled(false) for minimal overhead + * - Pre-allocated buffers: Reduced memory allocation in hot paths + */ + +#include +#include +#include +#include +#include + +#include "tent/metrics/tent_metrics.h" +#include "tent/metrics/config_loader.h" + +using namespace mooncake::tent; + +void printUsage(const char* program) { + std::cout << "Usage: " << program << " [OPTIONS]\n" + << "Options:\n" + << " --server Run in server mode (keep running)\n" + << " --port HTTP server port (default: 9100)\n" + << " --disabled Start with metrics collection disabled\n" + << " --help Show this help message\n"; +} + +int main(int argc, char* argv[]) { + std::cout << "=== TENT Metrics HTTP Server Example ===" << std::endl; + +#if TENT_METRICS_ENABLED + std::cout << "Compile-time metrics: ENABLED" << std::endl; +#else + std::cout << "Compile-time metrics: DISABLED (zero overhead)" << std::endl; +#endif + + // Parse command line arguments + bool server_mode = false; + bool start_disabled = false; + uint16_t port = 9100; + + for (int i = 1; i < argc; ++i) { + if (std::strcmp(argv[i], "--server") == 0) { + server_mode = true; + } else if (std::strcmp(argv[i], "--port") == 0 && i + 1 < argc) { + port = static_cast(std::atoi(argv[++i])); + } else if (std::strcmp(argv[i], "--disabled") == 0) { + start_disabled = true; + } else if (std::strcmp(argv[i], "--help") == 0) { + printUsage(argv[0]); + return 0; + } + } + + // 1. Load configuration (use defaults or from config file) + auto config = MetricsConfigLoader::loadWithDefaults(); + config.enabled = !start_disabled; // Can be disabled via --disabled flag + config.http_port = port; + config.http_host = "0.0.0.0"; + config.report_interval_seconds = + 0; // Disable periodic logging for this example + + std::cout << "\nConfiguration:" << std::endl; + std::cout << " - HTTP Server: " << config.http_host << ":" + << config.http_port << std::endl; + std::cout << " - Prometheus: " + << (config.enable_prometheus ? "enabled" : "disabled") + << std::endl; + std::cout << " - Runtime metrics: " + << (config.enabled ? "enabled" : "disabled") << std::endl; + + // 2. Initialize metrics system (this starts the HTTP server) + auto status = TentMetrics::instance().initialize(config); + if (!status.ok()) { + std::cerr << "Failed to initialize metrics: " << status.ToString() + << std::endl; + return 1; + } + + std::cout << "\nHTTP Server started. Available endpoints:" << std::endl; + std::cout << " - http://localhost:" << config.http_port + << "/metrics (Prometheus format)" << std::endl; + std::cout << " - http://localhost:" << config.http_port + << "/metrics/summary (Human readable)" << std::endl; + std::cout << " - http://localhost:" << config.http_port + << "/metrics/json (JSON format)" << std::endl; + std::cout << " - http://localhost:" << config.http_port + << "/health (Health check)" << std::endl; + + // 3. Simulate transfer operations with manual latency recording + std::cout << "\n--- Manual Latency Recording ---" << std::endl; + std::cout + << "Simulating transfer operations with explicit latency values..." + << std::endl; + + for (int i = 0; i < 10; ++i) { + // Simulate successful reads with manual latency + size_t read_bytes = 1024 * 1024 * (i + 1); // 1-10 MB + double read_latency = 0.001 + (i * 0.0005); // 1-5ms + TENT_RECORD_READ_COMPLETED(read_bytes, read_latency); + + // Simulate successful writes with manual latency + size_t write_bytes = 512 * 1024 * (i + 1); // 512KB - 5MB + double write_latency = 0.002 + (i * 0.0003); // 2-4.7ms + TENT_RECORD_WRITE_COMPLETED(write_bytes, write_latency); + + // Simulate some failures + if (i % 3 == 0) { + TENT_RECORD_READ_FAILED(1024); + } + if (i % 4 == 0) { + TENT_RECORD_WRITE_FAILED(512); + } + + std::cout << " Iteration " << (i + 1) << ": Read " + << read_bytes / 1024 / 1024 + << " MB (latency: " << read_latency * 1000 << "ms)" + << ", Write " << write_bytes / 1024 + << " KB (latency: " << write_latency * 1000 << "ms)" + << std::endl; + + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + + // 4. Demonstrate ScopedLatencyRecorder for automatic latency measurement + std::cout << "\n--- Automatic Latency Recording (ScopedLatencyRecorder) ---" + << std::endl; + std::cout << "Using RAII-style automatic latency measurement..." + << std::endl; + + // Example 1: Using TENT_SCOPED_READ_LATENCY macro + { + TENT_SCOPED_READ_LATENCY(2 * 1024 * 1024); // 2 MB read + // Simulate work (the latency is automatically measured) + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + std::cout << " Scoped read: 2 MB with ~5ms simulated work" + << std::endl; + } // Latency automatically recorded when scope ends + + // Example 2: Using TENT_SCOPED_WRITE_LATENCY macro + { + TENT_SCOPED_WRITE_LATENCY(1 * 1024 * 1024); // 1 MB write + // Simulate work + std::this_thread::sleep_for(std::chrono::milliseconds(3)); + std::cout << " Scoped write: 1 MB with ~3ms simulated work" + << std::endl; + } // Latency automatically recorded when scope ends + + // Example 3: Using ScopedLatencyRecorder directly with failure handling + { + ScopedLatencyRecorder recorder( + ScopedLatencyRecorder::OperationType::Read, 512 * 1024); + std::this_thread::sleep_for(std::chrono::milliseconds(2)); + // Simulate a failure condition + bool operation_failed = true; // Simulated failure + if (operation_failed) { + recorder.markFailed(); + std::cout << " Scoped read with failure: 512 KB marked as failed" + << std::endl; + } + } + + // Example 4: Multiple scoped operations in a loop + std::cout << "\n Running 5 scoped read operations..." << std::endl; + for (int i = 0; i < 5; ++i) { + TENT_SCOPED_READ_LATENCY(256 * 1024 * (i + 1)); // 256KB - 1.25MB + std::this_thread::sleep_for(std::chrono::milliseconds(1 + i)); + std::cout << " Scoped read " << (i + 1) << ": " << 256 * (i + 1) + << " KB with ~" << (1 + i) << "ms work" << std::endl; + } + + // 5. Demonstrate runtime enable/disable switch (only if not started with + // --disabled) + std::cout << "\n--- Runtime Enable/Disable Switch ---" << std::endl; + std::cout << "Current state: " + << (TentMetrics::isEnabled() ? "enabled" : "disabled") + << std::endl; + + if (!start_disabled) { + // Disable metrics at runtime + std::cout << "Disabling metrics collection..." << std::endl; + TentMetrics::setEnabled(false); + + // These calls will return immediately with minimal overhead + for (int i = 0; i < 1000; ++i) { + TENT_RECORD_READ_COMPLETED( + 1024, 0.001); // These are no-ops when disabled + } + std::cout << " 1000 record calls completed (no-op, metrics disabled)" + << std::endl; + + // Re-enable metrics + std::cout << "Re-enabling metrics collection..." << std::endl; + TentMetrics::setEnabled(true); + + // Now these will be recorded + TENT_RECORD_READ_COMPLETED(1024 * 1024, 0.005); + std::cout << " Recorded 1 MB read after re-enabling" << std::endl; + } else { + std::cout << " Skipping enable/disable demo (started with --disabled)" + << std::endl; + } + + // 6. Display summary + std::cout << "\n=== Metrics Summary ===" << std::endl; + std::cout << TentMetrics::instance().getSummaryString() << std::endl; + + // 7. Display Prometheus format (includes latency histogram buckets) + std::cout << "\n=== Prometheus Metrics (includes latency histograms) ===" + << std::endl; + std::string prometheus_metrics = + TentMetrics::instance().getPrometheusMetrics(); + if (!prometheus_metrics.empty()) { + std::cout << prometheus_metrics << std::endl; + } + + // 8. Keep server running for testing (or exit immediately) + if (server_mode) { + std::cout << "\nServer mode: Press Ctrl+C to exit..." << std::endl; + std::cout << "You can now query metrics via:" << std::endl; + std::cout << " curl http://localhost:" << config.http_port + << "/metrics # Prometheus format" << std::endl; + std::cout << " curl http://localhost:" << config.http_port + << "/metrics/summary # Human readable summary" << std::endl; + std::cout << " curl http://localhost:" << config.http_port + << "/metrics/json # JSON format" << std::endl; + std::cout << " curl http://localhost:" << config.http_port + << "/health # Health check" << std::endl; + + // Keep running until interrupted + while (true) { + std::this_thread::sleep_for(std::chrono::seconds(1)); + + // Simulate ongoing traffic + TENT_RECORD_READ_COMPLETED(1024 * 1024, 0.001); + TENT_RECORD_WRITE_COMPLETED(512 * 1024, 0.002); + } + } + + // 9. Cleanup + TentMetrics::instance().shutdown(); + + std::cout << "\nExample completed successfully!" << std::endl; + return 0; +}