fix: add stringSeparated output format for remote pipeline (main) (#10555)

cherry pick https://github.com/openobserve/openobserve/pull/10540 from
v40 to main

---------

Signed-off-by: Yashodhan Joshi <yjdoc2@gmail.com>
Co-authored-by: Yashodhan Joshi <yjdoc2@gmail.com>
Co-authored-by: Bhargav <BJP232004@GMAIL.COM>
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

src/config/src/meta/destinations.rs

@@ -149,6 +149,9 @@ pub enum HTTPOutputFormat {
     ESBulk {
         index: String,
     },
+    StringSeparated {
+        separator: String,
+    },
 }
 
 impl HTTPOutputFormat {
@@ -158,6 +161,8 @@ impl HTTPOutputFormat {
             Self::NDJSON => "application/x-ndjson",
             Self::NestedEvent => "application/x-ndjson",
             Self::ESBulk { .. } => "application/x-ndjson",
+            // this is not a json anymore, the handler must process it as a string
+            Self::StringSeparated { .. } => "text/plain",
         }
     }
 
@@ -205,6 +210,12 @@ impl HTTPOutputFormat {
                 temp.push(b'\n');
                 temp
             }
+            Self::StringSeparated { separator } => data
+                .iter()
+                .map(|x| x.to_string())
+                .collect::<Vec<_>>()
+                .join(separator)
+                .into_bytes(),
         }
     }
 }

web/src/components/pipeline/NodeForm/CreateDestinationForm.spec.ts

@@ -936,6 +936,325 @@ describe("CreateDestinationForm", () => {
     });
   });
 
