feat: add pipeline usage (#8295)

### **PR Type** Enhancement, Bug fix ___ ### **Description** - Add pipeline and remote usage tracking - Report usage for remote destination WAL writes - Switch pipeline channels to structured items - Fix size accounting across pipeline branches ___ ### Diagram Walkthrough ```mermaid flowchart LR A["Ingestion (Logs/Metrics)"] -- "pre-pipeline size" --> B["Usage: Pipeline"] B -- "execute pipeline" --> C["Pipeline Nodes (Condition/Function/Destination)"] C -- "local dest" --> D["Size by stream (no double count)"] C -- "remote dest WAL write" --> E["Usage: RemotePipeline"] ``` <details> <summary><h3> File Walkthrough</h3></summary> <table><thead><tr><th></th><th align="left">Relevant files</th></tr></thead><tbody><tr><td><strong>Enhancement</strong></td><td><details><summary>3 files</summary><table> <tr> <td><strong>usage.rs</strong><dd><code>Add Pipeline and RemotePipeline usage types and mapping</code>    </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8295/files#diff-b15a6adcb163dc27993c83528105a6726c44c86d8a84c5cf5d1d382c6ec8cc29">+22/-0</a>    </td> </tr> <tr> <td><strong>ingest.rs</strong><dd><code>Report size before pipeline; avoid double counting</code>              </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8295/files#diff-e62118e45be183266d39ffa08491f255017954235ff0f3d6937683e314da406f">+13/-7</a>    </td> </tr> <tr> <td><strong>batch_execution.rs</strong><dd><code>Add usage reporting; introduce PipelineItem; report remote WAL usage</code></dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8295/files#diff-3f6824a6eddaf1603e5cb0f1a4311ac6a45e1ffb967266ccc1f0bba944625e69">+158/-29</a></td> </tr> </table></details></td></tr><tr><td><strong>Bug fix</strong></td><td><details><summary>1 files</summary><table> <tr> <td><strong>handlers.rs</strong><dd><code>Avoid ownership by cloning stream params key</code>                          </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8295/files#diff-5eca8995a88fcecb6d6b8efd3c7c4496f4a96a4b060fb4ed85593574f7cf49eb">+1/-1</a>      </td> </tr> </table></details></td></tr><tr><td><strong>Formatting</strong></td><td><details><summary>3 files</summary><table> <tr> <td><strong>otlp.rs</strong><dd><code>Minor formatting and flow tidy in loop</code>                                      </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8295/files#diff-183498d98b17ddf344cbab4da6b2d9fe89ac81112db51655c142a89023b46ef3">+1/-0</a>      </td> </tr> <tr> <td><strong>json.rs</strong><dd><code>Minor formatting; prep for pipeline handling</code>                          </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8295/files#diff-506987e61095a54e47f7f208dad89f9f1bce595fa6bf1ba8a7d746f2b2ff6c4a">+1/-0</a>      </td> </tr> <tr> <td><strong>prom.rs</strong><dd><code>Minor formatting; prep for pipeline handling</code>                          </dd></td> <td><a href="https://github.com/openobserve/openobserve/pull/8295/files#diff-c4e9e1d49a2bd0fefe67684b3768c4792be63cebcf727d71a48141f5b5957b8c">+1/-0</a>      </td> </tr> </table></details></td></tr></tr></tbody></table> </details> ___ --------- Co-authored-by: sai nikhil kethe <nikhil@openobserve.ai> Co-authored-by: Shrinath Rao <shnath@openobserve.ai>
2026-11-04 08:09:27 +00:00 · 2025-09-17 19:02:41 +05:30
parent 4f79d76323
commit c77a559183
13 changed files with 331 additions and 82 deletions
@@ -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<UsageType> 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"),
@@ -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);
                                }
@@ -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;
                    }
