rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-12 06:52:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf: cache OTLP trace service and operation writes (#9432)

Browse Source 
* perf: cache OTLP trace service and operation writes

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* chore: fix CR issues for tests

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix: bound asynchronous trace auxiliary writes

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* perf: bound trace auxiliary cache by estimated memory

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* chore: size trace auxiliary cache from host or pod memory

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* chore: add metrics for cache hit and miss

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix: test

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* chore: unify frontend cache metrics

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* test: merge and reduce

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* chore: add init for the metrics

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

---------

Signed-off-by: shuiyisong <xixing.sys@gmail.com>
This commit is contained in:
shuiyisong
2026-10-10 02:45:43 +00:00
committed by GitHub
parent 6e1a137411
commit a907cbfbe4
32 changed files with 1784 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+6
View File
@@ -2459,7 +2459,9 @@ dependencies = [
"common-macro",
"common-meta",
"greptime-proto",
"lazy_static",
"meta-client",
"prometheus 0.14.0",
"serde",
"serde_json",
"snafu 0.8.6",
@@ -5604,6 +5606,7 @@ dependencies = [
"lazy_static",
"log-query",
"meta-client",
"moka",
"opentelemetry-proto",
"opentelemetry-semantic-conventions 0.31.0",
"operator",
@@ -10860,6 +10863,7 @@ dependencies = [
"chrono-tz",
"common-catalog",
"common-error",
"common-frontend",
"common-macro",
"common-meta",
"common-query",
@@ -13522,6 +13526,7 @@ dependencies = [
"common-catalog",
"common-decimal",
"common-error",
"common-frontend",
"common-grpc",
"common-macro",
"common-mem-prof",
@@ -15122,6 +15127,7 @@ dependencies = [
"store-api",
"substrait 1.4.0-alpha.0",
"table",
"temp-env",
"tempfile",
"time",
"tokio",
config/config.md
+2
View File
@@ -95,6 +95,7 @@
| `otlp` | -- | -- | OpenTelemetry protocol options. |
| `otlp.enable` | Bool | `true` | Whether to enable OpenTelemetry protocol in HTTP API. |
| `otlp.trace_ingest_chunk_size` | Integer | `512` | Maximum spans per trace ingest chunk. Set to 0 to disable splitting. |
| `otlp.trace_aux_cache_size` | String | Auto | Estimated memory budget for cached trace service/operation keys per frontend,<br/>shared across all catalogs, schemas, and trace tables. Set to 0 to disable caching.<br/>Defaults to 1/128 of the host or pod memory limit, with a minimum of 32MiB.<br/>Uses 32MiB if memory detection is unavailable.<br/>Excludes cache and allocator overhead; shared table names are charged per entry. |
| `otlp.experimental_enable_resource_info` | Bool | `false` | Whether to synthesize the `greptime_otel_resource_info` table from OTLP metric<br/>resource attributes, so metrics-only services reach the semantic graph. |
| `prom_store` | -- | -- | Prometheus remote storage options |
| `prom_store.enable` | Bool | `true` | Whether to enable Prometheus remote write and read in HTTP API. |
@@ -368,6 +369,7 @@
| `otlp` | -- | -- | OpenTelemetry protocol options. |
| `otlp.enable` | Bool | `true` | Whether to enable OpenTelemetry protocol in HTTP API. |
| `otlp.trace_ingest_chunk_size` | Integer | `512` | Maximum spans per trace ingest chunk. Set to 0 to disable splitting. |
| `otlp.trace_aux_cache_size` | String | Auto | Estimated memory budget for cached trace service/operation keys per frontend,<br/>shared across all catalogs, schemas, and trace tables. Set to 0 to disable caching.<br/>Defaults to 1/128 of the host or pod memory limit, with a minimum of 32MiB.<br/>Uses 32MiB if memory detection is unavailable.<br/>Excludes cache and allocator overhead; shared table names are charged per entry. |
| `otlp.experimental_enable_resource_info` | Bool | `false` | Whether to synthesize the `greptime_otel_resource_info` table from OTLP metric<br/>resource attributes, so metrics-only services reach the semantic graph. |
| `prom_store` | -- | -- | Prometheus remote storage options |
| `prom_store.enable` | Bool | `true` | Whether to enable Prometheus remote write and read in HTTP API. |
config/frontend.example.toml
+7
View File
@@ -310,6 +310,13 @@ enable = true
enable = true
## Maximum spans per trace ingest chunk. Set to 0 to disable splitting.
trace_ingest_chunk_size = 512
## Estimated memory budget for cached trace service/operation keys per frontend,
## shared across all catalogs, schemas, and trace tables. Set to 0 to disable caching.
## Defaults to 1/128 of the host or pod memory limit, with a minimum of 32MiB.
## Uses 32MiB if memory detection is unavailable.
## Excludes cache and allocator overhead; shared table names are charged per entry.
## @toml2docs:none-default="Auto"
#+ trace_aux_cache_size = "32MiB"
## Whether to synthesize the `greptime_otel_resource_info` table from OTLP metric
## resource attributes, so metrics-only services reach the semantic graph.
experimental_enable_resource_info = false
config/standalone.example.toml
+7
View File
@@ -282,6 +282,13 @@ enable = true
enable = true
## Maximum spans per trace ingest chunk. Set to 0 to disable splitting.
trace_ingest_chunk_size = 512
## Estimated memory budget for cached trace service/operation keys per frontend,
## shared across all catalogs, schemas, and trace tables. Set to 0 to disable caching.
## Defaults to 1/128 of the host or pod memory limit, with a minimum of 32MiB.
## Uses 32MiB if memory detection is unavailable.
## Excludes cache and allocator overhead; shared table names are charged per entry.
## @toml2docs:none-default="Auto"
#+ trace_aux_cache_size = "32MiB"
## Whether to synthesize the `greptime_otel_resource_info` table from OTLP metric
## resource attributes, so metrics-only services reach the semantic graph.
experimental_enable_resource_info = false
src/cmd/tests/load_config_test.rs
+23
View File
@@ -731,6 +731,29 @@ fn test_example_configs_document_actual_defaults() {
assert!(errors.is_empty(), "{}", errors.join("\n"));
}
#[test]
fn test_load_trace_aux_cache_size() {
let default = FrontendOptions::default().otlp.trace_aux_cache_size;
for (setting, size) in [
("", default),
("trace_aux_cache_size = 0", ReadableSize(0)),
("trace_aux_cache_size = \"1MiB\"", ReadableSize::mb(1)),
] {
let config = tempfile::NamedTempFile::new().unwrap();
std::fs::write(config.path(), format!("[otlp]\n{setting}\n")).unwrap();
let frontend =
GreptimeOptions::<FrontendOptions>::load_layered_options(config.path().to_str(), "")
.unwrap();
let standalone =
GreptimeOptions::<StandaloneOptions>::load_layered_options(config.path().to_str(), "")
.unwrap();
for options in [frontend.component, standalone.component.frontend_options()] {
assert_eq!(options.otlp.trace_aux_cache_size, size);
}
}
}
#[test]
fn test_load_removed_histogram_options() {
for enabled in [false, true] {
src/common/frontend/Cargo.toml
+2
View File
@@ -13,7 +13,9 @@ common-grpc.workspace = true
common-macro.workspace = true
common-meta.workspace = true
greptime-proto.workspace = true
lazy_static.workspace = true
meta-client.workspace = true
prometheus.workspace = true
serde.workspace = true
serde_json.workspace = true
snafu.workspace = true
src/common/frontend/src/lib.rs
+1
View File
@@ -18,6 +18,7 @@ use std::str::FromStr;
use snafu::OptionExt;
pub mod error;
pub mod metrics;
pub mod selector;
pub mod slow_query_event;
src/common/frontend/src/metrics.rs
+117
View File
@@ -0,0 +1,117 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Cache lookup counters shared by frontend, pipeline, and protocol handlers.
//!
//! The `type` label identifies `otlp_trace_aux`, `otlp_metrics_legacy`,
//! `pipeline_table`, `pipeline`, `pipeline_original`, `pipeline_failover`, or
//! `mysql_prepared_stmt`. Counters aggregate across catalogs and connections.
//! Trace lookups count distinct service/operation keys once per request, excluding
//! rechecks; legacy metric lookups count each requested metric name. Pipeline
//! loaders count successful coalesced waiters as hits and failed loads as misses.
//! Failover counts one outcome after resolving all candidate schemas; an ambiguous
//! result is a miss. Table and prepared-statement caches count each lookup.
//! Both outcomes for all known caches are exported from the first metric update.
use lazy_static::lazy_static;
use prometheus::{IntCounterVec, register_int_counter_vec};
const CACHE_TYPES: [&str; 7] = [
"otlp_trace_aux",
"otlp_metrics_legacy",
"pipeline_table",
"pipeline",
"pipeline_original",
"pipeline_failover",
"mysql_prepared_stmt",
];
lazy_static! {
static ref CACHE_COUNTERS: (IntCounterVec, IntCounterVec) = {
let hits = register_int_counter_vec!(
"greptime_frontend_cache_hit",
"frontend cache hit",
&["type"]
)
.unwrap();
let misses = register_int_counter_vec!(
"greptime_frontend_cache_miss",
"frontend cache miss",
&["type"]
)
.unwrap();
// Register both outcomes together so an all-hit or all-miss cache has both series.
for cache_type in CACHE_TYPES {
hits.with_label_values(&[cache_type]);
misses.with_label_values(&[cache_type]);
}
(hits, misses)
};
/// Frontend cache hits by cache type.
pub static ref CACHE_HIT: IntCounterVec = CACHE_COUNTERS.0.clone();
/// Frontend cache misses by cache type.
pub static ref CACHE_MISS: IntCounterVec = CACHE_COUNTERS.1.clone();
}
/// Records one completed cache lookup.
pub fn record_cache_lookup(cache_type: &str, hit: bool) {
if hit {
CACHE_HIT.with_label_values(&[cache_type]).inc();
} else {
CACHE_MISS.with_label_values(&[cache_type]).inc();
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_cache_lookup_counters() {
record_cache_lookup("pipeline", false);
// Scrape before accessing CACHE_HIT, which would initialize it and hide a missing series.
let families = prometheus::gather();
for (name, expected) in [
("greptime_frontend_cache_hit", 0.0),
("greptime_frontend_cache_miss", 1.0),
] {
let family = families
.iter()
.find(|family| family.name() == name)
.unwrap();
assert_eq!(family.get_metric().len(), CACHE_TYPES.len());
for cache_type in CACHE_TYPES {
let metric = family
.get_metric()
.iter()
.find(|metric| metric.get_label()[0].value() == cache_type)
.unwrap();
assert_eq!(metric.get_label()[0].name(), "type");
assert_eq!(
metric.get_counter().value(),
if cache_type == "pipeline" {
expected
} else {
0.0
}
);
}
}
record_cache_lookup("pipeline", true);
record_cache_lookup("pipeline", false);
assert_eq!(CACHE_HIT.with_label_values(&["pipeline"]).get(), 1);
assert_eq!(CACHE_MISS.with_label_values(&["pipeline"]).get(), 2);
}
}
src/common/query/src/lib.rs
+37 -2
View File
@@ -25,10 +25,13 @@ pub mod stream;
pub mod test_util;
use std::fmt::{Debug, Display, Formatter};
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use api::greptime_proto::v1::AddColumnLocation as Location;
use api::greptime_proto::v1::add_column_location::LocationType;
use common_error::ext::BoxedError;
use common_recordbatch::{
RecordBatches, SendableRecordBatchMapper, SendableRecordBatchStream, map_dictionary_to_values,
map_dictionary_to_values_schema,
@@ -74,6 +77,30 @@ pub struct OutputMeta {
/// May exist for query output. One can retrieve execution metrics from this plan.
pub plan: Option<Arc<dyn ExecutionPlan>>,
pub cost: OutputCost,
/// Process-local storage results attached by asynchronous ordinary-table batching.
/// Callers needing confirmed writes must await these. Dropping them does not cancel writes.
pub write_completions: Vec<WriteCompletion>,
}
/// Actual storage completion of an already accepted asynchronous write.
pub struct WriteCompletion(Pin<Box<dyn Future<Output = Result<(), BoxedError>> + Send>>);
impl WriteCompletion {
/// Wraps the storage result without changing how the write is scheduled.
pub fn new(completion: impl Future<Output = Result<(), BoxedError>> + Send + 'static) -> Self {
Self(Box::pin(completion))
}
/// Waits for storage success or failure, independently of queue admission.
pub async fn wait(self) -> Result<(), BoxedError> {
self.0.await
}
}
impl Debug for WriteCompletion {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
f.debug_struct("WriteCompletion").finish_non_exhaustive()
}
}
impl Output {
@@ -163,18 +190,26 @@ impl Debug for OutputData {
impl OutputMeta {
pub fn new(plan: Option<Arc<dyn ExecutionPlan>>, cost: usize) -> Self {
Self { plan, cost }
Self {
plan,
cost,
..Default::default()
}
}
pub fn new_with_plan(plan: Arc<dyn ExecutionPlan>) -> Self {
Self {
plan: Some(plan),
cost: 0,
..Default::default()
}
}
pub fn new_with_cost(cost: usize) -> Self {
Self { plan: None, cost }
Self {
cost,
..Default::default()
}
}
}
src/frontend/AGENTS.md
+14
View File
@@ -34,6 +34,9 @@ remote datanodes via `operator`/`client`.
| `heartbeat` | `src/frontend/src/heartbeat.rs` | Heartbeat to metasrv; handles suspend / cache invalidation |
| `service_config` | `src/frontend/src/service_config/` | Per-protocol option structs |
Shared frontend cache counters live in `src/common/frontend/src/metrics.rs`,
including caches owned by pipeline and protocol handlers.
## Request lifecycles
- **SQL query** (`instance.rs`): `do_query_inner` handles parsing, interceptors,
@@ -58,6 +61,17 @@ remote datanodes via `operator`/`client`.
the lazy schema handshake; checks permissions and reconciles missing columns
through `Inserter` once per stream, then reuses the refreshed table.
- **OTLP trace lookup tables** (`instance/otlp/trace_ingest.rs`): `Instance` owns
a shared cache of confirmed service/operation writes, which bypass batching.
`otlp.trace_aux_cache_size` sets its estimated byte budget at startup; zero disables it.
Async main writes retain completion results; a frontend task writes auxiliary
rows only for confirmed chunks before populating the cache.
Cold async trace requests reserve a shared auxiliary slot before main writes,
bounded separately by the table batcher's `max_inflight_requests`; confirmed
cache hits bypass this admission. The slot lasts through auxiliary completion.
These tables are ingestion-managed; manual mutation requires restarting
serving frontends.
- **Logical-table batching** (`instance/logical_batcher.rs`): `Services` initializes
one shared batcher for opted-in HTTP Prom and nonlegacy OTLP metric-engine
writes. Downstream routers can initialize it when enabling replacement endpoints.
src/frontend/Cargo.toml
+1
View File
@@ -58,6 +58,7 @@ humantime-serde.workspace = true
lazy_static.workspace = true
log-query.workspace = true
meta-client.workspace = true
moka = { workspace = true, features = ["sync"] }
opentelemetry-proto.workspace = true
opentelemetry-semantic-conventions = { version = "0.31", features = ["semconv_experimental"] }
operator.workspace = true
src/frontend/src/instance.rs
+32 -2
View File
@@ -48,8 +48,10 @@ use catalog::process_manager::{
use client::OutputData;
use common_base::Plugins;
use common_base::cancellation::CancellableFuture;
use common_batcher::request_limiter::RequestLimiter;
use common_error::ext::{BoxedError, ErrorExt};
use common_event_recorder::EventRecorderRef;
use common_frontend::metrics::{CACHE_HIT, CACHE_MISS};
use common_meta::cache::TableFlownodeSetCacheRef;
use common_meta::cache_invalidator::CacheInvalidatorRef;
use common_meta::key::TableMetadataManagerRef;
@@ -141,6 +143,8 @@ pub struct Instance {
slow_query_options: SlowQueryOptions,
influxdb_default_merge_mode: InfluxdbMergeMode,
trace_ingest_chunk_size: usize,
trace_aux_cache: otlp::TraceAuxCache,
trace_aux_limiter: Option<RequestLimiter>,
otlp_resource_info: bool,
suspend: Arc<AtomicBool>,
@@ -148,7 +152,7 @@ pub struct Instance {
// first layer key: db-string
// key: direct input metric name
// value: if runs in legacy mode
otlp_metrics_table_legacy_cache: DashMap<String, DashMap<String, bool>>,
otlp_metrics_table_legacy_cache: Arc<DashMap<String, DashMap<String, bool>>>,
}
impl Instance {
@@ -570,6 +574,12 @@ fn fast_legacy_check(
.iter()
.filter_map(|name| cache.get(name))
.collect::<Vec<_>>();
CACHE_HIT
.with_label_values(&["otlp_metrics_legacy"])
.inc_by(hit_cache.len() as u64);
CACHE_MISS
.with_label_values(&["otlp_metrics_legacy"])
.inc_by((names.len() - hit_cache.len()) as u64);
if !hit_cache.is_empty() {
let hit_legacy = hit_cache.iter().any(|en| *en.value());
let hit_prom = hit_cache.iter().any(|en| !*en.value());
@@ -3418,12 +3428,16 @@ mod tests {
}
#[tokio::test]
async fn test_event_recorder_is_exposed() -> TestResult<()> {
async fn test_instance_shared_state() -> TestResult<()> {
let instance =
test_instance_with_tables(test_table(1024, "source")?, test_table(1025, "target")?)
.await?;
let _event_recorder = instance.event_recorder();
assert!(Arc::ptr_eq(
&instance.otlp_metrics_table_legacy_cache,
&instance.clone().otlp_metrics_table_legacy_cache,
));
Ok(())
}
@@ -3554,11 +3568,20 @@ mod tests {
})?;
let ctx = test_query_ctx(1);
let handler: PipelineHandlerRef = Arc::new(instance.clone());
let hits = CACHE_HIT.with_label_values(&["pipeline_table"]);
let misses = CACHE_MISS.with_label_values(&["pipeline_table"]);
let before = (hits.get(), misses.get());
handler
.get_pipeline("pipeline", None, ctx.clone())
.await
.unwrap();
assert_eq!((hits.get(), misses.get()), (before.0 + 1, before.1 + 1));
handler
.get_pipeline("pipeline", None, ctx.clone())
.await
.unwrap();
assert_eq!((hits.get(), misses.get()), (before.0 + 3, before.1 + 1));
assert_permission_denied(
PipelineHandler::get_pipeline_str(&instance, "pipeline", None, ctx.clone()).await,
);
@@ -3877,11 +3900,15 @@ mod tests {
#[test]
fn test_fast_legacy_check_is_read_only() {
let hits = CACHE_HIT.with_label_values(&["otlp_metrics_legacy"]);
let misses = CACHE_MISS.with_label_values(&["otlp_metrics_legacy"]);
let before = (hits.get(), misses.get());
let cache = DashMap::new();
cache.insert("metric1".to_string(), true);
let names = vec!["metric1".to_string(), "metric2".to_string()];
assert_eq!(Some(true), fast_legacy_check(&cache, &names).unwrap());
assert_eq!((hits.get(), misses.get()), (before.0 + 1, before.1 + 1));
assert!(!cache.contains_key("metric2"));
cache_legacy_mode(&cache, &names, true).unwrap();
@@ -3893,6 +3920,9 @@ mod tests {
cache_incompatible.insert("metric1".to_string(), true);
cache_incompatible.insert("metric2".to_string(), false);
assert!(fast_legacy_check(&cache_incompatible, &names).is_err());
assert_eq!((hits.get(), misses.get()), (before.0 + 3, before.1 + 1));
assert_eq!(None, fast_legacy_check(&cache, &[]).unwrap());
assert_eq!((hits.get(), misses.get()), (before.0 + 3, before.1 + 1));
}
#[test]
src/frontend/src/instance/builder.rs
+15 -2
View File
@@ -22,6 +22,7 @@ use catalog::kvbackend::KvBackendCatalogManager;
use catalog::process_manager::ProcessManagerRef;
use catalog::system_schema::semantic_graph::EntityGraphProviderRef;
use common_base::Plugins;
use common_batcher::request_limiter::RequestLimiter;
use common_datasource::object_store::LocalFileAccess;
use common_event_recorder::{EventRecorderImpl, EventRecorderRef};
use common_meta::cache::{LayeredCacheRegistryRef, TableFlownodeSetCacheRef, TableRouteCacheRef};
@@ -50,8 +51,8 @@ use partition::manager::PartitionRuleManager;
use pipeline::pipeline_operator::PipelineOperator;
use query::QueryEngineFactory;
use query::region_query::RegionQueryHandlerFactoryRef;
use servers::batcher::BatchingProtocol;
use servers::batcher::table::TablePendingRowsBatcher;
use servers::batcher::{BatchingProtocol, pending_rows_batch_sync_enabled};
use snafu::{OptionExt, ResultExt};
use crate::error::{self, DataFusionSnafu, ExternalSnafu, Result};
@@ -60,6 +61,7 @@ use crate::frontend::FrontendOptions;
use crate::heartbeat::frontend_peer_addr;
use crate::instance::Instance;
use crate::instance::entity_graph::EntityGraphProviderImpl;
use crate::instance::otlp::TraceAuxCache;
use crate::instance::region_query::FrontendRegionQueryHandler;
use crate::service_config::BatcherOptions;
@@ -385,6 +387,15 @@ impl FrontendBuilder {
));
admin_event_recorder.install(&event_recorder);
let batcher_options = self.options.table_batcher_options();
let trace_aux_limiter = if batcher_options.pending_rows_batching_enabled()
&& !pending_rows_batch_sync_enabled()
{
RequestLimiter::try_new(batcher_options.max_inflight_requests)
} else {
None
};
Ok(Instance {
logical_batcher: Default::default(),
frontend_peer_addr,
@@ -400,10 +411,12 @@ impl FrontendBuilder {
event_recorder,
slow_query_recorder,
process_manager,
otlp_metrics_table_legacy_cache: DashMap::new(),
otlp_metrics_table_legacy_cache: Arc::new(DashMap::new()),
slow_query_options: self.options.slow_query.clone(),
influxdb_default_merge_mode: self.options.influxdb.default_merge_mode,
trace_ingest_chunk_size: self.options.otlp.trace_ingest_chunk_size,
trace_aux_cache: TraceAuxCache::new(self.options.otlp.trace_aux_cache_size.as_bytes()),
trace_aux_limiter,
otlp_resource_info: self.options.otlp.experimental_enable_resource_info,
suspend: Arc::new(AtomicBool::new(false)),
})
src/frontend/src/instance/otlp.rs
+1
View File
@@ -52,6 +52,7 @@ use table::requests::{
SOURCE_OPENTELEMETRY,
};
pub(crate) use self::trace_ingest::TraceAuxCache;
use self::trace_ingest::trace_conventions;
use crate::instance::Instance;
use crate::metrics::{OTLP_LOGS_ROWS, OTLP_METRICS_ROWS, OTLP_RESOURCE_INFO_WRITE_ERRORS};
src/frontend/src/instance/otlp/README.md
+52
View File
@@ -175,6 +175,58 @@ Accounting follows the main-table write:
accepted or rejected span counts; and
- failure details are bounded before they are folded into `TraceIngestOutcome`.
A cache shared by each frontend's `Instance` clones skips previously confirmed
auxiliary writes for v0, v1, and v2. `otlp.trace_aux_cache_size` sets an estimated
memory budget across all catalogs, schemas, and trace tables on each frontend.
By default, the budget is 1/128 of the pod's cgroup memory limit or total host
memory, with a minimum of `32MiB`. If memory detection is unavailable, it uses
`32MiB`. An explicit size overrides this default; `0` disables caching.
The estimate includes key structs and allocated string capacity, charging shared
table names per entry. It excludes cache and allocator overhead, so it is not a
strict process-memory limit. Changes take effect on frontend or standalone
restart. There is no time-based expiry.
Services and operations are cached independently. Only a successful
auxiliary write populates the cache. Auxiliary writes always bypass batching,
including when main spans use asynchronous batching. Failures, eviction,
frontend restarts, and concurrent misses can cause repeat writes. Write cost
reflects only the writes actually performed.
`greptime_frontend_cache_hit{type="otlp_trace_aux"}` and
`greptime_frontend_cache_miss{type="otlp_trace_aux"}` count distinct service and
operation keys at the first cache check in each request.
Requests with auxiliary admission count before acquiring a slot; other requests
count when preparing auxiliary writes. Admission and write rechecks do not count.
When the cache is disabled, all checked keys count as misses.
Successful auxiliary writes with `skip_wal` enabled are cached too. If a
datanode crashes before flushing those rows while the frontend survives, the
cache can suppress their recreation, even by later WAL-enabled requests, until
eviction or frontend restart. Missing service/operation discovery rows in this
case are an accepted limitation; the cache does not guarantee durability beyond
the write's WAL policy.
With asynchronous main-table batching, the response still acknowledges queue
admission. A frontend task waits for the storage results and writes auxiliary
rows only for successful chunks. Failed chunks do not populate the auxiliary
tables or cache. Deferred auxiliary failures are logged; they cannot change an
already returned response. Deferred writes retain the original request's row
admission, and their write cost is not included in the early response.
Async trace requests with cache misses acquire one shared auxiliary-work slot
before submitting main chunks. Each frontend limits these requests separately
from main-table batching, using `pending_rows_batcher.max_inflight_requests`.
The slot is held through auxiliary completion, including failures. Requests
whose service/operation entries are all cached bypass this admission and need
no deferred task. Cache lookups and updates hold no lock across storage writes.
The auxiliary tables are ingestion-managed. When caching is enabled, manual
`DROP`, `TRUNCATE`, `DELETE`, or other mutations require clearing the affected
caches or restarting all serving frontends before relying on ingestion to
populate the lookup rows again.
There is no automatic invalidation or public cache-clear command. In v0, fewer
duplicate rows are appended; distinct service/operation lookup results remain
the same.
A rejection detail carries the failing cause, not just a status code, so an
unusable attribute value names its column, source value, and target type. Repeats
of the same `(site, cause)` collapse into one entry with an occurrence count, and
src/frontend/src/instance/otlp/trace_ingest.rs
+335 -11
View File
@@ -13,6 +13,7 @@
// limitations under the License.
use std::collections::{BTreeMap, HashMap, HashSet};
use std::mem;
use std::sync::Arc;
use api::helper::ColumnDataTypeWrapper;
@@ -22,13 +23,17 @@ use api::v1::{
RowInsertRequest, RowInsertRequests, Rows, Value,
};
use client::Output;
use common_batcher::request_limiter::RequestLimiter;
use common_error::ext::{BoxedError, ErrorExt};
use common_error::status_code::StatusCode;
use common_meta::rpc::ddl::TriggerReason;
use common_query::WriteCompletion;
use common_telemetry::{debug, warn};
use datatypes::prelude::ConcreteDataType;
use moka::sync::Cache;
use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
use pipeline::PipelineWay;
use prometheus::IntCounter;
use servers::error::{self, Result as ServerResult};
use servers::otlp;
use servers::otlp::coerce::{coerce_value_data, is_supported_trace_coercion, trace_value_datatype};
@@ -43,6 +48,8 @@ use table::requests::{
SEMANTIC_TRACE_CONVENTIONS, SEMANTIC_VALUE_MIXED, SEMANTIC_VALUE_UNKNOWN, SIGNAL_TYPE_TRACE,
SOURCE_OPENTELEMETRY, TABLE_DATA_MODEL_TRACE_V1, TABLE_DATA_MODEL_TRACE_V2,
};
use table::table_name::TableName;
use tokio::sync::OwnedSemaphorePermit;
use crate::instance::Instance;
use crate::instance::otlp::trace_semconv::trace_semconv_fixed_type;
@@ -52,7 +59,131 @@ use crate::instance::otlp::trace_types::{
is_trace_reconcile_candidate_type, prepare_trace_column_rewrites, push_observed_trace_type,
truncate_for_diagnostics,
};
use crate::metrics::{OTLP_TRACES_FAILURE_COUNT, OTLP_TRACES_ROWS};
use crate::metrics::{
OTLP_TRACE_AUX_CACHE_HIT, OTLP_TRACE_AUX_CACHE_MISS, OTLP_TRACES_FAILURE_COUNT,
OTLP_TRACES_ROWS,
};
/// Confirmed auxiliary writes, shared across requests and instance clones.
///
/// ponytail: these tables are ingestion-managed; add invalidation if external
/// mutations must be supported without restarting every serving frontend.
#[derive(Clone)]
pub(crate) struct TraceAuxCache {
entries: Cache<TraceAuxCacheKey, ()>,
}
#[derive(Clone, Hash, PartialEq, Eq)]
struct TraceAuxCacheKey {
table: Arc<TableName>,
entry: TraceAuxEntry,
}
impl TraceAuxCacheKey {
/// Estimated key memory, excluding cache and allocator overhead.
fn size(&self) -> usize {
let entry_size = match &self.entry {
TraceAuxEntry::Service(service) => service.capacity(),
TraceAuxEntry::Operation(service, span, kind) => {
service.capacity() + span.capacity() + kind.capacity()
}
};
// Charge the shared table name per entry to keep the estimate conservative.
mem::size_of::<Self>()
+ mem::size_of::<TableName>()
+ self.table.catalog_name.capacity()
+ self.table.schema_name.capacity()
+ self.table.table_name.capacity()
+ entry_size
}
}
#[derive(Clone, Hash, PartialEq, Eq)]
enum TraceAuxEntry {
Service(String),
Operation(String, String, String),
}
impl TraceAuxCache {
/// Creates a cache with an estimated byte budget; zero disables caching.
pub(crate) fn new(max_size: u64) -> Self {
Self {
entries: Cache::builder()
.max_capacity(max_size)
.weigher(|key: &TraceAuxCacheKey, _: &()| {
u32::try_from(key.size()).unwrap_or(u32::MAX)
})
.build(),
}
}
/// Removes confirmed writes and returns keys to remember after insertion.
/// Only the first cache check in a request supplies metrics.
fn filter(
&self,
table: TableName,
aux_data: &mut TraceAuxData,
metrics: Option<(&IntCounter, &IntCounter)>,
) -> Vec<TraceAuxCacheKey> {
let lookups = aux_data.services.len() + aux_data.operations.len();
let table = Arc::new(table);
let mut pending = Vec::new();
let mut retain = |entry| {
let key = TraceAuxCacheKey {
table: table.clone(),
entry,
};
if self.entries.get(&key).is_some() {
false
} else {
pending.push(key);
true
}
};
aux_data
.services
.retain(|service| retain(TraceAuxEntry::Service(service.clone())));
aux_data.operations.retain(|(service, span, kind)| {
retain(TraceAuxEntry::Operation(
service.clone(),
span.clone(),
kind.clone(),
))
});
if let Some((hits, misses)) = metrics {
hits.inc_by((lookups - pending.len()) as u64);
misses.inc_by(pending.len() as u64);
}
pending
}
fn record(&self, keys: Vec<TraceAuxCacheKey>) {
for key in keys {
self.entries.insert(key, ());
}
}
/// Reserves background capacity only for misses, without holding a cache lock.
async fn acquire_for_misses(
&self,
limiter: &RequestLimiter,
table: TableName,
mut aux_data: TraceAuxData,
metrics: (&IntCounter, &IntCounter),
) -> ServerResult<Option<Arc<OwnedSemaphorePermit>>> {
self.filter(table.clone(), &mut aux_data, Some(metrics));
if aux_data.is_empty() {
return Ok(None);
}
let permit = limiter
.acquire()
.await
.map_err(|_| error::BatcherChannelClosedSnafu.build())?;
// Another request may have confirmed these entries while admission waited.
self.filter(table, &mut aux_data, None);
Ok((!aux_data.is_empty()).then_some(permit))
}
}
/// Merge converted V2 rows only when their fixed schemas agree.
fn merge_trace_v2_rows(rows: &mut Rows, batch: Rows) -> ServerResult<()> {
@@ -101,11 +232,52 @@ struct TraceChunkIngestContext<'a> {
/// Accumulates trace outcomes, auxiliary rows, and bounded failure details.
struct TraceIngestState {
needs_aux_write: bool,
aux_data: TraceAuxData,
pending_aux_data: Vec<(TraceAuxData, Vec<WriteCompletion>)>,
outcome: TraceIngestOutcome,
failure_messages: TraceFailureMessages,
}
impl TraceIngestState {
fn add_aux_data(&mut self, aux_data: TraceAuxData, completions: Vec<WriteCompletion>) {
if completions.is_empty() {
self.aux_data.services.extend(aux_data.services);
self.aux_data.operations.extend(aux_data.operations);
} else if !aux_data.is_empty() {
self.pending_aux_data.push((aux_data, completions));
}
}
}
/// Keeps confirmed chunks even if another chunk fails or loses its completion sender.
async fn confirmed_trace_aux_data(
pending: Vec<(TraceAuxData, Vec<WriteCompletion>)>,
) -> TraceAuxData {
let results = futures::future::join_all(pending.into_iter().map(
|(aux_data, completions)| async move {
for completion in completions {
completion.wait().await?;
}
Ok::<_, BoxedError>(aux_data)
},
))
.await;
let mut confirmed = TraceAuxData::default();
for result in results {
match result {
Ok(aux_data) => {
confirmed.services.extend(aux_data.services);
confirmed.operations.extend(aux_data.operations);
}
Err(err) => {
warn!(err; "Skipping auxiliary trace rows after asynchronous main write failure")
}
}
}
confirmed
}
/// Bounded, deduplicated failure details for one trace request.
///
/// Occurrences past [`TRACE_FAILURE_MESSAGE_LIMIT`] distinct failures are
@@ -876,6 +1048,32 @@ impl Instance {
conventions: &str,
ctx: QueryContextRef,
) -> ServerResult<TraceIngestOutcome> {
let aux_limiter = self
.trace_aux_limiter
.as_ref()
.filter(|_| ctx.batching_enabled());
// Reserve once per request, before any main chunks are submitted. Main
// batcher permits remain independent so multi-chunk requests can progress.
let aux_permit = if let Some(limiter) = aux_limiter {
let mut aux_data = TraceAuxData::default();
for span in groups.iter().flat_map(|group| &group.spans) {
aux_data.observe_span(span);
}
self.trace_aux_cache
.acquire_for_misses(
limiter,
TableName::new(ctx.current_catalog(), ctx.current_schema(), &table_name),
aux_data,
(&OTLP_TRACE_AUX_CACHE_HIT, &OTLP_TRACE_AUX_CACHE_MISS),
)
.await?
} else {
None
};
// Requests without auxiliary admission first check the cache when preparing writes.
let aux_cache_metrics = aux_limiter
.is_none()
.then_some((&*OTLP_TRACE_AUX_CACHE_HIT, &*OTLP_TRACE_AUX_CACHE_MISS));
let is_trace_v1_model = matches!(pipeline, PipelineWay::OtlpTraceDirectV1);
let data_model = match pipeline {
PipelineWay::OtlpTraceDirectV1 => Some(TABLE_DATA_MODEL_TRACE_V1),
@@ -918,7 +1116,11 @@ impl Instance {
table_name: &table_name,
};
let mut ingest_state = TraceIngestState {
// Confirmed hits stay satisfied for this request even if evicted
// later: cache eviction does not delete the stored lookup rows.
needs_aux_write: aux_limiter.is_none() || aux_permit.is_some(),
aux_data: TraceAuxData::default(),
pending_aux_data: Vec::new(),
outcome: TraceIngestOutcome::default(),
failure_messages: TraceFailureMessages::default(),
};
@@ -988,20 +1190,64 @@ impl Instance {
OTLP_TRACES_ROWS.inc_by(ingest_state.outcome.accepted_spans as u64);
if !ingest_state.pending_aux_data.is_empty() {
let pending = std::mem::take(&mut ingest_state.pending_aux_data);
let instance = self.clone();
let pipeline = match pipeline {
PipelineWay::OtlpTraceDirectV1 => PipelineWay::OtlpTraceDirectV1,
PipelineWay::OtlpTraceDirectV2 => PipelineWay::OtlpTraceDirectV2,
_ => PipelineWay::OtlpTraceDirectV0,
};
let table_name = table_name.clone();
let ctx = ctx.clone();
common_runtime::spawn_ingest(async move {
let _permit = aux_permit;
let aux_data = confirmed_trace_aux_data(pending).await;
let result: ServerResult<()> = async {
let (requests, keys) = instance.prepare_trace_aux_write(
aux_data,
&pipeline,
&table_name,
&ctx,
aux_cache_metrics,
)?;
if !requests.inserts.is_empty() {
instance
.insert_trace_aux_requests(requests, keys, &pipeline, ctx)
.await?;
}
Ok(())
}
.await;
if let Err(err) = result {
OTLP_TRACES_FAILURE_COUNT
.with_label_values(&["aux_table_update_failed"])
.inc();
warn!(table_name, err = ?err, "Failed to write auxiliary trace rows after asynchronous main write");
}
});
}
if !ingest_state.aux_data.is_empty() {
// Auxiliary trace tables are derived from spans whose main-table
// writes are already confirmed, so they never create new accepted
// spans and they do not affect rejected span counts.
let aux_requests = otlp::trace::to_grpc_insert_requests_for_aux_tables(
// Auxiliary trace tables are derived from accepted spans, so they
// never create new accepted spans or affect rejected span counts.
let aux_requests = self.prepare_trace_aux_write(
std::mem::take(&mut ingest_state.aux_data),
ingest_ctx.pipeline,
ingest_ctx.table_name,
&ctx,
aux_cache_metrics,
);
match aux_requests {
Ok((aux_requests, _)) if !aux_requests.inserts.is_empty() => {
Ok((aux_requests, pending_aux_keys)) if !aux_requests.inserts.is_empty() => {
match self
.insert_trace_requests(aux_requests, ingest_ctx.pipeline, ctx)
.insert_trace_aux_requests(
aux_requests,
pending_aux_keys,
ingest_ctx.pipeline,
ctx,
)
.await
{
Ok(output) => {
@@ -1076,6 +1322,72 @@ impl Instance {
Ok(ingest_state.outcome)
}
fn prepare_trace_aux_write(
&self,
mut aux_data: TraceAuxData,
pipeline: &PipelineWay,
table_name: &str,
ctx: &QueryContextRef,
metrics: Option<(&IntCounter, &IntCounter)>,
) -> ServerResult<(RowInsertRequests, Vec<TraceAuxCacheKey>)> {
let keys = self.trace_aux_cache.filter(
TableName::new(ctx.current_catalog(), ctx.current_schema(), table_name),
&mut aux_data,
metrics,
);
if aux_data.is_empty() {
return Ok((RowInsertRequests::default(), keys));
}
let (mut requests, _) =
otlp::trace::to_grpc_insert_requests_for_aux_tables(aux_data, pipeline, table_name)?;
// Remove fully cached tables before v1 reconciliation performs catalog lookups.
requests.inserts.retain(|request| {
request
.rows
.as_ref()
.is_some_and(|rows| !rows.rows.is_empty())
});
Ok((requests, keys))
}
/// Checks deferred cache population without confusing visible rows with completion.
#[cfg(feature = "testing")]
pub fn trace_aux_data_cached_for_test(
&self,
table_name: &str,
mut aux_data: TraceAuxData,
ctx: &QueryContextRef,
) -> bool {
self.trace_aux_cache
.filter(
TableName::new(ctx.current_catalog(), ctx.current_schema(), table_name),
&mut aux_data,
None,
)
.is_empty()
}
async fn insert_trace_aux_requests(
&self,
requests: RowInsertRequests,
keys: Vec<TraceAuxCacheKey>,
pipeline: &PipelineWay,
ctx: QueryContextRef,
) -> ServerResult<Output> {
// Cache entries require a completed write, even when main spans use
// batching with early acknowledgement.
let mut aux_ctx = ctx.fork();
aux_ctx.set_batching_enabled(false);
let output = self
.insert_trace_requests(requests, pipeline, Arc::new(aux_ctx))
.await?;
// ponytail: cache skip_wal writes too. After datanode data loss, lookup
// rows may stay missing until eviction/restart; require durable writes
// before caching if that limitation becomes unacceptable.
self.trace_aux_cache.record(keys);
Ok(output)
}
/// Ingest one owned trace chunk so successful spans can be moved into the
/// accepted set without extra cloning.
async fn ingest_trace_chunk(
@@ -1167,8 +1479,12 @@ impl Instance {
Self::add_trace_write_cost(&mut ingest_state.outcome, output.meta.cost);
ingest_state.outcome.accepted_spans += chunk_rows;
for span in &chunk {
ingest_state.aux_data.observe_span(span);
if ingest_state.needs_aux_write {
let mut aux_data = TraceAuxData::default();
for span in &chunk {
aux_data.observe_span(span);
}
ingest_state.add_aux_data(aux_data, output.meta.write_completions);
}
Ok(())
@@ -1345,7 +1661,11 @@ impl Instance {
Self::add_trace_write_cost(&mut ingest_state.outcome, output.meta.cost);
ingest_state.outcome.accepted_spans += span_count;
retry.add_to_aux_data(&mut ingest_state.aux_data);
if ingest_state.needs_aux_write {
let mut aux_data = TraceAuxData::default();
retry.add_to_aux_data(&mut aux_data);
ingest_state.add_aux_data(aux_data, output.meta.write_completions);
}
Ok(())
}
@@ -1428,7 +1748,11 @@ impl Instance {
Self::add_trace_write_cost(&mut ingest_state.outcome, output.meta.cost);
ingest_state.outcome.accepted_spans += 1;
span.add_to_aux_data(&mut ingest_state.aux_data);
if ingest_state.needs_aux_write {
let mut aux_data = TraceAuxData::default();
span.add_to_aux_data(&mut aux_data);
ingest_state.add_aux_data(aux_data, output.meta.write_completions);
}
}
Ok(())
src/frontend/src/instance/otlp/trace_ingest/tests.rs
+247 -3
View File
@@ -13,6 +13,7 @@
// limitations under the License.
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use api::v1::column_data_type_extension::TypeExt;
use api::v1::value::ValueData;
@@ -20,26 +21,269 @@ use api::v1::{
ColumnDataType, ColumnDataTypeExtension, ColumnSchema, JsonTypeExtension, Row,
RowInsertRequest, Rows, SemanticType, Value,
};
use common_batcher::request_limiter::RequestLimiter;
use common_error::ext::ErrorExt;
use common_error::status_code::StatusCode;
use common_query::WriteCompletion;
use datatypes::prelude::ConcreteDataType;
use datatypes::schema::{
ColumnSchema as DatatypesColumnSchema, SchemaBuilder as DatatypesSchemaBuilder,
};
use opentelemetry_proto::tonic::common::v1::any_value::Value as OtlpValue;
use opentelemetry_proto::tonic::common::v1::{AnyValue, ArrayValue, KeyValue};
use servers::otlp::trace::SERVICE_NAME_COLUMN;
use prometheus::IntCounter;
use servers::otlp::trace::attributes::Attributes;
use servers::otlp::trace::span::{SpanEvents, SpanLinks, TraceSpan};
use servers::otlp::trace::v1::{TraceBinaryType, TraceRetryColumn};
use servers::otlp::trace::{SERVICE_NAME_COLUMN, TraceAuxData};
use table::table_name::TableName;
use super::{
ChunkFailureReaction, Instance, TraceChunkRetry, TraceChunkSchemaState, TraceFailureMessages,
TraceRequestSchema, TraceRequestSchemaPlan, TraceSpanMetadata, TraceTablePreAlter, chunk_owned,
ChunkFailureReaction, Instance, TraceAuxCache, TraceAuxCacheKey, TraceAuxEntry,
TraceChunkRetry, TraceChunkSchemaState, TraceFailureMessages, TraceRequestSchema,
TraceRequestSchemaPlan, TraceSpanMetadata, TraceTablePreAlter, chunk_owned,
wrap_trace_alter_failure,
};
use crate::metrics::OTLP_TRACES_FAILURE_COUNT;
fn trace_aux_data(service: &str, span: &str, kind: &str) -> TraceAuxData {
TraceAuxData {
services: HashSet::from([service.to_string()]),
operations: HashSet::from([(service.to_string(), span.to_string(), kind.to_string())]),
}
}
fn trace_aux_cache_metrics() -> (IntCounter, IntCounter) {
(
IntCounter::new("test_trace_aux_cache_hit", "Cache hits").unwrap(),
IntCounter::new("test_trace_aux_cache_miss", "Cache misses").unwrap(),
)
}
#[tokio::test]
async fn test_trace_aux_waits_for_successful_chunk_completions() {
let (success_tx, success_rx) = tokio::sync::oneshot::channel();
let (failure_tx, failure_rx) = tokio::sync::oneshot::channel();
let completion = |receiver: tokio::sync::oneshot::Receiver<bool>| {
WriteCompletion::new(async move {
if receiver.await.unwrap_or(false) {
Ok(())
} else {
Err(common_error::ext::BoxedError::new(
servers::error::InternalSnafu {
err_msg: "main write failed",
}
.build(),
))
}
})
};
let mut confirmed = Box::pin(super::confirmed_trace_aux_data(vec![
(
trace_aux_data("good", "span", "server"),
vec![completion(success_rx)],
),
(
trace_aux_data("bad", "span", "server"),
vec![completion(failure_rx)],
),
]));
assert!(futures::poll!(confirmed.as_mut()).is_pending());
success_tx.send(true).unwrap();
assert!(futures::poll!(confirmed.as_mut()).is_pending());
// A lost completion sender must not count as successful storage.
drop(failure_tx);
let aux = confirmed.await;
assert_eq!(aux.services, HashSet::from(["good".to_string()]));
assert_eq!(
aux.operations,
trace_aux_data("good", "span", "server").operations
);
}
#[test]
fn test_trace_aux_cache_confirmed_writes() {
let cache = TraceAuxCache::new(1024);
let (hits, misses) = trace_aux_cache_metrics();
let metrics = Some((&hits, &misses));
let table = TableName::new("greptime", "public", "traces");
let mut data = trace_aux_data("svc", "span", "server");
let pending = cache.filter(table.clone(), &mut data, metrics);
assert_eq!(pending.len(), 2);
assert_eq!((hits.get(), misses.get()), (0, 2));
// Preparing a write must not suppress a retry after failure or cancellation.
let mut retry = trace_aux_data("svc", "span", "server");
assert_eq!(cache.filter(table.clone(), &mut retry, None).len(), 2);
assert_eq!((hits.get(), misses.get()), (0, 2));
cache.record(pending);
let mut repeated = trace_aux_data("svc", "span", "server");
assert!(
cache
.clone()
.filter(table.clone(), &mut repeated, metrics)
.is_empty()
);
assert!(repeated.is_empty());
assert_eq!((hits.get(), misses.get()), (2, 2));
cache.filter(table.clone(), &mut retry, None);
assert_eq!((hits.get(), misses.get()), (2, 2));
for (span, kind) in [("new_span", "server"), ("span", "client")] {
let mut data = trace_aux_data("svc", span, kind);
assert_eq!(cache.filter(table.clone(), &mut data, metrics).len(), 1);
assert!(data.services.is_empty());
assert_eq!(data.operations.len(), 1);
}
assert_eq!((hits.get(), misses.get()), (4, 4));
cache.filter(table.clone(), &mut TraceAuxData::default(), metrics);
assert_eq!((hits.get(), misses.get()), (4, 4));
let mut other_service = trace_aux_data("other", "span", "server");
assert_eq!(cache.filter(table, &mut other_service, None).len(), 2);
for other_table in [
TableName::new("other", "public", "traces"),
TableName::new("greptime", "other", "traces"),
TableName::new("greptime", "public", "other"),
] {
let mut data = trace_aux_data("svc", "span", "server");
assert_eq!(cache.filter(other_table, &mut data, None).len(), 2);
}
// An operation hit must not suppress a missing service entry.
let cache = TraceAuxCache::new(1024);
let table = TableName::new("greptime", "public", "traces");
let mut data = trace_aux_data("svc", "span", "server");
data.services.clear();
cache.record(cache.filter(table.clone(), &mut data, None));
let mut data = trace_aux_data("svc", "span", "server");
assert_eq!(cache.filter(table, &mut data, None).len(), 1);
assert_eq!(data.services.len(), 1);
assert!(data.operations.is_empty());
}
#[test]
fn test_trace_aux_cache_zero_size_disables_caching() {
let cache = TraceAuxCache::new(0);
let (hits, misses) = trace_aux_cache_metrics();
let table = TableName::new("greptime", "public", "traces");
let mut data = trace_aux_data("svc", "span", "server");
cache.record(cache.filter(table.clone(), &mut data, Some((&hits, &misses))));
let mut repeated = trace_aux_data("svc", "span", "server");
assert_eq!(
cache
.filter(table, &mut repeated, Some((&hits, &misses)))
.len(),
2
);
assert_eq!(repeated.services.len(), 1);
assert_eq!(repeated.operations.len(), 1);
assert_eq!(hits.get(), 0);
assert_eq!(misses.get(), 4);
}
#[test]
fn test_trace_aux_cache_accounts_for_key_memory() {
let cache = TraceAuxCache::new(1024);
let table = TableName::new("greptime", "public", "traces");
let mut data = trace_aux_data("svc", "span", "server");
cache.record(cache.filter(table.clone(), &mut data, None));
cache.entries.run_pending_tasks();
assert_eq!(cache.entries.entry_count(), 2);
assert!(cache.entries.weighted_size() > 2);
assert!(cache.entries.weighted_size() <= 1024);
// Spare string capacity counts even when the string content is empty.
let large = || String::with_capacity(2048);
let service = || TraceAuxEntry::Service("svc".to_string());
for (table, entry) in [
(TableName::new(large(), "public", "traces"), service()),
(TableName::new("greptime", large(), "traces"), service()),
(TableName::new("greptime", "public", large()), service()),
(table.clone(), TraceAuxEntry::Service(large())),
(
table.clone(),
TraceAuxEntry::Operation(large(), String::new(), String::new()),
),
(
table.clone(),
TraceAuxEntry::Operation(String::new(), large(), String::new()),
),
(
table,
TraceAuxEntry::Operation(String::new(), String::new(), large()),
),
] {
let cache = TraceAuxCache::new(1024);
cache.record(vec![TraceAuxCacheKey {
table: Arc::new(table),
entry,
}]);
cache.entries.run_pending_tasks();
assert_eq!(cache.entries.entry_count(), 0);
}
// Empty strings still retain the key and table structs.
let cache = TraceAuxCache::new(1);
cache.record(vec![TraceAuxCacheKey {
table: Arc::new(TableName::new("", "", "")),
entry: TraceAuxEntry::Service(String::new()),
}]);
cache.entries.run_pending_tasks();
assert_eq!(cache.entries.entry_count(), 0);
}
#[tokio::test]
async fn test_trace_aux_admission_rechecks_cache() {
let cache = TraceAuxCache::new(1024);
let (hits, misses) = trace_aux_cache_metrics();
let limiter = RequestLimiter::try_new(1).unwrap();
let table = TableName::new("greptime", "public", "traces");
let data = || trace_aux_data("svc", "span", "server");
let permit = cache
.acquire_for_misses(&limiter, table.clone(), data(), (&hits, &misses))
.await
.unwrap()
.unwrap();
let cloned_cache = cache.clone();
let cloned_limiter = limiter.clone();
let mut waiting = Box::pin(cloned_cache.acquire_for_misses(
&cloned_limiter,
table.clone(),
data(),
(&hits, &misses),
));
assert!(futures::poll!(waiting.as_mut()).is_pending());
assert_eq!((hits.get(), misses.get()), (0, 4));
// Successful storage populates the shared cache before releasing admission.
cache.record(cache.filter(table.clone(), &mut data(), None));
assert!(
cloned_cache
.acquire_for_misses(&cloned_limiter, table.clone(), data(), (&hits, &misses))
.await
.unwrap()
.is_none()
);
drop(permit);
assert!(waiting.await.unwrap().is_none());
assert_eq!((hits.get(), misses.get()), (2, 4));
// Rechecking a now-warm request must release its briefly acquired slot.
let other = cache.acquire_for_misses(
&limiter,
table,
trace_aux_data("svc", "other", "server"),
(&hits, &misses),
);
assert!(futures::poll!(Box::pin(other).as_mut()).is_ready());
assert_eq!((hits.get(), misses.get()), (3, 5));
}
#[test]
fn test_chunk_owned() {
let chunks = chunk_owned(vec![1, 2, 3], 2);
src/frontend/src/metrics.rs
+7
View File
@@ -12,6 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use common_frontend::metrics::{CACHE_HIT, CACHE_MISS};
use lazy_static::lazy_static;
use prometheus::*;
@@ -68,6 +69,12 @@ lazy_static! {
)
.unwrap();
/// Initial trace auxiliary-cache lookups, excluding admission and write rechecks.
pub static ref OTLP_TRACE_AUX_CACHE_HIT: IntCounter =
CACHE_HIT.with_label_values(&["otlp_trace_aux"]);
pub static ref OTLP_TRACE_AUX_CACHE_MISS: IntCounter =
CACHE_MISS.with_label_values(&["otlp_trace_aux"]);
/// The number of OpenTelemetry logs send by frontend node.
pub static ref OTLP_LOGS_ROWS: IntCounter = register_int_counter!(
"greptime_frontend_otlp_logs_rows",
src/frontend/src/service_config/otlp.rs
+44
View File
@@ -12,9 +12,12 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use common_base::readable_size::ReadableSize;
use common_stat::get_total_memory_readable;
use serde::{Deserialize, Serialize};
const DEFAULT_TRACE_INGEST_CHUNK_SIZE: usize = 512;
const MIN_TRACE_AUX_CACHE_SIZE: ReadableSize = ReadableSize::mb(32);
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
@@ -22,6 +25,12 @@ pub struct OtlpOptions {
pub enable: bool,
/// Maximum spans per trace ingest chunk. Set to 0 to disable splitting.
pub trace_ingest_chunk_size: usize,
/// Estimated memory budget for cached trace service/operation keys per frontend,
/// shared across all catalogs, schemas, and trace tables. Set to 0 to disable caching.
/// Defaults to 1/128 of the host or pod memory limit, with a minimum of 32MiB.
/// Uses 32MiB if memory detection is unavailable.
/// Excludes cache and allocator overhead; shared table names are charged per entry.
pub trace_aux_cache_size: ReadableSize,
/// Whether to synthesize the `greptime_otel_resource_info` descriptor
/// table from the resource attributes of OTLP metrics, so metrics-only
/// services reach the semantic graph. Off by default: it creates and
@@ -34,24 +43,51 @@ impl Default for OtlpOptions {
Self {
enable: true,
trace_ingest_chunk_size: DEFAULT_TRACE_INGEST_CHUNK_SIZE,
trace_aux_cache_size: default_trace_aux_cache_size(get_total_memory_readable()),
experimental_enable_resource_info: false,
}
}
}
fn default_trace_aux_cache_size(memory: Option<ReadableSize>) -> ReadableSize {
memory
.map(|size| size / 128)
.unwrap_or_default()
.max(MIN_TRACE_AUX_CACHE_SIZE)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_default_trace_aux_cache_size() {
for (memory, expected) in [
(None, ReadableSize::mb(32)),
(Some(ReadableSize(0)), ReadableSize::mb(32)),
(Some(ReadableSize::gb(1)), ReadableSize::mb(32)),
(Some(ReadableSize::gb(4)), ReadableSize::mb(32)),
(Some(ReadableSize::gb(8)), ReadableSize::mb(64)),
(Some(ReadableSize::gb(64)), ReadableSize::mb(512)),
] {
assert_eq!(default_trace_aux_cache_size(memory), expected);
}
}
#[test]
fn test_otlp_options() {
let default = OtlpOptions::default();
assert!(default.enable);
assert_eq!(default.trace_ingest_chunk_size, 512);
assert_eq!(
default.trace_aux_cache_size,
default_trace_aux_cache_size(get_total_memory_readable())
);
assert!(!default.experimental_enable_resource_info);
let options: OtlpOptions = toml::from_str("enable = false").unwrap();
assert!(!options.enable);
assert_eq!(options.trace_aux_cache_size, default.trace_aux_cache_size);
assert_eq!(
options.trace_ingest_chunk_size,
DEFAULT_TRACE_INGEST_CHUNK_SIZE
@@ -63,5 +99,13 @@ mod tests {
let serialized = toml::to_string(&options).unwrap();
assert_eq!(toml::from_str::<OtlpOptions>(&serialized).unwrap(), options);
for (value, size) in [("0", ReadableSize(0)), ("\"1MiB\"", ReadableSize::mb(1))] {
let options: OtlpOptions =
toml::from_str(&format!("trace_aux_cache_size = {value}")).unwrap();
assert_eq!(options.trace_aux_cache_size, size);
let serialized = toml::to_string(&options).unwrap();
assert_eq!(toml::from_str::<OtlpOptions>(&serialized).unwrap(), options);
}
}
}
src/operator/AGENTS.md
+2
View File
@@ -46,3 +46,5 @@ the default `meter-macros/noop` build and active metering when changing this pat
Finite requests split internally use `admit_write` / `admit_row_insert_batches`
before dispatch. Their database-scoped `QueryContext` admission prevents a second
row debit while `do_request` still records WCU for the actual writes.
Async table-batcher outputs carry storage completions in `OutputMeta`; keep
them when combining submissions so callers can distinguish admission from success.
src/operator/src/batcher.rs
+3 -1
View File
@@ -16,6 +16,7 @@ use std::sync::Arc;
use arrow::record_batch::RecordBatch;
use async_trait::async_trait;
use common_query::Output;
use session::context::QueryContextRef;
use table::metadata::TableInfoRef;
use tokio::sync::OwnedSemaphorePermit;
@@ -32,11 +33,12 @@ pub trait PendingRowsBatcher: Send + Sync {
/// Submits rows according to the acknowledgement policy, retaining the slot until
/// writing completes even when the response acknowledges queue admission only.
/// Cancelling the response wait does not retract an already enqueued submission.
/// Early acknowledgements retain actual storage results in the output metadata.
async fn submit(
&self,
table_info: TableInfoRef,
batch: RecordBatch,
ctx: QueryContextRef,
permit: Arc<OwnedSemaphorePermit>,
) -> Result<usize>;
) -> Result<Output>;
}
src/operator/src/insert.rs
+9 -6
View File
@@ -555,11 +555,14 @@ impl Inserter {
// Observe every table completion even when another table fails.
future::join_all(submissions).await
};
let affected_rows = results.into_iter().sum::<Result<usize>>()?;
Ok(Output::new(
OutputData::AffectedRows(affected_rows),
OutputMeta::new_with_cost(write_cost as _),
))
let mut affected_rows = 0;
let mut meta = OutputMeta::new_with_cost(write_cost as _);
for result in results {
let output = result?;
affected_rows += output.extract_rows_and_cost().0;
meta.write_completions.extend(output.meta.write_completions);
}
Ok(Output::new(OutputData::AffectedRows(affected_rows), meta))
}
/// Handles row inserts request with metric engine.
@@ -3302,7 +3305,7 @@ mod tests {
_batch: arrow::record_batch::RecordBatch,
_ctx: QueryContextRef,
_permit: Arc<tokio::sync::OwnedSemaphorePermit>,
) -> Result<usize> {
) -> Result<Output> {
panic!("empty writes must not submit a batch")
}
}
src/pipeline/Cargo.toml
+1
View File
@@ -19,6 +19,7 @@ chrono.workspace = true
chrono-tz.workspace = true
common-catalog.workspace = true
common-error.workspace = true
common-frontend.workspace = true
common-macro.workspace = true
common-meta.workspace = true
common-query.workspace = true
src/pipeline/src/manager/pipeline_cache.rs
+165 -53
View File
@@ -16,6 +16,7 @@ use std::future::Future;
use std::sync::Arc;
use std::time::Duration;
use common_frontend::metrics;
use datatypes::timestamp::TimestampNanosecond;
use moka::future::Cache;
@@ -32,7 +33,7 @@ const PIPELINES_CACHE_SIZE: u64 = 10000;
/// to encapsulate inner cache. Only public methods are exposed.
///
/// `pipelines` and `original_pipelines` are keyed by the *requested* schema so
/// a lookup is a single key probe, as [`Cache::try_get_with`] requires;
/// a lookup is a single key probe through [`Cache::entry`];
/// resolving it to a stored schema is the loader's job. `failover_cache` has no
/// loader and keeps the stored-schema key.
pub(crate) struct PipelineCache {
@@ -71,7 +72,7 @@ impl PipelineCache {
}
}
/// Concurrent misses on the same key share one `init` call.
/// Concurrent misses share one `init` call; successful waiters count as hits.
pub(crate) async fn get_pipeline_with(
&self,
schema: &str,
@@ -80,13 +81,17 @@ impl PipelineCache {
init: impl Future<Output = Result<Arc<Pipeline>>>,
) -> Result<Arc<Pipeline>> {
let key = generate_pipeline_cache_key(schema, name, version);
self.pipelines
.try_get_with(key, init)
.await
let entry = self.pipelines.entry(key).or_try_insert_with(init).await;
metrics::record_cache_lookup(
"pipeline",
entry.as_ref().is_ok_and(|entry| !entry.is_fresh()),
);
entry
.map(|entry| entry.into_value())
.map_err(|error| CacheLoadSnafu { error }.build())
}
/// Concurrent misses on the same key share one `init` call.
/// Concurrent misses share one `init` call; successful waiters count as hits.
pub(crate) async fn get_pipeline_str_with(
&self,
schema: &str,
@@ -95,9 +100,17 @@ impl PipelineCache {
init: impl Future<Output = Result<PipelineContent>>,
) -> Result<PipelineContent> {
let key = generate_pipeline_cache_key(schema, name, version);
self.original_pipelines
.try_get_with(key, init)
.await
let entry = self
.original_pipelines
.entry(key)
.or_try_insert_with(init)
.await;
metrics::record_cache_lookup(
"pipeline_original",
entry.as_ref().is_ok_and(|entry| !entry.is_fresh()),
);
entry
.map(|entry| entry.into_value())
.map_err(|error| CacheLoadSnafu { error }.build())
}
@@ -114,6 +127,7 @@ impl PipelineCache {
generate_pipeline_cache_key(schema, name, version),
] {
if let Some(content) = self.failover_cache.get(&key).await {
metrics::record_cache_lookup("pipeline_failover", true);
return Ok(Some(content));
}
}
@@ -126,7 +140,7 @@ impl PipelineCache {
.filter(|(k, _)| k.ends_with(&suffix))
.collect::<Vec<_>>();
match found.len() {
let result = match found.len() {
0 => Ok(None),
1 => Ok(Some(found.remove(0).1)),
_ => MultiPipelineWithDiffSchemaSnafu {
@@ -139,7 +153,12 @@ impl PipelineCache {
.join(","),
}
.fail(),
}
};
metrics::record_cache_lookup(
"pipeline_failover",
result.as_ref().is_ok_and(|content| content.is_some()),
);
result
}
pub(crate) async fn insert_failover_cache(&self, content: PipelineContent, with_latest: bool) {
@@ -191,7 +210,8 @@ impl PipelineCache {
mod tests {
use std::sync::atomic::{AtomicUsize, Ordering};
use tokio::sync::Barrier;
use common_frontend::metrics::{CACHE_HIT, CACHE_MISS};
use tokio::sync::Notify;
use super::*;
@@ -205,36 +225,149 @@ mod tests {
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[tokio::test]
async fn test_concurrent_misses_run_one_loader() {
const CONCURRENCY: usize = 8;
let cache = Arc::new(PipelineCache::new(Duration::from_secs(60)));
let loads = Arc::new(AtomicUsize::new(0));
let barrier = Arc::new(Barrier::new(CONCURRENCY));
let handles = (0..CONCURRENCY)
let hits = CACHE_HIT.with_label_values(&["pipeline_original"]);
let misses = CACHE_MISS.with_label_values(&["pipeline_original"]);
let before = (hits.get(), misses.get());
let cache = PipelineCache::new(Duration::from_secs(60));
let loads = AtomicUsize::new(0);
let release = Notify::new();
let mut requests = (0..CONCURRENCY)
.map(|_| {
let (cache, loads, barrier) = (cache.clone(), loads.clone(), barrier.clone());
tokio::spawn(async move {
barrier.wait().await;
Box::pin(cache.get_pipeline_str_with("db", "p", None, async {
loads.fetch_add(1, Ordering::SeqCst);
release.notified().await;
Ok(content_at(1))
}))
})
.collect::<Vec<_>>();
for request in &mut requests {
assert!(futures::poll!(request.as_mut()).is_pending());
}
assert_eq!(loads.load(Ordering::SeqCst), 1);
release.notify_waiters();
for request in requests {
assert_eq!(request.await.unwrap(), content_at(1));
}
assert_eq!(loads.load(Ordering::SeqCst), 1);
assert_eq!(
(hits.get(), misses.get()),
(before.0 + CONCURRENCY as u64 - 1, before.1 + 1)
);
}
#[tokio::test]
async fn test_pipeline_lookup_metrics() {
let cache = PipelineCache::new(Duration::from_secs(60));
for compiled in [false, true] {
let cache_type = if compiled {
"pipeline"
} else {
"pipeline_original"
};
let hits = CACHE_HIT.with_label_values(&[cache_type]);
let misses = CACHE_MISS.with_label_values(&[cache_type]);
let before = (hits.get(), misses.get());
let load = async |fail| {
if compiled {
cache
.get_pipeline_with("db", "p", None, async {
if fail {
return Err(crate::error::PipelineNotFoundSnafu {
name: "p",
version: None,
}
.build());
}
Ok(Arc::new(
crate::manager::table::PipelineTable::compile_pipeline(
"transform:\n - fields: [message]\n type: string",
)?,
))
})
.await
.map(|_| ())
} else {
cache
.get_pipeline_str_with("db", "p", None, async {
loads.fetch_add(1, Ordering::SeqCst);
// Hold the loader open so every caller is waiting on it.
tokio::time::sleep(Duration::from_millis(100)).await;
if fail {
return Err(crate::error::PipelineNotFoundSnafu {
name: "p",
version: None,
}
.build());
}
Ok(content_at(1))
})
.await
.unwrap()
})
})
.collect::<Vec<_>>();
for handle in handles {
assert_eq!(handle.await.unwrap(), content_at(1));
.map(|_| ())
}
};
assert!(load(true).await.is_err());
load(false).await.unwrap();
// A warm lookup must not evaluate the failing initializer.
load(true).await.unwrap();
assert_eq!((hits.get(), misses.get()), (before.0 + 1, before.1 + 2));
}
assert_eq!(loads.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn test_failover_lookup_metrics() {
let hits = CACHE_HIT.with_label_values(&["pipeline_failover"]);
let misses = CACHE_MISS.with_label_values(&["pipeline_failover"]);
let before = (hits.get(), misses.get());
let cache = PipelineCache::new(Duration::from_secs(60));
assert!(
cache
.get_failover_cache("db", "p", None)
.await
.unwrap()
.is_none()
);
let local = PipelineContent {
schema: "a".into(),
..content_at(1)
};
cache.insert_failover_cache(local.clone(), true).await;
// Direct schema lookup and cross-schema fallback each count one hit.
assert_eq!(
cache.get_failover_cache("a", "p", None).await.unwrap(),
Some(local.clone())
);
assert_eq!(
cache.get_failover_cache("db", "p", None).await.unwrap(),
Some(local)
);
cache
.insert_failover_cache(
PipelineContent {
schema: "b".into(),
..content_at(2)
},
true,
)
.await;
assert!(cache.get_failover_cache("db", "p", None).await.is_err());
// Global pipelines win over local entries inserted before or after them.
for (schema, version) in [(EMPTY_SCHEMA_NAME, 3), ("c", 4)] {
cache
.insert_failover_cache(
PipelineContent {
schema: schema.into(),
..content_at(version)
},
true,
)
.await;
assert_eq!(
cache.get_failover_cache("db", "p", None).await.unwrap(),
Some(content_at(3))
);
}
assert_eq!((hits.get(), misses.get()), (before.0 + 4, before.1 + 2));
}
#[tokio::test]
@@ -288,25 +421,4 @@ mod tests {
let failover = cache.get_failover_cache("b", "p", None).await.unwrap();
assert_eq!(failover.map(|c| c.version), Some(v2.version));
}
#[tokio::test]
async fn test_failover_serves_global_pipeline_to_unwarmed_schema() {
let cache = PipelineCache::new(Duration::from_secs(60));
let content = content_at(1);
cache.insert_failover_cache(content.clone(), true).await;
let found = cache.get_failover_cache("b", "p", None).await.unwrap();
assert_eq!(found, Some(content.clone()));
// A same-named pipeline under another schema must not shadow the global one.
let schema_local = PipelineContent {
schema: "x".to_string(),
..content_at(2)
};
cache.insert_failover_cache(schema_local, true).await;
let found = cache.get_failover_cache("b", "p", None).await.unwrap();
assert_eq!(found, Some(content));
}
}
src/pipeline/src/manager/pipeline_operator.rs
+4 -1
View File
@@ -19,6 +19,7 @@ use std::time::{Duration, Instant};
use api::v1::CreateTableExpr;
use catalog::{CatalogManagerRef, RegisterSystemTableRequest};
use common_catalog::consts::{DEFAULT_PRIVATE_SCHEMA_NAME, default_engine};
use common_frontend::metrics;
use common_meta::rpc::ddl::TriggerReason;
use common_telemetry::info;
use common_time::FOREVER;
@@ -164,7 +165,9 @@ impl PipelineOperator {
/// Get a pipeline table from the cache.
pub fn get_pipeline_table_from_cache(&self, catalog: &str) -> Option<PipelineTableRef> {
self.tables.read().unwrap().get(catalog).cloned()
let table = self.tables.read().unwrap().get(catalog).cloned();
metrics::record_cache_lookup("pipeline_table", table.is_some());
table
}
}
src/servers/AGENTS.md
+2
View File
@@ -41,3 +41,5 @@ cargo nextest run -p servers
```
Keep protocol translation here and permissions/database behavior in frontend.
Async table batching returns storage-completion futures with its early
acknowledgements. Keep completion reporting independent of the caller's lifetime.
src/servers/Cargo.toml
+1
View File
@@ -40,6 +40,7 @@ common-batcher.workspace = true
common-catalog.workspace = true
common-decimal.workspace = true
common-error.workspace = true
common-frontend.workspace = true
common-grpc.workspace = true
common-macro.workspace = true
common-mem-prof = { workspace = true, optional = true }
src/servers/src/batcher/table.rs
+50 -24
View File
@@ -30,6 +30,8 @@ use common_batcher::flush_limiter::FlushLimiter;
use common_batcher::flush_policy::timing::TimingFlushPolicy;
use common_batcher::request_limiter::RequestLimiter;
use common_batcher::worker_registry::WorkerRegistry;
use common_error::ext::BoxedError;
use common_query::{Output, WriteCompletion};
use operator::batcher::PendingRowsBatcher;
use operator::error::{BatchFlushSnafu, Result, UnexpectedSnafu};
use operator::insert::Inserter;
@@ -171,10 +173,10 @@ impl PendingRowsBatcher for TablePendingRowsBatcher {
batch: RecordBatch,
ctx: QueryContextRef,
permit: Arc<OwnedSemaphorePermit>,
) -> Result<usize> {
) -> Result<Output> {
let total_rows = batch.num_rows();
if total_rows == 0 {
return Ok(0);
return Ok(Output::new_with_affected_rows(0));
}
if table_info.catalog_name != ctx.current_catalog()
|| table_info.schema_name != ctx.current_schema()
@@ -213,19 +215,30 @@ impl PendingRowsBatcher for TablePendingRowsBatcher {
}
.fail();
}
if !self.pending_rows_batch_sync {
return Ok(total_rows);
let completion = async move {
response_rx
.await
.map_err(|_| {
UnexpectedSnafu {
violated: "batch worker stopped before reporting its write result"
.to_string(),
}
.build()
})?
.context(BatchFlushSnafu)
};
let mut output = Output::new_with_affected_rows(total_rows);
if self.pending_rows_batch_sync {
completion.await?;
} else {
output
.meta
.write_completions
.push(WriteCompletion::new(async move {
completion.await.map_err(BoxedError::new)
}));
}
response_rx
.await
.map_err(|_| {
UnexpectedSnafu {
violated: "batch worker stopped before reporting its write result".to_string(),
}
.build()
})?
.context(BatchFlushSnafu)?;
Ok(total_rows)
Ok(output)
}
}
@@ -386,17 +399,20 @@ mod tests {
.unwrap()
.unwrap();
let mut submitted = Some(submitted);
let mut completion = None;
if sync {
assert!(!submitted.as_ref().unwrap().is_finished());
} else {
assert_eq!(
timeout(Duration::from_secs(5), submitted.take().unwrap())
.await
.unwrap()
.unwrap()
.unwrap(),
1
);
let mut output = timeout(Duration::from_secs(5), submitted.take().unwrap())
.await
.unwrap()
.unwrap()
.unwrap();
assert_eq!(output.extract_rows_and_cost().0, 1);
assert_eq!(output.meta.write_completions.len(), 1);
let mut pending = Box::pin(output.meta.write_completions.pop().unwrap().wait());
assert!(futures::poll!(pending.as_mut()).is_pending());
completion = Some(pending);
}
// Early acknowledgement must not release capacity before completion.
let acquire = batcher.acquire();
@@ -420,9 +436,15 @@ mod tests {
.unwrap();
assert_eq!(result.is_err(), fail);
if !fail {
assert_eq!(result.unwrap(), 1);
let output = result.unwrap();
assert_eq!(output.extract_rows_and_cost().0, 1);
assert!(output.meta.write_completions.is_empty());
}
}
if let Some(completion) = completion {
let result = timeout(Duration::from_secs(5), completion).await.unwrap();
assert_eq!(result.is_err(), fail);
}
timeout(Duration::from_secs(5), acquire)
.await
.unwrap()
@@ -533,9 +555,13 @@ mod tests {
let (first_result, second_result) = timeout(Duration::from_secs(5), async {
if shared_request {
let permit = batcher.acquire().await.unwrap();
tokio::join!(
let (first, second) = tokio::join!(
batcher.submit(table.clone(), first, influx_ctx, permit.clone()),
batcher.submit(table.clone(), second, opentsdb_ctx, permit),
);
(
first.map(|output| output.extract_rows_and_cost().0),
second.map(|output| output.extract_rows_and_cost().0),
)
} else {
let inserter = Inserter::new(
src/servers/src/mysql/handler.rs
+21 -2
View File
@@ -23,6 +23,7 @@ use async_trait::async_trait;
use chrono::{NaiveDate, NaiveDateTime};
use common_catalog::parse_optional_catalog_and_schema_from_db_string;
use common_error::ext::ErrorExt;
use common_frontend::metrics;
use common_query::Output;
use common_telemetry::{debug, error, tracing, warn};
use common_time::Timezone;
@@ -182,8 +183,9 @@ impl MysqlInstanceShim {
/// Retrieve the query and logical plan by a given statement key
fn plan(&self, stmt_key: &str) -> Option<SqlPlan> {
let guard = self.prepared_stmts.read();
guard.get(stmt_key).cloned()
let plan = self.prepared_stmts.read().get(stmt_key).cloned();
metrics::record_cache_lookup("mysql_prepared_stmt", plan.is_some());
plan
}
/// Save the prepared statement and return the parameters and result columns
@@ -1053,6 +1055,23 @@ mod tests {
)
}
#[test]
fn test_prepared_statement_cache_metrics() {
use common_frontend::metrics::{CACHE_HIT, CACHE_MISS};
let hits = CACHE_HIT.with_label_values(&["mysql_prepared_stmt"]);
let misses = CACHE_MISS.with_label_values(&["mysql_prepared_stmt"]);
let before = (hits.get(), misses.get());
let mut shim = create_shim();
assert!(shim.plan("stmt").is_none());
shim.save_plan(SqlPlan::Shortcut("SELECT 1".into()), "stmt".into())
.unwrap();
assert!(shim.plan("stmt").is_some());
shim.do_close("stmt".into());
assert!(shim.plan("stmt").is_none());
assert_eq!((hits.get(), misses.get()), (before.0 + 1, before.1 + 2));
}
#[test]
fn test_batching_context() {
for enabled in [false, true] {
tests-integration/Cargo.toml
+1
View File
@@ -123,6 +123,7 @@ sqlx = { workspace = true, features = [
"chrono",
] }
store-api.workspace = true
temp-env.workspace = true
time = "0.3"
tokio-postgres = { workspace = true }
tokio-util.workspace = true
tests-integration/src/otlp.rs
+574 -3
View File
@@ -14,17 +14,28 @@
#[cfg(test)]
mod test {
use std::sync::Arc;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use TimeUnit as ArrowTimeUnit;
use api::region::RegionResponse;
use api::v1::region::{RegionRequest, region_request};
use client::{DEFAULT_CATALOG_NAME, OutputData};
use common_recordbatch::RecordBatches;
use common_meta::cache::LayeredCacheRegistryBuilder;
use common_meta::node_manager::{
Datanode, DatanodeManager, DatanodeRef, FlownodeManager, FlownodeRef, NodeManagerRef,
};
use common_meta::peer::Peer;
use common_query::request::QueryRequest;
use common_recordbatch::{RecordBatches, SendableRecordBatchStream};
use datatypes::arrow::array::AsArray;
use datatypes::arrow::datatypes::{
DataType, TimeUnit, TimestampMicrosecondType, TimestampMillisecondType,
TimestampNanosecondType, TimestampSecondType,
};
use frontend::instance::Instance;
use frontend::instance::builder::FrontendBuilder;
use frontend::service_config::BatcherOptions;
use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest;
use otel_arrow_rust::proto::opentelemetry::common::v1::any_value::Value as Val;
@@ -39,14 +50,574 @@ mod test {
use otel_arrow_rust::proto::opentelemetry::resource::v1::Resource;
use pipeline::{GreptimePipelineParams, PipelineWay};
use serde_json::json;
use servers::batcher::BatchingProtocol;
use servers::otlp::trace::{TraceAuxData, span};
use servers::query_handler::OpenTelemetryProtocolHandler;
use servers::query_handler::sql::SqlQueryHandler;
use session::context::QueryContext;
use session::protocol_ctx::{OtlpMetricCtx, ProtocolCtx};
use store_api::region_engine::RegionEngine;
use store_api::storage::RegionId;
use tokio::sync::oneshot;
use crate::standalone::GreptimeDbStandaloneBuilder;
use crate::standalone::{GreptimeDbStandalone, GreptimeDbStandaloneBuilder};
use crate::tests;
#[tokio::test(flavor = "multi_thread")]
async fn test_trace_aux_cache() {
check_trace_aux_cache(false).await;
}
#[tokio::test(flavor = "multi_thread")]
async fn test_trace_aux_cache_with_batching_v0_v1() {
check_trace_aux_cache(true).await;
}
#[test]
fn test_trace_aux_async_completion() {
temp_env::with_var("PENDING_ROWS_BATCH_SYNC", Some("false"), || {
tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.unwrap()
.block_on(check_trace_aux_async_completion());
});
}
#[test]
fn test_trace_aux_admission() {
temp_env::with_var("PENDING_ROWS_BATCH_SYNC", Some("false"), || {
tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.unwrap()
.block_on(async {
tokio::time::timeout(Duration::from_secs(60), check_trace_aux_admission())
.await
.expect("auxiliary admission must make progress");
});
});
}
struct TraceAuxWriteGate {
region_id: RegionId,
started: oneshot::Sender<()>,
release: oneshot::Receiver<bool>,
}
struct GatedTraceNodeManager {
inner: NodeManagerRef,
gate: Arc<Mutex<Option<TraceAuxWriteGate>>>,
}
#[async_trait::async_trait]
impl DatanodeManager for GatedTraceNodeManager {
async fn datanode(&self, peer: &Peer) -> DatanodeRef {
Arc::new(GatedTraceDatanode {
inner: self.inner.datanode(peer).await,
gate: self.gate.clone(),
})
}
}
#[async_trait::async_trait]
impl FlownodeManager for GatedTraceNodeManager {
async fn flownode(&self, peer: &Peer) -> FlownodeRef {
self.inner.flownode(peer).await
}
}
struct GatedTraceDatanode {
inner: DatanodeRef,
gate: Arc<Mutex<Option<TraceAuxWriteGate>>>,
}
#[async_trait::async_trait]
impl Datanode for GatedTraceDatanode {
async fn handle(
&self,
request: RegionRequest,
) -> common_meta::error::Result<RegionResponse> {
let gate = {
let mut gate = self.gate.lock().unwrap();
if gate.as_ref().is_some_and(|gate| {
matches!(&request.body, Some(region_request::Body::Inserts(inserts))
if inserts.requests.iter().any(|insert| insert.region_id == gate.region_id.as_u64()))
}) {
gate.take()
} else {
None
}
};
if let Some(gate) = gate {
gate.started.send(()).unwrap();
if !gate.release.await.unwrap() {
return common_meta::error::UnexpectedSnafu {
err_msg: "injected auxiliary write failure",
}
.fail();
}
}
self.inner.handle(request).await
}
async fn handle_query(
&self,
request: QueryRequest,
) -> common_meta::error::Result<SendableRecordBatchStream> {
self.inner.handle_query(request).await
}
}
async fn check_trace_aux_admission() {
let standalone = GreptimeDbStandaloneBuilder::new("trace_aux_admission")
.with_table_batcher(BatcherOptions {
protocols: vec![BatchingProtocol::Otlp],
pending_rows_flush_interval: Duration::from_secs(3600),
max_batch_rows: 1,
max_inflight_requests: 1,
..Default::default()
})
.build()
.await;
let source = standalone.fe_instance();
let gate = Arc::new(Mutex::new(None));
let cache_registry = Arc::new(
cache::with_default_composite_cache_registry(
LayeredCacheRegistryBuilder::default().add_cache_registry(
cache::build_fundamental_cache_registry(standalone.kv_backend.clone()),
),
)
.unwrap()
.build(),
);
let mut options = standalone.opts.frontend_options();
options.otlp.trace_ingest_chunk_size = 1;
let instance = Arc::new(
FrontendBuilder::new(
options,
standalone.kv_backend.clone(),
cache_registry,
source.catalog_manager().clone(),
Arc::new(GatedTraceNodeManager {
inner: source.node_manager().clone(),
gate: gate.clone(),
}),
source.procedure_executor().clone(),
Arc::new(catalog::process_manager::ProcessManager::new(
"trace_aux_admission".into(),
None,
)),
)
.try_build()
.await
.unwrap(),
);
for version in 0..=1 {
let table_name = format!("trace_aux_admission_v{version}");
let mut context = QueryContext::with(DEFAULT_CATALOG_NAME, "public");
context.set_extension(
common_catalog::consts::TRACE_TABLE_NAME_SESSION_KEY,
&table_name,
);
context.set_extension(table::requests::TRACE_TABLE_PARTITIONS_HINT_KEY, "1");
context.set_batching_enabled(true);
let ctx = Arc::new(context);
let write = async |request| {
instance
.traces(
instance.clone(),
request,
if version == 0 {
PipelineWay::OtlpTraceDirectV0
} else {
PipelineWay::OtlpTraceDirectV1
},
GreptimePipelineParams::default(),
table_name.clone(),
ctx.clone(),
)
.await
.unwrap()
};
let cached = |request: ExportTraceServiceRequest| {
let mut data = TraceAuxData::default();
for span in span::parse(request).iter().flat_map(|group| &group.spans) {
data.observe_span(span);
}
instance.trace_aux_data_cached_for_test(&table_name, data, &ctx)
};
let warm = trace_aux_request();
assert_eq!(1, write(warm.clone()).await.accepted_spans);
while !cached(warm.clone()) {
tokio::task::yield_now().await;
}
let table = instance
.catalog_manager()
.table(
DEFAULT_CATALOG_NAME,
"public",
&format!("{table_name}_operations"),
None,
)
.await
.unwrap()
.unwrap();
let route = instance
.partition_manager()
.find_physical_table_route(table.table_info().table_id())
.await
.unwrap();
let region_id = route.region_routes[0].region.id;
for success in [true, false] {
let mut cold = warm.clone();
let spans = &mut cold.resource_spans[0].scope_spans[0].spans;
spans[0].name = format!("cold_{success}");
spans.push(spans[0].clone());
let (started_tx, started_rx) = oneshot::channel();
let (release_tx, release_rx) = oneshot::channel();
*gate.lock().unwrap() = Some(TraceAuxWriteGate {
region_id,
started: started_tx,
release: release_rx,
});
// Two chunks must progress with just one main and one auxiliary slot.
assert_eq!(2, write(cold.clone()).await.accepted_spans);
started_rx.await.unwrap();
assert!(
!cached(cold.clone()),
"pending writes must not warm the cache"
);
let before = trace_aux_sequences(&standalone, &table_name).await;
let mut waiting = Box::pin(write(cold.clone()));
assert!(futures::poll!(waiting.as_mut()).is_pending());
// Main admission is free while auxiliary storage is blocked. Warm
// requests must bypass auxiliary admission and keep making progress.
assert_eq!(1, write(warm.clone()).await.accepted_spans);
assert!(futures::poll!(waiting.as_mut()).is_pending());
release_tx.send(success).unwrap();
assert_eq!(2, waiting.await.accepted_spans);
if !success {
// The failed write cannot satisfy the waiting request's miss.
while !cached(cold.clone()) {
tokio::task::yield_now().await;
}
}
let after = trace_aux_sequences(&standalone, &table_name).await;
assert_eq!(before[1], after[1], "cached service must not be rewritten");
assert_eq!(
before[2] + 1,
after[2],
"exactly one successful operation write"
);
}
}
}
fn trace_aux_request() -> ExportTraceServiceRequest {
serde_json::from_value(json!({
"resourceSpans": [{
"resource": {"attributes": [
{"key": "service.name", "value": {"stringValue": "svc"}}
]},
"scopeSpans": [{"spans": [{
"traceId": "c05d7a4ec8e1f231f02ed6e8da8655b4",
"spanId": "9630f2916e2f7909",
"name": "operation",
"kind": 2,
"startTimeUnixNano": "1736480942444376000",
"endTimeUnixNano": "1736480942444499000"
}]}]
}]
}))
.unwrap()
}
async fn trace_aux_sequences(standalone: &GreptimeDbStandalone, table_name: &str) -> Vec<u64> {
let instance = standalone.fe_instance();
let mut sequences = Vec::new();
for suffix in ["", "_services", "_operations"] {
let table = instance
.catalog_manager()
.table(
DEFAULT_CATALOG_NAME,
"public",
&format!("{table_name}{suffix}"),
None,
)
.await
.unwrap()
.unwrap();
let route = instance
.partition_manager()
.find_physical_table_route(table.table_info().table_id())
.await
.unwrap();
assert_eq!(route.region_routes.len(), 1);
sequences.push(
standalone
.mito_engine
.get_committed_sequence(route.region_routes[0].region.id)
.await
.unwrap(),
);
}
sequences
}
async fn check_trace_aux_async_completion() {
let standalone = GreptimeDbStandaloneBuilder::new("trace_aux_async_completion")
.with_table_batcher(BatcherOptions {
protocols: vec![BatchingProtocol::Otlp],
pending_rows_flush_interval: Duration::from_secs(3600),
max_batch_rows: 2,
..Default::default()
})
.build()
.await;
let instance = standalone.fe_instance();
// V2 bulk ingestion has a separate Variant/Struct conversion failure.
for version in 0..=1 {
let table_name = format!("trace_aux_async_v{version}");
let mut context = QueryContext::with(DEFAULT_CATALOG_NAME, "public");
context.set_extension(
common_catalog::consts::TRACE_TABLE_NAME_SESSION_KEY,
&table_name,
);
context.set_extension(table::requests::TRACE_TABLE_PARTITIONS_HINT_KEY, "1");
context.set_batching_enabled(true);
let ctx = Arc::new(context);
let request = trace_aux_request();
let write = async |request, ctx| {
let outcome = tokio::time::timeout(
Duration::from_secs(30),
instance.traces(
instance.clone(),
request,
if version == 0 {
PipelineWay::OtlpTraceDirectV0
} else {
PipelineWay::OtlpTraceDirectV1
},
GreptimePipelineParams::default(),
table_name.clone(),
ctx,
),
)
.await
.expect("async trace ingestion must return before the batch flush")
.unwrap();
assert_eq!((outcome.accepted_spans, outcome.rejected_spans), (1, 0));
assert!(outcome.error_message.is_none());
};
write(request.clone(), ctx.clone()).await;
assert_trace_v2_query(
instance,
&format!("SELECT COUNT(*) = 0 FROM {table_name}"),
ctx.clone(),
)
.await
.unwrap();
for suffix in ["_services", "_operations"] {
assert!(
instance
.catalog_manager()
.table(
DEFAULT_CATALOG_NAME,
"public",
&format!("{table_name}{suffix}"),
None
)
.await
.unwrap()
.is_none()
);
}
// The second submission deterministically triggers the main flush.
// Omitting the service avoids a second concurrent auxiliary write.
let mut flush_request = request.clone();
flush_request.resource_spans[0]
.resource
.as_mut()
.unwrap()
.attributes
.clear();
write(flush_request, ctx.clone()).await;
tokio::time::timeout(Duration::from_secs(30), async {
loop {
let mut aux_data = TraceAuxData::default();
for span in span::parse(request.clone())
.into_iter()
.flat_map(|group| group.spans)
{
aux_data.observe_span(&span);
}
if instance.trace_aux_data_cached_for_test(&table_name, aux_data, &ctx) {
break;
}
tokio::task::yield_now().await;
}
})
.await
.expect("deferred auxiliary writes must populate the cache");
for suffix in ["_services", "_operations"] {
assert_trace_v2_query(
instance,
&format!("SELECT COUNT(*) = 1 FROM {table_name}{suffix}"),
ctx.clone(),
)
.await
.unwrap();
}
assert_trace_v2_query(
instance,
&format!("SELECT COUNT(*) > 0 FROM {table_name}"),
ctx.clone(),
)
.await
.unwrap();
let before = trace_aux_sequences(&standalone, &table_name).await;
// A direct warm request finishes before we compare storage sequences.
let mut warm_ctx = ctx.fork();
warm_ctx.set_batching_enabled(false);
write(request, Arc::new(warm_ctx)).await;
let after = trace_aux_sequences(&standalone, &table_name).await;
assert!(after[0] > before[0]);
assert_eq!(after[1..], before[1..], "v{version}");
}
}
async fn check_trace_aux_cache(batching: bool) {
let standalone = GreptimeDbStandaloneBuilder::new("trace_aux_cache")
.with_table_batcher(if batching {
BatcherOptions {
protocols: vec![BatchingProtocol::Otlp],
// Two main spans trigger a flush, while each auxiliary table
// has only one row and would wait an hour if batched.
pending_rows_flush_interval: Duration::from_secs(3600),
max_batch_rows: 2,
..Default::default()
}
} else {
BatcherOptions::default()
})
.build()
.await;
let instance = standalone.fe_instance();
let cloned_instance = Arc::new(instance.as_ref().clone());
// V2 bulk writes currently fail on Variant/Struct schema conversion
// before reaching auxiliary ingestion; cover V2 with direct main writes.
let max_version = if batching { 1 } else { 2 };
for version in 0..=max_version {
let table_name = format!("trace_aux_v{version}");
let mut context = QueryContext::with(DEFAULT_CATALOG_NAME, "public");
context.set_extension(
common_catalog::consts::TRACE_TABLE_NAME_SESSION_KEY,
&table_name,
);
context.set_extension(table::requests::TRACE_TABLE_PARTITIONS_HINT_KEY, "1");
context.set_batching_enabled(batching);
let ctx = Arc::new(context);
let mut request = trace_aux_request();
if batching {
let spans = &mut request.resource_spans[0].scope_spans[0].spans;
spans.push(spans[0].clone());
}
let accepted_spans = if batching { 2 } else { 1 };
let write = async |frontend: &Arc<Instance>, request| {
tokio::time::timeout(
Duration::from_secs(30),
frontend.traces(
frontend.clone(),
request,
match version {
0 => PipelineWay::OtlpTraceDirectV0,
1 => PipelineWay::OtlpTraceDirectV1,
_ => PipelineWay::OtlpTraceDirectV2,
},
GreptimePipelineParams::default(),
table_name.clone(),
ctx.clone(),
),
)
.await
.expect("auxiliary writes must not wait for the batch flush interval")
.unwrap()
};
// An incompatible auxiliary column fails after the main span is
// accepted. No candidate key may be cached from the failed batch.
instance
.do_query(
&format!(
"CREATE TABLE {table_name}_operations (\
\"timestamp\" TIMESTAMP(9) TIME INDEX, service_name STRING, \
span_name STRING, span_kind BIGINT)"
),
ctx.clone(),
)
.await
.remove(0)
.unwrap();
let failed_aux = write(instance, request.clone()).await;
assert_eq!(
(failed_aux.accepted_spans, failed_aux.rejected_spans),
(accepted_spans, 0)
);
assert!(
failed_aux
.error_message
.unwrap()
.contains("Auxiliary trace tables")
);
instance
.do_query(&format!("DROP TABLE {table_name}_operations"), ctx.clone())
.await
.remove(0)
.unwrap();
let retry = write(instance, request.clone()).await;
assert_eq!(
(retry.accepted_spans, retry.rejected_spans),
(accepted_spans, 0)
);
assert!(retry.error_message.is_none(), "{retry:?}");
let before = trace_aux_sequences(&standalone, &table_name).await;
assert!(before.iter().all(|sequence| *sequence > 0));
// Storage sequences, unlike row counts in an upsert table, prove
// the writes were skipped. Cloned Instances must share the cache.
let warm = write(&cloned_instance, request.clone()).await;
assert_eq!(
(warm.accepted_spans, warm.rejected_spans),
(accepted_spans, 0)
);
assert!(warm.error_message.is_none(), "{warm:?}");
let after = trace_aux_sequences(&standalone, &table_name).await;
assert!(after[0] > before[0]);
assert_eq!(after[1..], before[1..], "v{version}");
request.resource_spans[0].scope_spans[0].spans[0].name = "new_operation".into();
let new_operation = write(instance, request).await;
assert_eq!(
(new_operation.accepted_spans, new_operation.rejected_spans),
(accepted_spans, 0)
);
assert!(new_operation.error_message.is_none(), "{new_operation:?}");
let new_sequences = trace_aux_sequences(&standalone, &table_name).await;
assert!(new_sequences[0] > after[0]);
assert_eq!(new_sequences[1], after[1]);
assert!(new_sequences[2] > after[2]);
}
}
#[tokio::test(flavor = "multi_thread")]
async fn test_trace_v2_on_standalone() -> Result<(), Box<dyn std::error::Error>> {
let standalone = GreptimeDbStandaloneBuilder::new("trace_v2_standalone")
tests-integration/tests/http.rs
+1
View File
@@ -3523,6 +3523,7 @@ fn drop_lines_with_inconsistent_results(input: String) -> String {
"disable_ec2_metadata =",
"cache_path =",
"cache_capacity =",
"trace_aux_cache_size =",
"memory_pool_size =",
"scan_memory_limit =",
"sas_token =",