Merge pull request #290 from binaryseed/metric-merge-opt

Optimize metric collection with counters

lib/new_relic/harvest/collector/metric/harvester.ex

@@ -42,17 +42,8 @@ defmodule NewRelic.Harvest.Collector.Metric.Harvester do
 
   def handle_cast(_late_msg, :completed), do: {:noreply, :completed}
 
-  def handle_cast({:report, report_metrics}, state) do
-    metrics =
-      report_metrics
-      |> List.wrap()
-      |> Enum.reduce(state.metrics, fn %{name: name, scope: scope} = metric, acc ->
-        Map.update(acc, {name, scope}, metric, fn existing ->
-          NewRelic.Metric.merge(existing, metric)
-        end)
-      end)
-
-    {:noreply, %{state | metrics: metrics}}
+  def handle_cast({:report, metrics}, state) do
+    {:noreply, %{state | metrics: merge(metrics, state)}}
   end
 
   def handle_call(_late_msg, _from, :completed), do: {:reply, :completed, :completed}
@@ -66,6 +57,12 @@ defmodule NewRelic.Harvest.Collector.Metric.Harvester do
     {:reply, build_metric_data(state.metrics), state}
   end
 
+  def merge(metrics, state) do
+    metrics
+    |> List.wrap()
+    |> Enum.reduce(state.metrics, &merge_metric/2)
+  end
+
   def send_harvest(state) do
     metric_data = build_metric_data(state.metrics)
 
@@ -89,22 +86,72 @@ defmodule NewRelic.Harvest.Collector.Metric.Harvester do
   end
 
   defp build_metric_data(metrics) do
-    metrics
-    |> Map.values()
-    |> Enum.map(&encode/1)
+    Enum.map(metrics, &build/1)
   end
 
-  def encode(%NewRelic.Metric{name: name, scope: scope} = m) do
+  @size 6
+  @call_count 1
+  @total_call_time 2
+  @total_exclusive_time 3
+  @min_call_time 4
+  @max_call_time 5
+  @sum_of_squares 6
+
+  defp merge_metric(metric, metrics_acc) do
+    case Map.get(metrics_acc, {metric.name, metric.scope}) do
+      nil ->
+        counter = new(@size, [])
+
+        add(counter, @call_count, round(metric.call_count))
+        add(counter, @total_call_time, encode(metric.total_call_time))
+        add(counter, @total_exclusive_time, encode(metric.total_exclusive_time))
+        add(counter, @min_call_time, encode(metric.min_call_time))
+        add(counter, @max_call_time, encode(metric.max_call_time))
+        add(counter, @sum_of_squares, encode(metric.sum_of_squares))
+
+        Map.put(metrics_acc, {metric.name, metric.scope}, counter)
+
+      counter ->
+        add(counter, @call_count, round(metric.call_count))
+        add(counter, @total_call_time, encode(metric.total_call_time))
+        add(counter, @total_exclusive_time, encode(metric.total_exclusive_time))
+
+        if metric.min_call_time < decode(get(counter, @min_call_time)),
+          do: put(counter, @min_call_time, encode(metric.max_call_time))
+
+        if metric.max_call_time > decode(get(counter, @max_call_time)),
+          do: put(counter, @max_call_time, encode(metric.max_call_time))
+
+        add(counter, @sum_of_squares, encode(metric.sum_of_squares))
+
+        metrics_acc
+    end
+  end
+
+  defp build({{name, scope}, counter}) do
     [
       %{name: to_string(name), scope: to_string(scope)},
       [
-        m.call_count,
-        m.total_call_time,
-        m.total_exclusive_time,
-        m.min_call_time,
-        m.max_call_time,
-        m.sum_of_squares
+        get(counter, @call_count),
+        decode(get(counter, @total_call_time)),
+        decode(get(counter, @total_exclusive_time)),
+        decode(get(counter, @min_call_time)),
+        decode(get(counter, @max_call_time)),
+        decode(get(counter, @sum_of_squares))
       ]
     ]
   end
+
+  @compile {:inline, new: 2, add: 3, put: 3, get: 2}
+  defp new(size, opts), do: :counters.new(size, opts)
+  defp add(counter, index, value), do: :counters.add(counter, index, value)
+  defp put(counter, index, value), do: :counters.put(counter, index, value)
+  defp get(counter, index), do: :counters.get(counter, index)
+
+  # counters store integers, so we encode values
+  # into integers keeping 3 decimal places of precision
+  @precision 1_000
+  @compile {:inline, encode: 1, decode: 1}
+  defp encode(val), do: round(val * @precision)
+  defp decode(val), do: val / @precision
 end

lib/new_relic/metric/metric.ex

@@ -1,27 +1,12 @@
 defmodule NewRelic.Metric do
+  @moduledoc false
+
   defstruct name: "",
             scope: "",
             call_count: 0,
-            max_call_time: 0,
-            min_call_time: 0,
-            sum_of_squares: 0,
             total_call_time: 0,
