feat(appengine): supoprt xandra telemetry bounce (#1462)

Re-bounces xandra telemetry under `astarte.appengine.database` to have
visibility over queries failures or exceptions.

Signed-off-by: Luca Zaninotto <luca.zaninotto@secomind.com>

Author: lusergit

apps/astarte_appengine_api/lib/astarte_appengine_api/config.ex

@@ -160,6 +160,16 @@ defmodule Astarte.AppEngine.API.Config do
           type: :binary,
           default: "astarte"
 
+  @envdoc """
+  "The handling method for database events. The default is `expose`, which means that the events are exposed trough telemetry. The other possible value, `log`, means that the events are logged instead."
+  """
+  app_env :database_events_handling_method,
+          :astarte_appengine_api,
+          :database_events_handling_method,
+          os_env: "DATABASE_EVENTS_HANDLING_METHOD",
+          type: Astarte.AppEngine.API.Config.TelemetryType,
+          default: :expose
+
   @doc """
   Returns the routing key used for Rooms AMQP events consumer. A constant for now.
   """

apps/astarte_appengine_api/lib/astarte_appengine_api/config/telemetry_type.ex

@@ -0,0 +1,42 @@
+#
+# This file is part of Astarte.
+#
+# Copyright 2025 SECO Mind Srl
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# SPDX-License-Identifier: Apache-2.0
+#
+
+defmodule Astarte.AppEngine.API.Config.TelemetryType do
+  @moduledoc """
+  The telemetry type that the node should use to report metrics.
+  """
+
+  use Skogsra.Type
+
+  @allowed_strategies ~w(expose log)
+
+  @impl Skogsra.Type
+  def cast(value) when value in @allowed_strategies do
+    case value do
+      "expose" -> {:ok, :expose}
+      "log" -> {:ok, :log}
+    end
+  end
+
+  @impl Skogsra.Type
+  def cast(_) do
+    :error
+  end
+end

apps/astarte_appengine_api/lib/astarte_appengine_api_web/telemetry.ex

@@ -19,6 +19,8 @@ defmodule Astarte.AppEngine.APIWeb.Telemetry do
   use Supervisor
   import Telemetry.Metrics
 
+  alias Astarte.AppEngine.API.Config
+  alias Astarte.AppEngine.APIWeb.Telemetry.DatabaseEvents
   alias Astarte.AppEngine.APIWeb.Telemetry.APIUsage
 
   def start_link(arg) do
@@ -144,10 +146,57 @@ defmodule Astarte.AppEngine.APIWeb.Telemetry do
       counter("astarte.appengine.channels.unwatch_request.count",
         tags: [:realm],
         description: "Trigger deinstallation requests count"
+      ),
+
+      # Database exception metrics
+      counter("astarte.appengine.database.execute_query.exception.count",
+        tags: [:query, :reason, :kind, :stacktrace],
+        tag_values: &to_valid_values/1,
+        unit: {:native, :second}
+      ),
+      counter("astarte.appengine.database.execute_query.stop.count",
+        tags: [:query, :reason],
+        tag_values: &to_valid_values/1,
+        unit: {:native, :second}
+      ),
+
+      # Database preparation metrics
+      counter("astarte.appengine.database.prepare_query.exception.count",
+        tags: [:query, :reason, :kind, :stacktrace],
+        tag_values: &to_valid_values/1,
+        unit: {:native, :second}
+      ),
+      counter("astarte.appengine.database.prepare_query.stop.count",
+        tags: [:query, :reason],
+        tag_values: &to_valid_values/1,
+        unit: {:native, :second}
+      ),
+
+      # Database connection metrics
+      counter(
+        "astarte.appengine.database.cluster.control_connection.failed_to_connect.count",
+        tag_values: &to_valid_values/1,
+        tags: [:cluster_name, :host, :reason]
+      ),
+      counter("astarte.appengine.database.failed_to_connect.conut",
+        tag_values: &to_valid_values/1,
+        tags: [:connection_name, :address, :port]
       )
     ]
   end
 
