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

snmpkit overhaul and etcd

Browse source
This commit is contained in:
Graham McIntire 2026-01-23 16:23:57 -06:00
parent f59cdbbd7a
commit 2b78b1a2d3
No known key found for this signature in database
25 changed files with 1070 additions and 726 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
k8s/deployment.yaml
View file

@ -11,7 +11,7 @@ spec:
rollingUpdate:
maxSurge: 1 # Allow 1 extra pod during rollout
maxUnavailable: 0 # Keep all pods running during rollout
minReadySeconds: 10 # Wait 10s after pod is ready before continuing
minReadySeconds: 20 # Wait 20s after pod is ready for cluster sync before continuing
selector:
matchLabels:
app: towerops
@ -22,7 +22,7 @@ spec:
spec:
# Use system-cluster-critical priority to ensure towerops starts after CNI is ready
priorityClassName: system-cluster-critical
terminationGracePeriodSeconds: 30 # Allow 30s for graceful shutdown
terminationGracePeriodSeconds: 60 # Allow 60s for Horde graceful shutdown (30s) + drain time
imagePullSecrets:
- name: gitlab-registry
securityContext:
@ -51,6 +51,17 @@ spec:
name: epmd
- containerPort: 9000
name: dist
lifecycle:
preStop:
exec:
command:
- /bin/sh
- -c
- |
# Stop accepting new connections
# Give etcd coordinator time to release locks (watches trigger instant failover)
# This sleep allows graceful shutdown before SIGTERM is sent
sleep 10
env:
- name: POD_IP
valueFrom:
@ -66,6 +77,8 @@ spec:
fieldPath: metadata.namespace
- name: DEPLOY_TIMESTAMP
value: "1970-01-01T00:00:00Z" # Placeholder, updated by GitLab CI on deploy
- name: ETCD_ENABLED
value: "true"
- name: RELEASE_DISTRIBUTION
value: "name"
- name: RELEASE_NODE
@ -125,4 +138,4 @@ spec:
periodSeconds: 3
timeoutSeconds: 2
successThreshold: 1
failureThreshold: 2
failureThreshold: 1 # Fast deregistration from service during shutdown

102
k8s/etcd-statefulset.yaml Normal file
View file

@ -0,0 +1,102 @@
---
# etcd StatefulSet for distributed coordination
# Used by Towerops for distributed process locking
apiVersion: v1
kind: Service
metadata:
name: etcd
namespace: towerops
labels:
app: etcd
spec:
ports:
- port: 2379
name: client
- port: 2380
name: peer
clusterIP: None
selector:
app: etcd
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: etcd
namespace: towerops
spec:
serviceName: etcd
replicas: 1
podManagementPolicy: Parallel
selector:
matchLabels:
app: etcd
template:
metadata:
labels:
app: etcd
spec:
containers:
- name: etcd
image: gcr.io/etcd-development/etcd:v3.5.11
ports:
- containerPort: 2379
name: client
- containerPort: 2380
name: peer
env:
- name: ETCD_NAME
valueFrom:
fieldRef:
fieldPath: metadata.name
- name: ETCD_INITIAL_CLUSTER
value: "etcd-0=http://etcd-0.etcd.towerops.svc.cluster.local:2380"
- name: ETCD_INITIAL_CLUSTER_STATE
value: "new"
- name: ETCD_INITIAL_CLUSTER_TOKEN
value: "towerops-etcd"
- name: ETCD_LISTEN_CLIENT_URLS
value: "http://0.0.0.0:2379"
- name: ETCD_ADVERTISE_CLIENT_URLS
value: "http://$(ETCD_NAME).etcd.towerops.svc.cluster.local:2379"
- name: ETCD_LISTEN_PEER_URLS
value: "http://0.0.0.0:2380"
- name: ETCD_INITIAL_ADVERTISE_PEER_URLS
value: "http://$(ETCD_NAME).etcd.towerops.svc.cluster.local:2380"
- name: ETCD_AUTO_COMPACTION_RETENTION
value: "1" # Keep 1 hour of history
- name: ETCD_QUOTA_BACKEND_BYTES
value: "8589934592" # 8GB
resources:
requests:
memory: "128Mi"
cpu: "100m"
limits:
memory: "512Mi"
cpu: "500m"
volumeMounts:
- name: etcd-data
mountPath: /var/run/etcd
livenessProbe:
httpGet:
path: /health
port: 2379
initialDelaySeconds: 30
periodSeconds: 10
timeoutSeconds: 5
failureThreshold: 5
readinessProbe:
httpGet:
path: /health
port: 2379
initialDelaySeconds: 20
periodSeconds: 10
timeoutSeconds: 5
failureThreshold: 10
volumeClaimTemplates:
- metadata:
name: etcd-data
spec:
accessModes: ["ReadWriteOnce"]
resources:
requests:
storage: 10Gi

1
k8s/kustomization.yaml
View file

@ -9,6 +9,7 @@ resources:
- service-headless.yaml
- valkey-statefulset.yaml
- valkey-service.yaml
- etcd-statefulset.yaml
- certificate.yaml
- ingressroute.yaml
- poddisruptionbudget.yaml

122
lib/snmpkit/snmp_lib/manager.ex
View file

@ -334,19 +334,15 @@ defmodule SnmpKit.SnmpLib.Manager do
opts = merge_default_opts(opts)
normalized_oids = Enum.map(oids, &normalize_oid/1)
case create_socket(opts) do
{:ok, socket} ->
results = get_multi_with_socket(socket, host, normalized_oids, opts)
:ok = close_socket(socket)
with {:ok, socket} <- create_socket(opts) do
results = get_multi_with_socket(socket, host, normalized_oids, opts)
:ok = close_socket(socket)
# Check if all operations failed due to network issues
case check_for_global_failure(results) do
{:global_failure, reason} -> {:error, reason}
:mixed_results -> {:ok, results}
end
{:error, reason} ->
{:error, reason}
# Check if all operations failed due to network issues
case check_for_global_failure(results) do
{:global_failure, reason} -> {:error, reason}
:mixed_results -> {:ok, results}
end
end
end
end
@ -475,49 +471,39 @@ defmodule SnmpKit.SnmpLib.Manager do
timeout = opts[:timeout] || @default_timeout
port_option = opts[:port] || @default_port
# Parse target to handle both host:port strings and :port option
{parsed_host, parsed_port} =
case Utils.parse_target(host) do
{:ok, %{host: h, port: p}} ->
# Check if host contained a port specification
if host_contains_port?(host) do
# Host:port format - use parsed port (backward compatibility)
{h, p}
else
# Host without port - use :port option
{h, port_option}
end
{:error, _} ->
# Parse failed - use original host and :port option
{host, port_option}
end
{parsed_host, parsed_port} = parse_host_and_port(host, port_option)
message = PDU.build_message(pdu, community, version)
Logger.debug("Built SNMP message: #{inspect(message)}")
with {:ok, packet} <- encode_pdu_message(message),
{:ok, response_packet} <- send_and_receive(socket, parsed_host, parsed_port, packet, timeout) do
decode_response_message(response_packet)
end
end
defp parse_host_and_port(host, port_option) do
case Utils.parse_target(host) do
{:ok, %{host: h, port: p}} ->
# Check if host contained a port specification
if host_contains_port?(host) do
# Host:port format - use parsed port (backward compatibility)
{h, p}
else
# Host without port - use :port option
{h, port_option}
end
{:error, _} ->
# Parse failed - use original host and :port option
{host, port_option}
end
end
defp encode_pdu_message(message) do
case PDU.encode_message(message) do
{:ok, packet} ->
Logger.debug("Encoded PDU packet for transmission")
case send_and_receive(socket, parsed_host, parsed_port, packet, timeout) do
{:ok, response_packet} ->
Logger.debug("Received response packet from network")
case PDU.decode_message(response_packet) do
{:ok, response_message} ->
Logger.debug("Decoded response message: #{inspect(response_message)}")
{:ok, response_message}
{:error, decode_reason} = decode_error ->
Logger.error("PDU decode failed: #{inspect(decode_reason)}")
decode_error
end
{:error, network_reason} = network_error ->
Logger.error("Network operation failed: #{inspect(network_reason)}")
network_error
end
{:ok, packet}
{:error, encode_reason} = encode_error ->
Logger.error("PDU encode failed: #{inspect(encode_reason)}")
@ -525,6 +511,20 @@ defmodule SnmpKit.SnmpLib.Manager do
end
end
defp decode_response_message(response_packet) do
Logger.debug("Received response packet from network")
case PDU.decode_message(response_packet) do
{:ok, response_message} ->
Logger.debug("Decoded response message: #{inspect(response_message)}")
{:ok, response_message}
{:error, decode_reason} = decode_error ->
Logger.error("PDU decode failed: #{inspect(decode_reason)}")
decode_error
end
end
defp send_and_receive(socket, host, port, packet, timeout) do
# Normal send-and-receive flow
case Transport.send_packet(socket, host, port, packet) do
@ -554,19 +554,23 @@ defmodule SnmpKit.SnmpLib.Manager do
# Multi-get implementation with connection reuse
defp get_multi_with_socket(socket, host, oids, opts) do
Enum.map(oids, fn oid ->
case perform_get_operation(socket, host, oid, opts) do
{:ok, response} ->
case extract_get_result_with_oid(response) do
{:ok, {oid, type, value}} -> {oid, type, value}
{:error, reason} -> {oid, {:error, reason}}
end
{:error, reason} ->
{oid, {:error, reason}}
end
perform_single_get(socket, host, oid, opts)
end)
end
defp perform_single_get(socket, host, oid, opts) do
case perform_get_operation(socket, host, oid, opts) do
{:ok, response} ->
case extract_get_result_with_oid(response) do
{:ok, {oid, type, value}} -> {oid, type, value}
{:error, reason} -> {oid, {:error, reason}}
end
{:error, reason} ->
{oid, {:error, reason}}
end
end
# Result extraction
defp extract_get_result(%{pdu: %{error_status: error_status}} = response) when error_status != 0 do
Logger.debug("Extracting error result - error_status: #{error_status}")

135
lib/snmpkit/snmp_lib/pdu/decoder.ex
View file

