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
This commit is contained in:
parent
7d73193dc6
commit
e0757fcd6a
5 changed files with 303 additions and 93 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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<string, string> 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;
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue