Create and Use Real-time Pipeline

Step 1: Open the pipeline editor

1. Log in to OpenObserve.
2. In the left navigation panel, go to Data > Pipelines. By default, the system opens the Pipelines tab.
3. Click New pipeline at the top-right corner.

Step 2: Enter a pipeline name

Provide a unique name for the pipeline.

Step 3: Configure the Source node

1. From the Source section, drag the Stream node into the editor. On an empty canvas, you can also click Choose a Source and select Stream.
2. Edit the source node:

• Select the Stream Type.

• Choose one of the following options:

  • Option A: Select an existing stream. Choose a stream from the Stream Name dropdown.
  • Option B: Create a new stream. Enable the Create new Stream toggle and create a new stream if necessary.

3. Click Save to confirm the source node.

Step 4: Configure the Transform node

1. From the Transform section, drag either a Conditions or Function node into the editor. You can also hover over a node and click + to pick the next step.

2. If you add a Conditions node:

   1. In the Associate Condition screen, select the field name in the Column drop-down list. This list displays the fields detected in the source stream. To use a field that is not listed, type its name and press Enter.
   2. Select an operator.
   3. Enter a value in the Value input box.
   4. Add more conditions with Condition, or nested AND/OR groups with Condition Group, if needed.
   Important

   Condition nodes require a stream schema.
   If you create a new stream in Step 3 and do not define any fields, the Condition editor does not display field names. To use conditions, add at least one field while creating the source stream or ingest sample data so that the schema is inferred.

   Guidelines

   • Use an empty string to check for empty values. For example, app_name != ""
   • Use null to check for null values. For example, app_name != null
   • If the condition does not match, the record is dropped.
   • If the record does not contain the specified field, it is also dropped.

3. If you add a Function node:

   Use a Function node to transform events using a VRL function.

   A Function does not require predefined fields. You can use it even if the source stream has no schema.
   

   To create a new function:

   

   1. In the Associate Function panel, enable the Create New Function toggle. The function editor opens in the panel.
   2. Enter the function name.
   3. Write a VRL function. Pipelines run VRL functions only.
   4. To test it, in the Query section select the stream type, stream name, and duration, and click Run query. The results appear under Events.
   5. Click Test Function and check the result under Output.
   6. Click Save to save the function.

4. The After Flattening toggle is enabled by default. It ensures the function processes normalized data. Disable this toggle only if you need the original structure.

   Guidelines

   • RBF (Run Before Flattening): Function executes before data structure is flattened.
   • RAF (Run After Flattening): Function executes after data structure is flattened.

5. Click Save to confirm the transform node.

Step 5: Configure the Destination node

A destination defines where the processed events are written. You can forward data to another stream or to an external system.

To add a stream destination:

1. Drag a Stream node into the editor.

2. In the Associate Stream panel, configure the destination stream.

3. Select the Stream Type, then select an existing stream in Stream Name, or type a new name to create a new stream.

   Note

   • For real-time pipelines, you can select logs, metrics, or traces as the destination stream type.
   • Enrichment_tables as a destination stream is only available for scheduled pipelines. It cannot be used as a destination in real-time pipelines.
   Dynamic stream names

   You can configure the stream name dynamically with curly braces.
   Example: static_text_{fieldname}_postfix
   Static text before and after the braces is optional.

   Default destination and data loss

   When you select a source stream for a real-time pipeline, OpenObserve automatically adds a destination that points back to the same stream. This default destination ensures that all ingested events continue to be stored in the source stream.
   If you remove this default destination and only keep filtered routes, events that do not match any condition are dropped and are not written to the source stream.
   Add at least one catch-all route back to the original stream if you want to preserve all events.

To add an external destination:

1. Drag a Remote node into the editor. The Remote node is available in the self-hosted Enterprise Edition.
2. In the External Destination panel, either select an existing destination or enable Create New Destination and fill in the required details such as destination type, name, URL, method, output format, and headers.
3. Click Save.

For detailed steps on configuring external destinations, see the Remote Destination guide.

Step 6: Connect the nodes

Use the connection icon to link Source > Transform > Destination. To remove a link, select the link and press delete on your keyboard.

Step 7: Save the pipeline

Click Save. The pipeline becomes active immediately.

Step 1: Ingest data

Use curl or any supported ingestion method. To find the sample curl command, go to Data Sources > Custom > Logs > Curl. Each new event ingested into the source stream is processed instantly.

Step 2: Verify output

1. Go to Logs.
2. From the stream selection dropdown, select the stream you added as destination.
3. Select the time range from the time range selector.
4. Click Run query to view the result.