Created result object for cleaner code

app/controllers/blazer/dashboards_controller.rb

@@ -37,9 +37,9 @@ module Blazer
         @data_sources.each do |data_source|
           query = data_source.smart_variables[var]
           if query
-            columns, rows, error, cached_at = data_source.run_statement(query)
-            ((@smart_vars[var] ||= []).concat(rows.map { |v| v.reverse })).uniq!
-            @sql_errors << error if error
+            result = data_source.run_statement(query)
+            ((@smart_vars[var] ||= []).concat(result.rows.map { |v| v.reverse })).uniq!
+            @sql_errors << result.error if result.error
           end
         end
       end

app/controllers/blazer/queries_controller.rb

@@ -54,9 +54,9 @@ module Blazer
       @bind_vars.each do |var|
         query = data_source.smart_variables[var]
         if query
-          columns, rows, error, cached_at = data_source.run_statement(query)
-          @smart_vars[var] = rows.map { |v| v.reverse }
-          @sql_errors << error if error
+          result = data_source.run_statement(query)
+          @smart_vars[var] = result.rows.map { |v| v.reverse }
+          @sql_errors << result.error if result.error
         end
       end
 
@@ -110,11 +110,10 @@ module Blazer
             break if result.any? || Time.now - wait_start > 3
           end
         else
-          result = @data_source.run_main_statement(@statement, options)
+          @result = @data_source.run_main_statement(@statement, options)
         end
 
-        if result.any?
-          @columns, @rows, @error, @cached_at, @just_cached = result
+        if @result
           @data_source.delete_results(@run_id) if @run_id
           render_run
         else
@@ -167,6 +166,12 @@ module Blazer
     end
 
     def render_run
+      @columns = @result.columns
+      @rows = @result.rows
+      @error = @result.error
+      @cached_at = @result.cached_at
+      @just_cached = @result.just_cached
+
       @checks = @query ? @query.checks : []
 
       @first_row = @rows.first || []
@@ -189,7 +194,7 @@ module Blazer
       @filename = @query.name.parameterize if @query
       @min_width_types = @columns.each_with_index.select { |c, i| @first_row[i].is_a?(Time) || @first_row[i].is_a?(String) || @data_source.smart_columns[c] }
 
-      @boom = Blazer.boom(@columns, @rows, @data_source)
+      @boom = @result.boom
 
       @linked_columns = @data_source.linked_columns
 

app/models/blazer/check.rb

@@ -15,7 +15,7 @@ module Blazer
       emails.to_s.downcase.split(",").map(&:strip)
     end
 
-    def update_state(columns, rows, error, data_source)
+    def update_state(result)
       check_type =
         if respond_to?(:check_type)
           self.check_type
@@ -25,17 +25,15 @@ module Blazer
           "bad_data"
         end
 
-      message = error
+      message = result.error
 
       self.state =
-        if error
-          if error == Blazer::TIMEOUT_MESSAGE
-            "timed out"
-          else
-            "error"
-          end
+        if result.timed_out?
+          "timed out"
+        elsif result.error
+          "error"
         elsif check_type == "anomaly"
-          anomaly, message = Blazer.detect_anomaly(columns, rows, data_source)
+          anomaly, message = result.detect_anomaly
           if anomaly.nil?
             "error"
           elsif anomaly
@@ -43,7 +41,7 @@ module Blazer
           else
             "passing"
           end
-        elsif rows.any?
+        elsif result.rows.any?
           check_type == "missing_data" ? "passing" : "failing"
         else
           check_type == "missing_data" ? "failing" : "passing"
@@ -53,7 +51,7 @@ module Blazer
       self.message = message if respond_to?(:message=)
 
       if respond_to?(:timeouts=)
-        if state == "timed out"
+        if result.timed_out?
           self.timeouts += 1
           self.state = "disabled" if timeouts >= 3
         else
@@ -63,7 +61,7 @@ module Blazer
 
       # do not notify on creation, except when not passing
       if (state_was || state != "passing") && state != state_was && emails.present?
