Add consumer offset_lag to rabbitmq-stream CLI command & Management

[gerhard]

This is an important metric to keep track of and be aware (maybe even alert on) when consumers fall behind consuming stream messages. While they should be able to catch up, if they fall behind too much and the stream gets truncated, they may miss on messages.

This is something that we want to expose via Prometheus metrics as well, but we've started closer to the core, CLI & Management.

While at it, we've made the commands use PrettyTable by default.

This should be merged as soon as it passes CI, we shouldn't wait on the Prometheus changes - they can come later.

Pair: @kjnilsson

[acogoluegnes]

Failures in the rabbit_stream_queue test suite:

rabbit_stream_queue_SUITE > single_node_parallel > consume_from_relative_time_offset
    #1. {error,
            {{shutdown,
                 {server_initiated_close,406,
                     <<"PRECONDITION_FAILED - invalid arg 'x-stream-offset' for queue 'consume_from_relative_time_offset' in vhost '/': invalid_stream_offset_arg">>}},
             {gen_server,call,
                 [<0.963.0>,
                  {subscribe,
                      {'basic.consume',0,
                          <<"consume_from_relative_time_offset">>,<<"ctag">>,
                          false,false,false,false,
                          [{<<"x-stream-offset">>,longstr,<<"100s">>}]},
                      <0.534.0>},
                  infinity]}}}

rabbit_stream_queue_SUITE > cluster_size_2_parallel > consume_from_relative_time_offset
    #1. {error,
            {{shutdown,
                 {server_initiated_close,406,
                     <<"PRECONDITION_FAILED - invalid arg 'x-stream-offset' for queue 'consume_from_relative_time_offset' in vhost '/': invalid_stream_offset_arg">>}},
             {gen_server,call,
                 [<0.1873.0>,
                  {subscribe,
                      {'basic.consume',0,
                          <<"consume_from_relative_time_offset">>,<<"ctag">>,
                          false,false,false,false,
                          [{<<"x-stream-offset">>,longstr,<<"100s">>}]},
                      <0.1380.0>},
                  infinity]}}}

rabbit_stream_queue_SUITE > cluster_size_3_parallel > consume_from_relative_time_offset
    #1. {error,
            {{shutdown,
                 {server_initiated_close,406,
                     <<"PRECONDITION_FAILED - invalid arg 'x-stream-offset' for queue 'consume_from_relative_time_offset' in vhost '/': invalid_stream_offset_arg">>}},
             {gen_server,call,
                 [<0.3521.0>,
                  {subscribe,
                      {'basic.consume',0,
                          <<"consume_from_relative_time_offset">>,<<"ctag">>,
                          false,false,false,false,
                          [{<<"x-stream-offset">>,longstr,<<"100s">>}]},
                      <0.2758.0>},
                  infinity]}}}

[fabiorosa-sn]

any idea on when this metric will be made available on prometheus ?

[michaelklishin]

@fabiorosa-sn this metric needs to be exposed in multiple places, for example, it is not actually exposed via CLI. rabbitmq_stream_management is the only place where it is exposed.

We do not make ETA promises. First step is usually expose something like this to rabbitmq-streams stream_status.

[johanrhodin]

@gomoripeti @markus812498 this might be something for us to contribute (the CLI + prometheus parts)

[markusgust]

it looks like it's already exposed to the stream CLI. Can look and see if I can figure out how exposing to prometheus works:

➜  sbin git:(main) ./rabbitmq-streams list_stream_consumers
Listing stream consumers ...
┌─────────────────┬────────┬───────────────────┬─────────────┬────────────┬─────────┬────────┬─────────────────┬────────────┐
│ subscription_id │ stream │ messages_consumed │ offset      │ offset_lag │ credits │ active │ activity_status │ properties │
├─────────────────┼────────┼───────────────────┼─────────────┼────────────┼─────────┼────────┼─────────────────┼────────────┤
│ 0               │ stream │ 1573021579        │ 31484410142 │ 0          │ 0       │ true   │ up              │ #{}        │
├─────────────────┼────────┼───────────────────┼─────────────┼────────────┼─────────┼────────┼─────────────────┼────────────┤
│ 0               │ stream │ 1573021579        │ 31484410142 │ 0          │ 0       │ true   │ up              │ #{}        │
└─────────────────┴────────┴───────────────────┴─────────────┴────────────┴─────────┴────────┴─────────────────┴────────────┘