+  describe("StringSeparated Output Format", () => {
+    it("should show separator field when stringseparated format is selected", async () => {
+      wrapper.vm.formData.destination_type = "custom";
+      wrapper.vm.formData.output_format = "stringseparated";
+      wrapper.vm.step = 2;
+      await wrapper.vm.$nextTick();
+      await flushPromises();
+
+      const separatorField = wrapper.find('[data-test="add-destination-separator-input"]');
+      expect(separatorField.exists()).toBe(true);
+    });
+
+    it("should hide separator field when other formats are selected", async () => {
+      wrapper.vm.formData.destination_type = "custom";
+      wrapper.vm.formData.output_format = "json";
+      wrapper.vm.step = 2;
+      await wrapper.vm.$nextTick();
+      await flushPromises();
+
+      const separatorField = wrapper.find('[data-test="add-destination-separator-input"]');
+      expect(separatorField.exists()).toBe(false);
+    });
+
+    it("should initialize separator as empty string", () => {
+      expect(wrapper.vm.formData.separator).toBe("");
+    });
+
+    it("should validate that separator is required for stringseparated format", async () => {
+      wrapper.vm.formData.destination_type = "custom";
+      wrapper.vm.formData.name = "test";
+      wrapper.vm.formData.url = "https://example.com";
+      wrapper.vm.formData.method = "post";
+      wrapper.vm.formData.output_format = "stringseparated";
+      wrapper.vm.formData.separator = "";
+      await wrapper.vm.$nextTick();
+
+      expect(wrapper.vm.canProceedStep2).toBe(false);
+    });
+
+    it("should pass validation when separator is provided for stringseparated format", async () => {
+      wrapper.vm.formData.destination_type = "custom";
+      wrapper.vm.formData.name = "test";
+      wrapper.vm.formData.url = "https://example.com";
+      wrapper.vm.formData.method = "post";
+      wrapper.vm.formData.output_format = "stringseparated";
+      wrapper.vm.formData.separator = "|";
+      await wrapper.vm.$nextTick();
+
+      expect(wrapper.vm.canProceedStep2).toBe(true);
+    });
+
+    it("should allow space as separator", async () => {
+      wrapper.vm.formData.destination_type = "custom";
+      wrapper.vm.formData.name = "test";
+      wrapper.vm.formData.url = "https://example.com";
+      wrapper.vm.formData.method = "post";
+      wrapper.vm.formData.output_format = "stringseparated";
+      wrapper.vm.formData.separator = " ";
+      await wrapper.vm.$nextTick();
+
+      expect(wrapper.vm.canProceedStep2).toBe(true);
+    });
+
+    it("should allow special characters as separator", async () => {
+      const specialChars = ["|", "\\n", "\\t", ",", ";", ":", "-"];
+
+      for (const char of specialChars) {
+        wrapper.vm.formData.destination_type = "custom";
+        wrapper.vm.formData.name = "test";
+        wrapper.vm.formData.url = "https://example.com";
+        wrapper.vm.formData.method = "post";
+        wrapper.vm.formData.output_format = "stringseparated";
+        wrapper.vm.formData.separator = char;
+        await wrapper.vm.$nextTick();
+
+        expect(wrapper.vm.canProceedStep2).toBe(true);
+      }
+    });
+
+    it("should format stringseparated as nested object in payload", async () => {
+      (destinationService.create as any).mockResolvedValue({ data: {} });
+
+      wrapper.vm.formData = {
+        name: "Test StringSeparated",
+        url: "https://example.com",
+        url_endpoint: "/api/logs",
+        method: "post",
+        output_format: "stringseparated",
+        separator: "|",
+        destination_type: "custom",
+        skip_tls_verify: false,
+        template: "",
+        headers: {},
+        emails: "",
+        type: "http",
+      };
+
+      wrapper.vm.apiHeaders = [
+        { key: "Authorization", value: "Bearer token123", uuid: "123" },
+      ];
+
+      await wrapper.vm.createDestination();
+      await flushPromises();
+
+      expect(destinationService.create).toHaveBeenCalledWith(
+        expect.objectContaining({
+          data: expect.objectContaining({
+            output_format: {
+              stringseparated: {
+                separator: "|",
+              },
+            },
+          }),
+        }),
+      );
+    });
+
+    it("should format stringseparated with newline separator in payload", async () => {
+      (destinationService.create as any).mockResolvedValue({ data: {} });
+
+      wrapper.vm.formData = {
+        name: "Test StringSeparated Newline",
+        url: "https://example.com",
+        url_endpoint: "/api/logs",
+        method: "post",
+        output_format: "stringseparated",
+        separator: "\\n",
+        destination_type: "custom",
+        skip_tls_verify: false,
+        template: "",
+        headers: {},
+        emails: "",
+        type: "http",
+      };
+
+      wrapper.vm.apiHeaders = [
+        { key: "Authorization", value: "Bearer token123", uuid: "123" },
+      ];
+
+      await wrapper.vm.createDestination();
+      await flushPromises();
+
+      expect(destinationService.create).toHaveBeenCalledWith(
+        expect.objectContaining({
+          data: expect.objectContaining({
+            output_format: {
+              stringseparated: {
+                separator: "\\n",
+              },
+            },
+          }),
+        }),
+      );
+    });
+
+    it("should parse stringseparated format when editing destination", () => {
+      const destination = {
+        name: "Test Destination",
+        url: "https://example.com/api/logs",
+        method: "post",
+        output_format: {
+          stringseparated: {
+            separator: "|",
+          },
+        },
+        destination_type_name: "custom",
+        skip_tls_verify: false,
+      };
+
+      wrapper.vm.populateFormForEdit(destination);
+
+      expect(wrapper.vm.formData.output_format).toBe("stringseparated");
+      expect(wrapper.vm.formData.separator).toBe("|");
+      expect(wrapper.vm.formData.esbulk_index).toBe("");
+    });
+
+    it("should parse stringseparated with newline separator when editing", () => {
+      const destination = {
+        name: "Test Destination",
+        url: "https://example.com/api/logs",
+        method: "post",
+        output_format: {
+          stringseparated: {
+            separator: "\\n",
+          },
+        },
+        destination_type_name: "custom",
+        skip_tls_verify: false,
+      };
+
+      wrapper.vm.populateFormForEdit(destination);
+
+      expect(wrapper.vm.formData.output_format).toBe("stringseparated");
+      expect(wrapper.vm.formData.separator).toBe("\\n");
+    });
+
+    it("should handle missing separator in stringseparated object when editing", () => {
+      const destination = {
+        name: "Test Destination",
+        url: "https://example.com/api/logs",
+        method: "post",
+        output_format: {
+          stringseparated: {},
+        },
+        destination_type_name: "custom",
+        skip_tls_verify: false,
+      };
+
+      wrapper.vm.populateFormForEdit(destination);
+
+      expect(wrapper.vm.formData.output_format).toBe("stringseparated");
+      expect(wrapper.vm.formData.separator).toBe("");
+    });
+
+    it("should reset separator when format changes from stringseparated to json", async () => {
+      wrapper.vm.formData.output_format = "stringseparated";
+      wrapper.vm.formData.separator = "|";
+      await wrapper.vm.$nextTick();
+
+      wrapper.vm.formData.output_format = "json";
+      await wrapper.vm.$nextTick();
+
+      // Separator should still have value but won't be used in payload
+      expect(wrapper.vm.formData.separator).toBe("|");
+    });
+
+    it("should clear separator when switching from stringseparated to esbulk in edit mode", () => {
+      const destination1 = {
+        name: "Test Destination",
+        url: "https://example.com/api/logs",
+        method: "post",
+        output_format: {
+          stringseparated: {
+            separator: "|",
+          },
+        },
+        destination_type_name: "custom",
+        skip_tls_verify: false,
+      };
+
+      wrapper.vm.populateFormForEdit(destination1);
+      expect(wrapper.vm.formData.separator).toBe("|");
+
+      // Now edit with esbulk format
+      const destination2 = {
+        name: "Test Destination",
+        url: "https://example.com/api/logs",
+        method: "post",
+        output_format: {
+          esbulk: {
+            index: "logs",
+          },
+        },
+        destination_type_name: "elasticsearch",
+        skip_tls_verify: false,
+      };
+
+      wrapper.vm.populateFormForEdit(destination2);
+      expect(wrapper.vm.formData.separator).toBe("");
+      expect(wrapper.vm.formData.esbulk_index).toBe("logs");
+    });
+
+    it("should reset separator field when form is reset", () => {
+      wrapper.vm.formData.separator = "|";
+      wrapper.vm.formData.output_format = "stringseparated";
+
+      wrapper.vm.resetForm();
+
+      expect(wrapper.vm.formData.separator).toBe("");
+      expect(wrapper.vm.formData.output_format).toBe("json");
+    });
+
+    // Note: The scenario where separator has a value but format is "json" is already covered by:
+    // 1. Line 1153-1163: Tests that separator is ignored when format changes to json
+    // 2. Line 560-600: Tests that json format works correctly in payload
+    // The payload construction logic is straightforward: if format === "stringseparated" use nested object, else use string
+
+    it("should validate separator field is not null", async () => {
+      wrapper.vm.formData.destination_type = "custom";
+      wrapper.vm.formData.name = "test";
+      wrapper.vm.formData.url = "https://example.com";
+      wrapper.vm.formData.method = "post";
+      wrapper.vm.formData.output_format = "stringseparated";
+      wrapper.vm.formData.separator = null as any;
+      await wrapper.vm.$nextTick();
+
+      expect(wrapper.vm.canProceedStep2).toBe(false);
+    });
+
+    it("should validate separator field is not undefined", async () => {
+      wrapper.vm.formData.destination_type = "custom";
+      wrapper.vm.formData.name = "test";
+      wrapper.vm.formData.url = "https://example.com";
+      wrapper.vm.formData.method = "post";
+      wrapper.vm.formData.output_format = "stringseparated";
+      wrapper.vm.formData.separator = undefined as any;
+      await wrapper.vm.$nextTick();
+
+      expect(wrapper.vm.canProceedStep2).toBe(false);
+    });
+
+    it("should include String Separated in outputFormats options", () => {
+      const outputFormats = [
+        { label: "JSON", value: "json" },
+        { label: "NDJSON", value: "ndjson" },
+        { label: "NestedEvent", value: "nestedevent" },
+        { label: "ESBulk", value: "esbulk" },
+        { label: "String Separated", value: "stringseparated" },
+      ];
+
+      // Check that stringseparated option exists
+      const stringSeparatedOption = outputFormats.find(
+        (opt) => opt.value === "stringseparated"
+      );
+      expect(stringSeparatedOption).toBeDefined();
+      expect(stringSeparatedOption?.label).toBe("String Separated");
+    });
+  });
+
   describe("Destination Metadata", () => {
     it("should have metadata field available", () => {
       // Metadata may be undefined initially until destination type requires it

web/src/components/pipeline/NodeForm/CreateDestinationForm.vue

@@ -279,6 +279,30 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
                 Index name where data will be written in Elasticsearch
               </template>
             </q-input>
+
+            <!-- StringSeparated Separator field - only shown when output format is stringseparated -->
+            <q-input
+              v-if="formData.output_format === 'stringseparated'"
+              data-test="add-destination-separator-input"
+              v-model="formData.separator"
+              :label="t('alert_destinations.separator') + ' *'"
+              :placeholder="t('alert_destinations.separator_placeholder')"
+              class="no-border showLabelOnTop q-mt-sm"
+              borderless
+              dense
+              flat
+              stack-label
+              :rules="[
+                (val: any) =>
+                  (val !== null && val !== undefined && val !== '') ||
+                  'Separator is required for StringSeparated format',
+              ]"
+              tabindex="0"
+            >
+              <template v-slot:hint>
+                {{ t('alert_destinations.separator_hint') }}
+              </template>
+            </q-input>
           </div>
 
           <!-- Destination-specific Metadata Section -->
@@ -648,6 +672,7 @@ const outputFormats = [
   { label: "NDJSON", value: "ndjson" },
   { label: "NestedEvent", value: "nestedevent" },
   { label: "ESBulk", value: "esbulk" },
+  { label: "String Separated", value: "stringseparated" },
 ];
 const destinationTypes = [
   {
@@ -710,6 +735,7 @@ const formData: Ref<DestinationData> = ref({
   output_format: "json",
   destination_type: "openobserve",
   esbulk_index: "",
+  separator: "",
 });
 
 // OpenObserve specific fields
@@ -861,13 +887,24 @@ const populateFormForEdit = (destination: any) => {
       formData.value.output_format = "esbulk";
       formData.value.esbulk_index =
         destination.output_format.esbulk.index || "default";
+      formData.value.separator = "";
+    } else if (
+      typeof destination.output_format === "object" &&
+      destination.output_format.stringseparated
+    ) {
+      formData.value.output_format = "stringseparated";
+      formData.value.separator =
+        destination.output_format.stringseparated.separator || "";
+      formData.value.esbulk_index = "";
     } else if (typeof destination.output_format === "string") {
       formData.value.output_format = destination.output_format;
       formData.value.esbulk_index = "";
+      formData.value.separator = "";
     }
   } else {
     formData.value.output_format = "json";
     formData.value.esbulk_index = "";
+    formData.value.separator = "";
   }
 
   // Use destination_type_name from backend, fallback to destination_type or default
@@ -1074,6 +1111,15 @@ const canProceedStep2 = computed(() => {
     );
   }
 
+  // Additional validation for StringSeparated format
+  if (formData.value.output_format === "stringseparated") {
+    return !!(
+      formData.value.separator !== null &&
+      formData.value.separator !== undefined &&
+      formData.value.separator !== ""
+    );
+  }
+
   return true;
 });
 
@@ -1218,13 +1264,24 @@ const createDestination = () => {
   const fullUrl = formData.value.url + (formData.value.url_endpoint || "");
 
   // Handle output format - for esbulk, format as JSON object with index
+  // For stringseparated, format as JSON object with separator
+  // For all other formats (json, nestedevent, etc.), keep as string
   let outputFormat: any = formData.value.output_format;
-  if (outputFormat === "esbulk" && formData.value.esbulk_index) {
+  if (formData.value.output_format === "esbulk") {
     outputFormat = {
       esbulk: {
         index: formData.value.esbulk_index,
       },
     };
+  } else if (outputFormat === "stringseparated" && formData.value.separator) {
+    outputFormat = {
+      stringseparated: {
+        separator: formData.value.separator,
+      },
+    };
+  } else {
+    // Keep output_format as string for json, nestedevent, and other formats
+    outputFormat = formData.value.output_format;
   }
 
   const payload: any = {
@@ -1334,6 +1391,7 @@ const resetForm = () => {
     output_format: "json",
     destination_type: defaultDestinationType,
     esbulk_index: "",
+    separator: "",
   };
   // Reset OpenObserve specific fields
   openobserveOrg.value = "default";

web/src/locales/languages/de.json

@@ -2036,7 +2036,10 @@
     "customWebhook": "Benutzerdefiniert (Webhook)",
     "customEmail": "Benutzerdefiniert (E-Mail)",
     "customAction": "Benutzerdefiniert (Aktion)",
-    "custom": "Benutzerdefiniert"
+    "custom": "Benutzerdefiniert",
+    "separator": "Trennzeichen",
+    "separator_placeholder": "Geben Sie ein Trennzeichen ein (z. B. |,\\n,\\ t, Leerzeichen)",
+    "separator_hint": "Trennzeichenfolge, die zwischen Datensätzen verwendet werden soll (Leerzeichen erlaubt)"
   },
   "metrics": {
     "searchMetric": "Suchen Sie nach einer Metrik",

web/src/locales/languages/en.json

@@ -2298,7 +2298,10 @@
     "customWebhook": "Custom (Webhook)",
     "customEmail": "Custom (Email)",
     "customAction": "Custom (Action)",
-    "custom": "Custom"
+    "custom": "Custom",
+    "separator": "Separator",
+    "separator_placeholder": "Enter separator (e.g., |, \\n, \\t, space)",
+    "separator_hint": "Separator string to use between records (spaces allowed)"
   },
   "pipeline_destinations": {
     "header": "Pipeline Destinations",

web/src/locales/languages/es.json

@@ -2036,7 +2036,10 @@
     "customWebhook": "Personalizado (Webhook)",
     "customEmail": "Personalizado (correo electrónico)",
     "customAction": "Personalizado (acción)",
-    "custom": "Personalizado"
+    "custom": "Personalizado",
+    "separator": "Separador",
+    "separator_placeholder": "Introduzca un separador (p. ej., |,\\n,\\ t, espacio)",
+    "separator_hint": "Cadena separadora para usar entre registros (se permiten espacios)"
   },
   "metrics": {
     "searchMetric": "Buscar una métrica",

web/src/locales/languages/fr.json

@@ -2036,7 +2036,10 @@
     "customWebhook": "Personnalisé (Webhook)",
     "customEmail": "Personnalisé (e-mail)",
     "customAction": "Personnalisé (Action)",
-    "custom": "Personnalisé"
+    "custom": "Personnalisé",
+    "separator": "Séparateur",
+    "separator_placeholder": "Entrez un séparateur (par exemple, |,\\n, \\ t, espace)",
+    "separator_hint": "Chaîne de séparation à utiliser entre les enregistrements (espaces autorisés)"
   },
   "metrics": {
     "searchMetric": "Rechercher une métrique",

web/src/locales/languages/it.json

@@ -2036,7 +2036,10 @@
     "customWebhook": "Personalizzato (Webhook)",
     "customEmail": "Personalizzato (email)",
     "customAction": "Personalizzato (azione)",
-    "custom": "Personalizzato"
+    "custom": "Personalizzato",
+    "separator": "Separatore",
+    "separator_placeholder": "Inserisci un separatore (ad esempio, |,\\n,\\ t, spazio)",
+    "separator_hint": "Stringa separatrice da utilizzare tra i record (spazi consentiti)"
   },
   "metrics": {
     "searchMetric": "Cerca una metrica",

web/src/locales/languages/ja.json

@@ -2053,7 +2053,10 @@
     "customWebhook": "カスタム (ウェブフック)",
     "customEmail": "カスタム (電子メール)",
     "customAction": "カスタム (アクション)",
-    "custom": "カスタム"
+    "custom": "カスタム",
+    "separator": "セパレーター",
+    "separator_placeholder": "セパレータを入力してください (例:|、\\n、\\ t、スペース)",
+    "separator_hint": "レコード間に使用する区切り文字列 (スペース可)"
   },
   "pipeline_destinations": {
     "header": "パイプライン送信先",

web/src/locales/languages/ko.json

@@ -2036,7 +2036,10 @@
     "customWebhook": "사용자 지정 (웹후크)",
     "customEmail": "사용자 지정 (이메일)",
     "customAction": "사용자 지정 (액션)",
-    "custom": "사용자 지정"
+    "custom": "사용자 지정",
+    "separator": "분리기",
+    "separator_placeholder": "구분자 입력 (예: |,\\n,\\ t, 공백)",
+    "separator_hint": "레코드 사이에 사용할 구분자 문자열 (공백 허용)"
   },
   "metrics": {
     "searchMetric": "지표 검색",

web/src/locales/languages/nl.json

@@ -2036,7 +2036,10 @@
     "customWebhook": "Op maat (Webhook)",
     "customEmail": "Aangepast (e-mail)",
     "customAction": "Aangepast (actie)",
-    "custom": "Op maat"
+    "custom": "Op maat",
+    "separator": "Separator",
+    "separator_placeholder": "Voer een scheidingsteken in (bijvoorbeeld |,\\n,\\ t, spatie)",
+    "separator_hint": "Scheidingstekenreeks om tussen records te gebruiken (spaties toegestaan)"
   },
   "metrics": {
     "searchMetric": "Zoek naar een statistiek",

web/src/locales/languages/pt.json

@@ -2036,7 +2036,10 @@
     "customWebhook": "Personalizado (Webhook)",
     "customEmail": "Personalizado (e-mail)",
     "customAction": "Personalizado (Ação)",
-    "custom": "Personalizado"
+    "custom": "Personalizado",
+    "separator": "Separador",
+    "separator_placeholder": "Insira o separador (por exemplo, |,\\n,\\ t, espaço)",
+    "separator_hint": "Cadeia de caracteres separadora a ser usada entre registros (espaços permitidos)"
   },
   "metrics": {
     "searchMetric": "Pesquisar uma métrica",

web/src/locales/languages/tr.json

@@ -2036,7 +2036,10 @@
     "customWebhook": "Özel (Webhook)",
     "customEmail": "Özel (E-posta)",
     "customAction": "Özel (Eylem)",
-    "custom": "Özel"
+    "custom": "Özel",
+    "separator": "Ayırıcı",
+    "separator_placeholder": "Ayırıcı girin (örn. |,\\ n,\\ t, boşluk)",
+    "separator_hint": "Kayıtlar arasında kullanılacak ayırıcı dize (boşluklara izin verilir)"
   },
   "metrics": {
     "searchMetric": "Bir Metrik Ara",

web/src/locales/languages/zh.json

@@ -2036,7 +2036,10 @@
     "customWebhook": "自定义（Webhook）",
     "customEmail": "自定义（电子邮件）",
     "customAction": "自定义（操作）",
-    "custom": "自定义"
+    "custom": "自定义",
+    "separator": "分隔符",
+    "separator_placeholder": "输入分隔符（例如，|、\\n、\\ t、空格）",
+    "separator_hint": "在记录之间使用的分隔字符串（允许空格）"
   },
   "metrics": {
     "searchMetric": "搜索指标",

web/src/ts/interfaces/alert.ts

@@ -117,10 +117,11 @@ export interface Destination {
   emails?: string;
   type: "http" | "email" | "sns" | "action";
   action_id?: string;
-  output_format?: "json" | "ndjson" | "nestedevent" | string; // string allows esbulk with dynamic index
+  output_format?: "json" | "ndjson" | "nestedevent" | string | any; // string allows esbulk with dynamic index, any for stringseparated object
   destination_type?: string; // Frontend internal use
   destination_type_name?: string; // From backend
   esbulk_index?: string; // For esbulk format index name
+  separator?: string; // For stringseparated format separator value
   metadata?: DestinationMetadata; // Destination-specific metadata as JSON object
 }
 
@@ -134,8 +135,9 @@ export interface DestinationPayload {
   emails?: string[];
   type: "http" | "email" | "sns" | "action";
   action_id?: string;
-  output_format?: "json" | "ndjson" | "nestedevent" | string; // string allows esbulk with dynamic index
+  output_format?: "json" | "ndjson" | "nestedevent" | string | any; // string allows esbulk with dynamic index, any for stringseparated object
   destination_type?: string; // New field added
+  separator?: string; // For stringseparated format separator value
 }
 
 // Destination object which is modified in frontend to display in table and form