<commit_before>require "sensu/daemon"
require "sensu/server/filter"
require "sensu/server/mutate"
require "sensu/server/handle"

module Sensu
  module Server
    class Process
      include Daemon
      include Filter
      include Mutate
      include Handle

      attr_reader :is_leader, :in_progress

      METRIC_CHECK_TYPE = "metric".freeze

      EVENT_FLAPPING_ACTION = "flapping".freeze

      DEFAULT_HANDLER_NAME = "default".freeze

      # Create an instance of the Sensu server process, start the
      # server within the EventMachine event loop, and set up server
      # process signal traps (for stopping).
      #
      # @param options [Hash]
      def self.run(options={})
        server = self.new(options)
        EM::run do
          server.start
          server.setup_signal_traps
        end
      end

      # Override Daemon initialize() to support Sensu server leader
      # election and the handling event count.
      #
      # @param options [Hash]
      def initialize(options={})
        super
        @is_leader = false
        @timers[:leader] = Array.new
        @in_progress = Hash.new(0)
      end

      # Set up the Redis and Transport connection objects, `@redis`
      # and `@transport`. This method updates the Redis on error
      # callback to reset the in progress check result counter. This
      # method "drys" up many instances of `setup_redis()` and
      # `setup_transport()`, particularly in the specs.
      #
      # @yield callback/block called after connecting to Redis and the
      #   Sensu Transport.
      def setup_connections
        setup_redis do
          @redis.on_error do |error|
            @logger.error("redis connection error", :error => error.to_s)
            @in_progress[:check_results] = 0
          end
          setup_transport do
            yield
          end
        end
      end

      # Create a registration check definition for a client. Client
      # definitions may contain `:registration` configuration,
      # containing custom attributes and handler information. By
      # default, the registration check definition sets the `:handler`
      # to `registration`. If the client provides its own
      # `:registration` configuration, it's deep merged with the
      # defaults. The check `:name`, `:output`, `:status`, `:issued`,
      # and `:executed` values are always overridden to guard against
      # an invalid definition.
      def create_registration_check(client)
        check = {:handler => "registration"}
        if client.has_key?(:registration)
          check = deep_merge(check, client[:registration])
        end
        timestamp = Time.now.to_i
        overrides = {
          :name => "registration",
          :output => "new client registration",
          :status => 1,
          :issued => timestamp,
          :executed => timestamp
        }
        check.merge(overrides)
      end

      # Create and process a client registration event. A registration
      # event is created when a Sensu client is first added to the
      # client registry. The `create_registration_check()` method is
      # called to create a registration check definition for the
      # client.
      #
      # @param client [Hash] definition.
      def create_client_registration_event(client)
        event = {
          :id => random_uuid,
          :client => client,
          :check => create_registration_check(client),
          :occurrences => 1,
          :action => :create,
          :timestamp => Time.now.to_i
        }
        process_event(event)
      end

      # Process an initial client registration, when it is first added
      # to the client registry. If a registration handler is defined
      # or the client specifies one, a client registration event is
      # created and processed (handled, etc.) for the client
      # (`create_client_registration_event()`).
      #
      # @param client [Hash] definition.
      def process_client_registration(client)
        if @settings.handler_exists?("registration") || client[:registration]
          create_client_registration_event(client)
        end
      end

      # Update the Sensu client registry, stored in Redis. Sensu
      # client data is used to provide additional event context and
      # enable agent health monitoring. The client registry supports
      # client signatures, unique string identifiers used for
      # keepalive and result source verification. If a client has a
      # signature, all further registry updates for the client must
      # have the same signature. A client can begin to use a signature
      # if one was not previously configured. JSON serialization is
      # used for the stored client data.
      #
      # @param client [Hash]
      # @yield [success] passes success status to optional
      #   callback/block.
      # @yieldparam success [TrueClass,FalseClass] indicating if the
      #   client registry update was a success or the client data was
      #   discarded due to client signature mismatch.
      def update_client_registry(client)
        @logger.debug("updating client registry", :client => client)
        client_key = "client:#{client[:name]}"
        signature_key = "#{client_key}:signature"
        @redis.setnx(signature_key, client[:signature]) do |created|
          process_client_registration(client) if created
          @redis.get(signature_key) do |signature|
            if (signature.nil? || signature.empty?) && client[:signature]
              @redis.set(signature_key, client[:signature])
            end
            if signature.nil? || signature.empty? || (client[:signature] == signature)
              @redis.multi
              @redis.set(client_key, Sensu::JSON.dump(client))
              @redis.sadd("clients", client[:name])
              @redis.exec do
                yield(true) if block_given?
              end
            else
              @logger.warn("invalid client signature", {
                :client => client,
                :signature => signature
              })
              @logger.warn("not updating client in the registry", :client => client)
              yield(false) if block_given?
            end
          end
        end
      end

      # Set up the client keepalive consumer, keeping the Sensu client
      # registry updated. The consumer receives JSON serialized client
      # keepalives from the transport, parses them, and calls
      # `update_client_registry()` with the client data to update the
      # registry. Transport message acknowledgements are used to
      # ensure the client registry is updated successfully. Keepalive
      # JSON parsing errors are logged.
      def setup_keepalives
        @logger.debug("subscribing to keepalives")
        @transport.subscribe(:direct, "keepalives", "keepalives", :ack => true) do |message_info, message|
          @logger.debug("received keepalive", :message => message)
          begin
            client = Sensu::JSON.load(message)
            update_client_registry(client) do
              @transport.ack(message_info)
            end
          rescue Sensu::JSON::ParseError => error
            @logger.error("failed to parse keepalive payload", {
              :message => message,
              :error => error.to_s
            })
            @transport.ack(message_info)
          end
        end
      end

      # Expand event handler sets, creating an array of handler
      # definitions. Handler sets cannot be deeply nested (by choice),
      # this method will return `nil` if an attempt is made to deeply
      # nest. If the provided handler definition is not a set, it is
      # returned.
      #
      # @param handler [Hash] definition.
      # @param depth [Integer] of the expansion.
      # @return [Array, Hash, Nil]
      def expand_handler_sets(handler, depth=0)
        if handler[:type] == "set"
          if depth < 2
            derive_handlers(handler[:handlers], depth + 1)
          else
            @logger.error("handler sets cannot be deeply nested", :handler => handler)
            nil
          end
        else
          handler
        end
      end

      # Derive an array of handler definitions from a list of handler
      # names. This method first checks for the existence of standard
      # handlers, followed by handler extensions. If a handler does
      # not exist for a name, it is logged and ignored. Duplicate
      # handler definitions are removed.
      #
      # @param handler_list [Array]
      # @param depth [Integer] of handler set expansion.
      # @return [Array]
      def derive_handlers(handler_list, depth=0)
        handler_list.compact.map { |handler_name|
          case
          when @settings.handler_exists?(handler_name)
            handler = @settings[:handlers][handler_name].merge(:name => handler_name)
            expand_handler_sets(handler, depth)
          when @extensions.handler_exists?(handler_name)
            @extensions[:handlers][handler_name]
          else
            @logger.error("unknown handler", :handler_name => handler_name)
            nil
          end
        }.flatten.compact.uniq
      end

      # Process an event: filter -> mutate -> handle.
      #
      # This method determines the appropriate handlers for an event,
      # filtering and mutating the event data for each of them. The
      # `@in_progress[:events]` is incremented by `1`, for each event
      # handler chain (filter -> mutate -> handle).
      #
      # @param event [Hash]
      def process_event(event)
        log_level = event[:check][:type] == METRIC_CHECK_TYPE ? :debug : :info
        @logger.send(log_level, "processing event", :event => event)
        handler_list = Array((event[:check][:handlers] || event[:check][:handler]) || DEFAULT_HANDLER_NAME)
        handlers = derive_handlers(handler_list)
        handlers.each do |handler|
          @in_progress[:events] += 1
          filter_event(handler, event) do |event|
            mutate_event(handler, event) do |event_data|
              handle_event(handler, event_data)
            end
          end
        end
      end

      # Run event bridge extensions, within the Sensu EventMachine
      # reactor (event loop). The extension API `safe_run()` method is
      # used to guard against most errors. Bridges are for relaying
      # Sensu event data to other services.
      #
      # @param event [Hash]
      def event_bridges(event)
        @extensions[:bridges].each do |name, bridge|
          bridge.safe_run(event) do |output, status|
            @logger.debug("bridge extension output", {
              :extension => bridge.definition,
              :output => output
            })
          end
        end
      end

      # Add a check result to an aggregate. A check aggregate uses the
      # check `:name` and the `:issued` timestamp as its unique
      # identifier. An aggregate uses several counters: the total
      # number of results in the aggregate, and a counter for each
      # check severity (ok, warning, etc). Check output is also
      # stored, to be summarized to aid in identifying outliers for a
      # check execution across a number of Sensu clients. JSON
      # serialization is used for storing check result data.
      #
      # @param client [Hash]
      # @param check [Hash]
      def aggregate_check_result(client, check)
        @logger.debug("adding check result to aggregate", {
          :client => client,
          :check => check
        })
        result_set = "#{check[:name]}:#{check[:issued]}"
        result_data = Sensu::JSON.dump(:output => check[:output], :status => check[:status])
        @redis.multi
        @redis.hset("aggregation:#{result_set}", client[:name], result_data)
        SEVERITIES.each do |severity|
          @redis.hsetnx("aggregate:#{result_set}", severity, 0)
        end
        severity = (SEVERITIES[check[:status]] || "unknown")
        @redis.hincrby("aggregate:#{result_set}", severity, 1)
        @redis.hincrby("aggregate:#{result_set}", "total", 1)
        @redis.sadd("aggregates:#{check[:name]}", check[:issued])
        @redis.sadd("aggregates", check[:name])
        @redis.exec
      end

      # Truncate check output. For metric checks, (`"type":
      # "metric"`), check output is truncated to a single line and a
      # maximum of 255 characters. Check output is currently left
      # unmodified for standard checks.
      #
      # @param check [Hash]
      # @return [Hash] check with truncated output.
      def truncate_check_output(check)
        case check[:type]
        when METRIC_CHECK_TYPE
          output_lines = check[:output].split("\n")
          output = output_lines.first || check[:output]
          if output_lines.length > 1 || output.length > 255
            output = output[0..255] + "\n..."
          end
          check.merge(:output => output)
        else
          check
        end
      end

      # Store check result data. This method stores check result data
      # and the 21 most recent check result statuses for a client/check
      # pair, this history is used for event context and flap detection.
      # The check execution timestamp is also stored, to provide an
      # indication of how recent the data is. Check output is
      # truncated by `truncate_check_output()` before it is stored.
      #
      # @param client [Hash]
      # @param check [Hash]
      # @yield [] callback/block called after the check data has been
      # stored (history, etc).
      def store_check_result(client, check)
        @logger.debug("storing check result", :check => check)
        result_key = "#{client[:name]}:#{check[:name]}"
        history_key = "history:#{result_key}"
        check_truncated = truncate_check_output(check)
        @redis.multi
        @redis.sadd("result:#{client[:name]}", check[:name])
        @redis.set("result:#{result_key}", Sensu::JSON.dump(check_truncated))
        @redis.rpush(history_key, check[:status])
        @redis.ltrim(history_key, -21, -1)
        @redis.exec do
          yield
        end
      end

      # Fetch the execution history for a client/check pair, the 21
      # most recent check result statuses. This method also calculates
      # the total state change percentage for the history, this value
      # is use for check state flap detection, using a similar
      # algorithm to Nagios:
      # http://nagios.sourceforge.net/docs/3_0/flapping.html
      #
      # @param client [Hash]
      # @param check [Hash]
      # @yield [history, total_state_change] callback/block to call
      #   with the check history and calculated total state change
      #   value.
      # @yieldparam history [Array] containing the last 21 check
      #   result exit status codes.
      # @yieldparam total_state_change [Float] percentage for the
      #   check history (exit status codes).
      def check_history(client, check)
        history_key = "history:#{client[:name]}:#{check[:name]}"
        @redis.lrange(history_key, -21, -1) do |history|
          total_state_change = 0
          unless history.length < 21
            state_changes = 0
            change_weight = 0.8
            previous_status = history.first
            history.each do |status|
              unless status == previous_status
                state_changes += change_weight
              end
              change_weight += 0.02
              previous_status = status
            end
            total_state_change = (state_changes.fdiv(20) * 100).to_i
          end
          yield(history, total_state_change)
        end
      end

      # Determine if a check state is flapping, rapidly changing
      # between an OK and non-OK state. Flap detection is only done
      # for checks that have defined low and hight flap detection
      # thresholds, `:low_flap_threshold` and `:high_flap_threshold`.
      # The `check_history()` method provides the check history and
      # more importantly the total state change precentage value that
      # is compared with the configured thresholds defined in the
      # check data. If a check hasn't been flapping, the
      # `:total_state_change` must be equal to or higher than the
      # `:high_flap_threshold` to be changed to flapping. If a check
      # has been flapping, the `:total_state_change` must be equal to
      # or lower than the `:low_flap_threshold` to no longer be
      # flapping. This method uses the same algorithm as Nagios:
      # http://nagios.sourceforge.net/docs/3_0/flapping.html
      #
      # @param stored_event [Hash]
      # @param check [Hash]
      # @return [TrueClass, FalseClass]
      def check_flapping?(stored_event, check)
        if check.has_key?(:low_flap_threshold) && check.has_key?(:high_flap_threshold)
          was_flapping = stored_event && stored_event[:action] == EVENT_FLAPPING_ACTION
          if was_flapping
            check[:total_state_change] > check[:low_flap_threshold]
          else
            check[:total_state_change] >= check[:high_flap_threshold]
          end
        else
          false
        end
      end

      # Update the event registry, stored in Redis. This method
      # determines if check data results in the creation or update of
      # event data in the registry. Existing event data for a
      # client/check pair is fetched, used in conditionals and the
      # composition of the new event data. If a check `:status` is not
      # `0`, or it has been flapping, an event is created/updated in
      # the registry. If there was existing event data, but the check
      # `:status` is now `0`, the event is removed (resolved) from the
      # registry. If the previous conditions are not met, and check
      # `:type` is `metric` and the `:status` is `0`, the event
      # registry is not updated, but the provided callback is called
      # with the event data. If none of the previous conditions are
      # met, the `@in_progress[:check_results]` counter is decremented
      # by `1`. All event data is sent to event bridge extensions,
      # including events that do not normally produce an action. JSON
      # serialization is used when storing data in the registry.
      #
      # @param client [Hash]
      # @param check [Hash]
      # @yield callback [event] callback/block called with the
      #   resulting event data if the event registry is updated, or
      #   the check is of type `:metric`.
      # @yieldparam event [Hash]
      def update_event_registry(client, check)
        @redis.hget("events:#{client[:name]}", check[:name]) do |event_json|
          stored_event = event_json ? Sensu::JSON.load(event_json) : nil
          flapping = check_flapping?(stored_event, check)
          event = {
            :id => (stored_event ? stored_event[:id] : random_uuid),
            :client => client,
            :check => check,
            :occurrences => 1,
            :action => (flapping ? :flapping : :create),
            :timestamp => Time.now.to_i
          }
          if check[:status] != 0 || flapping
            if stored_event && check[:status] == stored_event[:check][:status]
              event[:occurrences] = stored_event[:occurrences] + 1
            end
            @redis.hset("events:#{client[:name]}", check[:name], Sensu::JSON.dump(event)) do
              yield(event)
            end
          elsif stored_event
            event[:occurrences] = stored_event[:occurrences]
            event[:action] = :resolve
            unless check[:auto_resolve] == false && !check[:force_resolve]
              @redis.hdel("events:#{client[:name]}", check[:name]) do
                yield(event)
              end
            end
          elsif check[:type] == METRIC_CHECK_TYPE
            yield(event)
          else
            @in_progress[:check_results] -= 1
          end
          event_bridges(event)
        end
      end

      # Create a blank client (data) and add it to the client
      # registry. Only the client name is known, the other client
      # attributes must be updated via the API (POST /clients:client).
      # Dynamically created clients and those updated via the API will
      # have client keepalives disabled, `:keepalives` is set to
      # `false`.
      #
      # @param name [Hash] to use for the client.
      # @yield [client] callback/block to be called with the
      #   dynamically created client data.
      # @yieldparam client [Hash]
      def create_client(name)
        client = {
          :name => name,
          :address => "unknown",
          :subscriptions => [],
          :keepalives => false,
          :version => VERSION
        }
        update_client_registry(client) do
          yield(client)
        end
      end

      # Retrieve a client (data) from Redis if it exists. If a client
      # does not already exist, create one (a blank) using the
      # `client_key` as the client name. Dynamically create client
      # data can be updated using the API (POST /clients/:client). If
      # a client does exist and it has a client signature, the check
      # result must have a matching signature or it is discarded.
      #
      # @param result [Hash] data.
      # @yield [client] callback/block to be called with client data,
      #   either retrieved from Redis, or dynamically created.
      # @yieldparam client [Hash]
      def retrieve_client(result)
        client_key = result[:check][:source] || result[:client]
        @redis.get("client:#{client_key}") do |client_json|
          unless client_json.nil?
            client = Sensu::JSON.load(client_json)
            if client[:signature]
              if client[:signature] == result[:signature]
                yield(client)
              else
                @logger.warn("invalid check result signature", {
                  :result => result,
                  :client => client
                })
                @logger.warn("not retrieving client from the registry", :result => result)
              end
            else
              yield(client)
            end
          else
            create_client(client_key) do |client|
              yield(client)
            end
          end
        end
      end

      # Process a check result, storing its data, inspecting its
      # contents, and taking the appropriate actions (eg. update the
      # event registry). The `@in_progress[:check_results]` counter is
      # incremented by `1` prior to check result processing and then
      # decremented by `1` after updating the event registry. A check
      # result must have a valid client name, associated with a client
      # in the registry or one will be created. If a local check
      # definition exists for the check name, and the check result is
      # not from a standalone check execution, it's merged with the
      # check result for more context.
      #
      # @param result [Hash] data.
      def process_check_result(result)
        @in_progress[:check_results] += 1
        @logger.debug("processing result", :result => result)
        retrieve_client(result) do |client|
          check = case
          when @settings.check_exists?(result[:check][:name]) && !result[:check][:standalone]
            @settings[:checks][result[:check][:name]].merge(result[:check])
          else
            result[:check]
          end
          check[:type] ||= "standard"
          check[:origin] = result[:client] if check[:source]
          aggregate_check_result(client, check) if check[:aggregate]
          store_check_result(client, check) do
            check_history(client, check) do |history, total_state_change|
              check[:history] = history
              check[:total_state_change] = total_state_change
              update_event_registry(client, check) do |event|
                process_event(event)
                @in_progress[:check_results] -= 1
              end
            end
          end
        end
      end

      # Set up the check result consumer. The consumer receives JSON
      # serialized check results from the transport, parses them, and<commit_msg>[type-standard-constant] use frozen constant for default check type "standard"<commit_after>require "sensu/daemon"