@ -126,26 +126,18 @@ defmodule SnmpKit.SnmpLib.PDU.Decoder do
# Comprehensive decoding implementation (from SnmpSim)
defp decode_snmp_message_comprehensive(<<0x30, rest::binary>>) do
case parse_ber_length(rest) do
{:ok, {_content_length, content}} ->
case parse_snmp_message_fields(content) do
{:ok, {version, community, pdu_data}} ->
case parse_pdu_comprehensive(pdu_data) do
{:ok, pdu} ->
{:ok,
%{
version: version,
community: community,
pdu: pdu
}}
{:error, reason} ->
{:error, {:pdu_parse_error, reason}}
end
{:error, reason} ->
{:error, {:message_parse_error, reason}}
end
with {:ok, {_content_length, content}} <- parse_ber_length(rest),
{:ok, {version, community, pdu_data}} <- parse_snmp_message_fields(content),
{:ok, pdu} <- parse_pdu_comprehensive(pdu_data) do
{:ok,
%{
version: version,
community: community,
pdu: pdu
}}
else
{:error, reason} when is_tuple(reason) ->
{:error, reason}
{:error, reason} ->
{:error, {:message_parse_error, reason}}
@ -259,64 +251,71 @@ defmodule SnmpKit.SnmpLib.PDU.Decoder do
defp parse_ber_length_and_remaining(_), do: {:error, :invalid_length_format}
defp parse_pdu_comprehensive(<<tag, rest::binary>>) when tag in [0xA0, 0xA1, 0xA2, 0xA3, 0xA5] do
pdu_type =
case tag do
0xA0 -> :get_request
0xA1 -> :get_next_request
0xA2 -> :get_response
0xA3 -> :set_request
0xA5 -> :get_bulk_request
end
pdu_type = tag_to_pdu_type(tag)
case parse_ber_length_and_remaining(rest) do
{:ok, {_length, pdu_content, _remaining}} ->
case pdu_type do
:get_bulk_request ->
case parse_bulk_pdu_fields(pdu_content) do
{:ok, {request_id, non_repeaters, max_repetitions, varbinds}} ->
{:ok,
%{
type: pdu_type,
request_id: request_id,
non_repeaters: non_repeaters,
max_repetitions: max_repetitions,
varbinds: varbinds
}}
{:error, _reason} ->
{:ok, %{type: pdu_type, varbinds: [], non_repeaters: 0, max_repetitions: 0}}
end
_ ->
case parse_standard_pdu_fields(pdu_content) do
{:ok, {request_id, error_status, error_index, varbinds}} ->
{:ok,
%{
type: pdu_type,
request_id: request_id,
error_status: error_status,
error_index: error_index,
varbinds: varbinds
}}
{:error, _reason} ->
{:ok, %{type: pdu_type, varbinds: [], error_status: 0, error_index: 0}}
end
end
build_pdu_from_content(pdu_type, pdu_content)
{:error, _reason} ->
case pdu_type do
:get_bulk_request ->
{:ok, %{type: pdu_type, varbinds: [], non_repeaters: 0, max_repetitions: 0}}
_ ->
{:ok, %{type: pdu_type, varbinds: [], error_status: 0, error_index: 0}}
end
build_empty_pdu(pdu_type)
end
end
defp parse_pdu_comprehensive(_), do: {:error, :invalid_pdu}
defp tag_to_pdu_type(tag) do
case tag do
0xA0 -> :get_request
0xA1 -> :get_next_request
0xA2 -> :get_response
0xA3 -> :set_request
0xA5 -> :get_bulk_request
end
end
defp build_pdu_from_content(:get_bulk_request, pdu_content) do
case parse_bulk_pdu_fields(pdu_content) do
{:ok, {request_id, non_repeaters, max_repetitions, varbinds}} ->
{:ok,
%{
type: :get_bulk_request,
request_id: request_id,
non_repeaters: non_repeaters,
max_repetitions: max_repetitions,
varbinds: varbinds
}}
{:error, _reason} ->
{:ok, %{type: :get_bulk_request, varbinds: [], non_repeaters: 0, max_repetitions: 0}}
end
end
defp build_pdu_from_content(pdu_type, pdu_content) do
case parse_standard_pdu_fields(pdu_content) do
{:ok, {request_id, error_status, error_index, varbinds}} ->
{:ok,
%{
type: pdu_type,
request_id: request_id,
error_status: error_status,
error_index: error_index,
varbinds: varbinds
}}
{:error, _reason} ->
{:ok, %{type: pdu_type, varbinds: [], error_status: 0, error_index: 0}}
end
end
defp build_empty_pdu(:get_bulk_request) do
{:ok, %{type: :get_bulk_request, varbinds: [], non_repeaters: 0, max_repetitions: 0}}
end
defp build_empty_pdu(pdu_type) do
{:ok, %{type: pdu_type, varbinds: [], error_status: 0, error_index: 0}}
end
defp parse_standard_pdu_fields(data) do
with {:ok, {request_id, rest1}} <- parse_integer(data),
{:ok, {error_status, rest2}} <- parse_integer(rest1),

137
lib/snmpkit/snmp_lib/transport.ex
View file