@@ -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(
@@ -232,6 +232,7 @@ pub async fn ingest(org_id: &str, body: web::Bytes) -> Result<IngestionResponse>
                        if stream_params.stream_type != StreamType::Metrics {
                            continue;
                        }
+
                        // add partition keys
                        if !stream_partitioning_map.contains_key(stream_params.stream_name.as_str())
                        {
@@ -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())
                        {
@@ -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::<f64>()
+            / 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::<PipelineItem>(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<Sender<(usize, Value, bool)>>,
+    mut receiver: Receiver<PipelineItem>,
+    mut child_senders: Vec<Sender<PipelineItem>>,
    vrl_runtime: Option<(VRLResultResolver, bool)>,
    result_sender: Option<Sender<(usize, StreamParams, Value)>>,
    error_sender: Sender<(String, String, String, Option<String>)>,
@@ -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<PipelineItem>],
+    item: PipelineItem,
    node_type: &str,
 ) {
    if child_senders.len() == 1 {
@@ -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() {
@@ -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');
@@ -0,0 +1,11 @@
+  <svg
+    width="24"
+    height="24"
+    viewBox="0 0 24 24"
+    xmlns="http://www.w3.org/2000/svg"
+  >
+    <path
+    fill='#4637F5'
+      d="M22.0006 18.5019C22.0005 19.4686 21.6587 20.2935 20.9753 20.9768C20.2919 21.6601 19.4669 22.0016 18.5002 22.0016C17.5336 22.0015 16.7086 21.6597 16.0253 20.9763C15.3421 20.2929 15.0005 19.4679 15.0006 18.5012C15.0006 18.2012 15.0381 17.9096 15.1132 17.6262C15.1882 17.3429 15.2924 17.0763 15.4257 16.8263L11.6011 13.0009L8.85109 13.0006C8.63436 13.734 8.2218 14.3339 7.61342 14.8005C7.00505 15.2671 6.30086 15.5004 5.50086 15.5003C4.53419 15.5002 3.70922 15.1585 3.02595 14.4751C2.34269 13.7917 2.0011 12.9667 2.00119 12C2.00128 11.0333 2.34302 10.2084 3.02642 9.5251C3.70982 8.84183 4.53485 8.50024 5.50152 8.50033C6.30152 8.50041 7.00566 8.73381 7.61395 9.20053C8.22224 9.66725 8.63469 10.2673 8.85128 11.0006L11.6013 11.0009L15.4266 7.15127C15.2933 6.90126 15.1892 6.63875 15.1142 6.36374C15.0392 6.08873 15.0018 5.80123 15.0018 5.50123C15.0019 4.53456 15.3436 3.70959 16.027 3.02632C16.7104 2.34306 17.5355 2.00147 18.5021 2.00156C19.4688 2.00165 20.2938 2.34339 20.977 3.02679C21.6603 3.71019 22.0019 4.53522 22.0018 5.50189C22.0017 6.46856 21.66 7.29352 20.9766 7.97679C20.2932 8.66006 19.4681 9.00165 18.5015 9.00156C18.2015 9.00153 17.9098 8.964 17.6265 8.88898C17.3432 8.81395 17.0765 8.70976 16.8265 8.5764L13.4012 12.0011L16.8259 15.4264C17.0759 15.2931 17.3426 15.1889 17.6259 15.114C17.9092 15.039 18.2009 15.0015 18.5009 15.0016C19.4676 15.0016 20.2925 15.3434 20.9758 16.0268C21.6591 16.7102 22.0007 17.5352 22.0006 18.5019ZM20.0018 5.5017C20.0018 5.08503 19.856 4.73085 19.5644 4.43916C19.2728 4.14746 18.9186 4.0016 18.5019 4.00156C18.0853 4.00152 17.7311 4.14732 17.4394 4.43896C17.1477 4.7306 17.0018 5.08475 17.0018 5.50142C17.0018 5.91808 17.1476 6.27226 17.4392 6.56396C17.7308 6.85565 18.085 7.00152 18.5017 7.00156C18.9183 7.0016 19.2725 6.8558 19.5642 6.56416C19.8559 6.27252 20.0018 5.91837 20.0018 5.5017ZM7.00119 12.0005C7.00123 11.5838 6.85543 11.2296 6.56379 10.9379C6.27215 10.6462 5.918 10.5004 5.50133 10.5003C5.08466 10.5003 4.73048 10.6461 4.43879 10.9377C4.14709 11.2294 4.00123 11.5835 4.00119 12.0002C4.00115 12.4169 4.14695 12.771 4.43859 13.0627C4.73023 13.3544 5.08438 13.5003 5.50105 13.5003C5.91771 13.5004 6.27189 13.3546 6.56359 13.0629C6.85528 12.7713 7.00115 12.4171 7.00119 12.0005ZM20.0006 18.5017C20.0006 18.085 19.8548 17.7309 19.5632 17.4392C19.2715 17.1475 18.9174 17.0016 18.5007 17.0016C18.084 17.0015 17.7299 17.1473 17.4382 17.439C17.1465 17.7306 17.0006 18.0848 17.0006 18.5014C17.0005 18.9181 17.1463 19.2723 17.438 19.564C17.7296 19.8557 18.0838 20.0015 18.5004 20.0016C18.9171 20.0016 19.2713 19.8558 19.563 19.5642C19.8547 19.2725 20.0005 18.9184 20.0006 18.5017Z"
+    />
+  </svg>
@@ -0,0 +1,5 @@
+<svg width="24" height="24" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+<path fill-rule="evenodd" clip-rule="evenodd" d="M9.65407 11.1628V12.8372C9.65407 13.3368 9.85253 13.8159 10.2058 14.1692C10.5591 14.5225 11.0382 14.7209 11.5378 14.7209H13.2122C13.7118 14.7209 14.1909 14.5225 14.5442 14.1692C14.8975 13.8159 15.0959 13.3368 15.0959 12.8372V11.1628C15.0959 10.6632 14.8975 10.1841 14.5442 9.8308C14.1909 9.47753 13.7118 9.27907 13.2122 9.27907H11.5378C11.0382 9.27907 10.5591 9.47753 10.2058 9.8308C9.85253 10.1841 9.65407 10.6632 9.65407 11.1628ZM3.375 4.88372V6.55814C3.375 7.05773 3.57346 7.53686 3.92673 7.89013C4.28 8.2434 4.75913 8.44186 5.25872 8.44186H6.93314C7.43273 8.44186 7.91186 8.2434 8.26513 7.89013C8.6184 7.53686 8.81686 7.05773 8.81686 6.55814V4.88372C8.81686 4.38413 8.6184 3.905 8.26513 3.55173C7.91186 3.19846 7.43273 3 6.93314 3H5.25872C4.75913 3 4.28 3.19846 3.92673 3.55173C3.57346 3.905 3.375 4.38413 3.375 4.88372ZM15.9331 17.4419V19.1163C15.9331 19.6159 16.1316 20.095 16.4849 20.4483C16.8381 20.8015 17.3173 21 17.8169 21H19.4913C19.9909 21 20.47 20.8015 20.8233 20.4483C21.1765 20.095 21.375 19.6159 21.375 19.1163V17.4419C21.375 16.9423 21.1765 16.4631 20.8233 16.1099C20.47 15.7566 19.9909 15.5581 19.4913 15.5581H17.8169C17.3173 15.5581 16.8381 15.7566 16.4849 16.1099C16.1316 16.4631 15.9331 16.9423 15.9331 17.4419Z" fill="#DB2777"/>
+<path fill-rule="evenodd" clip-rule="evenodd" d="M8.18895 6.34883H19.0727C19.3501 6.34927 19.616 6.45967 19.8122 6.65584C20.0083 6.852 20.1187 7.11793 20.1192 7.39534V10.3256C20.1187 10.603 20.0083 10.8689 19.8122 11.0651C19.616 11.2612 19.3501 11.3716 19.0727 11.3721H16.561C16.3945 11.3721 16.2348 11.4382 16.117 11.556C15.9993 11.6738 15.9331 11.8335 15.9331 12C15.9331 12.1665 15.9993 12.3262 16.117 12.444C16.2348 12.5617 16.3945 12.6279 16.561 12.6279H19.0727C19.683 12.6279 20.269 12.3851 20.701 11.9539C21.1324 11.5218 21.3747 10.9362 21.375 10.3256V7.39534C21.375 6.78502 21.1322 6.19897 20.701 5.76697C20.2689 5.33564 19.6833 5.09327 19.0727 5.09302H8.18895C8.02242 5.09302 7.86271 5.15917 7.74496 5.27693C7.6272 5.39468 7.56105 5.55439 7.56105 5.72092C7.56105 5.88746 7.6272 6.04717 7.74496 6.16492C7.86271 6.28268 8.02242 6.34883 8.18895 6.34883ZM13.2122 17.6512H5.67733C5.39991 17.6507 5.13398 17.5403 4.93782 17.3442C4.74166 17.148 4.63126 16.8821 4.63081 16.6046V13.6744C4.63126 13.397 4.74166 13.1311 4.93782 12.9349C5.13398 12.7387 5.39991 12.6283 5.67733 12.6279H10.282C10.4485 12.6279 10.6082 12.5617 10.726 12.444C10.8437 12.3262 10.9099 12.1665 10.9099 12C10.9099 11.8335 10.8437 11.6738 10.726 11.556C10.6082 11.4382 10.4485 11.3721 10.282 11.3721H5.67733C5.067 11.3721 4.48095 11.6149 4.04895 12.046C3.61762 12.4782 3.37526 13.0638 3.375 13.6744V16.6046C3.375 17.215 3.61779 17.801 4.04895 18.233C4.48114 18.6644 5.06672 18.9067 5.67733 18.907H13.2122C13.3787 18.907 13.5385 18.8408 13.6562 18.7231C13.774 18.6053 13.8401 18.4456 13.8401 18.2791C13.8401 18.1125 13.774 17.9528 13.6562 17.8351C13.5385 17.7173 13.3787 17.6512 13.2122 17.6512Z" fill="#DB2777"/>
+<path fill-rule="evenodd" clip-rule="evenodd" d="M18.6793 13.2307L17.4494 12L18.6793 10.7693C18.7375 10.711 18.7838 10.6418 18.8153 10.5657C18.8468 10.4896 18.8631 10.408 18.8631 10.3256C18.8631 10.2432 18.8468 10.1616 18.8153 10.0854C18.7838 10.0093 18.7375 9.94013 18.6793 9.88185C18.621 9.82358 18.5518 9.77736 18.4757 9.74583C18.3995 9.71429 18.3179 9.69806 18.2355 9.69806C18.1531 9.69806 18.0715 9.71429 17.9954 9.74583C17.9193 9.77736 17.8501 9.82358 17.7918 9.88185L16.1174 11.5563C16.0591 11.6145 16.0128 11.6837 15.9812 11.7598C15.9497 11.836 15.9334 11.9176 15.9334 12C15.9334 12.0824 15.9497 12.164 15.9812 12.2402C16.0128 12.3163 16.0591 12.3855 16.1174 12.4437L17.7918 14.1181C17.8501 14.1764 17.9193 14.2226 17.9954 14.2542C18.0715 14.2857 18.1531 14.3019 18.2355 14.3019C18.3179 14.3019 18.3995 14.2857 18.4757 14.2542C18.5518 14.2226 18.621 14.1764 18.6793 14.1181C18.7375 14.0599 18.7838 13.9907 18.8153 13.9146C18.8468 13.8384 18.8631 13.7568 18.8631 13.6744C18.8631 13.592 18.8468 13.5104 18.8153 13.4343C18.7838 13.3581 18.7375 13.289 18.6793 13.2307ZM12.8188 20.3972L14.4932 18.7228C14.5515 18.6645 14.5978 18.5954 14.6294 18.5192C14.6609 18.4431 14.6772 18.3615 14.6772 18.2791C14.6772 18.1966 14.6609 18.115 14.6294 18.0389C14.5978 17.9628 14.5515 17.8936 14.4932 17.8353L12.8188 16.1609C12.7011 16.0432 12.5415 15.9771 12.3751 15.9771C12.2086 15.9771 12.049 16.0432 11.9314 16.1609C11.8137 16.2786 11.7476 16.4382 11.7476 16.6046C11.7476 16.7711 11.8137 16.9307 11.9314 17.0484L13.1612 18.2791L11.9314 19.5098C11.8731 19.568 11.8269 19.6372 11.7953 19.7133C11.7638 19.7895 11.7476 19.8711 11.7476 19.9535C11.7476 20.0359 11.7638 20.1175 11.7953 20.1936C11.8269 20.2698 11.8731 20.3389 11.9314 20.3972C11.9896 20.4555 12.0588 20.5017 12.1349 20.5332C12.2111 20.5648 12.2927 20.581 12.3751 20.581C12.4575 20.581 12.5391 20.5648 12.6152 20.5332C12.6913 20.5017 12.7605 20.4555 12.8188 20.3972Z" fill="#DB2777"/>
+</svg>
@@ -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"
    });
  });

@@ -100,7 +100,8 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
        </div>
        <!-- new section introduced to show the usage for ingestion , search , functions -->
        <div v-if="!dataLoading && Object.keys(usageData).length > 0" class="row wrap justify-evenly q-gutter-md ">
-            <div class="tile" style="min-width: 240px; flex-grow: 1; max-width: 100%; ">
+            <div class="tw-grid tw-grid-cols-3 tw-gap-4 tw-w-full">
+                <div class="tile" style="min-width: 240px; flex-grow: 1; max-width: 100%; ">
              <div class="tile-content q-pa-md rounded-borders text-center column justify-between "
              :class="store.state.theme === 'dark' ? 'dark-usage-tile-content' : 'light-usage-tile-content'"
               style="height: 100%; gap: 12px">
@@ -120,50 +121,94 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
               {{ usageData.ingestion }} {{ usageDataType.toUpperCase() }}
            </div>
            </div>
-            </div>
-            <div class="tile" style="min-width: 240px; flex-grow: 1; max-width: 100%;">
-              <div class="tile-content q-pa-md rounded-borders text-center column justify-between "
-              :class="store.state.theme === 'dark' ? 'dark-usage-tile-content' : 'light-usage-tile-content'"
-               style="height: 100%; gap: 12px">
-              <!-- Top Section (60%) -->
-              <div class="column justify-between">
-                <!-- Title row -->
-                <div class="row justify-between">
-                  <div class="usage-tile-title">Search</div>
-                  <div style="opacity: 0.8;">
-                    <img :src="searchIcon" />
-                  </div>
                </div>