require "sensu/server/filter"
require "sensu/server/mutate"
require "sensu/server/handle"

module Sensu
  module Server
    class Process
      include Daemon
      include Filter
      include Mutate
      include Handle

      attr_reader :is_leader, :in_progress

      STANDARD_CHECK_TYPE = "standard".freeze

      METRIC_CHECK_TYPE = "metric".freeze

      EVENT_FLAPPING_ACTION = "flapping".freeze

      DEFAULT_HANDLER_NAME = "default".freeze

      # Create an instance of the Sensu server process, start the
      # server within the EventMachine event loop, and set up server
      # process signal traps (for stopping).
      #
      # @param options [Hash]
      def self.run(options={})
        server = self.new(options)
        EM::run do
          server.start
          server.setup_signal_traps
        end
      end

      # Override Daemon initialize() to support Sensu server leader
      # election and the handling event count.
      #
      # @param options [Hash]
      def initialize(options={})
        super
        @is_leader = false
        @timers[:leader] = Array.new
        @in_progress = Hash.new(0)
      end

      # Set up the Redis and Transport connection objects, `@redis`
      # and `@transport`. This method updates the Redis on error
      # callback to reset the in progress check result counter. This
      # method "drys" up many instances of `setup_redis()` and
      # `setup_transport()`, particularly in the specs.
      #
      # @yield callback/block called after connecting to Redis and the
      #   Sensu Transport.
      def setup_connections
        setup_redis do
          @redis.on_error do |error|
            @logger.error("redis connection error", :error => error.to_s)
            @in_progress[:check_results] = 0
          end
          setup_transport do
            yield
          end
        end
      end

      # Create a registration check definition for a client. Client
      # definitions may contain `:registration` configuration,
      # containing custom attributes and handler information. By
      # default, the registration check definition sets the `:handler`
      # to `registration`. If the client provides its own
      # `:registration` configuration, it's deep merged with the
      # defaults. The check `:name`, `:output`, `:status`, `:issued`,
      # and `:executed` values are always overridden to guard against
      # an invalid definition.
      def create_registration_check(client)
        check = {:handler => "registration"}
        if client.has_key?(:registration)
          check = deep_merge(check, client[:registration])
        end
        timestamp = Time.now.to_i
        overrides = {
          :name => "registration",
          :output => "new client registration",
          :status => 1,
          :issued => timestamp,
          :executed => timestamp
        }
        check.merge(overrides)
      end

      # Create and process a client registration event. A registration
      # event is created when a Sensu client is first added to the
      # client registry. The `create_registration_check()` method is
      # called to create a registration check definition for the
      # client.
      #
      # @param client [Hash] definition.
      def create_client_registration_event(client)
        event = {
          :id => random_uuid,
          :client => client,
          :check => create_registration_check(client),
          :occurrences => 1,
          :action => :create,
          :timestamp => Time.now.to_i
        }
        process_event(event)
      end

      # Process an initial client registration, when it is first added
      # to the client registry. If a registration handler is defined
      # or the client specifies one, a client registration event is
      # created and processed (handled, etc.) for the client
      # (`create_client_registration_event()`).
      #
      # @param client [Hash] definition.
      def process_client_registration(client)
        if @settings.handler_exists?("registration") || client[:registration]
          create_client_registration_event(client)
        end
      end

      # Update the Sensu client registry, stored in Redis. Sensu
      # client data is used to provide additional event context and
      # enable agent health monitoring. The client registry supports
      # client signatures, unique string identifiers used for
      # keepalive and result source verification. If a client has a
      # signature, all further registry updates for the client must
      # have the same signature. A client can begin to use a signature
      # if one was not previously configured. JSON serialization is
      # used for the stored client data.
      #
      # @param client [Hash]
      # @yield [success] passes success status to optional
      #   callback/block.
      # @yieldparam success [TrueClass,FalseClass] indicating if the
      #   client registry update was a success or the client data was
      #   discarded due to client signature mismatch.
      def update_client_registry(client)
        @logger.debug("updating client registry", :client => client)
        client_key = "client:#{client[:name]}"
        signature_key = "#{client_key}:signature"
        @redis.setnx(signature_key, client[:signature]) do |created|
          process_client_registration(client) if created
          @redis.get(signature_key) do |signature|
            if (signature.nil? || signature.empty?) && client[:signature]
              @redis.set(signature_key, client[:signature])
            end
            if signature.nil? || signature.empty? || (client[:signature] == signature)
              @redis.multi
              @redis.set(client_key, Sensu::JSON.dump(client))
              @redis.sadd("clients", client[:name])
              @redis.exec do
                yield(true) if block_given?
              end
            else
              @logger.warn("invalid client signature", {
                :client => client,
                :signature => signature
              })
              @logger.warn("not updating client in the registry", :client => client)
              yield(false) if block_given?
            end
          end
        end
      end

      # Set up the client keepalive consumer, keeping the Sensu client
      # registry updated. The consumer receives JSON serialized client
      # keepalives from the transport, parses them, and calls
      # `update_client_registry()` with the client data to update the
      # registry. Transport message acknowledgements are used to
      # ensure the client registry is updated successfully. Keepalive
      # JSON parsing errors are logged.
      def setup_keepalives
        @logger.debug("subscribing to keepalives")
        @transport.subscribe(:direct, "keepalives", "keepalives", :ack => true) do |message_info, message|
          @logger.debug("received keepalive", :message => message)
          begin
            client = Sensu::JSON.load(message)
            update_client_registry(client) do
              @transport.ack(message_info)
            end
          rescue Sensu::JSON::ParseError => error
            @logger.error("failed to parse keepalive payload", {
              :message => message,
              :error => error.to_s
            })
            @transport.ack(message_info)
          end
        end
      end

      # Expand event handler sets, creating an array of handler
      # definitions. Handler sets cannot be deeply nested (by choice),
      # this method will return `nil` if an attempt is made to deeply
      # nest. If the provided handler definition is not a set, it is
      # returned.
      #
      # @param handler [Hash] definition.
      # @param depth [Integer] of the expansion.
      # @return [Array, Hash, Nil]
      def expand_handler_sets(handler, depth=0)
        if handler[:type] == "set"
          if depth < 2
            derive_handlers(handler[:handlers], depth + 1)
          else
            @logger.error("handler sets cannot be deeply nested", :handler => handler)
            nil
          end
        else
          handler
        end
      end

      # Derive an array of handler definitions from a list of handler
      # names. This method first checks for the existence of standard
      # handlers, followed by handler extensions. If a handler does
      # not exist for a name, it is logged and ignored. Duplicate
      # handler definitions are removed.
      #
      # @param handler_list [Array]
      # @param depth [Integer] of handler set expansion.
      # @return [Array]
      def derive_handlers(handler_list, depth=0)
        handler_list.compact.map { |handler_name|
          case
          when @settings.handler_exists?(handler_name)
            handler = @settings[:handlers][handler_name].merge(:name => handler_name)
            expand_handler_sets(handler, depth)
          when @extensions.handler_exists?(handler_name)
            @extensions[:handlers][handler_name]
          else
            @logger.error("unknown handler", :handler_name => handler_name)
            nil
          end
        }.flatten.compact.uniq
      end

      # Process an event: filter -> mutate -> handle.
      #
      # This method determines the appropriate handlers for an event,
      # filtering and mutating the event data for each of them. The
      # `@in_progress[:events]` is incremented by `1`, for each event
      # handler chain (filter -> mutate -> handle).
      #
      # @param event [Hash]
      def process_event(event)
        log_level = event[:check][:type] == METRIC_CHECK_TYPE ? :debug : :info
        @logger.send(log_level, "processing event", :event => event)
        handler_list = Array((event[:check][:handlers] || event[:check][:handler]) || DEFAULT_HANDLER_NAME)
        handlers = derive_handlers(handler_list)
        handlers.each do |handler|
          @in_progress[:events] += 1
          filter_event(handler, event) do |event|
            mutate_event(handler, event) do |event_data|
              handle_event(handler, event_data)
            end
          end
        end
      end

      # Run event bridge extensions, within the Sensu EventMachine
      # reactor (event loop). The extension API `safe_run()` method is
      # used to guard against most errors. Bridges are for relaying
      # Sensu event data to other services.
      #
      # @param event [Hash]
      def event_bridges(event)
        @extensions[:bridges].each do |name, bridge|
          bridge.safe_run(event) do |output, status|
            @logger.debug("bridge extension output", {
              :extension => bridge.definition,
              :output => output
            })
          end
        end
      end

      # Add a check result to an aggregate. A check aggregate uses the
      # check `:name` and the `:issued` timestamp as its unique
      # identifier. An aggregate uses several counters: the total
      # number of results in the aggregate, and a counter for each
      # check severity (ok, warning, etc). Check output is also
      # stored, to be summarized to aid in identifying outliers for a
      # check execution across a number of Sensu clients. JSON
      # serialization is used for storing check result data.
      #
      # @param client [Hash]
      # @param check [Hash]
      def aggregate_check_result(client, check)
        @logger.debug("adding check result to aggregate", {
          :client => client,
          :check => check
        })
        result_set = "#{check[:name]}:#{check[:issued]}"
        result_data = Sensu::JSON.dump(:output => check[:output], :status => check[:status])
        @redis.multi
        @redis.hset("aggregation:#{result_set}", client[:name], result_data)
        SEVERITIES.each do |severity|
          @redis.hsetnx("aggregate:#{result_set}", severity, 0)
        end
        severity = (SEVERITIES[check[:status]] || "unknown")
        @redis.hincrby("aggregate:#{result_set}", severity, 1)
        @redis.hincrby("aggregate:#{result_set}", "total", 1)
        @redis.sadd("aggregates:#{check[:name]}", check[:issued])
        @redis.sadd("aggregates", check[:name])
        @redis.exec
      end

      # Truncate check output. For metric checks, (`"type":
      # "metric"`), check output is truncated to a single line and a
      # maximum of 255 characters. Check output is currently left
      # unmodified for standard checks.
      #
      # @param check [Hash]
      # @return [Hash] check with truncated output.
      def truncate_check_output(check)
        case check[:type]
        when METRIC_CHECK_TYPE
          output_lines = check[:output].split("\n")
          output = output_lines.first || check[:output]
          if output_lines.length > 1 || output.length > 255
            output = output[0..255] + "\n..."
          end
          check.merge(:output => output)
        else
          check
        end
      end

      # Store check result data. This method stores check result data
      # and the 21 most recent check result statuses for a client/check
      # pair, this history is used for event context and flap detection.
      # The check execution timestamp is also stored, to provide an
      # indication of how recent the data is. Check output is
      # truncated by `truncate_check_output()` before it is stored.
      #
      # @param client [Hash]
      # @param check [Hash]
      # @yield [] callback/block called after the check data has been
      # stored (history, etc).
      def store_check_result(client, check)
        @logger.debug("storing check result", :check => check)
        result_key = "#{client[:name]}:#{check[:name]}"
        history_key = "history:#{result_key}"
        check_truncated = truncate_check_output(check)
        @redis.multi
        @redis.sadd("result:#{client[:name]}", check[:name])
        @redis.set("result:#{result_key}", Sensu::JSON.dump(check_truncated))
        @redis.rpush(history_key, check[:status])
        @redis.ltrim(history_key, -21, -1)
        @redis.exec do
          yield
        end
      end

      # Fetch the execution history for a client/check pair, the 21
      # most recent check result statuses. This method also calculates
      # the total state change percentage for the history, this value
      # is use for check state flap detection, using a similar
      # algorithm to Nagios:
      # http://nagios.sourceforge.net/docs/3_0/flapping.html
      #
      # @param client [Hash]
      # @param check [Hash]
      # @yield [history, total_state_change] callback/block to call
      #   with the check history and calculated total state change
      #   value.
      # @yieldparam history [Array] containing the last 21 check
      #   result exit status codes.
      # @yieldparam total_state_change [Float] percentage for the
      #   check history (exit status codes).
      def check_history(client, check)
        history_key = "history:#{client[:name]}:#{check[:name]}"
        @redis.lrange(history_key, -21, -1) do |history|
          total_state_change = 0
          unless history.length < 21
            state_changes = 0
            change_weight = 0.8
            previous_status = history.first
            history.each do |status|
              unless status == previous_status
                state_changes += change_weight
              end
              change_weight += 0.02
              previous_status = status
            end
            total_state_change = (state_changes.fdiv(20) * 100).to_i
          end
          yield(history, total_state_change)
        end
      end

      # Determine if a check state is flapping, rapidly changing
      # between an OK and non-OK state. Flap detection is only done
      # for checks that have defined low and hight flap detection
      # thresholds, `:low_flap_threshold` and `:high_flap_threshold`.
      # The `check_history()` method provides the check history and
      # more importantly the total state change precentage value that
      # is compared with the configured thresholds defined in the
      # check data. If a check hasn't been flapping, the
      # `:total_state_change` must be equal to or higher than the
      # `:high_flap_threshold` to be changed to flapping. If a check
      # has been flapping, the `:total_state_change` must be equal to
      # or lower than the `:low_flap_threshold` to no longer be
      # flapping. This method uses the same algorithm as Nagios:
      # http://nagios.sourceforge.net/docs/3_0/flapping.html
      #
      # @param stored_event [Hash]
      # @param check [Hash]
      # @return [TrueClass, FalseClass]
      def check_flapping?(stored_event, check)
        if check.has_key?(:low_flap_threshold) && check.has_key?(:high_flap_threshold)
          was_flapping = stored_event && stored_event[:action] == EVENT_FLAPPING_ACTION
          if was_flapping
            check[:total_state_change] > check[:low_flap_threshold]
          else
            check[:total_state_change] >= check[:high_flap_threshold]
          end
        else
          false
        end
      end

      # Update the event registry, stored in Redis. This method
      # determines if check data results in the creation or update of
      # event data in the registry. Existing event data for a
      # client/check pair is fetched, used in conditionals and the
      # composition of the new event data. If a check `:status` is not
      # `0`, or it has been flapping, an event is created/updated in
      # the registry. If there was existing event data, but the check
      # `:status` is now `0`, the event is removed (resolved) from the
      # registry. If the previous conditions are not met, and check
      # `:type` is `metric` and the `:status` is `0`, the event
      # registry is not updated, but the provided callback is called
      # with the event data. If none of the previous conditions are
      # met, the `@in_progress[:check_results]` counter is decremented
      # by `1`. All event data is sent to event bridge extensions,
      # including events that do not normally produce an action. JSON
      # serialization is used when storing data in the registry.
      #
      # @param client [Hash]
      # @param check [Hash]
      # @yield callback [event] callback/block called with the
      #   resulting event data if the event registry is updated, or
      #   the check is of type `:metric`.
      # @yieldparam event [Hash]
      def update_event_registry(client, check)
        @redis.hget("events:#{client[:name]}", check[:name]) do |event_json|
          stored_event = event_json ? Sensu::JSON.load(event_json) : nil
          flapping = check_flapping?(stored_event, check)
          event = {
            :id => (stored_event ? stored_event[:id] : random_uuid),
            :client => client,
            :check => check,
            :occurrences => 1,
            :action => (flapping ? :flapping : :create),
            :timestamp => Time.now.to_i
          }
          if check[:status] != 0 || flapping
            if stored_event && check[:status] == stored_event[:check][:status]
              event[:occurrences] = stored_event[:occurrences] + 1
            end
            @redis.hset("events:#{client[:name]}", check[:name], Sensu::JSON.dump(event)) do
              yield(event)
            end
          elsif stored_event
            event[:occurrences] = stored_event[:occurrences]
            event[:action] = :resolve
            unless check[:auto_resolve] == false && !check[:force_resolve]
              @redis.hdel("events:#{client[:name]}", check[:name]) do
                yield(event)
              end
            end
          elsif check[:type] == METRIC_CHECK_TYPE
            yield(event)
          else
            @in_progress[:check_results] -= 1
          end
          event_bridges(event)
        end
      end

      # Create a blank client (data) and add it to the client
      # registry. Only the client name is known, the other client
      # attributes must be updated via the API (POST /clients:client).
      # Dynamically created clients and those updated via the API will
      # have client keepalives disabled, `:keepalives` is set to
      # `false`.
      #
      # @param name [Hash] to use for the client.
      # @yield [client] callback/block to be called with the
      #   dynamically created client data.
      # @yieldparam client [Hash]
      def create_client(name)
        client = {
          :name => name,
          :address => "unknown",
          :subscriptions => [],
          :keepalives => false,
          :version => VERSION
        }
        update_client_registry(client) do
          yield(client)
        end
      end

      # Retrieve a client (data) from Redis if it exists. If a client
      # does not already exist, create one (a blank) using the
      # `client_key` as the client name. Dynamically create client
      # data can be updated using the API (POST /clients/:client). If
      # a client does exist and it has a client signature, the check
      # result must have a matching signature or it is discarded.
      #
      # @param result [Hash] data.
      # @yield [client] callback/block to be called with client data,
      #   either retrieved from Redis, or dynamically created.
      # @yieldparam client [Hash]
      def retrieve_client(result)
        client_key = result[:check][:source] || result[:client]
        @redis.get("client:#{client_key}") do |client_json|
          unless client_json.nil?
            client = Sensu::JSON.load(client_json)
            if client[:signature]
              if client[:signature] == result[:signature]
                yield(client)
              else
                @logger.warn("invalid check result signature", {
                  :result => result,
                  :client => client
                })
                @logger.warn("not retrieving client from the registry", :result => result)
              end
            else
              yield(client)
            end
          else
            create_client(client_key) do |client|
              yield(client)
            end
          end
        end
      end

      # Process a check result, storing its data, inspecting its
      # contents, and taking the appropriate actions (eg. update the
      # event registry). The `@in_progress[:check_results]` counter is
      # incremented by `1` prior to check result processing and then
      # decremented by `1` after updating the event registry. A check
      # result must have a valid client name, associated with a client
      # in the registry or one will be created. If a local check
      # definition exists for the check name, and the check result is
      # not from a standalone check execution, it's merged with the
      # check result for more context.
      #
      # @param result [Hash] data.
      def process_check_result(result)
        @in_progress[:check_results] += 1
        @logger.debug("processing result", :result => result)
        retrieve_client(result) do |client|
          check = case
          when @settings.check_exists?(result[:check][:name]) && !result[:check][:standalone]
            @settings[:checks][result[:check][:name]].merge(result[:check])
          else
            result[:check]
          end
          check[:type] ||= STANDARD_CHECK_TYPE
          check[:origin] = result[:client] if check[:source]
          aggregate_check_result(client, check) if check[:aggregate]
          store_check_result(client, check) do
            check_history(client, check) do |history, total_state_change|
              check[:history] = history
              check[:total_state_change] = total_state_change
              update_event_registry(client, check) do |event|
                process_event(event)
                @in_progress[:check_results] -= 1
              end
            end
          end
        end
      end

      # Set up the check result consumer. The consumer receives JSON
      # serialized check results from the transport, parses them, and
{'fraction_non_alphanumeric': 0.08913407301803757, 'fraction_numerical': 0.0027273290770470464, 'mean_word_length': 2.22623650179976, 'pattern_counts': {'":': 2, '<': 7, '<?xml version=': 0, '>': 117, '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': '2732107', 'n_tokens_mistral': 12961, 'n_tokens_neox': 12567, 'n_words': 4379}