Dashboard (#571)

* add observability for dashboard

* fix tests

* observer metric id -> pad

* observer: cleanup metrics on unlink/death

* add observer tests

* Update lib/membrane/core/observer.ex

Co-authored-by: Łukasz Kita <lukasz.kita0@gmail.com>

* WIP metric functions

* atomic demand metric

* demand metric, input queue size metric

* fix tests

* changes for CR

* rename observer to stalker

* refactor observability setup

* remove kino from deps

* fix stalker time rounding

* fixes for CR

* pad_data.metrics -> stalker_metrics

---------

Co-authored-by: Łukasz Kita <lukasz.kita0@gmail.com>

Author: mat-hek

lib/membrane/core/bin.ex

@@ -92,7 +92,7 @@ defmodule Membrane.Core.Bin do
       log_metadata: options.log_metadata
     }
 
-    Membrane.Core.Observability.setup(observability_config)
+    Membrane.Core.Stalker.register_component(options.stalker, observability_config)
     SubprocessSupervisor.set_parent_component(options.subprocess_supervisor, observability_config)
 
     clock_proxy = Membrane.Clock.start_link(proxy: true) ~> ({:ok, pid} -> pid)
@@ -122,7 +122,8 @@ defmodule Membrane.Core.Bin do
         },
         children_log_metadata: options.log_metadata,
         subprocess_supervisor: options.subprocess_supervisor,
-        resource_guard: resource_guard
+        resource_guard: resource_guard,
+        stalker: options.stalker
       }
       |> Child.PadSpecHandler.init_pads()
 

lib/membrane/core/bin/state.ex

@@ -45,10 +45,11 @@ defmodule Membrane.Core.Bin.State do
           terminating?: boolean(),
           resource_guard: Membrane.ResourceGuard.t(),
           setup_incomplete?: boolean(),
-          handling_action?: boolean()
+          handling_action?: boolean(),
+          stalker: Membrane.Core.Stalker.t()
         }
 
-  @enforce_keys [:module, :synchronization, :subprocess_supervisor, :resource_guard]
+  @enforce_keys [:module, :synchronization, :subprocess_supervisor, :resource_guard, :stalker]
   defstruct @enforce_keys ++
               [
                 internal_state: nil,

lib/membrane/core/child/pad_model.ex

@@ -43,7 +43,8 @@ defmodule Membrane.Core.Child.PadModel do
           options: %{optional(atom) => any},
           auto_demand_size: pos_integer() | nil,
           associated_pads: [Pad.ref()] | nil,
-          sticky_events: [Membrane.Event.t()]
+          sticky_events: [Membrane.Event.t()],
+          stalker_metrics: %{atom => :atomics.atomics_ref()}
         }
 
   @type pad_data :: bin_pad_data | element_pad_data

lib/membrane/core/element.ex

@@ -18,14 +18,13 @@ defmodule Membrane.Core.Element do
   use Bunch
   use GenServer
 
-  alias Membrane.Core.Element.DemandHandler
   alias Membrane.{Clock, Core, ResourceGuard, Sync}
-
-  alias Membrane.Core.{SubprocessSupervisor, TimerController}
+  alias Membrane.Core.Child.PadSpecHandler
 
   alias Membrane.Core.Element.{
     BufferController,
     DemandController,
+    DemandHandler,
     EffectiveFlowController,
     EventController,
     LifecycleController,
@@ -34,7 +33,10 @@ defmodule Membrane.Core.Element do
     StreamFormatController
   }
 
+  alias Membrane.Core.{SubprocessSupervisor, TimerController}
+
   require Membrane.Core.Message, as: Message
+  require Membrane.Core.Stalker, as: Stalker
   require Membrane.Core.Telemetry, as: Telemetry
   require Membrane.Logger
 
@@ -98,13 +100,13 @@ defmodule Membrane.Core.Element do
 
     observability_config = %{
       name: options.name,
-      component_type: :bin,
+      component_type: :element,
       pid: self(),
       parent_path: options.parent_path,
       log_metadata: options.log_metadata
     }
 
