blazer.rb 4.2 KB

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