Class: Karafka::Pro::Instrumentation::PerformanceTracker

Inherits:
Object
  • Object
show all
Includes:
Singleton
Defined in:
lib/karafka/pro/instrumentation/performance_tracker.rb

Overview

Note:

Even if we have some race-conditions here it is relevant due to the quantity of data. This is why we do not mutex it.

Tracker used to keep track of performance metrics It provides insights that can be used to optimize processing flow

Instance Method Summary collapse

Constructor Details

#initialize ⇒ PerformanceTracker

Note:

Samples are scoped by subscription group id so that two groups consuming the same topic name keep independent measurements - and so revoking a partition in one group never drops samples another group is still using.

Builds up nested concurrent hash for data tracking



52
53
54
55
56
57
58
59
60
 
# File 'lib/karafka/pro/instrumentation/performance_tracker.rb', line 52

def initialize
  @processing_times = Hash.new do |groups_hash, group_id|
    groups_hash[group_id] = Hash.new do |topics_hash, topic|
      topics_hash[topic] = Hash.new do |partitions_hash, partition|
        partitions_hash[partition] = []
      end
    end
  end
end

Instance Method Details

#on_consumer_consumed(event) ⇒ Object

Tracks time taken to process a single message of a given topic partition

Parameters:

  • event (Karafka::Core::Monitoring::Event) —

    event details



78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/karafka/pro/instrumentation/performance_tracker.rb', line 78

def on_consumer_consumed(event)
  consumer = event[:caller]
  messages = consumer.messages
  group_id = consumer.topic.subscription_group.id
  topic = messages.metadata.topic
  partition = messages.metadata.partition

  samples = @processing_times[group_id][topic][partition]
  samples << (event[:time] / messages.size)

  return unless samples.size > SAMPLES_COUNT

  samples.shift
end

#on_rebalance_partitions_revoked(event) ⇒ Object

Evicts the processing time samples of revoked partitions so the tracker does not retain them for the whole process lifetime. Without this every (topic, partition) ever consumed keeps its samples array forever - unbounded under regex pattern subscriptions that keep discovering new topic names. Mirrors the offset metadata fetcher, which also clears on revoke. Samples are scoped by subscription group, so we only ever evict the revoked group's data and never another group consuming the same topic.

Parameters:

  • event (Karafka::Core::Monitoring::Event) —

    rebalance revoked event details



100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/karafka/pro/instrumentation/performance_tracker.rb', line 100

def on_rebalance_partitions_revoked(event)
  group_id = event[:subscription_group_id]

  # Do not auto-vivify a branch for a group we have never tracked
  return unless @processing_times.key?(group_id)

  group_times = @processing_times[group_id]

  event[:tpl].to_h.each do |topic, partitions|
    next unless group_times.key?(topic)

    topic_times = group_times[topic]
    partitions.each { |partition| topic_times.delete(partition.partition) }
    group_times.delete(topic) if topic_times.empty?
  end

  @processing_times.delete(group_id) if group_times.empty?
end

#processing_time_p95(group_id, topic, partition) ⇒ Float

Returns p95 processing time of a single message from a single topic partition.

Parameters:

  • group_id (String) —

    subscription group id the topic partition belongs to

  • topic (String)
  • partition (Integer)

Returns:

  • (Float) —

    p95 processing time of a single message from a single topic partition



66
67
68
69
70
71
72
73
 
# File 'lib/karafka/pro/instrumentation/performance_tracker.rb', line 66

def processing_time_p95(group_id, topic, partition)
  values = @processing_times[group_id][topic][partition]

  return 0 if values.empty?
  return values.first if values.size == 1

  percentile(0.95, values)
end