-    Membrane.Core.Observability.setup(observability_config)
+    Membrane.Core.Stalker.register_component(options.stalker, observability_config)
     SubprocessSupervisor.set_parent_component(options.subprocess_supervisor, observability_config)
 
     {:ok, resource_guard} =
@@ -114,10 +116,46 @@ defmodule Membrane.Core.Element do
 
     ResourceGuard.register(resource_guard, fn -> Telemetry.report_terminate(:element) end)
 
+    self_pid = self()
+
+    Stalker.register_metric_function(:message_queue_length, fn ->
+      :erlang.process_info(self_pid, :message_queue_len) |> elem(1)
+    end)
+
+    Stalker.register_metric_function(:total_reductions, fn ->
+      :erlang.process_info(self_pid, :reductions) |> elem(1)
+    end)
+
     state =
-      Map.take(options, [:module, :name, :parent_clock, :sync, :parent, :subprocess_supervisor])
-      |> Map.put(:resource_guard, resource_guard)
-      |> State.new()
+      %State{
+        module: options.module,
+        type: options.module.membrane_element_type(),
+        name: options.name,
+        internal_state: nil,
+        parent_pid: options.parent,
+        supplying_demand?: false,
+        delayed_demands: MapSet.new(),
+        handle_demand_loop_counter: 0,
+        synchronization: %{
+          parent_clock: options.parent_clock,
+          timers: %{},
+          clock: nil,
+          stream_sync: options.sync,
+          latency: 0
+        },
+        initialized?: false,
+        playback: :stopped,
+        playback_queue: [],
+        resource_guard: resource_guard,
+        subprocess_supervisor: options.subprocess_supervisor,
+        terminating?: false,
+        setup_incomplete?: false,
+        effective_flow_control: :push,
+        handling_action?: false,
+        pads_to_snapshot: MapSet.new(),
+        stalker: options.stalker
+      }
+      |> PadSpecHandler.init_pads()
 
     state = LifecycleController.handle_init(options.user_options, state)
     {:ok, state, {:continue, :setup}}

lib/membrane/core/element/action_handler.ex

@@ -8,6 +8,8 @@ defmodule Membrane.Core.Element.ActionHandler do
 
   import Membrane.Pad, only: [is_pad_ref: 1]
 
+  alias Membrane.Telemetry
+
   alias Membrane.{
     ActionError,
     Buffer,
@@ -19,8 +21,6 @@ defmodule Membrane.Core.Element.ActionHandler do
     StreamFormat
   }
 
-  alias Membrane.Core.Child.PadModel
-
   alias Membrane.Core.Element.{
     DemandController,
     DemandHandler,
@@ -29,12 +29,12 @@ defmodule Membrane.Core.Element.ActionHandler do
     StreamFormatController
   }
 
-  alias Membrane.Core.{Events, Message, Telemetry, TimerController}
+  alias Membrane.Core.{Events, TimerController}
   alias Membrane.Element.Action
 
-  require Membrane.Core.Child.PadModel
-  require Membrane.Core.Message
-  require Membrane.Core.Telemetry
+  require Membrane.Core.Child.PadModel, as: PadModel
+  require Membrane.Core.Message, as: Message
+  require Membrane.Core.Telemetry, as: Telemetry
   require Membrane.Logger
 
   @impl CallbackHandler
@@ -320,10 +320,12 @@ defmodule Membrane.Core.Element.ActionHandler do
            end_of_stream?: false,
            stream_format: stream_format,
            pid: pid,
-           other_ref: other_ref
+           other_ref: other_ref,
+           stalker_metrics: stalker_metrics
          }
          when stream_format != nil <- pad_data do
       state = DemandController.decrease_demand_by_outgoing_buffers(pad_ref, buffers, state)
+      :atomics.add(stalker_metrics.total_buffers, 1, length(buffers))
       Message.send(pid, :buffer, buffers, for_pad: other_ref)
 
       PadModel.set_data!(state, pad_ref, :start_of_stream?, true)

lib/membrane/core/element/buffer_controller.ex

