Telemetry (#905)

config/config.exs

@@ -15,7 +15,7 @@ if config_env() == :test do
       bin: :all,
       pipeline: :all
     ],
-    events: :all
+    datapoints: :all
 
   # config to test legacy version of telemetry
   # config :membrane_core, :telemetry_flags, [

lib/membrane/bin/callback_context.ex

@@ -16,13 +16,15 @@ defmodule Membrane.Bin.CallbackContext do
   `c:Membrane.Bin.handle_crash_group_down/3`.
   """
   @type t :: %{
+          :children => %{Membrane.Child.name() => Membrane.ChildEntry.t()},
           :clock => Membrane.Clock.t(),
           :parent_clock => Membrane.Clock.t(),
-          :pads => %{Membrane.Pad.ref() => Membrane.Bin.PadData.t()},
+          :module => module(),
           :name => Membrane.Bin.name(),
-          :children => %{Membrane.Child.name() => Membrane.ChildEntry.t()},
+          :pads => %{Membrane.Pad.ref() => Membrane.Bin.PadData.t()},
           :playback => Membrane.Playback.t(),
           :resource_guard => Membrane.ResourceGuard.t(),
+          :setup_incomplete? => boolean(),
           :utility_supervisor => Membrane.UtilitySupervisor.t(),
           optional(:pad_options) => map(),
           optional(:members) => [Membrane.Child.name()],

lib/membrane/core/bin/callback_context.ex

@@ -15,13 +15,15 @@ defmodule Membrane.Core.Bin.CallbackContext do
   def from_state(state, optional_fields \\ []) do
     Map.new(optional_fields)
     |> Map.merge(%{
+      children: state.children,
       clock: state.synchronization.clock,
       parent_clock: state.synchronization.parent_clock,
       pads: state.pads_data,
+      module: state.module,
       name: state.name,
-      children: state.children,
       playback: state.playback,
       resource_guard: state.resource_guard,
+      setup_incomplete?: state.setup_incomplete?,
       utility_supervisor: state.subprocess_supervisor
     })
   end

lib/membrane/core/callback_handler.ex

@@ -8,6 +8,7 @@ defmodule Membrane.Core.CallbackHandler do
   use Bunch
 
   alias Membrane.CallbackError
+  alias Membrane.ComponentPath
 
   require Membrane.Logger
   require Membrane.Core.Telemetry, as: Telemetry
@@ -126,7 +127,7 @@ defmodule Membrane.Core.CallbackHandler do
   end
 
   @spec exec_callback(callback :: atom, args :: list, handler_params, state) ::
-          {list, internal_state}
+          {list, internal_state} | no_return()
   defp exec_callback(
          callback,
          args,
@@ -136,50 +137,57 @@ defmodule Membrane.Core.CallbackHandler do
     context = context_fun.(state)
     args = args ++ [context, internal_state]
 
-    callback_result =
-      try do
-        Telemetry.component_span(
-          state.__struct__,
-          callback,
-          fn ->
-            res = {_actions, new_internal_state} = apply(module, callback, args)
-
-            Telemetry.state_result(res, %{
-              args: args,
-              module: module,
-              internal_state_before: internal_state,
-              internal_state_after: new_internal_state,
-              component_context: context
-            })
+    try do
+      fn ->
+        apply(module, callback, args)
+        |> validate_callback_result!(module, callback)
+      end
+      |> report_telemetry(callback, args, state, context)
+    rescue
+      e in UndefinedFunctionError ->
+        _ignored =
+          with %{module: ^module, function: ^callback, arity: arity} <- e do
+            reraise CallbackError,
+                    [
+                      kind: :not_implemented,
+                      callback: {module, callback},
+                      arity: arity,
+                      args: args
+                    ],
+                    __STACKTRACE__
           end
-        )
-      rescue
-        e in UndefinedFunctionError ->
-          _ignored =
-            with %{module: ^module, function: ^callback, arity: arity} <- e do
-              reraise CallbackError,
-                      [
-                        kind: :not_implemented,
-                        callback: {module, callback},
-                        arity: arity,
-                        args: args
-                      ],
-                      __STACKTRACE__
-            end
 
-          reraise e, __STACKTRACE__
-      end
+        reraise e, __STACKTRACE__
+    end
+  end
 
-    case callback_result do
-      {actions, _state} when is_list(actions) ->
-        callback_result
+  defp validate_callback_result!({actions, _state} = callback_result, _module, _cb)
+       when is_list(actions) do
+    callback_result
+  end
 
-      _result ->
-        raise CallbackError,
-          kind: :bad_return,
-          callback: {module, callback},
-          value: callback_result
-    end
+  defp validate_callback_result!(callback_result, module, callback) do
+    raise CallbackError,
+      kind: :bad_return,
+      callback: {module, callback},
+      value: callback_result
+  end
+
+  defp report_telemetry(f, callback, args, state, context) do
+    Telemetry.span_component_callback(
+      f,
+      state.__struct__,
+      callback,
+      %{
+        callback: callback,
+        callback_args: args,
+        callback_context: context,
+        component_path: ComponentPath.get(),
+        component_type: state.module,
+        internal_state_before: state.internal_state,
+        internal_state_after: nil
+      }
+    )
   end
 
   @spec handle_callback_result(

lib/membrane/core/element/callback_context.ex

@@ -15,9 +15,11 @@ defmodule Membrane.Core.Element.CallbackContext do
       pads: state.pads_data,
       clock: state.synchronization.clock,
       parent_clock: state.synchronization.parent_clock,
+      module: state.module,
       name: state.name,
       playback: state.playback,
       resource_guard: state.resource_guard,
+      setup_incomplete?: state.setup_incomplete?,
       utility_supervisor: state.subprocess_supervisor
     })
   end

lib/membrane/core/filter_aggregator/context.ex

@@ -37,12 +37,14 @@ defmodule Membrane.Core.FilterAggregator.Context do
       |> Map.new(fn {name, description} -> {name, build_pad_data(description)} end)
 
     %{
+      module: module,
       pads: pads_data,
       clock: nil,
       name: name,
       parent_clock: nil,
       playback: :stopped,
       resource_guard: agg_ctx.resource_guard,
+      setup_incomplete?: true,
       utility_supervisor: agg_ctx.utility_supervisor
     }
   end

lib/membrane/core/pipeline/callback_context.ex

@@ -17,8 +17,10 @@ defmodule Membrane.Core.Pipeline.CallbackContext do
     |> Map.merge(%{
       clock: state.synchronization.clock_proxy,
       children: state.children,
+      module: state.module,
       playback: state.playback,
       resource_guard: state.resource_guard,
+      setup_incomplete?: state.setup_incomplete?,
       utility_supervisor: state.subprocess_supervisor
     })
   end