<commit_before>require 'triglav/agent/hdfs/watcher'
require 'triglav/agent/hdfs/connection'

module Triglav::Agent
  module Hdfs
    module Worker
      def initialize
        @timer = Timer.new
      end

      def reload
        $setting.reload
      end

      def run
        until @stop
          @timer.wait(watcher_interval) { process }
        end
      end

      def process
        $logger.debug { "Start Processing worker_id:#{worker_id}" }
        api_client = ApiClient.new # renew connection

        # It is possible to seperate agent process by prefixes of resource uris
        resource_uri_prefixes.each do |resource_uri_prefix|
          # list_aggregated_resources returns unique resources which we have to monitor
          if aggregated_resources = api_client.list_aggregated_resources(resource_uri_prefix)
            $logger.debug { "resource_uri_prefix:#{resource_uri_prefix} aggregated_resources.size:#{aggregated_resources.size}" }
            connection = Connection.new(get_connection_info(resource_uri_prefix))
            watcher = Watcher.new(connection)
            aggregated_resources.each do |resource|
              watcher.process(resource) {|events| api_client.send_messages(events) }
            end
          end
        end
        $logger.debug { "End Processing worker_id:#{worker_id}" }
      end

      def stop
        $logger.info { "Stop processing worker_id:#{worker_id}" }
        @stop = true
        @timer.signal
      end

      private

      def watcher_interval
        $setting.dig(:hdfs, :watcher_interval) || 60
      end

      def resource_uri_prefixes<commit_msg>Fix to restart via SIGHUP<commit_after>require 'triglav/agent/hdfs/watcher'
require 'triglav/agent/hdfs/connection'

module Triglav::Agent
  module Hdfs
    module Worker
      def initialize
      end

      def reload
        $logger.info { "Worker#reload worker_id:#{worker_id}" }
        $setting.reload
      end

      def run
        $logger.info { "Worker#run worker_id:#{worker_id}" }
        @timer = Timer.new
        @stop = false
        until @stop
          @timer.wait(watcher_interval) { process }
        end
      end

      def process
        $logger.debug { "Start Worker#process worker_id:#{worker_id}" }
        api_client = ApiClient.new # renew connection

        # It is possible to seperate agent process by prefixes of resource uris
        resource_uri_prefixes.each do |resource_uri_prefix|
          # list_aggregated_resources returns unique resources which we have to monitor
          if aggregated_resources = api_client.list_aggregated_resources(resource_uri_prefix)
            $logger.debug { "resource_uri_prefix:#{resource_uri_prefix} aggregated_resources.size:#{aggregated_resources.size}" }
            connection = Connection.new(get_connection_info(resource_uri_prefix))
            watcher = Watcher.new(connection)
            aggregated_resources.each do |resource|
              watcher.process(resource) {|events| api_client.send_messages(events) }
            end
          end
        end
        $logger.debug { "End Worker#process worker_id:#{worker_id}" }
      end

      def stop
        $logger.info { "Worker#stop worker_id:#{worker_id}" }
        @stop = true
        @timer.stop
      end

      private

      def watcher_interval
        $setting.dig(:hdfs, :watcher_interval) || 60
      end

      def resource_uri_prefixes
{'fraction_non_alphanumeric': 0.07872402692420252, 'fraction_numerical': 0.001170617500731636, 'mean_word_length': 2.4490413723511604, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '341974', 'n_tokens_mistral': 984, 'n_tokens_neox': 943, 'n_words': 218}