@@ -34,7 +34,13 @@ defmodule Membrane.Core.Element.BufferController do
   def handle_buffer(pad_ref, buffers, state) do
     withl pad: {:ok, data} <- PadModel.get_data(state, pad_ref),
           playback: %State{playback: :playing} <- state do
-      %{direction: :input, start_of_stream?: start_of_stream?} = data
+      %{
+        direction: :input,
+        start_of_stream?: start_of_stream?,
+        stalker_metrics: stalker_metrics
+      } = data
+
+      :atomics.add(stalker_metrics.total_buffers, 1, length(buffers))
 
       state =
         if start_of_stream? do
@@ -57,10 +63,11 @@ defmodule Membrane.Core.Element.BufferController do
   @spec do_handle_buffer(Pad.ref(), PadModel.pad_data(), [Buffer.t()] | Buffer.t(), State.t()) ::
           State.t()
   defp do_handle_buffer(pad_ref, %{flow_control: :auto} = data, buffers, state) do
-    %{demand: demand, demand_unit: demand_unit} = data
+    %{demand: demand, demand_unit: demand_unit, stalker_metrics: stalker_metrics} = data
     buf_size = Buffer.Metric.from_unit(demand_unit).buffers_size(buffers)
 
     state = PadModel.set_data!(state, pad_ref, :demand, demand - buf_size)
+    :atomics.put(stalker_metrics.demand, 1, demand - buf_size)
 
     state = AutoFlowUtils.auto_adjust_atomic_demand(pad_ref, state)
     exec_buffer_callback(pad_ref, buffers, state)

lib/membrane/core/element/demand_controller.ex

@@ -8,7 +8,6 @@ defmodule Membrane.Core.Element.DemandController do
   alias __MODULE__.AutoFlowUtils
 
   alias Membrane.Buffer
-  alias Membrane.Core.Child.PadModel
 
   alias Membrane.Core.Element.{
     AtomicDemand,
@@ -19,7 +18,7 @@ defmodule Membrane.Core.Element.DemandController do
 
   alias Membrane.Pad
 
-  require Membrane.Core.Child.PadModel
+  require Membrane.Core.Child.PadModel, as: PadModel
   require Membrane.Logger
 
   @spec snapshot_atomic_demand(Pad.ref(), State.t()) :: State.t()

lib/membrane/core/element/demand_controller/auto_flow_utils.ex

@@ -26,15 +26,19 @@ defmodule Membrane.Core.Element.DemandController.AutoFlowUtils do
 
   defp do_auto_adjust_atomic_demand(pad_data, state) when is_input_auto_pad_data(pad_data) do
     if increase_atomic_demand?(pad_data, state) do
-      diff = pad_data.auto_demand_size - pad_data.demand
-      :ok = AtomicDemand.increase(pad_data.atomic_demand, diff)
-
-      PadModel.set_data!(
-        state,
-        pad_data.ref,
-        :demand,
-        pad_data.auto_demand_size
-      )
+      %{
+        ref: ref,
+        auto_demand_size: auto_demand_size,
+        demand: demand,
+        atomic_demand: atomic_demand,
+        stalker_metrics: stalker_metrics
+      } = pad_data
+
+      diff = auto_demand_size - demand
+      :ok = AtomicDemand.increase(atomic_demand, diff)
+
+      :atomics.put(stalker_metrics.demand, 1, auto_demand_size)
+      PadModel.set_data!(state, ref, :demand, auto_demand_size)
     else
       state
     end

lib/membrane/core/element/input_queue.ex

@@ -14,6 +14,7 @@ defmodule Membrane.Core.Element.InputQueue do
   alias Membrane.Pad
   alias Membrane.StreamFormat
 
+  require Membrane.Core.Stalker, as: Stalker
   require Membrane.Core.Telemetry, as: Telemetry
   require Membrane.Logger
 
@@ -35,8 +36,9 @@ defmodule Membrane.Core.Element.InputQueue do
           demand: non_neg_integer(),
           inbound_metric: module(),
           outbound_metric: module(),
-          linked_output_ref: Pad.ref(),
-          atomic_demand: AtomicDemand.t()
+          pad_ref: Pad.ref(),
+          atomic_demand: AtomicDemand.t(),
+          stalker_metrics: %{atom() => any()}
         }
 
   @enforce_keys [
@@ -46,7 +48,8 @@ defmodule Membrane.Core.Element.InputQueue do
     :atomic_demand,
     :inbound_metric,
     :outbound_metric,
-    :linked_output_ref
+    :pad_ref,
+    :stalker_metrics
   ]
 
   defstruct @enforce_keys ++ [size: 0, demand: 0]
