From 537c9e245e0199254f81c85f82e46808bbe7e9f9 Mon Sep 17 00:00:00 2001 From: ohhAndy Date: Thu, 1 Oct 2026 10:12:07 -0400 Subject: [PATCH 1/6] Move EventParser and specs into lib tree --- .../kubernetes/workers/event_parser.rb | 79 +++++++++++++++++++ .../workers}/data/node_rebooted.yml | 0 .../kubernetes/workers}/data/pod_killing.yml | 0 .../workers}/data/pod_scheduled.yml | 0 .../data/replication_controller_event.yml | 0 .../workers/event_catcher_base_spec.rb} | 4 +- .../kubernetes/workers}/event_parser_spec.rb | 3 +- workers/event_catcher/event_parser.rb | 71 ----------------- 8 files changed, 81 insertions(+), 76 deletions(-) create mode 100644 lib/manageiq/providers/kubernetes/workers/event_parser.rb rename spec/{workers/event_catcher => lib/manageiq/providers/kubernetes/workers}/data/node_rebooted.yml (100%) rename spec/{workers/event_catcher => lib/manageiq/providers/kubernetes/workers}/data/pod_killing.yml (100%) rename spec/{workers/event_catcher => lib/manageiq/providers/kubernetes/workers}/data/pod_scheduled.yml (100%) rename spec/{workers/event_catcher => lib/manageiq/providers/kubernetes/workers}/data/replication_controller_event.yml (100%) rename spec/{workers/event_catcher/kubernetes_event_catcher_base_spec.rb => lib/manageiq/providers/kubernetes/workers/event_catcher_base_spec.rb} (99%) rename spec/{workers/event_catcher => lib/manageiq/providers/kubernetes/workers}/event_parser_spec.rb (97%) delete mode 100644 workers/event_catcher/event_parser.rb diff --git a/lib/manageiq/providers/kubernetes/workers/event_parser.rb b/lib/manageiq/providers/kubernetes/workers/event_parser.rb new file mode 100644 index 0000000000..accae30362 --- /dev/null +++ b/lib/manageiq/providers/kubernetes/workers/event_parser.rb @@ -0,0 +1,79 @@ +module ManageIQ + module Providers + module Kubernetes + module Workers + class EventParser + def self.extract_event_data(event) + involved_object = event.object.involvedObject + return {} if involved_object.nil? + + event_data = { + :timestamp => event.object.lastTimestamp || event.object.eventTime, + :kind => involved_object.kind, + :name => involved_object.name, + :namespace => involved_object.namespace, + :reason => event.object.reason, + :message => event.object.message, + :uid => involved_object.uid, + :event_uid => event.object.metadata.uid, + } + + event_data[:fieldpath] = involved_object.fieldPath unless involved_object.fieldPath.nil? + + return {} if event_data[:kind].nil? || event_data[:reason].nil? + + event_type_prefix = event_data[:kind].upcase + case event_data[:kind] + when 'Node' + event_data[:container_node_name] = event_data[:name] + when 'Pod' + /^spec.containers{(?.*)}$/ =~ event_data[:fieldpath] + event_data[:container_name] = container_name unless container_name.nil? + event_data[:container_group_name] = event_data[:name] + event_data[:container_namespace] = event_data[:namespace] + # TODO: ReplicationController is deprecated in favour of ReplicaSet/Deployment; + when 'ReplicationController' + event_type_prefix = 'REPLICATOR' + event_data[:container_replicator_name] = event_data[:name] + event_data[:container_namespace] = event_data[:namespace] + when 'ReplicaSet', 'Deployment', 'StatefulSet', 'DaemonSet', 'Job', 'CronJob' + event_data[:container_namespace] = event_data[:namespace] + end + + event_data[:event_type] = "#{event_type_prefix}_#{event_data[:reason].upcase}" + event_data + end + + def self.event_to_hash_from_data(event, ems_id = nil) + # TODO: ReplicationController is deprecated in favour of ReplicaSet/Deployment; + ems_ref_key = case event[:kind] + when 'Node' + :container_node_ems_ref + when 'Pod' + :container_group_ems_ref + when 'ReplicationController' + :container_replicator_ems_ref + end + + event_hash = { + :event_type => event[:event_type], + :source => 'KUBERNETES', + :timestamp => event[:timestamp], + :message => event[:message], + :container_node_name => event[:container_node_name], + :container_group_name => event[:container_group_name], + :container_replicator_name => event[:container_replicator_name], + :container_namespace => event[:container_namespace], + :container_name => event[:container_name], + :full_data => event, + :ems_id => ems_id, + :ems_ref => event[:event_uid], + } + event_hash[ems_ref_key] = event[:uid] if ems_ref_key + event_hash + end + end + end + end + end +end diff --git a/spec/workers/event_catcher/data/node_rebooted.yml b/spec/lib/manageiq/providers/kubernetes/workers/data/node_rebooted.yml similarity index 100% rename from spec/workers/event_catcher/data/node_rebooted.yml rename to spec/lib/manageiq/providers/kubernetes/workers/data/node_rebooted.yml diff --git a/spec/workers/event_catcher/data/pod_killing.yml b/spec/lib/manageiq/providers/kubernetes/workers/data/pod_killing.yml similarity index 100% rename from spec/workers/event_catcher/data/pod_killing.yml rename to spec/lib/manageiq/providers/kubernetes/workers/data/pod_killing.yml diff --git a/spec/workers/event_catcher/data/pod_scheduled.yml b/spec/lib/manageiq/providers/kubernetes/workers/data/pod_scheduled.yml similarity index 100% rename from spec/workers/event_catcher/data/pod_scheduled.yml rename to spec/lib/manageiq/providers/kubernetes/workers/data/pod_scheduled.yml diff --git a/spec/workers/event_catcher/data/replication_controller_event.yml b/spec/lib/manageiq/providers/kubernetes/workers/data/replication_controller_event.yml similarity index 100% rename from spec/workers/event_catcher/data/replication_controller_event.yml rename to spec/lib/manageiq/providers/kubernetes/workers/data/replication_controller_event.yml diff --git a/spec/workers/event_catcher/kubernetes_event_catcher_base_spec.rb b/spec/lib/manageiq/providers/kubernetes/workers/event_catcher_base_spec.rb similarity index 99% rename from spec/workers/event_catcher/kubernetes_event_catcher_base_spec.rb rename to spec/lib/manageiq/providers/kubernetes/workers/event_catcher_base_spec.rb index e646b282fb..56c33c7c60 100644 --- a/spec/workers/event_catcher/kubernetes_event_catcher_base_spec.rb +++ b/spec/lib/manageiq/providers/kubernetes/workers/event_catcher_base_spec.rb @@ -1,6 +1,4 @@ -require 'kubeclient' -require 'recursive-open-struct' -require_relative '../../../workers/event_catcher/event_parser' +require_relative '../../../lib/manageiq/providers/kubernetes/workers/event_parser' require_relative '../../../lib/manageiq/providers/kubernetes/workers/event_catcher_base' RSpec.describe KubernetesEventCatcherBase do diff --git a/spec/workers/event_catcher/event_parser_spec.rb b/spec/lib/manageiq/providers/kubernetes/workers/event_parser_spec.rb similarity index 97% rename from spec/workers/event_catcher/event_parser_spec.rb rename to spec/lib/manageiq/providers/kubernetes/workers/event_parser_spec.rb index 3456ea505f..e2e3a80fb0 100644 --- a/spec/workers/event_catcher/event_parser_spec.rb +++ b/spec/lib/manageiq/providers/kubernetes/workers/event_parser_spec.rb @@ -1,5 +1,4 @@ -require 'recursive-open-struct' -require_relative '../../../workers/event_catcher/event_parser' +require_relative '../../../lib/manageiq/providers/kubernetes/workers/event_parser' RSpec.describe EventParser do let(:fixture_dir) { File.expand_path('data', __dir__) } diff --git a/workers/event_catcher/event_parser.rb b/workers/event_catcher/event_parser.rb deleted file mode 100644 index f14855941b..0000000000 --- a/workers/event_catcher/event_parser.rb +++ /dev/null @@ -1,71 +0,0 @@ -class EventParser - def self.extract_event_data(event) - involved_object = event.object.involvedObject - return {} if involved_object.nil? - - event_data = { - :timestamp => event.object.lastTimestamp || event.object.eventTime, - :kind => involved_object.kind, - :name => involved_object.name, - :namespace => involved_object.namespace, - :reason => event.object.reason, - :message => event.object.message, - :uid => involved_object.uid, - :event_uid => event.object.metadata.uid, - } - - event_data[:fieldpath] = involved_object.fieldPath unless involved_object.fieldPath.nil? - - return {} if event_data[:kind].nil? || event_data[:reason].nil? - - event_type_prefix = event_data[:kind].upcase - case event_data[:kind] - when 'Node' - event_data[:container_node_name] = event_data[:name] - when 'Pod' - /^spec.containers{(?.*)}$/ =~ event_data[:fieldpath] - event_data[:container_name] = container_name unless container_name.nil? - event_data[:container_group_name] = event_data[:name] - event_data[:container_namespace] = event_data[:namespace] - # TODO: ReplicationController is deprecated in favour of ReplicaSet/Deployment; - when 'ReplicationController' - event_type_prefix = 'REPLICATOR' - event_data[:container_replicator_name] = event_data[:name] - event_data[:container_namespace] = event_data[:namespace] - when 'ReplicaSet', 'Deployment', 'StatefulSet', 'DaemonSet', 'Job', 'CronJob' - event_data[:container_namespace] = event_data[:namespace] - end - - event_data[:event_type] = "#{event_type_prefix}_#{event_data[:reason].upcase}" - event_data - end - - def self.event_to_hash_from_data(event, ems_id = nil) - # TODO: ReplicationController is deprecated in favour of ReplicaSet/Deployment; - ems_ref_key = case event[:kind] - when 'Node' - :container_node_ems_ref - when 'Pod' - :container_group_ems_ref - when 'ReplicationController' - :container_replicator_ems_ref - end - - event_hash = { - :event_type => event[:event_type], - :source => 'KUBERNETES', - :timestamp => event[:timestamp], - :message => event[:message], - :container_node_name => event[:container_node_name], - :container_group_name => event[:container_group_name], - :container_replicator_name => event[:container_replicator_name], - :container_namespace => event[:container_namespace], - :container_name => event[:container_name], - :full_data => event, - :ems_id => ems_id, - :ems_ref => event[:event_uid], - } - event_hash[ems_ref_key] = event[:uid] if ems_ref_key - event_hash - end -end From 1a5c670b4a2b657b6ab1cc1df7cb241d38b3d050 Mon Sep 17 00:00:00 2001 From: ohhAndy Date: Thu, 1 Oct 2026 10:15:02 -0400 Subject: [PATCH 2/6] Namespace EventCatcherBase and EventParser under ManageIQ::Providers::Kubernetes::Workers --- .../kubernetes/workers/event_catcher_base.rb | 417 +++++++++--------- .../workers/event_catcher_base_spec.rb | 19 +- .../kubernetes/workers/event_parser_spec.rb | 4 +- workers/event_catcher/event_catcher.rb | 2 +- workers/event_catcher/worker | 43 +- 5 files changed, 245 insertions(+), 240 deletions(-) diff --git a/lib/manageiq/providers/kubernetes/workers/event_catcher_base.rb b/lib/manageiq/providers/kubernetes/workers/event_catcher_base.rb index 19868c109a..42d1b31135 100644 --- a/lib/manageiq/providers/kubernetes/workers/event_catcher_base.rb +++ b/lib/manageiq/providers/kubernetes/workers/event_catcher_base.rb @@ -1,207 +1,220 @@ -class KubernetesEventCatcherBase - # Kinds not modelled in ManageIQ inventory; dropped before parsing regardless - # of reason. Not operator-configurable — there is no valid use-case for them. - DISABLED_KINDS = %w[Endpoints EndpointSlice Lease VirtualMachine VirtualMachineInstance].freeze - - # Fraction of a token's remaining TTL to use as a safety margin before forcing a reconnect. - TOKEN_REFRESH_MARGIN_RATIO = 0.1 - - def initialize(ems, endpoint, authentication, settings, messaging, logger) - @ems = ems - @endpoint = endpoint - @authentication = authentication - @settings = settings - @messaging = messaging - @logger = logger - ems_settings_key = "ems_#{ems['ems_type']}" - @filtered_events = (settings['blacklisted_event_names'] || settings.dig('ems', ems_settings_key, 'blacklisted_event_names') || []).map(&:to_s) - end - - def run! - notify_started - logger.info("#{log_prefix} Collecting events...") - version = nil - loop do - client = build_client - version = client.get_events.resourceVersion if version.nil? - logger.info("#{log_prefix} Watching from resourceVersion=#{version}") - version = watch_events(client, version) - end - rescue Interrupt - logger.info("#{log_prefix} Interrupted") - ensure - notify_stopping - end - - def stop! - end - - private - - def build_client - @token_expiry = nil # cleared before auth_options so a failed/incomplete fetch never leaves a stale expiry - client = Kubeclient::Client.new( - URI::HTTPS.build(:host => endpoint['hostname'], :port => (endpoint['port'] || 443).to_i), - 'v1', - :ssl_options => Kubeclient::Client::DEFAULT_SSL_OPTIONS.merge( - :verify_ssl => verify_ssl_mode, - :cert_store => ca_cert_store - ), - :auth_options => auth_options - ) - client.discover - client - end - - def watch_events(client, version) - watcher = client.watch_events(version) - timer = schedule_token_refresh(watcher) - - watcher.each do |event| - event_data = EventParser.extract_event_data(event) - - if event_data.empty? - if event.type == "ERROR" - code = event.object&.code - logger.warn("#{log_prefix} Received ERROR watch event (code=#{code}), clearing version and reconnecting") - version = nil if code == 410 - break +require 'kubeclient' +require 'manageiq-messaging' +require_relative 'event_parser' + +module ManageIQ + module Providers + module Kubernetes + module Workers + class EventCatcherBase + # Kinds not modelled in ManageIQ inventory; dropped before parsing regardless + # of reason. Not operator-configurable — there is no valid use-case for them. + DISABLED_KINDS = %w[Endpoints EndpointSlice Lease VirtualMachine VirtualMachineInstance].freeze + + # Fraction of a token's remaining TTL to use as a safety margin before forcing a reconnect. + TOKEN_REFRESH_MARGIN_RATIO = 0.1 + + def initialize(ems, endpoint, authentication, settings, messaging, logger) + @ems = ems + @endpoint = endpoint + @authentication = authentication + @settings = settings + @messaging = messaging + @logger = logger + ems_settings_key = "ems_#{ems['ems_type']}" + @filtered_events = (settings['blacklisted_event_names'] || settings.dig('ems', ems_settings_key, 'blacklisted_event_names') || []).map(&:to_s) + end + + def run! + notify_started + logger.info("#{log_prefix} Collecting events...") + version = nil + loop do + client = build_client + version = client.get_events.resourceVersion if version.nil? + logger.info("#{log_prefix} Watching from resourceVersion=#{version}") + version = watch_events(client, version) + end + rescue Interrupt + logger.info("#{log_prefix} Interrupted") + ensure + notify_stopping + end + + def stop! + end + + private + + def build_client + @token_expiry = nil # cleared before auth_options so a failed/incomplete fetch never leaves a stale expiry + client = Kubeclient::Client.new( + URI::HTTPS.build(:host => endpoint['hostname'], :port => (endpoint['port'] || 443).to_i), + 'v1', + :ssl_options => Kubeclient::Client::DEFAULT_SSL_OPTIONS.merge( + :verify_ssl => verify_ssl_mode, + :cert_store => ca_cert_store + ), + :auth_options => auth_options + ) + client.discover + client + end + + def watch_events(client, version) + watcher = client.watch_events(version) + timer = schedule_token_refresh(watcher) + + watcher.each do |event| + event_data = EventParser.extract_event_data(event) + + if event_data.empty? + if event.type == "ERROR" + code = event.object&.code + logger.warn("#{log_prefix} Received ERROR watch event (code=#{code}), clearing version and reconnecting") + version = nil if code == 410 + break + end + logger.info("#{log_prefix} Skipping event with no involvedObject (type=#{event.type})") + next + end + + if filtered?(event_data) + logger.debug("#{log_prefix} Filtered event kind=#{event_data[:kind]} reason=#{event_data[:reason]} event_type=#{event_data[:event_type]}") + next + end + + unless event_valid?(event_data) + logger.info("#{log_prefix} Invalid event (no timestamp) kind=#{event_data[:kind]} reason=#{event_data[:reason]}") + next + end + + logger.info("#{log_prefix} Queuing event kind=#{event_data[:kind]} reason=#{event_data[:reason]} name=#{event_data[:name]} namespace=#{event_data[:namespace]}") + + version = (event_data[:timestamp] && event.dig('metadata', 'resourceVersion')) || version + publish_events([EventParser.event_to_hash_from_data(event_data, ems['id'])]) + heartbeat + end + version + rescue EOFError, OpenSSL::SSL::SSLError, Kubeclient::HttpError => error + logger.info("#{log_prefix} Monitoring connection closed, reconnecting... #{error}") + version + ensure + timer&.kill + end + + attr_reader :ems, :endpoint, :authentication, :settings, :messaging, :logger, :filtered_events + + # Override in subclasses that issue short-lived tokens. + def token_expiry + nil + end + + def schedule_token_refresh(watcher) + expiry = token_expiry + return nil if expiry.nil? + + delay = token_refresh_delay(expiry) + logger.info("#{log_prefix} Token expires in #{(expiry - Time.now.utc).round}s, scheduling refresh in #{delay.round}s") + Thread.new do + sleep(delay) + watcher.finish + end + end + + def token_refresh_delay(expiry) + ttl = expiry - Time.now.utc + [ttl * (1 - TOKEN_REFRESH_MARGIN_RATIO), 0].max + end + + def filtered?(event_data) + DISABLED_KINDS.include?(event_data[:kind]) || filtered_events.include?(event_data[:event_type]) + end + + def event_valid?(event_data) + !event_data[:timestamp].nil? + end + + def publish_events(events) + events.each do |event| + messaging_client.publish_topic( + :service => 'manageiq.ems', + :sender => ems['id'], + :event => event[:event_type], + :payload => event + ) + end + end + + def messaging_client + @messaging_client ||= ManageIQ::Messaging::Client.open(messaging.merge(:client_ref => "kubernetes-event-catcher-#{ems['id']}")) + end + + def verify_ssl_mode + case endpoint['security_protocol'] + when nil, '' + endpoint['verify_ssl'].to_i != OpenSSL::SSL::VERIFY_NONE ? OpenSSL::SSL::VERIFY_PEER : OpenSSL::SSL::VERIFY_NONE + when 'ssl-without-validation' + OpenSSL::SSL::VERIFY_NONE + else + OpenSSL::SSL::VERIFY_PEER + end + end + + # Builds an OpenSSL::X509::Store from the endpoint's certificate_authority PEM, or nil if absent. + # Supports CA chains by splitting on PEM boundaries before loading. + def ca_cert_store + pem = endpoint['certificate_authority'] + return nil if pem.nil? || pem.strip.empty? + + store = OpenSSL::X509::Store.new + pem.split(/(?=-----BEGIN)/).each do |fragment| + store.add_cert(OpenSSL::X509::Certificate.new(fragment)) + end + store + end + + # Override in subclasses to provide provider-specific authentication options. + # Returns a hash passed directly to Kubeclient as :auth_options. + def auth_options + options = {} + options[:username] = authentication['userid'] if authentication['userid'] && authentication['password'] + options[:password] = authentication['password'] if options[:username] + options[:bearer_token] = authentication['auth_key'] if authentication['auth_key'] + options + end + + def notify_started + if ENV.fetch('NOTIFY_SOCKET', nil) + require 'sd_notify' + SdNotify.ready + elsif ENV.fetch('WORKER_HEARTBEAT_FILE', nil) + heartbeat_to_file + end + end + + def heartbeat + if ENV.fetch('NOTIFY_SOCKET', nil) + SdNotify.watchdog + elsif ENV.fetch('WORKER_HEARTBEAT_FILE', nil) + heartbeat_to_file + end + end + + def notify_stopping + SdNotify.stopping if ENV.fetch('NOTIFY_SOCKET', nil) + end + + def heartbeat_to_file + File.write(ENV.fetch('WORKER_HEARTBEAT_FILE'), Time.now.to_i + (settings.dig('worker_settings', 'heartbeat_timeout') || 120)) + end + + # Override in subclasses to customise the log prefix (e.g. to include the subclass name). + def log_prefix + 'MIQ(ManageIQ::Providers::Kubernetes::ContainerManager::EventCatcher)' + end end - logger.info("#{log_prefix} Skipping event with no involvedObject (type=#{event.type})") - next - end - - if filtered?(event_data) - logger.debug("#{log_prefix} Filtered event kind=#{event_data[:kind]} reason=#{event_data[:reason]} event_type=#{event_data[:event_type]}") - next - end - - unless event_valid?(event_data) - logger.info("#{log_prefix} Invalid event (no timestamp) kind=#{event_data[:kind]} reason=#{event_data[:reason]}") - next end - - logger.info("#{log_prefix} Queuing event kind=#{event_data[:kind]} reason=#{event_data[:reason]} name=#{event_data[:name]} namespace=#{event_data[:namespace]}") - - version = (event_data[:timestamp] && event.dig('metadata', 'resourceVersion')) || version - publish_events([EventParser.event_to_hash_from_data(event_data, ems['id'])]) - heartbeat - end - version - rescue EOFError, OpenSSL::SSL::SSLError, Kubeclient::HttpError => error - logger.info("#{log_prefix} Monitoring connection closed, reconnecting... #{error}") - version - ensure - timer&.kill - end - - attr_reader :ems, :endpoint, :authentication, :settings, :messaging, :logger, :filtered_events - - # Override in subclasses that issue short-lived tokens. - def token_expiry - nil - end - - def schedule_token_refresh(watcher) - expiry = token_expiry - return nil if expiry.nil? - - delay = token_refresh_delay(expiry) - logger.info("#{log_prefix} Token expires in #{(expiry - Time.now.utc).round}s, scheduling refresh in #{delay.round}s") - Thread.new do - sleep(delay) - watcher.finish - end - end - - def token_refresh_delay(expiry) - ttl = expiry - Time.now.utc - [ttl * (1 - TOKEN_REFRESH_MARGIN_RATIO), 0].max - end - - def filtered?(event_data) - DISABLED_KINDS.include?(event_data[:kind]) || filtered_events.include?(event_data[:event_type]) - end - - def event_valid?(event_data) - !event_data[:timestamp].nil? - end - - def publish_events(events) - events.each do |event| - messaging_client.publish_topic( - :service => 'manageiq.ems', - :sender => ems['id'], - :event => event[:event_type], - :payload => event - ) - end - end - - def messaging_client - @messaging_client ||= ManageIQ::Messaging::Client.open(messaging.merge(:client_ref => "kubernetes-event-catcher-#{ems['id']}")) - end - - def verify_ssl_mode - case endpoint['security_protocol'] - when nil, '' - endpoint['verify_ssl'].to_i != OpenSSL::SSL::VERIFY_NONE ? OpenSSL::SSL::VERIFY_PEER : OpenSSL::SSL::VERIFY_NONE - when 'ssl-without-validation' - OpenSSL::SSL::VERIFY_NONE - else - OpenSSL::SSL::VERIFY_PEER - end - end - - # Builds an OpenSSL::X509::Store from the endpoint's certificate_authority PEM, or nil if absent. - # Supports CA chains by splitting on PEM boundaries before loading. - def ca_cert_store - pem = endpoint['certificate_authority'] - return nil if pem.nil? || pem.strip.empty? - - store = OpenSSL::X509::Store.new - pem.split(/(?=-----BEGIN)/).each do |fragment| - store.add_cert(OpenSSL::X509::Certificate.new(fragment)) end - store - end - - # Override in subclasses to provide provider-specific authentication options. - # Returns a hash passed directly to Kubeclient as :auth_options. - def auth_options - options = {} - options[:username] = authentication['userid'] if authentication['userid'] && authentication['password'] - options[:password] = authentication['password'] if options[:username] - options[:bearer_token] = authentication['auth_key'] if authentication['auth_key'] - options - end - - def notify_started - if ENV['NOTIFY_SOCKET'] - SdNotify.ready - elsif ENV['WORKER_HEARTBEAT_FILE'] - heartbeat_to_file - end - end - - def heartbeat - if ENV['NOTIFY_SOCKET'] - SdNotify.watchdog - elsif ENV['WORKER_HEARTBEAT_FILE'] - heartbeat_to_file - end - end - - def notify_stopping - SdNotify.stopping if ENV['NOTIFY_SOCKET'] - end - - def heartbeat_to_file - File.write(ENV.fetch('WORKER_HEARTBEAT_FILE'), Time.now.to_i + (settings.dig('worker_settings', 'heartbeat_timeout') || 120)) - end - - # Override in subclasses to customise the log prefix (e.g. to include the subclass name). - def log_prefix - 'MIQ(ManageIQ::Providers::Kubernetes::ContainerManager::EventCatcher)' end end diff --git a/spec/lib/manageiq/providers/kubernetes/workers/event_catcher_base_spec.rb b/spec/lib/manageiq/providers/kubernetes/workers/event_catcher_base_spec.rb index 56c33c7c60..7f54d74035 100644 --- a/spec/lib/manageiq/providers/kubernetes/workers/event_catcher_base_spec.rb +++ b/spec/lib/manageiq/providers/kubernetes/workers/event_catcher_base_spec.rb @@ -1,7 +1,4 @@ -require_relative '../../../lib/manageiq/providers/kubernetes/workers/event_parser' -require_relative '../../../lib/manageiq/providers/kubernetes/workers/event_catcher_base' - -RSpec.describe KubernetesEventCatcherBase do +RSpec.describe ManageIQ::Providers::Kubernetes::Workers::EventCatcherBase do let(:ems) { {'id' => 1, 'type' => 'ManageIQ::Providers::Kubernetes::ContainerManager', 'ems_type' => 'kubernetes'} } let(:endpoint) { {'hostname' => 'localhost'} } let(:authentication) { {'auth_key' => 'test-token'} } @@ -40,7 +37,7 @@ def normal_event(kind, reason) describe '#filtered?' do described_class::DISABLED_KINDS.each do |kind| it "filters #{kind} events (disabled kind)" do - expect(base_catcher.send(:filtered?, EventParser.extract_event_data(normal_event(kind, 'SomeReason')))).to be(true) + expect(base_catcher.send(:filtered?, ManageIQ::Providers::Kubernetes::Workers::EventParser.extract_event_data(normal_event(kind, 'SomeReason')))).to be(true) end end @@ -50,18 +47,18 @@ def normal_event(kind, reason) it 'filters blacklisted events from scoped worker settings' do scoped_catcher = described_class.new(ems, endpoint, authentication, {'blacklisted_event_names' => ['NODE_REBOOTED']}, {}, logger) - expect(scoped_catcher.send(:filtered?, EventParser.extract_event_data(normal_event('Node', 'Rebooted')))).to be(true) + expect(scoped_catcher.send(:filtered?, ManageIQ::Providers::Kubernetes::Workers::EventParser.extract_event_data(normal_event('Node', 'Rebooted')))).to be(true) end it 'filters blacklisted events from full ems settings fallback' do settings['ems']['ems_kubernetes']['blacklisted_event_names'] = ['NODE_REBOOTED'] - expect(base_catcher.send(:filtered?, EventParser.extract_event_data(normal_event('Node', 'Rebooted')))).to be(true) + expect(base_catcher.send(:filtered?, ManageIQ::Providers::Kubernetes::Workers::EventParser.extract_event_data(normal_event('Node', 'Rebooted')))).to be(true) end it 'accepts non-disabled kind events with arbitrary reasons' do - expect(base_catcher.send(:filtered?, EventParser.extract_event_data(normal_event('Node', 'Unknown')))).to be(false) - expect(base_catcher.send(:filtered?, EventParser.extract_event_data(normal_event('Pod', 'CustomReason')))).to be(false) - expect(base_catcher.send(:filtered?, EventParser.extract_event_data(normal_event('Deployment', 'ScalingReplicaSet')))).to be(false) + expect(base_catcher.send(:filtered?, ManageIQ::Providers::Kubernetes::Workers::EventParser.extract_event_data(normal_event('Node', 'Unknown')))).to be(false) + expect(base_catcher.send(:filtered?, ManageIQ::Providers::Kubernetes::Workers::EventParser.extract_event_data(normal_event('Pod', 'CustomReason')))).to be(false) + expect(base_catcher.send(:filtered?, ManageIQ::Providers::Kubernetes::Workers::EventParser.extract_event_data(normal_event('Deployment', 'ScalingReplicaSet')))).to be(false) end end @@ -253,7 +250,7 @@ def normal_event(kind, reason) it 'skips and logs events with no involvedObject without crashing' do bare_event = double('WatchEvent', :type => 'BOOKMARK') - allow(EventParser).to receive(:extract_event_data).and_return({}) + allow(ManageIQ::Providers::Kubernetes::Workers::EventParser).to receive(:extract_event_data).and_return({}) allow(watcher).to receive(:each).and_yield(bare_event) expect(logger).to receive(:info).with(/Skipping event with no involvedObject/) expect(base_catcher.send(:watch_events, client, '42')).to eq('42') diff --git a/spec/lib/manageiq/providers/kubernetes/workers/event_parser_spec.rb b/spec/lib/manageiq/providers/kubernetes/workers/event_parser_spec.rb index e2e3a80fb0..80f62f9710 100644 --- a/spec/lib/manageiq/providers/kubernetes/workers/event_parser_spec.rb +++ b/spec/lib/manageiq/providers/kubernetes/workers/event_parser_spec.rb @@ -1,6 +1,4 @@ -require_relative '../../../lib/manageiq/providers/kubernetes/workers/event_parser' - -RSpec.describe EventParser do +RSpec.describe ManageIQ::Providers::Kubernetes::Workers::EventParser do let(:fixture_dir) { File.expand_path('data', __dir__) } def load_event(name) diff --git a/workers/event_catcher/event_catcher.rb b/workers/event_catcher/event_catcher.rb index 710888210f..a1631b627f 100644 --- a/workers/event_catcher/event_catcher.rb +++ b/workers/event_catcher/event_catcher.rb @@ -1,4 +1,4 @@ require_relative '../../lib/manageiq/providers/kubernetes/workers/event_catcher_base' -class EventCatcher < KubernetesEventCatcherBase +class EventCatcher < ManageIQ::Providers::Kubernetes::Workers::EventCatcherBase end diff --git a/workers/event_catcher/worker b/workers/event_catcher/worker index 73925f5e06..78519e2b25 100755 --- a/workers/event_catcher/worker +++ b/workers/event_catcher/worker @@ -1,25 +1,13 @@ #!/usr/bin/env ruby -require 'bundler/inline' - -gemfile(false) do - source 'https://rubygems.org' - gem 'kubeclient', '~> 4.7' - gem 'recursive-open-struct', '~> 1.1' - gem 'manageiq-messaging', '~> 2.1' - gem 'manageiq-loggers', '~> 2.0' - gem 'sd_notify', '~> 0.1.0' if ENV['APPLIANCE'] -end - -require 'json' -require 'uri' -require_relative 'event_parser' require_relative 'event_catcher' def build_logger - if ENV['CONTAINER'] + require 'manageiq-loggers' + + if ENV.fetch('CONTAINER', nil) ManageIQ::Loggers::Container.new - elsif ENV['APPLIANCE'] + elsif ENV.fetch('APPLIANCE', nil) ManageIQ::Loggers::Journald.new else ManageIQ::Loggers::Base.new($stdout) @@ -27,15 +15,24 @@ def build_logger end def parse_args + require 'json' + JSON.parse($stdin.read) rescue JSON::ParserError abort('Invalid argument format') end -args = parse_args -ems = args['ems'].find { |m| m['type'].end_with?('ContainerManager') } -endpoint = ems['endpoints'].find { |item| item['role'] == 'default' } -authentication = ems['authentications'].find { |item| item['authtype'] == 'bearer' || item['authtype'] == 'default' } -logger = build_logger -logger.progname = 'evm' -EventCatcher.new(ems, endpoint, authentication, args['settings'], args['messaging']&.transform_keys(&:to_sym) || {}, logger).run! +def main(args) + logger = build_logger + logger.progname = 'evm' + + ems = args['ems'].find { |m| m['type'].end_with?('ContainerManager') } + endpoint = ems['endpoints'].find { |item| item['role'] == 'default' } + authentication = ems['authentications'].find { |item| item['authtype'] == 'bearer' || item['authtype'] == 'default' } + settings = args['settings'] + messaging = args['messaging']&.transform_keys(&:to_sym) || {} + + EventCatcher.new(ems, endpoint, authentication, settings, messaging, logger).run! +end + +main(parse_args) From 38b602df7e673b91ce03b79bc5e4620ae710a87d Mon Sep 17 00:00:00 2001 From: ohhAndy Date: Thu, 1 Oct 2026 10:19:35 -0400 Subject: [PATCH 3/6] Fix rubocop: NegatedIfElseCondition and HashLikeCase --- .../kubernetes/workers/event_catcher_base.rb | 2 +- .../providers/kubernetes/workers/event_parser.rb | 13 +++++-------- 2 files changed, 6 insertions(+), 9 deletions(-) diff --git a/lib/manageiq/providers/kubernetes/workers/event_catcher_base.rb b/lib/manageiq/providers/kubernetes/workers/event_catcher_base.rb index 42d1b31135..b88956975f 100644 --- a/lib/manageiq/providers/kubernetes/workers/event_catcher_base.rb +++ b/lib/manageiq/providers/kubernetes/workers/event_catcher_base.rb @@ -153,7 +153,7 @@ def messaging_client def verify_ssl_mode case endpoint['security_protocol'] when nil, '' - endpoint['verify_ssl'].to_i != OpenSSL::SSL::VERIFY_NONE ? OpenSSL::SSL::VERIFY_PEER : OpenSSL::SSL::VERIFY_NONE + endpoint['verify_ssl'].to_i == OpenSSL::SSL::VERIFY_NONE ? OpenSSL::SSL::VERIFY_NONE : OpenSSL::SSL::VERIFY_PEER when 'ssl-without-validation' OpenSSL::SSL::VERIFY_NONE else diff --git a/lib/manageiq/providers/kubernetes/workers/event_parser.rb b/lib/manageiq/providers/kubernetes/workers/event_parser.rb index accae30362..7ac7c288b3 100644 --- a/lib/manageiq/providers/kubernetes/workers/event_parser.rb +++ b/lib/manageiq/providers/kubernetes/workers/event_parser.rb @@ -46,14 +46,11 @@ def self.extract_event_data(event) def self.event_to_hash_from_data(event, ems_id = nil) # TODO: ReplicationController is deprecated in favour of ReplicaSet/Deployment; - ems_ref_key = case event[:kind] - when 'Node' - :container_node_ems_ref - when 'Pod' - :container_group_ems_ref - when 'ReplicationController' - :container_replicator_ems_ref - end + ems_ref_key = { + 'Node' => :container_node_ems_ref, + 'Pod' => :container_group_ems_ref, + 'ReplicationController' => :container_replicator_ems_ref, + }[event[:kind]] event_hash = { :event_type => event[:event_type], From 3c21f623f99408eddb233c449317fbd0923f067e Mon Sep 17 00:00:00 2001 From: ohhAndy Date: Thu, 1 Oct 2026 10:27:03 -0400 Subject: [PATCH 4/6] Remove worker gemfile sync spec --- .../event_catcher/worker_gemfile_sync_spec.rb | 16 ---------------- 1 file changed, 16 deletions(-) delete mode 100644 spec/workers/event_catcher/worker_gemfile_sync_spec.rb diff --git a/spec/workers/event_catcher/worker_gemfile_sync_spec.rb b/spec/workers/event_catcher/worker_gemfile_sync_spec.rb deleted file mode 100644 index 646f363c9c..0000000000 --- a/spec/workers/event_catcher/worker_gemfile_sync_spec.rb +++ /dev/null @@ -1,16 +0,0 @@ -require 'bundler' - -RSpec.describe 'event catcher worker inline gems' do - let(:worker_path) { File.expand_path('../../../workers/event_catcher/worker', __dir__) } - let(:lockfile) { Bundler::LockfileParser.new(Bundler.read_file(File.expand_path('../../../Gemfile.lock', __dir__))) } - - it 'uses versions compatible with Gemfile.lock' do - source = File.read(worker_path) - gems = source.scan(/gem ['"]([^'"]+)['"], ['"]([^'"]+)['"]/).to_h - gems.each do |name, requirement| - locked = lockfile.specs.find { |spec| spec.name == name } - expect(locked).not_to be_nil - expect(Gem::Requirement.new(requirement).satisfied_by?(locked.version)).to be(true) - end - end -end From a780e27c094f818f7c9a2bcca26836851dcc4bdd Mon Sep 17 00:00:00 2001 From: ohhAndy Date: Thu, 1 Oct 2026 12:27:42 -0400 Subject: [PATCH 5/6] Update namespace and dir path for standalone worker --- workers/event_catcher/event_catcher.rb | 4 ---- .../container_manager/event_catcher/event_catcher.rb | 12 ++++++++++++ .../container_manager}/event_catcher/worker | 2 +- 3 files changed, 13 insertions(+), 5 deletions(-) delete mode 100644 workers/event_catcher/event_catcher.rb create mode 100644 workers/manageiq/providers/kubernetes/container_manager/event_catcher/event_catcher.rb rename workers/{ => manageiq/providers/kubernetes/container_manager}/event_catcher/worker (86%) diff --git a/workers/event_catcher/event_catcher.rb b/workers/event_catcher/event_catcher.rb deleted file mode 100644 index a1631b627f..0000000000 --- a/workers/event_catcher/event_catcher.rb +++ /dev/null @@ -1,4 +0,0 @@ -require_relative '../../lib/manageiq/providers/kubernetes/workers/event_catcher_base' - -class EventCatcher < ManageIQ::Providers::Kubernetes::Workers::EventCatcherBase -end diff --git a/workers/manageiq/providers/kubernetes/container_manager/event_catcher/event_catcher.rb b/workers/manageiq/providers/kubernetes/container_manager/event_catcher/event_catcher.rb new file mode 100644 index 0000000000..dba87ff9e5 --- /dev/null +++ b/workers/manageiq/providers/kubernetes/container_manager/event_catcher/event_catcher.rb @@ -0,0 +1,12 @@ +require_relative '../../../../../../lib/manageiq/providers/kubernetes/workers/event_catcher_base' + +module ManageIQ + module Providers + module Kubernetes + module ContainerManager + class EventCatcher < ManageIQ::Providers::Kubernetes::Workers::EventCatcherBase + end + end + end + end +end diff --git a/workers/event_catcher/worker b/workers/manageiq/providers/kubernetes/container_manager/event_catcher/worker similarity index 86% rename from workers/event_catcher/worker rename to workers/manageiq/providers/kubernetes/container_manager/event_catcher/worker index 78519e2b25..04335e4c6f 100755 --- a/workers/event_catcher/worker +++ b/workers/manageiq/providers/kubernetes/container_manager/event_catcher/worker @@ -32,7 +32,7 @@ def main(args) settings = args['settings'] messaging = args['messaging']&.transform_keys(&:to_sym) || {} - EventCatcher.new(ems, endpoint, authentication, settings, messaging, logger).run! + ManageIQ::Providers::Kubernetes::ContainerManager::EventCatcher.new(ems, endpoint, authentication, settings, messaging, logger).run! end main(parse_args) From ace07f94fd177f25cb83cf3626117feec6804e9e Mon Sep 17 00:00:00 2001 From: ohhAndy Date: Tue, 6 Oct 2026 14:04:02 -0400 Subject: [PATCH 6/6] Move standalone EventCatcher to Workers namespace, fix require to use load path, and add bundler setup to worker script --- .../providers/kubernetes/workers}/event_catcher.rb | 4 ++-- .../kubernetes/container_manager/event_catcher/worker | 7 +++++-- 2 files changed, 7 insertions(+), 4 deletions(-) rename {workers/manageiq/providers/kubernetes/container_manager/event_catcher => lib/manageiq/providers/kubernetes/workers}/event_catcher.rb (59%) diff --git a/workers/manageiq/providers/kubernetes/container_manager/event_catcher/event_catcher.rb b/lib/manageiq/providers/kubernetes/workers/event_catcher.rb similarity index 59% rename from workers/manageiq/providers/kubernetes/container_manager/event_catcher/event_catcher.rb rename to lib/manageiq/providers/kubernetes/workers/event_catcher.rb index dba87ff9e5..0574b342fd 100644 --- a/workers/manageiq/providers/kubernetes/container_manager/event_catcher/event_catcher.rb +++ b/lib/manageiq/providers/kubernetes/workers/event_catcher.rb @@ -1,9 +1,9 @@ -require_relative '../../../../../../lib/manageiq/providers/kubernetes/workers/event_catcher_base' +require 'manageiq/providers/kubernetes/workers/event_catcher_base' module ManageIQ module Providers module Kubernetes - module ContainerManager + module Workers class EventCatcher < ManageIQ::Providers::Kubernetes::Workers::EventCatcherBase end end diff --git a/workers/manageiq/providers/kubernetes/container_manager/event_catcher/worker b/workers/manageiq/providers/kubernetes/container_manager/event_catcher/worker index 04335e4c6f..746eb34852 100755 --- a/workers/manageiq/providers/kubernetes/container_manager/event_catcher/worker +++ b/workers/manageiq/providers/kubernetes/container_manager/event_catcher/worker @@ -1,6 +1,9 @@ #!/usr/bin/env ruby -require_relative 'event_catcher' +ENV['BUNDLE_GEMFILE'] ||= File.expand_path('../../../../../../Gemfile', __dir__) +require 'bundler/setup' if File.exist?(ENV['BUNDLE_GEMFILE']) + +require 'manageiq/providers/kubernetes/workers/event_catcher' def build_logger require 'manageiq-loggers' @@ -32,7 +35,7 @@ def main(args) settings = args['settings'] messaging = args['messaging']&.transform_keys(&:to_sym) || {} - ManageIQ::Providers::Kubernetes::ContainerManager::EventCatcher.new(ems, endpoint, authentication, settings, messaging, logger).run! + ManageIQ::Providers::Kubernetes::Workers::EventCatcher.new(ems, endpoint, authentication, settings, messaging, logger).run! end main(parse_args)