fix: format stream name for _bulk api (#3371)

src/handler/http/request/stream/mod.rs

@@ -19,7 +19,7 @@ use std::{
 };
 
 use actix_web::{delete, get, http, put, web, HttpRequest, HttpResponse, Responder};
-use config::meta::stream::{StreamSettings, StreamType};
+use config::meta::stream::{RoutingCondition, StreamSettings, StreamType};
 
 use crate::{
     common::{
@@ -30,7 +30,7 @@ use crate::{
         },
         utils::http::get_stream_type_from_request,
     },
-    service::stream,
+    service::{format_stream_name, stream},
 };
 
 /// GetSchema
@@ -93,10 +93,11 @@ async fn schema(
 #[put("/{org_id}/streams/{stream_name}/settings")]
 async fn settings(
     path: web::Path<(String, String)>,
-    settings: web::Json<StreamSettings>,
+    mut settings: web::Json<StreamSettings>,
     req: HttpRequest,
 ) -> Result<HttpResponse, Error> {
     let (org_id, stream_name) = path.into_inner();
+    let stream_name = format_stream_name(&stream_name);
     let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
     let stream_type = match get_stream_type_from_request(&query) {
         Ok(v) => {
@@ -123,6 +124,15 @@ async fn settings(
             );
         }
     };
+    // if routing is provided format the key using format_stream_name
+    if let Some(ref mut routing_map) = settings.routing {
+        let new_routing: hashbrown::HashMap<String, Vec<RoutingCondition>> = routing_map
+            .drain()
+            .map(|(key, value)| (format_stream_name(&key), value))
+            .collect();
+        settings.routing = Some(new_routing);
+    }
+
     let stream_type = stream_type.unwrap_or(StreamType::Logs);
     stream::save_stream_settings(&org_id, &stream_name, stream_type, settings.into_inner()).await
 }

src/handler/http/router/mod.rs

@@ -125,7 +125,7 @@ async fn audit_middleware(
         if res.response().error().is_none() {
             let body = if path.ends_with("/settings/logo") {
                 // Binary data, encode it with base64
-                general_purpose::STANDARD.encode(request_body.to_vec())
+                general_purpose::STANDARD.encode(&request_body)
             } else {
                 String::from_utf8(request_body.to_vec()).unwrap()
             };

src/service/ingestion/mod.rs

@@ -45,7 +45,7 @@ use crate::{
         },
         utils::functions::get_vrl_compiler_config,
     },
-    service::{db, format_partition_key},
+    service::{db, format_partition_key, format_stream_name},
 };
 
 pub mod grpc;
@@ -433,7 +433,7 @@ pub async fn get_stream_routing(
         .unwrap_or_default()
         .iter()
         .map(|(k, v)| Routing {
-            destination: k.to_string(),
+            destination: format_stream_name(k),
             routing: v.clone(),
         })
         .collect();

src/service/logs/bulk.rs

@@ -46,7 +46,7 @@ use crate::{
         stream::StreamParams,
     },
     service::{
-        db,
+        db, format_stream_name,
         ingestion::{evaluate_trigger, write_file, TriggerAlertData},
         metadata::{distinct_values::DvItem, write, MetadataItem, MetadataType},
         schema::{
@@ -132,6 +132,8 @@ pub async fn ingest(
             }
             (action, stream_name, doc_id) = ret.unwrap();
 
+            stream_name = format_stream_name(&stream_name);
+
             // skip blocked streams
             let key = format!("{org_id}/{}/{stream_name}", StreamType::Logs);
             if BLOCKED_STREAMS.contains(&key.as_str()) {