@@ -60,7 +63,7 @@ defmodule Membrane.Core.Element.InputQueue do
           inbound_demand_unit: Buffer.Metric.unit(),
           outbound_demand_unit: Buffer.Metric.unit(),
           atomic_demand: AtomicDemand.t(),
-          linked_output_ref: Pad.ref(),
+          pad_ref: Pad.ref(),
           log_tag: String.t(),
           target_size: pos_integer() | nil
         }) :: t()
@@ -69,7 +72,7 @@ defmodule Membrane.Core.Element.InputQueue do
       inbound_demand_unit: inbound_demand_unit,
       outbound_demand_unit: outbound_demand_unit,
       atomic_demand: atomic_demand,
-      linked_output_ref: linked_output_ref,
+      pad_ref: pad_ref,
       log_tag: log_tag,
       target_size: target_size
     } = config
@@ -81,14 +84,23 @@ defmodule Membrane.Core.Element.InputQueue do
 
     target_size = target_size || default_target_size
 
+    size_metric = :atomics.new(1, [])
+
+    Stalker.register_metric_function(
+      :input_queue_size,
+      fn -> :atomics.get(size_metric, 1) end,
+      pad: pad_ref
+    )
+
     %__MODULE__{
       q: @qe.new(),
       log_tag: log_tag,
       target_size: target_size,
       inbound_metric: inbound_metric,
       outbound_metric: outbound_metric,
       atomic_demand: atomic_demand,
-      linked_output_ref: linked_output_ref
+      pad_ref: pad_ref,
+      stalker_metrics: %{size: size_metric}
     }
     |> maybe_increase_atomic_demand()
   end
@@ -97,7 +109,8 @@ defmodule Membrane.Core.Element.InputQueue do
   def store(input_queue, type \\ :buffers, v)
 
   def store(input_queue, :buffers, v) when is_list(v) do
-    %__MODULE__{size: size, target_size: target_size} = input_queue
+    %__MODULE__{size: size, target_size: target_size, stalker_metrics: stalker_metrics} =
+      input_queue
 
     if size >= target_size do
       """
@@ -111,6 +124,7 @@ defmodule Membrane.Core.Element.InputQueue do
     %__MODULE__{size: size} = input_queue = do_store_buffers(input_queue, v)
 
     Telemetry.report_metric(:store, size, input_queue.log_tag)
+    :atomics.put(stalker_metrics.size, 1, size)
 
     input_queue
   end
@@ -160,7 +174,9 @@ defmodule Membrane.Core.Element.InputQueue do
     {out, input_queue} = do_take(input_queue, count)
     input_queue = maybe_increase_atomic_demand(input_queue)
 
-    Telemetry.report_metric(:take, input_queue.size, input_queue.log_tag)
+    %{size: size, stalker_metrics: stalker_metrics} = input_queue
+    Telemetry.report_metric(:take, size, input_queue.log_tag)
+    :atomics.put(stalker_metrics.size, 1, size)
 
     {out, input_queue}
   end
@@ -286,7 +302,7 @@ defmodule Membrane.Core.Element.InputQueue do
     diff = max(target_size - size - demand, div(target_size, 2))
 
     """
-    Increasing AtomicDemand linked to  #{inspect(input_queue.linked_output_ref)} by #{inspect(diff)}
+    Increasing AtomicDemand for pad  #{inspect(input_queue.pad_ref)} by #{inspect(diff)}
     """
     |> mk_log(input_queue)
     |> Membrane.Logger.debug_verbose()