-              </div>
-
-            <!-- Bottom Section (40%) -->
-            <div class="usage-data-to-display row items-end ">
-              {{ usageData.search }} {{ usageDataType.toUpperCase() }}
-            </div>
-            </div>
-            </div>
-            <div class="tile" style="min-width: 240px; flex-grow: 1; max-width: 100%;">
-              <div class="tile-content q-pa-md rounded-borders text-center column justify-between "
-              :class="store.state.theme === 'dark' ? 'dark-usage-tile-content' : 'light-usage-tile-content'"
-               style="height: 100%; gap: 12px">
-              <!-- Top Section (60%) -->
-              <div class="column justify-between">
-                <!-- Title row -->
-                <div class="row justify-between">
-                  <div class="usage-tile-title">Functions</div>
-                  <div style="opacity: 0.8;">
-                    <img :src="functionsIcon" />
-                  </div>
+                <div class="tile" style="min-width: 240px; flex-grow: 1; max-width: 100%;">
+                <div class="tile-content q-pa-md rounded-borders text-center column justify-between "
+                :class="store.state.theme === 'dark' ? 'dark-usage-tile-content' : 'light-usage-tile-content'"
+                style="height: 100%; gap: 12px">
+                <!-- Top Section (60%) -->
+                <div class="column justify-between">
+                    <!-- Title row -->
+                    <div class="row justify-between">
+                    <div class="usage-tile-title">Search</div>
+                    <div style="opacity: 0.8;">
+                        <img :src="searchIcon" />
+                    </div>
+                    </div>
                </div>
