From e0757fcd6a337487964e898de6dc5e552aa8fcd0 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Thu, 5 Mar 2026 11:16:28 -0600 Subject: [PATCH] feat: add LLDP topology discovery support from agent Integrates LLDP neighbor discovery from towerops-agent into Phoenix: - Updated protobuf definitions with LldpTopologyResult and LldpNeighbor messages - Regenerated Elixir protobuf code (agent.pb.ex) with LLDP_TOPOLOGY job type - Added validator for LLDP topology results with neighbor and management address validation - Added AgentChannel handler for "lldp_topology_result" events - Added Topology.upsert_neighbor/3 public API for storing agent-discovered neighbors - Stores LLDP neighbors in device_neighbors table via validated protobuf messages Agent changes were committed separately in towerops-agent repo (commit b6f60c8). Files modified: - priv/proto/agent.proto: Added LLDP message definitions - lib/towerops/proto/agent.pb.ex: Regenerated from proto file - lib/towerops/agent/validator.ex: Added validate_lldp_topology_result/1 - lib/towerops_web/channels/agent_channel.ex: Added lldp_topology_result handler - lib/towerops/topology.ex: Added upsert_neighbor/3 public wrapper --- lib/towerops/agent/validator.ex | 78 ++++++++ lib/towerops/proto/agent.pb.ex | 212 ++++++++++++--------- lib/towerops/topology.ex | 10 + lib/towerops_web/channels/agent_channel.ex | 75 ++++++++ priv/proto/agent.proto | 21 ++ 5 files changed, 303 insertions(+), 93 deletions(-) diff --git a/lib/towerops/agent/validator.ex b/lib/towerops/agent/validator.ex index 395888fc..273c209c 100644 --- a/lib/towerops/agent/validator.ex +++ b/lib/towerops/agent/validator.ex @@ -32,6 +32,8 @@ defmodule Towerops.Agent.Validator do alias Towerops.Agent.AgentHeartbeat alias Towerops.Agent.CredentialTestResult alias Towerops.Agent.InterfaceStat + alias Towerops.Agent.LldpNeighbor + alias Towerops.Agent.LldpTopologyResult alias Towerops.Agent.Metric alias Towerops.Agent.MetricBatch alias Towerops.Agent.MikrotikResult @@ -180,6 +182,23 @@ defmodule Towerops.Agent.Validator do end end + @doc """ + Validate LldpTopologyResult message. + + Checks device_id, job_id, local system name, neighbors count, and timestamp. + """ + @spec validate_lldp_topology_result(binary()) :: validation_result(LldpTopologyResult.t()) + def validate_lldp_topology_result(binary) when is_binary(binary) do + with {:ok, result} <- safe_decode(LldpTopologyResult, binary), + :ok <- validate_device_id(result.device_id), + :ok <- validate_job_id(result.job_id), + :ok <- validate_short_string(result.local_system_name, "local_system_name"), + :ok <- validate_lldp_neighbors(result.neighbors), + :ok <- validate_timestamp(result.timestamp) do + {:ok, result} + end + end + ## Private Validation Functions # Safe decode with error handling @@ -535,4 +554,63 @@ defmodule Towerops.Agent.Validator do end defp validate_mikrotik_sentences(_), do: {:error, {:invalid_sentences, "Sentences must be a list"}} + + # Validate LLDP neighbors list + defp validate_lldp_neighbors(neighbors) when is_list(neighbors) do + # Reasonably allow up to 1000 neighbors (large enterprise switches) + with :ok <- validate_neighbor_count(neighbors) do + validate_each_neighbor(neighbors) + end + end + + defp validate_lldp_neighbors(_), do: {:error, {:invalid_neighbors, "Neighbors must be a list"}} + + defp validate_neighbor_count(neighbors) when length(neighbors) > 1000 do + {:error, {:too_many_neighbors, "Neighbors list exceeds 1000 items"}} + end + + defp validate_neighbor_count(_), do: :ok + + defp validate_each_neighbor(neighbors) do + Enum.reduce_while(neighbors, :ok, fn neighbor, _acc -> + case validate_lldp_neighbor(neighbor) do + :ok -> {:cont, :ok} + error -> {:halt, error} + end + end) + end + + # Validate individual LLDP neighbor + defp validate_lldp_neighbor(%LldpNeighbor{} = neighbor) do + with :ok <- validate_short_string(neighbor.neighbor_name, "neighbor_name"), + :ok <- validate_short_string(neighbor.local_port, "local_port"), + :ok <- validate_short_string(neighbor.remote_port, "remote_port"), + :ok <- validate_short_string(neighbor.remote_port_id, "remote_port_id") do + validate_management_addresses(neighbor.management_addresses) + end + end + + # Validate management addresses list + defp validate_management_addresses(addrs) when is_list(addrs) do + with :ok <- validate_address_count(addrs) do + validate_each_address(addrs) + end + end + + defp validate_management_addresses(_), do: {:error, {:invalid_addresses, "Management addresses must be a list"}} + + defp validate_address_count(addrs) when length(addrs) > 10 do + {:error, {:too_many_addresses, "Management addresses exceeds 10 items"}} + end + + defp validate_address_count(_), do: :ok + + defp validate_each_address(addrs) do + Enum.reduce_while(addrs, :ok, fn addr, _acc -> + case validate_optional_ip(addr) do + :ok -> {:cont, :ok} + error -> {:halt, error} + end + end) + end end diff --git a/lib/towerops/proto/agent.pb.ex b/lib/towerops/proto/agent.pb.ex index 36d26213..5c8cafd6 100644 --- a/lib/towerops/proto/agent.pb.ex +++ b/lib/towerops/proto/agent.pb.ex @@ -12,6 +12,7 @@ defmodule Towerops.Agent.JobType do field :MIKROTIK, 2 field :TEST_CREDENTIALS, 3 field :PING, 4 + field :LLDP_TOPOLOGY, 5 end defmodule Towerops.Agent.QueryType do @@ -158,10 +159,7 @@ defmodule Towerops.Agent.Metric do json_name: "monitoringCheck", oneof: 0 - field :check_result, 5, - type: Towerops.Agent.CheckResult, - json_name: "checkResult", - oneof: 0 + field :check_result, 5, type: Towerops.Agent.CheckResult, json_name: "checkResult", oneof: 0 end defmodule Towerops.Agent.SensorReading do @@ -231,6 +229,110 @@ defmodule Towerops.Agent.MonitoringCheck do field :timestamp, 4, type: :int64 end +defmodule Towerops.Agent.Check do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.Check", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + oneof(:config, 0) + + field :id, 1, type: :string + field :check_type, 2, type: :string, json_name: "checkType" + field :interval_seconds, 3, type: :uint32, json_name: "intervalSeconds" + field :timeout_ms, 4, type: :uint32, json_name: "timeoutMs" + field :http, 5, type: Towerops.Agent.HttpCheckConfig, oneof: 0 + field :tcp, 6, type: Towerops.Agent.TcpCheckConfig, oneof: 0 + field :dns, 7, type: Towerops.Agent.DnsCheckConfig, oneof: 0 +end + +defmodule Towerops.Agent.HttpCheckConfig.HeadersEntry do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.HttpCheckConfig.HeadersEntry", + map: true, + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + field :key, 1, type: :string + field :value, 2, type: :string +end + +defmodule Towerops.Agent.HttpCheckConfig do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.HttpCheckConfig", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + field :url, 1, type: :string + field :method, 2, type: :string + field :expected_status, 3, type: :uint32, json_name: "expectedStatus" + field :verify_ssl, 4, type: :bool, json_name: "verifySsl" + field :headers, 5, repeated: true, type: Towerops.Agent.HttpCheckConfig.HeadersEntry, map: true + field :body, 6, type: :string + field :regex, 7, type: :string + field :follow_redirects, 8, type: :bool, json_name: "followRedirects" +end + +defmodule Towerops.Agent.TcpCheckConfig do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.TcpCheckConfig", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + field :host, 1, type: :string + field :port, 2, type: :uint32 + field :send, 3, type: :string + field :expect, 4, type: :string +end + +defmodule Towerops.Agent.DnsCheckConfig do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.DnsCheckConfig", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + field :hostname, 1, type: :string + field :server, 2, type: :string + field :record_type, 3, type: :string, json_name: "recordType" + field :expected, 4, type: :string +end + +defmodule Towerops.Agent.CheckResult do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.CheckResult", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + field :check_id, 1, type: :string, json_name: "checkId" + field :status, 2, type: :uint32 + field :output, 3, type: :string + field :response_time_ms, 4, type: :double, json_name: "responseTimeMs" + field :timestamp, 5, type: :int64 +end + +defmodule Towerops.Agent.CheckList do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.CheckList", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + field :checks, 1, repeated: true, type: Towerops.Agent.Check +end + defmodule Towerops.Agent.HeartbeatMetadata do @moduledoc false @@ -481,108 +583,32 @@ defmodule Towerops.Agent.MikrotikSentence do map: true end -# Service Checks (HTTP/TCP/DNS) - -defmodule Towerops.Agent.Check do +defmodule Towerops.Agent.LldpTopologyResult do @moduledoc false use Protobuf, - full_name: "towerops.agent.Check", + full_name: "towerops.agent.LldpTopologyResult", protoc_gen_elixir_version: "0.16.0", syntax: :proto3 - oneof(:config, 0) - - field :id, 1, type: :string - field :check_type, 2, type: :string, json_name: "checkType" - field :interval_seconds, 3, type: :uint32, json_name: "intervalSeconds" - field :timeout_ms, 4, type: :uint32, json_name: "timeoutMs" - field :http, 5, type: Towerops.Agent.HttpCheckConfig, oneof: 0 - field :tcp, 6, type: Towerops.Agent.TcpCheckConfig, oneof: 0 - field :dns, 7, type: Towerops.Agent.DnsCheckConfig, oneof: 0 -end - -defmodule Towerops.Agent.HttpCheckConfig.HeadersEntry do - @moduledoc false - - use Protobuf, - full_name: "towerops.agent.HttpCheckConfig.HeadersEntry", - map: true, - protoc_gen_elixir_version: "0.16.0", - syntax: :proto3 - - field :key, 1, type: :string - field :value, 2, type: :string -end - -defmodule Towerops.Agent.HttpCheckConfig do - @moduledoc false - - use Protobuf, - full_name: "towerops.agent.HttpCheckConfig", - protoc_gen_elixir_version: "0.16.0", - syntax: :proto3 - - field :url, 1, type: :string - field :method, 2, type: :string - field :expected_status, 3, type: :uint32, json_name: "expectedStatus" - field :verify_ssl, 4, type: :bool, json_name: "verifySsl" - field :headers, 5, repeated: true, type: Towerops.Agent.HttpCheckConfig.HeadersEntry, map: true - field :body, 6, type: :string - field :regex, 7, type: :string - field :follow_redirects, 8, type: :bool, json_name: "followRedirects" -end - -defmodule Towerops.Agent.TcpCheckConfig do - @moduledoc false - - use Protobuf, - full_name: "towerops.agent.TcpCheckConfig", - protoc_gen_elixir_version: "0.16.0", - syntax: :proto3 - - field :host, 1, type: :string - field :port, 2, type: :uint32 - field :send, 3, type: :string - field :expect, 4, type: :string -end - -defmodule Towerops.Agent.DnsCheckConfig do - @moduledoc false - - use Protobuf, - full_name: "towerops.agent.DnsCheckConfig", - protoc_gen_elixir_version: "0.16.0", - syntax: :proto3 - - field :hostname, 1, type: :string - field :server, 2, type: :string - field :record_type, 3, type: :string, json_name: "recordType" - field :expected, 4, type: :string -end - -defmodule Towerops.Agent.CheckResult do - @moduledoc false - - use Protobuf, - full_name: "towerops.agent.CheckResult", - protoc_gen_elixir_version: "0.16.0", - syntax: :proto3 - - field :check_id, 1, type: :string, json_name: "checkId" - field :status, 2, type: :uint32 - field :output, 3, type: :string - field :response_time_ms, 4, type: :double, json_name: "responseTimeMs" + field :device_id, 1, type: :string, json_name: "deviceId" + field :job_id, 2, type: :string, json_name: "jobId" + field :local_system_name, 3, type: :string, json_name: "localSystemName" + field :neighbors, 4, repeated: true, type: Towerops.Agent.LldpNeighbor field :timestamp, 5, type: :int64 end -defmodule Towerops.Agent.CheckList do +defmodule Towerops.Agent.LldpNeighbor do @moduledoc false use Protobuf, - full_name: "towerops.agent.CheckList", + full_name: "towerops.agent.LldpNeighbor", protoc_gen_elixir_version: "0.16.0", syntax: :proto3 - field :checks, 1, repeated: true, type: Towerops.Agent.Check + field :neighbor_name, 1, type: :string, json_name: "neighborName" + field :local_port, 2, type: :string, json_name: "localPort" + field :remote_port, 3, type: :string, json_name: "remotePort" + field :remote_port_id, 4, type: :string, json_name: "remotePortId" + field :management_addresses, 5, repeated: true, type: :string, json_name: "managementAddresses" end diff --git a/lib/towerops/topology.ex b/lib/towerops/topology.ex index c91a6e09..02f7f258 100644 --- a/lib/towerops/topology.ex +++ b/lib/towerops/topology.ex @@ -729,6 +729,16 @@ defmodule Towerops.Topology do {count, nil} end + @doc """ + Upserts a single LLDP neighbor record from agent discovery. + + Public wrapper for use by AgentChannel when processing LLDP topology results. + """ + @spec upsert_neighbor(String.t(), map(), DateTime.t()) :: {:ok, DeviceNeighbor.t()} | {:error, Ecto.Changeset.t()} + def upsert_neighbor(device_id, neighbor, timestamp) do + upsert_device_neighbor(device_id, neighbor, timestamp) + end + # Upserts a single device neighbor record defp upsert_device_neighbor(device_id, neighbor, now) do neighbor_device_id = find_device_by_lldp_neighbor(device_id, neighbor) diff --git a/lib/towerops_web/channels/agent_channel.ex b/lib/towerops_web/channels/agent_channel.ex index b71f88be..ba900ec4 100644 --- a/lib/towerops_web/channels/agent_channel.ex +++ b/lib/towerops_web/channels/agent_channel.ex @@ -23,6 +23,7 @@ defmodule ToweropsWeb.AgentChannel do alias Towerops.Agent.AgentJob alias Towerops.Agent.AgentJobList + alias Towerops.Agent.LldpTopologyResult alias Towerops.Agent.MikrotikCommand alias Towerops.Agent.MikrotikDevice alias Towerops.Agent.MikrotikResult @@ -40,6 +41,7 @@ defmodule ToweropsWeb.AgentChannel do alias Towerops.Snmp.AgentDiscovery alias Towerops.Snmp.Discovery alias Towerops.Snmp.SensorChangeDetector + alias Towerops.Topology require Logger @@ -627,6 +629,55 @@ defmodule ToweropsWeb.AgentChannel do end end + @spec handle_in(String.t(), %{required(String.t()) => base64_string()}, socket()) :: + {:noreply, socket()} + def handle_in("lldp_topology_result", %{"binary" => binary_b64}, socket) when is_binary(binary_b64) do + with {:ok, binary} <- safe_base64_decode(binary_b64), + {:ok, result} <- Validator.validate_lldp_topology_result(binary) do + maybe_debug_log(socket, "Received LLDP topology result from agent", + device_id: result.device_id, + job_id: result.job_id, + neighbor_count: length(result.neighbors), + local_system_name: result.local_system_name + ) + + # Store LLDP neighbors in database + case store_lldp_neighbors(result) do + {:ok, count} -> + Logger.info("Stored LLDP neighbors for device", + device_id: result.device_id, + neighbor_count: count + ) + + {:error, reason} -> + Logger.error("Failed to store LLDP neighbors", + device_id: result.device_id, + reason: inspect(reason) + ) + end + + {:noreply, socket} + else + {:error, {type, message}} -> + Logger.error("Invalid LLDP topology result from agent: #{type} - #{message}", + agent_token_id: socket.assigns.agent_token_id, + error_type: type, + error_message: message, + binary_size: byte_size(binary_b64) + ) + + {:noreply, socket} + + {:error, :base64_decode_failed} -> + Logger.error("Failed to decode LLDP topology result (invalid base64)", + agent_token_id: socket.assigns.agent_token_id, + binary_size: byte_size(binary_b64) + ) + + {:noreply, socket} + end + end + # Catch-all for unmatched events (diagnostic) def handle_in(event, payload, socket) do Logger.warning("Unhandled agent channel event", @@ -1285,6 +1336,30 @@ defmodule ToweropsWeb.AgentChannel do end end + defp store_lldp_neighbors(%LldpTopologyResult{} = result) do + now = DateTime.utc_now() + device_id = result.device_id + + # Store each neighbor using Topology module's upsert function + results = + Enum.map(result.neighbors, fn neighbor -> + # Call the private upsert function via the public Topology API + # The neighbor struct from protobuf matches the expected format + Topology.upsert_neighbor(device_id, neighbor, now) + end) + + success_count = Enum.count(results, &match?({:ok, _}, &1)) + {:ok, success_count} + rescue + e -> + Logger.error("Exception storing LLDP neighbors", + device_id: result.device_id, + error: Exception.message(e) + ) + + {:error, :storage_exception} + end + defp update_device_status_from_check(device, check_status) do new_status = if check_status in [:success, :up, :ok], do: :up, else: :down old_status = device.status diff --git a/priv/proto/agent.proto b/priv/proto/agent.proto index 47cf6b40..4d74ba78 100644 --- a/priv/proto/agent.proto +++ b/priv/proto/agent.proto @@ -2,6 +2,8 @@ syntax = "proto3"; package towerops.agent; +option go_package = "github.com/towerops-app/towerops-agent/pb"; + // Configuration received from the API message AgentConfig { string version = 1; @@ -171,6 +173,7 @@ enum JobType { MIKROTIK = 2; TEST_CREDENTIALS = 3; PING = 4; + LLDP_TOPOLOGY = 5; // Discover all LLDP neighbors for topology mapping } enum QueryType { @@ -271,3 +274,21 @@ message MikrotikResult { message MikrotikSentence { map attributes = 1; } + +// LLDP Topology Discovery Results + +message LldpTopologyResult { + string device_id = 1; + string job_id = 2; + string local_system_name = 3; + repeated LldpNeighbor neighbors = 4; + int64 timestamp = 5; +} + +message LldpNeighbor { + string neighbor_name = 1; + string local_port = 2; + string remote_port = 3; + string remote_port_id = 4; + repeated string management_addresses = 5; +}