<commit_before>module Metric::CiMixin::Processing
  def perf_process(interval_name, start_time, end_time, counters_data)
    unless Metric::Capture::VALID_CAPTURE_INTERVALS.include?(interval_name)
      raise ArgumentError, _("invalid interval_name '%{name}'") % {:name => interval_name}
    end

    log_header = "[#{interval_name}]"

    interval_orig = interval_name
    interval_name = 'hourly' if interval_name == 'historical'

    resources = counters_data.keys

    _log.info("#{log_header} Processing for #{log_specific_targets(resources)}, for range [#{start_time} - #{end_time}]...")

    _dummy, t = Benchmark.realtime_block(:total_time) do
      # Take the raw metrics and create hashes out of them
      rt_rows = {}

      counters_data.each do |resource, data|
        counters       = data[:counters]
        counter_values = data[:counter_values]

        Benchmark.realtime_block(:process_counter_values) do
          counter_values.each do |ts, cv|
            ts = Metric::Helper.nearest_realtime_timestamp(ts) if interval_name == 'realtime'

            col_vals = {}
            cv.each do |counter_id, value|
              counter = counters[counter_id]
              next if counter.nil? || counter[:capture_interval_name] != interval_name

              col = counter[:counter_key].to_sym
              unless Metric.column_names_symbols.include?(col)
                _log.debug("#{log_header} Column [#{col}] is not defined, skipping")
                next
              end

              col_vals.store_path(col, counter[:instance], [value, counter])
            end

            col_vals.each do |col, values_by_instance|
              # If there are multiple instances for a column, use the aggregate
              #   instance, if available, otherwise roll it up ourselves.
              value, counter = values_by_instance[""]
              if value.nil?
                value = 0
                counter = nil
                values_by_instance.each_value do |v, c|
                  value += v
                  counter = c
                end
              end

              # Create hashes for the rows
              rt = (rt_rows[[ts, resource]] ||= {
                :capture_interval_name => interval_name,
                :capture_interval      => counter[:capture_interval],
                :resource              => resource,
                :resource_name         => resource.name,
                :timestamp             => ts
              })
              rt[col], message = normalize_value(value, counter)
              _log.warn("#{log_header} #{log_target} Timestamp: [#{ts}], Column [#{col}]: '#{message}'") if message
            end
          end
        end
      end

      parameters = if ActiveMetrics::Base.connection_config[:adapter] == "miq_postgres"
                     # We can just pass original data to PG, with metrics grouped by timestamps, since that is how
                     # we store to PG now. It will spare quite some memory and time to not convert it to row_per_metric
                     # and than back to original format.
                     transform_parameters_row_with_all_metrics(resources, interval_name, start_time, end_time, rt_rows)
                   else
                     transform_parameters_row_per_metric(resources, interval_name, start_time, end_time, rt_rows)
                   end

      ActiveMetrics::Base.connection.write_multiple(parameters)

      resources.each do |resource|
        resource.update_attribute(:last_perf_capture_on, end_time) if resource.last_perf_capture_on.nil? || resource.last_perf_capture_on.utc.iso8601 < end_time
      end

      # Raise <class>_perf_complete alert event if realtime so alerts can be evaluated.
      resources.each do |resource|
        MiqEvent.raise_evm_alert_event_queue(resource, MiqEvent.event_name_for_target(resource, "perf_complete"))
      end

      resources.each do |resource|
        resource.perf_rollup_to_parents(interval_orig, start_time, end_time)
      end

      publish_metrics(rt_rows) if syndicate_metrics?
    end
    _log.info("#{log_header} Processing for #{log_specific_targets(resources)}, for range [#{start_time} - #{end_time}]...Complete - Timings: #{t.inspect}")
  end

  private

  def transform_parameters_row_per_metric(_resources, interval_name, _start_time, _end_time, rt_rows)
    rt_rows.flat_map do |(ts, _resource), rt|
      rt.merge!(Metric::Processing.process_derived_columns(rt[:resource], rt, interval_name == 'realtime' ? Metric::Helper.nearest_hourly_timestamp(ts) : nil))
      rt.delete_nils<commit_msg>cut up metrics perf_process<commit_after>module Metric::CiMixin::Processing
  def perf_process(interval_name, start_time, end_time, counters_data)
    unless Metric::Capture::VALID_CAPTURE_INTERVALS.include?(interval_name)
      raise ArgumentError, _("invalid interval_name '%{name}'") % {:name => interval_name}
    end

    log_header = "[#{interval_name}]"

    interval_orig = interval_name
    interval_name = 'hourly' if interval_name == 'historical'

    resources = counters_data.keys

    _log.info("#{log_header} Processing for #{log_specific_targets(resources)}, for range [#{start_time} - #{end_time}]...")

    _dummy, t = Benchmark.realtime_block(:total_time) do
      # Take the raw metrics and create hashes out of them
      rt_rows = {}

      counters_data.each do |resource, data|
        process_counter_values(resource, data, interval_name, rt_rows) ## rt_rows changes
      end

      parameters = convert_metrics(resources, interval_name, start_time, end_time, rt_rows)
      write_metrics(parameters)
      update_and_publish_metrics(interval_orig, start_time, end_time, resources, rt_rows)
    end
    _log.info("#{log_header} Processing for #{log_specific_targets(resources)}, for range [#{start_time} - #{end_time}]...Complete - Timings: #{t.inspect}")
  end

  def process_counter_values(resource, data, interval_name, rt_rows)
    log_header = "[#{interval_name}]"

    counters       = data[:counters]
    counter_values = data[:counter_values]

    Benchmark.realtime_block(:process_counter_values) do
      counter_values.each do |ts, cv|
        ts = Metric::Helper.nearest_realtime_timestamp(ts) if interval_name == 'realtime'

        col_vals = {}
        cv.each do |counter_id, value|
          counter = counters[counter_id]
          next if counter.nil? || counter[:capture_interval_name] != interval_name

          col = counter[:counter_key].to_sym
          unless Metric.column_names_symbols.include?(col)
            _log.debug("#{log_header} Column [#{col}] is not defined, skipping")
            next
          end

          col_vals.store_path(col, counter[:instance], [value, counter])
        end

        col_vals.each do |col, values_by_instance|
          # If there are multiple instances for a column, use the aggregate
          #   instance, if available, otherwise roll it up ourselves.
          value, counter = values_by_instance[""]
          if value.nil?
            value = 0
            counter = nil
            values_by_instance.each_value do |v, c|
              value += v
              counter = c
            end
          end

          # Create hashes for the rows
          rt = (rt_rows[[ts, resource]] ||= {
            :capture_interval_name => interval_name,
            :capture_interval      => counter[:capture_interval],
            :resource              => resource,
            :resource_name         => resource.name,
            :timestamp             => ts
          })
          rt[col], message = normalize_value(value, counter)
          _log.warn("#{log_header} #{log_target} Timestamp: [#{ts}], Column [#{col}]: '#{message}'") if message
        end
      end
    end
  end

  def convert_metrics(resources, interval_name, start_time, end_time, rt_rows)
    if ActiveMetrics::Base.connection_config[:adapter] == "miq_postgres"
      # We can just pass original data to PG, with metrics grouped by timestamps, since that is how
      # we store to PG now. It will spare quite some memory and time to not convert it to row_per_metric
      # and than back to original format.
      transform_parameters_row_with_all_metrics(resources, interval_name, start_time, end_time, rt_rows)
    else
      transform_parameters_row_per_metric(resources, interval_name, start_time, end_time, rt_rows)
    end
  end

  def write_metrics(parameters)
    ActiveMetrics::Base.connection.write_multiple(parameters)
  end

  def update_and_publish_metrics(interval_orig, start_time, end_time, resources, rt_rows)
    resources.each do |resource|
      resource.update_attribute(:last_perf_capture_on, end_time) if resource.last_perf_capture_on.nil? || resource.last_perf_capture_on.utc.iso8601 < end_time
    end

    # Raise <class>_perf_complete alert event if realtime so alerts can be evaluated.
    resources.each do |resource|
      MiqEvent.raise_evm_alert_event_queue(resource, MiqEvent.event_name_for_target(resource, "perf_complete"))
    end

    resources.each do |resource|
      resource.perf_rollup_to_parents(interval_orig, start_time, end_time)
    end

    publish_metrics(rt_rows) if syndicate_metrics?
  end

  private

  def transform_parameters_row_per_metric(_resources, interval_name, _start_time, _end_time, rt_rows)
    rt_rows.flat_map do |(ts, _resource), rt|
      rt.merge!(Metric::Processing.process_derived_columns(rt[:resource], rt, interval_name == 'realtime' ? Metric::Helper.nearest_hourly_timestamp(ts) : nil))
      rt.delete_nils
{'fraction_non_alphanumeric': 0.09423097057899696, 'fraction_numerical': 0.0010470107842110773, 'mean_word_length': 2.7904761904761903, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 17, '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': '7646206', 'n_tokens_mistral': 2775, 'n_tokens_neox': 2682, 'n_words': 654}