The Ingest Pipeline should support the ability for conditional branching of processor execution.

The proposal here is to define a CompareProcessor which will accept a list of conditions that must all be met for specific processors to be executed. This processor would have a conditions parameter that would accept various boolean operators, such as gt, lt, gte, lte, eq, neq. If all of the provided conditions evaluate to true, then a set of processors defined in a field potentially called if_processors would be executed. If any conditions in this set evaluate to false, then processors in the else_processors list option would be executed. This type of conditional can be though of as an "if/else" processor, where the conditions field is a compounded boolean expression joined with an AND. Continuing with this analogy -- the if_processors block would be the if block, and the else_processors would be the else block.

Whether both if_processors and else_processors configuration options would be necessary is still up in the air, at least one must be defined.

Since the CompareProcessor is just another processor, nested comparisons would be possible.

Here is a potential processor definition for the CompareProcessor

{
  "compare": {
    "conditions": [
      "<field_name1>" : { "<OP1>": x },
      "<field_name2>" : { "<OP2>": y },
      "<field_name3>" : { "<OP3>": z }
    ],
    "if_processors": [...],
    "else_processors": [...]
  }
}

Here is an example pipeline definition with a Compare Processor

{
  "description": "_description",
  "processors": [
    {
      "compare": {
        "conditions": [
          "ip_address": { "eq": "127.0.0.1" },
          "user_age": { "gt": 13 }
        ],
        "if_processors": [
          "compare" : {
            "conditions" : [
               "country_name" : { "neq": "USA" }
               "if_processors" : [
                 {
                   "geoip" : {
                     "source_field" : "ip"
                   }
                 }
               ]
            ]
          }
        ],
        "else_processors" : [
          {
            "mutate" : {
              "remove" : ["country_name"]
            }
          }
        ]
      },
      "mutate" : {
        "update" : {
          "field2" : "_value"
        }
      }
    }
  ]
}