@ -69,26 +69,26 @@ defmodule SnmpKit.SnmpLib.Transport do
@spec create_socket(binary() | :inet.socket_address(), port_number(), socket_options()) ::
{:ok, socket()} | {:error, atom()}
def create_socket(bind_address, port, options \\ []) do
case resolve_address(bind_address) do
{:ok, resolved_address} ->
if validate_port(port) do
merged_options = Keyword.merge(@default_socket_options, options)
with {:ok, resolved_address} <- resolve_address(bind_address),
:ok <- validate_port_result(port) do
open_udp_socket(resolved_address, port, options)
end
end
case :gen_udp.open(port, [:binary, {:ip, resolved_address} | merged_options]) do
{:ok, socket} ->
Logger.debug("Created UDP socket bound to #{format_endpoint(resolved_address, port)}")
defp validate_port_result(port) do
if validate_port(port), do: :ok, else: {:error, :invalid_port}
end
{:ok, socket}
defp open_udp_socket(resolved_address, port, options) do
merged_options = Keyword.merge(@default_socket_options, options)
{:error, reason} ->
Logger.error("Failed to create UDP socket: #{inspect(reason)}")
{:error, reason}
end
else
{:error, :invalid_port}
end
case :gen_udp.open(port, [:binary, {:ip, resolved_address} | merged_options]) do
{:ok, socket} ->
Logger.debug("Created UDP socket bound to #{format_endpoint(resolved_address, port)}")
{:ok, socket}
{:error, reason} ->
Logger.error("Failed to create UDP socket: #{inspect(reason)}")
{:error, reason}
end
end
@ -167,29 +167,18 @@ defmodule SnmpKit.SnmpLib.Transport do
# Allow port 0 for ephemeral ports in testing, but validate others
if port == 0 do
case resolve_address(bind_address) do
{:ok, resolved_address} ->
merged_options = Keyword.merge(@default_socket_options, server_options)
case :gen_udp.open(port, [:binary, {:ip, resolved_address} | merged_options]) do
{:ok, socket} ->
Logger.debug("Created UDP socket bound to #{format_endpoint(resolved_address, port)}")
{:ok, socket}
{:error, reason} ->
Logger.error("Failed to create UDP socket: #{inspect(reason)}")
{:error, reason}
end
{:error, reason} ->
{:error, reason}
end
create_ephemeral_server_socket(bind_address, port, server_options)
else
create_socket(bind_address, port, server_options)
end
end
defp create_ephemeral_server_socket(bind_address, port, server_options) do
with {:ok, resolved_address} <- resolve_address(bind_address) do
open_udp_socket(resolved_address, port, server_options)
end
end
@doc """
Sends a packet to the specified destination.
@ -371,58 +360,55 @@ defmodule SnmpKit.SnmpLib.Transport do
"""
@spec resolve_address(address()) :: {:ok, :inet.socket_address()} | {:error, atom()}
def resolve_address(address) when is_tuple(address) do
case address do
{a, b, c, d}
when is_integer(a) and is_integer(b) and is_integer(c) and is_integer(d) and
a >= 0 and a <= 255 and b >= 0 and b <= 255 and
c >= 0 and c <= 255 and d >= 0 and d <= 255 ->
{:ok, address}
_ ->
{:error, :invalid_ip_tuple}
if valid_ipv4_tuple?(address) do
{:ok, address}
else
{:error, :invalid_ip_tuple}
end
end
def resolve_address(address) when is_binary(address) do
case :inet.parse_address(String.to_charlist(address)) do
charlist = String.to_charlist(address)
case :inet.parse_address(charlist) do
{:ok, ip_tuple} ->
{:ok, ip_tuple}
{:error, :einval} ->
# Try hostname resolution
case :inet.gethostbyname(String.to_charlist(address)) do
{:ok, {:hostent, _name, _aliases, :inet, 4, [ip_tuple | _]}} ->
{:ok, ip_tuple}
{:error, reason} ->
Logger.error("Failed to resolve hostname #{address}: #{inspect(reason)}")
{:error, :hostname_resolution_failed}
end
resolve_hostname(charlist, address)
end
end
def resolve_address(address) when is_list(address) do
# Handle charlist input (from mix run -e single quotes)
case :inet.parse_address(address) do
{:ok, ip_tuple} ->
{:ok, ip_tuple}
{:error, :einval} ->
# Try hostname resolution
case :inet.gethostbyname(address) do
{:ok, {:hostent, _name, _aliases, :inet, 4, [ip_tuple | _]}} ->
{:ok, ip_tuple}
{:error, reason} ->
address_str = List.to_string(address)
Logger.error("Failed to resolve hostname #{address_str}: #{inspect(reason)}")
{:error, :hostname_resolution_failed}
end
address_str = List.to_string(address)
resolve_hostname(address, address_str)
end
end
def resolve_address(_), do: {:error, :invalid_address_format}
defp valid_ipv4_tuple?({a, b, c, d}) when is_integer(a) and is_integer(b) and is_integer(c) and is_integer(d) do
a in 0..255 and b in 0..255 and c in 0..255 and d in 0..255
end
defp valid_ipv4_tuple?(_), do: false
defp resolve_hostname(charlist, display_address) do
case :inet.gethostbyname(charlist) do
{:ok, {:hostent, _name, _aliases, :inet, 4, [ip_tuple | _]}} ->
{:ok, ip_tuple}
{:error, reason} ->
Logger.error("Failed to resolve hostname #{display_address}: #{inspect(reason)}")
{:error, :hostname_resolution_failed}
end
end
@doc """
Validates a port number.
@ -666,13 +652,8 @@ defmodule SnmpKit.SnmpLib.Transport do
remaining_timeout = min(per_recv_timeout, timeout - elapsed)
case receive_packet(socket, remaining_timeout) do
{:ok, {data, from_addr, from_port} = packet_info} ->
if filter_fn.(packet_info) do
{:ok, {data, from_addr, from_port}}
else
# Packet didn't match filter, continue waiting
receive_packet_filtered_loop(socket, filter_fn, timeout, per_recv_timeout, start_time)
end
{:ok, packet_info} ->
handle_received_packet(socket, filter_fn, timeout, per_recv_timeout, start_time, packet_info)
{:error, :timeout} ->
# Continue waiting if we haven't exceeded total timeout
@ -684,6 +665,22 @@ defmodule SnmpKit.SnmpLib.Transport do
end
end
defp handle_received_packet(
socket,
filter_fn,
timeout,
per_recv_timeout,
start_time,
{data, from_addr, from_port} = packet_info
) do
if filter_fn.(packet_info) do
{:ok, {data, from_addr, from_port}}
else
# Packet didn't match filter, continue waiting
receive_packet_filtered_loop(socket, filter_fn, timeout, per_recv_timeout, start_time)
end
end
defp collect_socket_stats(socket) do
stats = %{
socket_info: :inet.info(socket),

87
lib/snmpkit/snmp_lib/types.ex
View file

@ -198,14 +198,21 @@ defmodule SnmpKit.SnmpLib.Types do
def infer_type(nil), do: :null
def infer_type({a, b, c, d}) when is_integer(a) and is_integer(b) and is_integer(c) and is_integer(d) do
if a >= 0 and a <= 255 and b >= 0 and b <= 255 and c >= 0 and c <= 255 and d >= 0 and d <= 255 do
infer_tuple_type(a, b, c, d)
end
def infer_type(_), do: :unknown
defp infer_tuple_type(a, b, c, d) do
if valid_ip_octet?(a) and valid_ip_octet?(b) and valid_ip_octet?(c) and valid_ip_octet?(d) do
:ip_address
else
:unknown
end
end
def infer_type(_), do: :unknown
defp valid_ip_octet?(n) when is_integer(n), do: n >= 0 and n <= 255
defp valid_ip_octet?(_), do: false
@doc """
Decodes an SNMP typed value back to a native Elixir value.
@ -273,24 +280,27 @@ defmodule SnmpKit.SnmpLib.Types do
"""
@spec parse_ip_address(binary()) :: {:ok, {0..255, 0..255, 0..255, 0..255}} | {:error, atom()}
def parse_ip_address(ip_string) when is_binary(ip_string) do
case :inet.parse_address(String.to_charlist(ip_string)) do
{:ok, {a, b, c, d}}
when a >= 0 and a <= 255 and b >= 0 and b <= 255 and
c >= 0 and c <= 255 and d >= 0 and d <= 255 ->
{:ok, {a, b, c, d}}
{:ok, _} ->
{:error, :not_ipv4}
{:error, _} ->
{:error, :invalid_format}
end
ip_string
|> String.to_charlist()
|> :inet.parse_address()
|> validate_parsed_ip()
rescue
_ -> {:error, :invalid_format}
end
def parse_ip_address(_), do: {:error, :invalid_input}
defp validate_parsed_ip({:ok, {a, b, c, d} = tuple}) do
if valid_ip_octet?(a) and valid_ip_octet?(b) and valid_ip_octet?(c) and valid_ip_octet?(d) do
{:ok, tuple}
else
{:error, :not_ipv4}
end
end
defp validate_parsed_ip({:ok, _}), do: {:error, :not_ipv4}
defp validate_parsed_ip({:error, _}), do: {:error, :invalid_format}
## Type Validation
@doc """
@ -397,21 +407,28 @@ defmodule SnmpKit.SnmpLib.Types do
end
def validate_ip_address(value) when is_binary(value) do
validate_binary_ip_address(value)
end
def validate_ip_address(_), do: {:error, :invalid_format}
defp validate_binary_ip_address(value) do
# Check if it's a printable string (likely an IP address string)
if String.printable?(value) do
{:error, :invalid_format}
else
# It's binary data, check length
case byte_size(value) do
# Valid length but invalid values
4 -> {:error, :invalid_format}
# Wrong length
_ -> {:error, :invalid_length}
end
validate_binary_ip_length(value)
end
end
def validate_ip_address(_), do: {:error, :invalid_format}
defp validate_binary_ip_length(value) do
case byte_size(value) do
# Valid length but invalid values
4 -> {:error, :invalid_format}
# Wrong length
_ -> {:error, :invalid_length}
end
end
@doc """
Validates an SNMP integer value.
@ -920,22 +937,22 @@ defmodule SnmpKit.SnmpLib.Types do
defp encode_value_with_type(value, type, opts) do
validate = Keyword.get(opts, :validate, true)
case perform_encoding(value, type) do
{:ok, encoded_value} ->
if validate do
case validate_encoded_value(type, encoded_value) do
:ok -> {:ok, {type, encoded_value}}
{:error, reason} -> {:error, reason}
end
else
{:ok, {type, encoded_value}}
end
{:error, reason} ->
{:error, reason}
with {:ok, encoded_value} <- perform_encoding(value, type) do
maybe_validate_encoded(type, encoded_value, validate)
end
end
defp maybe_validate_encoded(type, encoded_value, true) do
case validate_encoded_value(type, encoded_value) do
:ok -> {:ok, {type, encoded_value}}
{:error, reason} -> {:error, reason}
end
end
defp maybe_validate_encoded(type, encoded_value, false) do
{:ok, {type, encoded_value}}
end
# Perform the actual encoding based on type
defp perform_encoding(value, :string) when is_binary(value), do: {:ok, value}
defp perform_encoding(value, :string) when is_list(value), do: {:ok, List.to_string(value)}

89
lib/snmpkit/snmp_lib/walker.ex
View file

@ -311,25 +311,7 @@ defmodule SnmpKit.SnmpLib.Walker do
{:ok, varbinds} ->
{valid_varbinds, continue?} = filter_table_varbinds(varbinds, state.table_prefix)
if continue? and valid_varbinds != [] do
# Get last OID for next request
last_oid =
case List.last(valid_varbinds) do
{oid, _type, _value} -> oid
{oid, _value} -> oid
end
new_state = %{
state
| current_oid: last_oid,
accumulated: valid_varbinds ++ state.accumulated
}
bulk_walk_loop(new_state)
else
{:ok, Enum.reverse(state.accumulated ++ valid_varbinds)}
end
process_bulk_walk_result(valid_varbinds, continue?, state, &bulk_walk_loop/1)
{:error, reason} ->
{:error, reason}
@ -343,30 +325,36 @@ defmodule SnmpKit.SnmpLib.Walker do
{:ok, varbinds} ->
{valid_varbinds, continue?} = filter_subtree_varbinds(varbinds, state.subtree_prefix)
if continue? and valid_varbinds != [] do
last_oid =
case List.last(valid_varbinds) do
{oid, _type, _value} -> oid
{oid, _value} -> oid
end
new_state = %{
state
| current_oid: last_oid,
accumulated: valid_varbinds ++ state.accumulated
}
bulk_walk_subtree_loop(new_state)
else
{:ok, Enum.reverse(state.accumulated ++ valid_varbinds)}
end
process_bulk_walk_result(valid_varbinds, continue?, state, &bulk_walk_subtree_loop/1)
{:error, reason} ->
{:error, reason}
end
end
defp process_bulk_walk_result(valid_varbinds, continue?, state, continue_fn) do
if continue? and valid_varbinds != [] do
last_oid = extract_last_oid(valid_varbinds)
new_state = %{
state
| current_oid: last_oid,
accumulated: valid_varbinds ++ state.accumulated
}
continue_fn.(new_state)
else
{:ok, Enum.reverse(state.accumulated ++ valid_varbinds)}
end
end
defp extract_last_oid(varbinds) do
case List.last(varbinds) do
{oid, _type, _value} -> oid
{oid, _value} -> oid
end
end
# Sequential walking fallback
defp sequential_walk_table(host, table_oid, opts) do
initial_state = %{
@ -501,27 +489,24 @@ defmodule SnmpKit.SnmpLib.Walker do
{:ok, varbinds} ->
{valid_varbinds, continue?} = filter_table_varbinds(varbinds, state.table_prefix)
if continue? and valid_varbinds != [] do
# Get last OID for next request
last_oid =
case List.last(valid_varbinds) do
{oid, _type, _value} -> oid
{oid, _value} -> oid
end
new_state = %{state | current_oid: last_oid}
{[valid_varbinds], new_state}
else
new_state = %{state | finished: true}
{[valid_varbinds], new_state}
end
process_stream_chunk_result(valid_varbinds, continue?, state)
{:error, _reason} ->
{:halt, nil}
end
end
defp process_stream_chunk_result(valid_varbinds, continue?, state) do
if continue? and valid_varbinds != [] do
last_oid = extract_last_oid(valid_varbinds)
new_state = %{state | current_oid: last_oid}
{[valid_varbinds], new_state}
else
new_state = %{state | finished: true}
{[valid_varbinds], new_state}
end
end
defp cleanup_streaming_state(_state), do: :ok
# Helper functions

52
lib/snmpkit/snmp_mgr/bulk.ex
View file

@ -44,36 +44,36 @@ defmodule SnmpKit.SnmpMgr.Bulk do
_ ->
oids_list = if is_list(oids), do: oids, else: [oids]
case resolve_oids(oids_list) do
{:ok, resolved_oids} ->
# For multiple OIDs, use non_repeaters to get single values for some
non_repeaters = Keyword.get(opts, :non_repeaters, @default_non_repeaters)
max_repetitions = Keyword.get(opts, :max_repetitions, @default_max_repetitions)
bulk_opts =
opts
|> Keyword.put(:non_repeaters, non_repeaters)
|> Keyword.put(:max_repetitions, max_repetitions)
|> Keyword.put(:version, :v2c)
# Use the first OID as the starting point for GETBULK
starting_oid = hd(resolved_oids)
case Core.send_get_bulk_request(target, starting_oid, bulk_opts) do
{:ok, results} ->
merged_opts = Config.merge_opts(opts)
{:ok, Format.enrich_varbinds(results, merged_opts)}
{:error, reason} ->
{:error, reason}
end
error ->
error
with {:ok, resolved_oids} <- resolve_oids(oids_list) do
perform_bulk_request(target, resolved_oids, opts)
end
end
end
defp perform_bulk_request(target, resolved_oids, opts) do
# For multiple OIDs, use non_repeaters to get single values for some
non_repeaters = Keyword.get(opts, :non_repeaters, @default_non_repeaters)
max_repetitions = Keyword.get(opts, :max_repetitions, @default_max_repetitions)
bulk_opts =
opts
|> Keyword.put(:non_repeaters, non_repeaters)
|> Keyword.put(:max_repetitions, max_repetitions)
|> Keyword.put(:version, :v2c)
# Use the first OID as the starting point for GETBULK
starting_oid = hd(resolved_oids)
case Core.send_get_bulk_request(target, starting_oid, bulk_opts) do
{:ok, results} ->
merged_opts = Config.merge_opts(opts)
{:ok, Format.enrich_varbinds(results, merged_opts)}
{:error, reason} ->
{:error, reason}
end
end
@doc """
Optimized table retrieval using GETBULK.

