Merge pull request #408 from XiXiaPdx/aggregate-dimensional-metrics

Aggregate dimensional metrics

lib/new_relic.ex

@@ -229,6 +229,17 @@ defmodule NewRelic do
   defdelegate report_custom_event(type, attributes),
     to: NewRelic.Harvest.Collector.CustomEvent.Harvester
 
+  @doc """
+  Report a Dimensional Metric.
+  Valid types: `:count`, `:gauge`, and `:summary`.
+
+  ```elixir
+  NewRelic.report_dimensional_metric(:count, "my_metric_name", 1, %{some: "attributes"})
+  ```
+  """
+  defdelegate report_dimensional_metric(type, name, value, attributes \\ %{}),
+    to: NewRelic.Harvest.TelemetrySdk.DimensionalMetrics.Harvester
+
   @doc """
   Report a Custom metric.
 

lib/new_relic/harvest/supervisor.ex

@@ -14,7 +14,8 @@ defmodule NewRelic.Harvest.Supervisor do
     Harvest.Collector.CustomEvent.HarvestCycle,
     Harvest.Collector.ErrorTrace.HarvestCycle,
     Harvest.TelemetrySdk.Logs.HarvestCycle,
-    Harvest.TelemetrySdk.Spans.HarvestCycle
+    Harvest.TelemetrySdk.Spans.HarvestCycle,
+    Harvest.TelemetrySdk.DimensionalMetrics.HarvestCycle
   ]
 
   def start_link(_) do

lib/new_relic/harvest/telemetry_sdk/api.ex

@@ -17,6 +17,14 @@ defmodule NewRelic.Harvest.TelemetrySdk.API do
     |> maybe_retry(url, payload)
   end
 
+  def dimensional_metric(metrics) do
+    url = url(:metric)
+    payload = {:metrics, metrics, generate_request_id()}
+
+    request(url, payload)
+    |> maybe_retry(url, payload)
+  end
+
   def request(url, payload) do
     post(url, payload)
   end

lib/new_relic/harvest/telemetry_sdk/config.ex

@@ -3,7 +3,8 @@ defmodule NewRelic.Harvest.TelemetrySdk.Config do
 
   @default %{
     logs_harvest_cycle: 5_000,
-    spans_harvest_cycle: 5_000
+    spans_harvest_cycle: 5_000,
+    dimensional_metrics_harvest_cycle: 5_000
   }
   def lookup(key) do
     Application.get_env(:new_relic_agent, key, @default[key])
@@ -18,7 +19,8 @@ defmodule NewRelic.Harvest.TelemetrySdk.Config do
 
     %{
       log: "https://#{env}log-api.#{region}newrelic.com/log/v1",
-      trace: trace_domain(env, region)
+      trace: trace_domain(env, region),
+      metric: metric_domain(env, region)
     }
   end
 
@@ -34,4 +36,8 @@ defmodule NewRelic.Harvest.TelemetrySdk.Config do
   defp trace_domain(_env, _region, infinite_tracing_host) do
     "https://#{infinite_tracing_host}/trace/v1"
   end
+
+  defp metric_domain(env, region) do
+    "https://#{env}metric-api.#{region}newrelic.com/metric/v1"
+  end
 end

lib/new_relic/harvest/telemetry_sdk/dimensional_metrics/harvester.ex

@@ -0,0 +1,158 @@
+defmodule NewRelic.Harvest.TelemetrySdk.DimensionalMetrics.Harvester do
+  use GenServer
+
+  @moduledoc false
+
+  alias NewRelic.Harvest
+  alias NewRelic.Harvest.TelemetrySdk
+
+  @valid_types [:count, :gauge, :summary]
+
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, [])
+  end
+
+  def init(_) do
+    {:ok,
+     %{
+       start_time: get_start_time(),
+       metrics: %{}
+     }}
+  end
+
+  # API
+
+  @spec report_dimensional_metric(:count | :gauge | :summary, atom() | binary(), any, map()) ::
+          :ok
+  def report_dimensional_metric(type, name, value, attributes) when type in @valid_types do
+    TelemetrySdk.DimensionalMetrics.HarvestCycle
+    |> Harvest.HarvestCycle.current_harvester()
+    |> GenServer.cast({:report, %{type: type, name: name, value: value, attributes: attributes}})
+  end
+
+  def gather_harvest,
+    do:
+      TelemetrySdk.DimensionalMetrics.HarvestCycle
+      |> Harvest.HarvestCycle.current_harvester()
+      |> GenServer.call(:gather_harvest)
+
+  # do not accept more report messages when harvest has already been reported
+  def handle_cast(_late_msg, :completed), do: {:noreply, :completed}
+
+  def handle_cast({:report, metric}, state) do
+    {:noreply, %{state | metrics: merge_metric(metric, state.metrics)}}
+  end
+
+  # do not resend metrics when harvest has already been reported
+  def handle_call(_late_msg, _from, :completed), do: {:reply, :completed, :completed}
+
+  def handle_call(:send_harvest, _from, state) do
+    send_harvest(state)
+    {:reply, :ok, :completed}
+  end
+
+  def handle_call(:gather_harvest, _from, state) do
+    {:reply, build_dimensional_metric_data(state.metrics, state), state}
+  end
+
+  # Helpers
+
+  defp merge_metric(
+         %{type: :summary, name: name, value: new_value, attributes: attributes},
+         metrics_acc
+       ) do
+    new_summary = %{
+      type: :summary,
+      name: name,
+      value: %{
+        count: 1,
+        min: new_value,
+        max: new_value,
+        sum: new_value
+      },
+      attributes: attributes
+    }
+
+    Map.update(
+      metrics_acc,
+      {:summary, name, attributes},
+      new_summary,
+      &update_metric(&1, new_value)
+    )
+  end
+
+  defp merge_metric(
+         %{type: type, name: name, value: new_value, attributes: attributes} = metric,
+         metrics_acc
+       ),
+       do:
+         Map.update(metrics_acc, {type, name, attributes}, metric, &update_metric(&1, new_value))
+
+  defp update_metric(
+         %{type: :count, value: value} = current_metric,
+         new_value
+       ),
+       do: %{current_metric | value: value + new_value}
+
+  defp update_metric(
+         %{type: :gauge} = current_metric,
+         new_value
+       ),
+       do: %{current_metric | value: new_value}
+
+  defp update_metric(
+         %{type: :summary} = current_metric,
+         new_value
+       ),
+       do: %{current_metric | value: update_summary_value_map(current_metric, new_value)}
+
+  defp update_summary_value_map(
+         %{type: :summary, value: value_map},
+         new_value
+       ) do
+    updated_sum_count = %{value_map | sum: value_map.sum + new_value, count: value_map.count + 1}
+
+    updated_min =
+      if new_value < value_map.min,
+        do: %{updated_sum_count | min: new_value},
+        else: updated_sum_count
+
+    if new_value > value_map.max, do: %{updated_min | max: new_value}, else: updated_min
+  end
+
+  defp send_harvest(state) do
+    metrics = Map.values(state.metrics)
+    TelemetrySdk.API.dimensional_metric(build_dimensional_metric_data(metrics, state))
+    log_harvest(length(metrics))
+  end
+
+  defp log_harvest(harvest_size) do
+    NewRelic.log(
+      :debug,
+      "Completed TelemetrySdk.DimensionalMetrics harvest - size: #{harvest_size}"
+    )
+  end
+
+  defp build_dimensional_metric_data(metrics, state) do
+    [
+      %{
+        metrics: metrics,
+        common: common(state.start_time)
+      }
+    ]
+  end
+
+  defp common(%{system: start_system_time, mono: start_mono}) do
+    %{
+      "timestamp" => start_system_time,
+      "interval.ms" => System.monotonic_time(:millisecond) - start_mono
+    }
+  end
+
+  defp get_start_time() do
+    %{
+      system: System.system_time(:millisecond),
+      mono: System.monotonic_time(:millisecond)
+    }
+  end
+end

lib/new_relic/harvest/telemetry_sdk/supervisor.ex

@@ -13,7 +13,8 @@ defmodule NewRelic.Harvest.TelemetrySdk.Supervisor do
   def init(_) do
     children = [
       data_supervisor(TelemetrySdk.Logs, :logs_harvest_cycle),
-      data_supervisor(TelemetrySdk.Spans, :spans_harvest_cycle)
+      data_supervisor(TelemetrySdk.Spans, :spans_harvest_cycle),
+      data_supervisor(TelemetrySdk.DimensionalMetrics, :dimensional_metrics_harvest_cycle)
     ]
 
     Supervisor.init(children, strategy: :one_for_all)

test/dimensional_metric_test.exs

@@ -0,0 +1,102 @@
+defmodule DimensionalMetricTest do
+  use ExUnit.Case
+
+  test "reports dimensional metrics" do
+    TestHelper.restart_harvest_cycle(
+      NewRelic.Harvest.TelemetrySdk.DimensionalMetrics.HarvestCycle
+    )
+
+    NewRelic.report_dimensional_metric(:count, "memory.foo_baz", 100, %{cpu: 1000})
+    NewRelic.report_dimensional_metric(:summary, "memory.foo_bar", 50, %{cpu: 2000})
+
+    [%{common: common, metrics: metrics_map}] =
+      TestHelper.gather_harvest(NewRelic.Harvest.TelemetrySdk.DimensionalMetrics.Harvester)
+
+    metrics = Map.values(metrics_map)
+    assert common["interval.ms"] > 0
+    assert common["timestamp"] > 0
+
+    assert length(metrics) == 2
+    [metric1, metric2] = metrics
+    assert metric1.name == "memory.foo_baz"
+    assert metric1.type == :count
+
+    assert metric2.name == "memory.foo_bar"
+    assert metric2.type == :summary
+
+    TestHelper.pause_harvest_cycle(NewRelic.Harvest.TelemetrySdk.DimensionalMetrics.HarvestCycle)
+  end
+
+  test "gauge dimensional metric is updated" do
+    TestHelper.restart_harvest_cycle(
+      NewRelic.Harvest.TelemetrySdk.DimensionalMetrics.HarvestCycle
+    )
+
+    NewRelic.report_dimensional_metric(:gauge, "mem_percent.foo_baz", 10, %{cpu: 1000})
+    NewRelic.report_dimensional_metric(:gauge, "mem_percent.foo_baz", 40, %{cpu: 1000})
+    NewRelic.report_dimensional_metric(:gauge, "mem_percent.foo_baz", 90, %{cpu: 1000})
+
+    [%{metrics: metrics_map}] =
+      TestHelper.gather_harvest(NewRelic.Harvest.TelemetrySdk.DimensionalMetrics.Harvester)
+
+    metrics = Map.values(metrics_map)
+
+    assert length(metrics) == 1
+    [metric] = metrics
+    assert metric.name == "mem_percent.foo_baz"
+    assert metric.type == :gauge
+    assert metric.value == 90
+
+    TestHelper.pause_harvest_cycle(NewRelic.Harvest.TelemetrySdk.DimensionalMetrics.HarvestCycle)
+  end
+
+  test "count dimensional metric is updated" do
+    TestHelper.restart_harvest_cycle(
+      NewRelic.Harvest.TelemetrySdk.DimensionalMetrics.HarvestCycle
+    )
+
+    NewRelic.report_dimensional_metric(:count, "OOM", 1, %{cpu: 1000})
+    NewRelic.report_dimensional_metric(:count, "OOM", 1, %{cpu: 1000})
+    NewRelic.report_dimensional_metric(:count, "OOM", 2, %{cpu: 1000})
+
+    [%{metrics: metrics_map}] =
+      TestHelper.gather_harvest(NewRelic.Harvest.TelemetrySdk.DimensionalMetrics.Harvester)
+
+    metrics = Map.values(metrics_map)
+
+    assert length(metrics) == 1
+    [metric] = metrics
+    assert metric.name == "OOM"
+    assert metric.type == :count
+    assert metric.value == 4
+
+    TestHelper.pause_harvest_cycle(NewRelic.Harvest.TelemetrySdk.DimensionalMetrics.HarvestCycle)
+  end
+
+  test "summary dimensional metric is updated" do
+    TestHelper.restart_harvest_cycle(
+      NewRelic.Harvest.TelemetrySdk.DimensionalMetrics.HarvestCycle
+    )
+
+    NewRelic.report_dimensional_metric(:summary, "duration", 40.5, %{cpu: 1000})
+    NewRelic.report_dimensional_metric(:summary, "duration", 20.5, %{cpu: 1000})
+    NewRelic.report_dimensional_metric(:summary, "duration", 9.5, %{cpu: 1000})
+    NewRelic.report_dimensional_metric(:summary, "duration", 55.5, %{cpu: 1000})
+
+    [%{metrics: metrics_map}] =
+      TestHelper.gather_harvest(NewRelic.Harvest.TelemetrySdk.DimensionalMetrics.Harvester)
+
+    metrics = Map.values(metrics_map)
+
+    assert length(metrics) == 1
+    [metric] = metrics
+    assert metric.name == "duration"
+    assert metric.type == :summary
+    assert metric.value.sum == 126
+    assert metric.value.min == 9.5
+    assert metric.value.max == 55.5
+    assert metric.value.count == 4
+
+    TestHelper.pause_harvest_cycle(NewRelic.Harvest.TelemetrySdk.DimensionalMetrics.HarvestCycle)
+  end
+end

test/integration/integration_test.exs

@@ -96,6 +96,25 @@ defmodule IntegrationTest do
     assert resp.status_code == 202
   end
 
+  test "Can post a dimensional metric" do
+    {:ok, resp} =
+      NewRelic.Harvest.TelemetrySdk.API.dimensional_metric([
+        %{
+          metrics: [
+            %{
+              attributes: %{cpu: 1000},
+              name: "mem_percent.foo_baz",
+              type: :gauge,
+              value: 90
+            }
+          ],
+          common: %{"timestamp" => System.system_time(:millisecond), "interval.ms" => 5000}
+        }
+      ])
+
+    assert resp.status_code == 202
+  end
+
   test "EnabledSupervisor starts" do
     NewRelic.EnabledSupervisorManager.start_child()