it fixes metric's bugs

Author: GriwMF

lib/stackify/metrics/metric_aggregate.rb

@@ -2,7 +2,7 @@ module Stackify::Metrics
   class MetricAggregate
 
     attr_accessor :name, :category, :value, :count, :occurred_utc,
-                  :monitor_id, :metric_type, :name_key
+                  :monitor_id, :metric_type, :name_key, :sent, :is_increment
 
     def initialize metric
       @name = metric.name
@@ -12,6 +12,8 @@ def initialize metric
       @count = 0
       @occurred_utc = metric.get_rounded_time
       @name_key = metric.calc_name_key
+      @sent = false
+      @is_increment = metric.is_increment
     end
 
     def aggregate_key

lib/stackify/metrics/metrics.rb

@@ -67,6 +67,10 @@ def time category, metric_name, start_time, advanced_settings = nil
         avarage_time category, metric_name, time_taken, advanced_settings
       end
 
+      def time_duration category, metric_name, duration_time, advanced_settings = nil
+        avarage_time category, metric_name, duration_time, advanced_settings
+      end
+
       def avarage_time category, metric_name, elapsed_time, advanced_settings = nil
         m = Metric.new category, metric_name, Stackify::Metrics::METRIC_TYPES[:counter_time]
         m.value = elapsed_time.round #seconds

lib/stackify/metrics/metrics_client.rb

@@ -60,83 +60,41 @@ def start_upload_metrics
       #read everything up to the start of the current minute
       read_queued_metrics_batch current_time
       handle_zero_reports current_time
-      get_for_recent = @aggregate_metrics.select do |_k, v|
-        v.occurred_utc < current_time && v.occurred_utc > current_time - 5.minutes
-      end
-      set_latest_aggregates get_for_recent
+
       selected_aggr_metrics = @aggregate_metrics.select { |_key, aggr| aggr.occurred_utc < current_time }
       first_50_metrics = Hash[selected_aggr_metrics.to_a.take 50]
       if first_50_metrics.length > 0
         #only getting metrics less than 10 minutes old to drop old data in case we get backed up
         #they are removed from the @aggregated_metrics in the upload function upon success
-        all_success = upload_aggregates(first_50_metrics.select { |_key, aggr| aggr.occurred_utc > current_time - 10.minutes })
-        all_success.map {|key, _aggr| @aggregate_metrics[key].occurred_utc = current_time }
+        upload_aggregates(first_50_metrics.select { |_key, aggr| aggr.occurred_utc > current_time - 10.minutes })
       end
       @aggregate_metrics.delete_if { |_key, aggr| aggr.occurred_utc < purge_older_than }
     end
 
-    def read_queued_metrics_batch chosen_time
-      batches = {}
-
+    def read_queued_metrics_batch current_time
       while  @metrics_queue.size > 0 do
+        break if Stackify::Utils.rounded_current_time.to_i != current_time.to_i
         metric = @metrics_queue.pop
         metric.calc_and_set_aggregate_key
-        unless batches.has_key? metric.aggregate_key
-          name_key = metric.calc_name_key
-          if metric.is_increment && @last_aggregates.has_key?(name_key)
-            #if wanting to do increments we need to grab the last value so we know what to increment
-            metric.value = @last_aggregates[name_key].value
-          end
-          batches[metric.aggregate_key] = MetricAggregate.new metric
-          #if it is nil don't do anything
-          #we are doing it where the aggregates are created so we don't do it one very single metric,
-          #just once per batch to optimize performance
-
-          @metric_settings[name_key] = metric.settings if metric.settings != nil
+        metric_for_aggregation = MetricAggregate.new(metric)
+        name_key = metric.calc_name_key
+        if @last_aggregates.has_key?(name_key) &&
+            (metric.is_increment || @last_aggregates[name_key].occurred_utc.to_i == current_time.to_i)
+          metric_for_aggregation.value = @last_aggregates[name_key].value
+          metric_for_aggregation.count = @last_aggregates[name_key].count
         end
-        batches[metric.aggregate_key].count += 1
-        if metric.is_increment
-          #add or subtract
-          batches[metric.aggregate_key].value += metric.value
-        elsif metric.metric_type == Stackify::Metrics::METRIC_TYPES[:metric_last]
-          #should end up the last value
-          batches[metric.aggregate_key].value = metric.value
-        else
-          batches[metric.aggregate_key].value += metric.value
-        end
-        #we don't need anything more this recent so bail
-        break if metric.occurred > chosen_time
-      end
-      batches.each_pair do |_key, aggregated_metric|
-        aggregate aggregated_metric
-      end
-    end
 
-    def aggregate am
-      agg_key = am.aggregate_key
-      if @aggregate_metrics.has_key? agg_key
-        agg = @aggregate_metrics[agg_key]
-      else
-        if @aggregate_metrics.length > 1000
-          str = 'No longer aggregating new metrics because more than 1000 are queued'
-          Stackify.internal_log :warn, str
-          return
-        end
-        Stackify.internal_log :debug, 'Creating aggregate for ' + agg_key
-        @aggregate_metrics[agg_key] = am
-        agg = Stackify::Metrics::Metric.new am.category, am.name, am.metric_type
-        agg = MetricAggregate.new agg
-        agg.occurred_utc = am.occurred_utc
-      end
+        @metric_settings[name_key] = metric.settings if metric.settings != nil
 
