refactor: migrate from etcd to Oban for distributed job coordination
Replaces etcd-based distributed locking with Oban's PostgreSQL-backed job queue. This simplifies the architecture by eliminating the need for a separate etcd cluster while providing better reliability and observability. Changes: - Add Oban dependency and migration (oban_jobs table) - Create DevicePollerCoordinator and DeviceMonitorCoordinator Oban workers - Remove EtcdCoordinator and EtcdLock modules - Update application supervisor to start Oban - Configure Oban with pollers (50 workers) and monitors (50 workers) queues - Remove etcd StatefulSet from Kubernetes manifests - Update monitoring supervisor documentation Benefits: - Simpler architecture (no etcd cluster to manage) - PostgreSQL-based (uses existing database) - Built-in uniqueness prevents duplicate jobs cluster-wide - Better observability with Oban Web UI - Automatic job rescue on node crashes - Easier local development (no etcd required) What was removed: - etcd StatefulSet (3 pods) - EtcdCoordinator module (320 lines) - EtcdLock module (158 lines) - eetcd dependency All 3,686 tests passing.
This commit is contained in:
parent
979d246160
commit
29593ac734
13 changed files with 230 additions and 559 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -6,7 +6,6 @@ resources:
|
|||
- deployment.yaml
|
||||
- service.yaml
|
||||
- service-headless.yaml
|
||||
- etcd-statefulset.yaml
|
||||
- certificate.yaml
|
||||
- ingressroute.yaml
|
||||
- poddisruptionbudget.yaml
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
|
|||
92
lib/towerops/workers/device_monitor_coordinator.ex
Normal file
92
lib/towerops/workers/device_monitor_coordinator.ex
Normal file
|
|
@ -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
|
||||
96
lib/towerops/workers/device_poller_coordinator.ex
Normal file
96
lib/towerops/workers/device_poller_coordinator.ex
Normal file
|
|
@ -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
|
||||
2
mix.exs
2
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"},
|
||||
|
|
|
|||
4
mix.lock
4
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"},
|
||||
|
|
|
|||
11
priv/repo/migrations/20260124220530_add_oban_jobs_table.exs
Normal file
11
priv/repo/migrations/20260124220530_add_oban_jobs_table.exs
Normal file
|
|
@ -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
|
||||
Loading…
Add table
Reference in a new issue