-        Blazer::CheckMailer.state_change(self, state, state_was, rows.size, message).deliver_later
+        Blazer::CheckMailer.state_change(self, state, state_was, result.rows.size, message).deliver_later
       end
       save! if changed?
     end

app/views/blazer/queries/run.html.erb

@@ -38,8 +38,8 @@
   <% if @rows.any? %>
     <% values = @rows.first %>
     <% chart_id = SecureRandom.hex %>
-    <% column_types = Blazer.column_types(@columns, @rows, @boom) %>
-    <% chart_type = Blazer.chart_type(column_types) %>
+    <% column_types = @result.column_types %>
+    <% chart_type = @result.chart_type %>
     <% chart_options = {id: chart_id, min: nil} %>
     <% series_library = {} %>
     <% target_index = @columns.index { |k| k.downcase == "target" } %>
@@ -81,7 +81,7 @@
       <%= line_chart @rows.group_by { |r| v = r[1]; (@boom[@columns[1]] || {})[v.to_s] || v }.each_with_index.map { |(name, v), i| {name: name, data: v.map { |v2| [v2[0], v2[2]] }, library: series_library[i]} }, chart_options %>
     <% elsif chart_type == "bar" %>
       <%= column_chart (values.size - 1).times.map { |i| name = @columns[i + 1]; {name: name, data: @rows.first(20).map { |r| [(@boom[@columns[0]] || {})[r[0].to_s] || r[0], r[i + 1]] } } }, id: chart_id %>
-    <% elsif values.size == 3 && column_types == ["string", "string", "numeric"] %>
+    <% elsif chart_type == "bar2" %>
       <% first_20 = @rows.group_by { |r| r[0] }.values.first(20).flatten(1) %>
       <% labels = first_20.map { |r| r[0] }.uniq %>
       <% series = first_20.map { |r| r[1] }.uniq %>

lib/blazer.rb

@@ -1,10 +1,11 @@
 require "csv"
 require "yaml"
 require "chartkick"
+require "safely/core"
 require "blazer/version"
 require "blazer/data_source"
+require "blazer/result"
 require "blazer/engine"
-require "safely/core"
 
 module Blazer
   class Error < StandardError; end
@@ -90,12 +91,12 @@ module Blazer
       Blazer.transform_statement.call(data_source, statement) if Blazer.transform_statement
 
       while tries <= 3
-        columns, rows, error, cached_at = data_source.run_statement(statement, refresh_cache: true, check: check, query: check.query)
-        if error == Blazer::TIMEOUT_MESSAGE
+        result = data_source.run_statement(statement, refresh_cache: true, check: check, query: check.query)
+        if result.timed_out?
           Rails.logger.info "[blazer timeout] query=#{check.query.name}"
           tries += 1
           sleep(10)
-        elsif error.to_s.start_with?("PG::ConnectionBad")
+        elsif result.error.to_s.start_with?("PG::ConnectionBad")
           data_source.reconnect
           Rails.logger.info "[blazer reconnect] query=#{check.query.name}"
           tries += 1
@@ -104,15 +105,15 @@ module Blazer
           break
         end
       end
-      check.update_state(columns, rows, error, data_source)
+      check.update_state(result)
       # TODO use proper logfmt
-      Rails.logger.info "[blazer check] query=#{check.query.name} state=#{check.state} rows=#{rows.try(:size)} error=#{error}"
+      Rails.logger.info "[blazer check] query=#{check.query.name} state=#{check.state} rows=#{result.rows.try(:size)} error=#{result.error}"
 
       instrument[:statement] = statement
       instrument[:data_source] = data_source
       instrument[:state] = check.state
-      instrument[:rows] = rows.try(:size)
-      instrument[:error] = error
+      instrument[:rows] = result.rows.try(:size)
+      instrument[:error] = result.error
       instrument[:tries] = tries
     end
   end
