wip

CHANGELOG.md

@@ -1,5 +1,17 @@
 # Karafka Web changelog
 
+## 0.6.0 (Unreleased)
+- **[Feature]** Introduce producers errors tracking.
+- [Improvement] Improve pagination by providing a "Go to first page" fast button.
+- [Improvement] Provide more explicit into on the consumers view consumers running.
+- [Improvement] Replace `compact` policy for states topics with `delete,compact` to optimize used storage.
+- [Improvement] Validate error reporting with unified error contract.
+- [Improvement] Use estimated errors count for counters presentation taken from the errors topic instead of materialization via from consumers states to allow for producers errors tracking.
+- [Improvement] Introduce `schema_version` to error reports.
+- [Fix] Fix missing empty `Process name` value in the errors index view.
+- [Refactor] Cleanup common components for errors extraction.
+- [Refactor] Remove not used and reduntant partials.
+
 ## 0.5.2 (2023-05-22)
 - Label ActiveJob consumers jobs with `active_job` tag.
 - Label Virtual Partitions consumers with `virtual` tag.

Gemfile.lock

@@ -1,11 +1,11 @@
 PATH
   remote: .
   specs:
-    karafka-web (0.5.2)
+    karafka-web (0.6.0)
       erubi (~> 1.4)
-      karafka (>= 2.0.40, < 3.0.0)
-      karafka-core (>= 2.0.12, < 3.0.0)
-      roda (~> 3.63)
+      karafka (>= 2.1.2, < 3.0.0)
+      karafka-core (>= 2.0.13, < 3.0.0)
+      roda (~> 3.68, >= 3.68)
       tilt (~> 2.0)
 
 GEM
@@ -26,15 +26,15 @@ GEM
     ffi (1.15.5)
     i18n (1.13.0)
       concurrent-ruby (~> 1.0)
-    karafka (2.0.41)
-      karafka-core (>= 2.0.12, < 3.0.0)
+    karafka (2.1.2)
+      karafka-core (>= 2.0.13, < 3.0.0)
       thor (>= 0.20)
-      waterdrop (>= 2.4.10, < 3.0.0)
+      waterdrop (>= 2.5.3, < 3.0.0)
       zeitwerk (~> 2.3)
-    karafka-core (2.0.12)
+    karafka-core (2.0.13)
       concurrent-ruby (>= 1.1)
-      karafka-rdkafka (>= 0.12.1)
-    karafka-rdkafka (0.12.1)
+      karafka-rdkafka (>= 0.12.3)
+    karafka-rdkafka (0.12.3)
       ffi (~> 1.15)
       mini_portile2 (~> 2.6)
       rake (> 12)
@@ -45,7 +45,7 @@ GEM
       rack (>= 3.0.0.beta1)
       webrick
     rake (13.0.6)
-    roda (3.67.0)
+    roda (3.68.0)
       rack
     rspec (3.12.0)
       rspec-core (~> 3.12.0)
@@ -70,8 +70,8 @@ GEM
     tilt (2.1.0)
     tzinfo (2.0.6)
       concurrent-ruby (~> 1.0)
-    waterdrop (2.5.2)
-      karafka-core (>= 2.0.12, < 3.0.0)
+    waterdrop (2.5.3)
+      karafka-core (>= 2.0.13, < 3.0.0)
       zeitwerk (~> 2.3)
     webrick (1.8.1)
     zeitwerk (2.6.8)

karafka-web.gemspec

@@ -17,9 +17,9 @@ Gem::Specification.new do |spec|
   spec.licenses    = %w[LGPL-3.0 Commercial]
 
   spec.add_dependency 'erubi', '~> 1.4'
-  spec.add_dependency 'karafka', '>= 2.0.40', '< 3.0.0'
-  spec.add_dependency 'karafka-core', '>= 2.0.12', '< 3.0.0'
-  spec.add_dependency 'roda', '~> 3.63'
+  spec.add_dependency 'karafka', '>= 2.1.2', '< 3.0.0'
+  spec.add_dependency 'karafka-core', '>= 2.0.13', '< 3.0.0'
+  spec.add_dependency 'roda', '~> 3.68', '>= 3.68'
   spec.add_dependency 'tilt', '~> 2.0'
 
   spec.add_development_dependency 'rackup', '~> 0.2'

lib/karafka/web/config.rb

@@ -24,20 +24,25 @@ class Config
           # Topic for storing states aggregated info
           setting :states, default: 'karafka_consumers_states'
         end
