blazer.rb 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138
  1. require "csv"
  2. require "yaml"
  3. require "chartkick"
  4. require "safely/core"
  5. require "blazer/version"
  6. require "blazer/data_source"
  7. require "blazer/result"
  8. require "blazer/run_statement"
  9. require "blazer/adapters/base_adapter"
  10. require "blazer/adapters/active_record_adapter"
  11. require "blazer/engine"
  12. module Blazer
  13. class Error < StandardError; end
  14. class TimeoutNotSupported < Error; end
  15. class << self
  16. attr_accessor :audit
  17. attr_reader :time_zone
  18. attr_accessor :user_name
  19. attr_accessor :user_class
  20. attr_accessor :user_method
  21. attr_accessor :before_action
  22. attr_accessor :from_email
  23. attr_accessor :cache
  24. attr_accessor :transform_statement
  25. attr_accessor :check_schedules
  26. attr_accessor :anomaly_checks
  27. attr_accessor :async
  28. attr_accessor :images
  29. end
  30. self.audit = true
  31. self.user_name = :name
  32. self.check_schedules = ["5 minutes", "1 hour", "1 day"]
  33. self.anomaly_checks = false
  34. self.async = false
  35. self.images = false
  36. TIMEOUT_MESSAGE = "Query timed out :("
  37. TIMEOUT_ERRORS = [
  38. "canceling statement due to statement timeout", # postgres
  39. "canceling statement due to conflict with recovery", # postgres
  40. "cancelled on user's request", # redshift
  41. "canceled on user's request", # redshift
  42. "system requested abort", # redshift
  43. "maximum statement execution time exceeded" # mysql
  44. ]
  45. BELONGS_TO_OPTIONAL = {}
  46. BELONGS_TO_OPTIONAL[:optional] = true if Rails::VERSION::MAJOR >= 5
  47. def self.time_zone=(time_zone)
  48. @time_zone = time_zone.is_a?(ActiveSupport::TimeZone) ? time_zone : ActiveSupport::TimeZone[time_zone.to_s]
  49. end
  50. def self.settings
  51. @settings ||= begin
  52. path = Rails.root.join("config", "blazer.yml").to_s
  53. if File.exist?(path)
  54. YAML.load(ERB.new(File.read(path)).result)
  55. else
  56. {}
  57. end
  58. end
  59. end
  60. def self.data_sources
  61. @data_sources ||= begin
  62. ds = Hash[
  63. settings["data_sources"].map do |id, s|
  64. [id, Blazer::DataSource.new(id, s)]
  65. end
  66. ]
  67. ds.default = ds.values.first
  68. ds
  69. end
  70. end
  71. def self.run_checks(schedule: nil)
  72. checks = Blazer::Check.includes(:query)
  73. checks = checks.where(schedule: schedule) if schedule
  74. checks.find_each do |check|
  75. next if check.state == "disabled"
  76. Safely.safely { run_check(check) }
  77. end
  78. end
  79. def self.run_check(check)
  80. rows = nil
  81. error = nil
  82. tries = 1
  83. ActiveSupport::Notifications.instrument("run_check.blazer", check_id: check.id, query_id: check.query.id, state_was: check.state) do |instrument|
  84. # try 3 times on timeout errors
  85. data_source = data_sources[check.query.data_source]
  86. statement = check.query.statement
  87. Blazer.transform_statement.call(data_source, statement) if Blazer.transform_statement
  88. while tries <= 3
  89. result = data_source.run_statement(statement, refresh_cache: true, check: check, query: check.query)
  90. if result.timed_out?
  91. Rails.logger.info "[blazer timeout] query=#{check.query.name}"
  92. tries += 1
  93. sleep(10)
  94. elsif result.error.to_s.start_with?("PG::ConnectionBad")
  95. data_source.reconnect
  96. Rails.logger.info "[blazer reconnect] query=#{check.query.name}"
  97. tries += 1
  98. sleep(10)
  99. else
  100. break
  101. end
  102. end
  103. check.update_state(result)
  104. # TODO use proper logfmt
  105. Rails.logger.info "[blazer check] query=#{check.query.name} state=#{check.state} rows=#{result.rows.try(:size)} error=#{result.error}"
  106. instrument[:statement] = statement
  107. instrument[:data_source] = data_source
  108. instrument[:state] = check.state
  109. instrument[:rows] = result.rows.try(:size)
  110. instrument[:error] = result.error
  111. instrument[:tries] = tries
  112. end
  113. end
  114. def self.send_failing_checks
  115. emails = {}
  116. Blazer::Check.includes(:query).where(state: ["failing", "error", "timed out", "disabled"]).find_each do |check|
  117. check.split_emails.each do |email|
  118. (emails[email] ||= []) << check
  119. end
  120. end
  121. emails.each do |email, checks|
  122. Blazer::CheckMailer.failing_checks(email, checks).deliver_later
  123. end
  124. end
  125. end