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

chore: add v2 version label to prom metrics (#8885)

Browse Source 
Signed-off-by: shuiyisong <xixing.sys@gmail.com>
This commit is contained in:
shuiyisong
2026-08-14 09:40:50 +00:00
committed by GitHub
parent ed4271af40
commit 072810159a
5 changed files with 42 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/servers/src/http/prom_store.rs
+14 -14
View File
@@ -47,9 +47,11 @@ use crate::http::header::{
};
use crate::pending_rows_batcher::PendingRowsBatcher;
use crate::prom_remote_write::decode::PromSeriesProcessor;
use crate::prom_remote_write::decode_remote_write_request;
use crate::prom_remote_write::v2::decode_remote_write_v2;
use crate::prom_remote_write::validation::PromValidationMode;
use crate::prom_remote_write::{
REMOTE_WRITE_V1_VERSION, REMOTE_WRITE_V2_VERSION, decode_remote_write_request,
};
use crate::prom_store::snappy_decompress;
use crate::query_handler::{PipelineHandlerRef, PromStoreProtocolHandlerRef, PromStoreResponse};
@@ -57,8 +59,6 @@ pub const PHYSICAL_TABLE_PARAM: &str = "physical_table";
pub const DEFAULT_ENCODING: &str = "snappy";
pub const VM_ENCODING: &str = "zstd";
pub const VM_PROTO_VERSION: &str = "1";
const REMOTE_WRITE_V1_VERSION: &str = "1.0";
const REMOTE_WRITE_V2_VERSION: &str = "2.0";
const REMOTE_WRITE_V1_PROTO: &str = "prometheus.WriteRequest";
const REMOTE_WRITE_V2_PROTO: &str = "io.prometheus.write.v2.Request";
const CONTENT_TYPE_PROTO_PARAM: &str = "proto";
@@ -194,11 +194,11 @@ async fn remote_write_v1(
{
Ok(outcome) => outcome,
Err(error) => {
record_remote_write_samples(&db, error.rows_written);
record_remote_write_samples(&db, REMOTE_WRITE_V1_VERSION, error.rows_written);
return Err(error.error);
}
};
record_remote_write_samples(&db, outcome.rows_written);
record_remote_write_samples(&db, REMOTE_WRITE_V1_VERSION, outcome.rows_written);
Ok((
StatusCode::NO_CONTENT,
@@ -264,8 +264,8 @@ async fn remote_write_v2(
{
Ok(outcome) => outcome,
Err(error) => {
record_remote_write_samples(&db, error.samples_written);
record_remote_write_histograms(&db, error.histograms_written);
record_remote_write_samples(&db, REMOTE_WRITE_V2_VERSION, error.samples_written);
record_remote_write_histograms(&db, REMOTE_WRITE_V2_VERSION, error.histograms_written);
return Ok(remote_write_v2_error_response(
error.error,
error.samples_written,
@@ -276,8 +276,8 @@ async fn remote_write_v2(
};
debug_assert_eq!(outcome.samples_written, sample_count);
debug_assert_eq!(outcome.histograms_written, histogram_count);
record_remote_write_samples(&db, outcome.samples_written);
record_remote_write_histograms(&db, outcome.histograms_written);
record_remote_write_samples(&db, REMOTE_WRITE_V2_VERSION, outcome.samples_written);
record_remote_write_histograms(&db, REMOTE_WRITE_V2_VERSION, outcome.histograms_written);
let mut headers = write_cost_header_map(outcome.write_cost);
append_remote_write_v2_written_headers(
@@ -322,7 +322,7 @@ fn prepare_remote_write_context(
query_ctx.set_extension(SEMANTIC_SOURCE_VERSION, remote_write_version);
query_ctx.set_extension(SEMANTIC_METRIC_METADATA_QUALITY, METADATA_QUALITY_INFERRED);
let timer = crate::metrics::METRIC_HTTP_PROM_STORE_WRITE_ELAPSED
.with_label_values(&[db.as_str()])
.with_label_values(&[db.as_str(), remote_write_version])
.start_timer();
(db, query_ctx, timer)
@@ -608,21 +608,21 @@ fn incomplete_prom_write_error() -> error::Error {
.build()
}
fn record_remote_write_samples(db: &str, rows: u64) {
fn record_remote_write_samples(db: &str, version: &str, rows: u64) {
if rows == 0 {
return;
}
crate::metrics::PROM_STORE_REMOTE_WRITE_SAMPLES
.with_label_values(&[db])
.with_label_values(&[db, version])
.inc_by(rows);
}
fn record_remote_write_histograms(db: &str, rows: u64) {
fn record_remote_write_histograms(db: &str, version: &str, rows: u64) {
if rows == 0 {
return;
}
crate::metrics::PROM_STORE_REMOTE_WRITE_HISTOGRAMS
.with_label_values(&[db])
.with_label_values(&[db, version])
.inc_by(rows);
}
src/servers/src/metrics.rs
+9 -14
View File
@@ -44,6 +44,7 @@ pub(crate) const METRIC_POSTGRES_EXTENDED_QUERY: &str = "extended";
pub(crate) const METRIC_METHOD_LABEL: &str = "method";
pub(crate) const METRIC_PATH_LABEL: &str = "path";
pub(crate) const METRIC_RESULT_LABEL: &str = "result";
pub(crate) const METRIC_VERSION_LABEL: &str = "version";
pub(crate) const METRIC_SUCCESS_VALUE: &str = "success";
pub(crate) const METRIC_FAILURE_VALUE: &str = "failure";
@@ -100,39 +101,33 @@ lazy_static! {
vec![0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0, 10.0, 60.0, 300.0]
)
.unwrap();
/// Http prometheus write duration per database.
/// Http prometheus write duration per database and remote write protocol version.
pub static ref METRIC_HTTP_PROM_STORE_WRITE_ELAPSED: HistogramVec = register_histogram_vec!(
"greptime_servers_http_prometheus_write_elapsed",
"servers http prometheus write elapsed",
&[METRIC_DB_LABEL],
&[METRIC_DB_LABEL, METRIC_VERSION_LABEL],
vec![0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0, 10.0, 60.0, 300.0]
)
.unwrap();
/// Prometheus remote write codec duration.
/// Prometheus remote write codec duration per protocol version.
pub static ref METRIC_HTTP_PROM_STORE_CODEC_ELAPSED: HistogramVec = register_histogram_vec!(
"greptime_servers_http_prometheus_codec_elapsed",
"servers http prometheus request codec duration",
&["type"],
&["type", METRIC_VERSION_LABEL],
)
.unwrap();
/// Decode duration of prometheus write request.
pub static ref METRIC_HTTP_PROM_STORE_DECODE_ELAPSED: Histogram = METRIC_HTTP_PROM_STORE_CODEC_ELAPSED
.with_label_values(&["decode"]);
/// Duration to convert prometheus write request to gRPC request.
pub static ref METRIC_HTTP_PROM_STORE_CONVERT_ELAPSED: Histogram = METRIC_HTTP_PROM_STORE_CODEC_ELAPSED
.with_label_values(&["convert"]);
/// The samples count of Prometheus remote write.
/// The samples count of Prometheus remote write per protocol version.
pub static ref PROM_STORE_REMOTE_WRITE_SAMPLES: IntCounterVec = register_int_counter_vec!(
"greptime_servers_prometheus_remote_write_samples",
"frontend prometheus remote write samples",
&[METRIC_DB_LABEL]
&[METRIC_DB_LABEL, METRIC_VERSION_LABEL]
)
.unwrap();
/// The native histograms count of Prometheus remote write.
/// The native histograms count of Prometheus remote write per protocol version.
pub static ref PROM_STORE_REMOTE_WRITE_HISTOGRAMS: IntCounterVec = register_int_counter_vec!(
"greptime_servers_prometheus_remote_write_histograms",
"frontend prometheus remote write native histograms",
&[METRIC_DB_LABEL]
&[METRIC_DB_LABEL, METRIC_VERSION_LABEL]
)
.unwrap();
pub static ref PENDING_BATCHES: IntGauge = register_int_gauge!(
src/servers/src/prom_remote_write/mod.rs
+8 -1
View File
@@ -37,6 +37,11 @@ use crate::prom_remote_write::row_builder::TablesBuilder;
use crate::prom_remote_write::validation::PromValidationMode;
use crate::prom_store::{snappy_decompress, zstd_decompress};
/// Prometheus remote write protocol versions, also used as the `version` label
/// of the remote write metrics.
pub const REMOTE_WRITE_V1_VERSION: &str = "1.0";
pub const REMOTE_WRITE_V2_VERSION: &str = "2.0";
lazy_static! {
static ref PROM_WRITE_REQUEST_POOL: Pool<PromWriteRequest<'static>> =
Pool::new(256, PromWriteRequest::default);
@@ -56,7 +61,9 @@ pub fn decode_remote_write_request(
prom_validation_mode: PromValidationMode,
processor: &mut PromSeriesProcessor,
) -> crate::error::Result<TablesBuilder<'static>> {
let _timer = crate::metrics::METRIC_HTTP_PROM_STORE_DECODE_ELAPSED.start_timer();
let _timer = crate::metrics::METRIC_HTTP_PROM_STORE_CODEC_ELAPSED
.with_label_values(&["decode", REMOTE_WRITE_V1_VERSION])
.start_timer();
// due to vmagent's limitation, there is a chance that vmagent is
// sending content type wrong so we have to apply a fallback with decoding
src/servers/src/prom_remote_write/v2.rs
+7 -3
View File
@@ -43,8 +43,8 @@ use table::requests::{
use crate::error::{self, Result};
use crate::prom_remote_write::row_builder::PromCtx;
use crate::prom_remote_write::try_decompress;
use crate::prom_remote_write::validation::validate_label_name;
use crate::prom_remote_write::{REMOTE_WRITE_V2_VERSION, try_decompress};
#[allow(deprecated)]
use crate::prom_store::{
DATABASE_LABEL, DATABASE_LABEL_ALT, METRIC_NAME_LABEL, PHYSICAL_TABLE_LABEL,
@@ -128,7 +128,9 @@ pub(crate) fn decode_remote_write_v2(
body: Bytes,
native_histograms_enabled: bool,
) -> Result<RemoteWriteV2WriteRequests> {
let decode_timer = crate::metrics::METRIC_HTTP_PROM_STORE_DECODE_ELAPSED.start_timer();
let decode_timer = crate::metrics::METRIC_HTTP_PROM_STORE_CODEC_ELAPSED
.with_label_values(&["decode", REMOTE_WRITE_V2_VERSION])
.start_timer();
// Match the v1 decoder's VictoriaMetrics fallback: some clients may send a
// mismatched content-encoding header, so try the other compression on failure.
@@ -140,7 +142,9 @@ pub(crate) fn decode_remote_write_v2(
let request = BorrowedRequest::decode(&buf).context(error::DecodePromRemoteRequestSnafu)?;
drop(decode_timer);
let _convert_timer = crate::metrics::METRIC_HTTP_PROM_STORE_CONVERT_ELAPSED.start_timer();
let _convert_timer = crate::metrics::METRIC_HTTP_PROM_STORE_CODEC_ELAPSED
.with_label_values(&["convert", REMOTE_WRITE_V2_VERSION])
.start_timer();
convert_remote_write_v2(request, native_histograms_enabled)
}
src/servers/src/prom_store.rs
+4 -1
View File
@@ -39,6 +39,7 @@ use snafu::{OptionExt, ResultExt, ensure};
use snap::raw::{Decoder, Encoder};
use crate::error::{self, Result};
use crate::prom_remote_write::REMOTE_WRITE_V1_VERSION;
use crate::row_writer::{self, MultiTableData};
pub const METRIC_NAME_LABEL: &str = "__name__";
@@ -458,7 +459,9 @@ fn recordbatch_to_timeseries(
}
pub fn to_grpc_row_insert_requests(request: &WriteRequest) -> Result<(RowInsertRequests, usize)> {
let _timer = crate::metrics::METRIC_HTTP_PROM_STORE_CONVERT_ELAPSED.start_timer();
let _timer = crate::metrics::METRIC_HTTP_PROM_STORE_CODEC_ELAPSED
.with_label_values(&["convert", REMOTE_WRITE_V1_VERSION])
.start_timer();
let mut multi_table_data = MultiTableData::new();