Pipelines are the core processing units that consume events from Kafka, apply detection rules, and output tagged events.

The Pipelines page shows the list of configured Pipelines. Each Pipeline has the following properties:

Run/Stop – Enable or disable the Pipeline. By default, when you configure a Pipeline, it's enabled. Running/stopping requires some time.
- Run – Creates FlinkDeployment in Kubernetes and starts processing
- Stop – Stops processing and removes FlinkDeployment
Name – The name of the Pipeline
Source Topic – The names of the Source Topics configured in the Pipeline
Destination Topic – The name of the Destination Topic configured in the Pipeline
Log Source – The name of Log Source configured in the Pipeline
Filters – The number of Filters configured in the Pipeline
Rules – The total number of rules in all Repositories configured in the Pipeline
Events – The number of events that has been tagged since the Pipeline has been created in the
Created – The date when the Pipeline was created
Status – The current status of the Pipeline
Actions
- View Pipeline
- Configure Pipeline

Note:

When a Pipeline is created and run, it starts reading Kafka Source Topic events from the moment of its creation
When Pipeline is stopped and then run again, event data is read from the moment of stopping

Create Pipeline

Before creating a Pipeline, ensure:

Source Topics exists in Kafka
Destination Topic exists in Kafka
Log Source is created and tested with the same Repositories and Source Topics that will be used in the Pipeline
Repositories contain rules and are synced (for Сloud Repositories)
Filters (if used) are created
There are enough Match Nodes in DetectFlow deployment and enough resources provided in Kubernetes. Each pipeline requires one Match Node. The guidelines on required resources for each Match Node depending on the EPS and the number of detection rules can be found in Appendix A of the DetectFlow Requirements and Deployment Guide

To create a Pipeline, follow these steps:

Click the New Pipeline button to open the New Pipeline page.
Fill in the Pipeline Name. You can use any naming convention. E. g. the name can be based on the log source that produces events to be processed.
Select a Log Source profile. Make sure you've tested the selected Log Source with the Repositories and Source Topics selected for this Pipeline.
Once you select a Log Source profile, the Source Topic and Repositories fields are automatically filled with the source topic and repositories selected for test during Log Source creation.
The Source Topic field is pre-filled based on the selected Log Source profile but you can add additional source topics. Source topics are Kafka topics with events you want to process.
Note:
- To be available for selection, a topic has to already exist in Kafka
- You can select multiple topics. In this case, if events from different topics have different format, pay attention to covering those different formats when creating the parsing script in the Log Source
- A Source Topic in one Pipeline can be a Destination Topic in another and vice versa. This ensures you can chain topics if needed
The Repositories field is pre-filled based on the selected Log Source profile but you can add additional repositories or folders.
Note:
- You can select multiple repositories or folders
- You can select a separate folder based on the log source from which the data is being sent to the source topic
- You can combine Cloud and Local Repositories
- It's recommended to create Repositories per log source
- Always ensure you test all the selected Repositories with the Log Source configured for a specific Source Topic
Select a Destination Topic.
Select Output Format:
- Preserve Source Format. Events will be written to the Destination Topic in the same format as they have in the Source Topic
- Apply Log Source Parsing. Events will be written to the Destination Topic in the parsed format according to the parsing rules you've set in the configurations of the Log Source specified for this Pipeline. E. g., if you have defined a parsing script parse_json(field="raw") and have a source format event:
```
{
    "raw": "{\"user\": \"john\", \"ip\": \"192.168.1.1\"}",
    "log": "192.168.1.1 - GET /api/users [2024-01-01 10:00:00]"
}
```
  it will be saved as:
```
{"user": "john", "ip": "192.168.1.1"}
```
Set the Save Untagged Events switch
- If enabled, all events (both tagged and untagged) from the Source Topic will be saved to the Destination Topic
- If disabled, only tagged events will be saved to the Destination Topic. Use this option to save on storage costs upstream by pushing only events that were matched with a detection rule
Optionally, specify additional settings:
- Select Filters. They allow you to filter out events before those events are matched to detection rules. This can be useful for filtering out known legit activity. Multiple Filters can be combined
- Set Custom Fields. These are fields that will be added at the end of each event saved to the Destination Topic. Specify them as key-value pairs in YAML format, e. g.
```
environment: production
region: us-east-1
source: kafka
```
Optionally, overwrite the general Pipeline Runtime settings so that this particular Pipeline has its own. While only Admin can modify the general Pipeline Runtime settings, any user can overwrite them for a particular Pipeline.
Click the Create Pipeline button.

Once the Pipeline is created:

It's run automatically. This process may take some time
You can check its status in the list of Pipelines on the Pipelines page
It's added to the Dashboard for monitoring

Possible Pipeline statuses:

Status	Meaning
Running	Operates normally, events are processed
Disabled	Disabled in DetectFlow by a user
Not Found	FlinkDeployment hasn't been found in Kubernetes
Unknown	Status retrieval error
Created	The Flink job has been created, but hasn't been started yet
Reconciling	Operator makes changes (e.g. config update or restarting)
Suspended	The Flink job has been suspended (with a savepoint created)
Failed	The Flink job has failed with an error
Cancelling	The Flink job is being stopped
Finished	The Flink job has finished
Restarting	The Flink job is restarting
Deploying	Deployment is being created/updated
Upgrading	Deployment is being updated to a new version
Rolled Back	Deployment has rolled back after an unsuccessful update
Stable	Deployment is stable, but the Flink job isn't running

A status may have one of three levels:

Green - informational (everything works fine)
Yellow - warning (pay attention, hover over the status chip to view a tooltip with the warning details)
Red - error (not operational, hover over the status chip to view a tooltip with the error details)

Pipeline Lifecycle

CREATE PIPELINE
- Save the pipeline config to PostgreSQL
- Create FlinkDeployment in Kubernetes
- Flink Operator spawns JobManager + TaskManagers
FLINK JOB STARTS
- Consume events from Source Topic (Kafka)
- Consume detection rules/Filters/parsers from the service sigma rules topic
- Apply parsing script to transform events
- Apply field mapping
- Apply Filters (if configured)
- Apply Sigma detection rules
- Write tagged events to Destination Topic (Kafka)
REAL-TIME UPDATES (Hot-Reload)
- Update rule/Filter/parser via API
- Admin Panel publishes to the service sigma rules topic
- Flink picks up changes WITHOUT restart

Hot-Reload vs Restart:

Change Type	Action	Downtime
Pipeline name	Change name in Postgres	None (hot-reload)
Rules (add/update/delete)	Kafka publishKafka publish	None (hot-reload)
Filters	Kafka publish	None (hot-reload)
Log Source (parser/mapping)	Kafka publish	None (hot-reload)
Custom Fields	Kafka publish	None (hot-reload)
Source/Destination Topic	Recreate FlinkDeployment (without savepoint)	60-90s
Enable/Disable	Create/Delete FlinkDeployment (with savepoint)	60-90s

View Pipeline Details

On the Pipelines page, click the eye icon next to the Pipeline you want to view.

On the Pipeline details page that appears, you can view the following:

Pipeline configuration:
- Source Topic
- Destination Topic
- Log Source
- Output Format
- Save Untagged Events
- Filters
Key metrics:
- Tagged Events. The number of events matched by the detection rules in this Pipeline since it was created
- Untagged Events. Events from the Source Topics that were processed but didn't match any of the applied detection rules (since the Pipeline was created)
- Active Rules. The total number of detection rules that are applied to the configured Source Topic events
- Matched Rules. The number of detection rules that have had a match with at least one event (since the Pipeline was created)
The list of detection rules against which the Source Topic events are matched. Each rule has the following info (to sort by a column, hover over its header and click the up or down arrow that appears):
- Rule Name. The title of the rule
- Repository. The Repository where the rule is located
- Created. The date when the rule was created
- Updated. The date when the rule was updated (if the rule belongs to a SOC Prime Cloud Repository, the update could have been made either in DetectFlow or on the SOC Prime Platform)
- Tagged Events. The number of events matched by this detection rule in this Pipeline since it was created
- Status
  - Active. The rule is applied to the Source Topic events
  - Inactive. The rule is not applied to the Source Topic events

By default, all rules are active, i.e. applied to the events in the associated Source Topic. You can activate and deactivate rules individually or in bulk. To do this, set checkmarks next to the rules you want to change the status for, and click Activate or Deactivate.

For example, you may want to deactivate a rule that needs fine-tuning.

Additionally, you can view only those rules that have matches with events using the Hide Unmatched Rules switch.

To view a rule, click on its name. A page with the rule's code appears.

From this page you also can:

For rules in Local Repositories:
- Delete the rule by clicking the Delete button
- Edit the rule by clicking the Edit button
For rules in Cloud Repositories:
- Edit the rule by clicking the Edit button
- Open the rule in Uncoder AI on the SOC Prime Platform by clicking the Open in Uncoder AI button (applicable only to rules synchronized with the SOC Prime Platform, a SOC Prime Platform account is required).

Edit or Delete Pipeline

To edit a Pipeline:

For the selected Pipeline, click on the gear icon in the Actions column on the Pipelines page.
On the Pipeline configuration page that appears, make the required adjustments and click the Update Pipeline button.

Note: The Pipeline must be enabled for changes to take effect.

To delete a Pipeline:

For the selected Pipeline, click on the gear icon in the Actions column on the Pipelines page.
On the Pipeline configuration page that appears, scroll to the bottom and click the Delete button.
Confirm the action.