diff --git a/lib/towerops/proto/agent.pb.ex b/lib/towerops/proto/agent.pb.ex new file mode 100644 index 00000000..195e666f --- /dev/null +++ b/lib/towerops/proto/agent.pb.ex @@ -0,0 +1,173 @@ +defmodule Towerops.Agent.AgentConfig do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.AgentConfig", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + field :version, 1, type: :string + field :poll_interval_seconds, 2, type: :uint32, json_name: "pollIntervalSeconds" + field :equipment, 3, repeated: true, type: Towerops.Agent.Equipment +end + +defmodule Towerops.Agent.Equipment do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.Equipment", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + field :id, 1, type: :string + field :name, 2, type: :string + field :ip_address, 3, type: :string, json_name: "ipAddress" + field :snmp, 4, type: Towerops.Agent.SnmpConfig + field :poll_interval_seconds, 5, type: :uint32, json_name: "pollIntervalSeconds" + field :sensors, 6, repeated: true, type: Towerops.Agent.Sensor + field :interfaces, 7, repeated: true, type: Towerops.Agent.Interface +end + +defmodule Towerops.Agent.SnmpConfig do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.SnmpConfig", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + field :enabled, 1, type: :bool + field :version, 2, type: :string + field :community, 3, type: :string + field :port, 4, type: :uint32 +end + +defmodule Towerops.Agent.Sensor.MetadataEntry do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.Sensor.MetadataEntry", + 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.Sensor do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.Sensor", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + field :id, 1, type: :string + field :type, 2, type: :string + field :oid, 3, type: :string + field :divisor, 4, type: :double + field :unit, 5, type: :string + field :metadata, 6, repeated: true, type: Towerops.Agent.Sensor.MetadataEntry, map: true +end + +defmodule Towerops.Agent.Interface do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.Interface", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + field :id, 1, type: :string + field :if_index, 2, type: :uint32, json_name: "ifIndex" + field :if_name, 3, type: :string, json_name: "ifName" +end + +defmodule Towerops.Agent.MetricBatch do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.MetricBatch", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + field :metrics, 1, repeated: true, type: Towerops.Agent.Metric +end + +defmodule Towerops.Agent.Metric do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.Metric", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + oneof(:metric_type, 0) + + field :sensor_reading, 1, + type: Towerops.Agent.SensorReading, + json_name: "sensorReading", + oneof: 0 + + field :interface_stat, 2, + type: Towerops.Agent.InterfaceStat, + json_name: "interfaceStat", + oneof: 0 +end + +defmodule Towerops.Agent.SensorReading do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.SensorReading", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + field :sensor_id, 1, type: :string, json_name: "sensorId" + field :value, 2, type: :double + field :status, 3, type: :string + field :timestamp, 4, type: :int64 +end + +defmodule Towerops.Agent.InterfaceStat do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.InterfaceStat", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + field :interface_id, 1, type: :string, json_name: "interfaceId" + field :if_in_octets, 2, type: :int64, json_name: "ifInOctets" + field :if_out_octets, 3, type: :int64, json_name: "ifOutOctets" + field :if_in_errors, 4, type: :int64, json_name: "ifInErrors" + field :if_out_errors, 5, type: :int64, json_name: "ifOutErrors" + field :if_in_discards, 6, type: :int64, json_name: "ifInDiscards" + field :if_out_discards, 7, type: :int64, json_name: "ifOutDiscards" + field :timestamp, 8, type: :int64 +end + +defmodule Towerops.Agent.HeartbeatMetadata do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.HeartbeatMetadata", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + field :version, 1, type: :string + field :hostname, 2, type: :string + field :uptime_seconds, 3, type: :uint64, json_name: "uptimeSeconds" +end + +defmodule Towerops.Agent.HeartbeatResponse do + @moduledoc false + + use Protobuf, + full_name: "towerops.agent.HeartbeatResponse", + protoc_gen_elixir_version: "0.16.0", + syntax: :proto3 + + field :status, 1, type: :string +end diff --git a/lib/towerops_web/controllers/api/agent_controller.ex b/lib/towerops_web/controllers/api/agent_controller.ex index 78a1ec13..67274106 100644 --- a/lib/towerops_web/controllers/api/agent_controller.ex +++ b/lib/towerops_web/controllers/api/agent_controller.ex @@ -10,6 +10,7 @@ defmodule ToweropsWeb.Api.AgentController do use ToweropsWeb, :controller + alias Towerops.Agent.MetricBatch alias Towerops.Agents alias Towerops.Snmp @@ -35,10 +36,24 @@ defmodule ToweropsWeb.Api.AgentController do 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, %{"metrics" => metrics}) when is_list(metrics) do + 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) @@ -148,8 +163,47 @@ defmodule ToweropsWeb.Api.AgentController do 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 + } + + _ -> + 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) diff --git a/mix.exs b/mix.exs index f173072c..54eaa8fa 100644 --- a/mix.exs +++ b/mix.exs @@ -59,6 +59,7 @@ defmodule Towerops.MixProject do {:ex_aws, "~> 2.5"}, {:ex_aws_ses, "~> 2.4"}, {:cbor, "~> 1.0"}, + {:protobuf, "~> 0.12"}, {:req, "~> 0.5"}, {:snmpkit, "~> 1.3"}, {:yaml_elixir, "~> 2.9"}, diff --git a/mix.lock b/mix.lock index 61811fb0..49f1e416 100644 --- a/mix.lock +++ b/mix.lock @@ -48,6 +48,7 @@ "plug": {:hex, :plug, "1.19.1", "09bac17ae7a001a68ae393658aa23c7e38782be5c5c00c80be82901262c394c0", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "560a0017a8f6d5d30146916862aaf9300b7280063651dd7e532b8be168511e62"}, "plug_crypto": {:hex, :plug_crypto, "2.1.1", "19bda8184399cb24afa10be734f84a16ea0a2bc65054e23a62bb10f06bc89491", [:mix], [], "hexpm", "6470bce6ffe41c8bd497612ffde1a7e4af67f36a15eea5f921af71cf3e11247c"}, "postgrex": {:hex, :postgrex, "0.21.1", "2c5cc830ec11e7a0067dd4d623c049b3ef807e9507a424985b8dcf921224cd88", [:mix], [{:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "27d8d21c103c3cc68851b533ff99eef353e6a0ff98dc444ea751de43eb48bdac"}, + "protobuf": {:hex, :protobuf, "0.16.0", "d1878725105d49162977cf3408ccc3eac4f3532e26e5a9e250f2c624175d10f6", [:mix], [{:jason, "~> 1.2", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm", "f0d0d3edd8768130f24cc2cfc41320637d32c80110e80d13f160fa699102c828"}, "req": {:hex, :req, "0.5.17", "0096ddd5b0ed6f576a03dde4b158a0c727215b15d2795e59e0916c6971066ede", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:ezstd, "~> 1.0", [hex: :ezstd, repo: "hexpm", optional: true]}, {:finch, "~> 0.17", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "0b8bc6ffdfebbc07968e59d3ff96d52f2202d0536f10fef4dc11dc02a2a43e39"}, "snmpkit": {:hex, :snmpkit, "1.3.19", "b09c38cea619a0a67d4fb0f3bfa3b9b749d42c400c2e7bd9446fef82f0d728a7", [:mix], [{:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: true]}, {:yaml_elixir, "~> 2.9", [hex: :yaml_elixir, repo: "hexpm", optional: true]}], "hexpm", "b05c1f3911204c4d781234187a6f1350c369fcffe2058a85b49735b1259eb973"}, "sobelow": {:hex, :sobelow, "0.14.1", "2f81e8632f15574cba2402bcddff5497b413c01e6f094bc0ab94e83c2f74db81", [:mix], [{:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "8fac9a2bd90fdc4b15d6fca6e1608efb7f7c600fa75800813b794ee9364c87f2"}, diff --git a/priv/proto/agent.proto b/priv/proto/agent.proto new file mode 100644 index 00000000..bacdbce5 --- /dev/null +++ b/priv/proto/agent.proto @@ -0,0 +1,83 @@ +syntax = "proto3"; + +package towerops.agent; + +// Configuration received from the API +message AgentConfig { + string version = 1; + uint32 poll_interval_seconds = 2; + repeated Equipment equipment = 3; +} + +message Equipment { + string id = 1; + string name = 2; + string ip_address = 3; + SnmpConfig snmp = 4; + uint32 poll_interval_seconds = 5; + repeated Sensor sensors = 6; + repeated Interface interfaces = 7; +} + +message SnmpConfig { + bool enabled = 1; + string version = 2; + string community = 3; + uint32 port = 4; +} + +message Sensor { + string id = 1; + string type = 2; + string oid = 3; + double divisor = 4; + string unit = 5; + map metadata = 6; +} + +message Interface { + string id = 1; + uint32 if_index = 2; + string if_name = 3; +} + +// Metrics submitted to the API +message MetricBatch { + repeated Metric metrics = 1; +} + +message Metric { + oneof metric_type { + SensorReading sensor_reading = 1; + InterfaceStat interface_stat = 2; + } +} + +message SensorReading { + string sensor_id = 1; + double value = 2; + string status = 3; + int64 timestamp = 4; // Unix timestamp in seconds +} + +message InterfaceStat { + string interface_id = 1; + int64 if_in_octets = 2; + int64 if_out_octets = 3; + int64 if_in_errors = 4; + int64 if_out_errors = 5; + int64 if_in_discards = 6; + int64 if_out_discards = 7; + int64 timestamp = 8; // Unix timestamp in seconds +} + +// Heartbeat metadata +message HeartbeatMetadata { + string version = 1; + string hostname = 2; + uint64 uptime_seconds = 3; +} + +message HeartbeatResponse { + string status = 1; +} diff --git a/test/towerops_web/controllers/api/agent_controller_test.exs b/test/towerops_web/controllers/api/agent_controller_test.exs index 86c90e0a..73814c7e 100644 --- a/test/towerops_web/controllers/api/agent_controller_test.exs +++ b/test/towerops_web/controllers/api/agent_controller_test.exs @@ -3,6 +3,10 @@ defmodule ToweropsWeb.Api.AgentControllerTest do import Towerops.AccountsFixtures + alias Towerops.Agent.InterfaceStat + alias Towerops.Agent.Metric + alias Towerops.Agent.MetricBatch + alias Towerops.Agent.SensorReading alias Towerops.Agents alias Towerops.Repo alias Towerops.Snmp.Device @@ -164,6 +168,45 @@ defmodule ToweropsWeb.Api.AgentControllerTest do assert %{"status" => "accepted", "received" => 2} = json_response(conn, 200) end + + test "accepts protobuf metrics", %{conn: conn, token: token} do + # Create protobuf metrics + sensor_reading = %SensorReading{ + sensor_id: Ecto.UUID.generate(), + value: 45.5, + status: "ok", + timestamp: DateTime.to_unix(DateTime.utc_now()) + } + + interface_stat = %InterfaceStat{ + interface_id: Ecto.UUID.generate(), + if_in_octets: 1_234_567, + if_out_octets: 9_876_543, + if_in_errors: 0, + if_out_errors: 0, + if_in_discards: 0, + if_out_discards: 0, + timestamp: DateTime.to_unix(DateTime.utc_now()) + } + + batch = %MetricBatch{ + metrics: [ + %Metric{metric_type: {:sensor_reading, sensor_reading}}, + %Metric{metric_type: {:interface_stat, interface_stat}} + ] + } + + # Encode to protobuf binary + encoded = MetricBatch.encode(batch) + + conn = + conn + |> put_req_header("authorization", "Bearer #{token}") + |> put_req_header("content-type", "application/x-protobuf") + |> post(~p"/api/v1/agent/metrics", encoded) + + assert %{"status" => "accepted", "received" => 2} = json_response(conn, 200) + end end describe "POST /api/v1/agent/heartbeat" do