diff --git a/src/handler/grpc/request/ingest.rs b/src/handler/grpc/request/ingest.rs index 3161523581..a6d0c99131 100644 --- a/src/handler/grpc/request/ingest.rs +++ b/src/handler/grpc/request/ingest.rs @@ -77,6 +77,11 @@ impl Ingest for Ingester { } } StreamType::Metrics => { + let stream_name = if stream_name.is_empty(){ + None + } else { + Some(stream_name.as_str()) + }; let log_ingestion_type: IngestionType = req .ingestion_type .unwrap_or_default() @@ -88,7 +93,7 @@ impl Ingest for Ingester { ))) } else { let data = bytes::Bytes::from(in_data.data); - crate::service::metrics::json::ingest(&org_id, data, internal_user) + crate::service::metrics::json::ingest(&org_id, stream_name, data, internal_user) .await .map(|_| ()) // we don't care about success response .map_err(|e| Error::IngestionError(format!("error in ingesting metrics {e}"))) diff --git a/src/handler/http/request/metrics/ingest.rs b/src/handler/http/request/metrics/ingest.rs index 75715b1f44..992b8f503c 100644 --- a/src/handler/http/request/metrics/ingest.rs +++ b/src/handler/http/request/metrics/ingest.rs @@ -80,7 +80,7 @@ pub async fn json( return MetaHttpResponse::too_many_requests(e); } - let mut resp = match metrics::json::ingest(&org_id, body, user).await { + let mut resp = match metrics::json::ingest(&org_id, None, body, user).await { Ok(v) => { if v.code == StatusCode::OK.as_u16() { MetaHttpResponse::json(v) diff --git a/src/job/promql_self_consume.rs b/src/job/promql_self_consume.rs index 51e991e315..a8b23f4b71 100644 --- a/src/job/promql_self_consume.rs +++ b/src/job/promql_self_consume.rs @@ -118,6 +118,7 @@ pub async fn run() -> Result<(), anyhow::Error> { let bytes = bytes::Bytes::from(metrics); match service::metrics::json::ingest( org, + None, bytes, IngestUser::SystemJob(SystemJobType::SelfMetricsPromql), ) diff --git a/src/service/metrics/json.rs b/src/service/metrics/json.rs index 0dc66cee01..89ac34b621 100644 --- a/src/service/metrics/json.rs +++ b/src/service/metrics/json.rs @@ -61,13 +61,16 @@ use crate::{ }, }; +const VALID_METRICS_TYPES: &[&str] = &["counter", "gauge", "histogram", "summary"]; + pub async fn ingest( org_id: &str, + stream_name: Option<&str>, body: Bytes, user: crate::common::meta::ingestion::IngestUser, ) -> Result { // check system resource - if let Err(e) = check_ingestion_allowed(org_id, StreamType::Metrics, None).await { + if let Err(e) = check_ingestion_allowed(org_id, StreamType::Metrics, stream_name).await { // we do not want to log trial period expired errors if matches!(e, infra::errors::Error::TrialPeriodExpired) { return Ok(IngestionResponse { @@ -117,18 +120,20 @@ pub async fn ingest( let mut record = flatten::flatten(record)?; // check data type let record = record.as_object_mut().unwrap(); - let stream_name = match record.get(NAME_LABEL).ok_or(anyhow!("missing __name__"))? { - json::Value::String(s) => format_stream_name(s.to_string()), - _ => { - return Err(anyhow::anyhow!("invalid __name__, need to be string")); - } - }; - let metrics_type = match record.get(TYPE_LABEL).ok_or(anyhow!("missing __type__"))? { - json::Value::String(s) => s.clone(), - _ => { - return Err(anyhow::anyhow!("invalid __type__, need to be string")); - } + let stream_name = match stream_name { + Some(name) => name.to_string(), + None => match record.get(NAME_LABEL).ok_or(anyhow!("missing __name__"))? { + json::Value::String(s) => format_stream_name(s.to_string()), + _ => { + return Err(anyhow::anyhow!("invalid __name__, need to be string")); + } + }, }; + let metrics_type = record + .get(TYPE_LABEL) + .and_then(|v| v.as_str()) + .unwrap_or("gauge") // default to gauge if __type__ is missing + .to_string(); // Start retrieve associated pipeline and initialize ExecutablePipeline let stream_param = StreamParams::new(org_id, &stream_name, StreamType::Metrics); @@ -148,37 +153,41 @@ pub async fn ingest( ) .await; - // check metrics type for Histogram & Summary - if metrics_type.to_lowercase() == "histogram" || metrics_type.to_lowercase() == "summary" { - if !stream_schema_map.contains_key(&stream_name) { - let mut schema = - infra::schema::get(org_id, &stream_name, StreamType::Metrics).await?; - if schema == Schema::empty() { - // create the metadata for the stream - let metadata = Metadata { - metric_family_name: stream_name.clone(), - metric_type: metrics_type.as_str().into(), - help: stream_name.clone().replace('_', " "), - unit: "".to_string(), - }; - let mut extra_metadata: HashMap = HashMap::new(); - extra_metadata.insert( - METADATA_LABEL.to_string(), - json::to_string(&metadata).unwrap(), - ); - schema = schema.with_metadata(extra_metadata); - db::schema::merge( - org_id, - &stream_name, - StreamType::Metrics, - &schema, - Some(now_micros()), - ) - .await?; - } - stream_schema_map.insert(stream_name.clone(), SchemaCache::new(schema)); + // check metrics type + if !VALID_METRICS_TYPES.contains(&metrics_type.to_lowercase().as_str()) { + return Err(anyhow::anyhow!( + "invalid metrics type, need to be one of: {}", + VALID_METRICS_TYPES.join(", ") + )); + } + + // check schema + if !stream_schema_map.contains_key(&stream_name) { + let mut schema = infra::schema::get(org_id, &stream_name, StreamType::Metrics).await?; + if schema == Schema::empty() { + // create the metadata for the stream + let metadata = Metadata { + metric_family_name: stream_name.clone(), + metric_type: metrics_type.as_str().into(), + help: stream_name.clone().replace('_', " "), + unit: "".to_string(), + }; + let mut extra_metadata: HashMap = HashMap::new(); + extra_metadata.insert( + METADATA_LABEL.to_string(), + json::to_string(&metadata).unwrap(), + ); + schema = schema.with_metadata(extra_metadata); + db::schema::merge( + org_id, + &stream_name, + StreamType::Metrics, + &schema, + Some(now_micros()), + ) + .await?; } - continue; + stream_schema_map.insert(stream_name.clone(), SchemaCache::new(schema)); } // check timestamp