-            total_exclusive_time: 0
-
-  @moduledoc false
-
-  def merge(one, two) do
-    %{
-      one
-      | call_count: one.call_count + two.call_count,
-        max_call_time: max(one.max_call_time, two.max_call_time),
-        min_call_time: calculate_min_call_time(one.min_call_time, two.min_call_time),
-        sum_of_squares: one.sum_of_squares + two.sum_of_squares,
-        total_call_time: one.total_call_time + two.total_call_time,
-        total_exclusive_time: one.total_exclusive_time + two.total_exclusive_time
-    }
-  end
-
-  defp calculate_min_call_time(cur, acc) when cur == 0 or acc == 0, do: max(cur, acc)
-  defp calculate_min_call_time(cur, acc), do: min(cur, acc)
+            total_exclusive_time: 0,
+            min_call_time: 0,
+            max_call_time: 0,
+            sum_of_squares: 0
 end

lib/new_relic/metric/metric_data.ex

@@ -399,14 +399,18 @@ defmodule NewRelic.Metric.MetricData do
     do: %Metric{
       name: :"Memory/Physical",
       call_count: 1,
-      total_call_time: memory_mb
+      total_call_time: memory_mb,
+      min_call_time: memory_mb,
+      max_call_time: memory_mb
     }
 
   def transform(:cpu, utilization: utilization),
     do: %Metric{
       name: :"CPU/User Time",
       call_count: 1,
-      total_call_time: utilization
+      total_call_time: utilization,
+      min_call_time: utilization,
+      max_call_time: utilization
     }
 
   def transform(:apdex, apdex: :satisfying, threshold: t),

lib/new_relic/transaction/complete.ex

@@ -605,7 +605,8 @@ defmodule NewRelic.Transaction.Complete do
     NewRelic.report_metric(:apdex, apdex: apdex, threshold: apdex_t())
   end
 
-  def apdex_t, do: Collector.AgentRun.apdex_t()
+  @default_apdex_t 2.0
+  def apdex_t, do: Collector.AgentRun.apdex_t() || @default_apdex_t
 
   defp parse_error_expected(%{expected: true}), do: true
   defp parse_error_expected(_), do: false

test/metric_test.exs

@@ -1,39 +1,6 @@
 defmodule MetricTest do
   use ExUnit.Case
 
-  test "merge two Metrics" do
-    one = %NewRelic.Metric{
-      name: "name",
-      scope: "scope",
-      call_count: 2,
-      max_call_time: 3.4,
-      min_call_time: 1.3,
-      sum_of_squares: 10,
-      total_call_time: 100,
-      total_exclusive_time: 90
-    }
-
-    two = %NewRelic.Metric{
-      name: "name",
-      scope: "scope",
-      call_count: 1,
-      max_call_time: 1,
-      min_call_time: 3.1,
-      sum_of_squares: 100,
-      total_call_time: 5,
-      total_exclusive_time: 1
-    }
-
-    aggregated = NewRelic.Metric.merge(one, two)
-
-    assert aggregated.call_count == 3
-    assert aggregated.max_call_time == 3.4
-    assert aggregated.min_call_time == 1.3
-    assert aggregated.sum_of_squares == 110
-    assert aggregated.total_call_time == 105
-    assert aggregated.total_exclusive_time == 91
-  end
-
   test "custom metrics" do
     TestHelper.restart_harvest_cycle(NewRelic.Harvest.Collector.Metric.HarvestCycle)
 
@@ -42,11 +9,12 @@ defmodule MetricTest do
 
     metrics = TestHelper.gather_harvest(NewRelic.Harvest.Collector.Metric.Harvester)
 
-    expected_count = 2
-    expected_value = 150
+    [_, [count, value, _, min, max, _]] = TestHelper.find_metric(metrics, "Custom/Foo/Bar", 2)
 
-    [_, [^expected_count, ^expected_value, _, _, _, _]] =
-      TestHelper.find_metric(metrics, "Custom/Foo/Bar", expected_count)
+    assert count == 2
+    assert value == 150.0
+    assert max == 100.0
+    assert min == 50.0
 
     TestHelper.pause_harvest_cycle(NewRelic.Harvest.Collector.Metric.HarvestCycle)
   end

test/metric_transaction_test.exs

@@ -163,7 +163,7 @@ defmodule MetricTransactionTest do
 
     metrics = TestHelper.gather_harvest(Collector.Metric.Harvester)
 
-    assert [_, [1, time, time, time, time, 0]] =
+    assert [_, [1, time, time, time, time, 0.0]] =
              TestHelper.find_metric(metrics, "WebFrontend/QueueTime")
 
     assert_in_delta time, 0.1, 0.02
@@ -179,7 +179,7 @@ defmodule MetricTransactionTest do
 
     apdex = TestHelper.find_metric(metrics, "Apdex", 0)
 
-    assert [_, [_, _, 1, _, _, _]] = apdex
+    assert [_, [_, _, 1.0, _, _, _]] = apdex
   end
 
   test "Custom transaction names" do

test/sampler_test.exs

@@ -128,7 +128,7 @@ defmodule SamplerTest do
 
     metrics = TestHelper.gather_harvest(Collector.Metric.Harvester)
 
-    assert [_, [_, 2, _, _, _, _]] =
+    assert [_, [_, 2.0, _, _, _, _]] =
              TestHelper.find_metric(
                metrics,
                "Supportability/ElixirAgent/ReporterCompleteTasksActive"