remove attributes check from otel logs, traces, metrics

nikhilsinhaparseable · nikhilsinhaparseable · commit 9045f6620afc · 2025-05-13T02:26:38.000-04:00
add check for fields count in any dataset
reject event if fields count exceeds `P_DATASET_FIELDS_ALLOWED_LIMIT`
default value is set to 250
diff --git a/src/cli.rs b/src/cli.rs
@@ -371,12 +371,12 @@ pub struct Options {
 
     #[arg(
         long,
-        env = "P_OTEL_ATTRIBUTES_ALLOWED_LIMIT",
-        default_value = "200",
-        value_parser = validation::validate_otel_attributes_allowed_limit,
-        help = "allowed limit for otel attributes"
+        env = "P_DATASET_FIELDS_ALLOWED_LIMIT",
+        default_value = "250",
+        value_parser = validation::validate_dataset_fields_allowed_limit,
+        help = "allowed limit for fields count in a dataset"
     )]
-    pub otel_attributes_allowed_limit: usize,
+    pub dataset_fields_allowed_limit: usize,
 }
 
 #[derive(Parser, Debug)]
diff --git a/src/handlers/http/ingest.rs b/src/handlers/http/ingest.rs
@@ -35,7 +35,6 @@ use crate::metadata::SchemaVersion;
 use crate::option::Mode;
 use crate::otel::logs::OTEL_LOG_KNOWN_FIELD_LIST;
 use crate::otel::metrics::OTEL_METRICS_KNOWN_FIELD_LIST;
-use crate::otel::otel_utils::OtelError;
 use crate::otel::traces::OTEL_TRACES_KNOWN_FIELD_LIST;
 use crate::parseable::{StreamNotFound, PARSEABLE};
 use crate::storage::{ObjectStorageError, StreamType};
