Real-Time Monitoring

Overview

Real-Time Monitoring provides continuous observation of platform health, quality metrics, agent operations, and system performance across the entire Prismatic Platform. Built on Elixir/OTP telemetry and Phoenix LiveView, it enables immediate detection and response to any deviation from expected behavior. The monitoring infrastructure spans every layer of the platform -- from individual GenServer process health to aggregate quality scores across all 99 umbrella applications.

Unlike traditional monitoring systems that poll at fixed intervals, the Prismatic Platform leverages the BEAM virtual machine's built-in observability and the :telemetry library to emit events at the point of occurrence. This event-driven architecture means anomalies are detected within milliseconds of occurrence, not at the next polling cycle. The combination of ETS-backed in-memory aggregation and PostgreSQL time-series storage enables both real-time dashboards and historical trend analysis with zero compromise on either capability.

The monitoring system is not an add-on observability layer but a core platform capability that powers Autonomous Self-Healing, feeds Quality Gates decisions, and enforces the NO MERCY zero-tolerance quality standards in production.

Monitoring Domains

The platform monitors eight distinct operational domains, each with specialized metrics, collection frequencies, and visualization dashboards:

Architecture

Event-Driven Data Flow

The monitoring architecture follows an event-driven pipeline from source emission through aggregation to visualization:

Source Events --> Telemetry Handlers --> ETS Aggregation --> PubSub Broadcast --> LiveView Render
     |                 |                    |                     |                    |
  :telemetry      Handler Module       Time-series           Phoenix.PubSub       Real-time
  .execute()      + ETS Counters       Rollup (1m/5m/1h)    Topic Broadcast      DOM Patch

This architecture ensures that monitoring adds negligible overhead to monitored operations. The :telemetry.execute/3 call is a simple function dispatch with no serialization, no network I/O, and no blocking. Handlers run asynchronously in the calling process, with ETS providing lock-free concurrent writes for counter updates.

Telemetry Event Flow

Every monitored operation emits structured telemetry events through a consistent pipeline:

defmodule PrismaticMonitoring.EventPipeline do
  @moduledoc """
  Core monitoring event pipeline.
  Demonstrates the three-stage flow from emission to visualization.
  """

  # Stage 1: Event emission at the source (in the monitored module)
  def emit_agent_operation(agent_name, operation, duration, result) do
    :telemetry.execute(
      [:prismatic, :agent, :execute, :stop],
      %{duration: duration},
      %{agent: agent_name, operation: operation, result: result}
    )
  end

  # Stage 2: Handler aggregation into ETS
  def handle_event(
    [:prismatic, :agent, :execute, :stop],
    %{duration: duration},
    %{agent: agent_name, result: result},
    _config
  ) do
    # Atomic counter update - lock-free, concurrent-safe
    :ets.update_counter(:agent_metrics, {agent_name, :count}, 1)
    :ets.insert(:agent_metrics, {{agent_name, :last_duration}, duration})

    # Track success/failure ratio
    result_key = if result == :ok, do: :success, else: :failure
    :ets.update_counter(:agent_metrics, {agent_name, result_key}, 1)

    # Stage 3: Broadcast to LiveView dashboards
    Phoenix.PubSub.broadcast(
      Prismatic.PubSub,
      "monitoring:agents",
      {:agent_metric_updated, agent_name}
    )
  end
end

LiveView Dashboard Infrastructure

The monitoring dashboards are built with Phoenix LiveView, providing true real-time updates without polling or manual refresh. LiveView's server-rendered approach means dashboard state is always consistent with actual platform state:

defmodule PrismaticWeb.MonitoringLive.AgentDashboard do
  use PrismaticWeb, :live_view

  @impl true
  def mount(_params, _session, socket) do
    if connected?(socket) do
      Phoenix.PubSub.subscribe(Prismatic.PubSub, "monitoring:agents")
    end

    {:ok, assign(socket, agents: fetch_all_agent_stats())}
  end

  @impl true
  def handle_info({:agent_metric_updated, agent_name}, socket) do
    updated_stats = fetch_agent_stats(agent_name)
    agents = Map.put(socket.assigns.agents, agent_name, updated_stats)
    {:noreply, assign(socket, :agents, agents)}
  end

  defp fetch_all_agent_stats do
    :ets.tab2list(:agent_metrics)
    |> Enum.group_by(fn {{agent, _metric}, _val} -> agent end)
    |> Map.new(fn {agent, entries} -> {agent, build_stats(entries)} end)
  end

  defp fetch_agent_stats(agent_name) do
    total = :ets.lookup_element(:agent_metrics, {agent_name, :count}, 2)
    success = :ets.lookup_element(:agent_metrics, {agent_name, :success}, 2)
    last_duration = :ets.lookup_element(:agent_metrics, {agent_name, :last_duration}, 2)

    %{
      total_ops: total,
      success_rate: if(total > 0, do: success / total * 100, else: 0.0),
      last_duration_ms: System.convert_time_unit(last_duration, :native, :millisecond)
    }
  end

  defp build_stats(entries) do
    Enum.into(entries, %{}, fn {{_agent, metric}, value} -> {metric, value} end)
  end
