filter more honeybadger alerts, format dates to users time zone, email template cleanup
This commit is contained in:
parent
b08daf9a88
commit
b30f2cf5af
16 changed files with 976 additions and 23 deletions
814
docs/FUTURE_IMPROVEMENTS.md
Normal file
814
docs/FUTURE_IMPROVEMENTS.md
Normal file
|
|
@ -0,0 +1,814 @@
|
|||
# Future Improvements
|
||||
|
||||
## Broadway + Oban: High-Volume Metric Ingestion
|
||||
|
||||
### Status: Planning
|
||||
**Target**: When reaching 100+ agents or experiencing API bottlenecks
|
||||
**Effort**: Medium (2-3 weeks)
|
||||
**Priority**: Medium (not urgent with current scale)
|
||||
|
||||
---
|
||||
|
||||
## Problem Statement
|
||||
|
||||
Current architecture uses synchronous agent metric submission:
|
||||
- Agent POSTs metrics → API validates → Direct DB insert → 200 OK
|
||||
- At scale (100s of agents, 1000s of devices), this creates:
|
||||
- **Thundering herd**: All agents polling on similar intervals
|
||||
- **API timeouts**: DB writes block the HTTP response
|
||||
- **Poor throughput**: One DB transaction per agent request
|
||||
- **No backpressure**: Peaks overwhelm the system
|
||||
|
||||
### Current Performance Profile
|
||||
|
||||
```
|
||||
500 agents × 60s interval = 8.3 req/sec average
|
||||
Peak (synchronized): 500 concurrent requests
|
||||
Each request: 10-50ms (DB insert + processing)
|
||||
Risk: Thundering herd, timeouts during peaks
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Proposed Solution: Broadway Message Processing Pipeline
|
||||
|
||||
### Architecture Overview
|
||||
|
||||
```
|
||||
┌─────────────────────────────────────────────────────────────────┐
|
||||
│ CURRENT (Oban) │
|
||||
├─────────────────────────────────────────────────────────────────┤
|
||||
│ DevicePollerWorker (Oban) ──┐ │
|
||||
│ DeviceMonitorWorker (Oban) ─┼─→ Direct SNMP → DB writes │
|
||||
│ Maintenance Workers (Oban) ─┘ │
|
||||
└─────────────────────────────────────────────────────────────────┘
|
||||
|
||||
┌─────────────────────────────────────────────────────────────────┐
|
||||
│ WITH BROADWAY (Proposed) │
|
||||
├─────────────────────────────────────────────────────────────────┤
|
||||
│ Agent POST /metrics ──→ Message Queue ──→ 202 Accepted │
|
||||
│ ↓ │
|
||||
│ Broadway Pipeline │
|
||||
│ ├─ Consumer (50 workers) │
|
||||
│ ├─ Batcher (1000 msgs) │
|
||||
│ └─ Bulk DB Insert │
|
||||
│ │
|
||||
│ Oban continues: │
|
||||
│ ├─ DevicePollerWorker (scheduled SNMP) │
|
||||
│ ├─ DeviceMonitorWorker (health checks) │
|
||||
│ └─ Maintenance workers (cleanup, etc.) │
|
||||
└─────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
### Role Separation
|
||||
|
||||
**Broadway** (async, high-volume, batchable):
|
||||
- ✅ Agent metric submissions (`POST /api/v1/agent/metrics`)
|
||||
- ✅ Agent heartbeats (optional - could batch these too)
|
||||
- ✅ High-throughput data ingestion
|
||||
- ✅ Batch database operations
|
||||
|
||||
**Oban** (scheduled, low-frequency, complex logic):
|
||||
- ✅ DevicePollerWorker (scheduled SNMP polling)
|
||||
- ✅ DeviceMonitorWorker (health checks)
|
||||
- ✅ Maintenance workers (cleanup, evaluators)
|
||||
- ✅ Discovery jobs (complex multi-step operations)
|
||||
|
||||
### Performance Characteristics After Broadway
|
||||
|
||||
```
|
||||
Agent POST: <5ms (just enqueue)
|
||||
Broadway pipeline:
|
||||
- 50 processor workers (parse/validate)
|
||||
- 10 batcher workers (1000 rows/batch)
|
||||
- Single INSERT for 1000 rows: ~50-100ms
|
||||
Throughput: 10k-50k metrics/sec (limited by DB, not API)
|
||||
Backpressure: Queue absorbs burst traffic
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Message Queue Options
|
||||
|
||||
### Option A: PostgreSQL-backed (Recommended Start)
|
||||
|
||||
**Pros**:
|
||||
- ✅ No new infrastructure (use existing PostgreSQL)
|
||||
- ✅ Transactional guarantees
|
||||
- ✅ Simple deployment
|
||||
- ✅ Easy development/testing
|
||||
|
||||
**Cons**:
|
||||
- ❌ Not ideal for extremely high throughput (10k+ msg/sec)
|
||||
- ❌ Adds load to primary database
|
||||
|
||||
**Implementation**: Custom Broadway producer reading from `agent_metrics_queue` table
|
||||
|
||||
**When to use**: Initial implementation, <100 agents, proof of concept
|
||||
|
||||
---
|
||||
|
||||
### Option B: RabbitMQ (Recommended Production)
|
||||
|
||||
**Pros**:
|
||||
- ✅ Battle-tested for high throughput
|
||||
- ✅ Built-in clustering and persistence
|
||||
- ✅ Management UI for monitoring
|
||||
- ✅ Proven at scale (100k+ msg/sec)
|
||||
|
||||
**Cons**:
|
||||
- ❌ Adds infrastructure dependency
|
||||
- ❌ Another service to monitor
|
||||
- ❌ Requires Kubernetes setup
|
||||
|
||||
**Implementation**: `broadway_rabbitmq` package with existing RabbitMQ cluster
|
||||
|
||||
**When to use**: 100+ agents, need guaranteed throughput, production workloads
|
||||
|
||||
---
|
||||
|
||||
### Option C: AWS SQS
|
||||
|
||||
**Pros**:
|
||||
- ✅ Fully managed, scales automatically
|
||||
- ✅ No infrastructure to maintain
|
||||
- ✅ Pay-per-use pricing
|
||||
|
||||
**Cons**:
|
||||
- ❌ Requires AWS
|
||||
- ❌ Potential latency (network calls)
|
||||
- ❌ Visibility timeout complexity
|
||||
|
||||
**Implementation**: `off_broadway_sqs` package
|
||||
|
||||
**When to use**: Running on AWS, want fully managed solution
|
||||
|
||||
---
|
||||
|
||||
## Implementation Details
|
||||
|
||||
### 1. Dependencies
|
||||
|
||||
```elixir
|
||||
# mix.exs
|
||||
defp deps do
|
||||
[
|
||||
{:broadway, "~> 1.0"},
|
||||
|
||||
# Choose one:
|
||||
{:broadway_rabbitmq, "~> 0.8"}, # Option B
|
||||
{:off_broadway_sqs, "~> 0.7"}, # Option C
|
||||
# Option A: custom producer (see below)
|
||||
|
||||
# ... existing deps
|
||||
]
|
||||
end
|
||||
```
|
||||
|
||||
### 2. Broadway Pipeline
|
||||
|
||||
```elixir
|
||||
# lib/towerops/pipelines/agent_metrics_pipeline.ex
|
||||
defmodule Towerops.Pipelines.AgentMetricsPipeline do
|
||||
use Broadway
|
||||
|
||||
alias Broadway.Message
|
||||
alias Towerops.Snmp
|
||||
|
||||
def start_link(_opts) do
|
||||
Broadway.start_link(__MODULE__,
|
||||
name: __MODULE__,
|
||||
producer: [
|
||||
module: {
|
||||
BroadwayRabbitMQ.Producer,
|
||||
queue: "agent_metrics",
|
||||
connection: [
|
||||
host: Application.get_env(:towerops, :rabbitmq_host),
|
||||
username: Application.get_env(:towerops, :rabbitmq_username),
|
||||
password: Application.get_env(:towerops, :rabbitmq_password)
|
||||
],
|
||||
on_failure: :reject_and_requeue
|
||||
},
|
||||
concurrency: 1
|
||||
],
|
||||
processors: [
|
||||
default: [
|
||||
concurrency: 50 # Parallel message processing
|
||||
]
|
||||
],
|
||||
batchers: [
|
||||
default: [
|
||||
batch_size: 1000, # Max messages per batch
|
||||
batch_timeout: 2_000, # Max wait time (ms)
|
||||
concurrency: 10 # Parallel batch processors
|
||||
]
|
||||
]
|
||||
)
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_message(_, %Message{data: data} = message, _context) do
|
||||
# Decode and validate the message
|
||||
# No DB writes here - just prepare data
|
||||
metric = Jason.decode!(data)
|
||||
|
||||
message
|
||||
|> Message.update_data(fn _ -> metric end)
|
||||
|> Message.put_batcher(:default)
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_batch(:default, messages, _batch_info, _context) do
|
||||
# Extract all metrics from the batch
|
||||
metrics = Enum.map(messages, & &1.data)
|
||||
|
||||
# Single multi-row insert for entire batch
|
||||
case Snmp.insert_metrics_batch(metrics) do
|
||||
{:ok, _} ->
|
||||
messages # All successful
|
||||
|
||||
{:error, reason} ->
|
||||
# Mark all messages as failed (will be retried)
|
||||
Enum.map(messages, &Message.failed(&1, reason))
|
||||
end
|
||||
end
|
||||
end
|
||||
```
|
||||
|
||||
### 3. Batch Insert Function
|
||||
|
||||
```elixir
|
||||
# lib/towerops/snmp.ex
|
||||
defmodule Towerops.Snmp do
|
||||
alias Towerops.Repo
|
||||
alias Towerops.Snmp.SensorReading
|
||||
|
||||
@doc """
|
||||
Insert a batch of sensor readings in a single transaction.
|
||||
|
||||
Expected input format from agents:
|
||||
[
|
||||
%{
|
||||
"equipment_id" => "uuid",
|
||||
"sensor_id" => "uuid",
|
||||
"value" => 42.5,
|
||||
"timestamp" => "2026-02-01T12:00:00Z"
|
||||
},
|
||||
...
|
||||
]
|
||||
"""
|
||||
def insert_metrics_batch(metrics) when is_list(metrics) do
|
||||
now = DateTime.utc_now()
|
||||
|
||||
# Transform to Ecto schema format
|
||||
entries = Enum.map(metrics, fn metric ->
|
||||
%{
|
||||
equipment_id: metric["equipment_id"],
|
||||
sensor_id: metric["sensor_id"],
|
||||
value: Decimal.new(metric["value"]),
|
||||
timestamp: parse_timestamp(metric["timestamp"]),
|
||||
inserted_at: now,
|
||||
updated_at: now
|
||||
}
|
||||
end)
|
||||
|
||||
# Single INSERT with multiple rows
|
||||
Repo.insert_all(SensorReading, entries,
|
||||
on_conflict: :nothing,
|
||||
returning: false
|
||||
)
|
||||
end
|
||||
|
||||
defp parse_timestamp(ts) when is_binary(ts) do
|
||||
case DateTime.from_iso8601(ts) do
|
||||
{:ok, dt, _} -> dt
|
||||
_ -> DateTime.utc_now()
|
||||
end
|
||||
end
|
||||
end
|
||||
```
|
||||
|
||||
### 4. Async API Endpoint
|
||||
|
||||
```elixir
|
||||
# lib/towerops_web/controllers/api/v1/agent_controller.ex
|
||||
defmodule ToweropsWeb.API.V1.AgentController do
|
||||
use ToweropsWeb, :controller
|
||||
|
||||
require Logger
|
||||
|
||||
# Current synchronous version (keep during transition)
|
||||
def metrics_sync(conn, %{"metrics" => metrics}) do
|
||||
# ... existing implementation
|
||||
end
|
||||
|
||||
# New async version with Broadway
|
||||
def metrics_async(conn, %{"metrics" => metrics}) do
|
||||
agent = conn.assigns.current_agent
|
||||
|
||||
# Quick validation only
|
||||
with :ok <- validate_agent_active(agent),
|
||||
:ok <- validate_metrics_schema(metrics) do
|
||||
|
||||
# Enqueue to message queue (non-blocking)
|
||||
message = Jason.encode!(%{
|
||||
agent_id: agent.id,
|
||||
organization_id: agent.organization_id,
|
||||
metrics: metrics,
|
||||
received_at: DateTime.utc_now()
|
||||
})
|
||||
|
||||
# This is fast - just writes to queue
|
||||
:ok = publish_to_queue("agent_metrics", message)
|
||||
|
||||
# Return immediately
|
||||
conn
|
||||
|> put_status(:accepted)
|
||||
|> json(%{status: "accepted", queued: length(metrics)})
|
||||
else
|
||||
{:error, reason} ->
|
||||
conn
|
||||
|> put_status(:bad_request)
|
||||
|> json(%{error: reason})
|
||||
end
|
||||
end
|
||||
|
||||
defp publish_to_queue(queue, message) do
|
||||
# RabbitMQ example
|
||||
AMQP.Basic.publish(
|
||||
ToweropsWeb.RabbitMQ.channel(),
|
||||
"",
|
||||
queue,
|
||||
message,
|
||||
persistent: true
|
||||
)
|
||||
end
|
||||
|
||||
defp validate_agent_active(%{active: true}), do: :ok
|
||||
defp validate_agent_active(_), do: {:error, "agent inactive"}
|
||||
|
||||
defp validate_metrics_schema(metrics) when is_list(metrics) do
|
||||
# Quick schema validation only
|
||||
if Enum.all?(metrics, &valid_metric?/1) do
|
||||
:ok
|
||||
else
|
||||
{:error, "invalid metric schema"}
|
||||
end
|
||||
end
|
||||
|
||||
defp valid_metric?(%{"equipment_id" => _, "sensor_id" => _, "value" => _}), do: true
|
||||
defp valid_metric?(_), do: false
|
||||
end
|
||||
```
|
||||
|
||||
### 5. Application Supervision Tree
|
||||
|
||||
```elixir
|
||||
# lib/towerops/application.ex
|
||||
defmodule Towerops.Application do
|
||||
def start(_type, _args) do
|
||||
children = [
|
||||
# ... existing children (Repo, PubSub, Endpoint, Oban, etc.)
|
||||
|
||||
# Add Broadway pipeline
|
||||
Towerops.Pipelines.AgentMetricsPipeline,
|
||||
]
|
||||
|
||||
opts = [strategy: :one_for_one, name: Towerops.Supervisor]
|
||||
Supervisor.start_link(children, opts)
|
||||
end
|
||||
end
|
||||
```
|
||||
|
||||
### 6. Router Updates
|
||||
|
||||
```elixir
|
||||
# lib/towerops_web/router.ex
|
||||
scope "/api/v1/agent", ToweropsWeb.API.V1 do
|
||||
pipe_through [:api, :api_agent_auth]
|
||||
|
||||
# Phase 1: Dual endpoints during migration
|
||||
post "/metrics", AgentController, :metrics_sync # Current
|
||||
post "/metrics/async", AgentController, :metrics_async # New
|
||||
|
||||
# Phase 3: After full migration
|
||||
# post "/metrics", AgentController, :metrics_async
|
||||
end
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Migration Path
|
||||
|
||||
### Phase 1: Dual Mode (Parallel Run)
|
||||
|
||||
**Duration**: 2-4 weeks
|
||||
**Goal**: Validate Broadway pipeline with real traffic
|
||||
|
||||
1. Deploy both endpoints (`/metrics` and `/metrics/async`)
|
||||
2. Update 10% of agents to use `/metrics/async`
|
||||
3. Monitor:
|
||||
- Queue depth
|
||||
- Processing latency
|
||||
- Error rates
|
||||
- Database performance
|
||||
|
||||
**Success Criteria**:
|
||||
- Zero data loss
|
||||
- <100ms p95 latency for queue insertion
|
||||
- <5s p95 latency for batch processing
|
||||
- No queue backlog growth
|
||||
|
||||
---
|
||||
|
||||
### Phase 2: Gradual Rollout
|
||||
|
||||
**Duration**: 2-4 weeks
|
||||
**Goal**: Full traffic migration
|
||||
|
||||
1. Increase to 25% of agents
|
||||
2. Monitor for 1 week
|
||||
3. Increase to 50% of agents
|
||||
4. Monitor for 1 week
|
||||
5. Increase to 100% of agents
|
||||
|
||||
**Rollback Plan**:
|
||||
- Keep sync endpoint active
|
||||
- Simple config change in agent to switch back
|
||||
- No data loss (queue persists messages)
|
||||
|
||||
---
|
||||
|
||||
### Phase 3: Deprecate Synchronous Endpoint
|
||||
|
||||
**Duration**: 1 week
|
||||
**Goal**: Clean up legacy code
|
||||
|
||||
1. Switch `/metrics` route to async handler
|
||||
2. Monitor for 1 week
|
||||
3. Remove `metrics_sync/2` function
|
||||
4. Update documentation
|
||||
|
||||
---
|
||||
|
||||
## Infrastructure Requirements
|
||||
|
||||
### PostgreSQL-backed (Phase 1)
|
||||
|
||||
**Database Schema**:
|
||||
```sql
|
||||
CREATE TABLE agent_metrics_queue (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
payload JSONB NOT NULL,
|
||||
inserted_at TIMESTAMP NOT NULL DEFAULT NOW(),
|
||||
processed_at TIMESTAMP,
|
||||
retry_count INTEGER DEFAULT 0,
|
||||
error TEXT
|
||||
);
|
||||
|
||||
CREATE INDEX idx_agent_metrics_queue_unprocessed
|
||||
ON agent_metrics_queue (inserted_at)
|
||||
WHERE processed_at IS NULL;
|
||||
```
|
||||
|
||||
**No additional services required**
|
||||
|
||||
---
|
||||
|
||||
### RabbitMQ (Production)
|
||||
|
||||
**Development** (`docker-compose.yml`):
|
||||
```yaml
|
||||
services:
|
||||
rabbitmq:
|
||||
image: rabbitmq:3.13-management-alpine
|
||||
ports:
|
||||
- "5672:5672" # AMQP
|
||||
- "15672:15672" # Management UI
|
||||
environment:
|
||||
RABBITMQ_DEFAULT_USER: towerops
|
||||
RABBITMQ_DEFAULT_PASS: ${RABBITMQ_PASSWORD}
|
||||
volumes:
|
||||
- rabbitmq_data:/var/lib/rabbitmq
|
||||
|
||||
volumes:
|
||||
rabbitmq_data:
|
||||
```
|
||||
|
||||
**Production** (Kubernetes):
|
||||
```yaml
|
||||
# k8s/rabbitmq/helmrelease.yaml
|
||||
apiVersion: helm.toolkit.fluxcd.io/v2beta1
|
||||
kind: HelmRelease
|
||||
metadata:
|
||||
name: rabbitmq
|
||||
namespace: towerops
|
||||
spec:
|
||||
chart:
|
||||
spec:
|
||||
chart: rabbitmq
|
||||
version: 12.x.x
|
||||
sourceRef:
|
||||
kind: HelmRepository
|
||||
name: bitnami
|
||||
values:
|
||||
auth:
|
||||
username: towerops
|
||||
existingPasswordSecret: rabbitmq-password
|
||||
persistence:
|
||||
enabled: true
|
||||
size: 20Gi
|
||||
clustering:
|
||||
enabled: true
|
||||
replicaCount: 3
|
||||
metrics:
|
||||
enabled: true
|
||||
serviceMonitor:
|
||||
enabled: true
|
||||
```
|
||||
|
||||
**Secrets** (1Password → Kubernetes):
|
||||
```bash
|
||||
# Create secret from 1Password
|
||||
kubectl create secret generic rabbitmq-password \
|
||||
--from-literal=rabbitmq-password=$(op read "op://towerops/rabbitmq/password") \
|
||||
-n towerops
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Monitoring & Observability
|
||||
|
||||
### Telemetry Metrics
|
||||
|
||||
```elixir
|
||||
# lib/towerops_web/telemetry.ex
|
||||
defp periodic_measurements do
|
||||
[
|
||||
# Existing metrics...
|
||||
|
||||
# Broadway pipeline metrics
|
||||
{Broadway, :metrics, [Towerops.Pipelines.AgentMetricsPipeline]},
|
||||
|
||||
# Queue depth (RabbitMQ)
|
||||
{ToweropsWeb.Metrics, :queue_depth, ["agent_metrics"]},
|
||||
|
||||
# Processing rate
|
||||
{ToweropsWeb.Metrics, :processing_rate, []},
|
||||
]
|
||||
end
|
||||
```
|
||||
|
||||
### LiveDashboard Additions
|
||||
|
||||
Add to `/admin/dashboard`:
|
||||
- **Queue Depth** - Messages waiting to be processed
|
||||
- **Processing Rate** - Messages/second throughput
|
||||
- **Batch Sizes** - Average messages per batch
|
||||
- **Failed Messages** - Retry queue depth
|
||||
- **Latency** - p50/p95/p99 processing time
|
||||
|
||||
### Alerts
|
||||
|
||||
Add to monitoring system (Prometheus/Grafana):
|
||||
- Queue depth > 10,000 messages (backlog warning)
|
||||
- Processing rate < 100 msg/sec (pipeline degraded)
|
||||
- Failed message rate > 5% (data quality issue)
|
||||
- Consumer crashes (Broadway supervisor restarts)
|
||||
|
||||
---
|
||||
|
||||
## Testing Strategy
|
||||
|
||||
### Unit Tests
|
||||
|
||||
```elixir
|
||||
# test/towerops/pipelines/agent_metrics_pipeline_test.exs
|
||||
defmodule Towerops.Pipelines.AgentMetricsPipelineTest do
|
||||
use Towerops.DataCase
|
||||
|
||||
alias Broadway.Message
|
||||
alias Towerops.Pipelines.AgentMetricsPipeline
|
||||
|
||||
describe "handle_message/3" do
|
||||
test "parses valid metric JSON" do
|
||||
data = Jason.encode!(%{
|
||||
"equipment_id" => "uuid",
|
||||
"sensor_id" => "uuid",
|
||||
"value" => 42.5
|
||||
})
|
||||
|
||||
message = %Message{data: data}
|
||||
result = AgentMetricsPipeline.handle_message(:default, message, %{})
|
||||
|
||||
assert result.batcher == :default
|
||||
assert result.data["value"] == 42.5
|
||||
end
|
||||
end
|
||||
|
||||
describe "handle_batch/4" do
|
||||
test "inserts batch of metrics" do
|
||||
messages = [
|
||||
%Message{data: %{"equipment_id" => "uuid1", ...}},
|
||||
%Message{data: %{"equipment_id" => "uuid2", ...}}
|
||||
]
|
||||
|
||||
result = AgentMetricsPipeline.handle_batch(:default, messages, %{}, %{})
|
||||
|
||||
assert length(result) == 2
|
||||
assert Enum.all?(result, &match?(%Message{status: :ok}, &1))
|
||||
end
|
||||
end
|
||||
end
|
||||
```
|
||||
|
||||
### Integration Tests
|
||||
|
||||
```elixir
|
||||
# test/towerops_web/controllers/api/v1/agent_controller_test.exs
|
||||
describe "POST /api/v1/agent/metrics/async" do
|
||||
test "enqueues metrics and returns 202", %{conn: conn, agent: agent} do
|
||||
metrics = [%{
|
||||
"equipment_id" => equipment.id,
|
||||
"sensor_id" => sensor.id,
|
||||
"value" => 42.5,
|
||||
"timestamp" => DateTime.utc_now() |> DateTime.to_iso8601()
|
||||
}]
|
||||
|
||||
conn = post(conn, ~p"/api/v1/agent/metrics/async", %{"metrics" => metrics})
|
||||
|
||||
assert %{"status" => "accepted", "queued" => 1} = json_response(conn, 202)
|
||||
|
||||
# Verify message in queue
|
||||
assert_queue_length("agent_metrics", 1)
|
||||
end
|
||||
end
|
||||
```
|
||||
|
||||
### Load Tests
|
||||
|
||||
```elixir
|
||||
# test/load/agent_metrics_load_test.exs
|
||||
# Use :fuse or similar to simulate 500 concurrent agents
|
||||
defmodule Towerops.Load.AgentMetricsTest do
|
||||
use ExUnit.Case
|
||||
|
||||
@tag timeout: :infinity
|
||||
@tag :load_test
|
||||
test "handles 500 concurrent agent submissions" do
|
||||
agents = create_agents(500)
|
||||
|
||||
# Simulate all agents posting at once
|
||||
tasks = Enum.map(agents, fn agent ->
|
||||
Task.async(fn ->
|
||||
submit_metrics(agent, generate_metrics(10))
|
||||
end)
|
||||
end)
|
||||
|
||||
results = Task.await_many(tasks, 30_000)
|
||||
|
||||
# All should succeed
|
||||
assert Enum.all?(results, &match?({:ok, _}, &1))
|
||||
|
||||
# Verify queue processed everything
|
||||
wait_for_queue_empty("agent_metrics", timeout: 60_000)
|
||||
|
||||
# Verify all metrics in database
|
||||
assert Repo.aggregate(SensorReading, :count) == 5000
|
||||
end
|
||||
end
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Performance Tuning
|
||||
|
||||
### Database Optimization
|
||||
|
||||
**Batch Insert Settings**:
|
||||
```elixir
|
||||
# config/runtime.exs
|
||||
config :towerops, Towerops.Repo,
|
||||
# Increase connection pool for Broadway workers
|
||||
pool_size: 20, # Up from 10
|
||||
|
||||
# Optimize for batch inserts
|
||||
prepare: :unnamed,
|
||||
timeout: 30_000
|
||||
```
|
||||
|
||||
**TimescaleDB Tuning**:
|
||||
```sql
|
||||
-- Increase batch size for continuous aggregates
|
||||
ALTER MATERIALIZED VIEW sensor_readings_1h SET (
|
||||
timescaledb.materialized_only = true,
|
||||
timescaledb.compress_after = '7 days'
|
||||
);
|
||||
|
||||
-- Add indexes for common queries
|
||||
CREATE INDEX CONCURRENTLY idx_sensor_readings_equipment_timestamp
|
||||
ON sensor_readings (equipment_id, timestamp DESC);
|
||||
```
|
||||
|
||||
### Broadway Tuning
|
||||
|
||||
```elixir
|
||||
# Processors: CPU-bound (parsing, validation)
|
||||
# Start with: processors_concurrency = num_cores * 2
|
||||
processors: [default: [concurrency: System.schedulers_online() * 2]]
|
||||
|
||||
# Batchers: I/O-bound (database writes)
|
||||
# Start with: batch_size = 1000, tune based on DB performance
|
||||
batchers: [
|
||||
default: [
|
||||
batch_size: 1000, # Increase if DB can handle larger batches
|
||||
batch_timeout: 2_000, # Decrease for lower latency
|
||||
concurrency: 10 # Increase if batchers become bottleneck
|
||||
]
|
||||
]
|
||||
```
|
||||
|
||||
### RabbitMQ Tuning
|
||||
|
||||
```elixir
|
||||
# Prefetch: How many unacked messages per consumer
|
||||
# Higher = better throughput, lower = better distribution
|
||||
connection: [
|
||||
prefetch_count: 100 # Start with 100, tune based on message size
|
||||
]
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Cost Analysis
|
||||
|
||||
### Infrastructure Costs (Estimated)
|
||||
|
||||
**PostgreSQL-backed**:
|
||||
- $0/month (uses existing database)
|
||||
- Adds ~10-20% CPU load to DB server
|
||||
|
||||
**RabbitMQ (Self-hosted Kubernetes)**:
|
||||
- 3 nodes × 2 vCPU × 4 GB RAM = ~$120/month
|
||||
- 20 GB persistent storage = ~$10/month
|
||||
- **Total**: ~$130/month
|
||||
|
||||
**AWS SQS**:
|
||||
- 1M requests = $0.40
|
||||
- 500 agents × 1 req/min × 60 min × 24 hrs × 30 days = 21.6M requests/month
|
||||
- **Total**: ~$8.64/month (plus data transfer costs)
|
||||
|
||||
### Development Costs
|
||||
|
||||
- **Phase 1** (PostgreSQL-backed): 1 week development + 1 week testing
|
||||
- **Phase 2** (RabbitMQ migration): 1 week development + 1 week testing
|
||||
- **Phase 3** (Cleanup): 2-3 days
|
||||
|
||||
**Total**: 4-5 weeks engineering time
|
||||
|
||||
---
|
||||
|
||||
## Decision Points
|
||||
|
||||
### When to Implement?
|
||||
|
||||
Implement Broadway when you hit **any** of these thresholds:
|
||||
|
||||
1. **Agent Count**: 100+ agents (current: <10)
|
||||
2. **API Latency**: p95 > 500ms for `/api/v1/agent/metrics`
|
||||
3. **Timeout Rate**: >1% of agent requests timing out
|
||||
4. **Database Load**: DB CPU > 70% during agent polling windows
|
||||
5. **Business Need**: SLA requires <100ms API response time
|
||||
|
||||
### Which Message Queue?
|
||||
|
||||
| Criteria | PostgreSQL | RabbitMQ | AWS SQS |
|
||||
|----------|-----------|----------|---------|
|
||||
| **Agents** | <100 | 100-1000 | Any |
|
||||
| **Throughput** | <1k msg/sec | <100k msg/sec | Unlimited |
|
||||
| **Infrastructure** | None | Kubernetes | AWS only |
|
||||
| **Ops Complexity** | Low | Medium | Low |
|
||||
| **Cost** | $0 | ~$130/mo | ~$9/mo |
|
||||
|
||||
**Recommended Path**:
|
||||
1. Start with **PostgreSQL-backed** (proof of concept)
|
||||
2. Migrate to **RabbitMQ** when hitting 100 agents
|
||||
3. Consider **SQS** only if already on AWS
|
||||
|
||||
---
|
||||
|
||||
## References
|
||||
|
||||
- [Broadway Documentation](https://hexdocs.pm/broadway)
|
||||
- [Building Custom Producers](https://samuelmullen.com/articles/building-custom-producers-with-elixirs-broadway)
|
||||
- [Broadway RabbitMQ](https://hexdocs.pm/broadway_rabbitmq)
|
||||
- [OffBroadway SQS](https://hexdocs.pm/off_broadway_sqs)
|
||||
- [Batch Insert Performance](https://hexdocs.pm/ecto/Ecto.Repo.html#c:insert_all/3)
|
||||
|
||||
---
|
||||
|
||||
## Related Documents
|
||||
|
||||
- `CLAUDE.md` - Project-specific development guidelines
|
||||
- `AGENTS.md` - Elixir/Phoenix/LiveView patterns
|
||||
- `docs/API.md` - Agent API documentation
|
||||
|
|
@ -40,8 +40,6 @@ defmodule Towerops.Accounts.UserNotifier do
|
|||
def deliver_update_email_instructions(user, url) do
|
||||
deliver(user.email, "Update email instructions", """
|
||||
|
||||
==============================
|
||||
|
||||
Hi #{user.email},
|
||||
|
||||
You can change your email by visiting the URL below:
|
||||
|
|
@ -49,8 +47,6 @@ defmodule Towerops.Accounts.UserNotifier do
|
|||
#{url}
|
||||
|
||||
If you didn't request this change, please ignore this.
|
||||
|
||||
==============================
|
||||
""")
|
||||
end
|
||||
|
||||
|
|
@ -67,8 +63,6 @@ defmodule Towerops.Accounts.UserNotifier do
|
|||
defp deliver_magic_link_instructions(user, url) do
|
||||
deliver(user.email, "Log in instructions", """
|
||||
|
||||
==============================
|
||||
|
||||
Hi #{user.email},
|
||||
|
||||
You can log into your account by visiting the URL below:
|
||||
|
|
@ -76,16 +70,12 @@ defmodule Towerops.Accounts.UserNotifier do
|
|||
#{url}
|
||||
|
||||
If you didn't request this email, please ignore this.
|
||||
|
||||
==============================
|
||||
""")
|
||||
end
|
||||
|
||||
defp deliver_confirmation_instructions(user, url) do
|
||||
deliver(user.email, "Confirmation instructions", """
|
||||
|
||||
==============================
|
||||
|
||||
Hi #{user.email},
|
||||
|
||||
You can confirm your account by visiting the URL below:
|
||||
|
|
@ -93,8 +83,6 @@ defmodule Towerops.Accounts.UserNotifier do
|
|||
#{url}
|
||||
|
||||
If you didn't create an account with us, please ignore this.
|
||||
|
||||
==============================
|
||||
""")
|
||||
end
|
||||
|
||||
|
|
@ -104,8 +92,6 @@ defmodule Towerops.Accounts.UserNotifier do
|
|||
def deliver_reset_password_instructions(user, url) do
|
||||
deliver(user.email, "Reset password instructions", """
|
||||
|
||||
==============================
|
||||
|
||||
Hi #{user.email},
|
||||
|
||||
You can reset your password by visiting the URL below:
|
||||
|
|
@ -115,8 +101,6 @@ defmodule Towerops.Accounts.UserNotifier do
|
|||
If you didn't request this change, please ignore this.
|
||||
|
||||
This link will expire in 1 hour.
|
||||
|
||||
==============================
|
||||
""")
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -3,8 +3,13 @@ defmodule Towerops.HoneybadgerFilter do
|
|||
Custom Honeybadger filter to exclude noisy errors during deployments
|
||||
and filter sensitive SNMP credentials from error reports.
|
||||
|
||||
Filters out OTP supervisor/gen_server errors that occur during normal
|
||||
node shutdown/restart.
|
||||
Filters out errors that occur during normal application shutdown/restart:
|
||||
- OTP supervisor/gen_server termination errors
|
||||
- Redix connection errors (tcp_closed, econnrefused)
|
||||
- DBConnection errors during shutdown
|
||||
- Normal process exits (:normal, :shutdown, {:shutdown, _})
|
||||
|
||||
Also filters sensitive data from error reports (passwords, tokens, SNMP communities).
|
||||
"""
|
||||
|
||||
use Honeybadger.Filter.Mixin
|
||||
|
|
@ -41,6 +46,18 @@ defmodule Towerops.HoneybadgerFilter do
|
|||
otp_gen_server_shutdown?(context) ->
|
||||
nil
|
||||
|
||||
# Redix connection errors during shutdown/restart
|
||||
redix_connection_error?(context) ->
|
||||
nil
|
||||
|
||||
# DBConnection errors during shutdown/restart
|
||||
db_connection_error?(context) ->
|
||||
nil
|
||||
|
||||
# Normal process shutdown (exit :normal, :shutdown, {:shutdown, _})
|
||||
normal_shutdown?(context) ->
|
||||
nil
|
||||
|
||||
true ->
|
||||
context
|
||||
|> super()
|
||||
|
|
@ -69,6 +86,60 @@ defmodule Towerops.HoneybadgerFilter do
|
|||
domain == ["otp"] and match?(["gen_server", "error_info", _], mfa)
|
||||
end
|
||||
|
||||
# Detect Redix connection errors during shutdown/restart
|
||||
defp redix_connection_error?(context) do
|
||||
reason = get_in(context, [:reason])
|
||||
message = get_in(context, [:message])
|
||||
|
||||
cond do
|
||||
# Redix connection closed errors
|
||||
is_tuple(reason) and elem(reason, 0) == :tcp_closed ->
|
||||
true
|
||||
|
||||
# Redix connection refused (during restart)
|
||||
is_tuple(reason) and elem(reason, 0) == :econnrefused ->
|
||||
true
|
||||
|
||||
# String message check for Redix errors
|
||||
is_binary(message) and String.contains?(message, "Redix") ->
|
||||
true
|
||||
|
||||
true ->
|
||||
false
|
||||
end
|
||||
end
|
||||
|
||||
# Detect DBConnection errors during shutdown/restart
|
||||
defp db_connection_error?(context) do
|
||||
reason = get_in(context, [:reason])
|
||||
message = get_in(context, [:message])
|
||||
|
||||
cond do
|
||||
# DBConnection.ConnectionError
|
||||
is_exception(reason) and reason.__struct__ == DBConnection.ConnectionError ->
|
||||
true
|
||||
|
||||
# String message check
|
||||
is_binary(message) and String.contains?(message, "DBConnection") ->
|
||||
true
|
||||
|
||||
true ->
|
||||
false
|
||||
end
|
||||
end
|
||||
|
||||
# Detect normal process shutdown
|
||||
defp normal_shutdown?(context) do
|
||||
reason = get_in(context, [:reason])
|
||||
|
||||
case reason do
|
||||
:normal -> true
|
||||
:shutdown -> true
|
||||
{:shutdown, _} -> true
|
||||
_ -> false
|
||||
end
|
||||
end
|
||||
|
||||
# Filter sensitive data from context
|
||||
defp filter_sensitive_data(nil), do: nil
|
||||
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ defmodule ToweropsWeb.Admin.DashboardLive do
|
|||
{:ok,
|
||||
socket
|
||||
|> assign(:page_title, "Admin Dashboard")
|
||||
|> assign(:timezone, socket.assigns.current_scope.timezone)
|
||||
|> assign(:user_count, user_count)
|
||||
|> assign(:org_count, org_count)
|
||||
|> assign(:recent_logs, recent_logs)}
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ defmodule ToweropsWeb.Admin.OrgLive.Index do
|
|||
{:ok,
|
||||
socket
|
||||
|> assign(:page_title, "All Organizations")
|
||||
|> assign(:timezone, socket.assigns.current_scope.timezone)
|
||||
|> assign(:organizations, orgs)}
|
||||
end
|
||||
|
||||
|
|
|
|||
|
|
@ -31,6 +31,7 @@ defmodule ToweropsWeb.Admin.UserLive.Index do
|
|||
{:ok,
|
||||
socket
|
||||
|> assign(:page_title, "All Users")
|
||||
|> assign(:timezone, socket.assigns.current_scope.timezone)
|
||||
|> assign(:users, users)
|
||||
|> assign(:client_ip, ip)}
|
||||
end
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ defmodule ToweropsWeb.AgentLive.Edit do
|
|||
{:ok,
|
||||
socket
|
||||
|> assign(:page_title, "Edit #{agent_token.name}")
|
||||
|> assign(:timezone, socket.assigns.current_scope.timezone)
|
||||
|> assign(:agent_token, agent_token)
|
||||
|> assign(:form, to_form(changeset))}
|
||||
end
|
||||
|
|
|
|||
|
|
@ -43,6 +43,7 @@ defmodule ToweropsWeb.AgentLive.Index do
|
|||
{:ok,
|
||||
socket
|
||||
|> assign(:page_title, "Remote Agents")
|
||||
|> assign(:timezone, socket.assigns.current_scope.timezone)
|
||||
|> assign(:is_superuser, is_superuser)
|
||||
|> assign(:agent_tokens, agent_tokens)
|
||||
|> assign(:cloud_pollers, cloud_pollers)
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@ defmodule ToweropsWeb.AgentLive.Show do
|
|||
{:ok,
|
||||
socket
|
||||
|> assign(:page_title, agent_token.name)
|
||||
|> assign(:timezone, socket.assigns.current_scope.timezone)
|
||||
|> assign(:agent_token, agent_token)
|
||||
|> assign(:polling_targets, polling_targets)
|
||||
|> assign(:direct_assignments, direct_assignments)
|
||||
|
|
|
|||
|
|
@ -167,11 +167,11 @@
|
|||
<div class="min-w-0 flex-1">
|
||||
<p class="text-sm font-medium text-gray-600 dark:text-gray-400">Last Seen</p>
|
||||
<p class="mt-2 text-lg font-semibold text-gray-900 dark:text-white">
|
||||
<.timestamp datetime={@agent_token.last_seen_at} />
|
||||
<.timestamp datetime={@agent_token.last_seen_at} timezone={@timezone} />
|
||||
</p>
|
||||
<%= if @agent_token.last_seen_at do %>
|
||||
<p class="mt-1 text-xs text-gray-500 dark:text-gray-400">
|
||||
<.timestamp datetime={@agent_token.last_seen_at} />
|
||||
<.timestamp datetime={@agent_token.last_seen_at} timezone={@timezone} />
|
||||
</p>
|
||||
<% end %>
|
||||
<%= if @agent_token.last_ip do %>
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ defmodule ToweropsWeb.DeviceLive.Index do
|
|||
{:ok,
|
||||
socket
|
||||
|> assign(:page_title, "Devices")
|
||||
|> assign(:timezone, socket.assigns.current_scope.timezone)
|
||||
|> assign(:device, device)
|
||||
|> assign(:grouped_devices, grouped_devices)
|
||||
|> assign(:has_sites, sites != [])
|
||||
|
|
|
|||
|
|
@ -235,6 +235,7 @@ defmodule ToweropsWeb.DeviceLive.Show do
|
|||
|
||||
socket
|
||||
|> assign(:page_title, device.name)
|
||||
|> assign(:timezone, socket.assigns.current_scope.timezone)
|
||||
|> assign(:device, device)
|
||||
|> assign(:recent_checks, recent_checks)
|
||||
|> assign(:metrics, metrics)
|
||||
|
|
|
|||
|
|
@ -176,7 +176,11 @@ defmodule ToweropsWeb.HelpLive.Index do
|
|||
<li>No ads or sponsored content.</li>
|
||||
<li>No data collection or sharing.</li>
|
||||
<li>
|
||||
Our remote agent is fully open source and ONLY collects monitoring jobs from the server and WILL NEVER allow us to access any part of your internal network.
|
||||
Our remote agent is
|
||||
<a href="https://github.com/towerops-app/towerops-agent/">
|
||||
fully open source
|
||||
</a>
|
||||
and ONLY collects monitoring jobs from the server and WILL NEVER allow us to access any part of your internal network.
|
||||
</li>
|
||||
</ul>
|
||||
</p>
|
||||
|
|
|
|||
|
|
@ -32,6 +32,7 @@ defmodule ToweropsWeb.NetworkMapLive do
|
|||
{:ok,
|
||||
socket
|
||||
|> assign(:page_title, "Network Map")
|
||||
|> assign(:timezone, socket.assigns.current_scope.timezone)
|
||||
|> assign(:loading, true)
|
||||
|> assign(:topology, default_topology)
|
||||
|> assign(:active_tab, "added")}
|
||||
|
|
|
|||
|
|
@ -80,13 +80,17 @@ defmodule ToweropsWeb.UserSettingsLive do
|
|||
user = socket.assigns.current_scope.user
|
||||
|
||||
case Accounts.update_user_profile(user, user_params) do
|
||||
{:ok, _updated_user} ->
|
||||
{:ok, updated_user} ->
|
||||
# Log profile update
|
||||
fields_changed = Map.keys(user_params)
|
||||
AuditLogger.log_user_profile_updated(nil, user.id, fields_changed)
|
||||
|
||||
# Update current_scope with new user data (including timezone)
|
||||
updated_scope = %{socket.assigns.current_scope | user: updated_user, timezone: updated_user.timezone || "UTC"}
|
||||
|
||||
socket =
|
||||
socket
|
||||
|> assign(:current_scope, updated_scope)
|
||||
|> put_flash(:info, "Profile updated successfully.")
|
||||
|> assign_profile_form()
|
||||
|
||||
|
|
@ -1376,7 +1380,10 @@ defmodule ToweropsWeb.UserSettingsLive do
|
|||
<%= for attempt <- @login_history do %>
|
||||
<tr>
|
||||
<td class="whitespace-nowrap py-4 pl-4 pr-3 text-sm text-gray-900 dark:text-white sm:pl-6">
|
||||
{format_timestamp_in_timezone(attempt.inserted_at, @timezone)}
|
||||
{format_timestamp_in_timezone(
|
||||
attempt.inserted_at,
|
||||
@current_scope.timezone
|
||||
)}
|
||||
</td>
|
||||
<td class="whitespace-nowrap px-3 py-4 text-sm">
|
||||
<%= if attempt.success do %>
|
||||
|
|
|
|||
|
|
@ -155,5 +155,69 @@ defmodule Towerops.HoneybadgerFilterTest do
|
|||
|
||||
assert filtered == nil
|
||||
end
|
||||
|
||||
test "filters Redix connection errors (tcp_closed)" do
|
||||
context = %{reason: {:tcp_closed, :some_data}}
|
||||
|
||||
filtered = HoneybadgerFilter.filter_context(context)
|
||||
|
||||
assert filtered == nil
|
||||
end
|
||||
|
||||
test "filters Redix connection errors (econnrefused)" do
|
||||
context = %{reason: {:econnrefused, :some_data}}
|
||||
|
||||
filtered = HoneybadgerFilter.filter_context(context)
|
||||
|
||||
assert filtered == nil
|
||||
end
|
||||
|
||||
test "filters Redix errors by message" do
|
||||
context = %{message: "Redix connection failed"}
|
||||
|
||||
filtered = HoneybadgerFilter.filter_context(context)
|
||||
|
||||
assert filtered == nil
|
||||
end
|
||||
|
||||
test "filters DBConnection errors by exception" do
|
||||
context = %{reason: %DBConnection.ConnectionError{message: "connection lost"}}
|
||||
|
||||
filtered = HoneybadgerFilter.filter_context(context)
|
||||
|
||||
assert filtered == nil
|
||||
end
|
||||
|
||||
test "filters DBConnection errors by message" do
|
||||
context = %{message: "DBConnection timeout"}
|
||||
|
||||
filtered = HoneybadgerFilter.filter_context(context)
|
||||
|
||||
assert filtered == nil
|
||||
end
|
||||
|
||||
test "filters normal shutdown" do
|
||||
context = %{reason: :normal}
|
||||
|
||||
filtered = HoneybadgerFilter.filter_context(context)
|
||||
|
||||
assert filtered == nil
|
||||
end
|
||||
|
||||
test "filters shutdown" do
|
||||
context = %{reason: :shutdown}
|
||||
|
||||
filtered = HoneybadgerFilter.filter_context(context)
|
||||
|
||||
assert filtered == nil
|
||||
end
|
||||
|
||||
test "filters shutdown with reason" do
|
||||
context = %{reason: {:shutdown, :application_stopped}}
|
||||
|
||||
filtered = HoneybadgerFilter.filter_context(context)
|
||||
|
||||
assert filtered == nil
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue