feat: Pipeline integration v2 (#4732)

This PR implements the new design of **Pipeline** introduced by #4432 

---

**2 types of pipelines differentiates by the source** 

Type I: **Realtime**
- source is a stream (Logs, Metrics, & Traces)
- takes effect when the source stream is being ingested

Type II: **Scheduled** (aka DerivedSteram)
- source is a SQL query
- takes effect based on the given schedule

The new pipeline is represented as a graph consists of different types
of Nodes that are connected by edges.
**Pipeline Data Model**
```rs
pub struct Pipeline {
    pub id: String,
    pub version: i32,
    pub enabled: bool,
    pub org: String,
    pub name: String,
    pub description: String,
    pub source: PipelineSource,
    pub nodes: Vec<Node>,
    pub edges: Vec<Edge>,
}
```
**4 types of node**
I. StreamNode: either used for source node of realtime pipeline or for
destination nodes
II. QueryNode: used for source node of scheduled pipeline
III. FunctionNode: used for executing vrl functions
IV. ConditionNode: used for checking routing conditions

Rules applied to validating a pipeline when it's created:
1. non-empty nodes list
2. non-empty edges list
3. 1st node in nodes list is either StreamNode or QueryNode
4. non-empty `conditions` in all ConditionNode nodes in nodes list
5. all leaf nodes are of type StreamNode
6. In the same branch, unchecked `after_flattened` FunctionNode can't
follow checked
   `after_flattened` checked FunctionNode

Pipeline execution is implemented with the struct `ExecutablePipeline`,
a ready-execute pipeline object cached in memory based on the Pipeline
objects modified on the UI side.
`ExecutablePipeline` object
- builds the relationships among all the nodes based on the edges
- topologically sorts the nodes based on the level to determine which
node to process first
- registers all the vrl function node once

`ExecutablePipeline` object processes ingested records in batch.
Starting from the source node, each connecting edge has a channel
connecting the parent and child node, used for passing the records. This
approach enables batch processing in parallel.

---

**Deprecated Features**
1. previous implementation of pipeline
2. Function x Stream associations
 - Functions can still be added/edited/removed as before
- Needs to create a pipeline in order to apply chosen functions to the
desired stream

The new release when this pr is merged will automatically migrate **old
pipelines** and **Function x Stream associations** to new pipeline
format and can be found in `Stream Pipelines` tab with `Migrated-`
prefix names.
**_Note: Auto generated pipelines are paused by default. Please verify
the pipelines before enabling them_**


Co-authored-by: Taiming Liu <liutaiming3@gmail.com>
Co-authored-by: Bhargav <BJP232004@GMAIL.COM>

Author: 3 people

Cargo.lock

@@ -99,6 +99,14 @@ pub async fn cli() -> Result<bool, anyhow::Error> {
                         .help("migrate to: sqlite, etcd, mysql, postgresql"),
                 ]),
             clap::Command::new("migrate-dashboards").about("migrate-dashboards"),