-      if am.metric_type == Stackify::Metrics::METRIC_TYPES[:metric_last]
-        agg.count = 1
-        agg.value = am.value
-      else
-        agg.count += am.count
-        agg.value += am.value
+        if metric.metric_type == Stackify::Metrics::METRIC_TYPES[:metric_last] && !metric.is_increment
+          metric_for_aggregation.value = metric.value
+          metric_for_aggregation.count = 1
+        else
+          metric_for_aggregation.value += metric.value
+          metric_for_aggregation.count += 1
+        end
+        @last_aggregates[name_key] = @aggregate_metrics[metric.aggregate_key] = metric_for_aggregation
       end
-      @aggregate_metrics[agg_key]= agg
     end
 
     def submit_metrics_task
@@ -153,29 +111,31 @@ def handle_zero_reports current_time
             @metric_settings.delete[aggregate.name_key]
             next
           end
-          #agg = MetricAggregate.new aggregate.category, aggregate.name, aggregate.metric_type
-          agg = aggregate
+          agg = aggregate.dup
           agg.occurred_utc = current_time
-          case aggregate.metric_type
-          when Stackify::Metrics::METRIC_TYPES[:metric_last]
-            settings.autoreport_last_value_if_nothing_reported = false #do not allow this
-          when Stackify::Metrics::METRIC_TYPES[:counter]
+
+          disabled_autoreport_last = [
+              Stackify::Metrics::METRIC_TYPES[:counter],
+              Stackify::Metrics::METRIC_TYPES[:timer]
+          ]
+          if disabled_autoreport_last.include? aggregate.metric_type
             settings.autoreport_last_value_if_nothing_reported = false #do not allow this
           end
+
           if settings.autoreport_zero_if_nothing_reported
-            agg.count = 1
+            agg.count = 0
             agg.value = 0
           elsif settings.autoreport_last_value_if_nothing_reported
-            agg.count = aggregate.value.to_i
+            agg.count = aggregate.count.to_i
             agg.value = aggregate.value
           else
             next
           end
           agg_key = agg.aggregate_key
           unless @aggregate_metrics.has_key? agg_key
-            agg.occurred_utc = current_time - 60
             agg.name_key = aggregate.name_key
-            Stackify.internal_log :debug, 'Creating 0 default value for ' + agg_key
+            agg.sent = false
+            Stackify.internal_log :debug, 'Creating default value for ' + agg_key
             @aggregate_metrics[agg_key] = agg
           end
         end
@@ -194,11 +154,6 @@ def set_latest_aggregates aggregates
     end
 
     def upload_aggregates aggr_metrics
-      s = ''
-      aggr_metrics.each_pair do |_k, m|
-        s = s + m.inspect.to_s + "\n --------------------------- \n"
-      end
-      Stackify.internal_log :debug, "Uploading Aggregate Metrics at #{ Time.now }: \n" + s
       all_success = true
       aggr_metrics.each_pair do |_key, metric|
         if @monitor_ids.has_key? metric.name_key
@@ -230,8 +185,7 @@ def upload_aggregates aggr_metrics
 
         #get identified once
         aggr_metrics_for_upload = aggr_metrics.select { |_key, aggr_metric| !aggr_metric.monitor_id.nil? }
-        response = @metrics_sender.upload_metrics aggr_metrics_for_upload
-        Stackify.internal_log :info, 'Metrics are uploaded successfully' if response.try(:status) == 200
+        @metrics_sender.upload_metrics aggr_metrics_for_upload
         all_success
       end
     end

lib/stackify/metrics/metrics_sender.rb

@@ -13,14 +13,19 @@ def monitor_info aggr_metric
 
     def upload_metrics aggr_metrics
       return true if aggr_metrics.nil? || aggr_metrics.length == 0
+      current_time = Stackify::Utils.rounded_current_time
       if Stackify.authorized?
         records = []
         aggr_metrics.each_pair do |_key, metric|
+          next if metric.sent || metric.occurred_utc.to_i >= current_time.to_i
           records << Stackify::Metrics::MetricForSubmit.new(metric).to_h
-          prms = [metric.category, metric.name, metric.count, metric.value, metric.monitor_id ]
-          Stackify.internal_log :debug, 'Uploading metric: %s: %s count %s, value %s, ID %s' %  prms
+          metric.sent = true
+        end
+        if records.any?
+          Stackify.internal_log :debug, "Uploading Aggregate Metrics at #{ Time.now }: \n" + JSON.pretty_generate(records)
+          response = send_request SUBMIT_METRIS_URI, records.to_json
+          Stackify.internal_log :info, 'Metrics are uploaded successfully' if response.try(:status) == 200
         end
-        send_request SUBMIT_METRIS_URI, records.to_json
       else
         Stackify.log_internal_error "Uploading of metrics is failed because of authorization failure"
       end