diff --git a/src/config/src/meta/self_reporting/usage.rs b/src/config/src/meta/self_reporting/usage.rs index 0271f0ae82..7471ac1054 100644 --- a/src/config/src/meta/self_reporting/usage.rs +++ b/src/config/src/meta/self_reporting/usage.rs @@ -155,6 +155,8 @@ pub enum UsageEvent { Ingestion, Search, Functions, + Pipeline, + RemotePipeline, Other, } @@ -164,6 +166,8 @@ impl std::fmt::Display for UsageEvent { UsageEvent::Ingestion => write!(f, "Ingestion"), UsageEvent::Search => write!(f, "Search"), UsageEvent::Functions => write!(f, "Functions"), + UsageEvent::Pipeline => write!(f, "Pipeline"), + UsageEvent::RemotePipeline => write!(f, "RemotePipeline"), UsageEvent::Other => write!(f, "Other"), } } @@ -177,6 +181,10 @@ impl From for UsageEvent { UsageEvent::Search } else if usage.is_function() { UsageEvent::Functions + } else if usage.is_pipeline() { + UsageEvent::Pipeline + } else if usage.is_remote_pipeline() { + UsageEvent::RemotePipeline } else { UsageEvent::Other } @@ -223,6 +231,10 @@ pub enum UsageType { SearchHistory, #[serde(rename = "functions")] Functions, + #[serde(rename = "pipeline")] + Pipeline, + #[serde(rename = "remote_pipeline")] + RemotePipeline, #[serde(rename = "data_retention")] Retention, #[serde(rename = "syslog")] @@ -267,6 +279,14 @@ impl UsageType { pub fn is_function(&self) -> bool { matches!(self, UsageType::Functions) } + + pub fn is_pipeline(&self) -> bool { + matches!(self, UsageType::Pipeline) + } + + pub fn is_remote_pipeline(&self) -> bool { + matches!(self, UsageType::RemotePipeline) + } } impl std::fmt::Display for UsageType { @@ -291,6 +311,8 @@ impl std::fmt::Display for UsageType { UsageType::SearchTopNValues => write!(f, "/_values"), UsageType::SearchHistory => write!(f, "/_search_history"), UsageType::Functions => write!(f, "functions"), + UsageType::Pipeline => write!(f, "pipeline"), + UsageType::RemotePipeline => write!(f, "remote_pipeline"), UsageType::Retention => write!(f, "data_retention"), UsageType::Syslog => write!(f, "syslog"), UsageType::EnrichmentTable => write!(f, "enrichment_table"), diff --git a/src/service/alerts/scheduler/handlers.rs b/src/service/alerts/scheduler/handlers.rs index 7f85fbafea..5ea02472f2 100644 --- a/src/service/alerts/scheduler/handlers.rs +++ b/src/service/alerts/scheduler/handlers.rs @@ -1468,7 +1468,7 @@ async fn handle_derived_stream_triggers( let (_, results): (Vec<_>, Vec<_>) = stream_pl_results.into_iter().unzip(); json_data_by_stream - .entry(stream_params) + .entry(stream_params.clone()) .or_default() .extend(results); } diff --git a/src/service/logs/ingest.rs b/src/service/logs/ingest.rs index 4bd2b91fba..5294155234 100644 --- a/src/service/logs/ingest.rs +++ b/src/service/logs/ingest.rs @@ -231,14 +231,16 @@ pub async fn ingest( None // `item` won't be flattened, no need to store original }; + // we report stream size before pushing data to pipeline + // this is to capture the actual size of stream at the time of ingestion + let _size = size_by_stream.entry(stream_name.clone()).or_insert(0); + *_size += estimate_json_bytes(&item); + if executable_pipeline.is_some() { // buffer the records, timestamp, and originals for pipeline batch processing pipeline_inputs.push(item); original_options.push(original_data); } else { - let _size = size_by_stream.entry(stream_name.clone()).or_insert(0); - *_size += estimate_json_bytes(&item); - // JSON Flattening - use per-stream flatten level let flatten_level = get_flatten_level(org_id, &stream_name, StreamType::Logs).await; let mut res = flatten::flatten_with_level(item, flatten_level)?; @@ -458,10 +460,14 @@ pub async fn ingest( ts_data.push((timestamp, local_val)); *fn_num = need_usage_report.then_some(function_no); - let _size = size_by_stream - .entry(destination_stream.clone()) - .or_insert(0); - *_size += original_size; + // Since we report the size for the original stream before the pipeline + // execution we need to skip reporting the actual size on disk. + if destination_stream.ne(&stream_name) { + let _size = size_by_stream + .entry(destination_stream.clone()) + .or_insert(0); + *_size += original_size; + } tokio::task::coop::consume_budget().await; } diff --git a/src/service/logs/otlp.rs b/src/service/logs/otlp.rs index 2b048e50fa..03eed8604b 100644 --- a/src/service/logs/otlp.rs +++ b/src/service/logs/otlp.rs @@ -345,6 +345,7 @@ pub async fn handle_request( if !derived_streams.contains(&destination_stream) { derived_streams.insert(destination_stream.clone()); } + if !user_defined_schema_map.contains_key(&destination_stream) { // a new dynamically created stream. need to check the two maps again crate::service::ingestion::get_uds_and_original_data_streams( diff --git a/src/service/metrics/json.rs b/src/service/metrics/json.rs index 0af5eadbd5..508107991f 100644 --- a/src/service/metrics/json.rs +++ b/src/service/metrics/json.rs @@ -232,6 +232,7 @@ pub async fn ingest(org_id: &str, body: web::Bytes) -> Result if stream_params.stream_type != StreamType::Metrics { continue; } + // add partition keys if !stream_partitioning_map.contains_key(stream_params.stream_name.as_str()) { diff --git a/src/service/metrics/prom.rs b/src/service/metrics/prom.rs index 25e5bed142..4556d90634 100644 --- a/src/service/metrics/prom.rs +++ b/src/service/metrics/prom.rs @@ -372,6 +372,7 @@ pub async fn remote_write( if stream_params.stream_type != StreamType::Metrics { continue; } + // add partition keys if !stream_partitioning_map.contains_key(stream_params.stream_name.as_str()) { diff --git a/src/service/pipeline/batch_execution.rs b/src/service/pipeline/batch_execution.rs index d87b52fdf8..5306c9914d 100644 --- a/src/service/pipeline/batch_execution.rs +++ b/src/service/pipeline/batch_execution.rs @@ -255,6 +255,34 @@ impl ExecutablePipeline { return Ok(HashMap::default()); } + // Report pipeline ingestion + let source_stream_params = self.get_source_stream_params(); + let source_size: f64 = records + .iter() + .map(|record| record.to_string().len() as f64) + .sum::() + / config::SIZE_IN_MB; + + if source_size > 0.0 { + let req_stats = config::meta::self_reporting::usage::RequestStats { + size: source_size, + records: batch_size as i64, + response_time: 0.0, + ..config::meta::self_reporting::usage::RequestStats::default() + }; + + crate::service::self_reporting::report_request_usage_stats( + req_stats, + org_id, + &self.id, + source_stream_params.stream_type, + config::meta::self_reporting::usage::UsageType::Pipeline, + 0, // No functions for source stream ingestion + chrono::Utc::now().timestamp_micros(), + ) + .await; + } + // result_channel let (result_sender, mut result_receiver) = channel::<(usize, StreamParams, Value)>(batch_size); @@ -267,7 +295,7 @@ impl ExecutablePipeline { let mut node_receivers = HashMap::new(); for node_id in &self.sorted_nodes { - let (sender, receiver) = channel::<(usize, Value, bool)>(batch_size); + let (sender, receiver) = channel::(batch_size); node_senders.insert(node_id.to_string(), sender); node_receivers.insert(node_id.to_string(), receiver); } @@ -348,7 +376,12 @@ impl ExecutablePipeline { }; let source_sender = node_senders.remove(&self.source_node_id).unwrap(); for (idx, record) in records.into_iter().enumerate() { - if let Err(send_err) = source_sender.send((idx, record, flattened)).await { + let pipeline_item = PipelineItem { + idx, + record, + flattened, + }; + if let Err(send_err) = source_sender.send(pipeline_item).await { log::error!( "[Pipeline]: Error sending original records into source Node for {send_err}" ); @@ -539,14 +572,21 @@ impl Default for ExecutablePipelineTraceInputs { } } +#[derive(Debug, Clone)] +struct PipelineItem { + idx: usize, + record: Value, + flattened: bool, +} + #[allow(clippy::too_many_arguments)] async fn process_node( pipeline_id: String, node_idx: usize, org_id: String, node: ExecutableNode, - mut receiver: Receiver<(usize, Value, bool)>, - mut child_senders: Vec>, + mut receiver: Receiver, + mut child_senders: Vec>, vrl_runtime: Option<(VRLResultResolver, bool)>, result_sender: Option>, error_sender: Sender<(String, String, String, Option)>, @@ -562,7 +602,12 @@ async fn process_node( // leaf node: `result_sender` guaranteed to be Some() // send received results directly via `result_sender` for collection let result_sender = result_sender.unwrap(); - while let Some((idx, mut record, flattened)) = receiver.recv().await { + while let Some(pipeline_item) = receiver.recv().await { + let PipelineItem { + idx, + mut record, + flattened, + } = pipeline_item; if !flattened && !record.is_null() && record.is_object() { record = match flatten::flatten_with_level( record, @@ -645,7 +690,12 @@ async fn process_node( } NodeData::Condition(condition_params) => { log::debug!("[Pipeline]: cond node {node_idx} starts processing"); - while let Some((idx, mut record, mut flattened)) = receiver.recv().await { + while let Some(pipeline_item) = receiver.recv().await { + let PipelineItem { + idx, + mut record, + mut flattened, + } = pipeline_item; // value must be flattened before condition params can take effect if !flattened && !record.is_null() && record.is_object() { record = match flatten::flatten_with_level( @@ -677,7 +727,11 @@ async fn process_node( { send_to_children( &mut child_senders, - (idx, record, flattened), + PipelineItem { + idx, + record, + flattened, + }, "ConditionNode", ) .await; @@ -691,7 +745,12 @@ async fn process_node( let mut runtime = crate::service::ingestion::init_functions_runtime(); let stream_name = stream_name.unwrap_or("pipeline".to_string()); let mut result_array_records = Vec::new(); - while let Some((idx, mut record, mut flattened)) = receiver.recv().await { + while let Some(pipeline_item) = receiver.recv().await { + let PipelineItem { + idx, + mut record, + mut flattened, + } = pipeline_item; if let Some((vrl_runtime, is_result_array_vrl)) = &vrl_runtime { if func_params.after_flatten && !flattened @@ -758,7 +817,11 @@ async fn process_node( flattened = false; // since apply_vrl_fn can produce unflattened data send_to_children( &mut child_senders, - (idx, record, flattened), + PipelineItem { + idx, + record, + flattened, + }, "FunctionNode", ) .await; @@ -806,7 +869,11 @@ async fn process_node( // use usize::MAX as a flag to disregard original_value send_to_children( &mut child_senders, - (usize::MAX, record.clone(), false), + PipelineItem { + idx: usize::MAX, + record: record.clone(), + flattened: false, + }, "FunctionNode", ) .await; @@ -835,7 +902,12 @@ async fn process_node( let max_ts = (Utc::now() + chrono::Duration::try_hours(cfg.limit.ingest_allowed_in_future).unwrap()) .timestamp_micros(); - while let Some((_, mut record, flattened)) = receiver.recv().await { + while let Some(pipeline_item) = receiver.recv().await { + let PipelineItem { + mut record, + flattened, + .. + } = pipeline_item; // handle timestamp before sending to remote_write service if !flattened && !record.is_null() && record.is_object() { record = match flatten::flatten_with_level( @@ -951,19 +1023,46 @@ async fn process_node( let mut remote_stream_for_batch = remote_stream.clone(); remote_stream_for_batch.org_id = org_id.clone().into(); + let records_len = records_to_write.len() as i64; + let writer = get_pipeline_wal_writer(&pipeline_id, remote_stream_for_batch).await?; - if let Err(e) = writer.write_wal(records_to_write).await { - let err_msg = format!( - "DestinationNode error persisting data for batch_key '{batch_key}' to be ingested externally: {e}" - ); - if let Err(send_err) = error_sender - .send((node.id.to_string(), node.node_type(), err_msg, None)) - .await - { - log::error!( - "[Pipeline]: DestinationNode failed sending errors for collection caused by: {send_err}" + match writer.write_wal(records_to_write).await { + Err(e) => { + let err_msg = format!( + "DestinationNode error persisting data for batch_key '{batch_key}' to be ingested externally: {e}" ); + if let Err(send_err) = error_sender + .send((node.id.to_string(), node.node_type(), err_msg, None)) + .await + { + log::error!( + "[Pipeline]: DestinationNode failed sending errors for collection caused by: {send_err}" + ); + } + } + Ok(data_size) => { + let data_size_mb = data_size as f64 / config::SIZE_IN_MB; + // Report remote destination usage after successful WAL write + if data_size_mb > 0.0 { + let req_stats = config::meta::self_reporting::usage::RequestStats { + size: data_size_mb, + records: records_len, + response_time: 0.0, + ..config::meta::self_reporting::usage::RequestStats::default() + }; + + crate::service::self_reporting::report_request_usage_stats( + req_stats, + &org_id, + &remote_stream.destination_name, + config::meta::stream::StreamType::Logs, // Default to Logs for remote destinations + config::meta::self_reporting::usage::UsageType::RemotePipeline, + 0, // No additional functions for remote destination + chrono::Utc::now().timestamp_micros(), + ) + .await; + } } } } else { @@ -1013,7 +1112,7 @@ pub async fn flush_all_buffers() -> Result<(), anyhow::Error> { let remote_stream = config::meta::stream::RemoteStreamParams { org_id: org_id.clone().into(), - destination_name: destination_name.into(), + destination_name: destination_name.clone().into(), }; let mut remote_stream_for_batch = remote_stream.clone(); @@ -1032,12 +1131,41 @@ pub async fn flush_all_buffers() -> Result<(), anyhow::Error> { let mut remote_stream_for_batch = remote_stream.clone(); remote_stream_for_batch.org_id = org_id.clone().into(); + let records_len = records_to_write.len(); + let writer = get_pipeline_wal_writer(&pipeline_id, remote_stream_for_batch).await?; - if let Err(e) = writer.write_wal(records_to_write).await { - let err_msg = format!( - "DestinationNode error persisting data for batch_key '{batch_key}' to be ingested externally: {e}" - ); - log::error!("{err_msg}"); + match writer.write_wal(records_to_write).await { + Err(e) => { + let err_msg = format!( + "DestinationNode error persisting data for batch_key '{batch_key}' to be ingested externally: {e}" + ); + log::error!("{err_msg}"); + } + Ok(data_size) => { + let data_size_mb = data_size as f64 / config::SIZE_IN_MB; + + // Report remote destination usage after successful WAL write + if data_size_mb > 0.0 { + let req_stats = config::meta::self_reporting::usage::RequestStats { + size: data_size_mb, + records: records_len as i64, + response_time: 0.0, + ..config::meta::self_reporting::usage::RequestStats::default() + }; + + crate::service::self_reporting::report_request_usage_stats( + req_stats, + &org_id, + &destination_name, + config::meta::stream::StreamType::Logs, /* Default to Logs for + * remote destination */ + config::meta::self_reporting::usage::UsageType::RemotePipeline, + 0, // No additional functions for remote destination + chrono::Utc::now().timestamp_micros(), + ) + .await; + } + } } } } @@ -1047,8 +1175,8 @@ pub async fn flush_all_buffers() -> Result<(), anyhow::Error> { } async fn send_to_children( - child_senders: &mut [Sender<(usize, Value, bool)>], - item: (usize, Value, bool), + child_senders: &mut [Sender], + item: PipelineItem, node_type: &str, ) { if child_senders.len() == 1 { diff --git a/tests/ui-testing/pages/logsPages/logsPage.js b/tests/ui-testing/pages/logsPages/logsPage.js index 4d7d25e53d..697d1c6b5d 100644 --- a/tests/ui-testing/pages/logsPages/logsPage.js +++ b/tests/ui-testing/pages/logsPages/logsPage.js @@ -1861,7 +1861,21 @@ export class LogsPage { async expectPageContainsText(text) { const logsPage = this.page.locator(this.qPageContainer); - return await expect(logsPage).toContainText(text); + + // Wait for the page to stabilize and check for "No events found" condition + await this.page.waitForLoadState('networkidle'); + + // If no data is available, trigger a refresh and wait + const pageText = await logsPage.textContent(); + if (pageText.includes('No events found')) { + console.log('No events found, attempting to refresh...'); + await this.clickRefreshButton(); + await this.page.waitForLoadState('networkidle'); + // Wait additional time for data to load + await this.page.waitForTimeout(3000); + } + + return await expect(logsPage).toContainText(text, { timeout: 10000 }); } async clickLogSearchIndexListFieldSearchInput() { diff --git a/tests/ui-testing/playwright-tests/Logs/logspage.spec.js b/tests/ui-testing/playwright-tests/Logs/logspage.spec.js index 824088b388..98787de07e 100644 --- a/tests/ui-testing/playwright-tests/Logs/logspage.spec.js +++ b/tests/ui-testing/playwright-tests/Logs/logspage.spec.js @@ -411,9 +411,14 @@ test.describe("Logs Page testcases", () => { await pm.logsPage.clickVrlEditor(); await page.waitForLoadState('domcontentloaded'); // Replace hard wait await pm.logsPage.clickRefreshButton(); + // Wait for VRL function to be applied and data to load + await page.waitForLoadState('networkidle'); + await page.waitForTimeout(3000); await pm.logsPage.clickMenuLinkMetricsItem(); await pm.logsPage.clickMenuLinkLogsItem(); await pm.logsPage.clickMenuLinkLogsItem(); + // Wait for page to stabilize after navigation + await page.waitForLoadState('networkidle'); await pm.logsPage.expectPageContainsText(".a=2"); testLogger.info('Function persistence test completed'); diff --git a/web/src/assets/images/usage/pipeline.svg b/web/src/assets/images/usage/pipeline.svg new file mode 100644 index 0000000000..30e457ac08 --- /dev/null +++ b/web/src/assets/images/usage/pipeline.svg @@ -0,0 +1,11 @@ + + + \ No newline at end of file diff --git a/web/src/assets/images/usage/remote_pipeline.svg b/web/src/assets/images/usage/remote_pipeline.svg new file mode 100644 index 0000000000..adae6ace7f --- /dev/null +++ b/web/src/assets/images/usage/remote_pipeline.svg @@ -0,0 +1,5 @@ + + + + + diff --git a/web/src/enterprise/components/billings/usage.spec.ts b/web/src/enterprise/components/billings/usage.spec.ts index 6e57592ce4..338897d108 100644 --- a/web/src/enterprise/components/billings/usage.spec.ts +++ b/web/src/enterprise/components/billings/usage.spec.ts @@ -160,7 +160,9 @@ describe("Usage Component", () => { expect(wrapper.vm.usageData).toEqual({ ingestion: "10.50", search: "5.25", - functions: "2.75" + functions: "2.75", + pipeline: "0.00", + remotepipeline: "0.00" }); }); @@ -328,7 +330,9 @@ describe("Usage Component", () => { expect(wrapper.vm.usageData).toEqual({ ingestion: "10.50", search: "5.25", - functions: "2.75" + functions: "2.75", + pipeline: "0.00", + remotepipeline: "0.00" }); }); diff --git a/web/src/enterprise/components/billings/usage.vue b/web/src/enterprise/components/billings/usage.vue index f6c0622e09..2aa22d56d4 100644 --- a/web/src/enterprise/components/billings/usage.vue +++ b/web/src/enterprise/components/billings/usage.vue @@ -100,7 +100,8 @@ along with this program. If not, see .
-
+
+
@@ -120,50 +121,94 @@ along with this program. If not, see . {{ usageData.ingestion }} {{ usageDataType.toUpperCase() }}
-
-
-
- -
- -
-
Search
-
- -
-
- - -
- {{ usageData.search }} {{ usageDataType.toUpperCase() }} -
-
-
-
-
- -
- -
-
Functions
-
- -
+
+
+ +
+ +
+
Search
+
+ +
+
-
- -
- {{ usageData.functions }} {{ usageDataType.toUpperCase() }} -
-
-
+ +
+ {{ usageData.search }} {{ usageDataType.toUpperCase() }} +
+
+
+
+
+ +
+ +
+
Functions
+
+ +
+
+
+ +
+ {{ usageData.functions }} {{ usageDataType.toUpperCase() }} +
+
+
+
+
+
+
+ +
+ +
+
Pipelines
+
+ +
+
+
+ + +
+ {{ usageData.pipeline }} {{ usageDataType.toUpperCase() }} +
+
+
+
+
+ +
+ +
+
Remote Pipelines
+
+ +
+
+
+ + +
+ {{ usageData.remotepipeline }} {{ usageDataType.toUpperCase() }} +
+
+
+
@@ -212,7 +257,9 @@ import CustomChartRenderer from "@/components/dashboards/panels/CustomChartRende const usageData = ref({ ingestion: "0.00", search: "0.00", - functions: "0.00" + functions: "0.00", + pipeline: "0.00", + remotepipeline: "0.00" }); let chartData: any = ref({}); onMounted(async () => { @@ -269,6 +316,8 @@ import CustomChartRenderer from "@/components/dashboards/panels/CustomChartRende const ingestionIcon = getImageURL("images/usage/ingestion.svg"); const searchIcon = getImageURL("images/usage/search.svg"); const functionsIcon = getImageURL("images/usage/function.svg"); + const pipelineIcon = getImageURL("images/usage/pipeline.svg"); + const remotePipelineIcon = getImageURL("images/usage/remote_pipeline.svg"); //this is the example data that needs to be used to get the chart in usage page //we just need to have the data in the format of the dataModel //eg: date and value and in the array format @@ -853,6 +902,8 @@ import CustomChartRenderer from "@/components/dashboards/panels/CustomChartRende usageData, getUsage, router, + pipelineIcon, + remotePipelineIcon, }; }, });