# frozen_string_literal: true module Karafka module Web module Ui module Models # Model that represents the general status of the Web UI. # # We use this data to display a status page that helps with debugging on what is missing # in the overall setup of the Web UI. # # People have various problems like too many partitions, not created topics, etc. and this # data and view aims to help them with understanding the current status of the setup class Status # Status of a single step of setup Step = Struct.new(:status, :details) do # @return [Boolean] is the given step successfully configured and working def success? status == :success || status == :warning end # @return [String] local namespace for partial of a given type def partial_namespace case status when :success then 'successes' when :warning then 'warnings' when :failure then 'failures' when :halted then 'failures' else raise ::Karafka::Errors::UnsupportedCaseError, status end end # @return [String] stringified status def to_s status.to_s end end # Is karafka-web enabled in the `karafka.rb` # Checks if the consumer group for web-ui is injected. # It does **not** check if the group is active because this may depend on the # configuration details, but for the Web-UI web app to work, the routing needs to be # aware of the deserializer, etc def enabled enabled = ::Karafka::App.routes.map(&:name).include?( ::Karafka::Web.config.processing.consumer_group ) Step.new( enabled ? :success : :failure, nil ) end # @return [Status::Step] were we able to connect to Kafka or not and how fast. # Some people try to work with Kafka over the internet with really high latency and this # should be highlighted in the UI as often the connection just becomes unstable def connection if enabled.success? # Do not connect more than once during the status object lifetime @connection_time || connect level = if @connection_time < 1_000 :success elsif @connection_time < 1_000_000 :warning else :failure end else level = :halted end Step.new( level, { time: @connection_time } ) end # @return [Status::Step] do all the needed topics exist def topics if connection.success? details = topics_details status = details.all? { |_, detail| detail[:present] } ? :success : :failure else status = :halted details = {} end Step.new( status, details ) end # @return [Status::Step] do we have all topics with expected number of partitions def partitions if topics.success? status = :success status = :failure if topics_details[topics_consumers_states][:partitions] != 1 status = :failure if topics_details[topics_consumers_reports][:partitions] != 1 status = :failure if topics_details[topics_consumers_metrics][:partitions] != 1 details = topics_details else status = :halted details = {} end Step.new( status, details ) end # @return [Status::Step] do we have correct replication for given env def replication if partitions.success? status = :success # low replication is not an error but just a warning and a potential problem # in case of a crash, this is why we do not fail but warn only status = :warning if topics_details.values.any? { |det| det[:replication] < 2 } # Allow for non-production setups to use replication 1 as it is not that relevant status = :success unless Karafka.env.production? details = topics_details else status = :halted details = {} end Step.new( status, details ) end # @return [Status::Step] Is the initial consumers state present in Kafka and that they # can be deserialized def initial_consumers_state details = { issue_type: :presence } if replication.success? begin @current_state ||= Models::ConsumersState.current status = @current_state ? :success : :failure rescue JSON::ParserError status = :failure details[:issue_type] = :deserialization end else status = :halted end Step.new( status, details ) end # @return [Status::Step] Is the initial consumers metrics record present in Kafka and # that they can be deserialized def initial_consumers_metrics details = { issue_type: :presence } if initial_consumers_state.success? begin @current_metrics ||= Models::ConsumersMetrics.current status = @current_metrics ? :success : :failure rescue JSON::ParserError status = :failure details[:issue_type] = :deserialization end else status = :halted end Step.new( status, details ) end # @return [Status::Step] could we read and operate on the current processes data (if any) def consumers_reports if initial_consumers_metrics.success? @processes ||= Models::Processes.active(@current_state) status = :success else status = :halted end Step.new(status, nil) rescue JSON::ParserError Step.new(:failure, nil) end # @return [Status::Step] Is there at least one active karafka server reporting to the # Web UI def live_reporting status = if consumers_reports.success? @processes.empty? ? :failure : :success else :halted end Step.new( status, nil ) end # @return [Status::Step] is there a subscription to our reports topic that is being # consumed actively. def state_calculation if live_reporting.success? @subscriptions ||= Models::Health .current(@current_state) .values.map { |consumer_group| consumer_group[:topics] } .flat_map(&:keys) status = @subscriptions.include?(topics_consumers_reports) ? :success : :failure else status = :halted end Step.new( status, nil ) end # @return [Status::Step] Are we able to actually digest the consumers reports with the # consumer that is consuming them. def consumers_reports_schema_state status = if state_calculation.success? @current_state[:schema_state] == 'compatible' ? :success : :failure else :halted end Step.new( status, nil ) end # @return [Status::Step] are there any active topics in the routing that are not present # in the cluster (does not apply to patterns) def routing_topics_presence if consumers_reports_schema_state.success? existing = @cluster_info.topics.map { |topic| topic[:topic_name] } missing = ::Karafka::App .routes .flat_map(&:topics) .flat_map { |topics| topics.map(&:itself) } .select(&:active?) .reject { |topic| topic.respond_to?(:patterns?) ? topic.patterns? : false } .map(&:name) .uniq .then { |routed_topics| routed_topics - existing } Step.new(missing.empty? ? :success : :warning, missing) else Step.new(:halted, []) end end # @return [Status::Step] is Pro enabled with all of its features. # @note It's not an error not to have it but we want to warn, that some of the features # may not work without Pro. def pro_subscription Step.new( ::Karafka.pro? ? :success : :warning, nil ) end private # @return [String] consumers states topic name def topics_consumers_states ::Karafka::Web.config.topics.consumers.states.to_s end # @return [String] consumers reports topic name def topics_consumers_reports ::Karafka::Web.config.topics.consumers.reports.to_s end # @return [String] consumers metrics topic name def topics_consumers_metrics ::Karafka::Web.config.topics.consumers.metrics.to_s end # @return [String] errors topic name def topics_errors ::Karafka::Web.config.topics.errors end # @return [Hash] hash with topics with which we work details (even if don't exist) def topics_details base = { present: false, partitions: 0, replication: 1 } topics = { topics_consumers_states => base.dup, topics_consumers_reports => base.dup, topics_consumers_metrics => base.dup, topics_errors => base.dup } @cluster_info.topics.each do |topic| name = topic[:topic_name] next unless topics.key?(name) topics[name].merge!( present: true, partitions: topic[:partition_count], replication: topic[:partitions].map { |part| part[:replica_count] }.max ) end topics end # Tries connecting with the cluster and saves the cluster info and the connection time # @note If fails, `connection_time` will be 1_000_000 def connect started = Time.now.to_f # For status we always need uncached data, otherwise status could cache outdated # info @cluster_info = Models::ClusterInfo.fetch(cached: false) @connection_time = (Time.now.to_f - started) * 1_000 rescue ::Rdkafka::RdkafkaError @connection_time = 1_000_000 end end end end end end