Implement an EventBus system to publish events and metrics

[pulse00]

I'd like to propose to continue the discussion started here in a separate issue.

Similar to the EventBus mechanism in the couchbase sdk, lettuce could introduce an EventBus emitting RedisEvents:

RedisClient client = new RedisClient("localhost")

client
    .eventBus()
    .get()
    .filter(event -> event.type().equals(EventType.SERVER))
    .subscribe(System.err::println);

This EventBus system could be used both by lettuce and spinach.

My proposal would be to initially create 2 types of events: EventType.SERVER (which would emit any information concerning the redis cluster) and EventType.CLIENT (which would emit any information concerning the client itself).

Questions

• How should the Observable emit events? Something like:

Schedulers.newThread().createWorker().schedulePeriodically(() -> subject.onNext(redisEvent), 0, 1, TimeUnit.SECONDS);

[mp911de]

Thanks, that's an amazing approach to creating more value for the user. I like the cluster events idea very much. The metrics/event bus story tackles a couple aspect:

• Where to manage/instantiate the event bus: There can be many RedisClient instances, and there are already ClientOptions. It would be great having a way to either distinct/combine/reuse event busses so one can decide where to collect metrics
• Emission of events and collecting data for the events: On which level aggregate network latencies. I tend to say per connection (IP quadruple) and per command type. That would result in a view that reveals latencies for a particular connection per command type (GET, SET, ...)
• Emission of events: Should be a configurable period that clears the aggregation values after emitting the events
• Aggregation: We certainly don't want to emit every command completion as event, rather aggregate a couple of commands until the metrics event is dispatched
• What means network latency to us? Until the first byte of the command comes back? Until the command is finished? There is a huge difference since a lot of things happen within the JDK until the command is completed. The streaming API, for instance, allows to pass in a callback that is invoked on every data element. One could Thread.sleep(1000) within the callback that causes the time until command completion to explode. The only reasonable solution seems to me to collect both times.
• Command-Retry also comes into play for the cluster part. A command can be rejected with the MOVED or ASK redirect and then the command is dispatched to the appropriate node

I'm not sure, whether this a 4.x or a 3.x(3.4) topic. Introducing rxjava to 3.4 is not a big deal. The result of the ticket and the metrics/event bus seems really amazing to me and I'm looking forward to get it into lettuce and make it available for clients built on top of lettuce such as spinach.

[mp911de]

See also #110. The event bus could be provided by a ClientConfig instead of client options. Client options are somewhat lightweight and can be replaced while using the client. The event bus is more like infrastructure and should not be changed during operations.

[mp911de]

Events:

• Connect/Disconnect
  • Payload: Host-Identifier, Connection?
• Exception in processing Leave it out for now.
  • Payload: Host-Identifier, Connection?, current active command?
• Latency Metrics event
  • Payload: Set of metrics, each latency metric has a key of the Host-Identifier and command type and contains the min/max/count and percentiles of the total duration (from writing the command until it is complete)

The event bus deprecates then the existing ConnectionEvents. Hope this makes sense.

[mp911de]

Command latency example:

Raw output:

CommandLatencyId [local:any -> localhost/127.0.0.1:6479, commandType=FLUSHDB]=[count=23, timeUnit=MICROSECONDS, 
    firstResponse=[min=166, max=1310, percentiles={50.0=227, 90.0=292, 95.0=395,  99.0=1310, 99.9=1310}], 
    completion=   [min=520, max=1892, percentiles={50.0=733, 90.0=937, 95.0=1187, 99.0=1892, 99.9=1892}]]

JSON:

{
  "[local:any -> localhost/127.0.0.1:6479, commandType=FLUSHALL]": {
    "count": 23,
    "timeUnit": "MICROSECONDS",
    "firstResponse": {
      "min": 151,
      "max": 1474,
      "percentiles": {
        "50.0": 232,
        "90.0": 307,
        "95.0": 323,
        "99.0": 1474,
        "99.9": 1474
      }
    },
    "completion": {
      "min": 548,
      "max": 4718,
      "percentiles": {
        "50.0": 794,
        "90.0": 1036,
        "95.0": 1073,
        "99.0": 4718,
        "99.9": 4718
      }
    }
  }
}

[mp911de]

Implemented.