Class: Karafka::Web::Pro::Ui::Controllers::Explorer::ExplorerController

Inherits:
BaseController show all
Defined in:
lib/karafka/web/pro/ui/controllers/explorer/explorer_controller.rb

Overview

Data explorer controller

Constant Summary

Constants inherited from Ui::Controllers::BaseController

Ui::Controllers::BaseController::Models

Instance Attribute Summary

Attributes inherited from Ui::Controllers::BaseController

#params, #session

Instance Method Summary collapse

Methods inherited from Ui::Controllers::BaseController

#cache, #initialize

Methods included from Ui::Controllers::Requests::Hookable

included, #run_after_hooks, #run_before_hooks

Constructor Details

This class inherits a constructor from Karafka::Web::Ui::Controllers::BaseController

Instance Method Details

#closest(topic_id, partition_id, time) ⇒ Object

Finds the closest offset matching the requested time and redirects to this location Note, that it redirects to closest but always younger.

Parameters:

  • topic_id (String)
  • partition_id (Integer)
  • time (Time) —

    time of the message



197
198
199
200
201
202
203
204
 
# File 'lib/karafka/web/pro/ui/controllers/explorer/explorer_controller.rb', line 197

def closest(topic_id, partition_id, time)
  target = Web::Ui::Lib::Admin.read_topic(topic_id, partition_id, 1, time).first

  partition_path = "explorer/topics/#{topic_id}/#{partition_id}"
  partition_path += "?offset=#{target.offset}" if target

  redirect(partition_path)
end

#index ⇒ Object

Lists all the topics we can explore



15
16
17
18
19
20
21
22
23
24
25
 
# File 'lib/karafka/web/pro/ui/controllers/explorer/explorer_controller.rb', line 15

def index
  @topics = Models::ClusterInfo
            .topics
            .sort_by { |topic| topic[:topic_name] }

  unless ::Karafka::Web.config.ui.visibility.internal_topics
    @topics.reject! { |topic| topic[:topic_name].start_with?('__') }
  end

  render
end

#partition(topic_id, partition_id) ⇒ Object

Shows messages available in a given partition

Parameters:

  • topic_id (String)
  • partition_id (Integer) 


61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/karafka/web/pro/ui/controllers/explorer/explorer_controller.rb', line 61

def partition(topic_id, partition_id)
  @visibility_filter = ::Karafka::Web.config.ui.policies.messages
  @topic_id = topic_id
  @partition_id = partition_id
  @watermark_offsets = Models::WatermarkOffsets.find(topic_id, partition_id)
  @partitions_count = Models::ClusterInfo.partitions_count(topic_id)

  previous_offset, @messages, next_offset = current_partition_data

  paginate(
    previous_offset,
    @params.current_offset,
    next_offset,
    # If message is an array, it means it's a compacted dummy offset representation
    @messages.map { |message| message.is_a?(Array) ? message.last : message.offset }
  )

  render
end

#recent(topic_id, partition_id) ⇒ Object

Displays the most recent message on a topic/partition

Parameters:

  • topic_id (String)
  • partition_id (Integer, nil) —

    partition we’re interested in or nil if we are interested in the most recent message from all the partitions



127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
 
# File 'lib/karafka/web/pro/ui/controllers/explorer/explorer_controller.rb', line 127

def recent(topic_id, partition_id)
  if partition_id
    active_partitions = [partition_id]
  else
    partitions_count = Models::ClusterInfo.partitions_count(topic_id)
    active_partitions, = Paginators::Partitions.call(partitions_count, 1)
  end

  recent = nil

  # This selects first pages with most recent messages and moves to next if first
  # contains only compacted data, etc.
  #
  # We do it until we find a message we could refer to (if doable) within first
  # ten pages
  10.times do |page|
    messages, = Models::Message.topic_page(topic_id, active_partitions, page + 1)

    # Selects newest out of all partitions
    # Reject compacted messages and transaction-related once
    recent = messages.reject { |message| message.is_a?(Array) }.max_by(&:timestamp)

    break if recent
  end

  recent || not_found!

  show(topic_id, recent.partition, recent.offset, paginate: false)
