rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-28 22:56:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: batch eligible OTLP metric rows

Browse Source 
Share the pending-row batching contract and lifecycle across Prometheus, OTLP HTTP, and OTel Arrow in `src/servers/` and `src/frontend/`.

Add OTLP batching configuration, protocol-aware submit observability, integration coverage in `tests-integration/`, and generated config/dashboard updates.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
This commit is contained in:
Lei, HUANG
2026-09-09 17:01:12 +08:00
parent 0d5d97e7cb
commit 731834a53f
28 changed files with 2436 additions and 270 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/config.md
+6 -4
View File
@@ -33,7 +33,7 @@
| `runtime.experimental_workload_scheduler.sample_every_polls` | Integer | `16` | Number of polls between scheduler fairness samples. Must be greater than zero. |
| `http` | -- | -- | The HTTP server options. |
| `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. |
| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.<br/>When Prometheus pending-row batching is enabled, a nonzero timeout less than or equal to the<br/>`prom_store.pending_rows_flush_interval` plus 1 second is adjusted to that value. |
| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.<br/>When synchronous pending-row batching serves an enabled Prometheus or OTLP HTTP endpoint,<br/>a nonzero timeout less than or equal to `prom_store.pending_rows_flush_interval` plus 1 second<br/>is adjusted to that value. |
| `http.body_limit` | String | `64MB` | HTTP request body limit.<br/>The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.<br/>Set to 0 to disable limit. |
| `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default<br/>This allows browser to access http APIs without CORS restrictions |
| `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. |
@@ -78,6 +78,7 @@
| `jaeger.enable` | Bool | `true` | Whether to enable Jaeger protocol in HTTP API. |
| `otlp` | -- | -- | OpenTelemetry protocol options. |
| `otlp.enable` | Bool | `true` | Whether to enable OpenTelemetry protocol in HTTP API. |
| `otlp.enable_metrics_batching` | Bool | `false` | Enable pending-row batching for eligible OTLP metrics over HTTP and OTel Arrow.<br/>Disabled by default. Requires Metric Engine, valid nonzero `prom_store.pending_rows_*` tuning,<br/>and `PENDING_ROWS_BATCH_SYNC=true`; otherwise requests use the original Inserter path.<br/>Legacy OTLP metric tables and ineligible converted request shapes, such as native histograms or<br/>mixed scalar/histogram requests, fall back to the Inserter.<br/>This startup option requires a restart. Before disabling it, drain pending rows and inflight<br/>requests. A failed multi-region flush may partially write rows, so retries can create duplicates. |
| `otlp.experimental_enable_exponential_histogram` | Bool | `false` | Experimental: enable cumulative OTLP exponential histogram ingestion. |
| `otlp.trace_ingest_chunk_size` | Integer | `512` | Maximum spans per trace ingest chunk. Set to 0 to disable splitting. |
| `otlp.experimental_enable_resource_info` | Bool | `false` | Whether to synthesize the `greptime_otel_resource_info` table from OTLP metric<br/>resource attributes, so metrics-only services reach the semantic graph. |
@@ -86,7 +87,7 @@
| `prom_store.with_metric_engine` | Bool | `true` | Whether to store the data from Prometheus remote write in metric engine. |
| `prom_store.prom_validation_mode` | String | `strict` | Whether to enable validation for Prometheus remote write requests.<br/>Available options:<br/>- strict: deny invalid UTF-8 strings (default).<br/>- lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).<br/>- unchecked: do not valid strings. |
| `prom_store.experimental_enable_prometheus_native_histogram` | Bool | `false` | Experimental: enable Prometheus remote write v2 native histogram ingestion. |
| `prom_store.pending_rows_flush_interval` | String | `0s` | Interval to flush pending rows batcher.<br/>Set to "0s" to disable batching mode in Prometheus Remote Write endpoint |
| `prom_store.pending_rows_flush_interval` | String | `0s` | Interval to flush pending rows batcher.<br/>Set to "0s" to disable batching for Prometheus Remote Write and force OTLP to use its Inserter. |
| `prom_store.max_batch_rows` | Integer | `100000` | Max rows per pending batch before triggering a flush. |
| `prom_store.max_concurrent_flushes` | Integer | `256` | Max number of concurrent batch flushes. |
| `prom_store.worker_channel_capacity` | Integer | `65526` | Capacity of the pending batch worker channel. |
@@ -272,7 +273,7 @@
| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 2). |
| `http` | -- | -- | The HTTP server options. |
| `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. |
| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.<br/>When Prometheus pending-row batching is enabled, a nonzero timeout less than or equal to the<br/>`prom_store.pending_rows_flush_interval` plus 1 second is adjusted to that value. |
| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.<br/>When synchronous pending-row batching serves an enabled Prometheus or OTLP HTTP endpoint,<br/>a nonzero timeout less than or equal to `prom_store.pending_rows_flush_interval` plus 1 second<br/>is adjusted to that value. |
| `http.body_limit` | String | `64MB` | HTTP request body limit.<br/>The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.<br/>Set to 0 to disable limit. |
| `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default<br/>This allows browser to access http APIs without CORS restrictions |
| `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. |
@@ -329,6 +330,7 @@
| `jaeger.enable` | Bool | `true` | Whether to enable Jaeger protocol in HTTP API. |
| `otlp` | -- | -- | OpenTelemetry protocol options. |
| `otlp.enable` | Bool | `true` | Whether to enable OpenTelemetry protocol in HTTP API. |
| `otlp.enable_metrics_batching` | Bool | `false` | Enable pending-row batching for eligible OTLP metrics over HTTP and OTel Arrow.<br/>Disabled by default. Requires Metric Engine, valid nonzero `prom_store.pending_rows_*` tuning,<br/>and `PENDING_ROWS_BATCH_SYNC=true`; otherwise requests use the original Inserter path.<br/>Legacy OTLP metric tables and ineligible converted request shapes, such as native histograms or<br/>mixed scalar/histogram requests, fall back to the Inserter.<br/>This startup option requires a restart. Before disabling it, drain pending rows and inflight<br/>requests. A failed multi-region flush may partially write rows, so retries can create duplicates. |
| `otlp.experimental_enable_exponential_histogram` | Bool | `false` | Experimental: enable cumulative OTLP exponential histogram ingestion. |
| `otlp.trace_ingest_chunk_size` | Integer | `512` | Maximum spans per trace ingest chunk. Set to 0 to disable splitting. |
| `otlp.experimental_enable_resource_info` | Bool | `false` | Whether to synthesize the `greptime_otel_resource_info` table from OTLP metric<br/>resource attributes, so metrics-only services reach the semantic graph. |
@@ -337,7 +339,7 @@
| `prom_store.with_metric_engine` | Bool | `true` | Whether to store the data from Prometheus remote write in metric engine. |
| `prom_store.prom_validation_mode` | String | `strict` | Whether to enable validation for Prometheus remote write requests.<br/>Available options:<br/>- strict: deny invalid UTF-8 strings (default).<br/>- lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).<br/>- unchecked: do not valid strings. |
| `prom_store.experimental_enable_prometheus_native_histogram` | Bool | `false` | Experimental: enable Prometheus remote write v2 native histogram ingestion. |
| `prom_store.pending_rows_flush_interval` | String | `0s` | Interval to flush pending rows batcher.<br/>Set to "0s" to disable batching mode in Prometheus Remote Write endpoint |
| `prom_store.pending_rows_flush_interval` | String | `0s` | Interval to flush pending rows batcher.<br/>Set to "0s" to disable batching for Prometheus Remote Write and force OTLP to use its Inserter. |
| `prom_store.max_batch_rows` | Integer | `100000` | Max rows per pending batch before triggering a flush. |
| `prom_store.max_concurrent_flushes` | Integer | `256` | Max number of concurrent batch flushes. |
| `prom_store.worker_channel_capacity` | Integer | `65526` | Capacity of the pending batch worker channel. |
config/frontend.example.toml
+12 -3
View File
@@ -52,8 +52,9 @@ default_column_prefix = "greptime"
## The address to bind the HTTP server.
addr = "127.0.0.1:4000"
## HTTP request timeout. Set to 0 to disable timeout.
## When Prometheus pending-row batching is enabled, a nonzero timeout less than or equal to the
## `prom_store.pending_rows_flush_interval` plus 1 second is adjusted to that value.
## When synchronous pending-row batching serves an enabled Prometheus or OTLP HTTP endpoint,
## a nonzero timeout less than or equal to `prom_store.pending_rows_flush_interval` plus 1 second
## is adjusted to that value.
timeout = "0s"
## HTTP request body limit.
## The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.
@@ -235,6 +236,14 @@ enable = true
[otlp]
## Whether to enable OpenTelemetry protocol in HTTP API.
enable = true
## Enable pending-row batching for eligible OTLP metrics over HTTP and OTel Arrow.
## Disabled by default. Requires Metric Engine, valid nonzero `prom_store.pending_rows_*` tuning,
## and `PENDING_ROWS_BATCH_SYNC=true`; otherwise requests use the original Inserter path.
## Legacy OTLP metric tables and ineligible converted request shapes, such as native histograms or
## mixed scalar/histogram requests, fall back to the Inserter.
## This startup option requires a restart. Before disabling it, drain pending rows and inflight
## requests. A failed multi-region flush may partially write rows, so retries can create duplicates.
enable_metrics_batching = false
## Experimental: enable cumulative OTLP exponential histogram ingestion.
experimental_enable_exponential_histogram = false
## Maximum spans per trace ingest chunk. Set to 0 to disable splitting.
@@ -258,7 +267,7 @@ prom_validation_mode = "strict"
## Experimental: enable Prometheus remote write v2 native histogram ingestion.
experimental_enable_prometheus_native_histogram = false
## Interval to flush pending rows batcher.
## Set to "0s" to disable batching mode in Prometheus Remote Write endpoint
## Set to "0s" to disable batching for Prometheus Remote Write and force OTLP to use its Inserter.
#+pending_rows_flush_interval = "0s"
## Max rows per pending batch before triggering a flush.
#+max_batch_rows = 100000
config/standalone.example.toml
+12 -3
View File
@@ -77,8 +77,9 @@ max_concurrent_queries = 0
## The address to bind the HTTP server.
addr = "127.0.0.1:4000"
## HTTP request timeout. Set to 0 to disable timeout.
## When Prometheus pending-row batching is enabled, a nonzero timeout less than or equal to the
## `prom_store.pending_rows_flush_interval` plus 1 second is adjusted to that value.
## When synchronous pending-row batching serves an enabled Prometheus or OTLP HTTP endpoint,
## a nonzero timeout less than or equal to `prom_store.pending_rows_flush_interval` plus 1 second
## is adjusted to that value.
timeout = "0s"
## HTTP request body limit.
## The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.
@@ -214,6 +215,14 @@ enable = true
[otlp]
## Whether to enable OpenTelemetry protocol in HTTP API.
enable = true
## Enable pending-row batching for eligible OTLP metrics over HTTP and OTel Arrow.
## Disabled by default. Requires Metric Engine, valid nonzero `prom_store.pending_rows_*` tuning,
## and `PENDING_ROWS_BATCH_SYNC=true`; otherwise requests use the original Inserter path.
## Legacy OTLP metric tables and ineligible converted request shapes, such as native histograms or
## mixed scalar/histogram requests, fall back to the Inserter.
## This startup option requires a restart. Before disabling it, drain pending rows and inflight
## requests. A failed multi-region flush may partially write rows, so retries can create duplicates.
enable_metrics_batching = false
## Experimental: enable cumulative OTLP exponential histogram ingestion.
experimental_enable_exponential_histogram = false
## Maximum spans per trace ingest chunk. Set to 0 to disable splitting.
@@ -237,7 +246,7 @@ prom_validation_mode = "strict"
## Experimental: enable Prometheus remote write v2 native histogram ingestion.
experimental_enable_prometheus_native_histogram = false
## Interval to flush pending rows batcher.
## Set to "0s" to disable batching mode in Prometheus Remote Write endpoint
## Set to "0s" to disable batching for Prometheus Remote Write and force OTLP to use its Inserter.
#+pending_rows_flush_interval = "0s"
## Max rows per pending batch before triggering a flush.
#+max_batch_rows = 100000
grafana/dashboards/metrics/cluster/dashboard.json
+2 -2
View File
@@ -4362,7 +4362,7 @@
"type": "prometheus",
"uid": "${metrics}"
},
"description": "Prometheus remote-write pending rows, batches, and workers. Rising pending rows indicate remote-write buffering backlog.",
"description": "Pending metric rows, batches, and workers. Rising pending rows indicate metric-row batching backlog.",
"fieldConfig": {
"defaults": {
"color": {
@@ -4482,7 +4482,7 @@
"refId": "C"
}
],
"title": "Prom Store Backlog",
"title": "Metric Row Batching Backlog",
"type": "timeseries"
},
{
grafana/dashboards/metrics/cluster/dashboard.md
+1 -1
View File
@@ -46,7 +46,7 @@
| Query Memory Usage | `sum by (instance, pod) (greptime_query_memory_pool_usage_bytes)` | `timeseries` | Query memory pool usage. Use this with query memory rejection panels to diagnose query saturation. | `prometheus` | `bytes` | `[{{instance}}]-[{{pod}}]` |
| Scan and Compaction Memory | `sum by (instance, pod) (greptime_mito_scan_memory_usage_bytes{instance=~"$datanode"})`<br/>`sum by (instance, pod) (greptime_mito_compaction_memory_in_use_bytes{instance=~"$datanode"})`<br/>`sum by (instance, pod) (greptime_mito_compaction_memory_limit_bytes{instance=~"$datanode"})` | `timeseries` | Datanode scan memory usage and compaction memory utilization. | `prometheus` | `bytes` | `scan-[{{instance}}]-[{{pod}}]` |
| Write Buffer and Active Stalling | `sum by (instance, pod) (greptime_mito_write_buffer_bytes{instance=~"$datanode"})`<br/>`sum by (instance, pod) (greptime_mito_write_stalling_count{instance=~"$datanode"})` | `timeseries` | Mito write buffer bytes and active write-stalling gauges. Growth here indicates write-path backpressure. | `prometheus` | `bytes` | `buffer-[{{instance}}]-[{{pod}}]` |
| Prom Store Backlog | `sum by (instance, pod) (greptime_prom_store_pending_rows{instance=~"$frontend"})`<br/>`sum by (instance, pod) (greptime_prom_store_pending_batches{instance=~"$frontend"})`<br/>`sum by (instance, pod) (greptime_prom_store_pending_workers{instance=~"$frontend"})` | `timeseries` | Prometheus remote-write pending rows, batches, and workers. Rising pending rows indicate remote-write buffering backlog. | `prometheus` | `short` | `rows-[{{instance}}]-[{{pod}}]` |
| Metric Row Batching Backlog | `sum by (instance, pod) (greptime_prom_store_pending_rows{instance=~"$frontend"})`<br/>`sum by (instance, pod) (greptime_prom_store_pending_batches{instance=~"$frontend"})`<br/>`sum by (instance, pod) (greptime_prom_store_pending_workers{instance=~"$frontend"})` | `timeseries` | Pending metric rows, batches, and workers. Rising pending rows indicate metric-row batching backlog. | `prometheus` | `short` | `rows-[{{instance}}]-[{{pod}}]` |
| Inflight Flush and Compaction | `sum by (instance, pod) (greptime_mito_inflight_flush_count{instance=~"$datanode"})`<br/>`sum by (instance, pod) (greptime_mito_inflight_compaction_count{instance=~"$datanode"})` | `timeseries` | Current in-flight flush and compaction tasks on datanodes. | `prometheus` | `short` | `flush-[{{instance}}]-[{{pod}}]` |
# Resources
| Title | Query | Type | Description | Datasource | Unit | Legend Format |
grafana/dashboards/metrics/cluster/dashboard.yaml
+2 -2
View File
@@ -632,9 +632,9 @@ groups:
type: prometheus
uid: ${metrics}
legendFormat: stalling-[{{instance}}]-[{{pod}}]
- title: Prom Store Backlog
- title: Metric Row Batching Backlog
type: timeseries
description: Prometheus remote-write pending rows, batches, and workers. Rising pending rows indicate remote-write buffering backlog.
description: Pending metric rows, batches, and workers. Rising pending rows indicate metric-row batching backlog.
unit: short
queries:
- expr: sum by (instance, pod) (greptime_prom_store_pending_rows{instance=~"$frontend"})
grafana/dashboards/metrics/standalone/dashboard.json
+2 -2
View File
@@ -4362,7 +4362,7 @@
"type": "prometheus",
"uid": "${metrics}"
},
"description": "Prometheus remote-write pending rows, batches, and workers. Rising pending rows indicate remote-write buffering backlog.",
"description": "Pending metric rows, batches, and workers. Rising pending rows indicate metric-row batching backlog.",
"fieldConfig": {
"defaults": {
"color": {
@@ -4482,7 +4482,7 @@
"refId": "C"
}
],
"title": "Prom Store Backlog",
"title": "Metric Row Batching Backlog",
"type": "timeseries"
},
{
grafana/dashboards/metrics/standalone/dashboard.md
+1 -1
View File
@@ -46,7 +46,7 @@
| Query Memory Usage | `sum by (instance, pod) (greptime_query_memory_pool_usage_bytes)` | `timeseries` | Query memory pool usage. Use this with query memory rejection panels to diagnose query saturation. | `prometheus` | `bytes` | `[{{instance}}]-[{{pod}}]` |
| Scan and Compaction Memory | `sum by (instance, pod) (greptime_mito_scan_memory_usage_bytes)`<br/>`sum by (instance, pod) (greptime_mito_compaction_memory_in_use_bytes)`<br/>`sum by (instance, pod) (greptime_mito_compaction_memory_limit_bytes)` | `timeseries` | Datanode scan memory usage and compaction memory utilization. | `prometheus` | `bytes` | `scan-[{{instance}}]-[{{pod}}]` |
| Write Buffer and Active Stalling | `sum by (instance, pod) (greptime_mito_write_buffer_bytes)`<br/>`sum by (instance, pod) (greptime_mito_write_stalling_count)` | `timeseries` | Mito write buffer bytes and active write-stalling gauges. Growth here indicates write-path backpressure. | `prometheus` | `bytes` | `buffer-[{{instance}}]-[{{pod}}]` |
| Prom Store Backlog | `sum by (instance, pod) (greptime_prom_store_pending_rows)`<br/>`sum by (instance, pod) (greptime_prom_store_pending_batches)`<br/>`sum by (instance, pod) (greptime_prom_store_pending_workers)` | `timeseries` | Prometheus remote-write pending rows, batches, and workers. Rising pending rows indicate remote-write buffering backlog. | `prometheus` | `short` | `rows-[{{instance}}]-[{{pod}}]` |
| Metric Row Batching Backlog | `sum by (instance, pod) (greptime_prom_store_pending_rows)`<br/>`sum by (instance, pod) (greptime_prom_store_pending_batches)`<br/>`sum by (instance, pod) (greptime_prom_store_pending_workers)` | `timeseries` | Pending metric rows, batches, and workers. Rising pending rows indicate metric-row batching backlog. | `prometheus` | `short` | `rows-[{{instance}}]-[{{pod}}]` |
| Inflight Flush and Compaction | `sum by (instance, pod) (greptime_mito_inflight_flush_count)`<br/>`sum by (instance, pod) (greptime_mito_inflight_compaction_count)` | `timeseries` | Current in-flight flush and compaction tasks on datanodes. | `prometheus` | `short` | `flush-[{{instance}}]-[{{pod}}]` |
# Resources
| Title | Query | Type | Description | Datasource | Unit | Legend Format |
grafana/dashboards/metrics/standalone/dashboard.yaml
+2 -2
View File
@@ -632,9 +632,9 @@ groups:
type: prometheus
uid: ${metrics}
legendFormat: stalling-[{{instance}}]-[{{pod}}]
- title: Prom Store Backlog
- title: Metric Row Batching Backlog
type: timeseries
description: Prometheus remote-write pending rows, batches, and workers. Rising pending rows indicate remote-write buffering backlog.
description: Pending metric rows, batches, and workers. Rising pending rows indicate metric-row batching backlog.
unit: short
queries:
- expr: sum by (instance, pod) (greptime_prom_store_pending_rows)
src/cmd/src/standalone.rs
+1
View File
@@ -1155,6 +1155,7 @@ mod tests {
fn test_toml() {
let opts = StandaloneOptions::default();
let toml_string = toml::to_string(&opts).unwrap();
assert!(toml_string.contains("enable_metrics_batching = false"));
assert!(toml_string.contains("experimental_enable_exponential_histogram = false"));
let parsed: StandaloneOptions = toml::from_str(&toml_string).unwrap();
assert_eq!(parsed.otlp, opts.otlp);
src/frontend/src/frontend.rs
+1
View File
@@ -222,6 +222,7 @@ mod tests {
fn test_toml() {
let opts = FrontendOptions::default();
let toml_string = toml::to_string(&opts).unwrap();
assert!(toml_string.contains("enable_metrics_batching = false"));
assert!(toml_string.contains("experimental_enable_exponential_histogram = false"));
let parsed: FrontendOptions = toml::from_str(&toml_string).unwrap();
assert_eq!(parsed.otlp, opts.otlp);
src/frontend/src/instance/otlp.rs
+360 -20
View File
@@ -35,8 +35,12 @@ use pipeline::{GreptimePipelineParams, PipelineWay};
use servers::error::{self, AuthSnafu, Result as ServerResult};
use servers::http::prom_store::PHYSICAL_TABLE_PARAM;
use servers::interceptor::{OpenTelemetryProtocolInterceptor, OpenTelemetryProtocolInterceptorRef};
use servers::metrics::METRIC_ROW_BATCH_ELIGIBILITY_FALLBACKS;
use servers::otlp;
use servers::otlp::trace::span::TraceSpanGroup;
use servers::pending_rows_batcher::{
MetricRowBatchProtocol, MetricRowBatcherRef, is_scalar_metric_batchable,
};
use servers::query_handler::{
MetricsIngestOutcome, OpenTelemetryProtocolHandler, PipelineHandlerRef, TraceIngestOutcome,
};
@@ -89,6 +93,7 @@ impl OpenTelemetryProtocolHandler for Instance {
async fn metrics(
&self,
request: ExportMetricsServiceRequest,
metric_row_batcher: Option<MetricRowBatcherRef>,
ctx: QueryContextRef,
) -> ServerResult<MetricsIngestOutcome> {
self.plugins
@@ -160,23 +165,40 @@ impl OpenTelemetryProtocolHandler for Instance {
Arc::new(c)
};
// OTLP tables have one sample field in both the legacy and physical paths.
let output = if metric_ctx.is_legacy || !metric_ctx.with_metric_engine {
self.handle_row_inserts(requests, ctx.clone(), false, true)
.await
.map_err(BoxedError::new)
.context(error::ExecuteGrpcQuerySnafu)
} else {
let physical_table = ctx
.extension(PHYSICAL_TABLE_PARAM)
.unwrap_or(GREPTIME_PHYSICAL_TABLE)
.to_string();
self.handle_metric_row_inserts(requests, ctx.clone(), physical_table)
.await
.map_err(BoxedError::new)
.context(error::ExecuteGrpcQuerySnafu)
}?;
outcome.write_cost = output.meta.cost;
let batching_candidate =
!metric_ctx.is_legacy && metric_ctx.with_metric_engine && metric_row_batcher.is_some();
let batchable = batching_candidate && is_scalar_metric_batchable(&requests)?;
let write_cost =
if let Some(metric_row_batcher) = metric_row_batcher.as_ref().filter(|_| batchable) {
metric_row_batcher
.submit(requests, ctx.clone(), MetricRowBatchProtocol::Otlp)
.await?;
0
} else {
if batching_candidate {
METRIC_ROW_BATCH_ELIGIBILITY_FALLBACKS
.with_label_values(&[MetricRowBatchProtocol::Otlp.as_str()])
.inc();
}
// OTLP tables have one sample field in both the legacy and physical paths.
let output = if metric_ctx.is_legacy || !metric_ctx.with_metric_engine {
self.handle_row_inserts(requests, ctx.clone(), false, true)
.await
.map_err(BoxedError::new)
.context(error::ExecuteGrpcQuerySnafu)
} else {
let physical_table = ctx
.extension(PHYSICAL_TABLE_PARAM)
.unwrap_or(GREPTIME_PHYSICAL_TABLE)
.to_string();
self.handle_metric_row_inserts(requests, ctx.clone(), physical_table)
.await
.map_err(BoxedError::new)
.context(error::ExecuteGrpcQuerySnafu)
}?;
output.meta.cost
};
outcome.write_cost = write_cost;
// Derived enrichment, written after the metric data is committed:
// failing here would make the client retry data the server already
@@ -321,16 +343,334 @@ impl OpenTelemetryProtocolHandler for Instance {
#[cfg(test)]
mod tests {
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use auth::{PermissionTableTarget, PermissionTableTargets};
use api::v1::RowInsertRequests;
use api::v1::meta::Role;
use async_trait::async_trait;
use auth::{
PermissionChecker, PermissionCheckerRef, PermissionReq, PermissionResp,
PermissionTableTarget, PermissionTableTargets, UserInfoRef,
};
use common_base::Plugins;
use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
use common_error::ext::ErrorExt;
use meta_client::client::MetaClientBuilder;
use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
use opentelemetry_proto::tonic::common::v1::{AnyValue, KeyValue, any_value};
use opentelemetry_proto::tonic::resource::v1::Resource;
use opentelemetry_proto::tonic::trace::v1::{ResourceSpans, ScopeSpans, Span};
use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest;
use otel_arrow_rust::proto::opentelemetry::common::v1::{
AnyValue as MetricAnyValue, KeyValue as MetricKeyValue, any_value as metric_any_value,
};
use otel_arrow_rust::proto::opentelemetry::metrics::v1::number_data_point::Value;
use otel_arrow_rust::proto::opentelemetry::metrics::v1::{
AggregationTemporality, ExponentialHistogram, ExponentialHistogramDataPoint, Gauge, Metric,
NumberDataPoint, ResourceMetrics, ScopeMetrics, metric,
};
use otel_arrow_rust::proto::opentelemetry::resource::v1::Resource as MetricResource;
use servers::metrics::METRIC_ROW_BATCH_ELIGIBILITY_FALLBACKS;
use servers::pending_rows_batcher::{MetricRowBatchProtocol, MetricRowBatcher};
use session::context::QueryContext;
use session::protocol_ctx::{OtlpMetricCtx, ProtocolCtx};
use super::trace_permission_targets;
use super::*;
use crate::frontend::FrontendOptions;
use crate::instance::builder::FrontendBuilder;
static FALLBACK_METRIC_ASSERTION_LOCK: tokio::sync::Mutex<()> =
tokio::sync::Mutex::const_new(());
#[derive(Default)]
struct FakeMetricRowBatcher {
requests: Mutex<Vec<RowInsertRequests>>,
protocols: Mutex<Vec<MetricRowBatchProtocol>>,
flush_error: Option<&'static str>,
accepted_rows: u64,
}
impl FakeMetricRowBatcher {
fn successful(accepted_rows: u64) -> Self {
Self {
accepted_rows,
..Default::default()
}
}
fn failing(message: &'static str) -> Self {
Self {
flush_error: Some(message),
..Default::default()
}
}
}
#[async_trait]
impl MetricRowBatcher for FakeMetricRowBatcher {
async fn submit(
&self,
requests: RowInsertRequests,
_ctx: QueryContextRef,
protocol: MetricRowBatchProtocol,
) -> ServerResult<u64> {
self.requests.lock().unwrap().push(requests);
self.protocols.lock().unwrap().push(protocol);
if let Some(message) = self.flush_error {
return Err(servers::error::InternalSnafu {
err_msg: message.to_string(),
}
.build());
}
Ok(self.accepted_rows)
}
}
async fn test_instance() -> Instance {
test_instance_with(FrontendOptions::default(), Plugins::new()).await
}
async fn test_instance_with(options: FrontendOptions, plugins: Plugins) -> Instance {
let meta_client = Arc::new(
MetaClientBuilder::new(0, Role::Frontend)
.enable_procedure()
.build(),
);
FrontendBuilder::new_test(&options, meta_client)
.with_plugin(plugins)
.try_build()
.await
.unwrap()
}
#[derive(Default)]
struct CountingPermissionChecker {
checks: AtomicUsize,
}
impl PermissionChecker for CountingPermissionChecker {
fn check_permission(
&self,
_user_info: UserInfoRef,
_req: PermissionReq,
) -> auth::error::Result<PermissionResp> {
self.checks.fetch_add(1, Ordering::Relaxed);
Ok(PermissionResp::Allow)
}
fn check_permission_with_table_targets(
&self,
_user_info: UserInfoRef,
_req: PermissionReq,
_targets: PermissionTableTargets,
) -> auth::error::Result<PermissionResp> {
self.checks.fetch_add(1, Ordering::Relaxed);
Ok(PermissionResp::Allow)
}
}
fn metric_query_ctx(with_metric_engine: bool) -> QueryContextRef {
let mut ctx = QueryContext::with(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME);
ctx.set_protocol_ctx(ProtocolCtx::OtlpMetric(OtlpMetricCtx {
with_metric_engine,
experimental_enable_exponential_histogram: true,
..Default::default()
}));
Arc::new(ctx)
}
fn metrics_request(metrics: Vec<Metric>) -> ExportMetricsServiceRequest {
ExportMetricsServiceRequest {
resource_metrics: vec![ResourceMetrics {
scope_metrics: vec![ScopeMetrics {
metrics,
..Default::default()
}],
..Default::default()
}],
}
}
fn gauge_metric(metric_name: &str) -> Metric {
Metric {
name: metric_name.to_string(),
data: Some(metric::Data::Gauge(Gauge {
data_points: vec![NumberDataPoint {
time_unix_nano: 1_000_000,
value: Some(Value::AsDouble(1.0)),
..Default::default()
}],
})),
..Default::default()
}
}
fn gauge_request(metric_name: &str) -> ExportMetricsServiceRequest {
metrics_request(vec![gauge_metric(metric_name)])
}
fn native_histogram_metric(metric_name: &str) -> Metric {
Metric {
name: metric_name.to_string(),
data: Some(metric::Data::ExponentialHistogram(ExponentialHistogram {
data_points: vec![ExponentialHistogramDataPoint {
time_unix_nano: 1_000_000,
count: 1,
sum: Some(1.0),
zero_count: 1,
..Default::default()
}],
aggregation_temporality: AggregationTemporality::Cumulative as i32,
})),
..Default::default()
}
}
async fn assert_metrics_falls_back(
instance: &Instance,
request: ExportMetricsServiceRequest,
ctx: QueryContextRef,
) {
let batcher = Arc::new(FakeMetricRowBatcher::default());
let error = instance
.metrics(request, Some(batcher.clone()), ctx)
.await
.unwrap_err();
let servers::error::Error::ExecuteGrpcQuery { source, .. } = error else {
panic!("expected ExecuteGrpcQuery from Inserter, got {error:?}");
};
let Some(crate::error::Error::TableOperation { source, .. }) =
source.as_any().downcast_ref::<crate::error::Error>()
else {
panic!("expected frontend TableOperation from Inserter, got {source:?}");
};
assert!(
matches!(
source,
operator::error::Error::SchemaNotFound { schema_info, .. }
if schema_info == DEFAULT_SCHEMA_NAME
),
"expected test-environment schema lookup failure, got {source:?}"
);
assert!(batcher.requests.lock().unwrap().is_empty());
}
#[tokio::test]
async fn test_metrics_routes_eligible_request_to_batcher() {
let instance = test_instance().await;
let batcher = Arc::new(FakeMetricRowBatcher::successful(7));
let outcome = instance
.metrics(
gauge_request("temperature"),
Some(batcher.clone()),
metric_query_ctx(true),
)
.await
.unwrap();
assert_eq!(1, outcome.accepted_data_points);
assert_eq!(0, outcome.rejected_data_points);
assert_eq!(0, outcome.write_cost);
let requests = batcher.requests.lock().unwrap();
assert_eq!(1, requests.len());
assert_eq!("temperature", requests[0].inserts[0].table_name);
assert_eq!(
vec![MetricRowBatchProtocol::Otlp],
*batcher.protocols.lock().unwrap()
);
}
#[tokio::test]
async fn test_metrics_legacy_mode_falls_back_to_inserter() {
let instance = test_instance().await;
let ctx = metric_query_ctx(true);
instance
.otlp_metrics_table_legacy_cache
.entry(ctx.get_db_string())
.or_default()
.insert("temperature".to_string(), true);
assert_metrics_falls_back(&instance, gauge_request("temperature"), ctx).await;
}
#[tokio::test]
async fn test_metrics_without_metric_engine_falls_back_to_inserter() {
let instance = test_instance().await;
assert_metrics_falls_back(
&instance,
gauge_request("temperature"),
metric_query_ctx(false),
)
.await;
}
#[tokio::test]
async fn test_metrics_native_histogram_falls_back_to_inserter() {
let _guard = FALLBACK_METRIC_ASSERTION_LOCK.lock().await;
let instance = test_instance().await;
let fallback = METRIC_ROW_BATCH_ELIGIBILITY_FALLBACKS
.with_label_values(&[MetricRowBatchProtocol::Otlp.as_str()]);
let initial = fallback.get();
assert_metrics_falls_back(
&instance,
metrics_request(vec![native_histogram_metric("latency")]),
metric_query_ctx(true),
)
.await;
assert_eq!(initial + 1, fallback.get());
}
#[tokio::test]
async fn test_metrics_mixed_request_falls_back_to_inserter() {
let _guard = FALLBACK_METRIC_ASSERTION_LOCK.lock().await;
let instance = test_instance().await;
assert_metrics_falls_back(
&instance,
metrics_request(vec![
gauge_metric("temperature"),
native_histogram_metric("latency"),
]),
metric_query_ctx(true),
)
.await;
}
#[tokio::test]
async fn test_metrics_batcher_error_skips_resource_info_insert() {
let checker = Arc::new(CountingPermissionChecker::default());
let plugins = Plugins::new();
plugins.insert::<PermissionCheckerRef>(checker.clone());
let mut options = FrontendOptions::default();
options.otlp.experimental_enable_resource_info = true;
let instance = test_instance_with(options, plugins).await;
let batcher = Arc::new(FakeMetricRowBatcher::failing("flush failed"));
let mut request = gauge_request("temperature");
request.resource_metrics[0].resource = Some(MetricResource {
attributes: vec![MetricKeyValue {
key: "service.name".to_string(),
value: Some(MetricAnyValue {
value: Some(metric_any_value::Value::StringValue("frontend".to_string())),
}),
}],
..Default::default()
});
let error = instance
.metrics(request, Some(batcher.clone()), metric_query_ctx(true))
.await
.unwrap_err();
assert!(error.to_string().contains("flush failed"), "{error:?}");
assert_eq!(1, batcher.requests.lock().unwrap().len());
assert_eq!(2, checker.checks.load(Ordering::Relaxed));
}
#[test]
fn test_trace_permission_targets() {
src/frontend/src/instance/prom_store.rs
+1 -1
View File
@@ -387,7 +387,7 @@ impl PendingRowsSchemaAlterer for Instance {
Ok(())
}
async fn add_missing_prom_tag_columns_batch(
async fn add_missing_metric_tag_columns_batch(
&self,
catalog: &str,
schema: &str,
src/frontend/src/server.rs
+643 -39
View File
@@ -38,7 +38,10 @@ use servers::interceptor::LogIngestInterceptorRef;
use servers::metrics_handler::MetricsHandler;
use servers::mysql::server::{MysqlServer, MysqlSpawnConfig, MysqlSpawnRef};
use servers::otel_arrow::OtelArrowServiceHandler;
use servers::pending_rows_batcher::{PendingRowsBatcher, pending_rows_batch_sync_enabled};
use servers::pending_rows_batcher::{
MetricRowBatcherRef, PendingRowsBatchMode, PendingRowsBatcher, PendingRowsBatcherOptions,
pending_rows_batch_sync_enabled,
};
use servers::postgres::PostgresServer;
use servers::request_memory_limiter::ServerMemoryLimiter;
use servers::server::{Server, ServerHandlers};
@@ -61,15 +64,148 @@ where
plugins: Plugins,
flight_handler: Option<FlightCraftRef>,
internal_flight_handler: Option<FlightCraftRef>,
metric_batching: ResolvedMetricBatching,
pending_rows_batcher: Option<Arc<PendingRowsBatcher>>,
#[cfg(any(test, feature = "testing"))]
metric_batcher_wiring: MetricBatcherWiring,
pub server_memory_limiter: ServerMemoryLimiter,
}
#[cfg(any(test, feature = "testing"))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum MetricBatchConsumer {
PrometheusHttp,
OtlpHttp,
ExternalOtelArrow,
InternalOtelArrow,
}
#[cfg(any(test, feature = "testing"))]
type MetricBatcherWiring =
Arc<std::sync::Mutex<Vec<(MetricBatchConsumer, Option<MetricRowBatcherRef>)>>>;
struct ResolvedMetricBatching {
options: Option<PendingRowsBatcherOptions>,
batch_mode: PendingRowsBatchMode,
with_metric_engine: bool,
prometheus_enabled: bool,
otlp_enabled: bool,
http_batch_consumer_enabled: bool,
}
/// Metric HTTP consumers registered outside OSS [`Services`] route assembly.
///
/// This input only describes route ownership. Runtime [`FrontendOptions`] remain
/// authoritative for Metric Engine, batching mode, tuning, and protocol behavior.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct ExternalMetricHttpConsumers {
/// An external Prometheus HTTP route consumes the shared metric batcher.
pub prometheus: bool,
/// An external OTLP HTTP route consumes the shared metric batcher.
pub otlp: bool,
}
impl ResolvedMetricBatching {
fn batcher_enabled(&self) -> bool {
self.prometheus_enabled || self.otlp_enabled
}
}
fn resolve_metric_batching(
opts: &FrontendOptions,
external_http_consumers: ExternalMetricHttpConsumers,
batch_mode: PendingRowsBatchMode,
) -> ResolvedMetricBatching {
let prom = &opts.prom_store;
let options = prom.pending_rows_batcher_options();
let valid_options = options.is_some();
let prometheus_enabled = (prom.enable || external_http_consumers.prometheus)
&& prom.with_metric_engine
&& valid_options;
let otlp_enabled = opts.otlp.enable_metrics_batching
&& prom.with_metric_engine
&& batch_mode.is_synchronous()
&& valid_options;
let http_batch_consumer_enabled =
prometheus_enabled || ((opts.otlp.enable || external_http_consumers.otlp) && otlp_enabled);
ResolvedMetricBatching {
options,
batch_mode,
with_metric_engine: prom.with_metric_engine,
prometheus_enabled,
otlp_enabled,
http_batch_consumer_enabled,
}
}
impl<T> Services<T>
where
T: Into<FrontendOptions> + Configurable + Clone,
{
pub fn new(opts: T, instance: Arc<Instance>, plugins: Plugins) -> Self {
Self::new_with_external_metric_http_consumers(
opts,
ExternalMetricHttpConsumers::default(),
instance,
plugins,
)
}
/// Creates services that share metric batching with externally owned HTTP routes.
pub fn new_with_external_metric_http_consumers(
opts: T,
external_http_consumers: ExternalMetricHttpConsumers,
instance: Arc<Instance>,
plugins: Plugins,
) -> Self {
let batch_mode = if pending_rows_batch_sync_enabled() {
PendingRowsBatchMode::Synchronous
} else {
PendingRowsBatchMode::Asynchronous
};
let metric_batching =
resolve_metric_batching(&opts.clone().into(), external_http_consumers, batch_mode);
Self::new_with_metric_batching(opts, instance, plugins, metric_batching)
}
#[cfg(test)]
fn new_with_synchronous_metric_batching(
opts: T,
instance: Arc<Instance>,
plugins: Plugins,
) -> Self {
let metric_batching = resolve_metric_batching(
&opts.clone().into(),
ExternalMetricHttpConsumers::default(),
PendingRowsBatchMode::Synchronous,
);
Self::new_with_metric_batching(opts, instance, plugins, metric_batching)
}
fn new_with_metric_batching(
opts: T,
instance: Arc<Instance>,
plugins: Plugins,
metric_batching: ResolvedMetricBatching,
) -> Self {
let feopts = opts.clone().into();
let pending_rows_batcher = metric_batching
.options
.as_ref()
.filter(|_| metric_batching.batcher_enabled())
.map(|options| {
PendingRowsBatcher::new(
instance.partition_manager().clone(),
instance.node_manager().clone(),
instance.catalog_manager().clone(),
instance.table_flownode_set_cache().clone(),
metric_batching.with_metric_engine,
instance.clone(),
options.clone(),
metric_batching.batch_mode,
)
});
// Create server request memory limiter for all server protocols
let server_memory_limiter = ServerMemoryLimiter::new(
feopts.max_in_flight_write_bytes.as_bytes(),
@@ -84,6 +220,10 @@ where
plugins,
flight_handler: None,
internal_flight_handler: None,
metric_batching,
pending_rows_batcher,
#[cfg(any(test, feature = "testing"))]
metric_batcher_wiring: Default::default(),
server_memory_limiter,
}
}
@@ -105,9 +245,10 @@ where
opts: &FrontendOptions,
request_memory_limiter: ServerMemoryLimiter,
) -> HttpServerBuilder {
let mut builder = HttpServerBuilder::new(effective_http_options(opts))
.with_memory_limiter(request_memory_limiter)
.with_sql_handler(self.instance.clone());
let mut builder =
HttpServerBuilder::new(effective_http_options(&opts.http, &self.metric_batching))
.with_memory_limiter(request_memory_limiter)
.with_sql_handler(self.instance.clone());
let validator = self.plugins.get::<LogValidatorRef>();
let ingest_interceptor = self.plugins.get::<LogIngestInterceptorRef<ServerError>>();
@@ -128,24 +269,10 @@ where
}
if opts.prom_store.enable {
let pending_rows_batcher = if opts.prom_store.with_metric_engine {
PendingRowsBatcher::try_new(
self.instance.partition_manager().clone(),
self.instance.node_manager().clone(),
self.instance.catalog_manager().clone(),
self.instance.table_flownode_set_cache().clone(),
opts.prom_store.with_metric_engine,
self.instance.clone(),
opts.prom_store.pending_rows_flush_interval,
opts.prom_store.max_batch_rows,
opts.prom_store.max_concurrent_flushes,
opts.prom_store.worker_channel_capacity,
opts.prom_store.max_inflight_requests,
opts.prom_store.flow_notification_queue_capacity,
)
} else {
None
};
let metric_row_batcher =
self.metric_row_batcher(self.metric_batching.prometheus_enabled);
#[cfg(any(test, feature = "testing"))]
self.record_metric_batcher(MetricBatchConsumer::PrometheusHttp, &metric_row_batcher);
builder = builder
.with_prom_handler(
self.instance.clone(),
@@ -154,16 +281,20 @@ where
opts.prom_store.prom_validation_mode,
opts.prom_store
.experimental_enable_prometheus_native_histogram,
pending_rows_batcher,
metric_row_batcher,
)
.with_prometheus_handler(self.instance.clone());
}
if opts.otlp.enable {
let metric_row_batcher = self.metric_row_batcher(self.metric_batching.otlp_enabled);
#[cfg(any(test, feature = "testing"))]
self.record_metric_batcher(MetricBatchConsumer::OtlpHttp, &metric_row_batcher);
builder = builder.with_otlp_handler(
self.instance.clone(),
opts.prom_store.with_metric_engine,
opts.otlp.experimental_enable_exponential_histogram,
metric_row_batcher,
);
}
@@ -218,6 +349,50 @@ where
}
}
fn metric_row_batcher(&self, enabled: bool) -> Option<MetricRowBatcherRef> {
if !enabled {
return None;
}
self.pending_rows_batcher
.as_ref()
.map(|batcher| batcher.clone() as MetricRowBatcherRef)
}
/// Returns the shared batcher when Prometheus pending-row batching is enabled.
pub fn prometheus_metric_row_batcher(&self) -> Option<MetricRowBatcherRef> {
self.metric_row_batcher(self.metric_batching.prometheus_enabled)
}
/// Returns the shared batcher when OTLP metric batching is enabled.
pub fn otlp_metric_row_batcher(&self) -> Option<MetricRowBatcherRef> {
self.metric_row_batcher(self.metric_batching.otlp_enabled)
}
#[cfg(any(test, feature = "testing"))]
fn record_metric_batcher(
&self,
consumer: MetricBatchConsumer,
batcher: &Option<MetricRowBatcherRef>,
) {
self.metric_batcher_wiring
.lock()
.unwrap()
.push((consumer, batcher.clone()));
}
/// Reports whether the production HTTP assembly configured an OTLP metric batcher.
#[cfg(feature = "testing")]
pub fn otlp_http_metric_batcher_configured(&self) -> Option<bool> {
self.metric_batcher_wiring
.lock()
.unwrap()
.iter()
.rev()
.find(|(consumer, _)| *consumer == MetricBatchConsumer::OtlpHttp)
.map(|(_, batcher)| batcher.is_some())
}
fn build_grpc_server(
&mut self,
grpc: &GrpcOptions,
@@ -264,6 +439,17 @@ where
.unwrap_or(default_flight_handler)
};
let metric_row_batcher = self.metric_row_batcher(self.metric_batching.otlp_enabled);
#[cfg(any(test, feature = "testing"))]
self.record_metric_batcher(
if external {
MetricBatchConsumer::ExternalOtelArrow
} else {
MetricBatchConsumer::InternalOtelArrow
},
&metric_row_batcher,
);
let grpc_server = builder
.name(name)
.database_handler(greptime_request_handler.clone())
@@ -271,6 +457,8 @@ where
.otel_arrow_handler(OtelArrowServiceHandler::new(
self.instance.clone(),
user_provider.clone(),
self.metric_batching.with_metric_engine,
metric_row_batcher,
))
.flight_handler(flight_handler)
.add_layer(axum::middleware::from_fn_with_state(
@@ -426,19 +614,22 @@ where
}
}
fn effective_http_options(opts: &FrontendOptions) -> HttpOptions {
effective_http_options_with_sync(opts, pending_rows_batch_sync_enabled())
}
fn effective_http_options_with_sync(opts: &FrontendOptions, batch_sync: bool) -> HttpOptions {
let mut http = opts.http.clone();
let flush_interval = opts.prom_store.pending_rows_flush_interval;
fn effective_http_options(
configured_http: &HttpOptions,
metric_batching: &ResolvedMetricBatching,
) -> HttpOptions {
let mut http = configured_http.clone();
let flush_interval = metric_batching
.options
.as_ref()
.map(PendingRowsBatcherOptions::flush_interval)
.unwrap_or_default();
let fallback_timeout = flush_interval.saturating_add(Duration::from_secs(1));
// In asynchronous batch mode submissions return right after enqueue and
// no request waits for a pending-row flush, so the timeout must not be
// raised either.
if !opts.prom_store.pending_rows_batching_enabled()
|| !batch_sync
if !metric_batching.batch_mode.is_synchronous()
|| !metric_batching.http_batch_consumer_enabled
|| http.timeout.is_zero()
|| http.timeout > fallback_timeout
{
@@ -506,6 +697,15 @@ mod tests {
}
}
fn resolved_http_options(
opts: &FrontendOptions,
batch_mode: PendingRowsBatchMode,
) -> HttpOptions {
let metric_batching =
resolve_metric_batching(opts, ExternalMetricHttpConsumers::default(), batch_mode);
effective_http_options(&opts.http, &metric_batching)
}
#[test]
fn test_effective_http_timeout_for_pending_rows() {
let cases = [
@@ -536,7 +736,7 @@ mod tests {
assert_eq!(
Duration::from_millis(expected),
effective_http_options_with_sync(&opts, true).timeout,
resolved_http_options(&opts, PendingRowsBatchMode::Synchronous).timeout,
"{name}"
);
}
@@ -553,19 +753,18 @@ mod tests {
assert_eq!(
Duration::from_millis(1000),
effective_http_options_with_sync(&opts, false).timeout,
resolved_http_options(&opts, PendingRowsBatchMode::Asynchronous).timeout,
);
assert_eq!(
Duration::from_millis(6000),
effective_http_options_with_sync(&opts, true).timeout,
resolved_http_options(&opts, PendingRowsBatchMode::Synchronous).timeout,
);
}
#[test]
fn test_effective_http_timeout_skips_fallback_when_batcher_disabled() {
// Mirrors the conditions under which `PendingRowsBatcher::try_new`
// returns `None`; in these cases no request can wait for a pending-row
// flush, so the timeout must not be raised.
// Invalid tuning cannot produce a resolved batcher, so no request can
// wait for a pending-row flush and the timeout must not be raised.
type KnobMutator = fn(&mut FrontendOptions);
let cases: [(&str, KnobMutator); 4] = [
("zero max_batch_rows", |opts| {
@@ -590,12 +789,362 @@ mod tests {
assert_eq!(
Duration::from_millis(1000),
effective_http_options_with_sync(&opts, true).timeout,
resolved_http_options(&opts, PendingRowsBatchMode::Synchronous).timeout,
"{name}"
);
}
}
#[test]
fn test_metric_batching_enablement_matrix() {
type OptionsMutator = fn(&mut FrontendOptions);
let cases: [(&str, OptionsMutator, bool, bool, bool, PendingRowsBatchMode); 8] = [
(
"prometheus only",
|_| {},
true,
false,
true,
PendingRowsBatchMode::Synchronous,
),
(
"otlp only with HTTP route disabled",
|opts| {
opts.prom_store.enable = false;
opts.otlp.enable = false;
opts.otlp.enable_metrics_batching = true;
},
false,
true,
true,
PendingRowsBatchMode::Synchronous,
),
(
"both consumers",
|opts| opts.otlp.enable_metrics_batching = true,
true,
true,
true,
PendingRowsBatchMode::Synchronous,
),
(
"both routes disabled",
|opts| {
opts.prom_store.enable = false;
opts.otlp.enable = false;
},
false,
false,
false,
PendingRowsBatchMode::Synchronous,
),
(
"metric engine disabled",
|opts| {
opts.prom_store.with_metric_engine = false;
opts.otlp.enable_metrics_batching = true;
},
false,
false,
false,
PendingRowsBatchMode::Synchronous,
),
(
"asynchronous mode preserves prometheus only",
|opts| opts.otlp.enable_metrics_batching = true,
true,
false,
true,
PendingRowsBatchMode::Asynchronous,
),
(
"prometheus disabled does not disable otlp",
|opts| {
opts.prom_store.enable = false;
opts.otlp.enable_metrics_batching = true;
},
false,
true,
true,
PendingRowsBatchMode::Synchronous,
),
(
"otlp batching disabled does not disable prometheus",
|opts| opts.otlp.enable_metrics_batching = false,
true,
false,
true,
PendingRowsBatchMode::Synchronous,
),
];
for (name, mutate, expected_prom, expected_otlp, expected_batcher, batch_mode) in cases {
let mut opts = FrontendOptions::default();
opts.prom_store.pending_rows_flush_interval = Duration::from_secs(5);
mutate(&mut opts);
let resolved =
resolve_metric_batching(&opts, ExternalMetricHttpConsumers::default(), batch_mode);
assert_eq!(expected_prom, resolved.prometheus_enabled, "{name}");
assert_eq!(expected_otlp, resolved.otlp_enabled, "{name}");
assert_eq!(batch_mode, resolved.batch_mode, "{name}");
assert_eq!(expected_batcher, resolved.batcher_enabled(), "{name}");
assert_eq!(
Duration::from_secs(5),
resolved.options.as_ref().unwrap().flush_interval(),
"{name}"
);
}
}
#[test]
fn test_metric_batching_snapshot_remains_authoritative_after_source_options_change() {
let mut opts = FrontendOptions::default();
opts.http.timeout = Duration::from_secs(1);
opts.prom_store.pending_rows_flush_interval = Duration::from_secs(5);
let resolved = resolve_metric_batching(
&opts,
ExternalMetricHttpConsumers::default(),
PendingRowsBatchMode::Synchronous,
);
opts.prom_store.pending_rows_flush_interval = Duration::from_secs(30);
opts.prom_store.max_batch_rows = 0;
assert_eq!(
Duration::from_secs(6),
effective_http_options(&opts.http, &resolved).timeout
);
assert!(resolved.options.is_some());
assert!(resolved.batcher_enabled());
}
#[test]
fn test_external_http_consumers_cannot_override_runtime_batching_configuration() {
type OptionsMutator = fn(&mut FrontendOptions);
struct TestCase {
name: &'static str,
external_http_consumers: ExternalMetricHttpConsumers,
batch_mode: PendingRowsBatchMode,
mutate: OptionsMutator,
expected_prometheus: bool,
expected_otlp: bool,
expected_flush_interval: Option<Duration>,
expected_timeout: Duration,
}
let cases = [
TestCase {
name: "external Prometheus uses runtime tuning",
external_http_consumers: ExternalMetricHttpConsumers {
prometheus: true,
otlp: false,
},
batch_mode: PendingRowsBatchMode::Synchronous,
mutate: |_| {},
expected_prometheus: true,
expected_otlp: false,
expected_flush_interval: Some(Duration::from_secs(5)),
expected_timeout: Duration::from_secs(6),
},
TestCase {
name: "external OTLP uses runtime tuning and batching flag",
external_http_consumers: ExternalMetricHttpConsumers {
prometheus: false,
otlp: true,
},
batch_mode: PendingRowsBatchMode::Synchronous,
mutate: |opts| opts.otlp.enable_metrics_batching = true,
expected_prometheus: false,
expected_otlp: true,
expected_flush_interval: Some(Duration::from_secs(5)),
expected_timeout: Duration::from_secs(6),
},
TestCase {
name: "external OTLP cannot enable runtime-disabled batching",
external_http_consumers: ExternalMetricHttpConsumers {
prometheus: false,
otlp: true,
},
batch_mode: PendingRowsBatchMode::Synchronous,
mutate: |_| {},
expected_prometheus: false,
expected_otlp: false,
expected_flush_interval: Some(Duration::from_secs(5)),
expected_timeout: Duration::from_secs(1),
},
TestCase {
name: "external OTLP cannot replace invalid runtime tuning",
external_http_consumers: ExternalMetricHttpConsumers {
prometheus: false,
otlp: true,
},
batch_mode: PendingRowsBatchMode::Synchronous,
mutate: |opts| {
opts.otlp.enable_metrics_batching = true;
opts.prom_store.max_batch_rows = 0;
},
expected_prometheus: false,
expected_otlp: false,
expected_flush_interval: None,
expected_timeout: Duration::from_secs(1),
},
TestCase {
name: "external OTLP cannot override asynchronous runtime mode",
external_http_consumers: ExternalMetricHttpConsumers {
prometheus: false,
otlp: true,
},
batch_mode: PendingRowsBatchMode::Asynchronous,
mutate: |opts| opts.otlp.enable_metrics_batching = true,
expected_prometheus: false,
expected_otlp: false,
expected_flush_interval: Some(Duration::from_secs(5)),
expected_timeout: Duration::from_secs(1),
},
];
for case in cases {
let mut opts = FrontendOptions::default();
opts.http.timeout = Duration::from_secs(1);
opts.prom_store.enable = false;
opts.prom_store.pending_rows_flush_interval = Duration::from_secs(5);
opts.otlp.enable = false;
(case.mutate)(&mut opts);
let resolved =
resolve_metric_batching(&opts, case.external_http_consumers, case.batch_mode);
assert_eq!(
case.expected_prometheus, resolved.prometheus_enabled,
"{}",
case.name
);
assert_eq!(case.expected_otlp, resolved.otlp_enabled, "{}", case.name);
assert_eq!(case.batch_mode, resolved.batch_mode, "{}", case.name);
assert_eq!(
case.expected_flush_interval,
resolved
.options
.as_ref()
.map(|options| options.flush_interval()),
"{}",
case.name
);
assert_eq!(
case.expected_timeout,
effective_http_options(&opts.http, &resolved).timeout,
"{}",
case.name
);
}
}
#[test]
fn test_metric_batching_invalid_options_disable_all_consumers() {
type KnobMutator = fn(&mut FrontendOptions);
let cases: [(&str, KnobMutator); 5] = [
("zero flush interval", |opts| {
opts.prom_store.pending_rows_flush_interval = Duration::ZERO
}),
("zero max batch rows", |opts| {
opts.prom_store.max_batch_rows = 0
}),
("zero max concurrent flushes", |opts| {
opts.prom_store.max_concurrent_flushes = 0
}),
("zero worker channel capacity", |opts| {
opts.prom_store.worker_channel_capacity = 0
}),
("zero max inflight requests", |opts| {
opts.prom_store.max_inflight_requests = 0
}),
];
for (name, disable_batcher) in cases {
let mut opts = FrontendOptions::default();
opts.prom_store.pending_rows_flush_interval = Duration::from_secs(5);
opts.otlp.enable_metrics_batching = true;
disable_batcher(&mut opts);
let resolved = resolve_metric_batching(
&opts,
ExternalMetricHttpConsumers::default(),
PendingRowsBatchMode::Synchronous,
);
assert!(!resolved.prometheus_enabled, "{name}");
assert!(!resolved.otlp_enabled, "{name}");
assert!(resolved.options.is_none(), "{name}");
}
}
#[test]
fn test_effective_http_timeout_for_enabled_sync_consumers() {
let cases = [
("prometheus", true, false, true, true, 6000),
("otlp HTTP", false, true, true, true, 6000),
("otlp Arrow only", false, true, false, true, 1000),
("asynchronous prometheus", true, false, true, false, 1000),
(
"otlp fallback in asynchronous mode",
false,
true,
true,
false,
1000,
),
];
for (name, prom_enable, otlp_batching, otlp_http, batch_sync, expected) in cases {
let mut opts = FrontendOptions::default();
opts.http.timeout = Duration::from_millis(1000);
opts.prom_store.enable = prom_enable;
opts.prom_store.pending_rows_flush_interval = Duration::from_millis(5000);
opts.otlp.enable = otlp_http;
opts.otlp.enable_metrics_batching = otlp_batching;
assert_eq!(
Duration::from_millis(expected),
resolved_http_options(
&opts,
if batch_sync {
PendingRowsBatchMode::Synchronous
} else {
PendingRowsBatchMode::Asynchronous
},
)
.timeout,
"{name}"
);
}
}
#[test]
fn test_enterprise_otlp_consumer_extends_timeout_with_oss_route_disabled() {
let mut opts = FrontendOptions::default();
opts.http.timeout = Duration::from_secs(1);
opts.prom_store.enable = false;
opts.prom_store.pending_rows_flush_interval = Duration::from_secs(5);
opts.otlp.enable = false;
opts.otlp.enable_metrics_batching = true;
let resolved = resolve_metric_batching(
&opts,
ExternalMetricHttpConsumers {
prometheus: false,
otlp: true,
},
PendingRowsBatchMode::Synchronous,
);
assert_eq!(
Duration::from_secs(6),
effective_http_options(&opts.http, &resolved).timeout
);
assert!(!opts.otlp.enable);
assert!(resolved.otlp_enabled);
assert!(resolved.http_batch_consumer_enabled);
}
#[tokio::test]
async fn test_database_sql_authentication_differs_between_public_and_internal_grpc() {
let options = FrontendOptions {
@@ -741,4 +1290,59 @@ mod tests {
// Assert
assert!(health_check.is_ok());
}
#[tokio::test]
async fn test_services_wires_same_metric_batcher_to_all_enabled_consumers() {
let options = FrontendOptions {
grpc: GrpcOptions::default().with_bind_addr("127.0.0.1:0"),
internal_grpc: Some(GrpcOptions::default().with_bind_addr("127.0.0.1:0")),
prom_store: crate::service_config::PromStoreOptions {
pending_rows_flush_interval: Duration::from_secs(1),
..Default::default()
},
otlp: crate::service_config::OtlpOptions {
enable_metrics_batching: true,
..Default::default()
},
..Default::default()
};
let meta_client = Arc::new(
MetaClientBuilder::new(0, Role::Frontend)
.enable_procedure()
.build(),
);
let instance = Arc::new(
FrontendBuilder::new_test(&options, meta_client)
.try_build()
.await
.unwrap(),
);
let services =
Services::new_with_synchronous_metric_batching(options, instance, Default::default());
let wiring = services.metric_batcher_wiring.clone();
let _handlers = services.build().unwrap();
let wiring = wiring.lock().unwrap();
assert_eq!(4, wiring.len());
let consumers = [
MetricBatchConsumer::PrometheusHttp,
MetricBatchConsumer::OtlpHttp,
MetricBatchConsumer::ExternalOtelArrow,
MetricBatchConsumer::InternalOtelArrow,
];
let first = wiring
.iter()
.find(|(consumer, _)| *consumer == consumers[0])
.and_then(|(_, batcher)| batcher.as_ref())
.unwrap();
for consumer in consumers {
let batcher = wiring
.iter()
.find(|(wired_consumer, _)| *wired_consumer == consumer)
.and_then(|(_, batcher)| batcher.as_ref())
.unwrap();
assert!(Arc::ptr_eq(first, batcher), "{consumer:?}");
}
}
}
src/frontend/src/service_config/otlp.rs
+15
View File
@@ -20,6 +20,13 @@ const DEFAULT_TRACE_INGEST_CHUNK_SIZE: usize = 512;
#[serde(default)]
pub struct OtlpOptions {
pub enable: bool,
/// Enables pending-row batching for eligible OTLP metrics over HTTP and OTel Arrow.
///
/// Disabled by default. Batching requires Metric Engine, valid nonzero
/// `prom_store.pending_rows_*` tuning, and synchronous pending-row mode.
/// Requests fall back to the Inserter when these conditions are not met or
/// when the converted rows are not scalar metrics supported by the batcher.
pub enable_metrics_batching: bool,
pub experimental_enable_exponential_histogram: bool,
/// Maximum spans per trace ingest chunk. Set to 0 to disable splitting.
pub trace_ingest_chunk_size: usize,
@@ -34,6 +41,7 @@ impl Default for OtlpOptions {
fn default() -> Self {
Self {
enable: true,
enable_metrics_batching: false,
experimental_enable_exponential_histogram: false,
trace_ingest_chunk_size: DEFAULT_TRACE_INGEST_CHUNK_SIZE,
experimental_enable_resource_info: false,
@@ -49,12 +57,14 @@ mod tests {
fn test_otlp_options() {
let default = OtlpOptions::default();
assert!(default.enable);
assert!(!default.enable_metrics_batching);
assert!(!default.experimental_enable_exponential_histogram);
assert_eq!(default.trace_ingest_chunk_size, 512);
assert!(!default.experimental_enable_resource_info);
let options: OtlpOptions = toml::from_str("enable = false").unwrap();
assert!(!options.enable);
assert!(!options.enable_metrics_batching);
assert!(!options.experimental_enable_exponential_histogram);
assert_eq!(
options.trace_ingest_chunk_size,
@@ -71,7 +81,12 @@ mod tests {
assert!(options.experimental_enable_exponential_histogram);
let serialized = toml::to_string(&options).unwrap();
assert!(serialized.contains("enable_metrics_batching = false"));
assert!(serialized.contains("experimental_enable_exponential_histogram = true"));
assert_eq!(toml::from_str::<OtlpOptions>(&serialized).unwrap(), options);
let options: OtlpOptions = toml::from_str("enable_metrics_batching = true").unwrap();
assert!(options.enable_metrics_batching);
assert!(options.enable);
}
}
src/frontend/src/service_config/prom_store.rs
+26 -10
View File
@@ -16,6 +16,7 @@ use std::num::NonZeroUsize;
use std::time::Duration;
use serde::{Deserialize, Serialize};
use servers::pending_rows_batcher::{PendingRowsBatcherOptions, PendingRowsBatcherTuning};
use servers::prom_remote_write::validation::PromValidationMode;
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
@@ -64,18 +65,21 @@ fn default_flow_notification_queue_capacity() -> NonZeroUsize {
}
impl PromStoreOptions {
pub(crate) fn pending_rows_batcher_options(&self) -> Option<PendingRowsBatcherOptions> {
PendingRowsBatcherOptions::try_new(PendingRowsBatcherTuning {
flush_interval: self.pending_rows_flush_interval,
max_batch_rows: self.max_batch_rows,
max_concurrent_flushes: self.max_concurrent_flushes,
worker_channel_capacity: self.worker_channel_capacity,
max_inflight_requests: self.max_inflight_requests,
flow_notification_queue_capacity: self.flow_notification_queue_capacity,
})
}
/// Returns whether the pending rows batcher can be enabled with these
/// options. Mirrors the enablement conditions of
/// `PendingRowsBatcher::try_new` in the servers crate, which returns
/// `None` when any of these knobs is zero.
/// options.
pub fn pending_rows_batching_enabled(&self) -> bool {
self.enable
&& self.with_metric_engine
&& !self.pending_rows_flush_interval.is_zero()
&& self.max_batch_rows > 0
&& self.max_concurrent_flushes > 0
&& self.worker_channel_capacity > 0
&& self.max_inflight_requests > 0
self.enable && self.with_metric_engine && self.pending_rows_batcher_options().is_some()
}
}
@@ -133,4 +137,16 @@ mod tests {
default_flow_notification_queue_capacity()
);
}
#[test]
fn test_pending_rows_options_validity_is_independent_of_prometheus_route() {
let options = PromStoreOptions {
enable: false,
pending_rows_flush_interval: Duration::from_secs(1),
..Default::default()
};
assert!(options.pending_rows_batcher_options().is_some());
assert!(!options.pending_rows_batching_enabled());
}
}
src/servers/src/http.rs
+6 -2
View File
@@ -73,7 +73,7 @@ use crate::http::result::null_result::NullResponse;
use crate::interceptor::LogIngestInterceptorRef;
use crate::metrics::http_metrics_layer;
use crate::metrics_handler::MetricsHandler;
use crate::pending_rows_batcher::PendingRowsBatcher;
use crate::pending_rows_batcher::MetricRowBatcherRef;
use crate::prometheus_handler::PrometheusHandlerRef;
use crate::query_handler::sql::ServerSqlQueryHandlerRef;
use crate::query_handler::{
@@ -682,7 +682,7 @@ impl HttpServerBuilder {
prom_store_with_metric_engine: bool,
prom_validation_mode: PromValidationMode,
experimental_enable_prometheus_native_histogram: bool,
pending_rows_batcher: Option<Arc<PendingRowsBatcher>>,
pending_rows_batcher: Option<MetricRowBatcherRef>,
) -> Self {
let state = PromStoreState {
prom_store_handler: handler,
@@ -717,6 +717,7 @@ impl HttpServerBuilder {
handler: OpenTelemetryProtocolHandlerRef,
with_metric_engine: bool,
experimental_enable_exponential_histogram: bool,
metric_row_batcher: Option<MetricRowBatcherRef>,
) -> Self {
Self {
router: self.router.nest(
@@ -725,6 +726,7 @@ impl HttpServerBuilder {
handler,
with_metric_engine,
experimental_enable_exponential_histogram,
metric_row_batcher,
),
),
..self
@@ -1392,6 +1394,7 @@ impl HttpServer {
otlp_handler: OpenTelemetryProtocolHandlerRef,
with_metric_engine: bool,
experimental_enable_exponential_histogram: bool,
metric_row_batcher: Option<MetricRowBatcherRef>,
) -> Router<S> {
Router::new()
.route("/v1/metrics", routing::post(otlp::metrics))
@@ -1405,6 +1408,7 @@ impl HttpServer {
with_metric_engine,
experimental_enable_exponential_histogram,
handler: otlp_handler,
metric_row_batcher,
})
}
src/servers/src/http/otlp.rs
+7 -1
View File
@@ -46,6 +46,7 @@ use crate::http::extractor::{
};
use crate::http::header::{CONTENT_TYPE_PROTOBUF, write_cost_header_map};
use crate::metrics::METRIC_HTTP_OPENTELEMETRY_LOGS_ELAPSED;
use crate::pending_rows_batcher::MetricRowBatcherRef;
use crate::query_handler::{
MetricsIngestOutcome, OpenTelemetryProtocolHandlerRef, PipelineHandler, TraceIngestOutcome,
};
@@ -79,6 +80,7 @@ pub struct OtlpState {
pub with_metric_engine: bool,
pub experimental_enable_exponential_histogram: bool,
pub handler: OpenTelemetryProtocolHandlerRef,
pub metric_row_batcher: Option<MetricRowBatcherRef>,
}
#[axum_macros::debug_handler]
@@ -110,6 +112,7 @@ pub async fn metrics(
with_metric_engine,
experimental_enable_exponential_histogram,
handler,
metric_row_batcher,
} = state;
query_ctx.set_protocol_ctx(ProtocolCtx::OtlpMetric(OtlpMetricCtx {
@@ -126,7 +129,10 @@ pub async fn metrics(
}));
let query_ctx = Arc::new(query_ctx);
match handler.metrics(request, query_ctx).await {
match handler
.metrics(request, metric_row_batcher, query_ctx)
.await
{
Ok(outcome) => {
if outcome.accepted_data_points == 0 && outcome.rejected_data_points > 0 {
Ok(OtlpMetricsResponse::Failure(outcome))
src/servers/src/http/prom_store.rs
+115 -28
View File
@@ -16,7 +16,6 @@ use std::sync::Arc;
use api::prom_store::remote::ReadRequest;
use api::v1::RowInsertRequests;
use async_trait::async_trait;
use axum::Extension;
use axum::body::Bytes;
use axum::extract::{Query, State};
@@ -45,7 +44,8 @@ use crate::http::extractor::PipelineInfo;
use crate::http::header::{
CONTENT_TYPE_PROTOBUF_STR, GREPTIME_DB_HEADER_METRICS, write_cost_header_map,
};
use crate::pending_rows_batcher::PendingRowsBatcher;
use crate::metrics::METRIC_ROW_BATCH_ELIGIBILITY_FALLBACKS;
use crate::pending_rows_batcher::{MetricRowBatchProtocol, MetricRowBatcher, MetricRowBatcherRef};
use crate::prom_remote_write::decode::PromSeriesProcessor;
use crate::prom_remote_write::v2::decode_remote_write_v2;
use crate::prom_remote_write::validation::PromValidationMode;
@@ -75,7 +75,7 @@ pub struct PromStoreState {
pub prom_store_with_metric_engine: bool,
pub prom_validation_mode: PromValidationMode,
pub experimental_enable_prometheus_native_histogram: bool,
pub pending_rows_batcher: Option<Arc<PendingRowsBatcher>>,
pub pending_rows_batcher: Option<MetricRowBatcherRef>,
}
#[derive(Debug, Serialize, Deserialize)]
@@ -352,18 +352,6 @@ struct PromWriteV2Error {
type PromWriteBatch = (QueryContextRef, RowInsertRequests);
#[async_trait]
trait PromWriteBatcher: Send + Sync {
async fn submit(&self, requests: RowInsertRequests, ctx: QueryContextRef) -> Result<u64>;
}
#[async_trait]
impl PromWriteBatcher for PendingRowsBatcher {
async fn submit(&self, requests: RowInsertRequests, ctx: QueryContextRef) -> Result<u64> {
PendingRowsBatcher::submit(self, requests, ctx).await
}
}
fn into_prom_write_batches(req: ContextReq, query_ctx: QueryContextRef) -> Vec<PromWriteBatch> {
req.as_req_iter(query_ctx).collect()
}
@@ -387,7 +375,7 @@ async fn preflight_prometheus_rows(
/// sample/histogram headers even when a later table write fails.
async fn write_prometheus_rows_with_progress(
prom_store_handler: PromStoreProtocolHandlerRef,
pending_rows_batcher: Option<Arc<PendingRowsBatcher>>,
pending_rows_batcher: Option<MetricRowBatcherRef>,
prom_store_with_metric_engine: bool,
mut batches: Vec<PromWriteBatch>,
) -> std::result::Result<PromWriteOutcome, PromWriteError> {
@@ -401,7 +389,7 @@ async fn write_prometheus_rows_with_progress(
let mut rows_written = 0;
for (temp_ctx, reqs) in batches {
let rows = batcher
.submit(reqs, temp_ctx)
.submit(reqs, temp_ctx, MetricRowBatchProtocol::Prometheus)
.await
.map_err(|error| PromWriteError {
error,
@@ -453,7 +441,7 @@ async fn write_prometheus_rows_with_progress(
async fn write_prometheus_v2_rows_with_progress(
prom_store_handler: PromStoreProtocolHandlerRef,
pending_rows_batcher: Option<Arc<PendingRowsBatcher>>,
pending_rows_batcher: Option<MetricRowBatcherRef>,
prom_store_with_metric_engine: bool,
sample_batches: Vec<PromWriteBatch>,
histogram_batches: Vec<PromWriteBatch>,
@@ -539,13 +527,18 @@ async fn write_prometheus_v2_rows_with_progress(
})
}
async fn write_batched_prometheus_v2_rows_with_progress<B: PromWriteBatcher + ?Sized>(
async fn write_batched_prometheus_v2_rows_with_progress(
prom_store_handler: PromStoreProtocolHandlerRef,
batcher: &B,
batcher: &dyn MetricRowBatcher,
prom_store_with_metric_engine: bool,
sample_batches: Vec<PromWriteBatch>,
histogram_batches: Vec<PromWriteBatch>,
) -> std::result::Result<PromWriteV2Outcome, PromWriteV2Error> {
if !histogram_batches.is_empty() {
METRIC_ROW_BATCH_ELIGIBILITY_FALLBACKS
.with_label_values(&[MetricRowBatchProtocol::Prometheus.as_str()])
.inc();
}
let sample_batch_count = sample_batches.len();
let mut batches = sample_batches;
batches.extend(histogram_batches);
@@ -563,7 +556,7 @@ async fn write_batched_prometheus_v2_rows_with_progress<B: PromWriteBatcher + ?S
let mut batches = batches.into_iter();
for (ctx, requests) in batches.by_ref().take(sample_batch_count) {
let rows = batcher
.submit(requests, ctx)
.submit(requests, ctx, MetricRowBatchProtocol::Prometheus)
.await
.map_err(|error| PromWriteV2Error {
error,
@@ -769,6 +762,9 @@ mod tests {
use session::context::{QueryContext, QueryContextRef};
use super::*;
use crate::pending_rows_batcher::{
MetricRowBatchProtocol, MetricRowBatcher, MetricRowBatcherRef,
};
use crate::prom_remote_write::validation::PromValidationMode;
use crate::query_handler::PromStoreProtocolHandler;
@@ -850,16 +846,21 @@ mod tests {
#[tokio::test]
async fn test_mixed_v2_preflights_all_then_batches_only_samples() {
let events = Arc::new(Mutex::new(Vec::new()));
let fallback = METRIC_ROW_BATCH_ELIGIBILITY_FALLBACKS
.with_label_values(&[MetricRowBatchProtocol::Prometheus.as_str()]);
let initial_fallbacks = fallback.get();
let handler: PromStoreProtocolHandlerRef = Arc::new(RecordingPromStoreHandler {
events: events.clone(),
});
let batcher = RecordingPromWriteBatcher {
let batcher: MetricRowBatcherRef = Arc::new(RecordingMetricRowBatcher {
events: events.clone(),
};
protocols: Arc::new(Mutex::new(Vec::new())),
rows_written: 1,
});
let Ok(outcome) = write_batched_prometheus_v2_rows_with_progress(
handler,
&batcher,
batcher.as_ref(),
true,
vec![test_prom_write_batch("sample")],
vec![test_prom_write_batch("histogram")],
@@ -880,6 +881,67 @@ mod tests {
],
*events.lock().unwrap()
);
assert_eq!(initial_fallbacks + 1, fallback.get());
}
#[tokio::test]
async fn test_prometheus_write_dispatches_through_metric_row_batcher_ref() {
let events = Arc::new(Mutex::new(Vec::new()));
let handler: PromStoreProtocolHandlerRef = Arc::new(RecordingPromStoreHandler {
events: events.clone(),
});
let protocols = Arc::new(Mutex::new(Vec::new()));
let batcher: MetricRowBatcherRef = Arc::new(RecordingMetricRowBatcher {
events: events.clone(),
protocols: protocols.clone(),
rows_written: 7,
});
let Ok(outcome) = write_prometheus_rows_with_progress(
handler,
Some(batcher),
true,
vec![test_prom_write_batch("sample")],
)
.await
else {
panic!("prometheus write through metric row batcher should succeed")
};
assert_eq!(7, outcome.rows_written);
assert_eq!(
vec!["pre:sample".to_string(), "batch:sample".to_string()],
*events.lock().unwrap()
);
assert_eq!(
vec![MetricRowBatchProtocol::Prometheus],
*protocols.lock().unwrap()
);
}
#[tokio::test]
async fn test_prometheus_write_propagates_metric_row_batcher_ref_error() {
let handler: PromStoreProtocolHandlerRef = Arc::new(RecordingPromStoreHandler {
events: Arc::new(Mutex::new(Vec::new())),
});
let batcher: MetricRowBatcherRef = Arc::new(FailingMetricRowBatcher);
let Err(write_error) = write_prometheus_rows_with_progress(
handler,
Some(batcher),
true,
vec![test_prom_write_batch("sample")],
)
.await
else {
panic!("metric row batcher error should propagate")
};
assert_eq!(0, write_error.rows_written);
assert!(matches!(
write_error.error,
error::Error::Internal { ref err_msg } if err_msg == "metric row batcher failed"
));
}
fn test_prom_write_batch(table_name: &str) -> PromWriteBatch {
@@ -904,15 +966,40 @@ mod tests {
));
}
struct RecordingPromWriteBatcher {
struct RecordingMetricRowBatcher {
events: Arc<Mutex<Vec<String>>>,
protocols: Arc<Mutex<Vec<MetricRowBatchProtocol>>>,
rows_written: u64,
}
#[async_trait]
impl PromWriteBatcher for RecordingPromWriteBatcher {
async fn submit(&self, requests: RowInsertRequests, _ctx: QueryContextRef) -> Result<u64> {
impl MetricRowBatcher for RecordingMetricRowBatcher {
async fn submit(
&self,
requests: RowInsertRequests,
_ctx: QueryContextRef,
protocol: MetricRowBatchProtocol,
) -> Result<u64> {
record_write_event(&self.events, "batch", &requests);
Ok(prom_write_row_count(&requests))
self.protocols.lock().unwrap().push(protocol);
Ok(self.rows_written)
}
}
struct FailingMetricRowBatcher;
#[async_trait]
impl MetricRowBatcher for FailingMetricRowBatcher {
async fn submit(
&self,
_requests: RowInsertRequests,
_ctx: QueryContextRef,
_protocol: MetricRowBatchProtocol,
) -> Result<u64> {
Err(error::InternalSnafu {
err_msg: "metric row batcher failed".to_string(),
}
.build())
}
}
src/servers/src/metrics.rs
+22
View File
@@ -48,6 +48,7 @@ pub(crate) const METRIC_VERSION_LABEL: &str = "version";
pub(crate) const METRIC_SUCCESS_VALUE: &str = "success";
pub(crate) const METRIC_FAILURE_VALUE: &str = "failure";
pub(crate) const METRIC_CANCELLED_VALUE: &str = "cancelled";
lazy_static! {
@@ -192,6 +193,27 @@ lazy_static! {
vec![0.0005, 0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0, 10.0, 60.0]
)
.unwrap();
pub static ref METRIC_ROW_BATCH_SUBMISSIONS: IntCounterVec = register_int_counter_vec!(
"greptime_servers_metric_row_batch_submissions_total",
"Total metric row batch submissions",
&[METRIC_PROTOCOL_LABEL]
)
.unwrap();
pub static ref METRIC_ROW_BATCH_ELIGIBILITY_FALLBACKS: IntCounterVec =
register_int_counter_vec!(
"greptime_servers_metric_row_batch_eligibility_fallbacks_total",
"Total metric requests that fell back because their rows were not batchable",
&[METRIC_PROTOCOL_LABEL]
)
.unwrap();
pub static ref METRIC_ROW_BATCH_WAIT_FLUSH_RESULT_ELAPSED: HistogramVec =
register_histogram_vec!(
"greptime_servers_metric_row_batch_wait_flush_result_elapsed",
"Elapsed time waiting for metric row batch flush results in seconds by result",
&[METRIC_PROTOCOL_LABEL, METRIC_RESULT_LABEL],
vec![0.0005, 0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0, 10.0, 60.0]
)
.unwrap();
/// Http prometheus read duration per database.
pub static ref METRIC_HTTP_PROM_STORE_READ_ELAPSED: HistogramVec = register_histogram_vec!(
"greptime_servers_http_prometheus_read_elapsed",
src/servers/src/otel_arrow.rs
+32 -3
View File
@@ -32,6 +32,7 @@ use tonic::{Request, Response, Status, Streaming};
use crate::error;
use crate::grpc::context_auth;
use crate::pending_rows_batcher::MetricRowBatcherRef;
use crate::query_handler::{MetricsIngestOutcome, OpenTelemetryProtocolHandlerRef};
const EXPONENTIAL_HISTOGRAM_UNSUPPORTED: &str = "OTel Arrow exponential histograms are unsupported because the Arrow wire format omits zero_threshold";
@@ -39,17 +40,33 @@ const EXPONENTIAL_HISTOGRAM_UNSUPPORTED: &str = "OTel Arrow exponential histogra
pub struct OtelArrowServiceHandler<T> {
handler: T,
user_provider: Option<UserProviderRef>,
with_metric_engine: bool,
metric_row_batcher: Option<MetricRowBatcherRef>,
}
impl<T> OtelArrowServiceHandler<T> {
pub fn new(handler: T, user_provider: Option<UserProviderRef>) -> Self {
pub fn new(
handler: T,
user_provider: Option<UserProviderRef>,
with_metric_engine: bool,
metric_row_batcher: Option<MetricRowBatcherRef>,
) -> Self {
Self {
handler,
user_provider,
with_metric_engine,
metric_row_batcher,
}
}
}
fn otel_arrow_metric_context(with_metric_engine: bool) -> OtlpMetricCtx {
OtlpMetricCtx {
with_metric_engine,
..Default::default()
}
}
fn batch_status(
batch_id: i64,
outcome: MetricsIngestOutcome,
@@ -90,11 +107,14 @@ impl ArrowMetricsService for OtelArrowServiceHandler<OpenTelemetryProtocolHandle
context_auth::check_auth(self.user_provider.clone(), &headers, query_ctx.clone()).await?;
let query_ctx = {
let mut ctx = query_ctx.fork();
ctx.set_protocol_ctx(ProtocolCtx::OtlpMetric(OtlpMetricCtx::default()));
ctx.set_protocol_ctx(ProtocolCtx::OtlpMetric(otel_arrow_metric_context(
self.with_metric_engine,
)));
Arc::new(ctx)
};
let handler = self.handler.clone();
let metric_row_batcher = self.metric_row_batcher.clone();
// handles incoming requests
common_runtime::spawn_global(async move {
@@ -144,7 +164,10 @@ impl ArrowMetricsService for OtelArrowServiceHandler<OpenTelemetryProtocolHandle
if !histogram.data_points.is_empty()
)
});
let outcome = match handler.metrics(request, query_ctx.clone()).await {
let outcome = match handler
.metrics(request, metric_row_batcher.clone(), query_ctx.clone())
.await
{
Ok(outcome) => outcome,
Err(error::Error::InvalidOtlpMetricInput { reason }) => {
let _ = sender
@@ -212,4 +235,10 @@ mod tests {
assert_eq!(ArrowStatusCode::InvalidArgument as i32, status.status_code);
assert_eq!(EXPONENTIAL_HISTOGRAM_UNSUPPORTED, status.status_message);
}
#[test]
fn otel_arrow_metric_context_uses_metric_engine_setting() {
assert!(otel_arrow_metric_context(true).with_metric_engine);
assert!(!otel_arrow_metric_context(false).with_metric_engine);
}
}
src/servers/src/pending_rows_batcher.rs
+906 -125
View File
File diff suppressed because it is too large Load Diff
src/servers/src/prom_row_builder.rs
+5 -5
View File
@@ -199,9 +199,9 @@ pub(crate) fn identify_missing_columns_from_proto(
Ok(missing)
}
/// Build a `Vec<ColumnSchema>` suitable for creating a new Prometheus logical table
/// Build a `Vec<ColumnSchema>` suitable for creating a new metric logical table
/// directly from the proto `rows.schema`, avoiding the round-trip through Arrow schema.
pub fn build_prom_create_table_schema_from_proto(
pub fn build_metric_create_table_schema_from_proto(
rows_schema: &[ColumnSchema],
) -> Result<Vec<ColumnSchema>> {
rows_schema
@@ -309,7 +309,7 @@ mod tests {
use arrow::datatypes::{DataType, Field, Schema as ArrowSchema, TimeUnit};
use super::{
build_prom_create_table_schema_from_proto, identify_missing_columns_from_proto,
build_metric_create_table_schema_from_proto, identify_missing_columns_from_proto,
rows_to_aligned_record_batch,
};
@@ -528,7 +528,7 @@ mod tests {
}
#[test]
fn test_build_prom_create_table_schema_from_proto() {
fn test_build_metric_create_table_schema_from_proto() {
let rows_schema = vec![
ColumnSchema {
column_name: "greptime_timestamp".to_string(),
@@ -550,7 +550,7 @@ mod tests {
},
];
let schema = build_prom_create_table_schema_from_proto(&rows_schema).unwrap();
let schema = build_metric_create_table_schema_from_proto(&rows_schema).unwrap();
assert_eq!(3, schema.len());
assert_eq!("greptime_timestamp", schema[0].column_name);
src/servers/src/query_handler.rs
+2
View File
@@ -54,6 +54,7 @@ use crate::error::Result;
use crate::http::jaeger::QueryTraceParams;
use crate::influxdb::InfluxdbRequest;
use crate::opentsdb::codec::DataPoint;
use crate::pending_rows_batcher::MetricRowBatcherRef;
pub type OpentsdbProtocolHandlerRef = Arc<dyn OpentsdbProtocolHandler + Send + Sync>;
pub type InfluxdbLineProtocolHandlerRef = Arc<dyn InfluxdbLineProtocolHandler + Send + Sync>;
pub type PromStoreProtocolHandlerRef = Arc<dyn PromStoreProtocolHandler + Send + Sync>;
@@ -141,6 +142,7 @@ pub trait OpenTelemetryProtocolHandler: PipelineHandler {
async fn metrics(
&self,
request: ExportMetricsServiceRequest,
metric_row_batcher: Option<MetricRowBatcherRef>,
ctx: QueryContextRef,
) -> Result<MetricsIngestOutcome>;
tests-integration/src/instance.rs
+1 -1
View File
@@ -777,7 +777,7 @@ mod tests {
};
let outcome =
OpenTelemetryProtocolHandler::metrics(&*instance, request, QueryContext::arc())
OpenTelemetryProtocolHandler::metrics(&*instance, request, None, QueryContext::arc())
.await
.unwrap();
let warning = outcome
tests-integration/src/otlp.rs
+1 -1
View File
@@ -116,7 +116,7 @@ mod test {
.is_ok()
);
let resp = instance.metrics(req, ctx.clone()).await;
let resp = instance.metrics(req, None, ctx.clone()).await;
assert!(resp.is_ok());
let mut output = instance
tests-integration/src/test_util.rs
+62 -12
View File
@@ -39,6 +39,7 @@ use common_test_util::temp_dir::{TempDir, create_temp_dir};
use common_wal::config::DatanodeWalConfig;
use datanode::config::{DatanodeOptions, StorageConfig};
use frontend::instance::Instance;
use frontend::server::Services;
use frontend::service_config::{MysqlOptions, PostgresOptions};
use mito2::gc::GcConfig;
use object_store::config::{
@@ -54,7 +55,10 @@ use servers::http::{HttpOptions, HttpServerBuilder};
use servers::metrics_handler::MetricsHandler;
use servers::mysql::server::{MysqlServer, MysqlSpawnConfig, MysqlSpawnRef};
use servers::otel_arrow::OtelArrowServiceHandler;
use servers::pending_rows_batcher::PendingRowsBatcher;
use servers::pending_rows_batcher::{
MetricRowBatcherRef, PendingRowsBatchMode, PendingRowsBatcher, PendingRowsBatcherOptions,
PendingRowsBatcherTuning,
};
use servers::postgres::PostgresServer;
use servers::prom_remote_write::validation::PromValidationMode;
use servers::query_handler::sql::SqlQueryHandler;
@@ -469,6 +473,42 @@ pub async fn setup_test_http_app_with_frontend(
setup_test_http_app_with_frontend_and_user_provider(store_type, name, None).await
}
/// Builds the real OTLP HTTP route with synchronous metric pending-row batching enabled.
pub async fn setup_test_otlp_metrics_app_with_frontend_batched(
store_type: StorageType,
name: &str,
) -> (Router, TestGuard) {
let instance = setup_standalone_instance(name, store_type).await;
let frontend = instance.fe_instance().clone();
let mut options = instance.opts.clone();
options.prom_store.with_metric_engine = true;
options.prom_store.pending_rows_flush_interval = Duration::from_millis(50);
options.otlp.enable = true;
options.otlp.enable_metrics_batching = true;
assert!(
servers::pending_rows_batcher::pending_rows_batch_sync_enabled(),
"OTLP batching integration tests require synchronous confirmation"
);
let services = Services::new(options.clone(), frontend.clone(), Plugins::new());
let frontend_options = options.clone().into();
let http_server =
services.http_server_builder(&frontend_options, services.server_memory_limiter.clone());
assert_eq!(
Some(true),
services.otlp_http_metric_batcher_configured(),
"production Services HTTP assembly must pass its batcher to OTLP"
);
let http_server = http_server
.with_greptime_config_options(options.to_toml().unwrap())
.build();
(
http_server.build(http_server.make_app()).unwrap(),
instance.guard,
)
}
pub async fn setup_test_http_app_with_frontend_and_slow_query_threshold(
store_type: StorageType,
name: &str,
@@ -490,7 +530,7 @@ pub async fn setup_test_http_app_with_frontend_and_slow_query_threshold(
.with_log_ingest_handler(instance.fe_instance().clone(), None, None)
.with_logs_handler(instance.fe_instance().clone())
.with_influxdb_handler(instance.fe_instance().clone())
.with_otlp_handler(instance.fe_instance().clone(), true, false)
.with_otlp_handler(instance.fe_instance().clone(), true, false, None)
.with_jaeger_handler(instance.fe_instance().clone())
.with_greptime_config_options(instance.opts.to_toml().unwrap())
.build();
@@ -558,6 +598,7 @@ pub async fn setup_test_http_app_with_frontend_and_custom_options(
instance.fe_instance().clone(),
true,
experimental_enable_exponential_histogram,
None,
)
.with_prometheus_handler(instance.fe_instance().clone())
.with_jaeger_handler(instance.fe_instance().clone())
@@ -671,20 +712,24 @@ async fn setup_test_prom_app_with_frontend_inner(
// Mirror the production wiring at `frontend::server`: build the batcher from the
// instance's managers. A short flush interval keeps the test responsive.
let pending_rows_batcher = if enable_batcher {
PendingRowsBatcher::try_new(
Some(PendingRowsBatcher::new(
frontend_ref.partition_manager().clone(),
frontend_ref.node_manager().clone(),
frontend_ref.catalog_manager().clone(),
frontend_ref.table_flownode_set_cache().clone(),
true,
frontend_ref.clone(),
Duration::from_millis(50),
1000,
4,
64,
64,
std::num::NonZeroUsize::new(1024).unwrap(),
)
PendingRowsBatcherOptions::try_new(PendingRowsBatcherTuning {
flush_interval: Duration::from_millis(50),
max_batch_rows: 1000,
max_concurrent_flushes: 4,
worker_channel_capacity: 64,
max_inflight_requests: 64,
flow_notification_queue_capacity: std::num::NonZeroUsize::new(1024).unwrap(),
})
.unwrap(),
PendingRowsBatchMode::Synchronous,
))
} else {
None
};
@@ -697,7 +742,7 @@ async fn setup_test_prom_app_with_frontend_inner(
true,
PromValidationMode::Strict,
experimental_enable_prometheus_native_histogram,
pending_rows_batcher,
pending_rows_batcher.map(|batcher| batcher as MetricRowBatcherRef),
)
.with_prometheus_handler(frontend_ref)
.with_greptime_config_options(instance.opts.datanode_options().to_toml().unwrap())
@@ -825,7 +870,12 @@ async fn setup_grpc_server_for_frontend_instance_with(
.database_handler(greptime_request_handler)
.flight_handler(flight_handler)
.prometheus_handler(fe_instance_ref.clone(), user_provider.clone())
.otel_arrow_handler(OtelArrowServiceHandler::new(fe_instance_ref, user_provider))
.otel_arrow_handler(OtelArrowServiceHandler::new(
fe_instance_ref,
user_provider,
true,
None,
))
.with_tls_config(grpc_config.tls)
.unwrap();
tests-integration/tests/http.rs
+190 -2
View File
@@ -80,7 +80,8 @@ use table::table_name::TableName;
use tests_integration::test_util::{
StorageType, setup_test_http_app, setup_test_http_app_with_frontend,
setup_test_http_app_with_frontend_and_slow_query_threshold,
setup_test_http_app_with_frontend_and_user_provider, setup_test_prom_app_with_frontend,
setup_test_http_app_with_frontend_and_user_provider,
setup_test_otlp_metrics_app_with_frontend_batched, setup_test_prom_app_with_frontend,
setup_test_prom_app_with_frontend_batched, setup_test_prom_app_with_frontend_native_histogram,
};
use urlencoding::encode;
@@ -2731,7 +2732,6 @@ pub async fn test_prometheus_remote_write(store_type: StorageType) {
r#"[["{\"protocol\":\"prometheus\",\"reason\":\"auto_alter\"}"]]"#,
)
.await;
guard.remove_all().await;
}
@@ -6969,6 +6969,194 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
guard.remove_all().await;
}
async fn send_otlp_metrics_json(client: &TestClient, content: &str) -> TestResponse {
let request: ExportMetricsServiceRequest = serde_json::from_str(content).unwrap();
send_req(
client,
vec![
(
HeaderName::from_static("content-type"),
HeaderValue::from_static("application/x-protobuf"),
),
(
HeaderName::from_static("x-greptime-otlp-metric-translation-strategy"),
HeaderValue::from_static("NoTranslation"),
),
],
"/v1/otlp/v1/metrics",
request.encode_to_vec(),
false,
)
.await
}
#[tokio::test(flavor = "multi_thread")]
async fn test_otlp_metrics_batching_flushes_multiple_small_requests() {
common_telemetry::init_default_ut_logging();
let (app, mut guard) = setup_test_otlp_metrics_app_with_frontend_batched(
StorageType::File,
"test_otlp_metrics_batching_flushes_multiple_small_requests",
)
.await;
let client = TestClient::new(app).await;
let responses = tokio::join!(
send_otlp_metrics_json(
&client,
r#"{"resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"task6_gauge","unit":"1","gauge":{"dataPoints":[{"timeUnixNano":"1000000","asDouble":1.25}]}}]}]}]}"#,
),
send_otlp_metrics_json(
&client,
r#"{"resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"task6_sum","unit":"requests","sum":{"dataPoints":[{"timeUnixNano":"2000000","asDouble":2.5}],"aggregationTemporality":1,"isMonotonic":true}}]}]}]}"#,
),
send_otlp_metrics_json(
&client,
r#"{"resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"task6_gauge","unit":"1","gauge":{"dataPoints":[{"timeUnixNano":"3000000","asDouble":3.75}]}}]}]}]}"#,
),
send_otlp_metrics_json(
&client,
r#"{"resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"task6_sum","unit":"requests","sum":{"dataPoints":[{"timeUnixNano":"4000000","asDouble":4.5}],"aggregationTemporality":1,"isMonotonic":true}}]}]}]}"#,
),
);
for response in [responses.0, responses.1, responses.2, responses.3] {
assert_eq!(StatusCode::OK, response.status());
let response = ExportMetricsServiceResponse::decode(response.bytes().await).unwrap();
assert!(response.partial_success.is_none());
}
validate_data(
"otlp_batched_gauge_rows",
&client,
"select greptime_timestamp, greptime_value from task6_gauge order by greptime_timestamp;",
"[[1,1.25],[3,3.75]]",
)
.await;
validate_data(
"otlp_batched_sum_rows",
&client,
"select greptime_timestamp, greptime_value from task6_sum order by greptime_timestamp;",
"[[2,2.5],[4,4.5]]",
)
.await;
guard.remove_all().await;
}
#[tokio::test(flavor = "multi_thread")]
async fn test_otlp_metrics_batching_preserves_fanout_semantic_options() {
common_telemetry::init_default_ut_logging();
let (app, mut guard) = setup_test_otlp_metrics_app_with_frontend_batched(
StorageType::File,
"test_otlp_metrics_batching_preserves_fanout_semantic_options",
)
.await;
let client = TestClient::new(app).await;
let content = r#"
{"resourceMetrics":[{"scopeMetrics":[{"metrics":[
{"name":"task6_histogram","unit":"ms","histogram":{"dataPoints":[{"attributes":[{"key":"route","value":{"stringValue":"read"}}],"timeUnixNano":"1000000","count":"3","sum":6.0,"bucketCounts":["1","2"],"explicitBounds":[1.0]}],"aggregationTemporality":2}},
{"name":"task6_summary","unit":"items","summary":{"dataPoints":[{"attributes":[{"key":"route","value":{"stringValue":"write"}}],"timeUnixNano":"2000000","count":"4","sum":10.0,"quantileValues":[{"quantile":0.5,"value":2.0},{"quantile":0.9,"value":4.0}]}]}}
]}]}]}
"#;
let response = send_otlp_metrics_json(&client, content).await;
assert_eq!(StatusCode::OK, response.status());
let response = ExportMetricsServiceResponse::decode(response.bytes().await).unwrap();
assert!(response.partial_success.is_none());
validate_data(
"otlp_batched_histogram_rows",
&client,
"select greptime_timestamp, route, le, greptime_value from task6_histogram_bucket order by le;",
"[[1,\"read\",\"1\",1.0],[1,\"read\",\"inf\",3.0]]",
)
.await;
validate_data(
"otlp_batched_histogram_count",
&client,
"select greptime_timestamp, route, greptime_value from task6_histogram_count;",
"[[1,\"read\",3.0]]",
)
.await;
validate_data(
"otlp_batched_histogram_sum",
&client,
"select greptime_timestamp, route, greptime_value from task6_histogram_sum;",
"[[1,\"read\",6.0]]",
)
.await;
validate_data(
"otlp_batched_summary_rows",
&client,
"select greptime_timestamp, route, quantile, greptime_value from task6_summary order by quantile;",
"[[2,\"write\",\"0.5\",2.0],[2,\"write\",\"0.9\",4.0]]",
)
.await;
validate_data(
"otlp_batched_summary_count",
&client,
"select greptime_timestamp, route, greptime_value from task6_summary_count;",
"[[2,\"write\",4.0]]",
)
.await;
validate_data(
"otlp_batched_summary_sum",
&client,
"select greptime_timestamp, route, greptime_value from task6_summary_sum;",
"[[2,\"write\",10.0]]",
)
.await;
validate_data(
"otlp_batched_fanout_semantics",
&client,
"select table_name from information_schema.tables where table_schema = 'public' and (\
(table_name = 'task6_histogram_bucket' and create_options like '%greptime.semantic.metric.type=histogram%' and create_options like '%greptime.semantic.metric.unit=ms%' and create_options like '%greptime.semantic.metric.temporality=cumulative%' and create_options like '%greptime.semantic.metric.metadata_quality=declared%' and create_options like '%greptime.semantic.signal_type=metric%' and create_options like '%greptime.semantic.source=opentelemetry%') or \
(table_name in ('task6_histogram_count', 'task6_histogram_sum') and create_options like '%greptime.semantic.metric.type=counter%' and create_options like '%greptime.semantic.metric.unit=ms%' and create_options like '%greptime.semantic.metric.temporality=cumulative%' and create_options like '%greptime.semantic.metric.metadata_quality=declared%' and create_options like '%greptime.semantic.signal_type=metric%' and create_options like '%greptime.semantic.source=opentelemetry%') or \
(table_name = 'task6_summary' and create_options like '%greptime.semantic.metric.type=summary%' and create_options like '%greptime.semantic.metric.unit=items%' and create_options like '%greptime.semantic.metric.metadata_quality=declared%' and create_options like '%greptime.semantic.signal_type=metric%' and create_options like '%greptime.semantic.source=opentelemetry%' and create_options not like '%greptime.semantic.metric.temporality=%') or \
(table_name in ('task6_summary_count', 'task6_summary_sum') and create_options like '%greptime.semantic.metric.type=counter%' and create_options like '%greptime.semantic.metric.unit=items%' and create_options like '%greptime.semantic.metric.metadata_quality=declared%' and create_options like '%greptime.semantic.signal_type=metric%' and create_options like '%greptime.semantic.source=opentelemetry%' and create_options not like '%greptime.semantic.metric.temporality=%')) order by table_name;",
"[[\"task6_histogram_bucket\"],[\"task6_histogram_count\"],[\"task6_histogram_sum\"],[\"task6_summary\"],[\"task6_summary_count\"],[\"task6_summary_sum\"]]",
)
.await;
guard.remove_all().await;
}
#[tokio::test(flavor = "multi_thread")]
async fn test_otlp_metrics_batching_evolves_new_tag_schema() {
common_telemetry::init_default_ut_logging();
let (app, mut guard) = setup_test_otlp_metrics_app_with_frontend_batched(
StorageType::File,
"test_otlp_metrics_batching_evolves_new_tag_schema",
)
.await;
let client = TestClient::new(app).await;
for content in [
r#"{"resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"task6_schema_gauge","gauge":{"dataPoints":[{"attributes":[{"key":"host","value":{"stringValue":"old"}}],"timeUnixNano":"1000000","asDouble":1.0}]}}]}]}]}"#,
r#"{"resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"task6_schema_gauge","gauge":{"dataPoints":[{"attributes":[{"key":"host","value":{"stringValue":"new"}},{"key":"zone","value":{"stringValue":"west"}}],"timeUnixNano":"2000000","asDouble":2.0}]}}]}]}]}"#,
] {
let response = send_otlp_metrics_json(&client, content).await;
assert_eq!(StatusCode::OK, response.status());
let response = ExportMetricsServiceResponse::decode(response.bytes().await).unwrap();
assert!(response.partial_success.is_none());
}
validate_data(
"otlp_batched_schema_evolution",
&client,
"select host, zone, greptime_value from task6_schema_gauge order by greptime_timestamp;",
"[[\"old\",null,1.0],[\"new\",\"west\",2.0]]",
)
.await;
validate_data(
"otlp_batched_schema_evolution_tag_semantics",
&client,
"select column_name, semantic_type from information_schema.columns where table_name = 'task6_schema_gauge' and column_name in ('host', 'zone') order by column_name;",
r#"[["host","TAG"],["zone","TAG"]]"#,
)
.await;
guard.remove_all().await;
}
pub async fn test_otlp_exponential_histogram(store_type: StorageType) {
use opentelemetry_proto::tonic::metrics::v1::{
AggregationTemporality, ExponentialHistogram, ExponentialHistogramDataPoint, Metric,