+  defp to_valid_values(%{query: query, reason: reason}) do
+    %{query: query.statement, reason: Xandra.Error.message(reason)}
+  end
+
+  defp to_valid_values(%{cluster_name: cluster_name, host: host, reason: reason}) do
+    %{cluster_name: cluster_name, host: inspect(host), reason: to_string(reason)}
+  end
+
+  defp to_valid_values(%{connection_name: connection_name, address: address, port: port}) do
+    %{connection_name: connection_name, address: inspect(address), port: inspect(port)}
+  end
+
   defp periodic_measurements do
     [
       # A module, function and arguments to be invoked periodically.
@@ -158,6 +207,36 @@ defmodule Astarte.AppEngine.APIWeb.Telemetry do
 
   defp attach_handlers do
     :telemetry.attach(APIUsage, [:cowboy, :request, :stop], &APIUsage.handle_event/4, nil)
+
+    :telemetry.attach_many(
+      DatabaseEvents,
+      xandra_events(),
+      &DatabaseEvents.handle_event/4,
+      Config.database_events_handling_method!()
+    )
+  end
+
+  defp xandra_events do
+    [
+      [:xandra, :connected],
+      [:xandra, :disconnected],
+      [:xandra, :failed_to_connect],
+      [:xandra, :prepared_cache, :hit],
+      [:xandra, :prepared_cache, :miss],
+      [:xandra, :prepare_query, :stop],
+      [:xandra, :execute_query, :stop],
+      [:xandra, :client_timeout],
+      [:xandra, :timed_out_response],
+      [:xandra, :server_warnings],
+      [:xandra, :cluster, :change_event],
+      [:xandra, :cluster, :control_connection, :connected],
+      [:xandra, :cluster, :control_connection, :disconnected],
+      [:xandra, :cluster, :control_connection, :failed_to_connect],
+      [:xandra, :cluster, :pool, :started],
+      [:xandra, :cluster, :pool, :restarted],
+      [:xandra, :cluster, :pool, :stopped],
+      [:xandra, :cluster, :discovered_peers]
+    ]
   end
 
   defp extract_phoenix_buckets_metadata(%{

apps/astarte_appengine_api/lib/astarte_appengine_api_web/telemetry/database_events.ex

@@ -0,0 +1,85 @@
+#
+# This file is part of Astarte.
+#
+# Copyright 2025 SECO Mind Srl
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# SPDX-License-Identifier: Apache-2.0
+#
+
+defmodule Astarte.AppEngine.APIWeb.Telemetry.DatabaseEvents do
+  @moduledoc """
+    Telemetry handler for database events.
+
+    This module listens to database events and emits telemetry events
+    with the relevant measurements and metadata.
+  """
+
+  @bounce_events [
+    [:prepare_query, :stop],
+    [:prepare_query, :exception],
+    [:execute, :stop],
+    [:execute, :exception],
+    [:cluster, :control_connection, :failed_to_connect],
+    [:failed_to_connect]
+  ]
+
+  require Logger
+  alias Astarte.AppEngine.APIWeb.TelemetryTaskSupervisor
+
+  @doc """
+  Handles telemetry events related to database operations.
+
+  See the documentation of Xandra for more details on the events:
+  https://hexdocs.pm/xandra/telemetry-events.html
+
+  This handler drops the `:xandra` prefix from the event name and
+  executes the telemetry event with the provided measurements and metadata.
+  """
+  def handle_event([:xandra | event], measurements, metadata, :expose) do
+    with :bounce <- validate_event(event) do
+      Task.Supervisor.start_child(TelemetryTaskSupervisor, fn ->
+        with :ok <- filter_event(event, metadata) do
+          :telemetry.execute(
+            [:astarte, :appengine, :database] ++ event,
+            measurements,
+            metadata
+          )
+        end
+      end)
+    end
+  end
+
+  def handle_event(event, measurements, metadata, :log) do
+    Xandra.Telemetry.handle_event(event, measurements, metadata, :no_config)
+  end
+
+  defp validate_event(event) do
+    case event in @bounce_events do
+      true -> :bounce
+      false -> :ok
+    end
+  end
+
+  defp filter_event([:execute_query, _], metadata), do: has_reason(metadata)
+  defp filter_event([:prepare_query, _], metadata), do: has_reason(metadata)
+  defp filter_event(_event, _metadata), do: :ok
+
+  defp has_reason(metadata) do
+    case Map.has_key?(metadata, :reason) do
+      true -> :ok
+      false -> :do_not_bounce
+    end
+  end
+end