diff --git a/config/config.md b/config/config.md
index a8d12c3557e..785d86e69df 100644
--- a/config/config.md
+++ b/config/config.md
@@ -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.
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. |
+| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.
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. |
| `http.body_limit` | String | `64MB` | HTTP request body limit.
The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.
Set to 0 to disable limit. |
| `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default
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.
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. |
| `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
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.
Available options:
- strict: deny invalid UTF-8 strings (default).
- lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).
- 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.
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.
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.
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.
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. |
+| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.
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. |
| `http.body_limit` | String | `64MB` | HTTP request body limit.
The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.
Set to 0 to disable limit. |
| `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default
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.
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. |
| `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
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.
Available options:
- strict: deny invalid UTF-8 strings (default).
- lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).
- 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.
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.
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. |
diff --git a/config/frontend.example.toml b/config/frontend.example.toml
index 7c6862916e4..80e18e7831d 100644
--- a/config/frontend.example.toml
+++ b/config/frontend.example.toml
@@ -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
diff --git a/config/standalone.example.toml b/config/standalone.example.toml
index b05cf0177d0..992c3297d4c 100644
--- a/config/standalone.example.toml
+++ b/config/standalone.example.toml
@@ -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
diff --git a/grafana/dashboards/metrics/cluster/dashboard.json b/grafana/dashboards/metrics/cluster/dashboard.json
index bd22a0da597..e20a23032a0 100644
--- a/grafana/dashboards/metrics/cluster/dashboard.json
+++ b/grafana/dashboards/metrics/cluster/dashboard.json
@@ -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"
},
{
diff --git a/grafana/dashboards/metrics/cluster/dashboard.md b/grafana/dashboards/metrics/cluster/dashboard.md
index b8e050114c6..b04b43b3d3b 100644
--- a/grafana/dashboards/metrics/cluster/dashboard.md
+++ b/grafana/dashboards/metrics/cluster/dashboard.md
@@ -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"})`
`sum by (instance, pod) (greptime_mito_compaction_memory_in_use_bytes{instance=~"$datanode"})`
`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"})`
`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"})`
`sum by (instance, pod) (greptime_prom_store_pending_batches{instance=~"$frontend"})`
`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"})`
`sum by (instance, pod) (greptime_prom_store_pending_batches{instance=~"$frontend"})`
`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"})`
`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 |
diff --git a/grafana/dashboards/metrics/cluster/dashboard.yaml b/grafana/dashboards/metrics/cluster/dashboard.yaml
index cd105edfcd2..9c139af4ae7 100644
--- a/grafana/dashboards/metrics/cluster/dashboard.yaml
+++ b/grafana/dashboards/metrics/cluster/dashboard.yaml
@@ -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"})
diff --git a/grafana/dashboards/metrics/standalone/dashboard.json b/grafana/dashboards/metrics/standalone/dashboard.json
index a90d05760e4..e6410fc780d 100644
--- a/grafana/dashboards/metrics/standalone/dashboard.json
+++ b/grafana/dashboards/metrics/standalone/dashboard.json
@@ -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"
},
{
diff --git a/grafana/dashboards/metrics/standalone/dashboard.md b/grafana/dashboards/metrics/standalone/dashboard.md
index 2ed84c59213..1187c2ff958 100644
--- a/grafana/dashboards/metrics/standalone/dashboard.md
+++ b/grafana/dashboards/metrics/standalone/dashboard.md
@@ -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)`
`sum by (instance, pod) (greptime_mito_compaction_memory_in_use_bytes)`
`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)`
`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)`
`sum by (instance, pod) (greptime_prom_store_pending_batches)`
`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)`
`sum by (instance, pod) (greptime_prom_store_pending_batches)`
`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)`
`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 |
diff --git a/grafana/dashboards/metrics/standalone/dashboard.yaml b/grafana/dashboards/metrics/standalone/dashboard.yaml
index 8da2a8fa6f1..437a34e4674 100644
--- a/grafana/dashboards/metrics/standalone/dashboard.yaml
+++ b/grafana/dashboards/metrics/standalone/dashboard.yaml
@@ -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)
diff --git a/src/cmd/src/standalone.rs b/src/cmd/src/standalone.rs
index 6c5ac25f641..a3422abd0c4 100644
--- a/src/cmd/src/standalone.rs
+++ b/src/cmd/src/standalone.rs
@@ -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);
diff --git a/src/frontend/src/frontend.rs b/src/frontend/src/frontend.rs
index bb15abd7bbf..649e3585f0c 100644
--- a/src/frontend/src/frontend.rs
+++ b/src/frontend/src/frontend.rs
@@ -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);
diff --git a/src/frontend/src/instance/otlp.rs b/src/frontend/src/instance/otlp.rs
index 4c80f1d17f2..eb1dda480db 100644
--- a/src/frontend/src/instance/otlp.rs
+++ b/src/frontend/src/instance/otlp.rs
@@ -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,
ctx: QueryContextRef,
) -> ServerResult {
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>,
+ protocols: Mutex>,
+ 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 {
+ 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 {
+ 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 {
+ 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) -> 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::()
+ 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::(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() {
diff --git a/src/frontend/src/instance/prom_store.rs b/src/frontend/src/instance/prom_store.rs
index e0b17b1a89e..af2a95857c7 100644
--- a/src/frontend/src/instance/prom_store.rs
+++ b/src/frontend/src/instance/prom_store.rs
@@ -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,
diff --git a/src/frontend/src/server.rs b/src/frontend/src/server.rs
index e112e5ae2a9..70490977efb 100644
--- a/src/frontend/src/server.rs
+++ b/src/frontend/src/server.rs
@@ -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,
internal_flight_handler: Option,
+ metric_batching: ResolvedMetricBatching,
+ pending_rows_batcher: Option>,
+ #[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)>>>;
+
+struct ResolvedMetricBatching {
+ options: Option,
+ 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 Services
where
T: Into + Configurable + Clone,
{
pub fn new(opts: T, instance: Arc, 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,
+ 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,
+ 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,
+ 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::();
let ingest_interceptor = self.plugins.get::>();
@@ -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 {
+ 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 {
+ 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 {
+ self.metric_row_batcher(self.metric_batching.otlp_enabled)
+ }
+
+ #[cfg(any(test, feature = "testing"))]
+ fn record_metric_batcher(
+ &self,
+ consumer: MetricBatchConsumer,
+ batcher: &Option,
+ ) {
+ 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 {
+ 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,
+ 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:?}");
+ }
+ }
}
diff --git a/src/frontend/src/service_config/otlp.rs b/src/frontend/src/service_config/otlp.rs
index d6a36f702e4..a556aa0a359 100644
--- a/src/frontend/src/service_config/otlp.rs
+++ b/src/frontend/src/service_config/otlp.rs
@@ -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::(&serialized).unwrap(), options);
+
+ let options: OtlpOptions = toml::from_str("enable_metrics_batching = true").unwrap();
+ assert!(options.enable_metrics_batching);
+ assert!(options.enable);
}
}
diff --git a/src/frontend/src/service_config/prom_store.rs b/src/frontend/src/service_config/prom_store.rs
index d2e386caae0..39c8e6e9026 100644
--- a/src/frontend/src/service_config/prom_store.rs
+++ b/src/frontend/src/service_config/prom_store.rs
@@ -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::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());
+ }
}
diff --git a/src/servers/src/http.rs b/src/servers/src/http.rs
index 311ed48d912..a7cfca2603a 100644
--- a/src/servers/src/http.rs
+++ b/src/servers/src/http.rs
@@ -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>,
+ pending_rows_batcher: Option,
) -> 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,
) -> 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,
) -> Router {
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,
})
}
diff --git a/src/servers/src/http/otlp.rs b/src/servers/src/http/otlp.rs
index 2d5385c5e33..a5da2ad8c37 100644
--- a/src/servers/src/http/otlp.rs
+++ b/src/servers/src/http/otlp.rs
@@ -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,
}
#[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))
diff --git a/src/servers/src/http/prom_store.rs b/src/servers/src/http/prom_store.rs
index cb7b6307c4b..3c0f5fc50ad 100644
--- a/src/servers/src/http/prom_store.rs
+++ b/src/servers/src/http/prom_store.rs
@@ -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>,
+ pub pending_rows_batcher: Option,
}
#[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;
-}
-
-#[async_trait]
-impl PromWriteBatcher for PendingRowsBatcher {
- async fn submit(&self, requests: RowInsertRequests, ctx: QueryContextRef) -> Result {
- PendingRowsBatcher::submit(self, requests, ctx).await
- }
-}
-
fn into_prom_write_batches(req: ContextReq, query_ctx: QueryContextRef) -> Vec {
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>,
+ pending_rows_batcher: Option,
prom_store_with_metric_engine: bool,
mut batches: Vec,
) -> std::result::Result {
@@ -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>,
+ pending_rows_batcher: Option,
prom_store_with_metric_engine: bool,
sample_batches: Vec,
histogram_batches: Vec,
@@ -539,13 +527,18 @@ async fn write_prometheus_v2_rows_with_progress(
})
}
-async fn write_batched_prometheus_v2_rows_with_progress(
+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,
histogram_batches: Vec,
) -> std::result::Result {
+ 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 PromWriteBatch {
@@ -904,15 +966,40 @@ mod tests {
));
}
- struct RecordingPromWriteBatcher {
+ struct RecordingMetricRowBatcher {
events: Arc>>,
+ protocols: Arc>>,
+ rows_written: u64,
}
#[async_trait]
- impl PromWriteBatcher for RecordingPromWriteBatcher {
- async fn submit(&self, requests: RowInsertRequests, _ctx: QueryContextRef) -> Result {
+ impl MetricRowBatcher for RecordingMetricRowBatcher {
+ async fn submit(
+ &self,
+ requests: RowInsertRequests,
+ _ctx: QueryContextRef,
+ protocol: MetricRowBatchProtocol,
+ ) -> Result {
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 {
+ Err(error::InternalSnafu {
+ err_msg: "metric row batcher failed".to_string(),
+ }
+ .build())
}
}
diff --git a/src/servers/src/metrics.rs b/src/servers/src/metrics.rs
index 0d70540be31..4b56185bd7d 100644
--- a/src/servers/src/metrics.rs
+++ b/src/servers/src/metrics.rs
@@ -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",
diff --git a/src/servers/src/otel_arrow.rs b/src/servers/src/otel_arrow.rs
index 730020161d3..65ea169d1a6 100644
--- a/src/servers/src/otel_arrow.rs
+++ b/src/servers/src/otel_arrow.rs
@@ -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 {
handler: T,
user_provider: Option,
+ with_metric_engine: bool,
+ metric_row_batcher: Option,
}
impl OtelArrowServiceHandler {
- pub fn new(handler: T, user_provider: Option) -> Self {
+ pub fn new(
+ handler: T,
+ user_provider: Option,
+ with_metric_engine: bool,
+ metric_row_batcher: Option,
+ ) -> 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 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);
+ }
}
diff --git a/src/servers/src/pending_rows_batcher.rs b/src/servers/src/pending_rows_batcher.rs
index 03eb90dce28..3c580d537de 100644
--- a/src/servers/src/pending_rows_batcher.rs
+++ b/src/servers/src/pending_rows_batcher.rs
@@ -22,7 +22,7 @@ use api::v1::meta::Peer;
use api::v1::region::{
BulkInsertRequest, RegionRequest, RegionRequestHeader, bulk_insert_request, region_request,
};
-use api::v1::{ArrowIpc, ColumnSchema, RowInsertRequests, Rows};
+use api::v1::{ArrowIpc, ColumnDataType, ColumnSchema, RowInsertRequests, Rows, SemanticType};
use arrow::array::Array;
use arrow::compute::{concat_batches, filter_record_batch};
use arrow::datatypes::{DataType as ArrowDataType, Schema as ArrowSchema, TimeUnit};
@@ -53,11 +53,13 @@ use crate::error;
use crate::error::{Error, Result};
use crate::metrics::{
FLOW_NOTIFICATION_DROPPED, FLUSH_DROPPED_ROWS, FLUSH_ELAPSED, FLUSH_FAILURES, FLUSH_ROWS,
- FLUSH_TOTAL, PENDING_BATCHES, PENDING_ROWS, PENDING_ROWS_BATCH_FLUSH_STAGE_ELAPSED,
- PENDING_ROWS_BATCH_INGEST_STAGE_ELAPSED, PENDING_WORKERS,
+ FLUSH_TOTAL, METRIC_CANCELLED_VALUE, METRIC_FAILURE_VALUE, METRIC_ROW_BATCH_SUBMISSIONS,
+ METRIC_ROW_BATCH_WAIT_FLUSH_RESULT_ELAPSED, METRIC_SUCCESS_VALUE, PENDING_BATCHES,
+ PENDING_ROWS, PENDING_ROWS_BATCH_FLUSH_STAGE_ELAPSED, PENDING_ROWS_BATCH_INGEST_STAGE_ELAPSED,
+ PENDING_WORKERS,
};
use crate::prom_row_builder::{
- 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,
};
@@ -69,9 +71,8 @@ const PENDING_ROWS_BATCH_SYNC_ENV: &str = "PENDING_ROWS_BATCH_SYNC";
/// before replying to the client (synchronous mode), controlled by the
/// `PENDING_ROWS_BATCH_SYNC` environment variable and defaulting to `true`.
///
-/// Callers that reason about how long a remote write request may block (e.g.
-/// the frontend HTTP timeout fallback) must consult this instead of
-/// duplicating the env lookup.
+/// Startup code must read this once and propagate the resolved mode to batcher
+/// construction and callers that derive timeout behavior.
pub fn pending_rows_batch_sync_enabled() -> bool {
std::env::var(PENDING_ROWS_BATCH_SYNC_ENV)
.ok()
@@ -79,9 +80,113 @@ pub fn pending_rows_batch_sync_enabled() -> bool {
.and_then(|v| v.parse::().ok())
.unwrap_or(true)
}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PendingRowsBatchMode {
+ Synchronous,
+ Asynchronous,
+}
+
+impl PendingRowsBatchMode {
+ pub fn is_synchronous(self) -> bool {
+ matches!(self, Self::Synchronous)
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct PendingRowsBatcherTuning {
+ pub flush_interval: Duration,
+ pub max_batch_rows: usize,
+ pub max_concurrent_flushes: usize,
+ pub worker_channel_capacity: usize,
+ pub max_inflight_requests: usize,
+ pub flow_notification_queue_capacity: NonZeroUsize,
+}
+
+#[derive(Debug, Clone)]
+pub struct PendingRowsBatcherOptions {
+ tuning: PendingRowsBatcherTuning,
+}
+
+impl PendingRowsBatcherOptions {
+ pub fn try_new(tuning: PendingRowsBatcherTuning) -> Option {
+ if tuning.flush_interval.is_zero()
+ || tuning.max_batch_rows == 0
+ || tuning.max_concurrent_flushes == 0
+ || tuning.worker_channel_capacity == 0
+ || tuning.max_inflight_requests == 0
+ {
+ return None;
+ }
+
+ Some(Self { tuning })
+ }
+
+ pub fn flush_interval(&self) -> Duration {
+ self.tuning.flush_interval
+ }
+}
const WORKER_IDLE_TIMEOUT_MULTIPLIER: u32 = 3;
const PHYSICAL_REGION_ESSENTIAL_COLUMN_COUNT: usize = 3;
const MAX_CONCURRENT_FLOW_NOTIFICATIONS: usize = 8;
+
+/// Batches protocol-neutral metric rows for ingestion.
+#[async_trait]
+pub trait MetricRowBatcher: Send + Sync {
+ /// Submits row insert requests and returns the number of accepted rows.
+ async fn submit(
+ &self,
+ requests: RowInsertRequests,
+ ctx: QueryContextRef,
+ protocol: MetricRowBatchProtocol,
+ ) -> Result;
+}
+
+pub type MetricRowBatcherRef = Arc;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum MetricRowBatchProtocol {
+ Prometheus,
+ Otlp,
+}
+
+impl MetricRowBatchProtocol {
+ pub const fn as_str(self) -> &'static str {
+ match self {
+ Self::Prometheus => "prometheus",
+ Self::Otlp => "otlp",
+ }
+ }
+}
+
+struct WaitFlushResultTimer {
+ started_at: Instant,
+ protocol: MetricRowBatchProtocol,
+ result_label: &'static str,
+}
+
+impl WaitFlushResultTimer {
+ fn start(protocol: MetricRowBatchProtocol) -> Self {
+ Self {
+ started_at: Instant::now(),
+ protocol,
+ result_label: METRIC_CANCELLED_VALUE,
+ }
+ }
+
+ fn finish_with_result(mut self, result_label: &'static str) {
+ self.result_label = result_label;
+ }
+}
+
+impl Drop for WaitFlushResultTimer {
+ fn drop(&mut self) {
+ METRIC_ROW_BATCH_WAIT_FLUSH_RESULT_ELAPSED
+ .with_label_values(&[self.protocol.as_str(), self.result_label])
+ .observe(self.started_at.elapsed().as_secs_f64());
+ }
+}
+
#[async_trait]
pub trait PendingRowsSchemaAlterer: Send + Sync {
/// Batch-create multiple logical tables that are missing.
@@ -97,7 +202,7 @@ pub trait PendingRowsSchemaAlterer: Send + Sync {
/// Batch-alter multiple logical tables to add missing tag columns.
/// Each entry is `(table_name, missing_column_names)`.
- async fn add_missing_prom_tag_columns_batch(
+ async fn add_missing_metric_tag_columns_batch(
&self,
catalog: &str,
schema: &str,
@@ -145,6 +250,10 @@ pub trait PhysicalFlushNodeRequester: Send + Sync {
) -> Result;
}
+type PhysicalFlushCatalogProviderRef = Arc;
+type PhysicalFlushPartitionProviderRef = Arc;
+type PhysicalFlushNodeRequesterRef = Arc;
+
#[derive(Clone)]
struct CatalogManagerPhysicalFlushAdapter {
catalog_manager: CatalogManagerRef,
@@ -341,83 +450,195 @@ fn batch_key_from_ctx(ctx: &QueryContextRef) -> BatchKey {
}
}
-/// Prometheus remote write pending rows batcher.
+/// Returns whether all non-empty inserts in `requests` can use the scalar metric batcher.
+///
+/// Empty inserts are ignored to match [`PendingRowsBatcher::submit`] filtering, and a request
+/// without rows is not batchable. Valid schemas outside the scalar metric shape return `false`,
+/// while malformed scalar row widths or value variants return an error before callers perform DDL.
+pub fn is_scalar_metric_batchable(requests: &RowInsertRequests) -> Result {
+ let mut has_rows = false;
+ let mut batchable = true;
+
+ for request in &requests.inserts {
+ let Some(rows) = &request.rows else {
+ continue;
+ };
+ if rows.rows.is_empty() {
+ continue;
+ }
+ has_rows = true;
+ if !is_scalar_metric_schema(&rows.schema) {
+ batchable = false;
+ continue;
+ }
+ validate_row_values(&request.table_name, rows)?;
+ }
+
+ Ok(has_rows && batchable)
+}
+
+fn validate_row_values(table_name: &str, rows: &Rows) -> Result<()> {
+ let datatypes = rows
+ .schema
+ .iter()
+ .map(|column| parse_column_datatype(table_name, column))
+ .collect::>>()?;
+
+ for (row_index, row) in rows.rows.iter().enumerate() {
+ ensure!(
+ row.values.len() == rows.schema.len(),
+ error::InvalidParameterSnafu {
+ reason: format!(
+ "Column count mismatch in table '{}' row {}, expected {}, got {}",
+ table_name,
+ row_index,
+ rows.schema.len(),
+ row.values.len()
+ )
+ }
+ );
+
+ for (column_index, (value, datatype)) in row.values.iter().zip(&datatypes).enumerate() {
+ let Some(value_datatype) = api::helper::proto_value_type(value) else {
+ continue;
+ };
+ ensure!(
+ value_datatype == *datatype,
+ error::InvalidParameterSnafu {
+ reason: format!(
+ "Value variant mismatch in table '{}' row {}, column '{}' (index {}), expected {:?}, got {:?}",
+ table_name,
+ row_index,
+ rows.schema[column_index].column_name,
+ column_index,
+ datatype,
+ value_datatype
+ )
+ }
+ );
+ }
+ }
+
+ Ok(())
+}
+
+fn parse_column_datatype(table_name: &str, column: &ColumnSchema) -> Result {
+ ColumnDataType::try_from(column.datatype).map_err(|_| {
+ error::InvalidParameterSnafu {
+ reason: format!(
+ "Unknown datatype {} for column '{}' in table '{}'",
+ column.datatype, column.column_name, table_name
+ ),
+ }
+ .build()
+ })
+}
+
+fn is_scalar_metric_schema(schema: &[ColumnSchema]) -> bool {
+ let timestamp_count = schema
+ .iter()
+ .filter(|column| {
+ column.datatype == ColumnDataType::TimestampMillisecond as i32
+ && column.semantic_type == SemanticType::Timestamp as i32
+ })
+ .count();
+ let field_count = schema
+ .iter()
+ .filter(|column| {
+ column.datatype == ColumnDataType::Float64 as i32
+ && column.semantic_type == SemanticType::Field as i32
+ })
+ .count();
+
+ timestamp_count == 1
+ && field_count == 1
+ && schema.iter().all(|column| {
+ (column.datatype == ColumnDataType::TimestampMillisecond as i32
+ && column.semantic_type == SemanticType::Timestamp as i32)
+ || (column.datatype == ColumnDataType::Float64 as i32
+ && column.semantic_type == SemanticType::Field as i32)
+ || (column.datatype == ColumnDataType::String as i32
+ && column.semantic_type == SemanticType::Tag as i32)
+ })
+ && schema
+ .iter()
+ .all(|column| column.datatype_extension.is_none())
+}
+
+/// Metric pending rows batcher.
pub struct PendingRowsBatcher {
workers: Arc>,
flush_interval: Duration,
max_batch_rows: usize,
- partition_manager: PartitionRuleManagerRef,
- node_manager: NodeManagerRef,
+ partition_provider: PhysicalFlushPartitionProviderRef,
+ node_requester: PhysicalFlushNodeRequesterRef,
catalog_manager: CatalogManagerRef,
+ flush_catalog_provider: PhysicalFlushCatalogProviderRef,
flow_notification_tx: mpsc::Sender,
flush_semaphore: Arc,
inflight_semaphore: Arc,
worker_channel_capacity: usize,
- prom_store_with_metric_engine: bool,
+ with_metric_engine: bool,
schema_alterer: PendingRowsSchemaAltererRef,
- pending_rows_batch_sync: bool,
+ batch_mode: PendingRowsBatchMode,
shutdown: broadcast::Sender<()>,
}
impl PendingRowsBatcher {
#[allow(clippy::too_many_arguments)]
- pub fn try_new(
+ pub fn new(
partition_manager: PartitionRuleManagerRef,
node_manager: NodeManagerRef,
catalog_manager: CatalogManagerRef,
table_flownode_set_cache: TableFlownodeSetCacheRef,
- prom_store_with_metric_engine: bool,
+ with_metric_engine: bool,
schema_alterer: PendingRowsSchemaAltererRef,
- flush_interval: Duration,
- max_batch_rows: usize,
- max_concurrent_flushes: usize,
- worker_channel_capacity: usize,
- max_inflight_requests: usize,
- flow_notification_queue_capacity: NonZeroUsize,
- ) -> Option> {
- // Disable the batcher if flush is disabled or configuration is invalid.
- // Zero values for these knobs either cause panics (e.g., zero-capacity channels)
- // or deadlocks (e.g., semaphores with no permits).
- if flush_interval.is_zero()
- || max_batch_rows == 0
- || max_concurrent_flushes == 0
- || worker_channel_capacity == 0
- || max_inflight_requests == 0
- {
- return None;
- }
-
+ options: PendingRowsBatcherOptions,
+ batch_mode: PendingRowsBatchMode,
+ ) -> Arc {
let (shutdown, _) = broadcast::channel(1);
- let pending_rows_batch_sync = pending_rows_batch_sync_enabled();
let workers = Arc::new(DashMap::new());
PENDING_WORKERS.set(workers.len() as i64);
let (flow_notification_tx, flow_notification_rx) =
- mpsc::channel(flow_notification_queue_capacity.get());
+ mpsc::channel(options.tuning.flow_notification_queue_capacity.get());
start_flow_notification_worker(
flow_notification_rx,
table_flownode_set_cache,
node_manager.clone(),
);
- Some(Arc::new(Self {
+ Arc::new(Self {
workers,
- flush_interval,
- max_batch_rows,
- partition_manager,
- node_manager,
+ flush_interval: options.tuning.flush_interval,
+ max_batch_rows: options.tuning.max_batch_rows,
+ partition_provider: Arc::new(PartitionManagerPhysicalFlushAdapter {
+ partition_manager,
+ }),
+ node_requester: Arc::new(NodeManagerPhysicalFlushAdapter { node_manager }),
+ flush_catalog_provider: Arc::new(CatalogManagerPhysicalFlushAdapter {
+ catalog_manager: catalog_manager.clone(),
+ }),
catalog_manager,
flow_notification_tx,
- prom_store_with_metric_engine,
+ with_metric_engine,
schema_alterer,
- flush_semaphore: Arc::new(Semaphore::new(max_concurrent_flushes)),
- inflight_semaphore: Arc::new(Semaphore::new(max_inflight_requests)),
- worker_channel_capacity,
- pending_rows_batch_sync,
+ flush_semaphore: Arc::new(Semaphore::new(options.tuning.max_concurrent_flushes)),
+ inflight_semaphore: Arc::new(Semaphore::new(options.tuning.max_inflight_requests)),
+ worker_channel_capacity: options.tuning.worker_channel_capacity,
+ batch_mode,
shutdown,
- }))
+ })
}
- pub async fn submit(&self, requests: RowInsertRequests, ctx: QueryContextRef) -> Result {
+ pub async fn submit(
+ &self,
+ requests: RowInsertRequests,
+ ctx: QueryContextRef,
+ protocol: MetricRowBatchProtocol,
+ ) -> Result {
+ METRIC_ROW_BATCH_SUBMISSIONS
+ .with_label_values(&[protocol.as_str()])
+ .inc();
let (table_batches, total_rows) = {
let _timer = PENDING_ROWS_BATCH_INGEST_STAGE_ELAPSED
.with_label_values(&["submit_build_and_align"])
@@ -479,18 +700,23 @@ impl PendingRowsBatcher {
}
}
- if self.pending_rows_batch_sync {
- let result = {
+ if self.batch_mode.is_synchronous() {
+ let wait_timer = WaitFlushResultTimer::start(protocol);
+ let flush_result = {
let _timer = PENDING_ROWS_BATCH_INGEST_STAGE_ELAPSED
.with_label_values(&["submit_wait_flush_result"])
.start_timer();
response_rx
.await
- .map_err(|_| error::BatcherChannelClosedSnafu.build())?
+ .map_err(|_| error::BatcherChannelClosedSnafu.build())
+ .and_then(|waiter_result| waiter_result.context(error::SubmitBatchSnafu))
};
- result
- .context(error::SubmitBatchSnafu)
- .map(|()| total_rows as u64)
+ wait_timer.finish_with_result(if flush_result.is_ok() {
+ METRIC_SUCCESS_VALUE
+ } else {
+ METRIC_FAILURE_VALUE
+ });
+ flush_result.map(|()| total_rows as u64)
} else {
Ok(total_rows as u64)
}
@@ -633,7 +859,7 @@ impl PendingRowsBatcher {
let _timer = PENDING_ROWS_BATCH_INGEST_STAGE_ELAPSED
.with_label_values(&["align_build_create_table_schema"])
.start_timer();
- build_prom_create_table_schema_from_proto(rows_schema)?
+ build_metric_create_table_schema_from_proto(rows_schema)?
};
plan.tables_to_create
.push(((*table_name).to_string(), request_schema));
@@ -672,7 +898,7 @@ impl PendingRowsBatcher {
catalog,
schema,
&create_refs,
- self.prom_store_with_metric_engine,
+ self.with_metric_engine,
ctx.clone(),
)
.await?;
@@ -771,7 +997,7 @@ impl PendingRowsBatcher {
.with_label_values(&["align_batch_add_missing_columns"])
.start_timer();
self.schema_alterer
- .add_missing_prom_tag_columns_batch(catalog, schema, &alter_refs, ctx.clone())
+ .add_missing_metric_tag_columns_batch(catalog, schema, &alter_refs, ctx.clone())
.await?;
}
@@ -868,16 +1094,17 @@ impl PendingRowsBatcher {
.flush_interval
.checked_mul(WORKER_IDLE_TIMEOUT_MULTIPLIER)
.unwrap_or(self.flush_interval);
+ let shutdown_rx = self.shutdown.subscribe();
start_worker(
key,
worker.tx.clone(),
self.workers.clone(),
rx,
- self.shutdown.clone(),
- self.partition_manager.clone(),
- self.node_manager.clone(),
- self.catalog_manager.clone(),
+ shutdown_rx,
+ self.partition_provider.clone(),
+ self.node_requester.clone(),
+ self.flush_catalog_provider.clone(),
self.flow_notification_tx.clone(),
self.flush_interval,
worker_idle_timeout,
@@ -889,6 +1116,18 @@ impl PendingRowsBatcher {
}
}
+#[async_trait]
+impl MetricRowBatcher for PendingRowsBatcher {
+ async fn submit(
+ &self,
+ requests: RowInsertRequests,
+ ctx: QueryContextRef,
+ protocol: MetricRowBatchProtocol,
+ ) -> Result {
+ PendingRowsBatcher::submit(self, requests, ctx, protocol).await
+ }
+}
+
impl Drop for PendingRowsBatcher {
fn drop(&mut self) {
let _ = self.shutdown.send(());
@@ -930,10 +1169,10 @@ fn start_worker(
worker_tx: mpsc::Sender,
workers: Arc>,
mut rx: mpsc::Receiver,
- shutdown: broadcast::Sender<()>,
- partition_manager: PartitionRuleManagerRef,
- node_manager: NodeManagerRef,
- catalog_manager: CatalogManagerRef,
+ mut shutdown_rx: broadcast::Receiver<()>,
+ partition_provider: PhysicalFlushPartitionProviderRef,
+ node_requester: PhysicalFlushNodeRequesterRef,
+ catalog_provider: PhysicalFlushCatalogProviderRef,
flow_notification_tx: mpsc::Sender,
flush_interval: Duration,
worker_idle_timeout: Duration,
@@ -944,13 +1183,13 @@ fn start_worker(
let mut batch = None;
let flush_timer = tokio::time::sleep(flush_interval);
tokio::pin!(flush_timer);
- let mut shutdown_rx = shutdown.subscribe();
let idle_deadline = tokio::time::Instant::now() + worker_idle_timeout;
let idle_timer = tokio::time::sleep_until(idle_deadline);
tokio::pin!(idle_timer);
loop {
tokio::select! {
+ biased;
cmd = rx.recv() => {
match cmd {
Some(WorkerCommand::Submit { table_batches, total_rows, ctx, response_tx, _permit }) => {
@@ -981,9 +1220,9 @@ fn start_worker(
&& let Some(flush) = drain_batch(&mut batch) {
spawn_flush(
flush,
- partition_manager.clone(),
- node_manager.clone(),
- catalog_manager.clone(),
+ partition_provider.clone(),
+ node_requester.clone(),
+ catalog_provider.clone(),
flow_notification_tx.clone(),
flush_semaphore.clone(),
).await;
@@ -991,11 +1230,11 @@ fn start_worker(
}
None => {
if let Some(flush) = drain_batch(&mut batch) {
- flush_batch_with_managers(
+ flush_batch(
flush,
- partition_manager.clone(),
- node_manager.clone(),
- catalog_manager.clone(),
+ partition_provider.as_ref(),
+ node_requester.as_ref(),
+ catalog_provider.as_ref(),
flow_notification_tx.clone(),
).await;
}
@@ -1030,9 +1269,9 @@ fn start_worker(
if let Some(flush) = drain_batch(&mut batch) {
spawn_flush(
flush,
- partition_manager.clone(),
- node_manager.clone(),
- catalog_manager.clone(),
+ partition_provider.clone(),
+ node_requester.clone(),
+ catalog_provider.clone(),
flow_notification_tx.clone(),
flush_semaphore.clone(),
).await;
@@ -1040,11 +1279,11 @@ fn start_worker(
}
_ = shutdown_rx.recv() => {
if let Some(flush) = drain_batch(&mut batch) {
- flush_batch_with_managers(
+ flush_batch(
flush,
- partition_manager.clone(),
- node_manager.clone(),
- catalog_manager.clone(),
+ partition_provider.as_ref(),
+ node_requester.as_ref(),
+ catalog_provider.as_ref(),
flow_notification_tx.clone(),
).await;
}
@@ -1103,9 +1342,9 @@ fn drain_batch(batch: &mut Option) -> Option {
async fn spawn_flush(
flush: FlushBatch,
- partition_manager: PartitionRuleManagerRef,
- node_manager: NodeManagerRef,
- catalog_manager: CatalogManagerRef,
+ partition_provider: PhysicalFlushPartitionProviderRef,
+ node_requester: PhysicalFlushNodeRequesterRef,
+ catalog_provider: PhysicalFlushCatalogProviderRef,
flow_notification_tx: mpsc::Sender,
semaphore: Arc,
) {
@@ -1113,11 +1352,11 @@ async fn spawn_flush(
Ok(permit) => {
tokio::spawn(async move {
let _permit = permit;
- flush_batch_with_managers(
+ flush_batch(
flush,
- partition_manager,
- node_manager,
- catalog_manager,
+ partition_provider.as_ref(),
+ node_requester.as_ref(),
+ catalog_provider.as_ref(),
flow_notification_tx,
)
.await;
@@ -1125,11 +1364,11 @@ async fn spawn_flush(
}
Err(err) => {
warn!(err; "Flush semaphore closed, flushing inline");
- flush_batch_with_managers(
+ flush_batch(
flush,
- partition_manager,
- node_manager,
- catalog_manager,
+ partition_provider.as_ref(),
+ node_requester.as_ref(),
+ catalog_provider.as_ref(),
flow_notification_tx,
)
.await;
@@ -1317,28 +1556,6 @@ async fn flush_region_writes_concurrently(
Ok(affected_rows)
}
-async fn flush_batch_with_managers(
- flush: FlushBatch,
- partition_manager: PartitionRuleManagerRef,
- node_manager: NodeManagerRef,
- catalog_manager: CatalogManagerRef,
- flow_notification_tx: mpsc::Sender,
-) {
- let partition_provider = PartitionManagerPhysicalFlushAdapter { partition_manager };
- let node_requester = NodeManagerPhysicalFlushAdapter {
- node_manager: node_manager.clone(),
- };
- let catalog_provider = CatalogManagerPhysicalFlushAdapter { catalog_manager };
- flush_batch(
- flush,
- &partition_provider,
- &node_requester,
- &catalog_provider,
- flow_notification_tx,
- )
- .await;
-}
-
async fn flush_batch(
flush: FlushBatch,
partition_manager: &(impl PhysicalFlushPartitionProvider + ?Sized),
@@ -1876,6 +2093,7 @@ mod tests {
use std::sync::{Arc, Mutex};
use std::time::Duration;
+ use api::helper::ColumnDataTypeWrapper;
use api::region::RegionResponse;
use api::v1::flow::{DirtyWindowRequests, FlowRequest, FlowResponse};
use api::v1::meta::Peer;
@@ -1883,7 +2101,7 @@ mod tests {
use api::v1::value::ValueData;
use api::v1::{
ColumnDataType, ColumnSchema, Row, RowInsertRequest, RowInsertRequests, Rows, SemanticType,
- Value,
+ StructValue, Value,
};
use arrow::array::{BinaryArray, BooleanArray, StringArray, TimestampMillisecondArray};
use arrow::datatypes::{DataType as ArrowDataType, Field, Schema as ArrowSchema};
@@ -1907,6 +2125,7 @@ mod tests {
BatchPutRequest, BatchPutResponse, DeleteRangeRequest, DeleteRangeResponse, PutRequest,
PutResponse, RangeRequest, RangeResponse,
};
+ use common_query::native_histogram::native_histogram_column_schema;
use common_query::request::QueryRequest;
use common_recordbatch::SendableRecordBatchStream;
use dashmap::DashMap;
@@ -1920,24 +2139,30 @@ mod tests {
use snafu::ResultExt;
use store_api::storage::RegionId;
use table::metadata::TableId;
+ use table::test_util::EmptyTable;
use table::test_util::table_info::test_table_info;
use tokio::sync::{Notify, Semaphore, broadcast, mpsc, oneshot};
use tokio::time::{advance, sleep};
use super::{
- BatchKey, Error, FlushBatch, FlushRegionWrite, FlushWaiter, PendingBatch,
- PendingRowsBatcher, PendingWorker, PhysicalFlushCatalogProvider,
- PhysicalFlushNodeRequester, PhysicalFlushPartitionProvider, PhysicalTableMetadata,
- PlannedRegionBatch, RecordBatchWithTsIdx, ResolvedRegionBatch, TableBatch, WorkerCommand,
- columns_taxonomy, drain_batch, encode_region_write_requests, extract_timestamps,
- flush_batch, flush_batch_physical, flush_region_writes_concurrently, greptime_timestamp,
- notify_flow_dirty_windows_after_flush, plan_region_batches, remove_worker_if_same_channel,
- should_close_worker_on_idle_timeout, should_dispatch_concurrently,
- start_flow_notification_worker, start_worker, strip_partition_columns_from_batch,
- transform_logical_batches_to_physical, try_enqueue_flow_notification,
+ BatchKey, Error, FlushBatch, FlushRegionWrite, FlushWaiter, MetricRowBatchProtocol,
+ PendingBatch, PendingRowsBatchMode, PendingRowsBatcher, PendingRowsSchemaAlterer,
+ PendingWorker, PhysicalFlushCatalogProvider, PhysicalFlushNodeRequester,
+ PhysicalFlushPartitionProvider, PhysicalTableMetadata, PlannedRegionBatch,
+ RecordBatchWithTsIdx, ResolvedRegionBatch, TableBatch, WorkerCommand, columns_taxonomy,
+ drain_batch, encode_region_write_requests, extract_timestamps, flush_batch,
+ flush_batch_physical, flush_region_writes_concurrently, greptime_timestamp,
+ is_scalar_metric_batchable, notify_flow_dirty_windows_after_flush, plan_region_batches,
+ remove_worker_if_same_channel, should_close_worker_on_idle_timeout,
+ should_dispatch_concurrently, start_flow_notification_worker, start_worker,
+ strip_partition_columns_from_batch, transform_logical_batches_to_physical,
+ try_enqueue_flow_notification,
};
use crate::error;
- use crate::metrics::FLOW_NOTIFICATION_DROPPED;
+ use crate::metrics::{
+ FLOW_NOTIFICATION_DROPPED, METRIC_FAILURE_VALUE, METRIC_ROW_BATCH_SUBMISSIONS,
+ METRIC_ROW_BATCH_WAIT_FLUSH_RESULT_ELAPSED, METRIC_SUCCESS_VALUE,
+ };
use crate::prom_row_builder::rows_to_aligned_record_batch;
fn mock_rows(row_count: usize, schema_name: &str) -> Rows {
@@ -1950,6 +2175,220 @@ mod tests {
}
}
+ fn scalar_insert(table_name: &str) -> RowInsertRequest {
+ RowInsertRequest {
+ table_name: table_name.to_string(),
+ rows: Some(Rows {
+ schema: vec![
+ ColumnSchema {
+ column_name: "ts".to_string(),
+ datatype: ColumnDataType::TimestampMillisecond as i32,
+ semantic_type: SemanticType::Timestamp as i32,
+ ..Default::default()
+ },
+ ColumnSchema {
+ column_name: "value".to_string(),
+ datatype: ColumnDataType::Float64 as i32,
+ semantic_type: SemanticType::Field as i32,
+ ..Default::default()
+ },
+ ColumnSchema {
+ column_name: "host".to_string(),
+ datatype: ColumnDataType::String as i32,
+ semantic_type: SemanticType::Tag as i32,
+ ..Default::default()
+ },
+ ],
+ rows: vec![Row {
+ values: vec![
+ Value {
+ value_data: Some(ValueData::TimestampMillisecondValue(1000)),
+ },
+ Value {
+ value_data: Some(ValueData::F64Value(1.0)),
+ },
+ Value {
+ value_data: Some(ValueData::StringValue("host-1".to_string())),
+ },
+ ],
+ }],
+ }),
+ }
+ }
+
+ fn scalar_request() -> RowInsertRequests {
+ RowInsertRequests {
+ inserts: vec![scalar_insert("cpu")],
+ }
+ }
+
+ #[test]
+ fn test_scalar_metric_batchability_accepts_ordinary_table() {
+ assert!(is_scalar_metric_batchable(&scalar_request()).unwrap());
+ }
+
+ #[test]
+ fn test_scalar_metric_batchability_accepts_multiple_logical_tables() {
+ let requests = RowInsertRequests {
+ inserts: vec![scalar_insert("cpu"), scalar_insert("memory")],
+ };
+
+ assert!(is_scalar_metric_batchable(&requests).unwrap());
+ }
+
+ #[test]
+ fn test_scalar_metric_batchability_rejects_native_histogram() {
+ let mut requests = scalar_request();
+ let rows = requests.inserts[0].rows.as_mut().unwrap();
+ rows.schema[1] = native_histogram_column_schema().unwrap();
+ rows.rows[0].values[1].value_data = Some(ValueData::StructValue(StructValue::default()));
+
+ assert!(!is_scalar_metric_batchable(&requests).unwrap());
+ }
+
+ #[test]
+ fn test_scalar_metric_batchability_rejects_datatype_extension() {
+ let mut requests = scalar_request();
+ requests.inserts[0].rows.as_mut().unwrap().schema[1].datatype_extension =
+ Some(Default::default());
+
+ assert!(!is_scalar_metric_batchable(&requests).unwrap());
+ }
+
+ #[test]
+ fn test_scalar_metric_batchability_rejects_multiple_fields() {
+ let mut requests = scalar_request();
+ let rows = requests.inserts[0].rows.as_mut().unwrap();
+ rows.schema.push(ColumnSchema {
+ column_name: "another_value".to_string(),
+ datatype: ColumnDataType::Float64 as i32,
+ semantic_type: SemanticType::Field as i32,
+ ..Default::default()
+ });
+ rows.rows[0].values.push(Value {
+ value_data: Some(ValueData::F64Value(2.0)),
+ });
+
+ assert!(!is_scalar_metric_batchable(&requests).unwrap());
+ }
+
+ #[test]
+ fn test_scalar_metric_batchability_rejects_non_string_tag() {
+ let mut requests = scalar_request();
+ let rows = requests.inserts[0].rows.as_mut().unwrap();
+ rows.schema[2].datatype = ColumnDataType::Int64 as i32;
+ rows.rows[0].values[2].value_data = Some(ValueData::I64Value(1));
+
+ assert!(!is_scalar_metric_batchable(&requests).unwrap());
+ }
+
+ #[test]
+ fn test_scalar_metric_batchability_falls_back_for_vector_binary_value() {
+ let mut requests = scalar_request();
+ let rows = requests.inserts[0].rows.as_mut().unwrap();
+ let (datatype, datatype_extension) = ColumnDataTypeWrapper::vector_datatype(3).into_parts();
+ rows.schema[2].datatype = datatype as i32;
+ rows.schema[2].datatype_extension = datatype_extension;
+ rows.rows[0].values[2].value_data = Some(ValueData::BinaryValue(vec![0; 12]));
+
+ assert!(!is_scalar_metric_batchable(&requests).unwrap());
+ }
+
+ #[test]
+ fn test_scalar_metric_batchability_checks_malformed_scalar_after_unsupported_insert() {
+ let mut vector = scalar_insert("vector");
+ let vector_rows = vector.rows.as_mut().unwrap();
+ let (datatype, datatype_extension) = ColumnDataTypeWrapper::vector_datatype(3).into_parts();
+ vector_rows.schema[2].datatype = datatype as i32;
+ vector_rows.schema[2].datatype_extension = datatype_extension;
+ vector_rows.rows[0].values[2].value_data = Some(ValueData::BinaryValue(vec![0; 12]));
+
+ let mut malformed = scalar_insert("malformed");
+ malformed.rows.as_mut().unwrap().rows[0].values.pop();
+ let requests = RowInsertRequests {
+ inserts: vec![vector, malformed],
+ };
+
+ assert!(is_scalar_metric_batchable(&requests).is_err());
+ }
+
+ #[test]
+ fn test_scalar_metric_batchability_rejects_non_millisecond_timestamp() {
+ let mut requests = scalar_request();
+ let rows = requests.inserts[0].rows.as_mut().unwrap();
+ rows.schema[0].datatype = ColumnDataType::TimestampNanosecond as i32;
+ rows.rows[0].values[0].value_data = Some(ValueData::TimestampNanosecondValue(1_000_000));
+
+ assert!(!is_scalar_metric_batchable(&requests).unwrap());
+ }
+
+ #[test]
+ fn test_scalar_metric_batchability_rejects_missing_timestamp_or_value() {
+ for missing_index in [0, 1] {
+ let mut requests = scalar_request();
+ let rows = requests.inserts[0].rows.as_mut().unwrap();
+ rows.schema.remove(missing_index);
+ rows.rows[0].values.remove(missing_index);
+
+ assert!(!is_scalar_metric_batchable(&requests).unwrap());
+ }
+ }
+
+ #[test]
+ fn test_scalar_metric_batchability_rejects_wrong_semantic_types() {
+ for (column_index, semantic_type) in [
+ (0, SemanticType::Tag),
+ (1, SemanticType::Tag),
+ (2, SemanticType::Field),
+ ] {
+ let mut requests = scalar_request();
+ requests.inserts[0].rows.as_mut().unwrap().schema[column_index].semantic_type =
+ semantic_type as i32;
+
+ assert!(!is_scalar_metric_batchable(&requests).unwrap());
+ }
+ }
+
+ #[test]
+ fn test_scalar_metric_batchability_errors_on_row_width_mismatch() {
+ let mut requests = scalar_request();
+ requests.inserts[0].rows.as_mut().unwrap().rows[0]
+ .values
+ .pop();
+
+ assert!(is_scalar_metric_batchable(&requests).is_err());
+ }
+
+ #[test]
+ fn test_scalar_metric_batchability_errors_on_value_variant_mismatch() {
+ let mut requests = scalar_request();
+ requests.inserts[0].rows.as_mut().unwrap().rows[0].values[2].value_data =
+ Some(ValueData::BoolValue(true));
+
+ assert!(is_scalar_metric_batchable(&requests).is_err());
+ }
+
+ #[test]
+ fn test_scalar_metric_batchability_accepts_null_cell() {
+ let mut requests = scalar_request();
+ requests.inserts[0].rows.as_mut().unwrap().rows[0].values[1].value_data = None;
+
+ assert!(is_scalar_metric_batchable(&requests).unwrap());
+ }
+
+ #[test]
+ fn test_scalar_metric_batchability_treats_empty_requests_as_not_batchable() {
+ assert!(!is_scalar_metric_batchable(&RowInsertRequests::default()).unwrap());
+
+ let requests = RowInsertRequests {
+ inserts: vec![RowInsertRequest {
+ table_name: "empty".to_string(),
+ rows: Some(Rows::default()),
+ }],
+ };
+ assert!(!is_scalar_metric_batchable(&requests).unwrap());
+ }
+
fn mock_tag_batch(tag_name: &str, tag_value: &str, ts: i64, val: f64) -> RecordBatch {
let schema = Arc::new(ArrowSchema::new(vec![
Field::new(
@@ -2344,6 +2783,26 @@ mod tests {
}
}
+ struct BlockingFlushNodeRequester {
+ write_started_tx: Mutex>>,
+ write_release: Arc,
+ }
+
+ #[async_trait]
+ impl PhysicalFlushNodeRequester for BlockingFlushNodeRequester {
+ async fn handle(
+ &self,
+ _peer: &Peer,
+ _request: RegionRequest,
+ ) -> error::Result {
+ if let Some(write_started_tx) = self.write_started_tx.lock().unwrap().take() {
+ let _ = write_started_tx.send(());
+ }
+ self.write_release.notified().await;
+ Ok(RegionResponse::new(0))
+ }
+ }
+
#[test]
fn test_collect_non_empty_table_rows_filters_empty_payloads() {
let requests = RowInsertRequests {
@@ -2404,6 +2863,328 @@ mod tests {
assert_eq!(ctx.current_catalog(), flush.ctx.current_catalog());
}
+ #[tokio::test]
+ async fn test_immediate_final_owner_drop_drains_worker_once_and_notifies_all_waiters() {
+ let ctx = session::context::QueryContext::arc();
+ let (first_response_tx, first_response_rx) = oneshot::channel();
+ let (second_response_tx, second_response_rx) = oneshot::channel();
+ let semaphore = Arc::new(Semaphore::new(2));
+ let first_permit = semaphore.clone().try_acquire_owned().unwrap();
+ let second_permit = semaphore.clone().try_acquire_owned().unwrap();
+ let writes = Arc::new(AtomicUsize::new(0));
+ let (flow_notification_tx, _flow_notification_rx) = mpsc::channel(1);
+ let (shutdown, _) = tokio::sync::broadcast::channel(1);
+ let batcher = Arc::new(PendingRowsBatcher {
+ workers: Arc::new(DashMap::new()),
+ flush_interval: Duration::from_secs(60),
+ max_batch_rows: 100,
+ partition_provider: Arc::new(MockFlushPartitionProvider {
+ partition_rule_calls: Arc::new(AtomicUsize::new(0)),
+ region_leader_calls: Arc::new(AtomicUsize::new(0)),
+ }),
+ node_requester: Arc::new(MockFlushNodeRequester {
+ writes: writes.clone(),
+ fail: false,
+ }),
+ catalog_manager: MemoryCatalogManager::with_default_setup(),
+ flush_catalog_provider: Arc::new(MockFlushCatalogProvider {
+ table: Some(mock_physical_table_metadata(1024)),
+ }),
+ flow_notification_tx,
+ flush_semaphore: Arc::new(Semaphore::new(1)),
+ inflight_semaphore: semaphore,
+ worker_channel_capacity: 2,
+ with_metric_engine: true,
+ schema_alterer: Arc::new(NoopSchemaAlterer),
+ batch_mode: PendingRowsBatchMode::Synchronous,
+ shutdown,
+ });
+ let worker = batcher.get_or_spawn_worker(BatchKey {
+ catalog: "greptime".to_string(),
+ schema: "public".to_string(),
+ physical_table: "phy".to_string(),
+ });
+ for (timestamp, response_tx, permit) in [
+ (1000, first_response_tx, first_permit),
+ (2000, second_response_tx, second_permit),
+ ] {
+ worker
+ .tx
+ .try_send(WorkerCommand::Submit {
+ table_batches: vec![(
+ "cpu".to_string(),
+ 42,
+ mock_aligned_tag_batch("tag1", "host-1", timestamp, 1.0),
+ )],
+ total_rows: 1,
+ ctx: ctx.clone(),
+ response_tx,
+ _permit: permit,
+ })
+ .unwrap();
+ }
+
+ let consumers = [
+ batcher.clone(),
+ batcher.clone(),
+ batcher.clone(),
+ batcher.clone(),
+ ];
+ drop(consumers);
+ assert_eq!(1, Arc::strong_count(&batcher));
+ assert_eq!(0, writes.load(Ordering::SeqCst));
+
+ drop(batcher);
+
+ let (first_result, second_result) = tokio::time::timeout(Duration::from_secs(5), async {
+ tokio::join!(first_response_rx, second_response_rx)
+ })
+ .await
+ .unwrap();
+ assert!(first_result.unwrap().is_ok());
+ assert!(second_result.unwrap().is_ok());
+ assert_eq!(1, writes.load(Ordering::SeqCst));
+ }
+
+ struct NoopSchemaAlterer;
+
+ #[async_trait]
+ impl PendingRowsSchemaAlterer for NoopSchemaAlterer {
+ async fn create_tables_if_missing_batch(
+ &self,
+ _catalog: &str,
+ _schema: &str,
+ _tables: &[(&str, &[ColumnSchema])],
+ _with_metric_engine: bool,
+ _ctx: session::context::QueryContextRef,
+ ) -> error::Result<()> {
+ Ok(())
+ }
+
+ async fn add_missing_metric_tag_columns_batch(
+ &self,
+ _catalog: &str,
+ _schema: &str,
+ _tables: &[(&str, &[String])],
+ _ctx: session::context::QueryContextRef,
+ ) -> error::Result<()> {
+ Ok(())
+ }
+ }
+
+ fn synchronous_submit_request() -> RowInsertRequests {
+ let mut requests = scalar_request();
+ requests.inserts[0].rows.as_mut().unwrap().schema[2].column_name = "tag1".to_string();
+ requests
+ }
+
+ fn test_synchronous_submit_batcher(
+ node_requester: Arc,
+ ) -> Arc {
+ let schema = Arc::new(
+ DtSchema::try_new(vec![
+ DtColumnSchema::new(
+ "ts",
+ datatypes::prelude::ConcreteDataType::timestamp_millisecond_datatype(),
+ false,
+ )
+ .with_time_index(true),
+ DtColumnSchema::new(
+ "value",
+ datatypes::prelude::ConcreteDataType::float64_datatype(),
+ true,
+ ),
+ DtColumnSchema::new(
+ "tag1",
+ datatypes::prelude::ConcreteDataType::string_datatype(),
+ true,
+ ),
+ ])
+ .unwrap(),
+ );
+ let table_info = test_table_info(42, "cpu", "public", "greptime", schema);
+ let catalog_manager =
+ MemoryCatalogManager::new_with_table(EmptyTable::from_table_info(&table_info));
+ let (flow_notification_tx, _flow_notification_rx) = mpsc::channel(1);
+ let (shutdown, _) = tokio::sync::broadcast::channel(1);
+
+ Arc::new(PendingRowsBatcher {
+ workers: Arc::new(DashMap::new()),
+ flush_interval: Duration::from_secs(60),
+ max_batch_rows: 1,
+ partition_provider: Arc::new(MockFlushPartitionProvider {
+ partition_rule_calls: Arc::new(AtomicUsize::new(0)),
+ region_leader_calls: Arc::new(AtomicUsize::new(0)),
+ }),
+ node_requester,
+ catalog_manager,
+ flush_catalog_provider: Arc::new(MockFlushCatalogProvider {
+ table: Some(mock_physical_table_metadata(1024)),
+ }),
+ flow_notification_tx,
+ flush_semaphore: Arc::new(Semaphore::new(1)),
+ inflight_semaphore: Arc::new(Semaphore::new(1)),
+ worker_channel_capacity: 1,
+ with_metric_engine: true,
+ schema_alterer: Arc::new(NoopSchemaAlterer),
+ batch_mode: PendingRowsBatchMode::Synchronous,
+ shutdown,
+ })
+ }
+
+ fn test_successful_synchronous_submit_batcher() -> (Arc, Arc) {
+ let writes = Arc::new(AtomicUsize::new(0));
+ let node_requester = Arc::new(MockFlushNodeRequester {
+ writes: writes.clone(),
+ fail: false,
+ });
+ (test_synchronous_submit_batcher(node_requester), writes)
+ }
+
+ fn test_failing_synchronous_submit_batcher() -> (Arc, Arc) {
+ let writes = Arc::new(AtomicUsize::new(0));
+ let node_requester = Arc::new(MockFlushNodeRequester {
+ writes: writes.clone(),
+ fail: true,
+ });
+ (test_synchronous_submit_batcher(node_requester), writes)
+ }
+
+ static METRIC_ASSERTION_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+
+ #[tokio::test]
+ async fn test_synchronous_submit_records_failed_flush_by_protocol() {
+ let _guard = METRIC_ASSERTION_LOCK.lock().await;
+
+ for protocol in [
+ MetricRowBatchProtocol::Prometheus,
+ MetricRowBatchProtocol::Otlp,
+ ] {
+ let failed_wait = METRIC_ROW_BATCH_WAIT_FLUSH_RESULT_ELAPSED
+ .with_label_values(&[protocol.as_str(), METRIC_FAILURE_VALUE]);
+ let initial_failed_waits = failed_wait.get_sample_count();
+ let (batcher, writes) = test_failing_synchronous_submit_batcher();
+
+ assert!(
+ batcher
+ .submit(
+ synchronous_submit_request(),
+ session::context::QueryContext::arc(),
+ protocol,
+ )
+ .await
+ .is_err()
+ );
+ assert_eq!(1, writes.load(Ordering::SeqCst));
+ assert_eq!(initial_failed_waits + 1, failed_wait.get_sample_count());
+ }
+ }
+
+ #[tokio::test]
+ async fn test_synchronous_submit_records_each_protocol_submission() {
+ let _guard = METRIC_ASSERTION_LOCK.lock().await;
+
+ for protocol in [
+ MetricRowBatchProtocol::Prometheus,
+ MetricRowBatchProtocol::Otlp,
+ ] {
+ let submissions = METRIC_ROW_BATCH_SUBMISSIONS.with_label_values(&[protocol.as_str()]);
+ let initial_submissions = submissions.get();
+ let (batcher, writes) = test_successful_synchronous_submit_batcher();
+
+ assert_eq!(
+ 1,
+ batcher
+ .submit(
+ synchronous_submit_request(),
+ session::context::QueryContext::arc(),
+ protocol,
+ )
+ .await
+ .unwrap()
+ );
+ assert_eq!(1, writes.load(Ordering::SeqCst));
+ assert_eq!(initial_submissions + 1, submissions.get());
+ }
+ }
+
+ #[tokio::test]
+ async fn test_synchronous_submit_records_successful_wait_by_protocol() {
+ let _guard = METRIC_ASSERTION_LOCK.lock().await;
+
+ for protocol in [
+ MetricRowBatchProtocol::Prometheus,
+ MetricRowBatchProtocol::Otlp,
+ ] {
+ let successful_wait = METRIC_ROW_BATCH_WAIT_FLUSH_RESULT_ELAPSED
+ .with_label_values(&[protocol.as_str(), METRIC_SUCCESS_VALUE]);
+ let initial_successful_waits = successful_wait.get_sample_count();
+ let (batcher, writes) = test_successful_synchronous_submit_batcher();
+
+ assert_eq!(
+ 1,
+ batcher
+ .submit(
+ synchronous_submit_request(),
+ session::context::QueryContext::arc(),
+ protocol,
+ )
+ .await
+ .unwrap()
+ );
+ assert_eq!(1, writes.load(Ordering::SeqCst));
+ assert_eq!(
+ initial_successful_waits + 1,
+ successful_wait.get_sample_count()
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn test_cancelled_synchronous_submit_records_only_cancelled_wait() {
+ let _guard = METRIC_ASSERTION_LOCK.lock().await;
+ let protocol = MetricRowBatchProtocol::Otlp;
+ let cancelled_wait = METRIC_ROW_BATCH_WAIT_FLUSH_RESULT_ELAPSED
+ .with_label_values(&[protocol.as_str(), "cancelled"]);
+ let successful_wait = METRIC_ROW_BATCH_WAIT_FLUSH_RESULT_ELAPSED
+ .with_label_values(&[protocol.as_str(), METRIC_SUCCESS_VALUE]);
+ let failed_wait = METRIC_ROW_BATCH_WAIT_FLUSH_RESULT_ELAPSED
+ .with_label_values(&[protocol.as_str(), METRIC_FAILURE_VALUE]);
+ let initial_cancelled_waits = cancelled_wait.get_sample_count();
+ let initial_successful_waits = successful_wait.get_sample_count();
+ let initial_failed_waits = failed_wait.get_sample_count();
+ let (write_started_tx, write_started_rx) = oneshot::channel();
+ let write_release = Arc::new(Notify::new());
+ let batcher = test_synchronous_submit_batcher(Arc::new(BlockingFlushNodeRequester {
+ write_started_tx: Mutex::new(Some(write_started_tx)),
+ write_release: write_release.clone(),
+ }));
+
+ let submit_task = tokio::spawn(async move {
+ batcher
+ .submit(
+ synchronous_submit_request(),
+ session::context::QueryContext::arc(),
+ protocol,
+ )
+ .await
+ });
+ tokio::time::timeout(Duration::from_secs(5), write_started_rx)
+ .await
+ .unwrap()
+ .unwrap();
+ submit_task.abort();
+ assert!(submit_task.await.unwrap_err().is_cancelled());
+
+ assert_eq!(
+ initial_cancelled_waits + 1,
+ cancelled_wait.get_sample_count()
+ );
+ assert_eq!(initial_successful_waits, successful_wait.get_sample_count());
+ assert_eq!(initial_failed_waits, failed_wait.get_sample_count());
+ write_release.notify_one();
+ }
+
#[test]
fn test_pending_batch_keeps_same_name_batches_with_distinct_table_ids() {
let ctx = session::context::QueryContext::arc();
diff --git a/src/servers/src/prom_row_builder.rs b/src/servers/src/prom_row_builder.rs
index aad12814683..eb620010da7 100644
--- a/src/servers/src/prom_row_builder.rs
+++ b/src/servers/src/prom_row_builder.rs
@@ -199,9 +199,9 @@ pub(crate) fn identify_missing_columns_from_proto(
Ok(missing)
}
-/// Build a `Vec` suitable for creating a new Prometheus logical table
+/// Build a `Vec` 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> {
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);
diff --git a/src/servers/src/query_handler.rs b/src/servers/src/query_handler.rs
index 35ec1cc2ff5..cbd27e66a7e 100644
--- a/src/servers/src/query_handler.rs
+++ b/src/servers/src/query_handler.rs
@@ -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;
pub type InfluxdbLineProtocolHandlerRef = Arc;
pub type PromStoreProtocolHandlerRef = Arc;
@@ -141,6 +142,7 @@ pub trait OpenTelemetryProtocolHandler: PipelineHandler {
async fn metrics(
&self,
request: ExportMetricsServiceRequest,
+ metric_row_batcher: Option,
ctx: QueryContextRef,
) -> Result;
diff --git a/tests-integration/src/instance.rs b/tests-integration/src/instance.rs
index 9f48d01a467..121abfd558d 100644
--- a/tests-integration/src/instance.rs
+++ b/tests-integration/src/instance.rs
@@ -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
diff --git a/tests-integration/src/otlp.rs b/tests-integration/src/otlp.rs
index 5d40a8bd7cb..6e10da3128a 100644
--- a/tests-integration/src/otlp.rs
+++ b/tests-integration/src/otlp.rs
@@ -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
diff --git a/tests-integration/src/test_util.rs b/tests-integration/src/test_util.rs
index 9d12848ce61..040d014487c 100644
--- a/tests-integration/src/test_util.rs
+++ b/tests-integration/src/test_util.rs
@@ -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();
diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs
index c696bd40ac4..551aa56c622 100644
--- a/tests-integration/tests/http.rs
+++ b/tests-integration/tests/http.rs
@@ -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,