+            clap::Command::new("migrate-pipeline").about("migrate pipelines")
+                .arg(
+                    clap::Arg::new("drop-table")
+                        .long("drop-table")
+                        .required(false)
+                        .num_args(0)
+                        .help("Drop existing Pipeline table first before migrating")
+                ),
             clap::Command::new("delete-parquet")
                 .about("delete parquet files from s3 and file_list")
                 .arg(
@@ -246,6 +254,11 @@ pub async fn cli() -> Result<bool, anyhow::Error> {
             println!("Running migration dashboard");
             migration::dashboards::run().await?
         }
+        "migrate-pipeline" => {
+            println!("Running migration pipeline");
+            let drop_table = command.get_flag("drop-table");
+            migration::pipeline_func::run(drop_table).await?;
+        }
         "delete-parquet" => {
             let file = command.get_one::<String>("file").unwrap();
             match file_list::delete_parquet_file(file, true).await {

src/cli/basic/cli.rs

@@ -15,7 +15,14 @@
 
 use std::sync::Arc;
 
-use config::{RwAHashMap, RwHashMap};
+use config::{
+    meta::{
+        alerts::{alert::Alert, destinations::Destination, templates::Template},
+        function::Transform,
+        stream::StreamParams,
+    },
+    RwAHashMap, RwHashMap,
+};
 use dashmap::DashMap;
 use hashbrown::HashMap;
 use infra::table::short_urls::ShortUrlRecord;
@@ -25,18 +32,12 @@ use vector_enrichment::TableRegistry;
 
 use crate::{
     common::meta::{
-        alerts::{alert::Alert, destinations::Destination, templates::Template},
-        dashboards::reports,
-        functions::{StreamFunctionsList, Transform},
-        maxmind::MaxmindClient,
-        organization::OrganizationSetting,
-        pipelines::PipeLine,
-        prom::ClusterLeader,
-        syslog::SyslogRoute,
-        user::User,
+        dashboards::reports, maxmind::MaxmindClient, organization::OrganizationSetting,
+        prom::ClusterLeader, syslog::SyslogRoute, user::User,
     },
     service::{
         db::scheduler as db_scheduler, enrichment::StreamTable, enrichment_table::geoip::Geoip,
+        pipeline::batch_execution::ExecutablePipeline,
     },
 };
 
@@ -47,8 +48,6 @@ pub static BUILD_DATE: &str = env!("GIT_BUILD_DATE");
 
 // global cache variables
 pub static KVS: Lazy<RwHashMap<String, bytes::Bytes>> = Lazy::new(Default::default);
-pub static STREAM_FUNCTIONS: Lazy<RwHashMap<String, StreamFunctionsList>> =
-    Lazy::new(DashMap::default);
 pub static QUERY_FUNCTIONS: Lazy<RwHashMap<String, Transform>> = Lazy::new(DashMap::default);
 pub static USERS: Lazy<RwHashMap<String, User>> = Lazy::new(DashMap::default);
 pub static USERS_RUM_TOKEN: Lazy<Arc<RwHashMap<String, User>>> =
@@ -83,6 +82,7 @@ pub static GEOIP_CITY_TABLE: Lazy<Arc<RwLock<Option<Geoip>>>> =
 pub static GEOIP_ASN_TABLE: Lazy<Arc<RwLock<Option<Geoip>>>> =
     Lazy::new(|| Arc::new(RwLock::new(None)));
 
+pub static STREAM_EXECUTABLE_PIPELINES: Lazy<RwAHashMap<StreamParams, ExecutablePipeline>> =
+    Lazy::new(Default::default);
 pub static USER_SESSIONS: Lazy<RwHashMap<String, String>> = Lazy::new(Default::default);
-pub static STREAM_PIPELINES: Lazy<RwHashMap<String, PipeLine>> = Lazy::new(DashMap::default);
 pub static SHORT_URLS: Lazy<RwHashMap<String, ShortUrlRecord>> = Lazy::new(DashMap::default);

src/common/infra/config.rs

@@ -13,10 +13,8 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-pub mod alerts;
 pub mod authz;
 pub mod dashboards;
-pub mod functions;
 pub mod http;
 pub mod ingestion;
 pub mod maxmind;

src/common/meta/mod.rs

@@ -13,11 +13,10 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+use config::meta::{alerts::alert::Alert, function::Transform};
 use serde::{Deserialize, Serialize};
 use utoipa::ToSchema;
 
-use super::{alerts::alert::Alert, functions::Transform};
-
 pub const DEFAULT_ORG: &str = "default";
 pub const CUSTOM: &str = "custom";
 pub const THRESHOLD: i64 = 9383939382;

src/common/meta/organization.rs

@@ -16,16 +16,15 @@
 use std::collections::HashMap;
 
 use config::{
-    meta::stream::{RoutingCondition, StreamType},
+    meta::{
+        alerts::derived_streams::DerivedStreamMeta,
+        stream::{RoutingCondition, StreamType},
+    },
     utils::json::Value,
 };
 use serde::{Deserialize, Serialize};
 use utoipa::ToSchema;
 
-use crate::common::meta::{
-    alerts::derived_streams::DerivedStreamMeta, functions::StreamFunctionsList,
-};
-
 #[derive(Clone, Debug, Serialize, Deserialize, ToSchema, PartialEq)]
 pub struct PipeLine {
     pub name: String,
@@ -42,42 +41,3 @@ pub struct PipeLine {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub meta: Option<HashMap<String, Value>>,
 }
-
-impl PipeLine {
-    pub fn into_response(self, functions: Option<StreamFunctionsList>) -> PipeLineResponse {
-        PipeLineResponse {
-            name: self.name,
-            description: self.description,
-            stream_name: self.stream_name,
-            stream_type: self.stream_type,
-            routing: self.routing,
-            derived_streams: self.derived_streams,
-            functions,
-            meta: self.meta,
-        }
-    }
-}
-
-#[derive(Clone, Debug, Serialize, Deserialize, ToSchema, PartialEq)]
-pub struct PipeLineResponse {
-    pub name: String,
-    #[serde(default)]
-    pub description: String,
-    #[serde(default)]
-    pub stream_name: String,
-    #[serde(default)]
-    pub stream_type: StreamType,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub routing: Option<HashMap<String, Vec<RoutingCondition>>>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub functions: Option<StreamFunctionsList>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub derived_streams: Option<Vec<DerivedStreamMeta>>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub meta: Option<HashMap<String, Value>>,
-}
-
-#[derive(Clone, Debug, Serialize, Deserialize, ToSchema)]
-pub struct PipeLineList {
-    pub list: Vec<PipeLineResponse>,
-}

src/common/meta/pipelines.rs

@@ -251,12 +251,6 @@ pub async fn add_zo_info(mut data: HashMap<String, json::Value>) -> HashMap<Stri
         format!("{:.0}", (traces_compressed_size / SIZE_IN_MB)).into(),
     );
 
-    let iter = STREAM_FUNCTIONS.iter().clone();
-    let mut ingest_functions = 0;
-    for item in iter {
-        ingest_functions += item.value().list.len()
-    }
-    data.insert("num_ingest_functions".to_string(), ingest_functions.into());
     data.insert(
         "num_query_functions".to_string(),
         QUERY_FUNCTIONS.len().into(),

src/common/meta/telemetry.rs

@@ -20,6 +20,7 @@ use version_compare::Version;
 pub mod dashboards;
 pub mod file_list;
 pub mod meta;
+pub mod pipeline_func;
 pub mod schema;
 
 pub async fn check_upgrade(old_ver: &str, new_ver: &str) -> Result<(), anyhow::Error> {
@@ -46,6 +47,11 @@ pub async fn check_upgrade(old_ver: &str, new_ver: &str) -> Result<(), anyhow::E
         upgrade_092_093().await?;
     }
 
+    let v131 = Version::from("v0.13.1").unwrap();
+    if old_ver < v131 {
+        upgrade_130_131().await?;
+    }
+
     Ok(())
 }
 
@@ -66,6 +72,13 @@ async fn upgrade_092_093() -> Result<(), anyhow::Error> {
     Ok(())
 }
 
+async fn upgrade_130_131() -> Result<(), anyhow::Error> {
+    // migrate pipelines and function associations
+    pipeline_func::run(false).await?;
+
+    Ok(())
+}
+
 pub async fn upgrade_resource_names() -> Result<(), anyhow::Error> {
     // The below migration requires ofga init ready, but on Router node,
     // we don't initialize ofga, hence the migration should not run on router