[RFC] Conditional Routing

[dlvenable]

Background

Data Prepper pipelines currently do not support conditionals or routing. Thus all events in Data Prepper must flow through all sinks and processors in a pipeline. Many users require the ability to route events to different sinks and processors depending on the specific event.

The following diagram outlines a common scenario: Users need to route data to different sinks depending on some property of the event.

Proposal

This RFC introduces a concept of a router to Data Prepper. Pipeline authors can define named routes in the router. Data Prepper will apply routes to individual Events before sending them to Sinks.

This GitHub issue focuses on using routing to route sinks. See #522 RFC for a proposal for routing through a processor chain.

The following diagram outlines where the router will sit and what it will perform.

Design

Data Prepper will introduce a new router component to the pipeline. This is at the same level of the YAML as the prepper and sink. The router will run after the Processor chain and before the Sinks. Data Prepper would evaluate these routes directly before passing the Events into the sinks.

log-pipeline:
  source:
    http:
  processor:
  router:
    - application-logs: '/log_type == "application"'
    - http-logs: '/log_type == "apache"'
  sink:
    - opensearch:
        hosts: [ "https://opensearch:9200" ]
        index: application_logs
        routes: [application-logs]
    - opensearch:
        hosts: [ "https://opensearch:9200" ]
        index: http_logs
        routes: [http-logs]
    - opensearch:
        hosts: [ "https://opensearch:9200" ]
        index: all_logs

Any sink with the routes property will only accept Events which match at least one of the routes. In the example above, application-logs is a named route. Data Prepper will only route events with the application-logs route to the first opensearch sink.

By default, Data Prepper will route all Events to a sink which does not define a route. Thus, in the example above, all Events will go into the third opensearch sink.

Alternatives

See the comments below for alternatives.

[graytaylor0]

I really like this! It is clever to set the routes in the EventMetadata. I do think users will want the remove_routes functionality, but there is no reason to add it immediately as it can be added easily in the future. From a user perspective, only typing a route for each sink/processor rather than a full conditional (like we had with the when conditional) is much cleaner, and has the added benefit of only evaluating conditionals once in the router processor.

The thing I would like some more clarity on is what really changes from the router's perspective for sinks vs processors. Event though sinks don't have ordering, it seems like the implementation would be the same.

[dapowers87]

In the first alternative, should the second router listing say apache instead of http-logs?

[dlvenable]

@dapowers87 , That was a typo, and I updated it. Thanks for catching and noting it!

[dapowers87]

I really like the idea of the first alternative. Having a router be a base component sounds like it could open itself up to a lot of neat ideas down the road.

[dapowers87]

What if we made router a base component and have pipeline flow in and out of them. Something like this?

pipeline-a:
  source:
    http:
  to-router: router-a

pipeline-b:
  processor:
    grok: ...
    add_entry: ...
  to-router: router-b

pipeline-c:
  processor:
    grok: ...
    add_entry: ...
  sink:
    opensearch: ...

pipeline-d:
  processor:
    grok: ...
    add_entry: ...
  to-router: router-c

pipeline-sink-a:
  sink:
    opensearch: ...
    stdout:

pipeline-sink-b:
  sink:
    opensearch: ...

router:
  - name: router-a
    set_routes:
      - pipeline-b: "some_conditional"
      - pipeline-c: "secondary_conditional"
  - name: router-b
    set_routes:
      - pipeline-c: "third_conditional"
      - pipeline-d: "fourth_conditional"
  - name: router-c
    set_routes:
      - pipeline-sink-a: "fifth_conditional"
      - pipeline-sink-b: "sixth_conditional"

[cmanning09]

This behavior will support other use-cases in the future. One example is to have a common, shared pipeline. Currently, Data Prepper does not support multiple sources in a pipeline. However, if that support is added, users could define a common middle pipeline which is shared across different pipelines.

Can this already be achieved like so:

