diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml index 8023a00e..e26d7855 100644 --- a/k8s/deployment.yaml +++ b/k8s/deployment.yaml @@ -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 diff --git a/k8s/etcd-statefulset.yaml b/k8s/etcd-statefulset.yaml new file mode 100644 index 00000000..81080926 --- /dev/null +++ b/k8s/etcd-statefulset.yaml @@ -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 diff --git a/k8s/kustomization.yaml b/k8s/kustomization.yaml index 2b2362fc..f6fce47d 100644 --- a/k8s/kustomization.yaml +++ b/k8s/kustomization.yaml @@ -9,6 +9,7 @@ resources: - service-headless.yaml - valkey-statefulset.yaml - valkey-service.yaml + - etcd-statefulset.yaml - certificate.yaml - ingressroute.yaml - poddisruptionbudget.yaml diff --git a/lib/snmpkit/snmp_lib/manager.ex b/lib/snmpkit/snmp_lib/manager.ex index 3415d1c1..a4cb3e54 100644 --- a/lib/snmpkit/snmp_lib/manager.ex +++ b/lib/snmpkit/snmp_lib/manager.ex @@ -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}") diff --git a/lib/snmpkit/snmp_lib/pdu/decoder.ex b/lib/snmpkit/snmp_lib/pdu/decoder.ex index 710495f7..fdc0d693 100644 --- a/lib/snmpkit/snmp_lib/pdu/decoder.ex +++ b/lib/snmpkit/snmp_lib/pdu/decoder.ex @@ -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(<>) 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), diff --git a/lib/snmpkit/snmp_lib/transport.ex b/lib/snmpkit/snmp_lib/transport.ex index 7d42a59d..2201d4a6 100644 --- a/lib/snmpkit/snmp_lib/transport.ex +++ b/lib/snmpkit/snmp_lib/transport.ex @@ -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), diff --git a/lib/snmpkit/snmp_lib/types.ex b/lib/snmpkit/snmp_lib/types.ex index 9411bbf8..5b00f841 100644 --- a/lib/snmpkit/snmp_lib/types.ex +++ b/lib/snmpkit/snmp_lib/types.ex @@ -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)} diff --git a/lib/snmpkit/snmp_lib/walker.ex b/lib/snmpkit/snmp_lib/walker.ex index 180a2281..45427271 100644 --- a/lib/snmpkit/snmp_lib/walker.ex +++ b/lib/snmpkit/snmp_lib/walker.ex @@ -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 diff --git a/lib/snmpkit/snmp_mgr/bulk.ex b/lib/snmpkit/snmp_mgr/bulk.ex index 846f2c24..2e14d96d 100644 --- a/lib/snmpkit/snmp_mgr/bulk.ex +++ b/lib/snmpkit/snmp_mgr/bulk.ex @@ -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. diff --git a/lib/snmpkit/snmp_mgr/core.ex b/lib/snmpkit/snmp_mgr/core.ex index 4a8903cb..280b6232 100644 --- a/lib/snmpkit/snmp_mgr/core.ex +++ b/lib/snmpkit/snmp_mgr/core.ex @@ -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 diff --git a/lib/snmpkit/snmp_mgr/format.ex b/lib/snmpkit/snmp_mgr/format.ex index 5958371b..58c9dd64 100644 --- a/lib/snmpkit/snmp_mgr/format.ex +++ b/lib/snmpkit/snmp_mgr/format.ex @@ -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 """ diff --git a/lib/snmpkit/snmp_mgr/walk.ex b/lib/snmpkit/snmp_mgr/walk.ex index 610e0503..1b467659 100644 --- a/lib/snmpkit/snmp_mgr/walk.ex +++ b/lib/snmpkit/snmp_mgr/walk.ex @@ -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) diff --git a/lib/towerops/etcd_lock.ex b/lib/towerops/etcd_lock.ex new file mode 100644 index 00000000..9fbf8061 --- /dev/null +++ b/lib/towerops/etcd_lock.ex @@ -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 diff --git a/lib/towerops/monitoring/device_monitor.ex b/lib/towerops/monitoring/device_monitor.ex index 2c15cc70..bc1f3d54 100644 --- a/lib/towerops/monitoring/device_monitor.ex +++ b/lib/towerops/monitoring/device_monitor.ex @@ -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 diff --git a/lib/towerops/monitoring/etcd_coordinator.ex b/lib/towerops/monitoring/etcd_coordinator.ex new file mode 100644 index 00000000..1e66ffa0 --- /dev/null +++ b/lib/towerops/monitoring/etcd_coordinator.ex @@ -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 diff --git a/lib/towerops/monitoring/supervisor.ex b/lib/towerops/monitoring/supervisor.ex index 46a6e625..66e3cc56 100644 --- a/lib/towerops/monitoring/supervisor.ex +++ b/lib/towerops/monitoring/supervisor.ex @@ -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, []) diff --git a/lib/towerops/snmp/poller_worker.ex b/lib/towerops/snmp/poller_worker.ex index 0daa32d2..5203bc34 100644 --- a/lib/towerops/snmp/poller_worker.ex +++ b/lib/towerops/snmp/poller_worker.ex @@ -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 diff --git a/mix.exs b/mix.exs index 447d24b5..2a2527e5 100644 --- a/mix.exs +++ b/mix.exs @@ -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"}, diff --git a/mix.lock b/mix.lock index 07407af5..0e965e84 100644 --- a/mix.lock +++ b/mix.lock @@ -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"}, diff --git a/test/snmpkit/snmp_lib/decoder_long_length_test.exs b/test/snmpkit/snmp_lib/decoder_long_length_test.exs index fb8952c2..88bcb815 100644 --- a/test/snmpkit/snmp_lib/decoder_long_length_test.exs +++ b/test/snmpkit/snmp_lib/decoder_long_length_test.exs @@ -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: <> @@ -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 diff --git a/test/snmpkit/snmp_lib/mib/compiler_test.exs b/test/snmpkit/snmp_lib/mib/compiler_test.exs index f4c07464..6097387a 100644 --- a/test/snmpkit/snmp_lib/mib/compiler_test.exs +++ b/test/snmpkit/snmp_lib/mib/compiler_test.exs @@ -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 diff --git a/test/snmpkit/snmp_lib/mib/comprehensive_mib_test.exs b/test/snmpkit/snmp_lib/mib/comprehensive_mib_test.exs index 677c0b8f..4a320fd1 100644 --- a/test/snmpkit/snmp_lib/mib/comprehensive_mib_test.exs +++ b/test/snmpkit/snmp_lib/mib/comprehensive_mib_test.exs @@ -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 diff --git a/test/snmpkit/snmp_lib/mib/docsis_mib_test.exs b/test/snmpkit/snmp_lib/mib/docsis_mib_test.exs index 37f9dab6..9815384d 100644 --- a/test/snmpkit/snmp_lib/mib/docsis_mib_test.exs +++ b/test/snmpkit/snmp_lib/mib/docsis_mib_test.exs @@ -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} diff --git a/test/snmpkit/snmp_lib/snmpv3_integration_test.exs b/test/snmpkit/snmp_lib/snmpv3_integration_test.exs index c0f1b539..d58fe6b2 100644 --- a/test/snmpkit/snmp_lib/snmpv3_integration_test.exs +++ b/test/snmpkit/snmp_lib/snmpv3_integration_test.exs @@ -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) diff --git a/test/snmpkit/snmp_lib/v3_encoder_test.exs b/test/snmpkit/snmp_lib/v3_encoder_test.exs index 7a088d1b..857faea0 100644 --- a/test/snmpkit/snmp_lib/v3_encoder_test.exs +++ b/test/snmpkit/snmp_lib/v3_encoder_test.exs @@ -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