@@ -468,8 +467,8 @@ pub enum PostError {
     KnownFormat(#[from] known_schema::Error),
     #[error("Ingestion is not allowed to stream {0} as it is already associated with a different OTEL format")]
     IncorrectLogFormat(String),
-    #[error("OtelError: {0}")]
-    OtelError(#[from] OtelError),
+    #[error("Ingestion failed for dataset {0} as fields count {1} exceeds the limit {2}, Parseable recommends creating a new dataset.")]
+    FieldsLimitExceeded(String, usize, usize),
 }
 
 impl actix_web::ResponseError for PostError {
@@ -498,7 +497,7 @@ impl actix_web::ResponseError for PostError {
             PostError::MissingTimePartition(_) => StatusCode::BAD_REQUEST,
             PostError::KnownFormat(_) => StatusCode::BAD_REQUEST,
             PostError::IncorrectLogFormat(_) => StatusCode::BAD_REQUEST,
-            PostError::OtelError(_) => StatusCode::BAD_REQUEST,
+            PostError::FieldsLimitExceeded(_, _, _) => StatusCode::BAD_REQUEST,
         }
     }
 
diff --git a/src/handlers/http/modal/utils/ingest_utils.rs b/src/handlers/http/modal/utils/ingest_utils.rs
@@ -54,6 +54,24 @@ pub async fn flatten_and_push_logs(
     log_source: &LogSource,
     p_custom_fields: &HashMap<String, String>,
 ) -> Result<(), PostError> {
+    // fetch the storage schema for the stream
+    let schema = PARSEABLE.get_stream(stream_name)?.get_schema();
+    //fetch the fields count from the schema
+    let fields_count = schema.fields().len();
+    if fields_count > PARSEABLE.options.dataset_fields_allowed_limit {
+        tracing::error!(
+            "Ingestion failed for dataset {0} as fields count {1} exceeds the limit {2}, Parseable recommends creating a new dataset.",
+            stream_name,
+            fields_count,
+            PARSEABLE.options.dataset_fields_allowed_limit);
+        // Return an error if the fields count exceeds the limit
+        return Err(PostError::FieldsLimitExceeded(
+            stream_name.to_string(),
+            fields_count,
+            PARSEABLE.options.dataset_fields_allowed_limit,
+        ));
+    }
+
     match log_source {
         LogSource::Kinesis => {
             //custom flattening required for Amazon Kinesis
@@ -65,24 +83,21 @@ pub async fn flatten_and_push_logs(
         LogSource::OtelLogs => {
             //custom flattening required for otel logs
             let logs: LogsData = serde_json::from_value(json)?;
-            let records = flatten_otel_logs(&logs)?;
-            for record in records {
+            for record in flatten_otel_logs(&logs) {
                 push_logs(stream_name, record, log_source, p_custom_fields).await?;
             }
         }
         LogSource::OtelTraces => {
             //custom flattening required for otel traces
             let traces: TracesData = serde_json::from_value(json)?;
-            let records = flatten_otel_traces(&traces)?;
-            for record in records {
+            for record in flatten_otel_traces(&traces) {
                 push_logs(stream_name, record, log_source, p_custom_fields).await?;
             }
         }
         LogSource::OtelMetrics => {
             //custom flattening required for otel metrics
             let metrics: MetricsData = serde_json::from_value(json)?;
-            let records = flatten_otel_metrics(metrics)?;
-            for record in records {
+            for record in flatten_otel_metrics(metrics) {
                 push_logs(stream_name, record, log_source, p_custom_fields).await?;
             }
         }
diff --git a/src/option.rs b/src/option.rs
@@ -95,8 +95,8 @@ pub mod validation {
 
     use super::{Compression, Mode};
 
-    // Maximum allowed otel attributes per event
-    const OTEL_ATTRIBUTES_ALLOWED_LIMIT: usize = 200;
+    // Maximum allowed count for fields in a dataset
+    const DATASET_FIELDS_ALLOWED_LIMIT: usize = 250;
 
     pub fn file_path(s: &str) -> Result<PathBuf, String> {
         if s.is_empty() {
@@ -177,18 +177,18 @@ pub mod validation {
         }
     }
 
-    pub fn validate_otel_attributes_allowed_limit(s: &str) -> Result<usize, String> {
+    pub fn validate_dataset_fields_allowed_limit(s: &str) -> Result<usize, String> {
         if let Ok(size) = s.parse::<usize>() {
-            if (1..=OTEL_ATTRIBUTES_ALLOWED_LIMIT).contains(&size) {
+            if (1..=DATASET_FIELDS_ALLOWED_LIMIT).contains(&size) {
                 Ok(size)
             } else {
                 Err(format!(
-                    "Invalid value for P_OTEL_ATTRIBUTES_ALLOWED_LIMIT. It should be between 1 and {}",
-                    OTEL_ATTRIBUTES_ALLOWED_LIMIT
+                    "Invalid value for P_DATASET_FIELDS_ALLOWED_LIMIT. It should be between 1 and {}",
+                    DATASET_FIELDS_ALLOWED_LIMIT
                 ))
             }
         } else {
-            Err("Invalid value for P_OTEL_ATTRIBUTES_ALLOWED_LIMIT. It should be given as integer value".to_string())
+            Err("Invalid value for P_DATASET_FIELDS_ALLOWED_LIMIT. It should be given as integer value".to_string())
         }
     }
 }
diff --git a/src/otel/logs.rs b/src/otel/logs.rs
@@ -15,15 +15,9 @@
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  *
  */
-
-use std::collections::HashSet;
-
-use crate::parseable::PARSEABLE;
-
 use super::otel_utils::collect_json_from_values;
 use super::otel_utils::convert_epoch_nano_to_timestamp;
 use super::otel_utils::insert_attributes;
-use super::otel_utils::OtelError;
 use opentelemetry_proto::tonic::logs::v1::LogRecord;
 use opentelemetry_proto::tonic::logs::v1::LogsData;
 use opentelemetry_proto::tonic::logs::v1::ScopeLogs;
@@ -150,10 +144,8 @@ fn flatten_scope_log(scope_log: &ScopeLogs) -> Vec<Map<String, Value>> {
 
 /// this function performs the custom flattening of the otel logs
 /// and returns a `Vec` of `Value::Object` of the flattened json
-pub fn flatten_otel_logs(message: &LogsData) -> Result<Vec<Value>, OtelError> {
+pub fn flatten_otel_logs(message: &LogsData) -> Vec<Value> {
     let mut vec_otel_json = Vec::new();
-    let known_fields: HashSet<&str> = OTEL_LOG_KNOWN_FIELD_LIST.iter().cloned().collect();
-
     for record in &message.resource_logs {
         let mut resource_log_json = Map::new();
         if let Some(resource) = &record.resource {
@@ -177,26 +169,9 @@ pub fn flatten_otel_logs(message: &LogsData) -> Result<Vec<Value>, OtelError> {
         for resource_logs_json in &mut vec_resource_logs_json {
             resource_logs_json.extend(resource_log_json.clone());
 
-            let attribute_count = resource_logs_json
-                .iter()
-                .filter(|(key, _)| !known_fields.contains(key.as_str()))
-                .count();
-            // Check if the number of attributes exceeds the allowed limit
-            if attribute_count > PARSEABLE.options.otel_attributes_allowed_limit {
-                tracing::error!(
-                    "OTEL logs ingestion failed because the number of attributes ({}) exceeded the threshold of {}",
-                    attribute_count,
-                    PARSEABLE.options.otel_attributes_allowed_limit
-                );
-                return Err(OtelError::AttributeCountExceeded(
-                    attribute_count,
-                    PARSEABLE.options.otel_attributes_allowed_limit,
-                ));
-            }
-
             vec_otel_json.push(Value::Object(resource_logs_json.clone()));
         }
     }
 
-    Ok(vec_otel_json)
+    vec_otel_json
 }
diff --git a/src/otel/metrics.rs b/src/otel/metrics.rs
@@ -15,19 +15,13 @@
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  *
  */
-
-use std::collections::HashSet;
-
 use opentelemetry_proto::tonic::metrics::v1::number_data_point::Value as NumberDataPointValue;
 use opentelemetry_proto::tonic::metrics::v1::{
     exemplar::Value as ExemplarValue, exponential_histogram_data_point::Buckets, metric, Exemplar,
     ExponentialHistogram, Gauge, Histogram, Metric, MetricsData, NumberDataPoint, Sum, Summary,
 };
 use serde_json::{Map, Value};
 
-use crate::otel::otel_utils::OtelError;
-use crate::parseable::PARSEABLE;
-
 use super::otel_utils::{
     convert_epoch_nano_to_timestamp, insert_attributes, insert_number_if_some,
 };
@@ -507,10 +501,8 @@ pub fn flatten_metrics_record(metrics_record: &Metric) -> Vec<Map<String, Value>
 
 /// this function performs the custom flattening of the otel metrics
 /// and returns a `Vec` of `Value::Object` of the flattened json
-pub fn flatten_otel_metrics(message: MetricsData) -> Result<Vec<Value>, OtelError> {
+pub fn flatten_otel_metrics(message: MetricsData) -> Vec<Value> {
     let mut vec_otel_json = Vec::new();
-    let known_fields: HashSet<&str> = OTEL_METRICS_KNOWN_FIELD_LIST.iter().cloned().collect();
-
     for record in &message.resource_metrics {
         let mut resource_metrics_json = Map::new();
         if let Some(resource) = &record.resource {
@@ -564,29 +556,11 @@ pub fn flatten_otel_metrics(message: MetricsData) -> Result<Vec<Value>, OtelErro
                 resource_metric_json.insert(key.clone(), value.clone());
             }
 
-            let attribute_count = resource_metric_json
-                .iter()
-                .filter(|(key, _)| !known_fields.contains(key.as_str()))
-                .count();
-
-            // Check if the number of attributes exceeds the allowed limit
-            if attribute_count > PARSEABLE.options.otel_attributes_allowed_limit {
-                tracing::error!(
-                    "OTEL metrics ingestion failed because the number of attributes ({}) exceeded the threshold of {}",
-                    attribute_count,
-                    PARSEABLE.options.otel_attributes_allowed_limit
-                );
-                return Err(OtelError::AttributeCountExceeded(
-                    attribute_count,
-                    PARSEABLE.options.otel_attributes_allowed_limit,
-                ));
-            }
-
             vec_otel_json.push(Value::Object(resource_metric_json.clone()));
         }
     }
 
-    Ok(vec_otel_json)
+    vec_otel_json
 }
 
 /// otel metrics event has json object for aggregation temporality
diff --git a/src/otel/otel_utils.rs b/src/otel/otel_utils.rs
@@ -20,7 +20,6 @@ use chrono::DateTime;
 use opentelemetry_proto::tonic::common::v1::{
     any_value::Value as OtelValue, AnyValue, ArrayValue, KeyValue, KeyValueList,
 };
-use serde::Serialize;
 use serde_json::{Map, Value};
 
 // Value can be one of types - String, Bool, Int, Double, ArrayValue, AnyValue, KeyValueList, Byte
@@ -194,9 +193,3 @@ pub fn convert_epoch_nano_to_timestamp(epoch_ns: i64) -> String {
     let dt = DateTime::from_timestamp_nanos(epoch_ns).naive_utc();
     dt.format("%Y-%m-%dT%H:%M:%S%.6fZ").to_string()
 }
-
-#[derive(Debug, thiserror::Error, Serialize)]
-pub enum OtelError {
-    #[error("Ingestion failed because the attributes count {0} exceeded the threshold of {1}")]
-    AttributeCountExceeded(usize, usize),
-}
diff --git a/src/otel/traces.rs b/src/otel/traces.rs
@@ -15,9 +15,6 @@
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  *
  */
-
-use std::collections::HashSet;
-
 use opentelemetry_proto::tonic::trace::v1::span::Event;
 use opentelemetry_proto::tonic::trace::v1::span::Link;
 use opentelemetry_proto::tonic::trace::v1::ScopeSpans;
@@ -26,9 +23,6 @@ use opentelemetry_proto::tonic::trace::v1::Status;
 use opentelemetry_proto::tonic::trace::v1::TracesData;
 use serde_json::{Map, Value};
 
-use crate::otel::otel_utils::OtelError;
-use crate::parseable::PARSEABLE;
-
 use super::otel_utils::convert_epoch_nano_to_timestamp;
 use super::otel_utils::insert_attributes;
 
@@ -105,9 +99,8 @@ fn flatten_scope_span(scope_span: &ScopeSpans) -> Vec<Map<String, Value>> {
 
 /// this function performs the custom flattening of the otel traces event
 /// and returns a `Vec` of `Value::Object` of the flattened json
-pub fn flatten_otel_traces(message: &TracesData) -> Result<Vec<Value>, OtelError> {
+pub fn flatten_otel_traces(message: &TracesData) -> Vec<Value> {
     let mut vec_otel_json = Vec::new();
-    let known_fields: HashSet<&str> = OTEL_TRACES_KNOWN_FIELD_LIST.iter().cloned().collect();
 
     for record in &message.resource_spans {
         let mut resource_span_json = Map::new();
@@ -135,29 +128,11 @@ pub fn flatten_otel_traces(message: &TracesData) -> Result<Vec<Value>, OtelError
                 resource_spans_json.insert(key.clone(), value.clone());
             }
 
-            let attribute_count = resource_spans_json
-                .iter()
-                .filter(|(key, _)| !known_fields.contains(key.as_str()))
-                .count();
-
-            // Check if the number of attributes exceeds the allowed limit
-            if attribute_count > PARSEABLE.options.otel_attributes_allowed_limit {
-                tracing::error!(
-                    "OTEL traces ingestion failed because the number of attributes ({}) exceeded the threshold of {}",
-                    attribute_count,
-                    PARSEABLE.options.otel_attributes_allowed_limit
-                );
-                return Err(OtelError::AttributeCountExceeded(
-                    attribute_count,
-                    PARSEABLE.options.otel_attributes_allowed_limit,
-                ));
-            }
-
             vec_otel_json.push(Value::Object(resource_spans_json.clone()));
         }
     }
 
-    Ok(vec_otel_json)
+    vec_otel_json
 }
 
 /// otel traces has json array of events

Original file line number	Diff line number	Diff line change
`@@ -95,8 +95,8 @@ pub mod validation {`
`95`	`95`
`96`	`96`	`use super::{Compression, Mode};`
`97`	`97`
`98`		`- // Maximum allowed otel attributes per event`
`99`		`- const OTEL_ATTRIBUTES_ALLOWED_LIMIT: usize = 200;`
	`98`	`+ // Maximum allowed count for fields in a dataset`
	`99`	`+ const DATASET_FIELDS_ALLOWED_LIMIT: usize = 250;`
`100`	`100`
`101`	`101`	`pub fn file_path(s: &str) -> Result<PathBuf, String> {`
`102`	`102`	`if s.is_empty() {`
`@@ -177,18 +177,18 @@ pub mod validation {`
`177`	`177`	`}`
`178`	`178`	`}`
`179`	`179`
`180`		`- pub fn validate_otel_attributes_allowed_limit(s: &str) -> Result<usize, String> {`
	`180`	`+ pub fn validate_dataset_fields_allowed_limit(s: &str) -> Result<usize, String> {`
`181`	`181`	`if let Ok(size) = s.parse::<usize>() {`
`182`		`- if (1..=OTEL_ATTRIBUTES_ALLOWED_LIMIT).contains(&size) {`
	`182`	`+ if (1..=DATASET_FIELDS_ALLOWED_LIMIT).contains(&size) {`
`183`	`183`	`Ok(size)`
`184`	`184`	`} else {`
`185`	`185`	`Err(format!(`
`186`		`- "Invalid value for P_OTEL_ATTRIBUTES_ALLOWED_LIMIT. It should be between 1 and {}",`
`187`		`- OTEL_ATTRIBUTES_ALLOWED_LIMIT`
	`186`	`+ "Invalid value for P_DATASET_FIELDS_ALLOWED_LIMIT. It should be between 1 and {}",`
	`187`	`+ DATASET_FIELDS_ALLOWED_LIMIT`
`188`	`188`	`))`
`189`	`189`	`}`
`190`	`190`	`} else {`
`191`		`- Err("Invalid value for P_OTEL_ATTRIBUTES_ALLOWED_LIMIT. It should be given as integer value".to_string())`
	`191`	`+ Err("Invalid value for P_DATASET_FIELDS_ALLOWED_LIMIT. It should be given as integer value".to_string())`
`192`	`192`	`}`
`193`	`193`	`}`
`194`	`194`	`}`