How to execute a processor AFTER a failed/successful output?

[static-moonlight]

I have a stream processor which reads messages, extracts certain information and puts them into a database. In rare cases this can fail due to duplications.

I want to add metrics to that stream processor, exporting information about successful and failed executions. However, metric is a processor, so how do I run it AFTER the output? How do I export metrics for these scenarios ...

• successful
• duplicate key
• other error

... for this simple setup:

input:
 [...] 

output:
  fallback:
    - sql_insert:
        [...]
    - switch:
        retry_until_success: false
        cases:
          - check: 'meta("fallback_error").contains("pq: duplicate key value violates unique constraint")'
            output:
              drop: {}
          - output:
              stdout:
                codec: lines

[mihaitodor]

Hi @static-moonlight 👋 For the successful case, you can use the broker output as a wrapper with pattern: fan_out_sequential. Add your current output as the first child and then have drop with an attached metric processor as the second child. For the failure case, you can wrap your current output in fallback and have this under the broker’s first child. The second child of fallback can use the fallback_error metadata in the attached processors to determine what metric to emit.

[static-moonlight]

you mean like this?

output:
  fallback:
    - broker:
        pattern: fan_out_sequential
        outputs:
          - sql_insert:
              [...]
          - drop: {}
            processors:
              metric: # success
                [...]
    - ????

You kind of lost me in the middle ...

[mihaitodor]

Yep!

output:
  broker:
    pattern: fan_out_sequential
    outputs:
      - fallback:
        - sql_insert:
          [...]
        - drop: {}
          processors:
            - metric: # failure
                      # you can use `switch` here based on `@fallback_error` to have different metrics for different errors
                
      - drop: {}
        processors:
          - metric: # success

[static-moonlight]

thanks

[static-moonlight]

wait a second ... the outer fallback doesn't do anything and is probably still there because i had some trouble to read your answer? :)

so, problably this?

output:
  broker:
    pattern: fan_out_sequential
    outputs:
      - fallback:
        - sql_insert:
          [...]
        - drop: {}
          processors:
            - metric: # failure
                      # you can use `switch` here based on `@fallback_error` to have different metrics for different errors
                
      - drop: {}
        processors:
          - metric: # success

[mihaitodor]

Ah yeah, sorry. I just copy / pasted and adjusted what you had before. Should've spotted that.

[static-moonlight]

thanks, i will try that out :)