From c818bd8540a090ad8e5f9147641558dedc6d34ca Mon Sep 17 00:00:00 2001 From: Jayamala Date: Mon, 6 May 2024 14:17:16 -0400 Subject: [PATCH] refractored the metrics --- lib/jobs.rb | 151 +++++++++++++++++++++++----------------------- run_dark_blue.rb | 6 +- test/test_jobs.rb | 64 ++++++++------------ 3 files changed, 101 insertions(+), 120 deletions(-) diff --git a/lib/jobs.rb b/lib/jobs.rb index 6b41e12..e18da2c 100644 --- a/lib/jobs.rb +++ b/lib/jobs.rb @@ -7,97 +7,97 @@ require_relative "bag_status" module Jobs class DarkBlueMetrics - - # for being unable to contact the push gateway class PushGatewayClientError < StandardError; end + def initialize(start_time:, end_time:) + @start_time = start_time + @end_time = end_time + end - def get_overall_metrics_current_run(start_time) - successful_transfer_status = [ + def get_success_count + successful_status = [ BagStatus::BAGGING, BagStatus::COPYING, BagStatus::COPIED, BagStatus::VALIDATING,BagStatus::VALIDATED, - BagStatus::BAGGED, BagStatus::PACKING,BagStatus::PACKED, BagStatus::DEPOSITING, BagStatus::DEPOSITED - ] - - # failed_transfer_status = [ - # BagStatus::FAILED, BagStatus::DEPOSIT_SKIPPED, BagStatus::VALIDATION_SKIPPED, BagStatus::VERIFY_FAILED - # ] - result = DatabaseSchema::StatusEvent + BagStatus::BAGGED, BagStatus::PACKING,BagStatus::PACKED, BagStatus::DEPOSITING, BagStatus::DEPOSITED] + success_count = DatabaseSchema::StatusEvent .join(:status, Sequel.qualify(:status, :id) => Sequel.qualify(:status_event, :status_id)) - .join(:bag, Sequel.qualify(:bag, :id) => Sequel.qualify(:status_event, :bag_id)) - .where { Sequel.qualify(:status_event,:timestamp) >= start_time } - .select(:timestamp, Sequel.qualify(:status, :name).as(:status_name), Sequel.qualify(:bag, :identifier).as(:bag_id), - Sequel.as(Sequel.function(:COUNT, Sequel.case({{Sequel.qualify(:status, :id) => successful_transfer_status} => 1}, 0)), - :success_count)) - - successful_bag_transfer = [] - failed_bag_transfer = [] - - # successful transfer - result.each do |row| - if row[:success_count]==successful_transfer_status.length - successful_bag_transfer << row[:bag_id] - else - failed_bag_transfer << row[:bag_id] - end - end - bag_transfer_metrics = registry.gauge( - :bag_transfer_metrics, - docstring: "Bag transfer metrics", - labels: [:date_of_processing , :success, :failure] - ) - labels = { - date_of_processing: Time.at(start_time).strftime("%m/%d/%Y"), - success: successful_bag_transfer.count, - failure: failed_bag_transfer.count - } - - bag_transfer_metrics.set(start_time, labels: labels) - gateway.add(registry) + .where { Sequel.qualify(:status_event,:timestamp) >= @start_time } + .select( :bag_id, Sequel.qualify(:status, :name)) + .group(:bag_id) + .having{count(Sequel.qualify(:status, :name)) == successful_status.length} + .count + success_count + end - successful_bag_id = registry.gauge( - :successful_bag_id, - docstring: "Successful Bag transfer metrics", - labels: [:bag_id] - ) + def get_failure_count + failed_status = [BagStatus::DEPOSIT_SKIPPED, BagStatus::FAILED, BagStatus::VALIDATION_SKIPPED, BagStatus::VERIFY_FAILED] + failure_count = DatabaseSchema::StatusEvent + .join(:status, Sequel.qualify(:status, :id) => Sequel.qualify(:status_event, :status_id)) + .where { Sequel.qualify(:status_event,:timestamp) >= @start_time } + .select( :bag_id, Sequel.qualify(:status, :name)) + .group(:bag_id) + .having(~Sequel.|(*failed_status.map {|status| Sequel.qualify(:status, :name) => status})) + .count + failure_count + end - labels = { - bag_id: successful_bag_transfer - } + def get_failed_bag_ids + failed_status = [BagStatus::DEPOSIT_SKIPPED, BagStatus::FAILED, BagStatus::VALIDATION_SKIPPED, BagStatus::VERIFY_FAILED] + failure_bag_ids = DatabaseSchema::Bag + .join(:status_event, Sequel.qualify(:status_event, :bag_id) => Sequel.qualify(:bag, :id)) + .join(:status, Sequel.qualify(:status, :id) => Sequel.qualify(:status_event, :status_id)) + .where(Sequel.qualify(:status_event, :timestamp) >= @start_time) + .where(Sequel.qualify(:status, :name).like("%#{failed_status.join('%')}%")) + .group(Sequel[:bag][:identifier]) + .select_map(Sequel[:bag][:identifier]) - successful_bag_id.set(start_time, labels: labels) - gateway.add(registry) + failure_bag_ids + end - failed_bag_id = registry.gauge( - :failed_bag_id, - docstring: "Failed Bag transfer metrics", - labels: [:bag_id] - ) + def set_success_count + dark_blue_success_count = registry.gauge( + :dark_blue_success_count, + docstring: "Successful number of bag transfer") + dark_blue_success_count.set(get_success_count) + end - labels = { - bag_id: failed_bag_transfer - } + def set_failed_count + dark_blue_failed_count = registry.gauge( + :dark_blue_failed_count, + docstring: "Failed number of bag transfer") + dark_blue_failed_count.set(get_failure_count) + end - failed_bag_id.set(start_time, labels: labels) - gateway.add(registry) + def set_failed_bag_id + dark_blue_failed_bag_ids = registry.gauge( + :dark_blue_failed_bag_ids, + docstring: "Failed bag transfer") + if !get_failed_bag_ids.empty? + dark_blue_failed_bag_ids.set(get_failed_bag_ids) + end end - def push_last_successful_run + def set_last_successful_run dark_blue_last_successful_run = registry.gauge( :dark_blue_last_successful_run, - {docstring: "Timestamp of the last successful run of the cron job"} - ) - time_in_milli_sec = (Time.now.to_i) * 1000 + docstring: "Timestamp of the last successful run of the cron job") + time_in_milli_sec = @start_time * 1000 dark_blue_last_successful_run.set(time_in_milli_sec) - gateway.add(registry) end - def push_processing_duration(start_time, end_time) + def set_processing_duration dark_blue_processing_duration = registry.gauge( :dark_blue_processing_duration, - {docstring: "Duration of processing in seconds for the cron job"} - ) - dark_blue_processing_duration.set(end_time - start_time) - gateway.add(registry) + docstring: "Duration of processing in seconds for the cron job") + dark_blue_processing_duration.set(@end_time - @start_time) + end + + def set_all_metrics + set_last_successful_run + set_processing_duration + set_success_count + set_failed_count + set_failed_bag_id + push_metrics end private @@ -112,11 +112,8 @@ def gateway ) end + def push_metrics + gateway.add(registry) + end end - - # class DarkBlueMetricsFactory - # def self.for(use_db:) - # DarkBlueMetrics.new if use_db? - # end - # end end diff --git a/run_dark_blue.rb b/run_dark_blue.rb index 2816e23..732297c 100644 --- a/run_dark_blue.rb +++ b/run_dark_blue.rb @@ -197,7 +197,5 @@ def self.parse(options) dark_blue_job.process end end_time = Time.now.to_i -metrics = Jobs::DarkBlueMetrics.new -metrics.get_overall_metrics_current_run(start_time) -metrics.push_processing_duration(start_time,end_time) -metrics.push_last_successful_run +metrics = Jobs::DarkBlueMetrics.new(start_time: start_time, end_time: end_time) +metrics.set_all_metrics diff --git a/test/test_jobs.rb b/test/test_jobs.rb index eba3671..d4530e6 100644 --- a/test/test_jobs.rb +++ b/test/test_jobs.rb @@ -8,7 +8,8 @@ class DarkBlueMetricTest < Minitest::Test def setup - @metrics = Jobs::DarkBlueMetrics.new + @time_now = Time.now + @metrics = Jobs::DarkBlueMetrics.new(start_time: @time_now, end_time:(@time_now + 5)) ENV["PROMETHEUS_PUSH_GATEWAY"] = "http://test.xyz" @registry = Minitest::Mock.new @@ -26,35 +27,29 @@ def test_gateway assert_instance_of Prometheus::Client::Push, result end - def test_push_last_successful_run - @gauge.expect(:set, nil,[Numeric]) - @registry.expect(:gauge, @gauge,[:dark_blue_last_successful_run, - {docstring: "Timestamp of the last successful run of the cron job"} - ]) - @gateway.expect(:add, nil, [@registry]) - - @metrics.instance_variable_set(:@registry, @registry) - @metrics.instance_variable_set(:@gateway, @gateway) - @metrics.push_last_successful_run - @registry.verify - @gauge.verify - @gateway.verify - end - - def test_push_processing_duration - expected_duration = 5 - time_now = Time.now - @gauge.expect(:set, nil,[Numeric]) - @registry.expect(:gauge, @gauge,[:dark_blue_processing_duration,{docstring: "Duration of processing in seconds for the cron job"} ]) - @gateway.expect(:add, nil, [@registry]) - - @metrics.instance_variable_set(:@registry, @registry) - @metrics.instance_variable_set(:@gateway, @gateway) - @metrics.push_processing_duration(time_now,(time_now + expected_duration)) - @registry.verify - @gauge.verify - @gateway.verify - end + # def test_set_last_successful_run + # expected_time = @time_now * 1000 + # @gauge.expect(:set, nil,[expected_time]) + + # @metrics.stub(:create_gauge, @gauge) do + # Time.stub :now, Time.at(expected_time) do + # @metrics.set_last_successful_run + # end + # end + # @gauge.verify + # end + + # def test_set_processing_duration + # expected_duration = 5 + # @gauge.expect(:set, nil,[expected_duration]) + + # @metrics.stub(:create_gauge, @gauge) do + # Time.stub :now, Time.at(expected_duration) do + # @metrics.set_processing_duration + # end + # end + # @gauge.verify + # end # TBD: # sleep(30) @@ -90,12 +85,3 @@ def test_push_processing_duration # bag_transfer_metrics.set(start_time + (60 * 60 * 24 * 3) , labels: labels) # gateway.add(registry) end - -# class DarkBlueMetricsFactoryTest < Minitest::Test -# def test_for_creates_db_repo -# db = Sequel.connect("mock://mysql2") -# repo = Job::DarkBlueMetricsFactory.for(use_db: db) -# assert repo.is_a?(Job::DarkBlueMetrics) -# end - -# end