diff --git a/config/runtime.exs b/config/runtime.exs index f78230aa..a0205059 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -79,6 +79,23 @@ if config_env() == :prod do # Configure Swoosh to use Req for API requests config :swoosh, :api_client, Swoosh.ApiClient.Req + # Configure Oban for distributed job processing + config :towerops, Oban, + repo: Towerops.Repo, + queues: [ + default: 10, + # SNMP polling jobs - one per device + pollers: 50, + # Device monitoring jobs - health checks + monitors: 50 + ], + plugins: [ + # Automatically delete completed jobs after 60 seconds + {Oban.Plugins.Pruner, max_age: 60}, + # Rescue orphaned jobs (when node crashes) + {Oban.Plugins.Reindexer, schedule: "0 2 * * *"} + ] + # ## SSL Support # # To get SSL working, you will need to add the `https` key diff --git a/config/test.exs b/config/test.exs index fb0ae1d4..36750e7c 100644 --- a/config/test.exs +++ b/config/test.exs @@ -20,6 +20,13 @@ config :phoenix_live_view, # Disable swoosh api client as it is only required for production adapters config :swoosh, :api_client, false +# Disable Oban job processing in tests (use testing mode) +config :towerops, Oban, + repo: Towerops.Repo, + testing: :manual, + queues: false, + plugins: false + # In test we don't send emails config :towerops, Towerops.Mailer, adapter: Swoosh.Adapters.Test diff --git a/k8s/etcd-statefulset.yaml b/k8s/etcd-statefulset.yaml deleted file mode 100644 index c467438b..00000000 --- a/k8s/etcd-statefulset.yaml +++ /dev/null @@ -1,111 +0,0 @@ ---- -# 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 - publishNotReadyAddresses: true - selector: - app: etcd ---- -apiVersion: apps/v1 -kind: StatefulSet -metadata: - name: etcd - namespace: towerops -spec: - serviceName: etcd - replicas: 3 - 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,etcd-1=http://etcd-1.etcd.towerops.svc.cluster.local:2380,etcd-2=http://etcd-2.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 - startupProbe: - httpGet: - path: /health - port: 2379 - initialDelaySeconds: 10 - periodSeconds: 5 - timeoutSeconds: 3 - failureThreshold: 30 - livenessProbe: - httpGet: - path: /health - port: 2379 - initialDelaySeconds: 10 - periodSeconds: 10 - timeoutSeconds: 5 - failureThreshold: 3 - readinessProbe: - httpGet: - path: /health - port: 2379 - initialDelaySeconds: 10 - periodSeconds: 5 - timeoutSeconds: 3 - failureThreshold: 3 - volumeClaimTemplates: - - metadata: - name: etcd-data - spec: - accessModes: ["ReadWriteOnce"] - resources: - requests: - storage: 10Gi diff --git a/k8s/kustomization.yaml b/k8s/kustomization.yaml index c91c930e..1456b5b8 100644 --- a/k8s/kustomization.yaml +++ b/k8s/kustomization.yaml @@ -6,7 +6,6 @@ resources: - deployment.yaml - service.yaml - service-headless.yaml - - etcd-statefulset.yaml - certificate.yaml - ingressroute.yaml - poddisruptionbudget.yaml diff --git a/lib/towerops/application.ex b/lib/towerops/application.ex index 5b870b30..641048cb 100644 --- a/lib/towerops/application.ex +++ b/lib/towerops/application.ex @@ -25,6 +25,7 @@ defmodule Towerops.Application do {Cluster.Supervisor, [topologies, [name: Towerops.ClusterSupervisor]]}, ToweropsWeb.Telemetry, Towerops.Repo, + {Oban, Application.fetch_env!(:towerops, Oban)}, {DNSCluster, query: Application.get_env(:towerops, :dns_cluster_query) || :ignore}, pubsub_spec(), # Load YAML profiles into ETS cache diff --git a/lib/towerops/etcd_lock.ex b/lib/towerops/etcd_lock.ex deleted file mode 100644 index 9fbf8061..00000000 --- a/lib/towerops/etcd_lock.ex +++ /dev/null @@ -1,158 +0,0 @@ -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/etcd_coordinator.ex b/lib/towerops/monitoring/etcd_coordinator.ex deleted file mode 100644 index 57927bbd..00000000 --- a/lib/towerops/monitoring/etcd_coordinator.ex +++ /dev/null @@ -1,271 +0,0 @@ -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") - - # Try to create watcher for lock releases (optimizes instant failover) - # If watcher creation fails, we'll rely on periodic rebalancing instead - 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) - - watcher = - case :eetcd_watch.watch(conn, watch_with_prefix) do - {:ok, watcher} -> - Logger.info("Watching etcd for lock releases (instant failover enabled)") - watcher - - {:error, reason} -> - Logger.warning( - "Failed to create etcd watcher: #{inspect(reason)}. " <> - "Relying on periodic rebalancing instead (60s intervals)" - ) - - nil - end - - # 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 - # eetcd is an Erlang library and expects charlists, not binary strings - endpoints = [ - ~c"etcd-0.etcd.towerops.svc.cluster.local:2379", - ~c"etcd-1.etcd.towerops.svc.cluster.local:2379", - ~c"etcd-2.etcd.towerops.svc.cluster.local:2379" - ] - - case :eetcd.open(:towerops_etcd, endpoints) do - {:ok, _pid} -> - # Return the connection name (atom), not the PID - # eetcd functions expect the connection name - {:ok, :towerops_etcd} - - {:error, _reason} = error -> - error - end - 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 66e3cc56..e40c192f 100644 --- a/lib/towerops/monitoring/supervisor.ex +++ b/lib/towerops/monitoring/supervisor.ex @@ -5,7 +5,6 @@ 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 @@ -31,29 +30,20 @@ defmodule Towerops.Monitoring.Supervisor do {DynamicSupervisor, name: Towerops.LocalPollerSupervisor, strategy: :one_for_one} ] - # Only start cleanup worker and etcd coordinator in production + # Only start cleanup worker in production children = if test_mode?() do base_children else - base_children ++ [NeighborCleanupWorker] ++ etcd_coordinator() + base_children ++ [NeighborCleanupWorker] end Supervisor.init(children, strategy: :one_for_one) end - # 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 """ Starts monitoring for a specific device (node-local). - Note: In production, monitors are started by EtcdCoordinator after acquiring locks. + Note: In production, monitors are coordinated by Oban workers to ensure cluster-wide uniqueness. This function is primarily for testing and manual operations. """ def start_monitor(device_id, lease_id \\ nil) do @@ -76,7 +66,7 @@ defmodule Towerops.Monitoring.Supervisor do @doc """ Starts an SNMP poller for a specific device (node-local). - Note: In production, pollers are started by EtcdCoordinator after acquiring locks. + Note: In production, pollers are coordinated by Oban workers to ensure cluster-wide uniqueness. This function is primarily for testing and manual operations. """ def start_snmp_poller(device_id, lease_id \\ nil) do diff --git a/lib/towerops/workers/device_monitor_coordinator.ex b/lib/towerops/workers/device_monitor_coordinator.ex new file mode 100644 index 00000000..613d9188 --- /dev/null +++ b/lib/towerops/workers/device_monitor_coordinator.ex @@ -0,0 +1,92 @@ +defmodule Towerops.Workers.DeviceMonitorCoordinator do + @moduledoc """ + Oban worker that coordinates device monitoring across the cluster. + + Uses Oban's unique job feature to ensure only one monitor per device + runs cluster-wide, replacing etcd-based locking. + """ + use Oban.Worker, + queue: :monitors, + unique: [ + period: 60, + keys: [:device_id], + states: [:available, :scheduled, :executing, :retryable] + ] + + alias Towerops.Devices + alias Towerops.Monitoring.Supervisor, as: MonitoringSupervisor + + require Logger + + # Monitoring interval - check every 30 seconds + @monitor_interval 30 + + @impl Oban.Worker + def perform(%Oban.Job{args: %{"device_id" => device_id}}) do + case Devices.get_device(device_id) do + nil -> + Logger.debug("Device #{device_id} no longer exists, skipping monitor coordination") + :ok + + _device -> + ensure_monitor_running(device_id) + + # Reschedule for next check + schedule_next_check(device_id) + + :ok + end + end + + @doc """ + Starts monitoring coordination for a device. + """ + def start_monitoring(device_id) do + %{device_id: device_id} + |> new() + |> Oban.insert() + end + + @doc """ + Stops monitoring for a device by cancelling its jobs. + """ + def stop_monitoring(device_id) do + import Ecto.Query + + # Cancel all pending jobs for this device + Oban.cancel_all_jobs( + from(j in Oban.Job, + where: j.worker == "Towerops.Workers.DeviceMonitorCoordinator", + where: fragment("args->>'device_id' = ?", ^device_id), + where: j.state in ["available", "scheduled", "executing", "retryable"] + ) + ) + + # Stop the running monitor GenServer + MonitoringSupervisor.stop_monitor(device_id) + end + + # Private functions + + defp ensure_monitor_running(device_id) do + case MonitoringSupervisor.start_monitor(device_id) do + {:ok, _pid} -> + Logger.debug("Started monitor for device #{device_id}") + :ok + + {:error, {:already_started, _pid}} -> + # Already running, which is fine + :ok + + {:error, reason} -> + Logger.warning("Failed to start monitor for device #{device_id}: #{inspect(reason)}") + :ok + end + end + + defp schedule_next_check(device_id) do + %{device_id: device_id} + |> new(schedule_in: @monitor_interval) + |> Oban.insert() + end +end diff --git a/lib/towerops/workers/device_poller_coordinator.ex b/lib/towerops/workers/device_poller_coordinator.ex new file mode 100644 index 00000000..b6972f01 --- /dev/null +++ b/lib/towerops/workers/device_poller_coordinator.ex @@ -0,0 +1,96 @@ +defmodule Towerops.Workers.DevicePollerCoordinator do + @moduledoc """ + Oban worker that coordinates SNMP polling across the cluster. + + Uses Oban's unique job feature to ensure only one poller per device + runs cluster-wide, replacing etcd-based locking. + """ + use Oban.Worker, + queue: :pollers, + unique: [ + period: 60, + keys: [:device_id], + states: [:available, :scheduled, :executing, :retryable] + ] + + alias Towerops.Devices + alias Towerops.Monitoring.Supervisor, as: MonitoringSupervisor + + require Logger + + @impl Oban.Worker + def perform(%Oban.Job{args: %{"device_id" => device_id}}) do + case Devices.get_device(device_id) do + nil -> + Logger.debug("Device #{device_id} no longer exists, skipping poll coordination") + :ok + + device -> + if device.snmp_enabled do + ensure_poller_running(device_id) + end + + # Reschedule for next poll interval (default 60s) + poll_interval = get_poll_interval(device) + schedule_next_poll(device_id, poll_interval) + + :ok + end + end + + @doc """ + Starts polling coordination for a device. + """ + def start_polling(device_id) do + %{device_id: device_id} + |> new() + |> Oban.insert() + end + + @doc """ + Stops polling for a device by cancelling its jobs. + """ + def stop_polling(device_id) do + import Ecto.Query + + # Cancel all pending jobs for this device + Oban.cancel_all_jobs( + from(j in Oban.Job, + where: j.worker == "Towerops.Workers.DevicePollerCoordinator", + where: fragment("args->>'device_id' = ?", ^device_id), + where: j.state in ["available", "scheduled", "executing", "retryable"] + ) + ) + + # Stop the running poller GenServer + MonitoringSupervisor.stop_snmp_poller(device_id) + end + + # Private functions + + defp ensure_poller_running(device_id) do + case MonitoringSupervisor.start_snmp_poller(device_id) do + {:ok, _pid} -> + Logger.debug("Started SNMP poller for device #{device_id}") + :ok + + {:error, {:already_started, _pid}} -> + # Already running, which is fine + :ok + + {:error, reason} -> + Logger.warning("Failed to start SNMP poller for device #{device_id}: #{inspect(reason)}") + :ok + end + end + + defp schedule_next_poll(device_id, interval_seconds) do + %{device_id: device_id} + |> new(schedule_in: interval_seconds) + |> Oban.insert() + end + + defp get_poll_interval(device) do + device.snmp_poll_interval || 60 + end +end diff --git a/mix.exs b/mix.exs index 2a2527e5..a7488959 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"}, - {:eetcd, "~> 0.6"}, + {:oban, "~> 2.18"}, {:bandit, "~> 1.5"}, {:phoenix_pubsub_redis, "~> 3.0"}, {:ecto_psql_extras, "~> 0.6"}, diff --git a/mix.lock b/mix.lock index 1de8fcde..000e206f 100644 --- a/mix.lock +++ b/mix.lock @@ -5,7 +5,6 @@ "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,7 +13,6 @@ "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"}, @@ -26,7 +24,6 @@ "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"}, "hpax": {:hex, :hpax, "1.0.3", "ed67ef51ad4df91e75cc6a1494f851850c0bd98ebc0be6e81b026e765ee535aa", [:mix], [], "hexpm", "8eab6e1cfa8d5918c2ce4ba43588e894af35dbd8e91e6e55c817bca5847df34a"}, @@ -42,6 +39,7 @@ "nimble_options": {:hex, :nimble_options, "1.1.1", "e3a492d54d85fc3fd7c5baf411d9d2852922f66e69476317787a7b2bb000a61b", [:mix], [], "hexpm", "821b2470ca9442c4b6984882fe9bb0389371b8ddec4d45a9504f00a66f650b44"}, "nimble_ownership": {:hex, :nimble_ownership, "1.0.2", "fa8a6f2d8c592ad4d79b2ca617473c6aefd5869abfa02563a77682038bf916cf", [:mix], [], "hexpm", "098af64e1f6f8609c6672127cfe9e9590a5d3fcdd82bc17a377b8692fd81a879"}, "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"}, + "oban": {:hex, :oban, "2.20.3", "e4d27336941955886cc7113420c32c63b70b64f10b27e08e3cf2b001153953cd", [:mix], [{:ecto_sql, "~> 3.10", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:ecto_sqlite3, "~> 0.9", [hex: :ecto_sqlite3, repo: "hexpm", optional: true]}, {:igniter, "~> 0.5", [hex: :igniter, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:myxql, "~> 0.7", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.20", [hex: :postgrex, repo: "hexpm", optional: true]}, {:telemetry, "~> 1.3", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "075ffbf1279a96bec495bc63d647b08929837d70bcc0427249ffe4d1dddaec33"}, "phoenix": {:hex, :phoenix, "1.8.3", "49ac5e485083cb1495a905e47eb554277bdd9c65ccb4fc5100306b350151aa95", [:mix], [{:bandit, "~> 1.0", [hex: :bandit, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 2.1", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:phoenix_template, "~> 1.0", [hex: :phoenix_template, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 2.0", [hex: :phoenix_view, repo: "hexpm", optional: true]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.7", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:plug_crypto, "~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:websock_adapter, "~> 0.5.3", [hex: :websock_adapter, repo: "hexpm", optional: false]}], "hexpm", "36169f95cc2e155b78be93d9590acc3f462f1e5438db06e6248613f27c80caec"}, "phoenix_ecto": {:hex, :phoenix_ecto, "4.7.0", "75c4b9dfb3efdc42aec2bd5f8bccd978aca0651dbcbc7a3f362ea5d9d43153c6", [:mix], [{:ecto, "~> 3.5", [hex: :ecto, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 2.14.2 or ~> 3.0 or ~> 4.1", [hex: :phoenix_html, repo: "hexpm", optional: true]}, {:plug, "~> 1.9", [hex: :plug, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.16 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}], "hexpm", "1d75011e4254cb4ddf823e81823a9629559a1be93b4321a6a5f11a5306fbf4cc"}, "phoenix_html": {:hex, :phoenix_html, "4.3.0", "d3577a5df4b6954cd7890c84d955c470b5310bb49647f0a114a6eeecc850f7ad", [:mix], [], "hexpm", "3eaa290a78bab0f075f791a46a981bbe769d94bc776869f4f3063a14f30497ad"}, diff --git a/priv/repo/migrations/20260124220530_add_oban_jobs_table.exs b/priv/repo/migrations/20260124220530_add_oban_jobs_table.exs new file mode 100644 index 00000000..48983da5 --- /dev/null +++ b/priv/repo/migrations/20260124220530_add_oban_jobs_table.exs @@ -0,0 +1,11 @@ +defmodule Towerops.Repo.Migrations.AddObanJobsTable do + use Ecto.Migration + + def up do + Oban.Migration.up(version: 12) + end + + def down do + Oban.Migration.down(version: 12) + end +end