fix: metrics not routing to sources properly

Author: Ziinc

lib/logflare/backends/source_sup.ex

@@ -20,7 +20,6 @@ defmodule Logflare.Backends.SourceSup do
   alias Logflare.Rules.Rule
   alias Logflare.Sources
   alias Logflare.Backends.AdaptorSupervisor
-  alias Logflare.Backends.UserMonitoring
 
   def child_spec(%Source{id: id} = arg) do
     %{

lib/logflare/backends/user_monitoring.ex

@@ -7,7 +7,6 @@ defmodule Logflare.Backends.UserMonitoring do
   import Logflare.Utils.Guards
   alias Logflare.Users
   alias Logflare.Sources
-  alias Logflare.Sources.Source
   alias Logflare.Logs
   alias Logflare.Logs.Processor
   alias Opentelemetry.Proto.Collector.Metrics.V1.ExportMetricsServiceRequest
@@ -43,15 +42,19 @@ defmodule Logflare.Backends.UserMonitoring do
 
   def metrics do
     [
+      sum("logflare.backends.ingest.ingested_bytes",
+        keep: &keep_metric_function/1,
+        description: "Amount of bytes ingested by backend for a source"
+      ),
+      # sum("logflare.endpoints.query.scanned_bytes",
+      #   keep: &keep_metric_function/1,
+      #   description: "Amount of bytes scanned by a Logflare Endpoint"
+      # ),
       counter("logflare.backends.ingest.ingested_count",
         measurement: :ingested_bytes,
         keep: &keep_metric_function/1,
         description: "Count of events ingested by backend for a source"
       ),
-      sum("logflare.backends.ingest.ingested_bytes",
-        keep: &keep_metric_function/1,
-        description: "Amount of bytes ingested by backend for a source"
-      ),
       sum("logflare.backends.ingest.egress.request_bytes",
         keep: &keep_metric_function/1,
         description:
@@ -68,7 +71,7 @@ defmodule Logflare.Backends.UserMonitoring do
   end
 
   # take all metadata string keys and non-nested values
-  defp extract_tags(metric, metadata) do
+  defp extract_tags(_metric, metadata) do
     for {key, value} <- metadata,
         is_binary(key) and !is_list_or_map(value) and value != nil,
         into: %{} do
@@ -78,30 +81,27 @@ defmodule Logflare.Backends.UserMonitoring do
 
   defp exporter_callback({:metrics, metrics}, config) do
     metrics
-    |> Enum.group_by(fn metric ->
-      metric
-      |> Protobuf.encode()
-      |> Protobuf.decode(Opentelemetry.Proto.Metrics.V1.Metric)
-      |> Map.get(:data)
-      |> Logflare.Logs.OtelMetric.handle_metric_data(%{})
-      |> hd()
+
+    |> OtelMetricExporter.Protocol.build_metric_service_request(config.resource)
+    |> Protobuf.encode()
+    |> Protobuf.decode(ExportMetricsServiceRequest)
+    |> Map.get(:resource_metrics)
+    |> Logs.OtelMetric.handle_batch(%{})
+    |> List.flatten()
+    |> Enum.group_by(fn event ->
+      event
       |> Map.get("attributes")
       |> Users.get_related_user_id()
     end)
-    |> Enum.each(fn {user_id, user_metrics} ->
+    |> Enum.each(fn {user_id, user_events} ->
       source =
         Sources.Cache.get_by(user_id: user_id, system_source_type: :metrics)
         |> Sources.Cache.preload_rules()
         |> Sources.refresh_source_metrics()
 
-      user_metrics
-      |> OtelMetricExporter.Protocol.build_metric_service_request(config.resource)
-      |> Protobuf.encode()
-      |> Protobuf.decode(ExportMetricsServiceRequest)
-      |> Map.get(:resource_metrics)
-      |> Processor.ingest(Logs.OtelMetric, source)
-    end)
+        Processor.ingest(user_events, Logs.Raw, source)
 
+    end)
     :ok
   end
 
@@ -135,12 +135,6 @@ defmodule Logflare.Backends.UserMonitoring do
       |> Sources.refresh_source_metrics()
       |> Sources.Cache.preload_rules()
 
-  defp get_system_source_metrics(user_id),
-    do:
-      Sources.Cache.get_by(user_id: user_id, system_source_type: :metrics)
-      |> Sources.refresh_source_metrics()
-      |> Sources.Cache.preload_rules()
-
   defp format_message(event),
     do:
       :logger_formatter.format(event, %{single_line: true, template: [:msg]})

lib/logflare_web/templates/source/edit.html.eex

@@ -310,7 +310,7 @@
   <% end %>
 
 
-  <% if @user.system_monitoring do %>
+  <%= if @user.system_monitoring do %>
     <%= section_header("Monitoring Labels") %>
     <p>Set labels for this source, to be used for monitoring. Labels are used to group metrics together when system monitoring is enabled. Values of the labels will be determined at ingest time.</p>
     <%= form_for @changeset, Routes.source_path(@conn, :update, @source),fn f -> %>

mix.exs

@@ -242,7 +242,7 @@ defmodule Logflare.Mixfile do
       {:opentelemetry_phoenix, "~> 2.0.0-rc.2"},
       {:opentelemetry_bandit, "~> 0.2.0-rc.1"},
       {:otel_metric_exporter,
-       git: "https://github.com/supabase/elixir-otel-metric-exporter", ref: "61e2d37"},
+       git: "https://github.com/supabase/elixir-otel-metric-exporter", ref: "e44bc0e"},
       {:live_monaco_editor, "~> 0.2"}
     ]
   end

mix.lock

@@ -110,7 +110,7 @@
   "opentelemetry_semantic_conventions": {:hex, :opentelemetry_semantic_conventions, "1.27.0", "acd0194a94a1e57d63da982ee9f4a9f88834ae0b31b0bd850815fe9be4bbb45f", [:mix, :rebar3], [], "hexpm", "9681ccaa24fd3d810b4461581717661fd85ff7019b082c2dff89c7d5b1fc2864"},
   "opentelemetry_telemetry": {:hex, :opentelemetry_telemetry, "1.1.2", "410ab4d76b0921f42dbccbe5a7c831b8125282850be649ee1f70050d3961118a", [:mix, :rebar3], [{:opentelemetry_api, "~> 1.3", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.1", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "641ab469deb181957ac6d59bce6e1321d5fe2a56df444fc9c19afcad623ab253"},
   "otel_http": {:hex, :otel_http, "0.2.0", "b17385986c7f1b862f5d577f72614ecaa29de40392b7618869999326b9a61d8a", [:rebar3], [], "hexpm", "f2beadf922c8cfeb0965488dd736c95cc6ea8b9efce89466b3904d317d7cc717"},
-  "otel_metric_exporter": {:git, "https://github.com/supabase/elixir-otel-metric-exporter", "61e2d371638e5ff7edad09639a0d9bc7a694d875", [ref: "61e2d37"]},
+  "otel_metric_exporter": {:git, "https://github.com/supabase/elixir-otel-metric-exporter", "e44bc0e7f3013933df022cae6d0b9b90320f3d1d", [ref: "e44bc0e"]},
   "parse_trans": {:hex, :parse_trans, "3.4.1", "6e6aa8167cb44cc8f39441d05193be6e6f4e7c2946cb2759f015f8c56b76e5ff", [:rebar3], [], "hexpm", "620a406ce75dada827b82e453c19cf06776be266f5a67cff34e1ef2cbb60e49a"},
   "pgoutput_decoder": {:hex, :pgoutput_decoder, "0.1.0", "d4ffae6e58a563f2e6de8a0495d9f9afbe2f4ac75d6805419cd4a0d05f414c00", [:mix], [], "hexpm", "4dbecbe4eb8de728178fd129ccba810bccafa9a8769c6714c8b7b22963081c27"},
   "phoenix": {:hex, :phoenix, "1.7.21", "14ca4f1071a5f65121217d6b57ac5712d1857e40a0833aff7a691b7870fc9a3b", [:mix], [{:castore, ">= 0.0.0", [hex: :castore, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 2.1", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:phoenix_template, "~> 1.0", [hex: :phoenix_template, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 2.0", [hex: :phoenix_view, repo: "hexpm", optional: true]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.7", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:plug_crypto, "~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:websock_adapter, "~> 0.5.3", [hex: :websock_adapter, repo: "hexpm", optional: false]}], "hexpm", "336dce4f86cba56fed312a7d280bf2282c720abb6074bdb1b61ec8095bdd0bc9"},

test/logflare/backends/user_monitoring_test.exs

@@ -28,6 +28,12 @@ defmodule Logflare.Backends.UserMonitoringTest do
     [source: source, source_b: source_b, user: user]
   end
 
+  def start_otel_exporter(_context) do
+    [spec] = UserMonitoring.get_otel_exporter()
+    start_supervised!(spec)
+    :ok
+  end
+
   describe "logs" do
     setup do
       :ok =
@@ -81,44 +87,16 @@ defmodule Logflare.Backends.UserMonitoringTest do
     end
   end
 
-  describe "metrics" do
-    setup do
-      insert(:plan)
-
-      user_1 = insert(:user)
-      user_1 = Users.preload_defaults(user_1)
-
-      backend_1 =
-        insert(:backend,
-          user: user_1,
-          type: :webhook,
-          config: %{url: "http://test.com"},
-          default_ingest?: false
-        )
-
-      user_2 = insert(:user)
-      user_2 = Users.preload_defaults(user_2)
-
-      backend_2 =
-        insert(:backend,
-          user: user_2,
-          type: :webhook,
-          config: %{url: "http://test.com"},
-          default_ingest?: false
-        )
-
-      {:ok, user_1: user_1, user_2: user_2, backend_1: backend_1, backend_2: backend_2}
-    end
-  end
 
   describe "system monitoring labels" do
+    setup :start_otel_exporter
     setup do
       start_supervised!(AllLogsLogged)
       insert(:plan)
       :ok
     end
 
-    test "applies to backend metrics" do
+test "backends.ingest.ingested_bytes and backends.ingest.ingested_count" do
       GoogleApi.BigQuery.V2.Api.Tabledata
       |> stub(:bigquery_tabledata_insert_all, fn _conn,
                                                  _project_id,
@@ -175,74 +153,60 @@ defmodule Logflare.Backends.UserMonitoringTest do
     end
 
     test "other users metrics" do
-      GoogleApi.BigQuery.V2.Api.Tabledata
-      |> stub(:bigquery_tabledata_insert_all, fn _conn,
-                                                 _project_id,
-                                                 _dataset_id,
-                                                 _table_name,
-                                                 _opts ->
-        {:ok, %GoogleApi.BigQuery.V2.Model.TableDataInsertAllResponse{insertErrors: nil}}
-      end)
+      pid = self()
 
       user =
         insert(:user, system_monitoring: true)
 
       other_user =
         insert(:user, system_monitoring: true)
 
-      %{id: source_id} = source = insert(:source, user: user, labels: "my_label=m.value")
+      source = insert(:source, user: user, labels: "my_label=m.value")
 
-      %{id: other_source_id} =
-        other_source = insert(:source, user: other_user, labels: "my_label=m.value")
+      other_source = insert(:source, user: other_user, labels: "my_label=m.value")
 
       metrics_source = insert(:source, user: user, system_source_type: :metrics)
       other_metrics_source = insert(:source, user: other_user, system_source_type: :metrics)
-      start_supervised!({SourceSup, metrics_source}, id: :metrics_source)
       start_supervised!({SourceSup, source}, id: :source)
       start_supervised!({SourceSup, other_source}, id: :other_source)
+      start_supervised!({SourceSup, metrics_source}, id: :metrics_source)
       start_supervised!({SourceSup, other_metrics_source}, id: :other_metrics_source)
+      GoogleApi.BigQuery.V2.Api.Tabledata
+      |> stub(:bigquery_tabledata_insert_all, fn _conn,
+                                                 _project_id,
+                                                 dataset_id,
+                                                 _table_name,
+                                                 opts ->
+                                                  if String.starts_with?(dataset_id, "#{other_user.id}") do
+                                                    send(pid, {:insert_all, opts[:body].rows})
+                                                  end
+                                                  {:ok, %GoogleApi.BigQuery.V2.Model.TableDataInsertAllResponse{insertErrors: nil}}
+      end)
+
+      :timer.sleep(500)
 
-      :timer.sleep(1000)
 
       assert {:ok, _} = Backends.ingest_logs([%{"metadata" => %{"value" => "test"}}], source)
 
       assert {:ok, _} =
-               Backends.ingest_logs([%{"metadata" => %{"value" => "test"}}], other_source)
+               Backends.ingest_logs([%{"metadata" => %{"value" => "different"}}], other_source)
+               :timer.sleep(1500)
 
-      TestUtils.retry_assert(fn ->
-        assert [_] = Backends.list_recent_logs_local(source)
-        assert [_] = Backends.list_recent_logs_local(other_source)
+      source_id = source.id
+      other_source_id = other_source.id
+      metrics_source_id = metrics_source.id
+      other_metrics_source_id = other_metrics_source.id
+      assert_receive {:insert_all, [%{json: %{"attributes" => _}} | _]= rows}, 5_000
 
-        assert [
-                 _ | _
-               ] = events = Backends.list_recent_logs_local(metrics_source)
+      rows = for row <- rows, do: row.json
 
-        assert Enum.all?(
-                 events,
-                 &match?(
-                   %LogEvent{
-                     body: %{"attributes" => %{"my_label" => "test", "source_id" => ^source_id}}
-                   },
-                   &1
-                 )
-               )
+      assert Enum.all?(rows, &match?(%{"attributes" => [%{"my_label" => "different"}]}, &1))
+      for row <- rows, attr <- row["attributes"] do
+        assert attr["source_id"] in [other_source_id, other_metrics_source_id]
+        refute attr["source_id"] in [source_id, metrics_source_id]
+      refute attr["my_label"] == "test"
+      end
 
-        assert [
-                 _ | _
-               ] = events = Backends.list_recent_logs_local(other_metrics_source)
-
-        assert Enum.all?(
-                 events,
-                 &match?(
-                   %LogEvent{
-                     body: %{
-                       "attributes" => %{"my_label" => "test", "source_id" => ^other_source_id}
-                     }
-                   },
-                   &1
-                 )
-               )
-      end)
     end
   end
 end