188
lib/snmpkit/snmp_mgr/core.ex
View file

@ -226,72 +226,63 @@ defmodule SnmpKit.SnmpMgr.Core do
"""
@spec send_set_request(target(), oid(), term(), opts()) :: snmp_result()
def send_set_request(target, oid, value, opts \\ []) do
# Parse target to extract host and port
{host, updated_opts} =
case Target.parse(target) do
{:ok, %{host: host, port: port}} ->
# Only use parsed port if the target actually contained a port specification
if target_contains_port?(target) do
# Target contained port - use parsed port
opts_with_port = Keyword.put(opts, :port, port)
{host, opts_with_port}
else
# Target didn't contain port - preserve user's port option
{host, opts}
end
{:error, _reason} ->
# Failed to parse, use as-is
{target, opts}
end
# Convert oid to proper format
oid_parsed =
case parse_oid(oid) do
{:ok, oid_list} -> oid_list
{:error, _} -> oid
end
# Convert value to snmp_lib format expected by SnmpKit.SnmpLib.Manager
typed_value =
cond do
# Already typed tuple - normalize type to manager's expected atoms
match?({t, _v} when is_atom(t), value) ->
normalize_manager_typed(value)
is_binary(value) ->
{:string, value}
is_integer(value) ->
{:integer, value}
# IPv4 tuple
match?(
{a, b, c, d} when is_integer(a) and is_integer(b) and is_integer(c) and is_integer(d),
value
) ->
{:ip_address, value}
true ->
{:opaque, value}
end
# Map options to snmp_lib format
{host, updated_opts} = parse_target_with_port(target, opts)
oid_parsed = parse_oid_or_fallback(oid)
typed_value = convert_value_to_typed(value)
snmp_lib_opts = map_options_to_snmp_lib(updated_opts)
# Use SnmpKit.SnmpLib.Manager for the actual operation
case Manager.set(host, oid_parsed, typed_value, snmp_lib_opts) do
{:ok, result} -> {:ok, result}
{:error, reason} -> {:error, reason}
end
end
defp parse_target_with_port(target, opts) do
case Target.parse(target) do
{:ok, %{host: host, port: port}} ->
if target_contains_port?(target) do
{host, Keyword.put(opts, :port, port)}
else
{host, opts}
end
{:error, _reason} ->
{target, opts}
end
end
defp parse_oid_or_fallback(oid) do
case parse_oid(oid) do
{:ok, oid_list} -> oid_list
{:error, _} -> oid
end
end
defp convert_value_to_typed(value) do
cond do
match?({t, _v} when is_atom(t), value) -> normalize_manager_typed(value)
is_binary(value) -> {:string, value}
is_integer(value) -> {:integer, value}
ipv4_tuple?(value) -> {:ip_address, value}
true -> {:opaque, value}
end
end
defp ipv4_tuple?({a, b, c, d}) when is_integer(a) and is_integer(b) and is_integer(c) and is_integer(d), do: true
defp ipv4_tuple?(_), do: false
# Normalize typed values to the atoms that SnmpKit.SnmpLib.Manager expects
defp normalize_manager_typed({type, val}) when is_atom(type) do
mapped_type = Map.get(@type_normalization, type, type)
{mapped_type, val}
end
# Extract varbinds from get_bulk results
defp extract_varbinds(%{"varbinds" => varbinds}) when is_list(varbinds), do: varbinds
defp extract_varbinds(results) when is_list(results), do: results
defp extract_varbinds(_), do: []
@doc """
Sends an SNMP GETBULK request (SNMPv2c only).
@ -313,55 +304,13 @@ defmodule SnmpKit.SnmpMgr.Core do
case version do
:v2c ->
# Parse target to extract host and port
{host, updated_opts} =
case Target.parse(target) do
{:ok, %{host: host, port: port}} ->
# Only use parsed port if the target actually contained a port specification
if target_contains_port?(target) do
# Target contained port - use parsed port
opts_with_port = Keyword.put(opts, :port, port)
{host, opts_with_port}
else
# Target didn't contain port - preserve user's port option
{host, opts}
end
{:error, _reason} ->
# Failed to parse, use as-is
{target, opts}
end
# Convert oid to proper format
oid_parsed =
case parse_oid(oid) do
{:ok, oid_list} -> oid_list
{:error, _} -> oid
end
# Map options to snmp_lib format
{host, updated_opts} = parse_target_with_port(target, opts)
oid_parsed = parse_oid_or_fallback(oid)
snmp_lib_opts = map_options_to_snmp_lib(updated_opts)
# Use SnmpKit.SnmpLib.Manager for the actual operation
case Manager.get_bulk(host, oid_parsed, snmp_lib_opts) do
{:ok, results} ->
# Process the results to extract varbinds in 3-tuple format
processed_results =
case results do
# Map format (snmp_lib v1.0.5+)
%{"varbinds" => varbinds} when is_list(varbinds) ->
varbinds
# Direct list format (older versions)
results when is_list(results) ->
results
# Other formats
_other ->
[]
end
{:ok, processed_results}
{:ok, extract_varbinds(results)}
{:error, reason} ->
{:error, reason}
@ -497,40 +446,29 @@ defmodule SnmpKit.SnmpMgr.Core do
end
def parse_oid(oid) when is_binary(oid) do
# Handle empty string case
case String.trim(oid) do
"" ->
{:ok, [1, 3]}
trimmed ->
# Try MIB registry first for symbolic names like "sysDescr.0"
case SnmpKit.SnmpLib.MIB.Registry.resolve_name(trimmed) do
{:ok, [_ | _] = oid_list} ->
{:ok, oid_list}
{:error, _} ->
# Try numeric string parsing
case OID.string_to_list(trimmed) do
{:ok, [_ | _] = oid_list} ->
{:ok, oid_list}
{:ok, []} ->
# Empty result fallback
{:ok, [1, 3]}
{:error, _} ->
# Fall back to MIB GenServer for container OIDs like "system", "interfaces"
case MIB.resolve(trimmed) do
{:ok, oid_list} -> {:ok, oid_list}
error -> error
end
end
end
"" -> {:ok, [1, 3]}
trimmed -> resolve_oid_string(trimmed)
end
end
def parse_oid(_), do: {:error, :invalid_oid_input}
defp resolve_oid_string(trimmed) do
with {:error, _} <- SnmpKit.SnmpLib.MIB.Registry.resolve_name(trimmed),
{:error, _} <- parse_numeric_oid(trimmed) do
MIB.resolve(trimmed)
end
end
defp parse_numeric_oid(oid_string) do
case OID.string_to_list(oid_string) do
{:ok, [_ | _] = oid_list} -> {:ok, oid_list}
{:ok, []} -> {:ok, [1, 3]}
error -> error
end
end
# Type information must never be inferred - it must be preserved from SNMP responses
# Removing type inference functions to prevent loss of critical type information
# Any SNMP operation that does not preserve type information should fail with an error

113
lib/snmpkit/snmp_mgr/format.ex
View file