@@ -129,123 +130,4 @@ module Blazer
       Blazer::CheckMailer.failing_checks(email, checks).deliver_later
     end
   end
-
-  def self.column_types(columns, rows, boom = {})
-    columns.each_with_index.map do |k, i|
-      v = (rows.find { |r| r[i] } || {})[i]
-      if boom[k]
-        "string"
-      elsif v.is_a?(Numeric)
-        "numeric"
-      elsif v.is_a?(Time) || v.is_a?(Date)
-        "time"
-      elsif v.nil?
-        nil
-      else
-        "string"
-      end
-    end
-  end
-
-  def self.chart_type(column_types)
-    if column_types.compact.size >= 2 && column_types.compact == ["time"] + (column_types.compact.size - 1).times.map { "numeric" }
-      "line"
-    elsif column_types == ["time", "string", "numeric"]
-      "line2"
-    elsif column_types.compact.size >= 2 && column_types == ["string"] + (column_types.compact.size - 1).times.map { "numeric" }
-      "bar"
-    end
-  end
-
-  def self.detect_anomaly(columns, rows, data_source)
-    anomaly = nil
-    message = nil
-
-    if rows.empty?
-      message = "No data"
-    else
-      boom = self.boom(columns, rows, data_source)
-      chart_type = self.chart_type(column_types(columns, rows, boom))
-      if chart_type == "line" || chart_type == "line2"
-        series = []
-
-        if chart_type == "line"
-          columns[1..-1].each_with_index.each do |k, i|
-            series << {name: k, data: rows.map{ |r| [r[0], r[i + 1]] }}
-          end
-        else
-          rows.group_by { |r| v = r[1]; (boom[columns[1]] || {})[v.to_s] || v }.each_with_index.map do |(name, v), i|
-            series << {name: name, data: v.map { |v2| [v2[0], v2[2]] }}
-          end
-        end
-
-        current_series = nil
-        begin
-          anomalies = []
-          series.each do |s|
-            current_series = s[:name]
-            anomalies << s[:name] if anomaly?(s[:data])
-          end
-          anomaly = anomalies.any?
-          if anomaly
-            if anomalies.size == 1
-              message = "Anomaly detected in #{anomalies.first}"
-            else
-              message = "Anomalies detected in #{anomalies.to_sentence}"
-            end
-          else
-            message = "No anomalies detected"
-          end
-        rescue => e
-          message = "#{current_series}: #{e.message}"
-        end
-      else
-        message = "Bad format"
-      end
-    end
-
-    [anomaly, message]
-  end
-
-  def self.anomaly?(series)
-    series = series.reject { |v| v[0].nil? }.sort_by { |v| v[0] }
-
-    csv_str =
-      CSV.generate do |csv|
-        csv << ["timestamp", "count"]
-        series.each do |row|
-          csv << row
-        end
-      end
-
-    timestamps = []
-    r_script = %x[which Rscript].chomp
-    raise "R not found" if r_script.empty?
-    output = %x[#{r_script} --vanilla #{File.expand_path("../blazer/detect_anomalies.R", __FILE__)} #{Shellwords.escape(csv_str)}]
-    if output.empty?
-      raise "Unknown R error"
-    end
-
-    rows = CSV.parse(output, headers: true)
-    error = rows.first && rows.first["x"]
-    raise error if error
-
-    rows.each do |row|
-      timestamps << Time.parse(row["timestamp"])
-    end
-    timestamps.include?(series.last[0].to_time)
-  end
-
-  def self.boom(columns, rows, data_source)
-    boom = {}
-    columns.each_with_index do |key, i|
-      query = data_source.smart_columns[key]
-      if query
-        values = rows.map { |r| r[i] }.compact.uniq
-        columns, rows2, error, cached_at = data_source.run_statement(ActiveRecord::Base.send(:sanitize_sql_array, [query.sub("{value}", "(?)"), values]))
-        boom[key] = Hash[rows2.map { |k, v| [k.to_s, v] }]
-      end
-    end
-    boom
-  end
 end

lib/blazer/data_source.rb

@@ -110,32 +110,34 @@ module Blazer
       end
 
       start_time = Time.now
-      columns, rows, error, cached_at, just_cached = run_statement(statement, options.merge(with_just_cached: true))
+      result = run_statement(statement, options.merge(with_just_cached: true))
       duration = Time.now - start_time
 
       if Blazer.audit
         audit.duration = duration if audit.respond_to?(:duration=)
-        audit.error = error if audit.respond_to?(:error=)
-        audit.timed_out = error == Blazer::TIMEOUT_MESSAGE if audit.respond_to?(:timed_out=)
-        audit.cached = cached_at.present? if audit.respond_to?(:cached=)
-        if !cached_at && duration >= 10
+        audit.error = result.error if audit.respond_to?(:error=)
+        audit.timed_out = result.timed_out? if audit.respond_to?(:timed_out=)
+        audit.cached = result.cached? if audit.respond_to?(:cached=)
+        if !result.cached? && duration >= 10
           audit.cost = cost(statement) if audit.respond_to?(:cost=)
         end
         audit.save! if audit.changed?
       end
 
-      if query && error != Blazer::TIMEOUT_MESSAGE
+      if query && !result.timed_out?
         query.checks.each do |check|
-          check.update_state(columns, rows, error, self)
+          check.update_state(result)
         end
       end
 
-      [columns, rows, error, cached_at, just_cached]
+      result
     end
 
     def read_cache(cache_key)
       value = Blazer.cache.read(cache_key)
-      Marshal.load(value) if value
+      if value
+        Blazer::Result.new(*Marshal.load(value))
+      end
     end
 
     def run_results(run_id)
@@ -155,7 +157,7 @@ module Blazer
       run_id = options[:run_id]
       cache_key = statement_cache_key(statement) if cache_mode != "off"
       if cache_mode != "off" && !options[:refresh_cache]
-        columns, rows, error, cached_at = read_cache(cache_key)
+        result = read_cache(cache_key)
       end
 
       unless rows
@@ -173,12 +175,10 @@ module Blazer
         if options[:check]
           comment << ",check_id:#{options[:check].id},check_emails:#{options[:check].emails}"
         end
-        columns, rows, error, just_cached = run_statement_helper(statement, comment, options[:run_id])
+        result = run_statement_helper(statement, comment, options[:run_id])
       end
 
-      output = [columns, rows, error, cached_at]
-      output << just_cached if options[:with_just_cached]
-      output
+      result
     end
 
     def clear_cache(statement)
@@ -203,8 +203,8 @@ module Blazer
     end
 
     def tables
-      columns, rows, error, cached_at = run_statement(connection_model.send(:sanitize_sql_array, ["SELECT table_name FROM information_schema.tables WHERE table_schema IN (?) ORDER BY table_name", schemas]))
-      rows.map(&:first)
+      result = run_statement(connection_model.send(:sanitize_sql_array, ["SELECT table_name FROM information_schema.tables WHERE table_schema IN (?) ORDER BY table_name", schemas]))
+      result.rows.map(&:first)
     end
 
     def postgresql?
@@ -279,7 +279,7 @@ module Blazer
         Blazer.cache.write(run_cache_key(run_id), cache_data, expires_in: 30.seconds)
       end
 
-      [columns, rows, error, cache && !cache_data.nil?]
+      Blazer::Result.new(self, columns, rows, error, nil, cache && !cache_data.nil?)
     end
 
     def adapter_name

lib/blazer/result.rb

@@ -0,0 +1,147 @@
+module Blazer
+  class Result
+    attr_reader :data_source, :columns, :rows, :error, :cached_at, :just_cached
+
+    def initialize(data_source, columns, rows, error, cached_at, just_cached)
+      @data_source = data_source
+      @columns = columns
+      @rows = rows
+      @error = error
+      @cached_at = cached_at
+      @just_cached = just_cached
+    end
+
+    def timed_out?
+      error == Blazer::TIMEOUT_MESSAGE
+    end
+
+    def cached?
+      cached_at.present?
+    end
+
+    def boom
+      @boom ||= begin
+        boom = {}
+        columns.each_with_index do |key, i|
+          query = data_source.smart_columns[key]
+          if query
+            values = rows.map { |r| r[i] }.compact.uniq
+            result = data_source.run_statement(ActiveRecord::Base.send(:sanitize_sql_array, [query.sub("{value}", "(?)"), values]))
+            boom[key] = Hash[result.rows.map { |k, v| [k.to_s, v] }]
+          end
+        end
+        boom
+      end
+    end
+
+    def column_types
+      @column_types ||= begin
+        columns.each_with_index.map do |k, i|
+          v = (rows.find { |r| r[i] } || {})[i]
+          if boom[k]
+            "string"
+          elsif v.is_a?(Numeric)
+            "numeric"
+          elsif v.is_a?(Time) || v.is_a?(Date)
+            "time"
+          elsif v.nil?
+            nil
+          else
+            "string"
+          end
+        end
+      end
+    end
+
+    def chart_type
+      @chart_type ||= begin
+        if column_types.compact.size >= 2 && column_types.compact == ["time"] + (column_types.compact.size - 1).times.map { "numeric" }
+          "line"
+        elsif column_types == ["time", "string", "numeric"]
+          "line2"
+        elsif column_types.compact.size >= 2 && column_types == ["string"] + (column_types.compact.size - 1).times.map { "numeric" }
+          "bar"
+        elsif column_types == ["string", "string", "numeric"]
+          "bar2"
+        end
+      end
+    end
+
+    def detect_anomaly
+      anomaly = nil
+      message = nil
+
+      if rows.empty?
+        message = "No data"
+      else
+        if chart_type == "line" || chart_type == "line2"
+          series = []
+
+          if chart_type == "line"
+            columns[1..-1].each_with_index.each do |k, i|
+              series << {name: k, data: rows.map{ |r| [r[0], r[i + 1]] }}
+            end
+          else
+            rows.group_by { |r| v = r[1]; (boom[columns[1]] || {})[v.to_s] || v }.each_with_index.map do |(name, v), i|
+              series << {name: name, data: v.map { |v2| [v2[0], v2[2]] }}
+            end
+          end
+
+          current_series = nil
+          begin
+            anomalies = []
+            series.each do |s|
+              current_series = s[:name]
+              anomalies << s[:name] if anomaly?(s[:data])
+            end
+            anomaly = anomalies.any?
+            if anomaly
+              if anomalies.size == 1
+                message = "Anomaly detected in #{anomalies.first}"
+              else
+                message = "Anomalies detected in #{anomalies.to_sentence}"
+              end
+            else
+              message = "No anomalies detected"
+            end
+          rescue => e
+            message = "#{current_series}: #{e.message}"
+          end
+        else
+          message = "Bad format"
+        end
+      end
+
+      [anomaly, message]
+    end
+
+    def anomaly?(series)
+      series = series.reject { |v| v[0].nil? }.sort_by { |v| v[0] }
+
+      csv_str =
+        CSV.generate do |csv|
+          csv << ["timestamp", "count"]
+          series.each do |row|
+            csv << row
+          end
+        end
+
+      timestamps = []
+      r_script = %x[which Rscript].chomp
+      raise "R not found" if r_script.empty?
+      output = %x[#{r_script} --vanilla #{File.expand_path("../blazer/detect_anomalies.R", __FILE__)} #{Shellwords.escape(csv_str)}]
+      if output.empty?
+        raise "Unknown R error"
+      end
+
+      rows = CSV.parse(output, headers: true)
+      error = rows.first && rows.first["x"]
+      raise error if error
+
+      rows.each do |row|
+        timestamps << Time.parse(row["timestamp"])
+      end
+      timestamps.include?(series.last[0].to_time)
+    end
+  end
+end