Subscription processor pipeline

[bewing]

Right now, formatters/processors are assigned to a specific output. I would consider it useful to be able to apply a set of processors only to events for a given subscription. One way to accomplish this would be to chain "filter" formatters, where you can have a formatter that evaluates a condition, and has its own list of processors for further event processing if it evaluates to true.

[karimra]

all processors have a condition that tells if they need to be executed on the event message or not, it takes a jq expression

[bewing]

I did a quick spot check, because I didn't remember seeing this in the documentation that I reviewed -- while condition is present on some processors, it doesn't appear to be universal.

$ git grep CheckCondition
formatters/event_add_tag/event_add_tag.go:                      ok, err := formatters.CheckCondition(p.code, e)
formatters/event_allow/event_allow.go:                  ok, err := formatters.CheckCondition(d.code, e)
formatters/event_drop/event_drop.go:                    ok, err := formatters.CheckCondition(d.code, e)
formatters/event_trigger/event_trigger.go:              res, err := formatters.CheckCondition(p.code, e)
formatters/event_write/event_write.go:                  ok, err := formatters.CheckCondition(p.code, e)
formatters/processors.go:func CheckCondition(code *gojq.Code, e *EventMsg) (bool, error) {
formatters/processors_test.go:func TestCheckCondition(t *testing.T) {
formatters/processors_test.go:                          ok, err := CheckCondition(code, in)

[karimra]

Agree, it's not present for all processors. I didn't see a need for it be present for all of them ( which is wrong )

A solution for this issue would be to implement the condition check for all processors, applying a processor to messages from a subscription sub1 would just need condition: '.name == "sub1"'

[Sparc0]

Is this implemented or is it just a idea how to solve the issue?

Because i have a processor i only want to run on the subscription named interface_state To only allow events for interfaces i am interested in. But if i uncomment the condition it will allow all interfaces.

intf-keep-events:
  event-allow:
    #condition: '.name == "interface_state"'
    tags:
      - "ae\\w+"
      - "ge-.{5}"
      - "xe-.{5}"
      - "et-.{5}"
      - "fxp\\w?"
      - "em\\w?"
      - "irb"

This processor will drop all my other paths i subscribe on for cpu, memory etc. Also when i start gNMIc i still see some events that i fetch using mode: once. Are processors applied on this mode even?

I could use event-drop instead to drop all interfaces i dont want but using the event-allow would make the config more understandable.

And no i cant do a more specific subscribe call for interface since JunOS wont let me fetch all ae interfaces when i do

/interfaces/interface[name=ae]/state

But it let me fetch all et interfaces when i do

/interfaces/interface[name=et]/state

I have a JTAC ticket about this already.

[karimra]

@Sparc0 condition is implemented for the processor event-allow: https://gnmic.kmrd.dev/user_guide/event_processors/event_allow/

Processors apply to updates from a once subscription, providing the output is influxdb, prometheus or the format is event

[Sparc0]

@karimra so docs says condition or one of the regular expressions under tags, tag-names, values or value-names. so the way i used condition in my example above is not meant to work as i thought then?

[karimra]

Yes, an event message is allowed if the condition is true OR any of the regular expressions matches. event-allow is kind of an exception... condition is for more complex cases, while the other fields are for simple regex matching.

You could combine all those regular expressions with the subscription name under condition:

(.name == "interface_state") 
    and
    (
        .tags.interface_name | 
        (
            test("ae\\w+")
            or
            test("ge-.{5}")
            or
            test("xe-.{5}")
            or
            test("et-.{5}")
            or
            test("fxp\\w?")
            or
            test("em\\w?")
            or
            test("irb")
        )
    )

Note that the subscription name is also present as a tag .tags.subscription-name, so you can also use this condition:

(.tags."subscription-name" == "interface_state") 
    and
    (
        .tags.interface_name | 
        (
            test("ae\\w+")
            or
            test("ge-.{5}")
            or
            test("xe-.{5}")
            or
            test("et-.{5}")
            or
            test("fxp\\w?")
            or
            test("em\\w?")
            or
            test("irb")
        )
    )