+
+        setting :producers do
+          # Reports containing particular producers instances informations and metrics.
+          # It is similar in nature to the consumers reports one.
+          setting :reports, default: 'karafka_producers_reports'
+        end
       end
 
       # Tracking and reporting related settings
       setting :tracking do
-        # Collects the metrics we will be dispatching
-        # Tracks and reports the collected metrics
-        setting :reporter, default: Tracking::Reporter.new
-
         # How often should we report data from a single process
         # You may set it to a lower value in development but in production and scale, every
         # 5 seconds should be enough
         setting :interval, default: 5_000
 
         setting :consumers do
+          # Reports the metrics collected in the sampler
+          setting :reporter, default: Tracking::Consumers::Reporter.new
+
           setting :sampler, default: Tracking::Consumers::Sampler.new
 
           setting :listeners, default: [
@@ -51,7 +56,14 @@ class Config
         end
 
         setting :producers do
-          setting :listeners, default: []
+          setting :reporter, default: Tracking::Producers::Reporter.new
+
+          setting :sampler, default: Tracking::Producers::Sampler.new
+
+          setting :listeners, default: [
+            Tracking::Producers::Listeners::Errors.new,
+            Tracking::Producers::Listeners::Reporter.new
+          ]
         end
       end
 

lib/karafka/web/installer.rb

@@ -10,7 +10,7 @@ class Installer
       # @param replication_factor [Integer] replication factor we want to use (1 by default)
       def bootstrap!(replication_factor: 1)
         bootstrap_topics!(replication_factor)
-        bootstrap_state!
+        bootstrap_consumers_state!
       end
 
       # Removes all the Karafka topics and creates them again with the same replication factor
@@ -142,7 +142,7 @@ def bootstrap_topics!(replication_factor = 1)
       end
 
       # Creates the initial state record with all values being empty
-      def bootstrap_state!
+      def bootstrap_consumers_state!
         ::Karafka.producer.produce_sync(
           topic: Karafka::Web.config.topics.consumers.states,
           key: Karafka::Web.config.topics.consumers.states,
@@ -151,7 +151,6 @@ def bootstrap_state!
             stats: {
               batches: 0,
               messages: 0,
-              errors: 0,
               retries: 0,
               dead: 0,
               busy: 0,

lib/karafka/web/tracking/consumers/contracts/consumer_group.rb

@@ -8,7 +8,7 @@ module Consumers
         module Contracts
           # Expected data for each consumer group
           # It's mostly about subscription groups details
-          class ConsumerGroup < BaseContract
+          class ConsumerGroup < Tracking::Contracts::Base
             configure
 
             required(:id) { |val| val.is_a?(String) && !val.empty? }

lib/karafka/web/tracking/consumers/contracts/job.rb

@@ -6,7 +6,7 @@ module Tracking
       module Consumers
         module Contracts
           # Contract for the job reporting details
-          class Job < BaseContract
+          class Job < Tracking::Contracts::Base
             configure
 
             required(:consumer) { |val| val.is_a?(String) }

lib/karafka/web/tracking/consumers/contracts/partition.rb

@@ -6,7 +6,7 @@ module Tracking
       module Consumers
         module Contracts
           # Partition metrics required for web to operate
-          class Partition < BaseContract
+          class Partition < Tracking::Contracts::Base
             configure
 
             required(:id) { |val| val.is_a?(Integer) && val >= 0 }

lib/karafka/web/tracking/consumers/contracts/report.rb

@@ -9,7 +9,7 @@ module Contracts
           #
           # Any outgoing reporting needs to match this format for it to work with the statuses
           # consumer.
-          class Report < BaseContract
+          class Report < Tracking::Contracts::Base
             configure
 
             required(:schema_version) { |val| val.is_a?(String) }

lib/karafka/web/tracking/consumers/contracts/subscription_group.rb

@@ -7,7 +7,7 @@ module Consumers
         module Contracts
           # Expected data for each subscription group
           # It's mostly about topics details
-          class SubscriptionGroup < BaseContract
+          class SubscriptionGroup < Tracking::Contracts::Base
             configure
 
             required(:id) { |val| val.is_a?(String) && !val.empty? }

lib/karafka/web/tracking/consumers/contracts/topic.rb

@@ -6,7 +6,9 @@ module Tracking
       module Consumers
         module Contracts
           # Expected topic information that needs to go out
-          class Topic < BaseContract
+          class Topic < Tracking::Contracts::Base
+            configure
+
             required(:name) { |val| val.is_a?(String) && !val.empty? }
             required(:partitions) { |val| val.is_a?(Hash) }
 

lib/karafka/web/tracking/consumers/listeners/errors.rb

@@ -9,6 +9,11 @@ module Listeners
           class Errors < Base
             include Tracking::Helpers::ErrorInfo
 
+            # Schema used by consumers error reporting
+            SCHEMA_VERSION = '1.0.0'
+
+            private_constant :SCHEMA_VERSION
+
             # Collects errors info and counts errors
             #
             # @param event [Karafka::Core::Monitoring::Event]
@@ -25,6 +30,7 @@ def on_error_occurred(event)
                 error_class, error_message, backtrace = extract_error_info(event[:error])
 
                 sampler.errors << {
+                  schema_version: SCHEMA_VERSION,
                   type: event[:type],
                   error_class: error_class,
                   error_message: error_message,

lib/karafka/web/tracking/consumers/reporter.rb

@@ -23,7 +23,8 @@ class Reporter
           def initialize
             # Move back so first report is dispatched fast to indicate, that the process is alive
             @tracked_at = monotonic_now - 10_000
-            @contract = Consumers::Contracts::Report.new
+            @report_contract = Consumers::Contracts::Report.new
+            @error_contract = Tracking::Contracts::Error.new
           end
 
           # Dispatches the current state from sampler to appropriate topics
@@ -43,7 +44,7 @@ def report(forced: false)
 
               report = sampler.to_report
 
-              @contract.validate!(report)
+              @report_contract.validate!(report)
 
               process_name = report[:process][:name]
 
@@ -59,6 +60,8 @@ def report(forced: false)
 
               # Report errors that occurred (if any)
               messages += sampler.errors.map do |error|
+                @error_contract.validate!(error)
+
                 {
                   topic: Karafka::Web.config.topics.errors,
                   payload: error.to_json,

lib/karafka/web/tracking/contracts/base.rb

@@ -0,0 +1,33 @@
+# frozen_string_literal: true
+
+module Karafka
+  module Web
+    module Tracking
+      module Contracts
+        # Base for all the metric related contracts
+        class Base < ::Karafka::Core::Contractable::Contract
+          class << self
+            # This layer is not for users extensive feedback, thus we can easily use the minimum
+            # error messaging there is.
+            def configure
+              super do |config|
+                config.error_messages = YAML.safe_load(
+                  File.read(
+                    File.join(Karafka::Web.gem_root, 'config', 'locales', 'errors.yml')
+                  )
+                ).fetch('en').fetch('validations').fetch('web')
+              end
+            end
+          end
+
+          # @param data [Hash] data for validation
+          # @return [Boolean] true if all good
+          # @raise [Errors::ContractError] invalid report
+          def validate!(data)
+            super(data, Errors::Tracking::ContractError)
+          end
+        end
+      end
+    end
+  end
+end

lib/karafka/web/tracking/contracts/error.rb

@@ -0,0 +1,31 @@
+# frozen_string_literal: true
+
+module Karafka
+  module Web
+    module Tracking
+      module Contracts
+        # Contract for error reporting
+        # Since producers and consumers report their errors to the same topic, we need to have
+        # a unified contract for both
+        class Error < Base
+          configure
+
+          required(:schema_version) { |val| val.is_a?(String) }
+          required(:type) { |val| val.is_a?(String) && !val.empty? }
+          required(:error_class) { |val| val.is_a?(String) && !val.empty? }
+          required(:error_message) { |val| val.is_a?(String) }
+          required(:backtrace) { |val| val.is_a?(String) }
+          required(:details) { |val| val.is_a?(Hash) }
+          required(:occurred_at) { |val| val.is_a?(Float) }
+
+          nested(:process) do
+            required(:name) { |val| val.is_a?(String) && !val.empty? }
+            # Tags may not be present for producers because they may operate from outside of
+            # karafka taggable process
+            optional(:tags) { |val| val.is_a?(Karafka::Core::Taggable::Tags) }
+          end
+        end
+      end
+    end
+  end
+end

lib/karafka/web/tracking/helpers/error_info.rb

@@ -3,7 +3,9 @@
 module Karafka
   module Web
     module Tracking
+      # Namespace for tracking related helpers
       module Helpers
+        # Module containing some helper methods useful for extracting extra errors info
         module ErrorInfo
           # Extracts the basic error info
           #