end

#show(topic_id, partition_id, offset, paginate: true) ⇒ Object

Displays given message

Parameters:

  • topic_id (String)
  • partition_id (Integer)
  • offset (Integer) —

    offset of the message we want to display

  • paginate (Boolean) (defaults to: true) —

    do we want to have pagination



87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/karafka/web/pro/ui/controllers/explorer/explorer_controller.rb', line 87

def show(topic_id, partition_id, offset, paginate: true)
  Lib::PatternsDetector.new.call

  @visibility_filter = ::Karafka::Web.config.ui.policies.messages
  @topic_id = topic_id
  @partition_id = partition_id
  @offset = offset

  watermark_offsets = Models::WatermarkOffsets.find(topic_id, partition_id)

  @message = Models::Message.find(
    @topic_id,
    @partition_id,
    @offset,
    watermark_offsets: watermark_offsets
  )

  # Only if it is not a compacted message we can deserialize stuff
  if @message
    @safe_key = Web::Pro::Ui::Lib::SafeRunner.new { @message.key }.tap(&:call)
    @safe_headers = Web::Pro::Ui::Lib::SafeRunner.new { @message.headers }.tap(&:call)
    @safe_payload = Web::Pro::Ui::Lib::SafeRunner.new { @message.payload }.tap(&:call)
  end

  # This may be off for certain views like recent view where we are interested only
  # in the most recent all the time. It does not make any sense to display pagination
  # there
  #
  # We need watermark offsets to decide if we can paginate left and right
  # We check that because it may have got here via the transactional flow
  paginate(offset, watermark_offsets.low, watermark_offsets.high) if paginate

  render
end

#surrounding(topic_id, partition_id, offset) ⇒ Object

Computes a page on which the given offset is in the middle of the page if possible Useful often when debugging to be able to quickly jump to the historical location of message and its surrounding to understand failure

Parameters:

  • topic_id (String)
  • partition_id (Integer)
  • offset (Integer) —

    offset of the message we want to display



164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
 
# File 'lib/karafka/web/pro/ui/controllers/explorer/explorer_controller.rb', line 164

def surrounding(topic_id, partition_id, offset)
  watermark_offsets = Models::WatermarkOffsets.find(topic_id, partition_id)

  not_found! if offset < watermark_offsets.low
  not_found! if offset >= watermark_offsets.high

  # Assume we start from this offset
  shift = 0
  elements = 0

  # Position the offset as close to the middle of offset based page as possible
  ::Karafka::Web.config.ui.per_page.times do
    break if elements >= ::Karafka::Web.config.ui.per_page

    elements += 1 if offset + shift < watermark_offsets.high

    if offset - shift > watermark_offsets.low
      shift += 1
      elements += 1
    end
  end

  target = offset - shift

  redirect("explorer/topics/#{topic_id}/#{partition_id}?offset=#{target}")
end

#topic(topic_id) ⇒ Object

Note:

This view may not be 100% accurate because we merge multiple partitions data into a single view and this is never accurate. It can be used however to quickly look at most recent data flowing, etc, hence it is still useful for aggregated metrics information

Note:

We cannot use offset references here because each of the partitions may have completely different values

Displays aggregated messages from (potentially) all partitions of a topic

Parameters:

  • topic_id (String) 


38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/karafka/web/pro/ui/controllers/explorer/explorer_controller.rb', line 38

def topic(topic_id)
  @visibility_filter = ::Karafka::Web.config.ui.policies.messages

  @topic_id = topic_id
  @partitions_count = Models::ClusterInfo.partitions_count(topic_id)

  @active_partitions, materialized_page, @limited = Paginators::Partitions.call(
    @partitions_count, @params.current_page
  )

  @messages, next_page = Models::Message.topic_page(
    topic_id, @active_partitions, materialized_page
  )

  paginate(@params.current_page, next_page)

  render
end