Telemetry (#905)

CHANGELOG.md

@@ -5,6 +5,7 @@
  * Implement `Membrane.Connector`. [#904](https://github.com/membraneframework/membrane_core/pull/904)
  * Implememnt diamonds detection. [#909](https://github.com/membraneframework/membrane_core/pull/909)
  * Incorporate `Membrane.Funnel`, `Membrane.Tee` and `Membane.Fake.Sink`. [#922](https://github.com/membraneframework/membrane_core/issues/922)
+ * Added new, revised Telemetry system [#905](https://github.com/membraneframework/membrane_core/pull/918)
 
 ## 1.1.2
  * Add new callback `handle_child_terminated/3` along with new assertions. [#894](https://github.com/membraneframework/membrane_core/pull/894)

lib/membrane/core/callback_handler.ex

@@ -133,7 +133,8 @@ defmodule Membrane.Core.CallbackHandler do
          %{context: context_fun},
          %{module: module, internal_state: internal_state} = state
        ) do
-    args = args ++ [context_fun.(state), internal_state]
+    context = context_fun.(state)
+    args = args ++ [context, internal_state]
 
     callback_result =
       try do
@@ -143,7 +144,13 @@ defmodule Membrane.Core.CallbackHandler do
           fn ->
             res = {_actions, new_internal_state} = apply(module, callback, args)
 
-            Telemetry.state_result(res, args, internal_state, new_internal_state, state)
+            Telemetry.state_result(res, %{
+              args: args,
+              module: module,
+              internal_state_before: internal_state,
+              internal_state_after: new_internal_state,
+              component_context: context
+            })
           end
         )
       rescue

lib/membrane/core/element/manual_flow_controller/input_queue.ex

@@ -10,10 +10,10 @@ defmodule Membrane.Core.Element.ManualFlowController.InputQueue do
 
   alias Membrane.Buffer
   alias Membrane.Core.Element.AtomicDemand
+  alias Membrane.Core.Telemetry
   alias Membrane.Event
   alias Membrane.Pad
   alias Membrane.StreamFormat
-  alias Membrane.Core.Telemetry
 
   require Membrane.Core.Stalker, as: Stalker
   require Membrane.Logger
@@ -174,6 +174,7 @@ defmodule Membrane.Core.Element.ManualFlowController.InputQueue do
     input_queue = maybe_increase_atomic_demand(input_queue)
 
     %{size: size, stalker_metrics: stalker_metrics} = input_queue
+    Telemetry.report_take(size, input_queue.log_tag)
     :atomics.put(stalker_metrics.size, 1, size)
 
     {out, input_queue}

lib/membrane/core/element/stream_format_controller.ex

@@ -33,7 +33,7 @@ defmodule Membrane.Core.Element.StreamFormatController do
     withl pad: {:ok, data} <- PadModel.get_data(state, pad_ref),
           playback: %State{playback: :playing} <- state do
       %{direction: :input} = data
-      Telemetry.report_stream_format(stream_format, %{pad_ref: inspect(pad_ref)})
+      Telemetry.report_stream_format(stream_format, inspect(pad_ref))
 
       queue = data.input_queue
 

lib/membrane/core/legacy_telemetry.ex

@@ -1,5 +1,10 @@
 defmodule Membrane.Core.LegacyTelemetry do
   @moduledoc false
+
+  # This module is deprecated and will be removed in the next major release (2.0)
+  # All contents below starting from the next comment are from Legacy Telemetry system
+  # to retain compatibility with the old telemetry system in case of outdated configuration.
+
   # This module provides a way to gather events from running Membrane components, as well
   # as exposing these events in a format idiomatic to [Telemetry](https://hexdocs.pm/telemetry/)
   # library. It uses compile time flags from `config.exs` to determine which events should be
@@ -14,7 +19,7 @@ defmodule Membrane.Core.LegacyTelemetry do
   @doc """
   Reports metrics such as input buffer's size inside functions, incoming events and received stream format.
   """
-  def report_metric(metric, value, log_tag \\ nil) do
+  defmacro report_metric(metric, value, log_tag \\ nil) do
     event =
       quote do
         [:membrane, :metric, :value]
@@ -40,7 +45,7 @@ defmodule Membrane.Core.LegacyTelemetry do
   Given list of buffers (or a single buffer) calculates total size of their payloads in bits
   and reports it.
   """
-  def report_bitrate(buffers) do
+  defmacro report_bitrate(buffers) do
     event =
       quote do
         [:membrane, :metric, :value]
@@ -56,7 +61,7 @@ defmodule Membrane.Core.LegacyTelemetry do
               Enum.reduce(
                 List.wrap(unquote(buffers)),
                 0,
-                &(Membrane.Core.LegacyTelemetry.get_payload_size(&1.payload) + &2)
+                &(unquote(__MODULE__).get_payload_size(&1.payload) + &2)
               )
         }
       end
@@ -68,6 +73,7 @@ defmodule Membrane.Core.LegacyTelemetry do
     )
   end
 
+  @spec get_payload_size(any()) :: non_neg_integer()
   def get_payload_size(payload) when is_bitstring(payload) do
     Membrane.Payload.size(payload)
   end
@@ -87,7 +93,7 @@ defmodule Membrane.Core.LegacyTelemetry do
   @doc """
   Reports new link connection being initialized in pipeline.
   """
-  def report_link(from, to) do
+  defmacro report_link(from, to) do
     event =
       quote do
         [:membrane, :link, :new]
@@ -100,10 +106,10 @@ defmodule Membrane.Core.LegacyTelemetry do
           from: inspect(unquote(from).child),
           to: inspect(unquote(to).child),
           pad_from:
-            Membrane.Core.LegacyTelemetry.__get_public_pad_name__(unquote(from).pad_ref)
+            unquote(__MODULE__).__get_public_pad_name__(unquote(from).pad_ref)
             |> inspect(),
           pad_to:
-            Membrane.Core.LegacyTelemetry.__get_public_pad_name__(unquote(to).pad_ref)
+            unquote(__MODULE__).__get_public_pad_name__(unquote(to).pad_ref)
             |> inspect()
         }
       end
@@ -114,7 +120,7 @@ defmodule Membrane.Core.LegacyTelemetry do
   @doc """
   Reports a pipeline/bin/element initialization.
   """
-  def report_init(type) when type in [:pipeline, :bin, :element] do
+  defmacro report_init(type) when type in [:pipeline, :bin, :element] do
     event =
       quote do
         [:membrane, unquote(type), :init]
@@ -141,7 +147,7 @@ defmodule Membrane.Core.LegacyTelemetry do
   @doc """
   Reports a pipeline/bin/element termination.
   """
-  def report_terminate(type) when type in [:pipeline, :bin, :element] do
+  defmacro report_terminate(type) when type in [:pipeline, :bin, :element] do
     event =
       quote do
         [:membrane, unquote(type), :terminate]

lib/membrane/core/parent/child_life_controller/link_utils.ex

@@ -5,8 +5,8 @@ defmodule Membrane.Core.Parent.ChildLifeController.LinkUtils do
 
   alias Membrane.Child
   alias Membrane.Core.{Bin, Message, Parent}
-  alias Membrane.Core.Telemetry
   alias Membrane.Core.Bin.PadController
+  alias Membrane.Core.Telemetry
 
   alias Membrane.Core.Parent.{
     ChildLifeController,