end

Alert Thresholds

The monitoring system defines four severity levels with automated response escalation:

Domain-Specific Thresholds

Each monitoring domain has configurable thresholds calibrated to the platform's operational requirements:

BEAM Observability Integration

The platform extends standard BEAM virtual machine observability tools with custom instrumentation for deep runtime analysis:

defmodule PrismaticMonitoring.SystemProbe do
  @moduledoc """
  Periodic system health probe using BEAM introspection.
  Collects system-level metrics and emits telemetry events.
  """

  use GenServer

  @probe_interval :timer.seconds(10)

  def init(state) do
    :erlang.system_flag(:scheduler_wall_time, true)
    schedule_probe()
    {:ok, state}
  end

  def handle_info(:probe, state) do
    metrics = collect_system_metrics()

    :telemetry.execute(
      [:prismatic, :system, :health],
      metrics,
      %{node: Node.self(), timestamp: System.system_time(:millisecond)}
    )

    schedule_probe()
    {:noreply, %{state | last_metrics: metrics}}
  end

  defp collect_system_metrics do
    memory = :erlang.memory()

    %{
      total_memory: memory[:total],
      process_memory: memory[:processes],
      ets_memory: memory[:ets],
      binary_memory: memory[:binary],
      atom_memory: memory[:atom],
      process_count: :erlang.system_info(:process_count),
      port_count: :erlang.system_info(:port_count),
      run_queue: :erlang.statistics(:run_queue),
      scheduler_count: :erlang.system_info(:schedulers_online),
      ets_table_count: length(:ets.all())
    }
  end

  defp schedule_probe, do: Process.send_after(self(), :probe, @probe_interval)
end

EASM Real-Time Scoring

The Perimeter module's External Attack Surface Management integrates directly with the monitoring infrastructure for continuous security posture assessment:

Asset Discovery --> Vulnerability Scan --> Risk Scoring --> Rating Calculation --> Dashboard
      |                  |                    |                  |                   |
   DNS/Cert          CVE Match           Evidence-based      A-F Grade          LiveView
   Enumeration       Port Analysis       NABLA Scoring       300-900 Score      Real-time

Time-Series Aggregation

Raw telemetry events are aggregated into time-series buckets for historical analysis and trend detection. The tiered storage approach balances query performance against retention costs:

Raw Events (ETS) --> 1-min Rollup (ETS) --> 5-min Rollup (PG) --> Hourly (PG) --> Daily (PG)
     |                    |                      |                    |              |
  Instant            Counters              Aggregates          Summaries       Long-term
  Access             + Rates               + Percentiles       + Trends        Archives

Structured Logging Integration

All monitoring events emit structured logs in JSON format for correlation with external log aggregation systems:

Logger.info("Agent operation completed",
  agent: "blue-drift-detector",
  operation: :drift_scan,
  duration_ms: 142,
  result: :ok,
  findings: 0,
  telemetry_event: [:prismatic, :agent, :execute, :stop]
)

# Output (JSON structured):
# {"level":"info","msg":"Agent operation completed",
#  "agent":"blue-drift-detector","operation":"drift_scan",
#  "duration_ms":142,"result":"ok","findings":0,
#  "telemetry_event":"prismatic.agent.execute.stop",
#  "timestamp":"2026-02-15T10:30:42.123Z"}

Page Load Performance Monitoring

A dedicated monitoring subsystem enforces the platform's strict page load performance standard:

Violations at the P99 level trigger automatic investigation. Violations at the hard limit trigger merge blocking in the CI pipeline.

Integration Points

Real-Time Monitoring connects to every major platform subsystem, serving as the observability backbone:

• Powers Autonomous Self-Healing detection and response triggers
• Feeds Quality Gates validation with real-time quality metrics
• Monitored by Color Teams for security posture and drift assessment
• Built on Telemetry Integration event infrastructure
• Enforces NO MERCY zero-tolerance quality standards in production
• Supports Intelligence Synthesis with operational performance metrics
• Integrates with AIAD Standard for comprehensive agent health tracking
• Validates NABLA Axioms compliance through continuous signal monitoring
• Provides evidence for Trinity Gate structural consistency validation

Commands

Connect & Contribute

Created by Tomáš Korcak (korczis) | Open Source under GHL

• GitHub | GitLab | LinkedIn | Contact
• Developer Portal | Architecture | Meet the Creator