application-log-pipeline:
  source:
    http:
  prepper:
     ...
  sink:
    - pipeline:
        name: "log-pipeline"

appach-log-pipeline:
  source:
    http:
  prepper:
     ...
  sink:
    - pipeline:
         name: "log-pipeline"

log-pipeline:
  source:
    http:
  prepper:
     ...
  sink:
    - opensearch:

An alternative I would like to propose is making a router a type of sink. It reduces the routing and/or conditional checks in processors & core. It does not require building a concept of routes into data prepper core and only requires us to develop a single plugin. It does create more pipelines which may be a downside depending on how you look at it. Here is an example:

log-pipeline:
  source:
    http:
  prepper:
      ... generic log processing
  sink:
    -router:
      - '/type == "application"':
        - opensearch:
            hosts: [ "https://opensearch:9200" ]
            index: application_logs
      - '/type == "apache"':
        - pipeline:
             name: "apache-log-pipeline"

appache-log-pipeline:
  source:
    pipeline: log-pipeline
  prepper:
     ... unique appache log processing
  sink:
    - opensearch:
         hosts: [ "https://opensearch:9200" ]
         index: appache_logs

[cmanning09]

has the added benefit of only evaluating conditionals once in the router processor.

@graytaylor0 , the original conditional would be evaluated once but every downstream processor / sink / core will need to evaluate the route.

[graytaylor0]

has the added benefit of only evaluating conditionals once in the router processor.

@graytaylor0 , the original conditional would be evaluated once but every downstream processor / sink / core will need to evaluate the route.

Yes but evaluating if a String is in a Set is less computationally intensive than evaluating the conditional itself.

[graytaylor0]

Can this already be achieved like so:

  source:
    http:
  prepper:
     ...
  sink:
    - pipeline:
        name: "log-pipeline"

appach-log-pipeline:
  source:
    http:
  prepper:
     ...
  sink:
    - pipeline:
         name: "log-pipeline"

log-pipeline:
  source:
    http:
  prepper:
     ...
  sink:
    - opensearch:

I could be wrong about this, but I am fairly sure that this would not work. I think that in the current state, a pipeline can only receive Events from another pipeline by specifying that pipeline as the source.

application-log-pipeline:
  source:
    http:
  prepper:
     ...
  sink:
    - pipeline:
        name: "log-pipeline"

appach-log-pipeline:
  source:
    pipeline: 
       name: "application-log-pipeline"
  prepper:
     ...
  sink:
    - pipeline:
         name: "log-pipeline"

[dlvenable]

This behavior will support other use-cases in the future. One example is to have a common, shared pipeline. Currently, Data Prepper does not support multiple sources in a pipeline. However, if that support is added, users could define a common middle pipeline which is shared across different pipelines.

Can this already be achieved like so:

@cmanning09 ,

There are a few reasons we cannot do this now.

1. As @graytaylor0 pointed out, pipeline authors must specify the pipeline source in their pipeline. They cannot support "any pipeline" as a source.
2. Pipelines cannot have multiple sources. As I noted in the issue description, this is a separate issue.
3. Some users want to have pipelines where events flow through a shared pipeline and then out to different final destination sinks. So for example, application logs go through the shared pipeline and then out to an application logs OpenSearch. HTTP logs could also go through the shared pipeline and then out to an HTTP-logs OpenSearch cluster.

It is for the third reason that I proposed a router processor. These routers would apply routes to Events early in a pipeline and then allow them to route differently later.

An alternative I would like to propose is making a router a type of sink. It reduces the routing and/or conditional checks in processors & core. It does not require building a concept of routes into data prepper core and only requires us to develop a single plugin. It does create more pipelines which may be a downside depending on how you look at it. Here is an example:

This approach is somewhat similar to some of the alternatives I described (though it moves from Core to a plugin). If Data Prepper had a router sink, how would Data Prepper handle the shard pipeline use-case that I just described?