-              </div>

-            <!-- Bottom Section (40%) -->
-            <div class="usage-data-to-display row items-end ">
-              {{ usageData.functions }} {{ usageDataType.toUpperCase() }}
-            </div>
-            </div>
-            </div>
+                <!-- Bottom Section (40%) -->
+                <div class="usage-data-to-display row items-end ">
+                {{ usageData.search }} {{ usageDataType.toUpperCase() }}
+                </div>
+                </div>
+                </div>
+                <div class="tile" style="min-width: 240px; flex-grow: 1; max-width: 100%;">
+                <div class="tile-content q-pa-md rounded-borders text-center column justify-between "
+                :class="store.state.theme === 'dark' ? 'dark-usage-tile-content' : 'light-usage-tile-content'"
+                style="height: 100%; gap: 12px">
+                <!-- Top Section (60%) -->
+                <div class="column justify-between">
+                    <!-- Title row -->
+                    <div class="row justify-between">
+                    <div class="usage-tile-title">Functions</div>
+                    <div style="opacity: 0.8;">
+                        <img :src="functionsIcon" />
+                    </div>
+                    </div>
+                </div>

+                <!-- Bottom Section (40%) -->
+                <div class="usage-data-to-display row items-end ">
+                {{ usageData.functions }} {{ usageDataType.toUpperCase() }}
+                </div>
+                </div>
+                </div>
+            </div>
+            <div class="tw-grid tw-grid-cols-2 tw-gap-4 tw-w-full">
+                <div class="tile" style="min-width: 240px; flex-grow: 1; max-width: 100%;">
+                <div class="tile-content q-pa-md rounded-borders text-center column justify-between "
+                :class="store.state.theme === 'dark' ? 'dark-usage-tile-content' : 'light-usage-tile-content'"
+                style="height: 100%; gap: 12px">
+                <!-- Top Section (60%) -->
+                <div class="column justify-between">
+                    <!-- Title row -->
+                    <div class="row justify-between">
+                    <div class="usage-tile-title">Pipelines</div>
+                    <div style="opacity: 0.8;" class="tw-bg-[#E6EFFE] tw-flex tw-items-center tw-justify-center tw-rounded-[9px] tw-h-[33px] tw-w-[33px]">
+                        <img :src="pipelineIcon" class="tw-h-[18px] tw-w-[18px] tw-mx-[7px] tw-my-[7px]" />
+                    </div>
+                    </div>
+                </div>
+
+                <!-- Bottom Section (40%) -->
+                <div class="usage-data-to-display row items-end ">
+                {{ usageData.pipeline }} {{ usageDataType.toUpperCase() }}
+                </div>
+                </div>
+                </div>
+                <div class="tile" style="min-width: 240px; flex-grow: 1; max-width: 100%;">
+                <div class="tile-content q-pa-md rounded-borders text-center column justify-between "
+                :class="store.state.theme === 'dark' ? 'dark-usage-tile-content' : 'light-usage-tile-content'"
+                style="height: 100%; gap: 12px">
+                <!-- Top Section (60%) -->
+                <div class="column justify-between">
+                    <!-- Title row -->
+                    <div class="row justify-between">
+                    <div class="usage-tile-title">Remote Pipelines</div>
+                    <div style="opacity: 0.8;" class="tw-bg-[#F2DCF5] tw-flex tw-items-center tw-justify-center tw-rounded-[9px] tw-h-[33px] tw-w-[33px]">
+                        <img :src="remotePipelineIcon" class="tw-h-[18px] tw-w-[18px] tw-mx-[7px] tw-my-[7px]" />
+                    </div>
+                    </div>
+                </div>
+
+                <!-- Bottom Section (40%) -->
+                <div class="usage-data-to-display row items-end ">
+                {{ usageData.remotepipeline }} {{ usageDataType.toUpperCase() }}
+                </div>
+                </div>
+                </div>
+            </div>
        </div>
        <!-- charts section -->
        <div>
@@ -212,7 +257,9 @@ import CustomChartRenderer from "@/components/dashboards/panels/CustomChartRende
      const usageData = ref<any>({
        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,
      };
    },
  });