graham/towerops
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

protobufs

Browse source
This commit is contained in:
Graham McIntire 2026-01-09 17:33:01 -06:00
parent 0f238ac33d
commit 3ef04f261f
No known key found for this signature in database
6 changed files with 356 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

173
lib/towerops/proto/agent.pb.ex Normal file
View file

@ -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

56
lib/towerops_web/controllers/api/agent_controller.ex
View file

@ -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)

1
mix.exs
View file

@ -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"},

1
mix.lock
View file

@ -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"},

83
priv/proto/agent.proto Normal file
View file

@ -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<string, string> 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;
}

43
test/towerops_web/controllers/api/agent_controller_test.exs
View file

@ -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