starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-06 08:09:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: otlp http for logs & metrics (#1482)

Browse Source
This commit is contained in:
Ashish Kolhe
2023-09-11 15:25:31 +05:30
committed by GitHub
parent b4f3dc4c3c
commit 56655cdbef
23 changed files with 1698 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+62 -3
View File
@@ -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"
Cargo.toml
+1 -2
View File
@@ -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"
src/common/meta/usage.rs
+5 -1
View File
@@ -91,7 +91,8 @@ impl From<UsageType> 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(),
}
}
}
src/handler/grpc/request/logs.rs
+5 -2
View File
@@ -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()))
}
src/handler/grpc/request/metrics/ingester.rs
+5 -2
View File
@@ -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()))
}
src/handler/grpc/request/traces.rs
+5 -2
View File
@@ -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()))
}
src/handler/grpc/request/usage.rs
+1 -1
View File
@@ -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(),
src/handler/http/request/logs/ingest.rs
+37 -1
View File
@@ -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<String>,
thread_id: web::Data<usize>,
req: HttpRequest,
body: web::Bytes,
) -> Result<HttpResponse, Error> {
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(),
)))
}
}
src/handler/http/request/metrics/ingest.rs
+38 -2
View File
@@ -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<String>,
thread_id: web::Data<usize>,
req: HttpRequest,
body: web::Bytes,
) -> Result<HttpResponse, Error> {
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(),
)))
}
}
src/handler/http/request/mod.rs
+3
View File
@@ -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";
src/handler/http/request/traces/mod.rs
+27 -4
View File
@@ -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<String>,
thread_id: web::Data<usize>,
req: HttpRequest,
body: web::Bytes,
) -> Result<HttpResponse, Error> {
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(),
)),
)
}
}
src/handler/http/router/mod.rs
+6 -3
View File
@@ -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),
);
}
src/service/ingestion/grpc.rs
+10
View File
@@ -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<AnyValue>) -> json::Value {
match attr_val {
Some(local_val) => match &local_val.value {
@@ -112,6 +114,14 @@ pub fn get_exemplar_val(attr_val: &Option<exemplar::Value>) -> 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;
src/service/ingestion/mod.rs
+16
View File
@@ -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;
src/service/logs/mod.rs
+3 -18
View File
@@ -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<Field>) -> (Option<String>, 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<String, Schema>,
src/service/logs/json_no_fn.rs → src/service/logs/otlp_grpc.rs
+25 -14
View File
@@ -204,6 +204,7 @@ pub async fn handle_grpc_request(
org_id: &str,
thread_id: usize,
request: ExportLogsServiceRequest,
is_grpc: bool,
) -> Result<HttpResponse, anyhow::Error> {
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<String, Vec<String>> = 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());
}
}
src/service/logs/otlp_http.rs
+359
View File
@@ -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<HttpResponse, std::io::Error> {
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<HttpResponse, std::io::Error> {
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<String, Schema> = AHashMap::new();
let mut stream_alerts_map: AHashMap<String, Vec<Alert>> = AHashMap::new();
let mut stream_status = StreamStatus::new(stream_name);
let mut trigger: Option<Trigger> = 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<String> = 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<String, Vec<String>> = 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<String, json::Value> = 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::<i64>().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));
}
src/service/metrics/mod.rs
+12 -2
View File
@@ -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<Metadata> {
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
}
src/service/metrics/otlp.rs → src/service/metrics/otlp_grpc.rs
+47 -20
View File
@@ -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<HttpResponse, anyhow::Error> {
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<String, PartitioningDetails> = 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<String, String>,
@@ -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);
src/service/metrics/otlp_http.rs
+1006
View File
File diff suppressed because it is too large Load Diff
src/service/search/datafusion/transform_udf.rs
+5 -2
View File
@@ -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,
src/service/traces/mod.rs
+4 -2
View File
@@ -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");
src/service/traces/otlp_http.rs
+16 -21
View File
@@ -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::*;