[TENT] Add Metrics System with HTTP Server and Prometheus Integration (#1355)
* feat(metrics): add TENT metrics system with HTTP server and Prometheus integration - Add comprehensive metrics system based on yalantinglibs for monitoring data transfer performance - Implement HTTP server with endpoints for Prometheus, JSON, and human-readable metrics - Add compile-time and runtime performance optimization with zero-overhead when disabled - Integrate metrics into TransferEngine with automatic latency tracking - Add configuration loader supporting config files and environment variables - Include example application demonstrating metrics usage - Add documentation for metrics system configuration and usage Signed-off-by: staryxchen <staryxchen@tencent.com> * Update docs/source/design/tent/metrics.md Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> * refactor(metrics): simplify config loading with explicit priority - Replace indirect environment config loading with direct parsing - Implement clear priority: file config > environment variables > defaults - Add validation for environment variable values - Remove redundant default value comparisons Signed-off-by: staryxchen <staryxchen@tencent.com> * refactor(transfer_engine): extract metrics recording logic into dedicated method - Add recordTaskCompletionMetrics method to TransferEngineImpl class - Replace duplicate metrics recording code in getTransferStatus methods with calls to new method - Centralize task completion metrics logic for better maintainability Signed-off-by: staryxchen <staryxchen@tencent.com> * build(metrics): improve yalantinglibs dependency handling - Change warning to fatal error when TENT_METRICS_ENABLED is ON but yalantinglibs is missing - Provide clearer warning message when metrics are disabled Signed-off-by: staryxchen <staryxchen@tencent.com> * refactor(metrics): replace manual JSON construction with nlohmann/json library - Use nlohmann/json for cleaner and more maintainable JSON serialization - Remove manual string stream manipulation and formatting - Improve code readability and reduce error-prone manual concatenation Signed-off-by: staryxchen <staryxchen@tencent.com> * style: reformat code with clang-format Signed-off-by: staryxchen <staryxchen@tencent.com> * refactor(config): centralize parsing utilities in ConfigHelper - Move parsing functions from MetricsConfigLoader to ConfigHelper - Add applyEnvironmentOverrides method to reduce code duplication - Update includes and comments to reflect new structure Signed-off-by: staryxchen <staryxchen@tencent.com> * test: add unit tests for metrics config loader and reorganize test structure - Move examples directory to tests directory in CMakeLists.txt - Add comprehensive unit tests for MetricsConfigLoader functionality - Include tests for config parsing, environment variable loading, and validation - Rename and relocate tent_metrics_example.cpp to tests directory Signed-off-by: staryxchen <staryxchen@tencent.com> * style: reformat code lines for better readability Signed-off-by: staryxchen <staryxchen@tencent.com> * fix(build): remove redundant Asio dependency from metrics CMakeLists - Remove Asio dependency search and linking as yalantinglibs bundles it internally - Add clarifying comment about bundled Asio in yalantinglibs Signed-off-by: staryxchen <staryxchen@tencent.com> --------- Signed-off-by: staryxchen <staryxchen@tencent.com> Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
This commit is contained in:
Stary
GitHub
parent
4e3c1555d3
commit
abb429f935
|
|
@ -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<double>{/* 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<double> 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<double> 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
|
||||
```
|
||||
|
||||
|
|
@ -7,4 +7,5 @@ endif()
|
|||
set(CMAKE_POSITION_INDEPENDENT_CODE ON)
|
||||
|
||||
add_subdirectory(src)
|
||||
add_subdirectory(plugins)
|
||||
add_subdirectory(plugins)
|
||||
add_subdirectory(tests)
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -19,10 +19,12 @@
|
|||
#include <tent/common/status.h>
|
||||
#include <tent/common/types.h>
|
||||
|
||||
#include <algorithm>
|
||||
#include <filesystem>
|
||||
#include <fstream>
|
||||
#include <mutex>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
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<double> parseDoubleArray(const std::string& str);
|
||||
};
|
||||
|
||||
} // namespace tent
|
||||
|
|
|
|||
|
|
@ -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<double> latency_buckets;
|
||||
std::vector<double> 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
|
||||
|
|
@ -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 <atomic>
|
||||
#include <condition_variable>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
#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 <ylt/metric.hpp>
|
||||
#include <ylt/coro_http/coro_http_server.hpp>
|
||||
#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<bool> runtime_enabled_{true};
|
||||
|
||||
std::atomic<bool> initialized_{false};
|
||||
MetricsConfig config_;
|
||||
|
||||
#if TENT_METRICS_ENABLED
|
||||
// Initialize HTTP server with endpoints
|
||||
void initHttpServer();
|
||||
|
||||
// HTTP server for metrics endpoint
|
||||
std::unique_ptr<coro_http::coro_http_server> http_server_;
|
||||
|
||||
// Periodic metric reporting thread
|
||||
std::thread metric_report_thread_;
|
||||
std::atomic<bool> metric_report_running_{false};
|
||||
std::mutex metric_report_mutex_;
|
||||
std::condition_variable metric_report_cv_;
|
||||
|
||||
// Counters - stored as pointers for unified management
|
||||
std::vector<ylt::metric::counter_t*> 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<ylt::metric::histogram_t*> histograms_;
|
||||
// Store bucket boundaries separately since ylt histogram doesn't expose
|
||||
// them publicly
|
||||
std::vector<std::vector<double>> 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<double> 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<double> 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<double>(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
|
||||
|
|
@ -17,6 +17,7 @@
|
|||
|
||||
#include <array>
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <cstddef>
|
||||
#include <cstdint>
|
||||
#include <memory>
|
||||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
|
|
@ -14,6 +14,10 @@
|
|||
|
||||
#include "tent/common/config.h"
|
||||
|
||||
#include <glog/logging.h>
|
||||
|
||||
#include <sstream>
|
||||
|
||||
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<uint16_t>(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<double> ConfigHelper::parseDoubleArray(const std::string& str) {
|
||||
std::vector<double> 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
|
||||
|
|
@ -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()
|
||||
|
|
@ -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 <glog/logging.h>
|
||||
|
||||
#include <cstdlib>
|
||||
|
||||
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<uint16_t>(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<uint16_t>(
|
||||
config.get(config_keys::METRICS_HTTP_PORT,
|
||||
static_cast<int>(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<uint16_t>(
|
||||
config.get(config_keys::METRICS_HTTP_SERVER_THREADS,
|
||||
static_cast<int>(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<double>(config_keys::METRICS_LATENCY_BUCKETS);
|
||||
if (!latency_buckets_array.empty()) {
|
||||
metrics_config.latency_buckets = latency_buckets_array;
|
||||
}
|
||||
|
||||
auto size_buckets_array =
|
||||
config.getArray<double>(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<uint16_t>(
|
||||
config->get(config_keys::METRICS_HTTP_PORT,
|
||||
static_cast<int>(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<uint16_t>(
|
||||
config->get(config_keys::METRICS_HTTP_SERVER_THREADS,
|
||||
static_cast<int>(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<double>(config_keys::METRICS_LATENCY_BUCKETS);
|
||||
if (!latency_buckets_array.empty()) {
|
||||
metrics_config.latency_buckets = latency_buckets_array;
|
||||
}
|
||||
|
||||
auto size_buckets_array =
|
||||
config->getArray<double>(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
|
||||
|
|
@ -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 <glog/logging.h>
|
||||
#include <tent/thirdparty/nlohmann/json.h>
|
||||
#include <sstream>
|
||||
#include <iomanip>
|
||||
|
||||
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<double> 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<std::mutex> 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<coro_http_server>(
|
||||
config_.http_server_threads, config_.http_port);
|
||||
|
||||
// Register /metrics endpoint for Prometheus
|
||||
http_server_->set_http_handler<GET>(
|
||||
"/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<GET>(
|
||||
"/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<GET>(
|
||||
"/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<GET>(
|
||||
"/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<double>(bytes));
|
||||
read_requests_total_.inc();
|
||||
read_size_.observe(static_cast<int64_t>(bytes));
|
||||
if (latency_seconds > 0.0) {
|
||||
// Convert seconds to microseconds for histogram (int64_t internally)
|
||||
int64_t latency_us = static_cast<int64_t>(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<double>(bytes));
|
||||
write_requests_total_.inc();
|
||||
write_size_.observe(static_cast<int64_t>(bytes));
|
||||
if (latency_seconds > 0.0) {
|
||||
// Convert seconds to microseconds for histogram (int64_t internally)
|
||||
int64_t latency_us = static_cast<int64_t>(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<int64_t>(
|
||||
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<uint64_t>(read_reqs) << " reqs, "
|
||||
<< static_cast<uint64_t>(read_fails) << " fails) | "
|
||||
<< "Write: " << formatBytes(write_bytes) << " ("
|
||||
<< static_cast<uint64_t>(write_reqs) << " reqs, "
|
||||
<< static_cast<uint64_t>(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
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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<ProxyManager>(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<TransportType, size_t> 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<double>(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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
@ -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 <gtest/gtest.h>
|
||||
|
||||
#include <cstdlib>
|
||||
#include <string>
|
||||
|
||||
#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
|
||||
|
|
@ -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 <chrono>
|
||||
#include <cstdlib>
|
||||
#include <cstring>
|
||||
#include <iostream>
|
||||
#include <thread>
|
||||
|
||||
#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 <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<uint16_t>(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;
|
||||
}
|
||||
Loading…
Reference in New Issue