defmodule ToweropsWeb.Api.AgentController do @moduledoc """ API controller for remote agent communication. Provides endpoints for agents to: - Fetch polling configuration - Submit metrics - Send heartbeats """ use ToweropsWeb, :controller alias Towerops.Agent.AgentConfig alias Towerops.Agent.HeartbeatMetadata alias Towerops.Agent.HeartbeatResponse alias Towerops.Agent.MetricBatch alias Towerops.Agents alias Towerops.Snmp @doc """ GET /api/v1/agent/config Returns polling configuration for all equipment that should be polled by this agent. This includes directly assigned equipment plus equipment that inherits the agent from site or organization defaults. Supports both JSON and Protocol Buffers response formats based on Accept header. """ def get_config(conn, _params) do agent_token = conn.assigns.current_agent_token equipment_list = Agents.list_agent_polling_targets(agent_token.id) config = %{ version: "1.0", poll_interval_seconds: 60, equipment: Enum.map(equipment_list, &build_equipment_config/1) } # Check Accept header for protobuf support case get_req_header(conn, "accept") do ["application/x-protobuf" | _] -> # Convert to protobuf and send proto_config = convert_config_to_proto(config) binary = AgentConfig.encode(proto_config) conn |> put_resp_content_type("application/x-protobuf") |> send_resp(200, binary) _ -> # Default to JSON json(conn, config) end end @doc """ POST /api/v1/agent/metrics Accepts a batch of metrics from the agent and processes them asynchronously. Supports both JSON (legacy) and Protocol Buffers (efficient). """ def submit_metrics(conn, params) do agent_token = conn.assigns.current_agent_token metrics = case get_req_header(conn, "content-type") do ["application/x-protobuf" | _] -> # Decode protobuf {:ok, body, _conn} = Plug.Conn.read_body(conn) batch = MetricBatch.decode(body) convert_protobuf_metrics(batch.metrics) _ -> # JSON format (fallback) Map.get(params, "metrics", []) end _ = Task.start(fn -> process_metrics(agent_token, metrics) end) json(conn, %{status: "accepted", received: length(metrics)}) end @doc """ POST /api/v1/agent/heartbeat Updates the agent's last_seen_at timestamp and metadata. Supports both JSON and Protocol Buffers formats. """ def heartbeat(conn, params) do require Logger agent_token = conn.assigns.current_agent_token ip = get_client_ip(conn) # Log heartbeat request for debugging intermittent 502 errors Logger.debug( "Heartbeat received from agent", agent_token_id: agent_token.id, client_ip: ip, content_type: List.first(get_req_header(conn, "content-type")), request_id: conn.assigns[:request_id] ) # Parse metadata based on content type metadata = case get_req_header(conn, "content-type") do ["application/x-protobuf" | _] -> # Decode protobuf case Plug.Conn.read_body(conn) do {:ok, body, _conn} -> try do proto_metadata = HeartbeatMetadata.decode(body) %{ "version" => proto_metadata.version, "hostname" => proto_metadata.hostname, "uptime_seconds" => proto_metadata.uptime_seconds } rescue e -> require Logger Logger.error("Failed to decode protobuf heartbeat: #{inspect(e)}") # Return empty metadata on decode error %{} end {:error, reason} -> require Logger Logger.error("Failed to read heartbeat body: #{inspect(reason)}") %{} end _ -> # JSON format (fallback) %{ "version" => Map.get(params, "version"), "hostname" => Map.get(params, "hostname"), "uptime_seconds" => Map.get(params, "uptime_seconds") } end # In test environment, update synchronously to avoid DB ownership issues # In production, update asynchronously for better performance if Application.get_env(:towerops, :env) == :test do Agents.update_agent_token_heartbeat(agent_token.id, ip, metadata) else Task.start(fn -> try do Agents.update_agent_token_heartbeat(agent_token.id, ip, metadata) rescue e -> Logger.error( "Failed to update agent heartbeat", agent_token_id: agent_token.id, error: inspect(e), stacktrace: Exception.format_stacktrace(__STACKTRACE__) ) end end) end # Return response in appropriate format case get_req_header(conn, "content-type") do ["application/x-protobuf" | _] -> proto_response = %HeartbeatResponse{status: "ok"} binary = HeartbeatResponse.encode(proto_response) conn |> put_resp_content_type("application/x-protobuf") |> send_resp(200, binary) _ -> json(conn, %{status: "ok"}) end end # Private helpers defp build_equipment_config(equipment) do device = equipment.snmp_device %{ id: equipment.id, name: equipment.name, ip_address: equipment.ip_address, snmp: %{ enabled: equipment.snmp_enabled, version: equipment.snmp_version, community: equipment.snmp_community, port: equipment.snmp_port || 161 }, poll_interval_seconds: equipment.check_interval_seconds || 60, sensors: build_sensor_config(device), interfaces: build_interface_config(device) } end defp build_sensor_config(nil), do: [] defp build_sensor_config(device) do Enum.map(device.sensors, fn sensor -> %{ id: sensor.id, type: sensor.sensor_type, oid: sensor.sensor_oid, divisor: sensor.sensor_divisor, unit: sensor.sensor_unit, metadata: sensor.metadata } end) end defp build_interface_config(nil), do: [] defp build_interface_config(device) do Enum.map(device.interfaces, fn interface -> %{ id: interface.id, if_index: interface.if_index, if_name: interface.if_name } end) end defp process_metrics(_agent_token, metrics) do Enum.each(metrics, &process_single_metric/1) end defp process_single_metric(%{"type" => "sensor_reading"} = m) do Snmp.create_sensor_reading(%{ sensor_id: m["sensor_id"], value: m["value"], status: m["status"] || "ok", checked_at: parse_timestamp(m["timestamp"]) }) end defp process_single_metric(%{"type" => "interface_stat"} = m) do Snmp.create_interface_stat(%{ interface_id: m["interface_id"], if_in_octets: m["if_in_octets"], if_out_octets: m["if_out_octets"], if_in_errors: m["if_in_errors"], if_out_errors: m["if_out_errors"], if_in_discards: m["if_in_discards"], if_out_discards: m["if_out_discards"], checked_at: parse_timestamp(m["timestamp"]) }) end defp process_single_metric(%{"type" => "neighbor_discovery"} = m) do process_neighbor_discovery(m) end defp process_single_metric(_), do: :ok defp process_neighbor_discovery(m) do case Snmp.get_interface(m["interface_id"]) do nil -> :ok interface -> Snmp.upsert_neighbor(%{ interface_id: m["interface_id"], equipment_id: interface.equipment_id, protocol: m["protocol"], remote_chassis_id: m["remote_chassis_id"], remote_system_name: m["remote_system_name"], remote_system_description: m["remote_system_description"], remote_platform: m["remote_platform"], remote_port_id: m["remote_port_id"], remote_port_description: m["remote_port_description"], remote_address: m["remote_address"], remote_capabilities: m["remote_capabilities"] || [], last_discovered_at: parse_timestamp(m["timestamp"]) }) end end defp convert_protobuf_metrics(metrics) do metrics |> Enum.map(fn metric -> case metric.metric_type do {:sensor_reading, sr} -> %{ "type" => "sensor_reading", "sensor_id" => sr.sensor_id, "value" => sr.value, "status" => sr.status, "timestamp" => sr.timestamp } {:interface_stat, is} -> %{ "type" => "interface_stat", "interface_id" => is.interface_id, "if_in_octets" => is.if_in_octets, "if_out_octets" => is.if_out_octets, "if_in_errors" => is.if_in_errors, "if_out_errors" => is.if_out_errors, "if_in_discards" => is.if_in_discards, "if_out_discards" => is.if_out_discards, "timestamp" => is.timestamp } {:neighbor_discovery, nd} -> %{ "type" => "neighbor_discovery", "interface_id" => nd.interface_id, "protocol" => nd.protocol, "remote_chassis_id" => nd.remote_chassis_id, "remote_system_name" => nd.remote_system_name, "remote_system_description" => nd.remote_system_description, "remote_platform" => nd.remote_platform, "remote_port_id" => nd.remote_port_id, "remote_port_description" => nd.remote_port_description, "remote_address" => nd.remote_address, "remote_capabilities" => nd.remote_capabilities, "timestamp" => nd.timestamp } _ -> nil end end) |> Enum.reject(&is_nil/1) end defp parse_timestamp(nil), do: DateTime.truncate(DateTime.utc_now(), :second) defp parse_timestamp(ts) when is_integer(ts) do ts |> DateTime.from_unix!(:second) |> DateTime.truncate(:second) end defp parse_timestamp(ts) when is_binary(ts) do case DateTime.from_iso8601(ts) do {:ok, dt, _} -> DateTime.truncate(dt, :second) _ -> DateTime.truncate(DateTime.utc_now(), :second) end end # Convert internal config format to protobuf defp convert_config_to_proto(config) do %AgentConfig{ version: config.version, poll_interval_seconds: config.poll_interval_seconds, equipment: Enum.map(config.equipment, &convert_equipment_to_proto/1) } end defp convert_equipment_to_proto(eq) do %Towerops.Agent.Equipment{ id: eq.id, name: eq.name, ip_address: eq.ip_address, snmp: convert_snmp_to_proto(eq.snmp), poll_interval_seconds: eq.poll_interval_seconds, sensors: Enum.map(eq.sensors, &convert_sensor_to_proto/1), interfaces: Enum.map(eq.interfaces, &convert_interface_to_proto/1) } end defp convert_snmp_to_proto(snmp) do %Towerops.Agent.SnmpConfig{ enabled: snmp.enabled, version: snmp.version, community: snmp.community, port: snmp.port } end defp convert_sensor_to_proto(sensor) do %Towerops.Agent.Sensor{ id: sensor.id, type: sensor.type, oid: sensor.oid, divisor: sensor.divisor || 0.0, unit: sensor.unit || "", metadata: convert_metadata_to_proto(sensor.metadata) } end defp convert_interface_to_proto(interface) do %Towerops.Agent.Interface{ id: interface.id, if_index: interface.if_index, if_name: interface.if_name } end defp convert_metadata_to_proto(nil), do: %{} defp convert_metadata_to_proto(metadata) when is_map(metadata) do Map.new(metadata, fn {k, v} -> {to_string(k), to_string(v)} end) end # Get the real client IP, checking forwarded headers first defp get_client_ip(conn) do # Check X-Forwarded-For header first (set by load balancers/proxies) case get_req_header(conn, "x-forwarded-for") do [forwarded | _] -> # X-Forwarded-For can have multiple IPs (client, proxy1, proxy2...) # Take the first one (original client) forwarded |> String.split(",") |> List.first() |> String.trim() [] -> # Fall back to X-Real-IP case get_req_header(conn, "x-real-ip") do [real_ip | _] -> String.trim(real_ip) [] -> # Fall back to conn.remote_ip to_string(:inet_parse.ntoa(conn.remote_ip)) end end end end