Add Prometheus metrics for DarkBlue Service

Gemfile

@@ -6,6 +6,7 @@ gem "canister", "~> 0.9.2"
 gem "faraday", "~> 2.9"
 gem "faraday-retry", "~> 2.2"
 gem "mysql2", "~> 0.5.6"
+gem "prometheus-client", "~> 4.2.2"
 gem "rexml", "~> 3.2.6"
 gem "semantic_logger", "~> 4.15"
 gem "sequel", "~> 5.77"

Gemfile.lock

@@ -45,6 +45,7 @@ GEM
     mysql2 (0.5.6)
     net-http (0.4.1)
       uri
+    prometheus-client (4.2.2)
     pastel (0.8.0)
       tty-color (~> 0.5)
     rake (13.1.0)
@@ -73,6 +74,7 @@ DEPENDENCIES
   faraday-retry (~> 2.2)
   minitest (~> 5.20)
   mysql2 (~> 0.5.6)
+  prometheus-client (~> 4.2.2)
   rake (~> 13.1)
   rexml (~> 3.2.6)
   semantic_logger (~> 4.15)

docker-compose.yml

@@ -39,5 +39,16 @@ services:
       interval: 5s
       timeout: 5s
       retries: 3   
+  prometheus:
+    image: prom/prometheus
+    ports:
+      - 9090:9090
+    volumes:
+      - ./prometheus.yml:/etc/prometheus/prometheus.yml
+
+  pushgateway:
+    image: prom/pushgateway
+    ports:
+      - 9091:9091   
 volumes:
   database:

example.env

@@ -63,3 +63,6 @@ APTRUST_REMOTE_SETTINGS_RESTORE_PATH=umich.edu
 APTRUST_REMOTE_SETTINGS_BUCKET_REGION=
 APTRUST_REMOTE_SETTINGS_AWS_ACCESS_KEY_ID=
 APTRUST_REMOTE_SETTINGS_AWS_SECRET_ACCESS_KEY=
+
+# Metrics push gateway url 
+PROMETHEUS_PUSH_GATEWAY=

lib/config.rb

@@ -202,13 +202,20 @@ def get_subset_by_key_stem(stem)
     keyword_init: true
   )
 
+  Metrics = Struct.new(
+    "MetricsConfig",
+    :push_gateway_url,
+    keyword_init: true
+  )
+
   Config = Struct.new(
     "Config",
     :settings,
     :database,
     :repository,
     :dark_blue,
     :aptrust,
+    :metrics,
     keyword_init: true
   )
 
@@ -310,7 +317,8 @@ def self.create_config(data)
             base_url: data.get_value(key: "APTRUST_API_BASE_URL")
           ),
           remote: create_remote_config(data.get_subset_by_key_stem("APTRUST_REMOTE_"))
-        )
+        ),
+        metrics: Metrics.new(push_gateway_url: data.get_value(key: "PROMETHEUS_PUSH_GATEWAY"))
       )
     end
 

lib/metrics_collector.rb

@@ -0,0 +1,121 @@
+require "time"
+
+require "prometheus/client"
+require "prometheus/client/push"
+require "prometheus/client/registry"
+
+require_relative "../services"
+require_relative "bag_status"
+
+module DarkBlueMetrics
+  class PushGatewayClientError < StandardError; end
+
+  class Timer
+    def self.time_processing
+      start_time = Time.now.to_i
+      yield
+      end_time = Time.now.to_i
+      [start_time, end_time]
+    end
+  end
+
+  class MetricsProvider
+    def initialize(start_time:, end_time:, status_event_repo:, push_gateway_url:)
+      @start_time = start_time
+      @end_time = end_time
+      @status_event_repo = status_event_repo
+      @push_gateway_url = push_gateway_url
+    end
+
+    def get_latest_bag_events_by_time
+      st_time = Time.at(@start_time)
+      @status_event_repo.get_latest_event_for_bags(start_time: st_time)
+    end
+
+    def get_success_count(events_by_time)
+      success_count = events_by_time.select { |e| e.status == BagStatus::DEPOSITED }
+      success_count.count
+    end
+
+    def get_failure_count(events_by_time)
+      failure_count = events_by_time.select { |e| e.status == BagStatus::FAILED }
+      failure_count.count
+    end
+
+    def get_failed_bag_ids(events_by_time)
+      events_by_time.select { |e| e.status == BagStatus::FAILED }
+    end
+
+    def set_success_count(events_by_time)
+      dark_blue_success_count = registry.gauge(
+        :dark_blue_success_count,
+        docstring: "Successful number of bag transfer"
+      )
+      dark_blue_success_count.set(get_success_count(events_by_time))
+    end
+
+    def set_failed_count(events_by_time)
+      dark_blue_failed_count = registry.gauge(
+        :dark_blue_failed_count,
+        docstring: "Failed number of bag transfer"
+      )
+      dark_blue_failed_count.set(get_failure_count(events_by_time))
+    end
+
+    def set_failed_bag_id(events_by_time)
+      dark_blue_failed_bag_ids = registry.counter(
+        :dark_blue_failed_bag_ids,
+        docstring: "Failed bag transfer",
+        labels: [:failed_id]
+      )
+      get_failed_ids = get_failed_bag_ids(events_by_time)
+      get_failed_ids.each do |e|
+        dark_blue_failed_bag_ids.increment(labels: {failed_id: e.bag_identifier})
+      end
+    end
+
+    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")
+      return unless dark_blue_last_successful_run
+      # converting starttime to milliseconds to support converting epoch time to datetime
+      # https://github.com/grafana/grafana/issues/6297
+      time_in_milli_sec = (@start_time * 1000)
+      dark_blue_last_successful_run.set(time_in_milli_sec)
+    end
+
+    def set_processing_duration
+      dark_blue_processing_duration = registry.gauge(:dark_blue_processing_duration,
+        docstring: "Duration of processing in seconds for the cron job")
+      return unless dark_blue_processing_duration
+      dark_blue_processing_duration.set(@end_time - @start_time)
+    end
+
+    def set_all_metrics
+      set_last_successful_run
+      set_processing_duration
+      latest_events = get_latest_bag_events_by_time
+      set_success_count(latest_events)
+      set_failed_count(latest_events)
+      set_failed_bag_id(latest_events)
+      push_metrics
+    end
+
+    private
+
+    def registry
+      @registry ||= Prometheus::Client::Registry.new
+    end
+
+    def gateway
+      @gateway ||= Prometheus::Client::Push.new(
+        job: "DarkBlueMetric",
+        gateway: @push_gateway_url
+      )
+    end
+
+    def push_metrics
+      gateway.add(registry)
+    end
+  end
+end

