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,147 @@
+defmodule NewRelic.Harvest.TelemetrySdk.DimensionalMetrics.Harvester do
+  use GenServer
+
+  @moduledoc false
+
+  alias NewRelic.Harvest
+  alias NewRelic.Harvest.TelemetrySdk
+
+  @interval_ms TelemetrySdk.Config.lookup(:dimensional_metrics_harvest_cycle)
+
+  @valid_types [:count, :gauge, :summary]
+
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, [])
+  end
+
+  def init(_) do
+    {:ok,
+     %{
+       start_time_ms: System.system_time(:millisecond),
+       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: type, name: name, value: new_value, attributes: attributes} = metric,
+         metrics_acc
+       ) do
+    attributes_hash = :erlang.phash2(attributes)
+
+    case Map.get(metrics_acc, {type, name, attributes_hash}) do
+      nil ->
+        case type do
+          :summary ->
+            new_summary = %{
+              type: type,
+              name: name,
+              value: %{
+                count: 1,
+                min: new_value,
+                max: new_value,
+                sum: new_value
+              },
+              attributes: attributes
+            }
+
+            Map.put(metrics_acc, {type, name, attributes_hash}, new_summary)
+
+          _ ->
+            Map.put(metrics_acc, {type, name, attributes_hash}, metric)
+        end
+
+      %{type: :count, value: current_value} = current_metric ->
+        Map.put(metrics_acc, {type, name, attributes_hash}, %{
+          current_metric
+          | value: current_value + new_value
+        })
+
+      %{type: :gauge} = current_metric ->
+        Map.put(metrics_acc, {type, name, attributes_hash}, %{current_metric | value: new_value})
+
+      %{type: :summary} = current_metric ->
+        Map.put(
+          metrics_acc,
+          {type, name, attributes_hash},
+          %{current_metric | value: update_summary_value_map(current_metric, new_value)}
+        )
+    end
+  end
+
+  defp update_summary_value_map(
+         %{type: :summary, value: value_map} = current_metric,
+         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)
+      }
+    ]
+  end
+
+  defp common(%{start_time_ms: start_time_ms}) do
+    %{"timestamp" => start_time_ms, "interval.ms" => @interval_ms}
+  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/telemetry_sdk/dimensional_metrics_harvester_test.exs

@@ -0,0 +1,59 @@
+defmodule TelemetrySdk.DimensionalMetricsHarvesterTest do
+  use ExUnit.Case
+
+  alias NewRelic.Harvest
+  alias NewRelic.Harvest.TelemetrySdk
+
+  test "Harvester collects dimensional metrics" do
+    {:ok, harvester} =
+      DynamicSupervisor.start_child(
+        TelemetrySdk.DimensionalMetrics.HarvesterSupervisor,
+        TelemetrySdk.DimensionalMetrics.Harvester
+      )
+
+    metric1 = %{type: :gauge, name: "cpu", value: 10, attributes: %{k8: true, id: 123}}
+    GenServer.cast(harvester, {:report, metric1})
+
+    metrics = GenServer.call(harvester, :gather_harvest)
+    assert length(metrics) > 0
+  end
+
+  test "harvest cycle" do
+    Application.put_env(:new_relic_agent, :dimensional_metrics_harvest_cycle, 300)
+    TestHelper.restart_harvest_cycle(TelemetrySdk.DimensionalMetrics.HarvestCycle)
+
+    first = Harvest.HarvestCycle.current_harvester(TelemetrySdk.DimensionalMetrics.HarvestCycle)
+    Process.monitor(first)
+
+    # Wait until harvest swap
+    assert_receive {:DOWN, _ref, _, ^first, :shutdown}, 1000
+
+    second = Harvest.HarvestCycle.current_harvester(TelemetrySdk.DimensionalMetrics.HarvestCycle)
+    Process.monitor(second)
+
+    refute first == second
+    assert Process.alive?(second)
+
+    TestHelper.pause_harvest_cycle(TelemetrySdk.DimensionalMetrics.HarvestCycle)
+    Application.delete_env(:new_relic_agent, :dimensional_metrics_harvest_cycle)
+
+    # Ensure the last harvester has shut down
+    assert_receive {:DOWN, _ref, _, ^second, :shutdown}, 1000
+  end
+
+  test "Ignore late reports" do
+    TestHelper.restart_harvest_cycle(TelemetrySdk.DimensionalMetrics.HarvestCycle)
+
+    harvester =
+      TelemetrySdk.DimensionalMetrics.HarvestCycle
+      |> Harvest.HarvestCycle.current_harvester()
+
+    assert :ok == GenServer.call(harvester, :send_harvest)
+
+    GenServer.cast(harvester, {:report, :late_msg})
+
+    assert :completed == GenServer.call(harvester, :send_harvest)
+
+    TestHelper.pause_harvest_cycle(TelemetrySdk.DimensionalMetrics.HarvestCycle)
+  end
+end