@ -85,41 +85,7 @@ defmodule SnmpKit.SnmpMgr.Format do
{"1.3.6.1.2.1.4.20.1.1.192.168.1.1", :ip_address, "192.168.1.1"}
"""
def pretty_print({oid, type, value}) do
formatted_value =
case type do
:timeticks ->
uptime(value)
:ip_address ->
ip_address(value)
:counter32 ->
"#{value} (Counter32)"
:counter64 ->
"#{value} (Counter64)"
:gauge32 ->
"#{value} (Gauge32)"
:unsigned32 ->
"#{value} (Unsigned32)"
:octet_string ->
format_octet_string(value)
:object_identifier ->
case value do
oid_list when is_list(oid_list) -> Enum.join(oid_list, ".")
oid_string when is_binary(oid_string) -> oid_string
other -> inspect(other)
end
_ ->
inspect(value)
end
{oid, type, formatted_value}
{oid, type, format_by_type(type, value)}
end
@doc """
@ -225,55 +191,50 @@ defmodule SnmpKit.SnmpMgr.Format do
include_names = Keyword.get(opts, :include_names, true)
include_formatted = Keyword.get(opts, :include_formatted, true)
{oid_string, oid_list} =
cond do
is_list(oid_any) ->
{Enum.join(oid_any, "."), oid_any}
{oid_string, oid_list} = normalize_oid(oid_any)
name = get_mib_name(oid_string, include_names)
is_binary(oid_any) ->
case OID.string_to_list(oid_any) do
{:ok, list} -> {oid_any, list}
_ -> {oid_any, []}
end
%{oid: oid_string, oid_list: oid_list, type: type, value: value}
|> maybe_add_name(name, include_names)
|> maybe_add_formatted(type, value, include_formatted)
end
true ->
str = to_string(oid_any)
defp normalize_oid(oid) when is_list(oid), do: {Enum.join(oid, "."), oid}
case OID.string_to_list(str) do
{:ok, list} -> {str, list}
_ -> {str, []}
end
end
defp normalize_oid(oid) when is_binary(oid) do
case OID.string_to_list(oid) do
{:ok, list} -> {oid, list}
_ -> {oid, []}
end
end
name =
if include_names do
try do
case MIB.reverse_lookup(oid_string) do
{:ok, mib_name} -> mib_name
_ -> nil
end
catch
:exit, _ -> nil
end
end
defp normalize_oid(oid) do
str = to_string(oid)
enriched = %{
oid: oid_string,
oid_list: oid_list,
type: type,
value: value
}
case OID.string_to_list(str) do
{:ok, list} -> {str, list}
_ -> {str, []}
end
end
enriched = if include_names, do: Map.put(enriched, :name, name), else: enriched
defp get_mib_name(_oid_string, false), do: nil
enriched =
if include_formatted do
Map.put(enriched, :formatted, format_by_type(type, value))
else
enriched
end
defp get_mib_name(oid_string, true) do
case MIB.reverse_lookup(oid_string) do
{:ok, mib_name} -> mib_name
_ -> nil
end
catch
:exit, _ -> nil
end
enriched
defp maybe_add_name(enriched, _name, false), do: enriched
defp maybe_add_name(enriched, name, true), do: Map.put(enriched, :name, name)
defp maybe_add_formatted(enriched, _type, _value, false), do: enriched
defp maybe_add_formatted(enriched, type, value, true) do
Map.put(enriched, :formatted, format_by_type(type, value))
end
@doc """

51
lib/snmpkit/snmp_mgr/walk.ex
View file

@ -120,38 +120,10 @@ defmodule SnmpKit.SnmpMgr.Walk do
defp walk_from_oid(target, current_oid, root_oid, acc, remaining, opts) when remaining > 0 do
case Core.send_get_next_request(target, current_oid, opts) do
{:ok, {next_oid_string, type, value}} ->
case OID.string_to_list(next_oid_string) do
{:ok, next_oid} ->
if still_in_scope?(next_oid, root_oid) do
new_acc = [{next_oid_string, type, value} | acc]
walk_from_oid(target, next_oid, root_oid, new_acc, remaining - 1, opts)
else
# Walked beyond the root scope
{:ok, Enum.reverse(acc)}
end
handle_walk_response(target, root_oid, acc, remaining, opts, next_oid_string, type, value)
{:error, _} ->
{:ok, Enum.reverse(acc)}
end
{:error, {:snmp_error, :endOfMibView}} ->
# Reached end of MIB
{:ok, Enum.reverse(acc)}
{:error, {:snmp_error, :noSuchName}} ->
# No more objects
{:ok, Enum.reverse(acc)}
{:error, :end_of_mib_view} ->
# Reached end of MIB (alternative format)
{:ok, Enum.reverse(acc)}
{:error, :no_such_name} ->
# No more objects (alternative format)
{:ok, Enum.reverse(acc)}
{:error, _} = error ->
error
{:error, error} ->
handle_walk_error(error, acc)
end
end
@ -160,6 +132,23 @@ defmodule SnmpKit.SnmpMgr.Walk do
{:ok, Enum.reverse(acc)}
end
defp handle_walk_response(target, root_oid, acc, remaining, opts, next_oid_string, type, value) do
with {:ok, next_oid} <- OID.string_to_list(next_oid_string),
true <- still_in_scope?(next_oid, root_oid) do
new_acc = [{next_oid_string, type, value} | acc]
walk_from_oid(target, next_oid, root_oid, new_acc, remaining - 1, opts)
else
{:error, _} -> {:ok, Enum.reverse(acc)}
false -> {:ok, Enum.reverse(acc)}
end
end
defp handle_walk_error({:snmp_error, :endOfMibView}, acc), do: {:ok, Enum.reverse(acc)}
defp handle_walk_error({:snmp_error, :noSuchName}, acc), do: {:ok, Enum.reverse(acc)}
defp handle_walk_error(:end_of_mib_view, acc), do: {:ok, Enum.reverse(acc)}
defp handle_walk_error(:no_such_name, acc), do: {:ok, Enum.reverse(acc)}
defp handle_walk_error(error, _acc), do: {:error, error}
defp still_in_scope?(current_oid, root_oid) do
# Check if current OID is still within the root OID scope
List.starts_with?(current_oid, root_oid)

158
lib/towerops/etcd_lock.ex Normal file
View file

@ -0,0 +1,158 @@
defmodule Towerops.EtcdLock do
@moduledoc """
Distributed lock implementation using etcd with automatic keepalive.
This module provides distributed locking across the Towerops cluster using
etcd's lease mechanism. Locks automatically expire if the holder crashes,
and the etcd client automatically renews leases for running processes.
## Features
- Atomic lock acquisition using etcd transactions
- Automatic lease renewal (no manual heartbeat needed)
- TTL-based expiration on crashes
- Compare-and-swap semantics for safety
"""
require Logger
@doc """
Acquires a distributed lock with automatic keepalive.
Creates an etcd lease and attempts to atomically acquire the lock.
The lease is automatically renewed by the etcd client until released.
## Parameters
- `conn` - etcd connection
- `key` - lock key (will be prefixed with "towerops/locks/")
- `ttl_seconds` - lease TTL (default: 60s)
## Returns
- `{:ok, lease_id}` if lock acquired
- `{:error, :already_locked}` if lock is held by another process
"""
def acquire(conn, key, ttl_seconds \\ 60) do
full_key = lock_key(key)
value = node_identifier()
# Create lease with auto-keepalive
case :eetcd_lease.grant(conn, ttl_seconds) do
{:ok, %{~c"ID" => lease_id}} ->
# Start keepalive for the lease
{:ok, _pid} = :eetcd_lease.keep_alive(conn, lease_id)
# Attempt atomic lock acquisition
case put_if_not_exists(conn, full_key, value, lease_id) do
{:ok, true} ->
Logger.debug("Acquired etcd lock: #{key} (lease: #{lease_id})")
{:ok, lease_id}
{:ok, false} ->
# Lock already held - revoke our unused lease
:eetcd_lease.revoke(conn, lease_id)
{:error, :already_locked}
{:error, reason} = error ->
Logger.error("Failed to acquire lock #{key}: #{inspect(reason)}")
:eetcd_lease.revoke(conn, lease_id)
error
end
{:error, reason} = error ->
Logger.error("Failed to create lease for lock #{key}: #{inspect(reason)}")
error
end
end
@doc """
Releases a distributed lock.
Revokes the lease, which automatically deletes the key and triggers
watchers on other nodes.
## Parameters
- `conn` - etcd connection
- `lease_id` - lease ID returned from acquire/3
"""
def release(conn, lease_id) do
case :eetcd_lease.revoke(conn, lease_id) do
{:ok, _} ->
Logger.debug("Released etcd lock (lease: #{lease_id})")
:ok
{:error, reason} = error ->
Logger.error("Failed to release lock (lease: #{lease_id}): #{inspect(reason)}")
error
end
end
@doc """
Acquires a lock, executes a function, then releases the lock.
Ensures the lock is always released, even if the function raises an exception.
## Parameters
- `conn` - etcd connection
- `key` - lock key
- `fun` - function to execute while holding the lock
- `opts` - options (`:ttl` - lease TTL in seconds, default: 60)
## Returns
Result of the function, or `{:error, :already_locked}` if lock acquisition fails.
"""
def with_lock(conn, key, fun, opts \\ []) do
ttl = Keyword.get(opts, :ttl, 60)
case acquire(conn, key, ttl) do
{:ok, lease_id} ->
try do
fun.()
after
release(conn, lease_id)
end
{:error, _reason} = error ->
error
end
end
# Private functions
defp put_if_not_exists(conn, key, value, lease_id) do
# Use etcd transaction: only put if key doesn't exist (mod_revision == 0)
# Create comparison: key mod_revision == 0 means key doesn't exist
cmp = :eetcd_compare.new(key)
if_cond = [:eetcd_compare.mod_revision(cmp, "=", 0)]
# Create put operation with lease
kv_ctx = :eetcd_kv.new(conn)
kv_with_key = :eetcd_kv.with_key(kv_ctx, key)
kv_with_value = :eetcd_kv.with_value(kv_with_key, value)
kv_with_lease = :eetcd_kv.with_lease(kv_with_value, lease_id)
then_ops = [:eetcd_op.put(kv_with_lease)]
else_ops = []
case :eetcd_kv.txn(conn, if_cond, then_ops, else_ops) do
{:ok, %{succeeded: true}} ->
{:ok, true}
{:ok, %{succeeded: false}} ->
{:ok, false}
{:error, reason} ->
{:error, reason}
end
end
defp lock_key(key), do: "towerops/locks/#{key}"
defp node_identifier do
"#{node()}:#{System.system_time(:millisecond)}"
end
end

18
lib/towerops/monitoring/device_monitor.ex
View file

@ -23,17 +23,19 @@ defmodule Towerops.Monitoring.DeviceMonitor do
@doc """
Starts a monitor for the given device ID.
"""
def start_link(device_id) do
GenServer.start_link(__MODULE__, device_id, name: via_tuple(device_id))
def start_link(opts) do
device_id = Keyword.fetch!(opts, :device_id)
lease_id = Keyword.get(opts, :lease_id)
GenServer.start_link(__MODULE__, {device_id, lease_id}, name: via_tuple(device_id))
end
@doc """
Triggers an immediate check for the device.
Cluster-aware: finds the monitor regardless of which node it's on.
"""
def trigger_check(device_id) do
# Check if monitor process is running (cluster-aware lookup)
case Horde.Registry.lookup(Towerops.Monitoring.Registry, device_id) do
# Check if monitor process is running (node-local lookup)
case Registry.lookup(Towerops.Monitoring.Registry, device_id) do
[{_pid, _}] ->
# Process exists, send cast
GenServer.cast(via_tuple(device_id), :check_now)
@ -47,7 +49,7 @@ defmodule Towerops.Monitoring.DeviceMonitor do
# Server Callbacks
@impl true
def init(device_id) do
def init({device_id, lease_id}) do
device = Devices.get_device!(device_id)
if device.monitoring_enabled do
@ -55,7 +57,7 @@ defmodule Towerops.Monitoring.DeviceMonitor do
send(self(), :check_device)
end
{:ok, %{device_id: device_id}}
{:ok, %{device_id: device_id, lease_id: lease_id}}
end
@impl true
@ -246,6 +248,6 @@ defmodule Towerops.Monitoring.DeviceMonitor do
end
defp via_tuple(device_id) do
{:via, Horde.Registry, {Towerops.Monitoring.Registry, device_id}}
{:via, Registry, {Towerops.Monitoring.Registry, device_id}}
end
end

248
lib/towerops/monitoring/etcd_coordinator.ex Normal file
View file

@ -0,0 +1,248 @@
defmodule Towerops.Monitoring.EtcdCoordinator do
@moduledoc """
Coordinates distributed SNMP polling and monitoring across the cluster using etcd.
This GenServer maintains an etcd connection and watches for lock releases,
enabling instant failover when processes need to be restarted on other nodes.
## Responsibilities
- Connect to etcd cluster on startup
- Acquire locks for all devices that should have pollers/monitors
- Watch for lock releases (when other nodes die)
- Automatically start processes when locks become available
"""
use GenServer
alias Towerops.Devices
alias Towerops.EtcdLock
alias Towerops.Monitoring.DeviceMonitor
alias Towerops.Snmp.PollerWorker
require Logger
@rebalance_interval 60_000
def start_link(opts) do
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
end
@impl true
def init(_opts) do
# Connect to etcd cluster
case connect_etcd() do
{:ok, conn} ->
Logger.info("Connected to etcd cluster")
# Watch for lock releases (triggers instant failover)
watch_ctx = :eetcd_watch.new()
watch_with_key = :eetcd_watch.with_key(watch_ctx, "towerops/locks/")
watch_with_prefix = :eetcd_watch.with_prefix(watch_with_key)
{:ok, watcher} = :eetcd_watch.watch(conn, watch_with_prefix)
Logger.info("Watching etcd for lock releases")
# Start acquiring locks for all devices
send(self(), :acquire_locks)
# Schedule periodic rebalancing (safety net for missed watch events)
schedule_rebalance()
{:ok, %{conn: conn, watcher: watcher, owned_leases: %{}}}
{:error, reason} ->
Logger.error("Failed to connect to etcd: #{inspect(reason)}")
{:stop, {:etcd_connection_failed, reason}}
end
end
@impl true
def handle_info(:acquire_locks, state) do
# Try to acquire locks for all pollers and monitors
acquire_all_poller_locks(state.conn, state.owned_leases)
acquire_all_monitor_locks(state.conn, state.owned_leases)
{:noreply, state}
end
@impl true
def handle_info(:rebalance, state) do
# Periodic rebalance - acquire any orphaned locks
Logger.debug("Running periodic etcd rebalance")
send(self(), :acquire_locks)
schedule_rebalance()
{:noreply, state}
end
@impl true
def handle_info({:watcher, _watcher, events}, state) do
# Instant notification when locks are released
Enum.each(events, &handle_watch_event(&1, state.conn))
{:noreply, state}
end
@impl true
def terminate(_reason, state) do
# Release all our locks on shutdown
Enum.each(state.owned_leases, fn {_key, lease_id} ->
EtcdLock.release(state.conn, lease_id)
end)
:ok
end
# Public API
@doc """
Returns the etcd connection for direct use by other modules.
"""
def get_connection do
GenServer.call(__MODULE__, :get_connection)
end
@impl true
def handle_call(:get_connection, _from, state) do
{:reply, {:ok, state.conn}, state}
end
# Private functions
defp connect_etcd do
# Connect to etcd StatefulSet in Kubernetes
endpoints = [
"etcd-0.etcd.towerops.svc.cluster.local:2379",
"etcd-1.etcd.towerops.svc.cluster.local:2379",
"etcd-2.etcd.towerops.svc.cluster.local:2379"
]
:eetcd.open(:towerops_etcd, endpoints)
end
defp handle_watch_event(%{type: :delete, kv: %{key: key}}, conn) do
# Lock was released - try to acquire it
device_id = extract_device_id(key)
type = extract_lock_type(key)
Logger.info("Lock released: #{type}/#{device_id}, attempting to acquire")
case type do
:poller -> try_start_poller(conn, device_id)
:monitor -> try_start_monitor(conn, device_id)
_ -> :ok
end
end
defp handle_watch_event(_event, _conn), do: :ok
defp acquire_all_poller_locks(conn, owned_leases) do
devices = Devices.list_snmp_enabled_devices()
Enum.each(devices, fn device ->
key = "poller/#{device.id}"
# Skip if we already own this lock
if !Map.has_key?(owned_leases, key) do
try_start_poller(conn, device.id)
end
end)
end
defp acquire_all_monitor_locks(conn, owned_leases) do
devices = Devices.list_monitored_devices()
Enum.each(devices, fn device ->
key = "monitor/#{device.id}"
# Skip if we already own this lock
if !Map.has_key?(owned_leases, key) do
try_start_monitor(conn, device.id)
end
end)
end
defp try_start_poller(conn, device_id) do
key = "poller/#{device_id}"
case EtcdLock.acquire(conn, key, 60) do
{:ok, lease_id} ->
Logger.info("Acquired poller lock for device #{device_id}")
spec = {PollerWorker, device_id: device_id, lease_id: lease_id}
case DynamicSupervisor.start_child(Towerops.LocalPollerSupervisor, spec) do
{:ok, _pid} ->
# Track the lease
GenServer.cast(__MODULE__, {:track_lease, key, lease_id})
:ok
{:error, reason} ->
Logger.error("Failed to start poller for #{device_id}: #{inspect(reason)}")
# Release the lock since we couldn't start the worker
EtcdLock.release(conn, lease_id)
end
{:error, :already_locked} ->
# Another node has it
:ok
{:error, reason} ->
Logger.error("Failed to acquire poller lock for #{device_id}: #{inspect(reason)}")
end
end
defp try_start_monitor(conn, device_id) do
key = "monitor/#{device_id}"
case EtcdLock.acquire(conn, key, 60) do
{:ok, lease_id} ->
Logger.info("Acquired monitor lock for device #{device_id}")
spec = {DeviceMonitor, device_id: device_id, lease_id: lease_id}
case DynamicSupervisor.start_child(Towerops.LocalMonitorSupervisor, spec) do
{:ok, _pid} ->
# Track the lease
GenServer.cast(__MODULE__, {:track_lease, key, lease_id})
:ok
{:error, reason} ->
Logger.error("Failed to start monitor for #{device_id}: #{inspect(reason)}")
# Release the lock since we couldn't start the worker
EtcdLock.release(conn, lease_id)
end
{:error, :already_locked} ->
# Another node has it
:ok
{:error, reason} ->
Logger.error("Failed to acquire monitor lock for #{device_id}: #{inspect(reason)}")
end
end
@impl true
def handle_cast({:track_lease, key, lease_id}, state) do
owned_leases = Map.put(state.owned_leases, key, lease_id)
{:noreply, %{state | owned_leases: owned_leases}}
end
defp extract_device_id(key) do
# "towerops/locks/poller/device-id" -> "device-id"
key
|> String.split("/")
|> List.last()
end
defp extract_lock_type(key) do
# "towerops/locks/poller/device-id" -> :poller
cond do
String.contains?(key, "/poller/") -> :poller
String.contains?(key, "/monitor/") -> :monitor
true -> :unknown
end
end
defp schedule_rebalance do
Process.send_after(self(), :rebalance, @rebalance_interval)
end
end

117
lib/towerops/monitoring/supervisor.ex
View file

@ -5,12 +5,11 @@ defmodule Towerops.Monitoring.Supervisor do
use Supervisor
alias Towerops.Monitoring.DeviceMonitor
alias Towerops.Monitoring.EtcdCoordinator
alias Towerops.Snmp.NeighborCleanupWorker
alias Towerops.Snmp.PollerRegistry
alias Towerops.Snmp.PollerWorker
# Note: DynamicSupervisor names are process names for Horde, not module aliases
require Logger
def start_link(init_arg) do
@ -20,73 +19,55 @@ defmodule Towerops.Monitoring.Supervisor do
@impl true
def init(_init_arg) do
base_children = [
# Horde.Registry for distributed process naming (cluster-aware)
{Horde.Registry, keys: :unique, name: Towerops.Monitoring.Registry, members: :auto},
# Horde.Registry for SNMP poller processes (cluster-aware)
{Horde.Registry, keys: :unique, name: PollerRegistry, members: :auto},
# Local Registry for process naming (node-local)
{Registry, keys: :unique, name: Towerops.Monitoring.Registry},
# Local Registry for SNMP poller processes (node-local)
{Registry, keys: :unique, name: PollerRegistry},
# Task.Supervisor for parallel polling operations (local to each node)
{Task.Supervisor, name: Towerops.Snmp.PollerTaskSupervisor},
# Horde.DynamicSupervisor for monitor workers (distributed across cluster)
{Horde.DynamicSupervisor, name: DynamicSupervisor, strategy: :one_for_one, members: :auto},
# Horde.DynamicSupervisor for SNMP poller workers (distributed across cluster)
{Horde.DynamicSupervisor, name: PollerSupervisor, strategy: :one_for_one, members: :auto}
# Local DynamicSupervisor for monitor workers (node-local)
{DynamicSupervisor, name: Towerops.LocalMonitorSupervisor, strategy: :one_for_one},
# Local DynamicSupervisor for SNMP poller workers (node-local)
{DynamicSupervisor, name: Towerops.LocalPollerSupervisor, strategy: :one_for_one}
]
# Only start cleanup worker in non-test environments
# Only start cleanup worker and etcd coordinator in production
children =
if test_mode?() do
base_children
else
base_children ++ [NeighborCleanupWorker]
end
# Start all monitors after supervisor is initialized (but not in test mode)
# In test mode, the Repo uses Sandbox pool which we can detect
_ =
if !test_mode?() do
Task.start(fn ->
# Wait briefly for the supervisor tree to be fully initialized
Process.sleep(100)
try do
start_all_monitors()
Logger.info("Started all devices monitors")
rescue
error ->
Logger.error("Failed to start monitors: #{inspect(error)}")
end
try do
start_all_snmp_pollers()
Logger.info("Started all SNMP pollers")
rescue
error ->
Logger.error("Failed to start SNMP pollers: #{inspect(error)}")
end
end)
base_children ++ [NeighborCleanupWorker] ++ etcd_coordinator()
end
Supervisor.init(children, strategy: :one_for_one)
end
@doc """
Starts monitoring for a specific device.
Cluster-aware: only one monitor per device across all nodes.
"""
def start_monitor(device_id) do
spec = {DeviceMonitor, device_id}
Horde.DynamicSupervisor.start_child(DynamicSupervisor, spec)
# Start etcd coordinator only in production (when ETCD_ENABLED=true)
defp etcd_coordinator do
if System.get_env("ETCD_ENABLED") == "true" do
[EtcdCoordinator]
else
[]
end
end
@doc """
Stops monitoring for a specific device.
Cluster-aware: stops the monitor regardless of which node it's running on.
Starts monitoring for a specific device (node-local).
Note: In production, monitors are started by EtcdCoordinator after acquiring locks.
This function is primarily for testing and manual operations.
"""
def start_monitor(device_id, lease_id \\ nil) do
spec = {DeviceMonitor, device_id: device_id, lease_id: lease_id}
DynamicSupervisor.start_child(Towerops.LocalMonitorSupervisor, spec)
end
@doc """
Stops monitoring for a specific device (node-local).
"""
def stop_monitor(device_id) do
case Horde.Registry.lookup(Towerops.Monitoring.Registry, device_id) do
case Registry.lookup(Towerops.Monitoring.Registry, device_id) do
[{pid, _}] ->
Horde.DynamicSupervisor.terminate_child(DynamicSupervisor, pid)
DynamicSupervisor.terminate_child(Towerops.LocalMonitorSupervisor, pid)
[] ->
:ok
@ -94,46 +75,28 @@ defmodule Towerops.Monitoring.Supervisor do
end
@doc """
Starts monitors for all devices that has monitoring enabled.
Starts an SNMP poller for a specific device (node-local).
Note: In production, pollers are started by EtcdCoordinator after acquiring locks.
This function is primarily for testing and manual operations.
"""
def start_all_monitors do
Enum.each(Towerops.Devices.list_monitored_devices(), fn device ->
start_monitor(device.id)
end)
def start_snmp_poller(device_id, lease_id \\ nil) do
spec = {PollerWorker, device_id: device_id, lease_id: lease_id}
DynamicSupervisor.start_child(Towerops.LocalPollerSupervisor, spec)
end
@doc """
Starts an SNMP poller for a specific device.
Cluster-aware: only one poller per device across all nodes.
"""
def start_snmp_poller(device_id) do
spec = {PollerWorker, device_id}
Horde.DynamicSupervisor.start_child(PollerSupervisor, spec)
end
@doc """
Stops an SNMP poller for a specific device.
Cluster-aware: stops the poller regardless of which node it's running on.
Stops an SNMP poller for a specific device (node-local).
"""
def stop_snmp_poller(device_id) do
case Horde.Registry.lookup(PollerRegistry, device_id) do
case Registry.lookup(PollerRegistry, device_id) do
[{pid, _}] ->
Horde.DynamicSupervisor.terminate_child(PollerSupervisor, pid)
DynamicSupervisor.terminate_child(Towerops.LocalPollerSupervisor, pid)
[] ->
:ok
end
end
@doc """
Starts SNMP pollers for all devices that has SNMP enabled.
"""
def start_all_snmp_pollers do
Enum.each(Towerops.Devices.list_snmp_enabled_devices(), fn device ->
start_snmp_poller(device.id)
end)
end
# Check if running in test mode by inspecting the Repo pool configuration
defp test_mode? do
config = Application.get_env(:towerops, Towerops.Repo, [])

13
lib/towerops/snmp/poller_worker.ex
View file

@ -33,8 +33,11 @@ defmodule Towerops.Snmp.PollerWorker do
@doc """
Starts a poller for the given device ID.
"""
def start_link(device_id) do
GenServer.start_link(__MODULE__, device_id, name: via_tuple(device_id))
def start_link(opts) do
device_id = Keyword.fetch!(opts, :device_id)
lease_id = Keyword.get(opts, :lease_id)
GenServer.start_link(__MODULE__, {device_id, lease_id}, name: via_tuple(device_id))
end
@doc """
@ -54,7 +57,7 @@ defmodule Towerops.Snmp.PollerWorker do
# Server Callbacks
@impl true
def init(device_id) do
def init({device_id, lease_id}) do
device = Devices.get_device!(device_id)
# device
@ -70,7 +73,7 @@ defmodule Towerops.Snmp.PollerWorker do
end
end
{:ok, %{device_id: device_id}}
{:ok, %{device_id: device_id, lease_id: lease_id}}
end
@impl true
@ -1526,6 +1529,6 @@ defmodule Towerops.Snmp.PollerWorker do
end
defp via_tuple(device_id) do
{:via, Horde.Registry, {PollerRegistry, device_id}}
{:via, Registry, {PollerRegistry, device_id}}
end
end

