diff --git a/Cargo.lock b/Cargo.lock index ec340490f6..da75fd28d4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4228,7 +4228,7 @@ dependencies = [ "once_cell", "opentelemetry 0.18.0", "opentelemetry-otlp", - "opentelemetry-proto", + "opentelemetry-proto 0.2.0", "parking_lot 0.12.1", "parquet", "prometheus", @@ -4292,6 +4292,16 @@ dependencies = [ "opentelemetry_sdk 0.18.0", ] +[[package]] +name = "opentelemetry" +version = "0.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5f4b8347cc26099d3aeee044065ecc3ae11469796b4d65d065a23a584ed92a6f" +dependencies = [ + "opentelemetry_api 0.19.0", + "opentelemetry_sdk 0.19.0", +] + [[package]] name = "opentelemetry" version = "0.20.0" @@ -4327,7 +4337,7 @@ dependencies = [ "http", "opentelemetry 0.18.0", "opentelemetry-http", - "opentelemetry-proto", + "opentelemetry-proto 0.1.0", "prost", "reqwest", "serde", @@ -4346,11 +4356,24 @@ dependencies = [ "futures-util", "opentelemetry 0.18.0", "prost", + "tonic 0.8.3", + "tonic-build 0.8.4", +] + +[[package]] +name = "opentelemetry-proto" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "045f8eea8c0fa19f7d48e7bc3128a39c2e5c533d5c61298c548dfefc1064474c" +dependencies = [ + "futures", + "futures-util", + "opentelemetry 0.19.0", + "prost", "protobuf", "serde", "serde_json", "tonic 0.8.3", - "tonic-build 0.8.4", ] [[package]] @@ -4378,6 +4401,22 @@ dependencies = [ "thiserror", ] +[[package]] +name = "opentelemetry_api" +version = "0.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed41783a5bf567688eb38372f2b7a8530f5a607a4b49d38dd7573236c23ca7e2" +dependencies = [ + "fnv", + "futures-channel", + "futures-util", + "indexmap 1.9.3", + "once_cell", + "pin-project-lite", + "thiserror", + "urlencoding", +] + [[package]] name = "opentelemetry_api" version = "0.20.0" @@ -4416,6 +4455,26 @@ dependencies = [ "tokio-stream", ] +[[package]] +name = "opentelemetry_sdk" +version = "0.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b3a2a91fdbfdd4d212c0dcc2ab540de2c2bcbbd90be17de7a7daf8822d010c1" +dependencies = [ + "async-trait", + "crossbeam-channel", + "dashmap", + "fnv", + "futures-channel", + "futures-executor", + "futures-util", + "once_cell", + "opentelemetry_api 0.19.0", + "percent-encoding", + "rand", + "thiserror", +] + [[package]] name = "opentelemetry_sdk" version = "0.20.0" diff --git a/Cargo.toml b/Cargo.toml index 95f178177d..c9aeb5a6b0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -105,13 +105,12 @@ opentelemetry-otlp = { version = "0.11", features = [ "serialize", "reqwest-client", ] } -opentelemetry-proto = { version = "0.1", features = [ +opentelemetry-proto = { version = "0.2", features = [ "gen-tonic", "traces", "logs", "metrics", "with-serde", - "build-server", ] } parking_lot = "0.12" prometheus = "0.13" diff --git a/src/common/meta/usage.rs b/src/common/meta/usage.rs index d4e536cc7f..940c868eac 100644 --- a/src/common/meta/usage.rs +++ b/src/common/meta/usage.rs @@ -91,7 +91,8 @@ impl From for UsageEvent { | UsageType::GCPSubscription | UsageType::EnrichmentTable | UsageType::Syslog - | UsageType::JsonMetrics => UsageEvent::Ingestion, + | UsageType::JsonMetrics + | UsageType::Logs => UsageEvent::Ingestion, UsageType::Search | UsageType::SearchAround | UsageType::SearchTopNValues @@ -110,6 +111,8 @@ pub enum UsageType { Json, #[serde(rename = "logs/_multi")] Multi, + #[serde(rename = "/v1/logs")] + Logs, #[serde(rename = "/traces")] Traces, #[serde(rename = "/v1/write")] @@ -155,6 +158,7 @@ impl ToString for UsageType { UsageType::SearchTopNValues => "/_values".to_owned(), UsageType::GCPSubscription => "/gcp/_sub".to_owned(), UsageType::MetricSearch => "/metrics/_search".to_owned(), + UsageType::Logs => "/v1/logs".to_owned(), } } } diff --git a/src/handler/grpc/request/logs.rs b/src/handler/grpc/request/logs.rs index 74f0ee9700..ee403df34b 100644 --- a/src/handler/grpc/request/logs.rs +++ b/src/handler/grpc/request/logs.rs @@ -45,14 +45,17 @@ impl LogsService for LogsServer { return Err(Status::invalid_argument(msg)); } - let resp = crate::service::logs::json_no_fn::handle_grpc_request( + let resp = crate::service::logs::otlp_grpc::handle_grpc_request( org_id.unwrap().to_str().unwrap(), 0, in_req, + true, ) .await; if resp.is_ok() { - return Ok(Response::new(ExportLogsServiceResponse {})); + return Ok(Response::new(ExportLogsServiceResponse { + partial_success: None, + })); } else { Err(Status::internal(resp.err().unwrap().to_string())) } diff --git a/src/handler/grpc/request/metrics/ingester.rs b/src/handler/grpc/request/metrics/ingester.rs index 3f1fe5488b..b33ff1a455 100644 --- a/src/handler/grpc/request/metrics/ingester.rs +++ b/src/handler/grpc/request/metrics/ingester.rs @@ -45,14 +45,17 @@ impl MetricsService for Ingester { return Err(Status::invalid_argument(msg)); } - let resp = crate::service::metrics::otlp::handle_grpc_request( + let resp = crate::service::metrics::otlp_grpc::handle_grpc_request( org_id.unwrap().to_str().unwrap(), 0, in_req, + true, ) .await; if resp.is_ok() { - return Ok(Response::new(ExportMetricsServiceResponse {})); + return Ok(Response::new(ExportMetricsServiceResponse { + partial_success: None, + })); } else { Err(Status::internal(resp.err().unwrap().to_string())) } diff --git a/src/handler/grpc/request/traces.rs b/src/handler/grpc/request/traces.rs index 9f6ace8cf9..5cce37a116 100644 --- a/src/handler/grpc/request/traces.rs +++ b/src/handler/grpc/request/traces.rs @@ -12,8 +12,9 @@ // See the License for the specific language governing permissions and // limitations under the License. +use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest; use opentelemetry_proto::tonic::collector::trace::v1::{ - trace_service_server::TraceService, ExportTraceServiceRequest, ExportTraceServiceResponse, + trace_service_server::TraceService, ExportTraceServiceResponse, }; use tonic::Status; use tonic::{codegen::*, Response}; @@ -46,7 +47,9 @@ impl TraceService for TraceServer { let resp = handle_trace_request(org_id.unwrap().to_str().unwrap(), 0, in_req, true).await; if resp.is_ok() { - return Ok(Response::new(ExportTraceServiceResponse {})); + return Ok(Response::new(ExportTraceServiceResponse { + partial_success: None, + })); } else { Err(Status::internal(resp.err().unwrap().to_string())) } diff --git a/src/handler/grpc/request/usage.rs b/src/handler/grpc/request/usage.rs index 9eb0ed91bf..9f70c42aa1 100644 --- a/src/handler/grpc/request/usage.rs +++ b/src/handler/grpc/request/usage.rs @@ -33,7 +33,7 @@ impl Usage for UsageServerImpl { let report_to_org_id = metadata.get(&CONFIG.grpc.org_header_key); let in_data = req.data.unwrap_or_default(); log::info!("UsageServer: report_usage received data"); - let resp = crate::service::logs::json_no_fn::ingest( + let resp = crate::service::logs::otlp_grpc::ingest( report_to_org_id.unwrap().to_str().unwrap(), &report_to_stream, in_data.data.into(), diff --git a/src/handler/http/request/logs/ingest.rs b/src/handler/http/request/logs/ingest.rs index db7ac236c5..1dee093ac4 100644 --- a/src/handler/http/request/logs/ingest.rs +++ b/src/handler/http/request/logs/ingest.rs @@ -12,10 +12,12 @@ // See the License for the specific language governing permissions and // limitations under the License. -use actix_web::{http, post, web, HttpResponse}; +use actix_web::{http, post, web, HttpRequest, HttpResponse}; use std::io::Error; use crate::common::meta::http::HttpResponse as MetaHttpResponse; +use crate::handler::http::request::{CONTENT_TYPE_JSON, CONTENT_TYPE_PROTO}; +use crate::service::logs::otlp_http::{logs_json_handler, logs_proto_handler}; use crate::{ common::meta::ingestion::{GCPIngestionRequest, KinesisFHRequest}, service::logs, @@ -200,3 +202,37 @@ pub async fn handle_gcp_request( }, ) } + +/** LogsIngest */ +#[utoipa::path( + context_path = "/api", + tag = "Logs", + operation_id = "PostLogs", + request_body(content = String, description = "ExportLogsServiceRequest", content_type = "application/x-protobuf"), + responses( + (status = 200, description="Success", content_type = "application/json", body = IngestionResponse, example = json!({"code": 200})), + (status = 500, description="Failure", content_type = "application/json", body = HttpResponse), + ) +)] +#[post("/{org_id}/v1/logs")] +pub async fn otlp_logs_write( + org_id: web::Path, + thread_id: web::Data, + req: HttpRequest, + body: web::Bytes, +) -> Result { + let org_id = org_id.into_inner(); + let content_type = req.headers().get("Content-Type").unwrap().to_str().unwrap(); + if content_type.eq(CONTENT_TYPE_PROTO) { + log::info!("otlp::logs_proto_handler"); + logs_proto_handler(&org_id, **thread_id, body).await + } else if content_type.starts_with(CONTENT_TYPE_JSON) { + log::info!("otlp::logs_json_handler"); + logs_json_handler(&org_id, **thread_id, body).await + } else { + Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( + http::StatusCode::BAD_REQUEST.into(), + "Bad Request".to_string(), + ))) + } +} diff --git a/src/handler/http/request/metrics/ingest.rs b/src/handler/http/request/metrics/ingest.rs index 576a5966db..0e7f87e648 100644 --- a/src/handler/http/request/metrics/ingest.rs +++ b/src/handler/http/request/metrics/ingest.rs @@ -12,11 +12,13 @@ // See the License for the specific language governing permissions and // limitations under the License. -use actix_web::{http, post, web, HttpResponse}; +use actix_web::{http, post, web, HttpRequest, HttpResponse}; use std::io::Error; use crate::common::meta::http::HttpResponse as MetaHttpResponse; -use crate::service::metrics; +use crate::handler::http::request::{CONTENT_TYPE_JSON, CONTENT_TYPE_PROTO}; +use crate::service::metrics::otlp_http::{metrics_json_handler, metrics_proto_handler}; +use crate::service::metrics::{self}; /** _json ingestion API */ #[utoipa::path( @@ -52,3 +54,37 @@ pub async fn json( }, ) } + +/** MetricsIngest */ +#[utoipa::path( + context_path = "/api", + tag = "Metrics", + operation_id = "PostMetrics", + request_body(content = String, description = "ExportMetricsServiceRequest", content_type = "application/x-protobuf"), + responses( + (status = 200, description="Success", content_type = "application/json", body = IngestionResponse, example = json!({"code": 200})), + (status = 500, description="Failure", content_type = "application/json", body = HttpResponse), + ) +)] +#[post("/{org_id}/v1/metrics")] +pub async fn otlp_metrics_write( + org_id: web::Path, + thread_id: web::Data, + req: HttpRequest, + body: web::Bytes, +) -> Result { + let org_id = org_id.into_inner(); + let content_type = req.headers().get("Content-Type").unwrap().to_str().unwrap(); + if content_type.eq(CONTENT_TYPE_PROTO) { + log::info!("otlp::metrics_proto_handler"); + metrics_proto_handler(&org_id, **thread_id, body).await + } else if content_type.starts_with(CONTENT_TYPE_JSON) { + log::info!("otlp::metrics_json_handler"); + metrics_json_handler(&org_id, **thread_id, body).await + } else { + Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( + http::StatusCode::BAD_REQUEST.into(), + "Bad Request".to_string(), + ))) + } +} diff --git a/src/handler/http/request/mod.rs b/src/handler/http/request/mod.rs index 75df8cd343..c81523f4e3 100644 --- a/src/handler/http/request/mod.rs +++ b/src/handler/http/request/mod.rs @@ -27,3 +27,6 @@ pub mod stream; pub mod syslog; pub mod traces; pub mod users; + +pub const CONTENT_TYPE_JSON: &str = "application/json"; +pub const CONTENT_TYPE_PROTO: &str = "application/x-protobuf"; diff --git a/src/handler/http/request/traces/mod.rs b/src/handler/http/request/traces/mod.rs index c3b38d3408..7770269b02 100644 --- a/src/handler/http/request/traces/mod.rs +++ b/src/handler/http/request/traces/mod.rs @@ -15,10 +15,11 @@ use actix_web::{http, post, web, HttpRequest, HttpResponse}; use std::io::Error; -use crate::{common::meta, service::traces::otlp_http}; - -pub const CONTENT_TYPE_JSON: &str = "application/json"; -pub const CONTENT_TYPE_PROTO: &str = "application/x-protobuf"; +use crate::{ + common::meta, + handler::http::request::{CONTENT_TYPE_JSON, CONTENT_TYPE_PROTO}, + service::traces::otlp_http, +}; /** TracesIngest */ #[utoipa::path( @@ -53,3 +54,25 @@ pub async fn traces_write( ) } } +#[post("/{org_id}/v1/traces")] +pub async fn otlp_traces_write( + org_id: web::Path, + thread_id: web::Data, + req: HttpRequest, + body: web::Bytes, +) -> Result { + let org_id = org_id.into_inner(); + let content_type = req.headers().get("Content-Type").unwrap().to_str().unwrap(); + if content_type.eq(CONTENT_TYPE_PROTO) { + otlp_http::traces_proto(&org_id, **thread_id, body).await + } else if content_type.starts_with(CONTENT_TYPE_JSON) { + otlp_http::traces_json(&org_id, **thread_id, body).await + } else { + Ok( + HttpResponse::BadRequest().json(meta::http::HttpResponse::error( + http::StatusCode::BAD_REQUEST.into(), + "Bad Request".to_string(), + )), + ) + } +} diff --git a/src/handler/http/router/mod.rs b/src/handler/http/router/mod.rs index 584932dfa0..8d987c3d14 100644 --- a/src/handler/http/router/mod.rs +++ b/src/handler/http/router/mod.rs @@ -28,7 +28,7 @@ use utoipa_swagger_ui::SwaggerUi; use super::auth::{validator, validator_aws, validator_gcp}; use super::request::{ alerts::*, dashboards::*, enrichment_table, functions, kv, logs, metrics, organization, prom, - search, status, stream, syslog, traces::*, users, + search, status, stream, syslog, traces, users, }; use crate::common::infra::config::CONFIG; @@ -161,7 +161,7 @@ pub fn get_service_routes(cfg: &mut web::ServiceConfig) { .service(list_dashboards) .service(get_dashboard) .service(delete_dashboard) - .service(traces_write) + .service(traces::traces_write) .service(save_alert) .service(get_alert) .service(list_alerts) @@ -197,7 +197,10 @@ pub fn get_service_routes(cfg: &mut web::ServiceConfig) { .service(syslog::delete_route) .service(syslog::update_route) .service(syslog::toggle_state) - .service(enrichment_table::save_enrichment_table), + .service(enrichment_table::save_enrichment_table) + .service(metrics::ingest::otlp_metrics_write) + .service(logs::ingest::otlp_logs_write) + .service(traces::otlp_traces_write), ); } diff --git a/src/service/ingestion/grpc.rs b/src/service/ingestion/grpc.rs index 134a234dde..9ddd666262 100644 --- a/src/service/ingestion/grpc.rs +++ b/src/service/ingestion/grpc.rs @@ -18,6 +18,8 @@ use opentelemetry_proto::tonic::{ metrics::v1::{exemplar, number_data_point}, }; +use super::get_value; + pub fn get_val(attr_val: &Option) -> json::Value { match attr_val { Some(local_val) => match &local_val.value { @@ -112,6 +114,14 @@ pub fn get_exemplar_val(attr_val: &Option) -> json::Value { } } +pub fn get_val_for_attr(attr_val: json::Value) -> json::Value { + let local_val = attr_val.as_object().unwrap(); + if let Some((_key, value)) = local_val.into_iter().next() { + return serde_json::Value::String(get_value(value)); + }; + ().into() +} + #[cfg(test)] mod tests { use opentelemetry_proto::tonic::common::v1::AnyValue; diff --git a/src/service/ingestion/mod.rs b/src/service/ingestion/mod.rs index 9c609c3587..2f079d014c 100644 --- a/src/service/ingestion/mod.rs +++ b/src/service/ingestion/mod.rs @@ -330,6 +330,22 @@ pub fn write_file( req_stats } +pub fn get_value(value: &Value) -> String { + if value.is_boolean() { + value.as_bool().unwrap().to_string() + } else if value.is_f64() { + value.as_f64().unwrap().to_string() + } else if value.is_i64() { + value.as_i64().unwrap().to_string() + } else if value.is_u64() { + value.as_u64().unwrap().to_string() + } else if value.is_string() { + value.as_str().unwrap().to_string() + } else { + "".to_string() + } +} + #[cfg(test)] mod tests { use std::collections::HashMap; diff --git a/src/service/logs/mod.rs b/src/service/logs/mod.rs index f5a984788d..7995c37fa7 100644 --- a/src/service/logs/mod.rs +++ b/src/service/logs/mod.rs @@ -32,14 +32,15 @@ use crate::common::{ }; use crate::service::schema::check_for_schema; -use super::ingestion::get_wal_time_key; +use super::ingestion::{get_wal_time_key, get_value}; pub mod bulk; pub mod gcs_pub_sub; pub mod json; -pub mod json_no_fn; pub mod kinesis_firehose; pub mod multi; +pub mod otlp_grpc; +pub mod otlp_http; pub mod syslog; static BULK_OPERATORS: [&str; 3] = ["create", "index", "update"]; @@ -198,22 +199,6 @@ pub fn cast_to_type(mut value: Value, delta: Vec) -> (Option, Opt } } -pub fn get_value(value: &Value) -> String { - if value.is_boolean() { - value.as_bool().unwrap().to_string() - } else if value.is_f64() { - value.as_f64().unwrap().to_string() - } else if value.is_i64() { - value.as_i64().unwrap().to_string() - } else if value.is_u64() { - value.as_u64().unwrap().to_string() - } else if value.is_string() { - value.as_str().unwrap().to_string() - } else { - "".to_string() - } -} - async fn add_valid_record( stream_meta: StreamMeta, stream_schema_map: &mut AHashMap, diff --git a/src/service/logs/json_no_fn.rs b/src/service/logs/otlp_grpc.rs similarity index 94% rename from src/service/logs/json_no_fn.rs rename to src/service/logs/otlp_grpc.rs index bf52ce5f1b..624f6fb228 100644 --- a/src/service/logs/json_no_fn.rs +++ b/src/service/logs/otlp_grpc.rs @@ -204,6 +204,7 @@ pub async fn handle_grpc_request( org_id: &str, thread_id: usize, request: ExportLogsServiceRequest, + is_grpc: bool, ) -> Result { if !cluster::is_ingester(&cluster::LOCAL_NODE_ROLE) { return Ok( @@ -217,7 +218,7 @@ pub async fn handle_grpc_request( if !db::file_list::BLOCKED_ORGS.is_empty() && db::file_list::BLOCKED_ORGS.contains(&org_id) { return Ok(HttpResponse::Forbidden().json(MetaHttpResponse::error( http::StatusCode::FORBIDDEN.into(), - "Quota exceeded for this organisation".to_string(), + "Quota exceeded for this organization".to_string(), ))); } let start = std::time::Instant::now(); @@ -253,7 +254,7 @@ pub async fn handle_grpc_request( let mut data_buf: AHashMap> = AHashMap::new(); for resource_log in &request.resource_logs { - for instrumentation_logs in &resource_log.instrumentation_library_logs { + for instrumentation_logs in &resource_log.scope_logs { for log_record in &instrumentation_logs.log_records { let mut rec = json::json!({}); @@ -265,7 +266,7 @@ pub async fn handle_grpc_request( } None => {} } - match &instrumentation_logs.instrumentation_library { + match &instrumentation_logs.scope { Some(lib) => { rec["instrumentation_library_name"] = serde_json::Value::String(lib.name.to_owned()); @@ -370,10 +371,16 @@ pub async fn handle_grpc_request( // only one trigger per request, as it updates etcd super::evaluate_trigger(trigger, stream_alerts_map).await; + let ep = if is_grpc { + "grpc/export/logs" + } else { + "/api/org/v1/logs" + }; + let time = start.elapsed().as_secs_f64(); metrics::HTTP_RESPONSE_TIME .with_label_values(&[ - "/api/org/ingest/logs/_json", + ep, "200", org_id, stream_name, @@ -382,7 +389,7 @@ pub async fn handle_grpc_request( .observe(time); metrics::HTTP_INCOMING_REQUESTS .with_label_values(&[ - "/api/org/ingest/logs/_json", + ep, "200", org_id, stream_name, @@ -401,7 +408,9 @@ pub async fn handle_grpc_request( 0, ) .await; - let res = ExportLogsServiceResponse {}; + let res = ExportLogsServiceResponse { + partial_success: None, + }; let mut out = BytesMut::with_capacity(res.encoded_len()); res.encode(&mut out).expect("Out of memory"); @@ -416,12 +425,12 @@ pub mod test { use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest; use opentelemetry_proto::tonic::common::v1::any_value::Value::{IntValue, StringValue}; - use opentelemetry_proto::tonic::common::v1::InstrumentationLibrary; - use opentelemetry_proto::tonic::common::v1::KeyValue; - use opentelemetry_proto::tonic::logs::v1::InstrumentationLibraryLogs; - use opentelemetry_proto::tonic::logs::v1::ResourceLogs; + use opentelemetry_proto::tonic::common::v1::{InstrumentationScope, KeyValue}; + use opentelemetry_proto::tonic::logs::v1::{ResourceLogs, ScopeLogs}; use opentelemetry_proto::tonic::{common::v1::AnyValue, logs::v1::LogRecord}; + use crate::service::logs::otlp_grpc::handle_grpc_request; + #[tokio::test] async fn test_handle_logs_request() { let org_id = "test_org_id"; @@ -455,17 +464,19 @@ pub mod test { ..Default::default() }; - let ins = InstrumentationLibraryLogs { - instrumentation_library: Some(InstrumentationLibrary { + let ins = ScopeLogs { + scope: Some(InstrumentationScope { name: "test".to_string(), version: "1.0.0".to_string(), + attributes: vec![], + dropped_attributes_count: 0, }), log_records: vec![log_rec], ..Default::default() }; let res_logs = ResourceLogs { - instrumentation_library_logs: vec![ins], + scope_logs: vec![ins], ..Default::default() }; @@ -473,7 +484,7 @@ pub mod test { resource_logs: vec![res_logs], }; - let result = super::handle_grpc_request(org_id, thread_id, request).await; + let result = handle_grpc_request(org_id, thread_id, request, true).await; assert!(result.is_ok()); } } diff --git a/src/service/logs/otlp_http.rs b/src/service/logs/otlp_http.rs new file mode 100644 index 0000000000..242b5dd124 --- /dev/null +++ b/src/service/logs/otlp_http.rs @@ -0,0 +1,359 @@ +// Copyright 2023 Zinc Labs Inc. +// +// 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. + +use actix_web::{http, web, HttpResponse}; +use ahash::AHashMap; +use arrow_schema::Schema; +use bytes::BytesMut; +use chrono::{Duration, Utc}; +use opentelemetry_proto::tonic::collector::logs::v1::{ + ExportLogsPartialSuccess, ExportLogsServiceRequest, ExportLogsServiceResponse, +}; +use prost::Message; + +use crate::{ + common::{ + infra::{cluster, config::CONFIG, metrics}, + meta::{ + alert::{Alert, Trigger}, + http::HttpResponse as MetaHttpResponse, + ingestion::StreamStatus, + stream::StreamParams, + usage::UsageType, + StreamType, + }, + utils::{flatten, json}, + }, + handler::http::request::CONTENT_TYPE_JSON, + service::{ + db, + ingestion::{grpc::get_val_for_attr, write_file}, + schema::stream_schema_exists, + usage::report_request_usage_stats, + }, +}; + +use super::StreamMeta; + +const SERVICE_NAME: &str = "service.name"; +const SERVICE: &str = "service"; + +pub async fn logs_proto_handler( + org_id: &str, + thread_id: usize, + body: web::Bytes, +) -> Result { + let request = ExportLogsServiceRequest::decode(body).expect("Invalid protobuf"); + match super::otlp_grpc::handle_grpc_request(org_id, thread_id, request, false).await { + Ok(res) => Ok(res), + Err(e) => { + log::error!("error while handling request: {}", e); + Ok( + HttpResponse::InternalServerError().json(MetaHttpResponse::error( + http::StatusCode::INTERNAL_SERVER_ERROR.into(), + e.to_string(), + )), + ) + } + } +} + +//example at: https://opentelemetry.io/docs/specs/otel/protocol/file-exporter/#examples +pub async fn logs_json_handler( + org_id: &str, + thread_id: usize, + body: web::Bytes, +) -> Result { + if !cluster::is_ingester(&cluster::LOCAL_NODE_ROLE) { + return Ok( + HttpResponse::InternalServerError().json(MetaHttpResponse::error( + http::StatusCode::INTERNAL_SERVER_ERROR.into(), + "not an ingester".to_string(), + )), + ); + } + + if !db::file_list::BLOCKED_ORGS.is_empty() && db::file_list::BLOCKED_ORGS.contains(&org_id) { + return Ok(HttpResponse::Forbidden().json(MetaHttpResponse::error( + http::StatusCode::FORBIDDEN.into(), + "Quota exceeded for this organization".to_string(), + ))); + } + + let start = std::time::Instant::now(); + let stream_name = "default"; + let mut stream_schema_map: AHashMap = AHashMap::new(); + let mut stream_alerts_map: AHashMap> = AHashMap::new(); + let mut stream_status = StreamStatus::new(stream_name); + let mut trigger: Option = None; + + let mut min_ts = + (Utc::now() + Duration::hours(CONFIG.limit.ingest_allowed_upto)).timestamp_micros(); + + let stream_schema = stream_schema_exists( + org_id, + stream_name, + StreamType::Logs, + &mut stream_schema_map, + ) + .await; + + let mut partition_keys: Vec = vec![]; + if stream_schema.has_partition_keys { + let partition_det = + crate::service::ingestion::get_stream_partition_keys(stream_name, &stream_schema_map) + .await; + partition_keys = partition_det.partition_keys; + } + + // Start get stream alerts + let key = format!("{}/{}/{}", &org_id, StreamType::Logs, &stream_name); + crate::service::ingestion::get_stream_alerts(key, &mut stream_alerts_map).await; + // End get stream alert + + let mut buf: AHashMap> = AHashMap::new(); + + let body: json::Value = match json::from_slice(body.as_ref()) { + Ok(v) => v, + Err(e) => { + return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( + http::StatusCode::BAD_REQUEST.into(), + format!("Invalid json: {}", e), + ))) + } + }; + + let logs = match body.get("resourceLogs") { + Some(v) => match v.as_array() { + Some(v) => v, + None => { + return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( + http::StatusCode::BAD_REQUEST.into(), + format!("Invalid json: the structure must be {{\"resourceLogs\":[]}}"), + ))) + } + }, + None => { + return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( + http::StatusCode::BAD_REQUEST.into(), + format!("Invalid json: the structure must be {{\"resourceLogs\":[]}}"), + ))) + } + }; + + for res_log in logs.iter() { + let mut service_att_map: json::Map = json::Map::new(); + if res_log.get("resource").is_some() { + let resource = res_log.get("resource").unwrap().as_object().unwrap(); + if resource.get("attributes").is_some() { + let attributes = resource.get("attributes").unwrap().as_array().unwrap(); + for res_attr in attributes { + let local_attr = res_attr.as_object().unwrap(); + if local_attr + .get("key") + .unwrap() + .as_str() + .unwrap() + .eq(SERVICE_NAME) + { + let loc_service_name = + local_attr.get("value").unwrap().as_object().unwrap(); + for item in loc_service_name { + service_att_map.insert(SERVICE_NAME.to_string(), item.1.clone()); + } + } else { + service_att_map.insert( + format!( + "{}_{}", + SERVICE, + local_attr.get("key").unwrap().as_str().unwrap() + ), + get_val_for_attr(local_attr.get("value").unwrap().clone()), + ); + } + } + } + } + let scope_resources = res_log.get("scopeLogs"); + let inst_resources = if scope_resources.is_some() { + scope_resources.unwrap().as_array().unwrap() + } else { + res_log + .get("instrumentationLibrarySpans") + .unwrap() + .as_array() + .unwrap() + }; + for inst_log in inst_resources { + if inst_log.get("logRecords").is_some() { + let log_records = inst_log.get("logRecords").unwrap().as_array().unwrap(); + for log in log_records { + let start_time: i64 = match log.get("timeUnixNano").unwrap() { + json::Value::Number(v) => v.as_u64().unwrap() as i64, + json::Value::String(v) => v.parse::().unwrap(), + _ => 0, + }; + + let timestamp = if start_time > 0 { + start_time / 1000 + } else { + Utc::now().timestamp_micros() + }; + + let mut value: json::Value = json::to_value(log).unwrap(); + + //JSON Flattening + value = flatten::flatten(&value).unwrap(); + + // get json object + let local_val = value.as_object_mut().unwrap(); + + if log.get("attributes").is_some() { + let attributes = log.get("attributes").unwrap().as_array().unwrap(); + for res_attr in attributes { + let local_attr = res_attr.as_object().unwrap(); + + local_val.insert( + local_attr.get("key").unwrap().as_str().unwrap().to_owned(), + get_val_for_attr(local_attr.get("value").unwrap().clone()), + ); + } + } + //remove attributes after adding + local_val.remove("attributes"); + + //remove body before adding + local_val.remove("body_stringvalue"); + + if log.get("body").is_some() { + let body = log.get("body").unwrap().get("stringValue").unwrap(); + local_val.insert("body".to_owned(), body.clone()); + } + + // check ingestion time + let earliest_time = + Utc::now() + Duration::hours(0 - CONFIG.limit.ingest_allowed_upto); + if timestamp < earliest_time.timestamp_micros() { + stream_status.status.failed += 1; // to old data, just discard + stream_status.status.error = super::get_upto_discard_error(); + continue; + } + if timestamp < min_ts { + min_ts = timestamp; + } + + local_val.insert( + CONFIG.common.column_timestamp.clone(), + json::Value::Number(timestamp.into()), + ); + + local_val.append(&mut service_att_map.clone()); + + /* local_val.insert( + "service_attributes".to_owned(), + json::Value::Object(service_att_map.clone()), + ); */ + + let local_trigger = super::add_valid_record( + StreamMeta { + org_id: org_id.to_string(), + stream_name: stream_name.to_string(), + partition_keys: partition_keys.clone(), + stream_alerts_map: stream_alerts_map.clone(), + }, + &mut stream_schema_map, + &mut stream_status.status, + &mut buf, + local_val, + ) + .await; + + if local_trigger.is_some() { + trigger = Some(local_trigger.unwrap()); + } + } + } + } + } + + // write to file + let mut stream_file_name = "".to_string(); + let mut req_stats = write_file( + buf, + thread_id, + StreamParams { + org_id, + stream_name, + stream_type: StreamType::Logs, + }, + &mut stream_file_name, + None, + ); + + /* if stream_file_name.is_empty() { + return Ok(IngestionResponse::new( + http::StatusCode::OK.into(), + vec![stream_status], + )); + } */ + + // only one trigger per request, as it updates etcd + super::evaluate_trigger(trigger, stream_alerts_map).await; + + let time = start.elapsed().as_secs_f64(); + metrics::HTTP_RESPONSE_TIME + .with_label_values(&[ + "/api/org/v1/logs", + "200", + org_id, + stream_name, + StreamType::Logs.to_string().as_str(), + ]) + .observe(time); + metrics::HTTP_INCOMING_REQUESTS + .with_label_values(&[ + "/api/org/v1/logs", + "200", + org_id, + stream_name, + StreamType::Logs.to_string().as_str(), + ]) + .inc(); + + req_stats.response_time = start.elapsed().as_secs_f64(); + //metric + data usage + report_request_usage_stats( + req_stats, + org_id, + stream_name, + StreamType::Logs, + UsageType::Json, + 0, + ) + .await; + + let res = ExportLogsServiceResponse { + partial_success: Some(ExportLogsPartialSuccess { + rejected_log_records: stream_status.status.failed as i64, + error_message: stream_status.status.error, + }), + }; + let mut out = BytesMut::with_capacity(res.encoded_len()); + res.encode(&mut out).expect("Out of memory"); + + return Ok(HttpResponse::Ok() + .status(http::StatusCode::OK) + .content_type(CONTENT_TYPE_JSON) + .body(out)); +} diff --git a/src/service/metrics/mod.rs b/src/service/metrics/mod.rs index 560c047ea5..e74f4999af 100644 --- a/src/service/metrics/mod.rs +++ b/src/service/metrics/mod.rs @@ -15,12 +15,16 @@ use datafusion::arrow::datatypes::Schema; use crate::common; -use crate::common::meta::prom::{Metadata, METADATA_LABEL}; +use crate::common::infra::config::CONFIG; +use crate::common::meta::prom::{Metadata, METADATA_LABEL, VALUE_LABEL}; pub mod json; -pub mod otlp; +pub mod otlp_grpc; +pub mod otlp_http; pub mod prom; +const EXCLUDE_LABELS: [&str; 5] = [VALUE_LABEL, "start_time", "is_monotonic", "exemplars", "le"]; + pub fn get_prom_metadata_from_schema(schema: &Schema) -> Option { let metadata = schema.metadata.get(METADATA_LABEL)?; let metadata: Metadata = common::utils::json::from_str(metadata).unwrap(); @@ -56,3 +60,9 @@ pub fn signature_without_labels( }); Signature(hasher.finalize().into()) } + +fn get_exclude_labels() -> Vec<&'static str> { + let mut vec: Vec<&str> = EXCLUDE_LABELS.to_vec(); + vec.push(&CONFIG.common.column_timestamp); + vec +} diff --git a/src/service/metrics/otlp.rs b/src/service/metrics/otlp_grpc.rs similarity index 93% rename from src/service/metrics/otlp.rs rename to src/service/metrics/otlp_grpc.rs index 74e9e51293..38ec0199cb 100644 --- a/src/service/metrics/otlp.rs +++ b/src/service/metrics/otlp_grpc.rs @@ -14,7 +14,7 @@ use crate::{ common::{ - infra::{cluster, config::CONFIG}, + infra::{cluster, config::CONFIG, metrics}, meta::{ self, alert::{Alert, Trigger}, @@ -51,12 +51,13 @@ use opentelemetry_proto::tonic::{ }; use prost::Message; -const EXCLUDE_LABELS: [&str; 5] = [VALUE_LABEL, "start_time", "is_monotonic", "exemplars", "le"]; +use super::get_exclude_labels; pub async fn handle_grpc_request( org_id: &str, thread_id: usize, request: ExportMetricsServiceRequest, + is_grpc: bool, ) -> Result { if !cluster::is_ingester(&cluster::LOCAL_NODE_ROLE) { return Ok( @@ -81,8 +82,8 @@ pub async fn handle_grpc_request( let mut stream_partitioning_map: AHashMap = AHashMap::new(); for resource_metric in &request.resource_metrics { - for instrumentation_metric in &resource_metric.instrumentation_library_metrics { - for metric in &instrumentation_metric.metrics { + for scope_metric in &resource_metric.scope_metrics { + for metric in &scope_metric.metrics { let metric_name = &metric.name; // check for schema let schema_exists = stream_schema_exists( @@ -125,7 +126,7 @@ pub async fn handle_grpc_request( } None => {} } - match &instrumentation_metric.instrumentation_library { + match &scope_metric.scope { Some(lib) => { rec["instrumentation_library_name"] = serde_json::Value::String(lib.name.to_owned()); @@ -161,7 +162,7 @@ pub async fn handle_grpc_request( &mut prom_meta, ), Data::Summary(summary) => { - process_summary(&mut rec, summary, &mut metadata, &mut prom_meta) + process_summary(&rec, summary, &mut metadata, &mut prom_meta) } }, None => vec![], @@ -298,6 +299,32 @@ pub async fn handle_grpc_request( 0, ) .await; + + let ep = if is_grpc { + "grpc/export/metrics" + } else { + "/api/org/v1/metrics" + }; + + let time = start.elapsed().as_secs_f64(); + metrics::HTTP_RESPONSE_TIME + .with_label_values(&[ + ep, + "200", + org_id, + &stream_name, + StreamType::Metrics.to_string().as_str(), + ]) + .observe(time); + metrics::HTTP_INCOMING_REQUESTS + .with_label_values(&[ + ep, + "200", + org_id, + &stream_name, + StreamType::Metrics.to_string().as_str(), + ]) + .inc(); } // only one trigger per request, as it updates etcd @@ -322,7 +349,9 @@ pub async fn handle_grpc_request( } } - let res = ExportMetricsServiceResponse {}; + let res = ExportMetricsServiceResponse { + partial_success: None, + }; let mut out = BytesMut::with_capacity(res.encoded_len()); res.encode(&mut out).expect("Out of memory"); @@ -374,24 +403,19 @@ fn process_sum( process_aggregation_temporality(rec, sum.aggregation_temporality); rec["is_monotonic"] = sum.is_monotonic.into(); for data_point in &sum.data_points { - process_data_point(rec, data_point); - let val_map = rec.as_object_mut().unwrap(); + let mut dp_rec = rec.clone(); + process_data_point(&mut dp_rec, data_point); + let val_map = dp_rec.as_object_mut().unwrap(); let vec: Vec<&str> = get_exclude_labels(); let hash = super::signature_without_labels(val_map, &vec); val_map.insert(HASH_LABEL.to_string(), json::Value::String(hash.into())); - records.push(rec.clone()); + records.push(dp_rec.clone()); } records } -fn get_exclude_labels() -> Vec<&'static str> { - let mut vec: Vec<&str> = EXCLUDE_LABELS.to_vec(); - vec.push(&CONFIG.common.column_timestamp); - vec -} - fn process_histogram( rec: &mut json::Value, hist: &Histogram, @@ -408,7 +432,8 @@ fn process_histogram( let mut records = vec![]; process_aggregation_temporality(rec, hist.aggregation_temporality); for data_point in &hist.data_points { - for mut bucket_rec in process_hist_data_point(rec, data_point) { + let mut dp_rec = rec.clone(); + for mut bucket_rec in process_hist_data_point(&mut dp_rec, data_point) { let val_map = bucket_rec.as_object_mut().unwrap(); let vec: Vec<&str> = get_exclude_labels(); let hash = super::signature_without_labels(val_map, &vec); @@ -434,7 +459,8 @@ fn process_exponential_histogram( let mut records = vec![]; process_aggregation_temporality(rec, hist.aggregation_temporality); for data_point in &hist.data_points { - for mut bucket_rec in process_exp_hist_data_point(rec, data_point) { + let mut dp_rec = rec.clone(); + for mut bucket_rec in process_exp_hist_data_point(&mut dp_rec, data_point) { let val_map = bucket_rec.as_object_mut().unwrap(); let vec: Vec<&str> = get_exclude_labels(); let hash = super::signature_without_labels(val_map, &vec); @@ -446,7 +472,7 @@ fn process_exponential_histogram( } fn process_summary( - rec: &mut json::Value, + rec: &json::Value, summary: &Summary, metadata: &mut prom::Metadata, prom_meta: &mut AHashMap, @@ -460,7 +486,8 @@ fn process_summary( let mut records = vec![]; for data_point in &summary.data_points { - for mut bucket_rec in process_summary_data_point(rec, data_point) { + let mut dp_rec = rec.clone(); + for mut bucket_rec in process_summary_data_point(&mut dp_rec, data_point) { let val_map = bucket_rec.as_object_mut().unwrap(); let vec: Vec<&str> = get_exclude_labels(); let hash = super::signature_without_labels(val_map, &vec); diff --git a/src/service/metrics/otlp_http.rs b/src/service/metrics/otlp_http.rs new file mode 100644 index 0000000000..66891a6b9b --- /dev/null +++ b/src/service/metrics/otlp_http.rs @@ -0,0 +1,1006 @@ +// Copyright 2023 Zinc Labs Inc. +// +// 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. + +use crate::{ + common::{ + infra::{cluster, config::CONFIG, metrics}, + meta::{ + self, + alert::{Alert, Trigger}, + http::HttpResponse as MetaHttpResponse, + prom::{self, MetricType, HASH_LABEL, METADATA_LABEL, NAME_LABEL, VALUE_LABEL}, + stream::{PartitioningDetails, StreamParams}, + usage::UsageType, + StreamType, + }, + utils::{flatten, json}, + }, + handler::http::request::CONTENT_TYPE_JSON, + service::{ + db, + ingestion::{chk_schema_by_record, grpc::get_val_for_attr, write_file}, + schema::{set_schema_metadata, stream_schema_exists}, + stream::unwrap_partition_time_level, + usage::report_request_usage_stats, + }, +}; +use actix_web::{http, web, HttpResponse}; +use ahash::AHashMap; +use bytes::BytesMut; +use chrono::Utc; +use datafusion::arrow::datatypes::Schema; +use opentelemetry::trace::{SpanId, TraceId}; +use opentelemetry_proto::tonic::{ + collector::metrics::v1::{ExportMetricsServiceRequest, ExportMetricsServiceResponse}, + metrics::v1::*, +}; +use prost::Message; + +use super::{get_exclude_labels, otlp_grpc::handle_grpc_request}; + +const SERVICE: &str = "service"; + +pub async fn metrics_proto_handler( + org_id: &str, + thread_id: usize, + body: web::Bytes, +) -> Result { + let request = ExportMetricsServiceRequest::decode(body).expect("Invalid protobuf"); + match handle_grpc_request(org_id, thread_id, request, false).await { + Ok(res) => Ok(res), + Err(e) => { + log::error!("error while handling request: {}", e); + Ok( + HttpResponse::InternalServerError().json(MetaHttpResponse::error( + http::StatusCode::INTERNAL_SERVER_ERROR.into(), + e.to_string(), + )), + ) + } + } +} + +pub async fn metrics_json_handler( + org_id: &str, + thread_id: usize, + body: web::Bytes, +) -> Result { + if !cluster::is_ingester(&cluster::LOCAL_NODE_ROLE) { + return Ok( + HttpResponse::InternalServerError().json(MetaHttpResponse::error( + http::StatusCode::INTERNAL_SERVER_ERROR.into(), + "not an ingester".to_string(), + )), + ); + } + + if !db::file_list::BLOCKED_ORGS.is_empty() && db::file_list::BLOCKED_ORGS.contains(&org_id) { + return Ok(HttpResponse::Forbidden().json(MetaHttpResponse::error( + http::StatusCode::FORBIDDEN.into(), + "Quota exceeded for this organization".to_string(), + ))); + } + + let start = std::time::Instant::now(); + let mut metric_data_map: AHashMap>> = AHashMap::new(); + let mut metric_schema_map: AHashMap = AHashMap::new(); + let mut stream_alerts_map: AHashMap> = AHashMap::new(); + let mut stream_trigger_map: AHashMap = AHashMap::new(); + let mut stream_partitioning_map: AHashMap = AHashMap::new(); + + let body: json::Value = match json::from_slice(body.as_ref()) { + Ok(v) => v, + Err(e) => { + return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( + http::StatusCode::BAD_REQUEST.into(), + format!("Invalid json: {}", e), + ))) + } + }; + + let res_metrics = match body.get("resourceMetrics") { + Some(v) => match v.as_array() { + Some(v) => v, + None => { + return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( + http::StatusCode::BAD_REQUEST.into(), + format!("Invalid json: the structure must be {{\"resourceMetrics\":[]}}"), + ))) + } + }, + None => { + return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( + http::StatusCode::BAD_REQUEST.into(), + format!("Invalid json: the structure must be {{\"resourceMetrics\":[]}}"), + ))) + } + }; + + for res_metric in res_metrics.iter() { + let mut service_att_map: json::Map = json::Map::new(); + if res_metric.get("resource").is_some() { + let resource = res_metric.get("resource").unwrap().as_object().unwrap(); + if resource.get("attributes").is_some() { + let attributes = resource.get("attributes").unwrap().as_array().unwrap(); + for res_attr in attributes { + let local_attr = res_attr.as_object().unwrap(); + + service_att_map.insert( + format!( + "{}_{}", + SERVICE, + local_attr.get("key").unwrap().as_str().unwrap() + ), + get_val_for_attr(local_attr.get("value").unwrap().clone()), + ); + } + } + } + let scope_resources = res_metric.get("scopeMetrics"); + let inst_resources = if scope_resources.is_some() { + scope_resources.unwrap().as_array().unwrap() + } else { + continue; + }; + for inst_metrics in inst_resources { + if inst_metrics.get("metrics").is_some() { + let metrics = inst_metrics.get("metrics").unwrap().as_array().unwrap(); + for metric in metrics { + // parse metadata + let metric_name = metric.get("name").unwrap().as_str().unwrap(); + // check for schema + let schema_exists = stream_schema_exists( + org_id, + metric_name, + StreamType::Metrics, + &mut metric_schema_map, + ) + .await; + + // get partition keys + if !stream_partitioning_map.contains_key(metric_name) { + let partition_det = crate::service::ingestion::get_stream_partition_keys( + metric_name, + &metric_schema_map, + ) + .await; + stream_partitioning_map + .insert(metric_name.to_owned(), partition_det.clone()); + } + let partition_det = stream_partitioning_map.get(metric_name).unwrap(); + let partition_keys = partition_det.partition_keys.clone(); + let partition_time_level = unwrap_partition_time_level( + partition_det.partition_time_level, + StreamType::Metrics, + ); + + // Start get stream alerts + let key = format!("{}/{}/{}", &org_id, StreamType::Metrics, metric_name); + crate::service::ingestion::get_stream_alerts(key, &mut stream_alerts_map).await; + // End get stream alert + + let buf = metric_data_map.entry(metric_name.to_owned()).or_default(); + let mut rec = json::json!({}); + + rec[NAME_LABEL] = metric_name.to_owned().into(); + + for (key, value) in &service_att_map { + rec[key] = value.clone(); + } + + if let Some(lib) = inst_metrics.get("scope") { + let lib = lib.as_object().unwrap(); + if let Some(v) = lib.get("name") { + rec["instrumentation_library_name"] = + serde_json::Value::String(v.as_str().unwrap().to_string()); + } + if let Some(v) = lib.get("version") { + rec["instrumentation_library_version"] = + serde_json::Value::String(v.as_str().unwrap().to_string()); + } + }; + + let metadata = prom::Metadata { + metric_family_name: metric_name.to_owned(), + metric_type: MetricType::Unknown, + help: metric + .get("description") + .unwrap_or(&json::json!("")) + .as_str() + .unwrap() + .to_owned(), + unit: metric.get("unit").unwrap().as_str().unwrap().to_owned(), + }; + let mut prom_meta: AHashMap = AHashMap::new(); + + let records = if metric.get("sum").is_some() { + let sum = metric.get("sum").unwrap().as_object().unwrap(); + process_sum(&mut rec, sum, &mut metadata.clone(), &mut prom_meta) + } else if metric.get("histogram").is_some() { + let histogram = metric.get("histogram").unwrap().as_object().unwrap(); + process_histogram( + &mut rec, + histogram, + &mut metadata.clone(), + &mut prom_meta, + ) + } else if metric.get("summary").is_some() { + let summary = metric.get("summary").unwrap().as_object().unwrap(); + process_summary(&rec, summary, &mut metadata.clone(), &mut prom_meta) + } else if metric.get("gauge").is_some() { + let gauge = metric.get("gauge").unwrap().as_object().unwrap(); + process_gauge(&mut rec, gauge, &mut metadata.clone(), &mut prom_meta) + } else if metric.get("exponentialHistogram").is_some() { + let exp = metric + .get("exponentialHistogram") + .unwrap() + .as_object() + .unwrap(); + process_exponential_histogram( + &mut rec, + exp, + &mut metadata.clone(), + &mut prom_meta, + ) + } else { + continue; + }; + + let mut extra_metadata: AHashMap = AHashMap::new(); + extra_metadata.insert( + METADATA_LABEL.to_string(), + json::to_string(&metadata).unwrap(), + ); + set_schema_metadata(org_id, metric_name, StreamType::Metrics, extra_metadata) + .await + .unwrap(); + + if !schema_exists.has_metadata { + set_schema_metadata(org_id, metric_name, StreamType::Metrics, prom_meta) + .await + .unwrap(); + } + + for mut rec in records { + // flattening + rec = flatten::flatten(&rec).expect("failed to flatten"); + // get json object + let val_map: &mut serde_json::Map = + rec.as_object_mut().unwrap(); + + let timestamp = val_map + .get(&CONFIG.common.column_timestamp) + .unwrap() + .as_i64() + .unwrap_or(Utc::now().timestamp_micros()); + + let value_str = json::to_string(&val_map).unwrap(); + chk_schema_by_record( + &mut metric_schema_map, + org_id, + StreamType::Metrics, + metric_name, + timestamp, + &value_str, + ) + .await; + + // get hour key + let hour_key = crate::service::ingestion::get_wal_time_key( + timestamp, + &partition_keys, + partition_time_level, + val_map, + None, + ); + let hour_buf = buf.entry(hour_key).or_default(); + hour_buf.push(value_str); + + // real time alert + if !stream_alerts_map.is_empty() { + // Start check for alert trigger + let key = + format!("{}/{}/{}", &org_id, StreamType::Metrics, metric_name); + if let Some(alerts) = stream_alerts_map.get(&key) { + for alert in alerts { + if alert.is_real_time { + let set_trigger = meta::alert::Evaluate::evaluate( + &alert.condition, + val_map.clone(), + ); + if set_trigger { + stream_trigger_map.insert( + metric_name.to_owned(), + Trigger { + timestamp, + is_valid: true, + alert_name: alert.name.clone(), + stream: metric_name.to_owned(), + org: org_id.to_string(), + stream_type: StreamType::Metrics, + last_sent_at: 0, + count: 0, + is_ingest_time: true, + }, + ); + } + } + } + } + // End check for alert trigger + } + } + } + } + } + } + + let time = start.elapsed().as_secs_f64(); + for (stream_name, stream_data) in metric_data_map { + // stream_data could be empty if metric value is nan, check it + if stream_data.is_empty() { + continue; + } + + // write to file + let mut stream_file_name = "".to_string(); + + // check if we are allowed to ingest + if db::compact::retention::is_deleting_stream( + org_id, + &stream_name, + StreamType::Metrics, + None, + ) { + log::warn!("stream [{stream_name}] is being deleted"); + continue; + } + + let time_level = if let Some(details) = stream_partitioning_map.get(&stream_name) { + details.partition_time_level + } else { + Some(CONFIG.limit.metrics_file_retention.as_str().into()) + }; + + let mut req_stats = write_file( + stream_data, + thread_id, + StreamParams { + org_id, + stream_name: &stream_name, + stream_type: StreamType::Metrics, + }, + &mut stream_file_name, + time_level, + ); + + req_stats.response_time += time; + report_request_usage_stats( + req_stats, + org_id, + &stream_name, + StreamType::Metrics, + UsageType::Metrics, + 0, + ) + .await; + + let time = start.elapsed().as_secs_f64(); + metrics::HTTP_RESPONSE_TIME + .with_label_values(&[ + "/api/org/v1/metrics", + "200", + org_id, + &stream_name, + StreamType::Metrics.to_string().as_str(), + ]) + .observe(time); + metrics::HTTP_INCOMING_REQUESTS + .with_label_values(&[ + "/api/org/v1/metrics", + "200", + org_id, + &stream_name, + StreamType::Metrics.to_string().as_str(), + ]) + .inc(); + } + + // only one trigger per request, as it updates etcd + for (_, entry) in &stream_trigger_map { + let mut alerts = stream_alerts_map + .get(&format!( + "{}/{}/{}", + entry.org, + StreamType::Metrics, + entry.stream + )) + .unwrap() + .clone(); + + alerts.retain(|alert| alert.name.eq(&entry.alert_name)); + if !alerts.is_empty() { + crate::service::ingestion::send_ingest_notification( + entry.clone(), + alerts.first().unwrap().clone(), + ) + .await; + } + } + + let res = ExportMetricsServiceResponse { + partial_success: None, + }; + let mut out = BytesMut::with_capacity(res.encoded_len()); + res.encode(&mut out).expect("Out of memory"); + + return Ok(HttpResponse::Ok() + .status(http::StatusCode::OK) + .content_type(CONTENT_TYPE_JSON) + .body(out)); +} + +fn process_sum( + rec: &mut json::Value, + sum: &json::Map, + metadata: &mut prom::Metadata, + prom_meta: &mut AHashMap, +) -> Vec { + // set metadata + metadata.metric_type = MetricType::Counter; + prom_meta.insert( + meta::prom::METADATA_LABEL.to_string(), + json::to_string(&metadata).unwrap(), + ); + + let mut records = vec![]; + process_aggregation_temporality( + rec, + sum.get("aggregationTemporality").unwrap().as_u64().unwrap(), + ); + rec["is_monotonic"] = sum.get("isMonotonic").unwrap().as_bool().unwrap().into(); + for data_point in sum.get("dataPoints").unwrap().as_array().unwrap_or(&vec![]) { + let dp = data_point.as_object().unwrap(); + let mut dp_rec = rec.clone(); + process_data_point(&mut dp_rec, dp); + let val_map = dp_rec.as_object_mut().unwrap(); + + let vec: Vec<&str> = get_exclude_labels(); + + let hash = super::signature_without_labels(val_map, &vec); + val_map.insert(HASH_LABEL.to_string(), json::Value::String(hash.into())); + records.push(dp_rec.clone()); + } + records +} + +fn process_histogram( + rec: &mut json::Value, + hist: &json::Map, + metadata: &mut prom::Metadata, + prom_meta: &mut AHashMap, +) -> Vec { + // set metadata + metadata.metric_type = MetricType::Histogram; + prom_meta.insert( + meta::prom::METADATA_LABEL.to_string(), + json::to_string(&metadata).unwrap(), + ); + + let mut records = vec![]; + process_aggregation_temporality( + rec, + hist.get("aggregationTemporality") + .unwrap() + .as_u64() + .unwrap(), + ); + for data_point in hist + .get("dataPoints") + .unwrap() + .as_array() + .unwrap_or(&vec![]) + { + let dp = data_point.as_object().unwrap(); + let mut dp_rec = rec.clone(); + for mut bucket_rec in process_hist_data_point(&mut dp_rec, dp) { + let val_map = bucket_rec.as_object_mut().unwrap(); + let vec: Vec<&str> = get_exclude_labels(); + let hash = super::signature_without_labels(val_map, &vec); + val_map.insert(HASH_LABEL.to_string(), json::Value::String(hash.into())); + records.push(bucket_rec); + } + } + records +} + +fn process_summary( + rec: &json::Value, + summary: &json::Map, + metadata: &mut prom::Metadata, + prom_meta: &mut AHashMap, +) -> Vec { + // set metadata + metadata.metric_type = MetricType::Summary; + prom_meta.insert( + meta::prom::METADATA_LABEL.to_string(), + json::to_string(&metadata).unwrap(), + ); + + let mut records = vec![]; + for data_point in summary + .get("dataPoints") + .unwrap() + .as_array() + .unwrap_or(&vec![]) + { + let dp = data_point.as_object().unwrap(); + let mut dp_rec = rec.clone(); + for mut bucket_rec in process_summary_data_point(&mut dp_rec, dp) { + let val_map = bucket_rec.as_object_mut().unwrap(); + let vec: Vec<&str> = get_exclude_labels(); + let hash = super::signature_without_labels(val_map, &vec); + val_map.insert(HASH_LABEL.to_string(), json::Value::String(hash.into())); + records.push(bucket_rec); + } + } + records +} + +fn process_gauge( + rec: &mut json::Value, + gauge: &json::Map, + metadata: &mut prom::Metadata, + prom_meta: &mut AHashMap, +) -> Vec { + let mut records = vec![]; + + // set metadata + metadata.metric_type = MetricType::Gauge; + prom_meta.insert( + meta::prom::METADATA_LABEL.to_string(), + json::to_string(&metadata).unwrap(), + ); + + for data_point in gauge + .get("dataPoints") + .unwrap() + .as_array() + .unwrap_or(&vec![]) + { + let dp = data_point.as_object().unwrap(); + process_data_point(rec, dp); + let val_map = rec.as_object_mut().unwrap(); + let hash = super::signature_without_labels(val_map, &[VALUE_LABEL]); + val_map.insert(HASH_LABEL.to_string(), json::Value::String(hash.into())); + records.push(rec.clone()); + } + records +} + +fn process_exponential_histogram( + rec: &mut json::Value, + hist: &json::Map, + metadata: &mut prom::Metadata, + prom_meta: &mut AHashMap, +) -> Vec { + // set metadata + metadata.metric_type = MetricType::ExponentialHistogram; + prom_meta.insert( + meta::prom::METADATA_LABEL.to_string(), + json::to_string(&metadata).unwrap(), + ); + let mut records = vec![]; + process_aggregation_temporality( + rec, + hist.get("aggregationTemporality") + .unwrap() + .as_u64() + .unwrap(), + ); + for data_point in hist + .get("dataPoints") + .unwrap() + .as_array() + .unwrap_or(&vec![]) + { + let mut dp_rec = rec.clone(); + let dp = data_point.as_object().unwrap(); + for mut bucket_rec in process_exp_hist_data_point(&mut dp_rec, dp) { + let val_map = bucket_rec.as_object_mut().unwrap(); + let vec: Vec<&str> = get_exclude_labels(); + let hash = super::signature_without_labels(val_map, &vec); + val_map.insert(HASH_LABEL.to_string(), json::Value::String(hash.into())); + records.push(bucket_rec); + } + } + records +} + +fn process_data_point(rec: &mut json::Value, data_point: &json::Map) { + for attr in data_point + .get("attributes") + .unwrap() + .as_array() + .unwrap_or(&vec![]) + { + let attr = attr.as_object().unwrap(); + if let Some(v) = attr.get("value") { + rec[attr.get("key").unwrap().as_str().unwrap()] = get_attribute_value(v) + } + } + let ts = data_point + .get("timeUnixNano") + .unwrap() + .as_str() + .unwrap() + .parse::() + .unwrap(); + rec["start_time"] = data_point.get("startTimeUnixNano").unwrap().clone(); + rec[&CONFIG.common.column_timestamp] = (ts / 1000).into(); + + set_data_point_value(rec, data_point); + + if let Some(v) = data_point.get("flags") { + rec["flag"] = if v.as_u64().unwrap() == 1 { + DataPointFlags::FlagNoRecordedValue.as_str_name() + } else { + DataPointFlags::FlagNone.as_str_name() + } + .into(); + } + + process_exemplars(rec, data_point); +} + +fn process_hist_data_point( + rec: &mut json::Value, + data_point: &json::Map, +) -> Vec { + let mut bucket_recs = vec![]; + + for attr in data_point + .get("attributes") + .unwrap() + .as_array() + .unwrap_or(&vec![]) + { + let attr = attr.as_object().unwrap(); + if let Some(v) = attr.get("value") { + rec[attr.get("key").unwrap().as_str().unwrap()] = get_attribute_value(v); + } + } + let ts = data_point + .get("timeUnixNano") + .unwrap() + .as_str() + .unwrap() + .parse::() + .unwrap(); + rec["start_time"] = data_point.get("startTimeUnixNano").unwrap().clone(); + rec[&CONFIG.common.column_timestamp] = (ts / 1000).into(); + if let Some(v) = data_point.get("flags") { + rec["flag"] = if v.as_u64().unwrap() == 1 { + DataPointFlags::FlagNoRecordedValue.as_str_name() + } else { + DataPointFlags::FlagNone.as_str_name() + } + .into(); + } + process_exemplars(rec, data_point); + // add count record + let mut count_rec = rec.clone(); + count_rec[VALUE_LABEL] = data_point.get("count").unwrap().as_str().into(); + count_rec[NAME_LABEL] = format!("{}_count", count_rec[NAME_LABEL]).into(); + bucket_recs.push(count_rec); + + // add sum record + let mut sum_rec = rec.clone(); + sum_rec[VALUE_LABEL] = data_point.get("sum").unwrap().as_str().into(); + sum_rec[NAME_LABEL] = format!("{}_sum", sum_rec[NAME_LABEL]).into(); + bucket_recs.push(sum_rec); + + // add bucket records + + let buckets = data_point.get("bucketCounts").unwrap().as_array().unwrap(); + let explicit_bounds = data_point + .get("explicitBounds") + .unwrap() + .as_array() + .unwrap(); + let last_index = buckets.len() - 1; + for i in 0..last_index { + let mut bucket_rec = rec.clone(); + if let Some(val) = buckets.get(i) { + bucket_rec[VALUE_LABEL] = (*val).clone() + } + if let Some(val) = explicit_bounds.get(i) { + bucket_rec["le"] = (*val).clone() + } + if i == last_index { + bucket_rec["le"] = std::f64::INFINITY.into(); + } + bucket_recs.push(bucket_rec); + } + bucket_recs +} + +fn process_exp_hist_data_point( + rec: &mut json::Value, + data_point: &json::Map, +) -> Vec { + let mut bucket_recs = vec![]; + + for attr in data_point + .get("attributes") + .unwrap() + .as_array() + .unwrap_or(&vec![]) + { + let attr = attr.as_object().unwrap(); + if let Some(v) = attr.get("value") { + rec[attr.get("key").unwrap().as_str().unwrap()] = get_attribute_value(v); + } + } + let ts = data_point + .get("timeUnixNano") + .unwrap() + .as_str() + .unwrap() + .parse::() + .unwrap(); + rec["start_time"] = data_point.get("startTimeUnixNano").unwrap().clone(); + rec[&CONFIG.common.column_timestamp] = (ts / 1000).into(); + if let Some(v) = data_point.get("flags") { + rec["flag"] = if v.as_u64().unwrap() == 1 { + DataPointFlags::FlagNoRecordedValue.as_str_name() + } else { + DataPointFlags::FlagNone.as_str_name() + } + .into(); + } + process_exemplars(rec, data_point); + + // add count record + let mut count_rec = rec.clone(); + count_rec[VALUE_LABEL] = data_point.get("count").unwrap().as_str().into(); + count_rec[NAME_LABEL] = format!("{}_count", count_rec[NAME_LABEL]).into(); + bucket_recs.push(count_rec); + + // add sum record + let mut sum_rec = rec.clone(); + sum_rec[VALUE_LABEL] = data_point.get("sum").unwrap().as_str().into(); + sum_rec[NAME_LABEL] = format!("{}_sum", sum_rec[NAME_LABEL]).into(); + bucket_recs.push(sum_rec); + + let base = 2 ^ (2 ^ -data_point.get("scale").unwrap().as_i64().unwrap()); + // add negative bucket records + if let Some(buckets) = data_point.get("negative") { + let buckets = buckets.as_object().unwrap(); + let offset = buckets.get("offset").unwrap().as_i64().unwrap(); + for (i, val) in buckets + .get("bucket_counts") + .unwrap() + .as_array() + .unwrap() + .iter() + .enumerate() + { + let mut bucket_rec = rec.clone(); + bucket_rec[VALUE_LABEL] = (*val).clone(); + bucket_rec["le"] = (base ^ (offset + (i as i64) + 1)).into(); + bucket_recs.push(bucket_rec); + } + } + // add positive bucket records + if let Some(buckets) = data_point.get("positive") { + let buckets = buckets.as_object().unwrap(); + let offset = buckets.get("offset").unwrap().as_i64().unwrap(); + for (i, val) in buckets + .get("bucket_counts") + .unwrap() + .as_array() + .unwrap() + .iter() + .enumerate() + { + let mut bucket_rec = rec.clone(); + bucket_rec[VALUE_LABEL] = (*val).clone(); + bucket_rec["le"] = (base ^ (offset + (i as i64) + 1)).into(); + bucket_recs.push(bucket_rec); + } + } + + bucket_recs +} + +fn process_summary_data_point( + rec: &mut json::Value, + data_point: &json::Map, +) -> Vec { + let mut bucket_recs = vec![]; + + for attr in data_point + .get("attributes") + .unwrap() + .as_array() + .unwrap_or(&vec![]) + { + let attr = attr.as_object().unwrap(); + if let Some(v) = attr.get("value") { + rec[attr.get("key").unwrap().as_str().unwrap()] = get_attribute_value(v) + } + } + let ts = data_point + .get("timeUnixNano") + .unwrap() + .as_str() + .unwrap() + .parse::() + .unwrap(); + rec["start_time"] = data_point.get("startTimeUnixNano").unwrap().clone(); + rec[&CONFIG.common.column_timestamp] = (ts / 1000).into(); + if let Some(v) = data_point.get("flags") { + rec["flag"] = if v.as_u64().unwrap() == 1 { + DataPointFlags::FlagNoRecordedValue.as_str_name() + } else { + DataPointFlags::FlagNone.as_str_name() + } + .into(); + } + // add count record + let mut count_rec = rec.clone(); + count_rec[VALUE_LABEL] = data_point.get("count").unwrap().as_str().into(); + count_rec[NAME_LABEL] = format!("{}_count", count_rec[NAME_LABEL]).into(); + bucket_recs.push(count_rec); + + // add sum record + let mut sum_rec = rec.clone(); + sum_rec[VALUE_LABEL] = data_point.get("sum").unwrap().as_str().into(); + sum_rec[NAME_LABEL] = format!("{}_sum", sum_rec[NAME_LABEL]).into(); + bucket_recs.push(sum_rec); + + // add bucket records + let buckets = data_point + .get("quantileValues") + .unwrap() + .as_array() + .unwrap(); + for value in buckets { + let mut bucket_rec = rec.clone(); + let value = value.as_object().unwrap(); + bucket_rec[VALUE_LABEL] = value.get("value").unwrap().clone(); + bucket_rec["quantile"] = value.get("quantile").unwrap().clone(); + bucket_recs.push(bucket_rec); + } + bucket_recs +} + +fn process_aggregation_temporality(rec: &mut json::Value, val: u64) { + rec["aggregation_temporality"] = match val { + 0 => AggregationTemporality::Unspecified.as_str_name(), + 1 => AggregationTemporality::Delta.as_str_name(), + 2 => AggregationTemporality::Cumulative.as_str_name(), + _ => AggregationTemporality::Unspecified.as_str_name(), + } + .into(); +} + +fn process_exemplars(rec: &mut json::Value, data_point: &json::Map) { + let exemplars = match data_point.get("exemplars") { + Some(v) => match v.as_array() { + Some(v) => v, + None => { + return; + } + }, + None => { + return; + } + }; + let mut exemplar_coll = vec![]; + for exemplar in exemplars { + let exemp = exemplar.as_object().unwrap(); + let mut exemplar_rec = json::json!({}); + for attr in exemp + .get("filteredAttributes") + .unwrap() + .as_array() + .unwrap_or(&vec![]) + { + let attr = attr.as_object().unwrap(); + if let Some(v) = attr.get("value") { + exemplar_rec[attr.get("key").unwrap().as_str().unwrap()] = + get_metric_value(v).into(); + } + } + set_data_point_value(&mut exemplar_rec, exemp); + exemplar_rec[&CONFIG.common.column_timestamp] = + (exemplar.get("timeUnixNano").unwrap().as_u64().unwrap() / 1000).into(); + + let trace_id_bytes = hex::decode(exemplar.get("traceId").unwrap().as_str().unwrap()) + .expect("Failed to decode hex string"); + + let mut trace_id_array = [0u8; 16]; + trace_id_array.copy_from_slice(&trace_id_bytes); + + match TraceId::from_bytes(trace_id_array) { + TraceId::INVALID => {} + _ => { + exemplar_rec["trace_id"] = TraceId::from_bytes(trace_id_array).to_string().into(); + } + }; + + let span_id_bytes = hex::decode(exemplar.get("spanId").unwrap().as_str().unwrap()) + .expect("Failed to decode hex string"); + + let mut span_id_array = [0u8; 8]; + span_id_array.copy_from_slice(&span_id_bytes); + + match SpanId::from_bytes(span_id_array) { + SpanId::INVALID => {} + _ => { + exemplar_rec["span_id"] = SpanId::from_bytes(span_id_array).to_string().into(); + } + }; + + exemplar_coll.push(exemplar_rec) + } + rec["exemplars"] = exemplar_coll.into(); +} + +fn set_data_point_value(rec: &mut json::Value, data_point: &json::Map) { + if data_point.get("asInt").is_some() { + rec[VALUE_LABEL] = (data_point.get("asInt").unwrap().as_i64().unwrap() as f64).into(); + } else { + rec[VALUE_LABEL] = data_point.get("asDouble").unwrap().clone(); + } +} + +fn get_attribute_value(val: &json::Value) -> json::Value { + if val.get("stringValue").is_some() { + val.get("stringValue").unwrap().clone() + } else if val.get("boolValue").is_some() { + val.get("boolValue").unwrap().clone() + } else if val.get("intValue").is_some() { + val.get("intValue").unwrap().clone() + } else if val.get("doubleValue").is_some() { + val.get("doubleValue").unwrap().clone() + } else if val.get("arrayValue").is_some() { + val.get("arrayValue").unwrap().clone() + } else if val.get("kvlistValue").is_some() { + val.get("kvlistValue").unwrap().clone() + } else if val.get("bytesValue").is_some() { + val.get("bytesValue").unwrap().clone() + } else { + json::Value::Null + } +} + +fn get_metric_value(val: &json::Value) -> f64 { + if val.get("stringValue").is_some() { + val.get("stringValue") + .unwrap() + .as_str() + .unwrap() + .parse::() + .unwrap() + } else if val.get("intValue").is_some() { + val.get("intValue").unwrap().as_i64().unwrap() as f64 + } else if val.get("doubleValue").is_some() { + val.get("doubleValue").unwrap().as_f64().unwrap() + } else { + 0.0 + } +} diff --git a/src/service/search/datafusion/transform_udf.rs b/src/service/search/datafusion/transform_udf.rs index d5426716b2..12b1f0663f 100644 --- a/src/service/search/datafusion/transform_udf.rs +++ b/src/service/search/datafusion/transform_udf.rs @@ -26,8 +26,11 @@ use std::{collections::BTreeMap, sync::Arc}; use vector_enrichment::TableRegistry; use vrl::compiler::{runtime::Runtime, CompilationResult, Program, TargetValueRef, VrlRuntime}; -use crate::common::{infra::config::QUERY_FUNCTIONS, utils::json}; -use crate::service::{ingestion::compile_vrl_function, logs::get_value}; +use crate::service::ingestion::compile_vrl_function; +use crate::{ + common::{infra::config::QUERY_FUNCTIONS, utils::json}, + service::ingestion::get_value, +}; fn create_user_df( fn_name: &str, diff --git a/src/service/traces/mod.rs b/src/service/traces/mod.rs index 52747eadc6..1b5671b09f 100644 --- a/src/service/traces/mod.rs +++ b/src/service/traces/mod.rs @@ -134,7 +134,7 @@ pub async fn handle_trace_request( ); } } - let inst_resources = res_span.instrumentation_library_spans; + let inst_resources = res_span.scope_spans; for inst_span in inst_resources { let spans = inst_span.spans; for span in spans { @@ -379,7 +379,9 @@ pub async fn handle_trace_request( } } - let res = ExportTraceServiceResponse {}; + let res = ExportTraceServiceResponse { + partial_success: None, + }; let mut out = BytesMut::with_capacity(res.encoded_len()); res.encode(&mut out).expect("Out of memory"); diff --git a/src/service/traces/otlp_http.rs b/src/service/traces/otlp_http.rs index b83be8eba1..416101b3df 100644 --- a/src/service/traces/otlp_http.rs +++ b/src/service/traces/otlp_http.rs @@ -20,26 +20,28 @@ use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest; use prost::Message; use std::{fs::OpenOptions, io::Error}; -use crate::common::{ - infra::{cluster, config::CONFIG, metrics}, - meta::{ - alert::{Alert, Evaluate, Trigger}, - http::HttpResponse as MetaHttpResponse, - stream::{PartitionTimeLevel, StreamParams}, - traces::{Event, Span, SpanRefType}, - usage::UsageType, - StreamType, - }, - utils::{flatten, json}, -}; use crate::service::{ db, format_partition_key, format_stream_name, ingestion::write_file, - logs::get_value, schema::{add_stream_schema, stream_schema_exists}, stream::unwrap_partition_time_level, usage::report_request_usage_stats, }; +use crate::{ + common::{ + infra::{cluster, config::CONFIG, metrics}, + meta::{ + alert::{Alert, Evaluate, Trigger}, + http::HttpResponse as MetaHttpResponse, + stream::{PartitionTimeLevel, StreamParams}, + traces::{Event, Span, SpanRefType}, + usage::UsageType, + StreamType, + }, + utils::{flatten, json}, + }, + service::ingestion::grpc::get_val_for_attr, +}; const PARENT_SPAN_ID: &str = "reference.parent_span_id"; const PARENT_TRACE_ID: &str = "reference.parent_trace_id"; @@ -130,6 +132,7 @@ pub async fn traces_json( // End Register Transforms for stream */ let mut service_name: String = traces_stream_name.to_string(); + //let export_req: ExportTraceServiceRequest = json::from_slice(body.as_ref()).unwrap(); let body: json::Value = match json::from_slice(body.as_ref()) { Ok(v) => v, Err(e) => { @@ -471,14 +474,6 @@ pub async fn traces_json( //Ok(HttpResponse::Ok().into()) } -fn get_val_for_attr(attr_val: json::Value) -> json::Value { - let local_val = attr_val.as_object().unwrap(); - if let Some((_key, value)) = local_val.into_iter().next() { - return serde_json::Value::String(get_value(value)); - }; - ().into() -} - #[cfg(test)] mod tests { use super::*;