lib/status_event_repository.rb

@@ -87,6 +87,14 @@ def get_latest_event_for_bag(bag_identifier:)
         .sort_by(&:timestamp).reverse
       (events.length > 0) ? events[0] : nil
     end
+
+    def get_latest_event_for_bags(start_time:)
+      events_by_id = @status_events.group_by(&:bag_identifier)
+      events_by_id.map do |id, events_for_id|
+        event_by_time = events_for_id.select { |e| e.timestamp >= start_time }
+        event_by_time.max_by(&:timestamp) unless event_by_time.empty?
+      end.compact
+    end
   end
 
   class StatusEventDatabaseRepository
@@ -142,6 +150,14 @@ def get_all_for_bag_identifier(identifier)
         .map { |se| convert_to_struct(se) }
     end
 
+    # https://sequel.jeremyevans.net/rdoc/classes/Sequel/SQL/Window.html
+    def get_latest_event_for_bags(start_time:)
+      latest_events = base_query.where { timestamp >= start_time }
+        .select_append { row_number.function.over(partition: :bag_id, order: Sequel.desc(:timestamp)).as(:rn) }
+        .from_self.where(rn: 1)
+      latest_events.all.map { |se| convert_to_struct(se) }
+    end
+
     def get_latest_event_for_bag(bag_identifier:)
       event = base_query
         .where(bag: DatabaseSchema::Bag.where(identifier: bag_identifier))

prometheus.yml

@@ -0,0 +1,8 @@
+scrape_configs:
+- job_name: darkbluejob
+  honor_labels: true
+  honor_timestamps: true
+  scrape_interval: 50ms
+  metrics_path: /metrics
+  static_configs:
+    - targets: ['pushgateway:9091']

run_dark_blue.rb

@@ -7,12 +7,13 @@
 
 require_relative "lib/archivematica"
 require_relative "lib/bag_repository"
+require_relative "lib/bag_validator"
 require_relative "lib/data_transfer"
 require_relative "lib/dispatcher"
+require_relative "lib/metrics_collector"
 require_relative "lib/remote_client"
 require_relative "lib/repository_package_repository"
 require_relative "lib/status_event_repository"
-require_relative "lib/bag_validator"
 
 class DarkBlueError < StandardError
 end
@@ -35,7 +36,7 @@ def initialize(config)
         type: config.aptrust.remote.type,
         settings: config.aptrust.remote.settings
       ),
-      status_event_repo: StatusEventRepository::StatusEventRepositoryFactory.for(use_db: DB),
+      status_event_repo: S.status_event_repo,
       bag_repo: BagRepository::BagRepositoryFactory.for(use_db: DB)
     )
     @arch_configs = config.dark_blue.archivematicas
@@ -201,8 +202,14 @@ def self.parse(options)
 dark_blue_job = DarkBlueJob.new(config)
 
 options = DarkBlueParser.parse ARGV
-if options.packages.length > 0
-  dark_blue_job.redeliver_packages(options.packages)
-else
-  dark_blue_job.process
-end
+
+start_time, end_time = DarkBlueMetrics::Timer.time_processing {
+  if options.packages.length > 0
+    dark_blue_job.redeliver_packages(options.packages)
+  else
+    dark_blue_job.process
+  end
+}
+metrics = DarkBlueMetrics::MetricsProvider.new(start_time: start_time, end_time: end_time,
+  status_event_repo: S.status_event_repo, push_gateway_url: config.metrics.push_gateway_url)
+metrics.set_all_metrics

services.rb

@@ -44,3 +44,8 @@ def self.included(klass)
     password: db_config.password,
     fractional_seconds: true)
 end
+
+S.register(:status_event_repo) do
+  db = S.config.database && S.dbconnect
+  StatusEventRepository::StatusEventRepositoryFactory.for(use_db: db)
+end