2
mix.exs
View file

@ -69,7 +69,7 @@ defmodule Towerops.MixProject do
{:jason, "~> 1.2"},
{:dns_cluster, "~> 0.2.0"},
{:libcluster, "~> 3.4"},
{:horde, "~> 0.10.0"},
{:eetcd, "~> 0.6"},
{:bandit, "~> 1.5"},
{:phoenix_pubsub_redis, "~> 3.0"},
{:ecto_psql_extras, "~> 0.6"},

3
mix.lock
View file

@ -5,6 +5,7 @@
"cbor": {:hex, :cbor, "1.0.1", "39511158e8ea5a57c1fcb9639aaa7efde67129678fee49ebbda780f6f24959b0", [:mix], [], "hexpm", "5431acbe7a7908f17f6a9cd43311002836a34a8ab01876918d8cfb709cd8b6a2"},
"cc_precompiler": {:hex, :cc_precompiler, "0.1.11", "8c844d0b9fb98a3edea067f94f616b3f6b29b959b6b3bf25fee94ffe34364768", [:mix], [{:elixir_make, "~> 0.7", [hex: :elixir_make, repo: "hexpm", optional: false]}], "hexpm", "3427232caf0835f94680e5bcf082408a70b48ad68a5f5c0b02a3bea9f3a075b9"},
"comeonin": {:hex, :comeonin, "5.5.1", "5113e5f3800799787de08a6e0db307133850e635d34e9fab23c70b6501669510", [:mix], [], "hexpm", "65aac8f19938145377cee73973f192c5645873dcf550a8a6b18187d17c13ccdb"},
"cowlib": {:hex, :cowlib, "2.13.0", "db8f7505d8332d98ef50a3ef34b34c1afddec7506e4ee4dd4a3a266285d282ca", [:make, :rebar3], [], "hexpm", "e1e1284dc3fc030a64b1ad0d8382ae7e99da46c3246b815318a4b848873800a4"},
"credo": {:hex, :credo, "1.7.15", "283da72eeb2fd3ccf7248f4941a0527efb97afa224bcdef30b4b580bc8258e1c", [:mix], [{:bunt, "~> 0.2.1 or ~> 1.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "291e8645ea3fea7481829f1e1eb0881b8395db212821338e577a90bf225c5607"},
"db_connection": {:hex, :db_connection, "2.9.0", "a6a97c5c958a2d7091a58a9be40caf41ab496b0701d21e1d1abff3fa27a7f371", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "17d502eacaf61829db98facf6f20808ed33da6ccf495354a41e64fe42f9c509c"},
"decimal": {:hex, :decimal, "2.3.0", "3ad6255aa77b4a3c4f818171b12d237500e63525c2fd056699967a3e7ea20f62", [:mix], [], "hexpm", "a4d66355cb29cb47c3cf30e71329e58361cfcb37c34235ef3bf1d7bf3773aeac"},
@ -14,6 +15,7 @@
"ecto": {:hex, :ecto, "3.13.5", "9d4a69700183f33bf97208294768e561f5c7f1ecf417e0fa1006e4a91713a834", [:mix], [{:decimal, "~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "df9efebf70cf94142739ba357499661ef5dbb559ef902b68ea1f3c1fabce36de"},
"ecto_psql_extras": {:hex, :ecto_psql_extras, "0.8.8", "aa02529c97f69aed5722899f5dc6360128735a92dd169f23c5d50b1f7fdede08", [:mix], [{:ecto_sql, "~> 3.7", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:postgrex, "> 0.16.0", [hex: :postgrex, repo: "hexpm", optional: false]}, {:table_rex, "~> 3.1.1 or ~> 4.0", [hex: :table_rex, repo: "hexpm", optional: false]}], "hexpm", "04c63d92b141723ad6fed2e60a4b461ca00b3594d16df47bbc48f1f4534f2c49"},
"ecto_sql": {:hex, :ecto_sql, "3.13.4", "b6e9d07557ddba62508a9ce4a484989a5bb5e9a048ae0e695f6d93f095c25d60", [:mix], [{:db_connection, "~> 2.4.1 or ~> 2.5", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 3.13.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.7", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.19 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "2b38cf0749ca4d1c5a8bcbff79bbe15446861ca12a61f9fba604486cb6b62a14"},
"eetcd": {:hex, :eetcd, "0.6.0", "f0be33bdb520c642be7fef40a344c33c4d7dd603ce2dd133610caf6efd56e499", [:rebar3], [{:gun, "2.1.0", [hex: :gun, repo: "hexpm", optional: false]}], "hexpm", "9e97b6aea4da9a62d8cda9a65028ee335c6ec966b2044bf8d3ba0e6d8e7d6b37"},
"elixir_make": {:hex, :elixir_make, "0.9.0", "6484b3cd8c0cee58f09f05ecaf1a140a8c97670671a6a0e7ab4dc326c3109726", [:mix], [], "hexpm", "db23d4fd8b757462ad02f8aa73431a426fe6671c80b200d9710caf3d1dd0ffdb"},
"elixir_uuid": {:hex, :elixir_uuid, "1.2.1", "dce506597acb7e6b0daeaff52ff6a9043f5919a4c3315abb4143f0b00378c097", [:mix], [], "hexpm", "f7eba2ea6c3555cea09706492716b0d87397b88946e6380898c2889d68585752"},
"erlex": {:hex, :erlex, "0.2.8", "cd8116f20f3c0afe376d1e8d1f0ae2452337729f68be016ea544a72f767d9c12", [:mix], [], "hexpm", "9d66ff9fedf69e49dc3fd12831e12a8a37b76f8651dd21cd45fcf5561a8a7590"},
@ -25,6 +27,7 @@
"fine": {:hex, :fine, "0.1.4", "b19a89c1476c7c57afb5f9314aed5960b5bc95d5277de4cb5ee8e1d1616ce379", [:mix], [], "hexpm", "be3324cc454a42d80951cf6023b9954e9ff27c6daa255483b3e8d608670303f5"},
"gen_smtp": {:hex, :gen_smtp, "1.3.0", "62c3d91f0dcf6ce9db71bcb6881d7ad0d1d834c7f38c13fa8e952f4104a8442e", [:rebar3], [{:ranch, ">= 1.8.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "0b73fbf069864ecbce02fe653b16d3f35fd889d0fdd4e14527675565c39d84e6"},
"gettext": {:hex, :gettext, "1.0.2", "5457e1fd3f4abe47b0e13ff85086aabae760497a3497909b8473e0acee57673b", [:mix], [{:expo, "~> 0.5.1 or ~> 1.0", [hex: :expo, repo: "hexpm", optional: false]}], "hexpm", "eab805501886802071ad290714515c8c4a17196ea76e5afc9d06ca85fb1bfeb3"},
"gun": {:hex, :gun, "2.1.0", "b4e4cbbf3026d21981c447e9e7ca856766046eff693720ba43114d7f5de36e87", [:make, :rebar3], [{:cowlib, "2.13.0", [hex: :cowlib, repo: "hexpm", optional: false]}], "hexpm", "52fc7fc246bfc3b00e01aea1c2854c70a366348574ab50c57dfe796d24a0101d"},
"heroicons": {:git, "https://github.com/tailwindlabs/heroicons.git", "0435d4ca364a608cc75e2f8683d374e55abbae26", [tag: "v2.2.0", sparse: "optimized", depth: 1]},
"honeybadger": {:hex, :honeybadger, "0.24.1", "13ffe56b4d148649c8fbb0e091fefecc5d8e8eb7ade684b6900085a947d741d5", [:mix], [{:ecto, ">= 2.0.0", [hex: :ecto, repo: "hexpm", optional: true]}, {:hackney, "~> 1.1", [hex: :hackney, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:phoenix, ">= 1.0.0 and < 2.0.0", [hex: :phoenix, repo: "hexpm", optional: true]}, {:plug, ">= 1.0.0 and < 2.0.0", [hex: :plug, repo: "hexpm", optional: true]}, {:process_tree, "~> 0.2.1", [hex: :process_tree, repo: "hexpm", optional: false]}, {:req, "~> 0.5.0", [hex: :req, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "0c97d5a82c42298b9935dbc0a7e3c14372c8f55257f603828258ef9f7e0da892"},
"horde": {:hex, :horde, "0.10.0", "31c6a633057c3ec4e73064d7b11ba409c9f3c518aa185377d76bee441b76ceb0", [:mix], [{:delta_crdt, "~> 0.6.2", [hex: :delta_crdt, repo: "hexpm", optional: false]}, {:libring, "~> 1.7", [hex: :libring, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:telemetry_poller, "~> 0.5.0 or ~> 1.0", [hex: :telemetry_poller, repo: "hexpm", optional: false]}], "hexpm", "0b51c435cb698cac9bf9c17391dce3ebb1376ae6154c81f077fc61db771b9432"},

3
test/snmpkit/snmp_lib/decoder_long_length_test.exs
View file

@ -2,6 +2,7 @@ defmodule SnmpKit.SnmpLib.DecoderLongLengthTest do
use ExUnit.Case, async: true
alias SnmpKit.SnmpLib.PDU.Decoder
alias SnmpKit.SnmpMgr.Format
# BER length encoder supporting short form and 1–2 byte long form
defp len_bytes(n) when n < 128, do: <<n>>
@ -51,7 +52,7 @@ defmodule SnmpKit.SnmpLib.DecoderLongLengthTest do
assert value == s
# And the formatter should render it as plain text, not hex
formatted = SnmpKit.SnmpMgr.Format.format_by_type(:octet_string, value)
formatted = Format.format_by_type(:octet_string, value)
assert formatted == s
end

3
test/snmpkit/snmp_lib/mib/compiler_test.exs
View file

@ -9,6 +9,7 @@ defmodule SnmpKit.SnmpLib.MIB.CompilerTest do
use ExUnit.Case, async: true
alias SnmpKit.SnmpLib.MIB.Compiler
alias SnmpKit.SnmpLib.MIB.Parser
# Skip these tests if yecc is not available (requires :parsetools)
@moduletag :yecc_required
@ -131,7 +132,7 @@ defmodule SnmpKit.SnmpLib.MIB.CompilerTest do
# Test that Compiler properly wraps Parser functionality
assert {:ok, compiled} = Compiler.compile_string(mib_content)
assert {:ok, parsed} = SnmpKit.SnmpLib.MIB.Parser.parse(mib_content)
assert {:ok, parsed} = Parser.parse(mib_content)
# Verify the compiler adds the expected structure
assert compiled.name == parsed.name

27
test/snmpkit/snmp_lib/mib/comprehensive_mib_test.exs
View file

@ -168,24 +168,19 @@ defmodule SnmpKit.SnmpLib.MIB.ComprehensiveMibTest do
end
defp test_single_mib_file(file_path, file_name) do
case File.read(file_path) do
{:ok, content} ->
case Parser.parse(content) do
{:ok, mib} when is_map(mib) ->
definitions_count =
case mib do
%{definitions: defs} when is_list(defs) -> length(defs)
_ -> 1
end
{:ok, {file_name, definitions_count}}
{:error, reason} ->
{:error, {file_name, reason}}
end
with {:ok, content} <- File.read(file_path),
{:ok, mib} when is_map(mib) <- Parser.parse(content) do
definitions_count = get_definitions_count(mib)
{:ok, {file_name, definitions_count}}
else
{:error, reason} when is_binary(reason) ->
{:error, {file_name, "File read error: #{reason}"}}
{:error, reason} ->
{:error, {file_name, "File read error: #{reason}"}}
{:error, {file_name, reason}}
end
end
defp get_definitions_count(%{definitions: defs}) when is_list(defs), do: length(defs)
defp get_definitions_count(_), do: 1
end

19
test/snmpkit/snmp_lib/mib/docsis_mib_test.exs
View file

@ -267,18 +267,15 @@ defmodule SnmpKit.SnmpLib.MIB.DocsisMibTest do
defp test_mib_group(mibs, _group_name) do
Enum.map(mibs, fn {name, path} ->
result =
case File.read(path) do
{:ok, content} ->
case Parser.parse(content) do
{:ok, mib} ->
{:ok, mib}
{:error, error} ->
{:error, [error]}
end
{:error, reason} ->
with {:ok, content} <- File.read(path),
{:ok, mib} <- Parser.parse(content) do
{:ok, mib}
else
{:error, reason} when is_binary(reason) ->
{:error, [%{type: :file_error, message: "File not found: #{reason}"}]}
{:error, error} ->
{:error, [error]}
end
{name, result}

21
test/snmpkit/snmp_lib/snmpv3_integration_test.exs
View file

@ -502,29 +502,12 @@ defmodule SnmpKit.SnmpLib.SNMPv3IntegrationTest do
end
defp create_user_with_protocols(engine_id, auth_protocol, priv_protocol) do
auth_key_size =
case auth_protocol do
:md5 -> 16
:sha1 -> 20
:sha256 -> 32
:sha384 -> 48
:sha512 -> 64
end
priv_key_size =
case priv_protocol do
:des -> 8
:aes128 -> 16
:aes192 -> 24
:aes256 -> 32
end
%{
security_name: "protocol_user",
auth_protocol: auth_protocol,
priv_protocol: priv_protocol,
auth_key: :crypto.strong_rand_bytes(auth_key_size),
priv_key: :crypto.strong_rand_bytes(priv_key_size),
auth_key: :crypto.strong_rand_bytes(auth_key_size(auth_protocol)),
priv_key: :crypto.strong_rand_bytes(priv_key_size(priv_protocol)),
engine_id: engine_id,
engine_boots: 1,
engine_time: System.system_time(:second)

68
test/snmpkit/snmp_lib/v3_encoder_test.exs
View file

@ -482,53 +482,37 @@ defmodule SnmpKit.SnmpLib.PDU.V3EncoderTest do
auth_protocol = Keyword.get(opts, :auth_protocol, :sha256)
priv_protocol = Keyword.get(opts, :priv_protocol, :aes128)
auth_key =
case security_level do
:no_auth_no_priv -> <<>>
_ -> :crypto.strong_rand_bytes(32)
end
priv_key =
case security_level do
:auth_priv ->
# Generate key based on protocol requirements
key_size =
case priv_protocol do
:des -> 8
:aes128 -> 16
:aes192 -> 24
:aes256 -> 32
# default fallback
_ -> 16
end
:crypto.strong_rand_bytes(key_size)
_ ->
<<>>
end
actual_auth_protocol =
case security_level do
:no_auth_no_priv -> :none
_ -> auth_protocol
end
actual_priv_protocol =
case security_level do
:auth_priv -> priv_protocol
_ -> :none
end
%{
security_name: "test_user",
auth_protocol: actual_auth_protocol,
priv_protocol: actual_priv_protocol,
auth_key: auth_key,
priv_key: priv_key,
auth_protocol: get_auth_protocol(security_level, auth_protocol),
priv_protocol: get_priv_protocol(security_level, priv_protocol),
auth_key: generate_auth_key(security_level),
priv_key: generate_priv_key(security_level, priv_protocol),
engine_id: "test_engine_id",
engine_boots: 1,
engine_time: System.system_time(:second)
}
end
defp get_auth_protocol(:no_auth_no_priv, _), do: :none
defp get_auth_protocol(_, auth_protocol), do: auth_protocol
defp get_priv_protocol(:auth_priv, priv_protocol), do: priv_protocol
defp get_priv_protocol(_, _), do: :none
defp generate_auth_key(:no_auth_no_priv), do: <<>>
defp generate_auth_key(_), do: :crypto.strong_rand_bytes(32)
defp generate_priv_key(:auth_priv, priv_protocol) do
key_size = get_priv_key_size(priv_protocol)
:crypto.strong_rand_bytes(key_size)
end
defp generate_priv_key(_, _), do: <<>>
defp get_priv_key_size(:des), do: 8
defp get_priv_key_size(:aes128), do: 16
defp get_priv_key_size(:aes192), do: 24
defp get_priv_key_size(:aes256), do: 32
defp get_priv_key_size(_), do: 16
end