diff --git a/README.md b/README.md index 83f0e1a..dfcf26d 100644 --- a/README.md +++ b/README.md @@ -42,6 +42,180 @@ This project follows Clean Architecture principles with clear separation of conc - **Configuration Management**: Koanf for flexible configuration loading (file + environment variables) - **Structured Logging**: Zap logger with configurable levels and formats - **Batch Processing**: Efficient batch message processing and database inserts +- **AFTN Protocol Validation**: Optional AFTN/ICAO protocol compliance validation with detailed error tracking +- **Serial Reader Health Monitoring**: Automatic detection of message flow interruptions and sequence gaps + +## AFTN Protocol Validation + +The system includes comprehensive AFTN (Aeronautical Fixed Telecommunication Network) protocol validation to ensure incoming telegrams comply with ICAO standards. + +### Features + +1. **Protocol Field Validation** + - Priority indicators: FF (Flash), GG (Immediate), QU (Distress), DD (Delay), SS (Service), KK (Correction) + - ICAO addresses: 4-character alphanumeric validation + - DateTime formats: DDHHMM with range validation (DD:01-31, HH:00-23, MM:00-59) + +2. **Serial Reader Health Monitoring** + - Automatic detection of message flow interruptions + - Configurable gap threshold (default: 2 minutes) + - Message sequence gap detection (missing sequence numbers) + - Real-time health status metrics + +3. **Dead-Letter Queue (DLQ) Routing** + - Invalid telegrams automatically routed to DLQ for offline review + - Detailed error categorization (priority_indicator, icao_address, datetime, multiple_errors) + - Preserves original message content for debugging + +4. **Observability** + - Prometheus metrics for validation errors, message gaps, and health status + - Structured logging with error context + - OpenTelemetry tracing integration + +### Configuration + +AFTN validation is **disabled by default** for safe rollout. Enable it in your configuration file: + +```toml +[aftn] +# Enable AFTN protocol validation +validation_enabled = true + +# Time threshold after which serial reader is considered stalled +message_gap_threshold = "2m" + +# Enable detection of missing message sequence numbers +enable_sequence_gap_detection = true +``` + +### Metrics + +The system exposes the following Prometheus metrics: + +| Metric | Type | Description | +|--------|------|-------------| +| `caatsm_aftn_validation_errors_total` | Counter | Total AFTN validation errors by error_type label | +| `caatsm_message_gap_seconds` | Gauge | Time in seconds since last message received | +| `caatsm_message_sequence_gap_total` | Counter | Number of detected sequence gaps (missing messages) | +| `caatsm_serial_reader_healthy` | Gauge | Health status: 1=healthy, 0=stalled | + +Example Prometheus queries: + +```promql +# AFTN validation error rate +rate(caatsm_aftn_validation_errors_total[5m]) + +# Current message gap +caatsm_message_gap_seconds{stream="TELEGRAM", consumer="telegram-consumer"} + +# Serial reader health +caatsm_serial_reader_healthy{stream="TELEGRAM", consumer="telegram-consumer"} + +# Sequence gap rate +rate(caatsm_message_sequence_gap_total[5m]) +``` + +### Alerts + +Pre-configured Prometheus alert rules are available in `configs/prometheus-alerts.yml`: + +- **SerialReaderStalled** (critical): No messages for > 2 minutes +- **HighMessageGap** (warning): Gap > 60 seconds +- **MessageSequenceGaps** (warning): Missing sequence numbers detected +- **HighAFTNValidationErrorRate** (warning): > 5% of messages failing validation +- **ConsumerLagGrowing** (warning): Pending messages increasing +- **ConsumerCriticallyBehind** (critical): > 5000 pending messages + +To install the alerts: + +```bash +# Copy alerts to Prometheus server +cp configs/prometheus-alerts.yml /path/to/prometheus/rules/ + +# Add to prometheus.yml +rule_files: + - "rules/prometheus-alerts.yml" + +# Reload Prometheus +curl -X POST http://localhost:9090/-/reload +``` + +### Grafana Dashboard + +A pre-built Grafana dashboard is available in `configs/grafana-dashboard-aftn.json` with panels for: + +- Serial reader health status (stat panel with color coding) +- Message gap time series +- AFTN validation errors by type +- AFTN error rate percentage +- Sequence gap rate +- Consumer pending messages +- Message processing throughput by status + +To import the dashboard: + +1. Open Grafana UI +2. Navigate to Dashboards → Import +3. Upload `configs/grafana-dashboard-aftn.json` +4. Select your Prometheus datasource +5. Click "Import" + +### Error Handling + +When AFTN validation fails: + +1. **Message Status**: Set to `aftn_error` +2. **Error Recording**: Error details stored in `error_reason` field +3. **DLQ Routing**: Message published to DLQ subject (if configured) +4. **Metrics**: Validation error counter incremented with error type label +5. **Logging**: Warning logged with error context and message preview +6. **Tracing**: Error recorded in OpenTelemetry span + +Example log entry: + +```json +{ + "level": "warn", + "msg": "AFTN validation failed", + "status": "aftn_error", + "error_type": "priority_indicator", + "content_preview": "ZCZC TMQ2526 141605\nXX ZBTJZPZX\n...", + "error": "AFTN validation error [priority_indicator]: must be one of FF, GG, QU, DD, SS, KK (value: \"XX\")" +} +``` + +### Testing AFTN Validation + +To test AFTN validation in development: + +```bash +# Enable validation in config.dev.toml +[aftn] +validation_enabled = true + +# Start the application +make run + +# Send a telegram with invalid priority indicator +# The message will be rejected and routed to DLQ + +# Check DLQ for rejected messages +nats sub caatsm.dlq + +# Check metrics +curl http://localhost:2112/metrics | grep aftn +``` + +### Disabling AFTN Validation + +AFTN validation can be disabled at runtime without code changes: + +```toml +[aftn] +validation_enabled = false # Disable validation +``` + +This allows for gradual rollout and quick rollback if issues arise. ## Contributor Guide diff --git a/configs/config.dev.toml b/configs/config.dev.toml index 333e70e..b1a68c6 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -118,3 +118,12 @@ health_timeout = "2s" enabled = true # Set to true when switching to JetStream mode # subject: NATS subject where failed messages will be published for manual inspection subject = "caatsm.dlq" + +[aftn] +# AFTN Protocol Validation and Monitoring +# validation_enabled: Enable AFTN protocol validation for telegrams (disabled by default for safe rollout) +validation_enabled = false +# message_gap_threshold: Duration after which serial reader is considered stalled (no messages received) +message_gap_threshold = "2m" +# enable_sequence_gap_detection: Monitor for missing sequence numbers in telegram stream +enable_sequence_gap_detection = true diff --git a/configs/grafana-dashboard-aftn.json b/configs/grafana-dashboard-aftn.json new file mode 100644 index 0000000..14e13f9 --- /dev/null +++ b/configs/grafana-dashboard-aftn.json @@ -0,0 +1,863 @@ +{ + "annotations": { + "list": [ + { + "builtIn": 1, + "datasource": { + "type": "grafana", + "uid": "-- Grafana --" + }, + "enable": true, + "hide": true, + "iconColor": "rgba(0, 211, 255, 1)", + "name": "Annotations & Alerts", + "type": "dashboard" + } + ] + }, + "editable": true, + "fiscalYearStartMonth": 0, + "graphTooltip": 1, + "id": null, + "links": [], + "panels": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "Health status of the serial reader. 1 = healthy (messages flowing), 0 = stalled (no messages received for > 2 minutes)", + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [ + { + "options": { + "0": { + "color": "red", + "index": 1, + "text": "STALLED" + }, + "1": { + "color": "green", + "index": 0, + "text": "HEALTHY" + } + }, + "type": "value" + } + ], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "red", + "value": null + }, + { + "color": "green", + "value": 1 + } + ] + }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { + "h": 6, + "w": 6, + "x": 0, + "y": 0 + }, + "id": 1, + "options": { + "colorMode": "background", + "graphMode": "none", + "justifyMode": "center", + "orientation": "auto", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "text": {}, + "textMode": "value_and_name", + "wideLayout": true + }, + "pluginVersion": "10.0.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "expr": "caatsm_serial_reader_healthy{stream=\"$stream\", consumer=\"$consumer\"}", + "refId": "A" + } + ], + "title": "Serial Reader Health", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "Time in seconds since the last message was received from the serial reader", + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "yellow", + "value": 60 + }, + { + "color": "red", + "value": 120 + } + ] + }, + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { + "h": 6, + "w": 6, + "x": 6, + "y": 0 + }, + "id": 2, + "options": { + "colorMode": "background", + "graphMode": "area", + "justifyMode": "center", + "orientation": "auto", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showPercentChange": false, + "text": {}, + "textMode": "value_and_name", + "wideLayout": true + }, + "pluginVersion": "10.0.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "expr": "caatsm_message_gap_seconds{stream=\"$stream\", consumer=\"$consumer\"}", + "refId": "A" + } + ], + "title": "Message Gap (seconds)", + "type": "stat" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "Total number of AFTN validation errors in the last 5 minutes", + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 20, + "gradientMode": "none", + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 2, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { + "h": 6, + "w": 6, + "x": 12, + "y": 0 + }, + "id": 3, + "options": { + "legend": { + "calcs": [ + "last" + ], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "10.0.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "expr": "sum(rate(caatsm_aftn_validation_errors_total[5m])) by (error_type)", + "legendFormat": "{{error_type}}", + "refId": "A" + } + ], + "title": "AFTN Validation Errors by Type (5m rate)", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "Percentage of messages failing AFTN validation", + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "max": 100, + "min": 0, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "yellow", + "value": 1 + }, + { + "color": "red", + "value": 5 + } + ] + }, + "unit": "percent" + }, + "overrides": [] + }, + "gridPos": { + "h": 6, + "w": 6, + "x": 18, + "y": 0 + }, + "id": 4, + "options": { + "displayMode": "gradient", + "maxVizHeight": 300, + "minVizHeight": 10, + "minVizWidth": 0, + "namePlacement": "auto", + "orientation": "horizontal", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showUnfilled": true, + "sizing": "auto", + "text": {}, + "valueMode": "color" + }, + "pluginVersion": "10.0.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "expr": "100 * (sum(rate(caatsm_aftn_validation_errors_total[5m])) / sum(rate(caatsm_processed_total[5m])))", + "refId": "A" + } + ], + "title": "AFTN Error Rate %", + "type": "bargauge" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "Time series showing message gap evolution over time", + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "Seconds", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + }, + "insertNulls": false, + "lineInterpolation": "smooth", + "lineWidth": 2, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "line" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 120 + } + ] + }, + "unit": "s" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 6 + }, + "id": 5, + "options": { + "legend": { + "calcs": [ + "mean", + "max" + ], + "displayMode": "table", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "pluginVersion": "10.0.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "expr": "caatsm_message_gap_seconds{stream=\"$stream\", consumer=\"$consumer\"}", + "legendFormat": "{{stream}}/{{consumer}}", + "refId": "A" + } + ], + "title": "Message Gap Over Time", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "Rate of sequence gaps detected (missing message sequence numbers)", + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "Gaps/sec", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 20, + "gradientMode": "none", + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + }, + "insertNulls": false, + "lineInterpolation": "stepAfter", + "lineWidth": 2, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 1 + } + ] + }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 6 + }, + "id": 6, + "options": { + "legend": { + "calcs": [ + "sum" + ], + "displayMode": "table", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "pluginVersion": "10.0.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "expr": "rate(caatsm_message_sequence_gap_total{stream=\"$stream\", consumer=\"$consumer\"}[5m])", + "legendFormat": "{{stream}}/{{consumer}}", + "refId": "A" + } + ], + "title": "Sequence Gap Rate (5m)", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "Number of pending messages in the JetStream consumer queue", + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "Messages", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + }, + "insertNulls": false, + "lineInterpolation": "smooth", + "lineWidth": 2, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "line" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "yellow", + "value": 1000 + }, + { + "color": "red", + "value": 5000 + } + ] + }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 14 + }, + "id": 7, + "options": { + "legend": { + "calcs": [ + "mean", + "max" + ], + "displayMode": "table", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "none" + } + }, + "pluginVersion": "10.0.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "expr": "caatsm_nats_consumer_pending_messages{stream=\"$stream\", consumer=\"$consumer\"}", + "legendFormat": "{{stream}}/{{consumer}}", + "refId": "A" + } + ], + "title": "Consumer Pending Messages", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "description": "Message processing throughput by status", + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "Messages/sec", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 20, + "gradientMode": "none", + "hideFrom": { + "tooltip": false, + "viz": false, + "legend": false + }, + "insertNulls": false, + "lineInterpolation": "smooth", + "lineWidth": 2, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "normal" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "short" + }, + "overrides": [ + { + "matcher": { + "id": "byRegexp", + "options": ".*error.*" + }, + "properties": [ + { + "id": "color", + "value": { + "fixedColor": "red", + "mode": "fixed" + } + } + ] + }, + { + "matcher": { + "id": "byRegexp", + "options": ".*parsed.*" + }, + "properties": [ + { + "id": "color", + "value": { + "fixedColor": "green", + "mode": "fixed" + } + } + ] + } + ] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 14 + }, + "id": 8, + "options": { + "legend": { + "calcs": [ + "mean" + ], + "displayMode": "table", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "pluginVersion": "10.0.0", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "expr": "sum(rate(caatsm_processed_total[5m])) by (status)", + "legendFormat": "{{status}}", + "refId": "A" + } + ], + "title": "Message Processing Rate by Status", + "type": "timeseries" + } + ], + "schemaVersion": 39, + "tags": [ + "caatsm", + "aftn", + "aviation", + "telegram" + ], + "templating": { + "list": [ + { + "current": { + "selected": false, + "text": "Prometheus", + "value": "Prometheus" + }, + "hide": 0, + "includeAll": false, + "label": "Datasource", + "multi": false, + "name": "DS_PROMETHEUS", + "options": [], + "query": "prometheus", + "refresh": 1, + "regex": "", + "skipUrlSync": false, + "type": "datasource" + }, + { + "current": { + "selected": false, + "text": "TELEGRAM", + "value": "TELEGRAM" + }, + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "definition": "label_values(caatsm_nats_consumer_pending_messages, stream)", + "hide": 0, + "includeAll": false, + "label": "Stream", + "multi": false, + "name": "stream", + "options": [], + "query": { + "qryType": 1, + "query": "label_values(caatsm_nats_consumer_pending_messages, stream)", + "refId": "PrometheusVariableQueryEditor-VariableQuery" + }, + "refresh": 1, + "regex": "", + "skipUrlSync": false, + "sort": 0, + "type": "query" + }, + { + "current": { + "selected": false, + "text": "telegram-consumer", + "value": "telegram-consumer" + }, + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "definition": "label_values(caatsm_nats_consumer_pending_messages{stream=\"$stream\"}, consumer)", + "hide": 0, + "includeAll": false, + "label": "Consumer", + "multi": false, + "name": "consumer", + "options": [], + "query": { + "qryType": 1, + "query": "label_values(caatsm_nats_consumer_pending_messages{stream=\"$stream\"}, consumer)", + "refId": "PrometheusVariableQueryEditor-VariableQuery" + }, + "refresh": 1, + "regex": "", + "skipUrlSync": false, + "sort": 0, + "type": "query" + } + ] + }, + "time": { + "from": "now-1h", + "to": "now" + }, + "timepicker": {}, + "timezone": "browser", + "title": "CAATSM AFTN Health & Validation", + "uid": "caatsm-aftn-health", + "version": 1, + "weekStart": "" +} diff --git a/configs/prometheus-alerts.yml b/configs/prometheus-alerts.yml new file mode 100644 index 0000000..61c216d --- /dev/null +++ b/configs/prometheus-alerts.yml @@ -0,0 +1,224 @@ +# Prometheus Alert Rules for CAATSM AFTN Telegram Processor +# +# Installation: +# 1. Copy this file to your Prometheus server's rules directory +# 2. Add to prometheus.yml: +# rule_files: +# - "prometheus-alerts.yml" +# 3. Reload Prometheus configuration +# +# Alert Severity Levels: +# - critical: Immediate action required (pages on-call) +# - warning: Investigation needed (notify team channel) + +groups: + - name: caatsm_aftn_health + interval: 30s + rules: + # Critical: Serial reader has stopped publishing messages + - alert: SerialReaderStalled + expr: caatsm_serial_reader_healthy == 0 + for: 2m + labels: + severity: critical + component: serial_reader + annotations: + summary: "Serial reader stalled for {{ $labels.stream }}/{{ $labels.consumer }}" + description: | + No messages have been received from the serial reader for more than 2 minutes. + This indicates the serial port reader may have crashed or the hardware connection is broken. + + Current gap: {{ with query "caatsm_message_gap_seconds{stream=\"" }}{{ . | first | value | humanizeDuration }}{{ end }} + Stream: {{ $labels.stream }} + Consumer: {{ $labels.consumer }} + + ACTION REQUIRED: + 1. Check serial reader process status + 2. Verify serial port connection + 3. Check hardware status + 4. Review serial reader logs + + # Warning: Message gap is growing but not yet critical + - alert: HighMessageGap + expr: caatsm_message_gap_seconds > 60 and caatsm_serial_reader_healthy == 1 + for: 1m + labels: + severity: warning + component: serial_reader + annotations: + summary: "High message gap detected: {{ $labels.stream }}/{{ $labels.consumer }}" + description: | + Message gap is {{ $value }}s but still below critical threshold. + This may indicate slow message processing or reduced incoming message rate. + + Stream: {{ $labels.stream }} + Consumer: {{ $labels.consumer }} + + # Warning: Sequence gaps detected (missing messages) + - alert: MessageSequenceGaps + expr: rate(caatsm_message_sequence_gap_total[5m]) > 0 + for: 2m + labels: + severity: warning + component: serial_reader + annotations: + summary: "Message sequence gaps detected: {{ $labels.stream }}/{{ $labels.consumer }}" + description: | + Missing message sequence numbers detected at {{ $value | humanize }} gaps/sec. + This indicates messages are being lost or skipped in the stream. + + Stream: {{ $labels.stream }} + Consumer: {{ $labels.consumer }} + Rate: {{ $value | humanize }} gaps/sec + + Possible causes: + - Serial reader buffer overflow + - Network packet loss (if messages forwarded over network) + - Stream retention limits exceeded + - Consumer processing too slow + + # Warning: High AFTN validation error rate + - alert: HighAFTNValidationErrorRate + expr: | + ( + sum(rate(caatsm_aftn_validation_errors_total[5m])) by (stream, consumer) + / + sum(rate(caatsm_processed_total[5m])) by (stream, consumer) + ) > 0.05 + for: 5m + labels: + severity: warning + component: aftn_validator + annotations: + summary: "High AFTN validation error rate: {{ $value | humanizePercentage }}" + description: | + More than 5% of incoming telegrams are failing AFTN protocol validation. + Current error rate: {{ $value | humanizePercentage }} + + This may indicate: + - Upstream system sending malformed telegrams + - Serial port data corruption + - Configuration mismatch + + Check DLQ for error details and patterns. + + # Warning: Consumer lag is growing + - alert: ConsumerLagGrowing + expr: | + deriv(caatsm_nats_consumer_pending_messages[5m]) > 10 + for: 3m + labels: + severity: warning + component: consumer + annotations: + summary: "Consumer lag growing: {{ $labels.stream }}/{{ $labels.consumer }}" + description: | + Consumer pending messages is growing at {{ $value | humanize }} msgs/sec. + Current pending: {{ with query "caatsm_nats_consumer_pending_messages" }}{{ . | first | value }}{{ end }} + + This indicates the consumer cannot keep up with incoming message rate. + + Stream: {{ $labels.stream }} + Consumer: {{ $labels.consumer }} + + # Critical: Consumer critically behind + - alert: ConsumerCriticallyBehind + expr: caatsm_nats_consumer_pending_messages > 5000 + for: 5m + labels: + severity: critical + component: consumer + annotations: + summary: "Consumer critically behind: {{ $value }} pending messages" + description: | + Consumer has {{ $value }} pending messages - critically behind. + This will cause message processing delays and may trigger stream retention limits. + + Stream: {{ $labels.stream }} + Consumer: {{ $labels.consumer }} + Pending: {{ $value }} + + ACTION REQUIRED: + 1. Check processor performance and errors + 2. Check database connection and performance + 3. Consider scaling consumers horizontally + 4. Review stream retention settings + + # Warning: High processing failure rate + - alert: HighProcessingFailureRate + expr: | + ( + sum(rate(caatsm_messages_total{result="fail"}[5m])) by (stream, consumer) + / + sum(rate(caatsm_messages_total[5m])) by (stream, consumer) + ) > 0.10 + for: 5m + labels: + severity: warning + component: processor + annotations: + summary: "High processing failure rate: {{ $value | humanizePercentage }}" + description: | + More than 10% of messages are failing to process. + Current failure rate: {{ $value | humanizePercentage }} + + Stream: {{ $labels.stream }} + Consumer: {{ $labels.consumer }} + + Check application logs for error details. + + # Warning: High publish failure rate + - alert: HighPublishFailureRate + expr: | + sum(rate(caatsm_publish_failures_total[5m])) by (category) + / + sum(rate(caatsm_processed_total[5m])) by (category) > 0.05 + for: 5m + labels: + severity: warning + component: publisher + annotations: + summary: "High publish failure rate for {{ $labels.category }}: {{ $value | humanizePercentage }}" + description: | + More than 5% of {{ $labels.category }} messages failing to publish. + Current failure rate: {{ $value | humanizePercentage }} + + Category: {{ $labels.category }} + + Check NATS JetStream connectivity and publisher logs. + + # Warning: DLQ publish failures (messages lost) + - alert: DLQPublishFailures + expr: rate(caatsm_dlq_publish_failures_total[5m]) > 0 + for: 2m + labels: + severity: warning + component: dlq + annotations: + summary: "DLQ publish failures detected" + description: | + Failed messages cannot be published to DLQ - messages may be lost! + Failure rate: {{ $value | humanize }} msgs/sec + + Stream: {{ $labels.stream }} + Consumer: {{ $labels.consumer }} + + Check DLQ subject configuration and NATS JetStream health. + + - name: caatsm_aftn_validation_details + interval: 1m + rules: + # Recording rule: AFTN error rate by type + - record: caatsm:aftn_validation_error_rate:5m + expr: | + rate(caatsm_aftn_validation_errors_total[5m]) + + # Recording rule: Total processing rate + - record: caatsm:processing_rate:5m + expr: | + sum(rate(caatsm_processed_total[5m])) by (stream, consumer, status) + + # Recording rule: Average message gap + - record: caatsm:message_gap_seconds:avg + expr: | + avg(caatsm_message_gap_seconds) by (stream, consumer) diff --git a/go.mod b/go.mod index 588bafb..f4baee1 100644 --- a/go.mod +++ b/go.mod @@ -12,7 +12,7 @@ require ( github.com/knadh/koanf/v2 v2.3.0 github.com/nats-io/nats.go v1.47.0 github.com/onsi/ginkgo/v2 v2.27.2 - github.com/onsi/gomega v1.38.2 + github.com/onsi/gomega v1.38.3 github.com/prometheus/client_golang v1.23.2 github.com/stretchr/testify v1.11.1 github.com/testcontainers/testcontainers-go v0.30.0 diff --git a/internal/adapter/dto/telegram.go b/internal/adapter/dto/telegram.go index 59eac43..1675038 100644 --- a/internal/adapter/dto/telegram.go +++ b/internal/adapter/dto/telegram.go @@ -14,6 +14,7 @@ const ( MessageStatusParsed MessageStatus = "parsed" MessageStatusHeaderError MessageStatus = "header_error" MessageStatusBodyError MessageStatus = "body_error" + MessageStatusAFTNError MessageStatus = "aftn_error" ) // ParsedTelegram holds the parsed data from an aviation message. diff --git a/internal/adapter/validator/aftn.go b/internal/adapter/validator/aftn.go new file mode 100644 index 0000000..06f4023 --- /dev/null +++ b/internal/adapter/validator/aftn.go @@ -0,0 +1,176 @@ +package validator + +import ( + "caatsm/internal/adapter/dto" + "fmt" + "regexp" + "strconv" + "strings" +) + +// AFTNError represents an AFTN protocol violation +type AFTNError struct { + Field string // e.g., "priority_indicator", "icao_address" + Value string + Message string +} + +func (e *AFTNError) Error() string { + return fmt.Sprintf("AFTN validation error [%s]: %s (value: %q)", e.Field, e.Message, e.Value) +} + +// AFTN field validators +var ( + // Priority indicators: FF (Flash), GG (Immediate), QU (Distress), DD (Delay), SS (Service), KK (Correction) + validPriorities = map[string]bool{ + "FF": true, "GG": true, "QU": true, + "DD": true, "SS": true, "KK": true, + } + + // ICAO address: 4 uppercase alphanumeric characters + icaoAddressPattern = regexp.MustCompile(`^[A-Z0-9]{4}$`) + + // DateTime: DDHHMM (6 digits) + dateTimePattern = regexp.MustCompile(`^\d{6}$`) +) + +// ValidatePriorityIndicator validates AFTN priority indicator +func ValidatePriorityIndicator(priority string) error { + priority = strings.TrimSpace(strings.ToUpper(priority)) + if priority == "" { + return nil // Optional field + } + if !validPriorities[priority] { + return &AFTNError{ + Field: "priority_indicator", + Value: priority, + Message: "must be one of FF, GG, QU, DD, SS, KK", + } + } + return nil +} + +// ValidateICAOAddress validates 4-character ICAO address +func ValidateICAOAddress(address string) error { + address = strings.TrimSpace(strings.ToUpper(address)) + if address == "" { + return nil // Optional field + } + if !icaoAddressPattern.MatchString(address) { + return &AFTNError{ + Field: "icao_address", + Value: address, + Message: "must be 4 uppercase alphanumeric characters", + } + } + return nil +} + +// ValidateDateTime validates DDHHMM format +func ValidateDateTime(dt string) error { + dt = strings.TrimSpace(dt) + if dt == "" { + return nil // Optional field + } + if !dateTimePattern.MatchString(dt) { + return &AFTNError{ + Field: "datetime", + Value: dt, + Message: "must be 6 digits (DDHHMM format)", + } + } + // Additional semantic validation + if len(dt) == 6 { + day := dt[0:2] + hour := dt[2:4] + minute := dt[4:6] + // Basic range checks + if !isValidRange(day, 1, 31) || !isValidRange(hour, 0, 23) || !isValidRange(minute, 0, 59) { + return &AFTNError{ + Field: "datetime", + Value: dt, + Message: "invalid date/time ranges (DD:01-31, HH:00-23, MM:00-59)", + } + } + } + return nil +} + +// ValidateTelegram validates all AFTN fields in ParsedTelegram +func ValidateTelegram(telegram *dto.ParsedTelegram) error { + if telegram == nil { + return nil + } + + var errors []error + + // Validate priority indicator + if err := ValidatePriorityIndicator(telegram.PriorityIndicator); err != nil { + errors = append(errors, err) + } + + // Validate primary address (ICAO) + if err := ValidateICAOAddress(telegram.PrimaryAddress); err != nil { + errors = append(errors, err) + } + + // Validate originator (ICAO) + if err := ValidateICAOAddress(telegram.Originator); err != nil { + errors = append(errors, err) + } + + // Validate datetime + if err := ValidateDateTime(telegram.DateTime); err != nil { + errors = append(errors, err) + } + + // Validate originator datetime + if err := ValidateDateTime(telegram.OriginatorDateTime); err != nil { + errors = append(errors, err) + } + + if len(errors) > 0 { + return &AFTNValidationErrors{Errors: errors} + } + return nil +} + +// AFTNValidationErrors wraps multiple validation errors +type AFTNValidationErrors struct { + Errors []error +} + +func (e *AFTNValidationErrors) Error() string { + messages := make([]string, len(e.Errors)) + for i, err := range e.Errors { + messages[i] = err.Error() + } + return fmt.Sprintf("AFTN validation failed: %s", strings.Join(messages, "; ")) +} + +// IsAFTNError checks if error is an AFTN validation error +func IsAFTNError(err error) bool { + if err == nil { + return false + } + _, ok1 := err.(*AFTNError) + _, ok2 := err.(*AFTNValidationErrors) + return ok1 || ok2 +} + +// GetAFTNErrorType extracts the error type for metrics labeling +func GetAFTNErrorType(err error) string { + if aftnErr, ok := err.(*AFTNError); ok { + return aftnErr.Field + } + if _, ok := err.(*AFTNValidationErrors); ok { + return "multiple_errors" + } + return "unknown" +} + +// isValidRange checks if a numeric string is within the specified range +func isValidRange(s string, min, max int) bool { + val, err := strconv.Atoi(s) + return err == nil && val >= min && val <= max +} diff --git a/internal/adapter/validator/aftn_test.go b/internal/adapter/validator/aftn_test.go new file mode 100644 index 0000000..ff52638 --- /dev/null +++ b/internal/adapter/validator/aftn_test.go @@ -0,0 +1,341 @@ +package validator_test + +import ( + "caatsm/internal/adapter/dto" + "caatsm/internal/adapter/validator" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("AFTN Validator", func() { + Describe("ValidatePriorityIndicator", func() { + Context("with valid priority indicators", func() { + It("accepts FF (Flash)", func() { + err := validator.ValidatePriorityIndicator("FF") + Expect(err).To(BeNil()) + }) + + It("accepts GG (Immediate)", func() { + err := validator.ValidatePriorityIndicator("GG") + Expect(err).To(BeNil()) + }) + + It("accepts QU (Distress)", func() { + err := validator.ValidatePriorityIndicator("QU") + Expect(err).To(BeNil()) + }) + + It("accepts DD (Delay)", func() { + err := validator.ValidatePriorityIndicator("DD") + Expect(err).To(BeNil()) + }) + + It("accepts SS (Service)", func() { + err := validator.ValidatePriorityIndicator("SS") + Expect(err).To(BeNil()) + }) + + It("accepts KK (Correction)", func() { + err := validator.ValidatePriorityIndicator("KK") + Expect(err).To(BeNil()) + }) + + It("accepts lowercase with trimming", func() { + err := validator.ValidatePriorityIndicator(" ff ") + Expect(err).To(BeNil()) + }) + + It("accepts empty string (optional field)", func() { + err := validator.ValidatePriorityIndicator("") + Expect(err).To(BeNil()) + }) + }) + + Context("with invalid priority indicators", func() { + It("rejects invalid code XX", func() { + err := validator.ValidatePriorityIndicator("XX") + Expect(err).ToNot(BeNil()) + Expect(validator.IsAFTNError(err)).To(BeTrue()) + Expect(validator.GetAFTNErrorType(err)).To(Equal("priority_indicator")) + }) + + It("rejects single character", func() { + err := validator.ValidatePriorityIndicator("F") + Expect(err).ToNot(BeNil()) + }) + + It("rejects three characters", func() { + err := validator.ValidatePriorityIndicator("FFF") + Expect(err).ToNot(BeNil()) + }) + }) + }) + + Describe("ValidateICAOAddress", func() { + Context("with valid ICAO addresses", func() { + It("accepts ZBTJ (Beijing)", func() { + err := validator.ValidateICAOAddress("ZBTJ") + Expect(err).To(BeNil()) + }) + + It("accepts KLAX (Los Angeles)", func() { + err := validator.ValidateICAOAddress("KLAX") + Expect(err).To(BeNil()) + }) + + It("accepts ZGGG (Guangzhou)", func() { + err := validator.ValidateICAOAddress("ZGGG") + Expect(err).To(BeNil()) + }) + + It("accepts alphanumeric codes like Z999", func() { + err := validator.ValidateICAOAddress("Z999") + Expect(err).To(BeNil()) + }) + + It("accepts 1ABC", func() { + err := validator.ValidateICAOAddress("1ABC") + Expect(err).To(BeNil()) + }) + + It("accepts lowercase with trimming", func() { + err := validator.ValidateICAOAddress(" zbtj ") + Expect(err).To(BeNil()) + }) + + It("accepts empty string (optional field)", func() { + err := validator.ValidateICAOAddress("") + Expect(err).To(BeNil()) + }) + }) + + Context("with invalid ICAO addresses", func() { + It("rejects too short (3 chars)", func() { + err := validator.ValidateICAOAddress("ZBT") + Expect(err).ToNot(BeNil()) + Expect(validator.IsAFTNError(err)).To(BeTrue()) + Expect(validator.GetAFTNErrorType(err)).To(Equal("icao_address")) + }) + + It("rejects too long (5 chars)", func() { + err := validator.ValidateICAOAddress("ZBTJX") + Expect(err).ToNot(BeNil()) + }) + + It("rejects special characters", func() { + err := validator.ValidateICAOAddress("ZB-J") + Expect(err).ToNot(BeNil()) + }) + + It("rejects spaces", func() { + err := validator.ValidateICAOAddress("ZB J") + Expect(err).ToNot(BeNil()) + }) + }) + }) + + Describe("ValidateDateTime", func() { + Context("with valid datetime values", func() { + It("accepts 151430 (15th day, 14:30)", func() { + err := validator.ValidateDateTime("151430") + Expect(err).To(BeNil()) + }) + + It("accepts 010000 (1st day, 00:00)", func() { + err := validator.ValidateDateTime("010000") + Expect(err).To(BeNil()) + }) + + It("accepts 312359 (31st day, 23:59)", func() { + err := validator.ValidateDateTime("312359") + Expect(err).To(BeNil()) + }) + + It("accepts empty string (optional field)", func() { + err := validator.ValidateDateTime("") + Expect(err).To(BeNil()) + }) + }) + + Context("with invalid datetime values", func() { + It("rejects non-numeric", func() { + err := validator.ValidateDateTime("15A430") + Expect(err).ToNot(BeNil()) + Expect(validator.IsAFTNError(err)).To(BeTrue()) + Expect(validator.GetAFTNErrorType(err)).To(Equal("datetime")) + }) + + It("rejects too short (5 digits)", func() { + err := validator.ValidateDateTime("15143") + Expect(err).ToNot(BeNil()) + }) + + It("rejects too long (7 digits)", func() { + err := validator.ValidateDateTime("1514301") + Expect(err).ToNot(BeNil()) + }) + + It("rejects invalid day (00)", func() { + err := validator.ValidateDateTime("001430") + Expect(err).ToNot(BeNil()) + }) + + It("rejects invalid day (32)", func() { + err := validator.ValidateDateTime("321430") + Expect(err).ToNot(BeNil()) + }) + + It("rejects invalid hour (24)", func() { + err := validator.ValidateDateTime("152430") + Expect(err).ToNot(BeNil()) + }) + + It("rejects invalid minute (60)", func() { + err := validator.ValidateDateTime("151460") + Expect(err).ToNot(BeNil()) + }) + + It("rejects invalid minute (99)", func() { + err := validator.ValidateDateTime("151499") + Expect(err).ToNot(BeNil()) + }) + }) + }) + + Describe("ValidateTelegram", func() { + Context("with valid telegram", func() { + It("accepts telegram with all valid fields", func() { + telegram := &dto.ParsedTelegram{ + PriorityIndicator: "FF", + PrimaryAddress: "ZBTJ", + Originator: "KLAX", + DateTime: "151430", + OriginatorDateTime: "151425", + } + err := validator.ValidateTelegram(telegram) + Expect(err).To(BeNil()) + }) + + It("accepts telegram with empty optional fields", func() { + telegram := &dto.ParsedTelegram{ + PriorityIndicator: "", + PrimaryAddress: "ZBTJ", + Originator: "", + DateTime: "151430", + OriginatorDateTime: "", + } + err := validator.ValidateTelegram(telegram) + Expect(err).To(BeNil()) + }) + + It("accepts nil telegram", func() { + err := validator.ValidateTelegram(nil) + Expect(err).To(BeNil()) + }) + }) + + Context("with invalid telegram fields", func() { + It("reports invalid priority indicator", func() { + telegram := &dto.ParsedTelegram{ + PriorityIndicator: "XX", + PrimaryAddress: "ZBTJ", + DateTime: "151430", + } + err := validator.ValidateTelegram(telegram) + Expect(err).ToNot(BeNil()) + Expect(validator.IsAFTNError(err)).To(BeTrue()) + }) + + It("reports invalid primary address", func() { + telegram := &dto.ParsedTelegram{ + PriorityIndicator: "FF", + PrimaryAddress: "TOOLONG", + DateTime: "151430", + } + err := validator.ValidateTelegram(telegram) + Expect(err).ToNot(BeNil()) + Expect(validator.IsAFTNError(err)).To(BeTrue()) + }) + + It("reports invalid originator", func() { + telegram := &dto.ParsedTelegram{ + PriorityIndicator: "FF", + PrimaryAddress: "ZBTJ", + Originator: "KL", + DateTime: "151430", + } + err := validator.ValidateTelegram(telegram) + Expect(err).ToNot(BeNil()) + Expect(validator.IsAFTNError(err)).To(BeTrue()) + }) + + It("reports invalid datetime", func() { + telegram := &dto.ParsedTelegram{ + PriorityIndicator: "FF", + PrimaryAddress: "ZBTJ", + DateTime: "321430", + } + err := validator.ValidateTelegram(telegram) + Expect(err).ToNot(BeNil()) + Expect(validator.IsAFTNError(err)).To(BeTrue()) + }) + + It("reports multiple errors", func() { + telegram := &dto.ParsedTelegram{ + PriorityIndicator: "XX", + PrimaryAddress: "TOOLONG", + Originator: "KL", + DateTime: "321430", + OriginatorDateTime: "991499", + } + err := validator.ValidateTelegram(telegram) + Expect(err).ToNot(BeNil()) + Expect(validator.IsAFTNError(err)).To(BeTrue()) + Expect(validator.GetAFTNErrorType(err)).To(Equal("multiple_errors")) + Expect(err.Error()).To(ContainSubstring("AFTN validation failed")) + }) + }) + }) + + Describe("IsAFTNError", func() { + It("returns true for AFTNError", func() { + err := validator.ValidatePriorityIndicator("XX") + Expect(validator.IsAFTNError(err)).To(BeTrue()) + }) + + It("returns true for AFTNValidationErrors", func() { + telegram := &dto.ParsedTelegram{ + PriorityIndicator: "XX", + PrimaryAddress: "TOOLONG", + } + err := validator.ValidateTelegram(telegram) + Expect(validator.IsAFTNError(err)).To(BeTrue()) + }) + + It("returns false for nil error", func() { + Expect(validator.IsAFTNError(nil)).To(BeFalse()) + }) + }) + + Describe("GetAFTNErrorType", func() { + It("extracts field name from AFTNError", func() { + err := validator.ValidatePriorityIndicator("XX") + Expect(validator.GetAFTNErrorType(err)).To(Equal("priority_indicator")) + }) + + It("returns 'multiple_errors' for AFTNValidationErrors", func() { + telegram := &dto.ParsedTelegram{ + PriorityIndicator: "XX", + PrimaryAddress: "TOOLONG", + } + err := validator.ValidateTelegram(telegram) + Expect(validator.GetAFTNErrorType(err)).To(Equal("multiple_errors")) + }) + + It("returns 'unknown' for non-AFTN errors", func() { + errorType := validator.GetAFTNErrorType(nil) + Expect(errorType).To(Equal("unknown")) + }) + }) +}) diff --git a/internal/adapter/validator/validator_suite_test.go b/internal/adapter/validator/validator_suite_test.go new file mode 100644 index 0000000..b8fe992 --- /dev/null +++ b/internal/adapter/validator/validator_suite_test.go @@ -0,0 +1,13 @@ +package validator_test + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestValidator(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Validator Suite") +} diff --git a/internal/app/processor.go b/internal/app/processor.go index 655952c..a317ede 100644 --- a/internal/app/processor.go +++ b/internal/app/processor.go @@ -3,6 +3,8 @@ package app import ( "caatsm/internal/adapter/dto" "caatsm/internal/adapter/parser" + "caatsm/internal/adapter/validator" + "caatsm/internal/infra/config" "caatsm/internal/infra/log" "caatsm/internal/infra/telemetry" "caatsm/internal/port" @@ -25,6 +27,7 @@ type MessageProcessor struct { publisher port.Publisher logger *zap.Logger telemetry telemetry.Recorder + cfg *config.Config } // ProcessingStatus represents the outcome of the processing pipeline @@ -45,6 +48,7 @@ func NewMessageProcessor( publisher port.Publisher, rec telemetry.Recorder, logger *zap.Logger, + cfg *config.Config, ) *MessageProcessor { return &MessageProcessor{ parser: parser, @@ -52,6 +56,7 @@ func NewMessageProcessor( publisher: publisher, logger: logger, telemetry: rec, + cfg: cfg, } } @@ -138,6 +143,31 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string) } parsed.ErrorReason = "" + // AFTN protocol validation (if enabled) + if p.cfg.AFTN.ValidationEnabled { + if err := validator.ValidateTelegram(parsed); err != nil { + parsed.Status = dto.MessageStatusAFTNError + parsed.ErrorReason = err.Error() + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + span.SetAttributes( + attribute.String("aftn.error_type", validator.GetAFTNErrorType(err)), + ) + p.telemetry.RecordAFTNValidationError(ctx, validator.GetAFTNErrorType(err)) + p.persistRaw(ctx, parsed) + msgLogger.With(zap.String("status", string(parsed.Status))). + Warn("AFTN validation failed", + zap.String("error_type", validator.GetAFTNErrorType(err)), + zap.String("content_preview", truncateContent(parsed.Content, 256)), + zap.Error(err), + ) + latency := parsed.ParsedAt.Sub(receivedAt) + p.telemetry.RecordFailure("aftn_validator") + p.telemetry.RecordProcessingResult(ctx, string(parsed.Status), parsed.Category, latency) + return Permanent(fmt.Errorf("AFTN validation error: %w", err)) + } + } + // Log parsing result span.SetAttributes( attribute.String("telegram.status", string(parsed.Status)), diff --git a/internal/app/processor_bench_test.go b/internal/app/processor_bench_test.go index 1357dd6..b5e73dd 100644 --- a/internal/app/processor_bench_test.go +++ b/internal/app/processor_bench_test.go @@ -3,9 +3,11 @@ package app import ( "caatsm/internal/adapter/dto" "caatsm/internal/adapter/parser" + "caatsm/internal/infra/config" "caatsm/internal/infra/telemetry" "context" "testing" + "time" "go.uber.org/zap" ) @@ -64,8 +66,15 @@ func createBenchmarkProcessor() *MessageProcessor { mockPub := &mockPublisher{} logger := zap.NewNop() recorder := telemetry.NewNoop() + cfg := &config.Config{ + AFTN: config.AFTNConfig{ + ValidationEnabled: false, + MessageGapThreshold: 2 * time.Minute, + EnableSequenceGapDetection: true, + }, + } - return NewMessageProcessor(aviationParser, mockRepo, mockPub, recorder, logger) + return NewMessageProcessor(aviationParser, mockRepo, mockPub, recorder, logger, cfg) } // BenchmarkHandleARR benchmarks processing ARR messages end-to-end @@ -133,8 +142,15 @@ func BenchmarkHandleParseOnly(b *testing.B) { mockPub := &mockPublisher{} logger := zap.NewNop() recorder := telemetry.NewNoop() + cfg := &config.Config{ + AFTN: config.AFTNConfig{ + ValidationEnabled: false, + MessageGapThreshold: 2 * time.Minute, + EnableSequenceGapDetection: true, + }, + } - processor := NewMessageProcessor(aviationParser, mockRepo, mockPub, recorder, logger) + processor := NewMessageProcessor(aviationParser, mockRepo, mockPub, recorder, logger, cfg) ctx := context.Background() b.ResetTimer() diff --git a/internal/app/processor_test.go b/internal/app/processor_test.go index a9de2ef..7f70f64 100644 --- a/internal/app/processor_test.go +++ b/internal/app/processor_test.go @@ -6,8 +6,9 @@ import ( "strings" "time" - "caatsm/internal/adapter/parser" "caatsm/internal/adapter/dto" + "caatsm/internal/adapter/parser" + "caatsm/internal/infra/config" "caatsm/internal/infra/telemetry" "caatsm/internal/port" @@ -122,7 +123,7 @@ var _ = Describe("MessageProcessor", func() { }, err: errors.New("parse failure"), } - proc = NewMessageProcessor(parserStub, repo, pub, telemetry.NewNoop(), logger) + proc = NewMessageProcessor(parserStub, repo, pub, telemetry.NewNoop(), logger, newTestConfig()) err := proc.Handle(ctx, []byte("raw"), "msg-6") Expect(err).To(HaveOccurred()) @@ -146,7 +147,17 @@ var _ = Describe("MessageProcessor", func() { }) func newTestProcessor(p parser.Parser, repo port.Repository, pub port.Publisher) *MessageProcessor { - return NewMessageProcessor(p, repo, pub, telemetry.NewNoop(), zap.NewNop()) + return NewMessageProcessor(p, repo, pub, telemetry.NewNoop(), zap.NewNop(), newTestConfig()) +} + +func newTestConfig() *config.Config { + return &config.Config{ + AFTN: config.AFTNConfig{ + ValidationEnabled: false, // Disabled by default for tests + MessageGapThreshold: 2 * time.Minute, + EnableSequenceGapDetection: true, + }, + } } type stubParser struct { diff --git a/internal/infra/config/config.go b/internal/infra/config/config.go index 7f36b17..a56c974 100644 --- a/internal/infra/config/config.go +++ b/internal/infra/config/config.go @@ -22,6 +22,7 @@ type Config struct { Telemetry TelemetryConfig `koanf:"telemetry"` Monitoring MonitoringConfig `koanf:"monitoring"` DLQ DLQConfig `koanf:"dlq"` + AFTN AFTNConfig `koanf:"aftn"` // Legacy fields for backward compatibility during migration Subscription SubscriptionConfig `koanf:"subscription"` Timeouts TimeoutsConfig `koanf:"timeouts"` @@ -146,6 +147,19 @@ type DLQConfig struct { Subject string `koanf:"subject"` } +// AFTNConfig defines AFTN protocol validation and monitoring settings +type AFTNConfig struct { + // ValidationEnabled enables AFTN protocol validation + ValidationEnabled bool `koanf:"validation_enabled"` + + // MessageGapThreshold is the duration after which the serial reader + // is considered stalled (no messages received). Default: 2 minutes. + MessageGapThreshold time.Duration `koanf:"message_gap_threshold"` + + // EnableSequenceGapDetection enables monitoring for missing sequence numbers + EnableSequenceGapDetection bool `koanf:"enable_sequence_gap_detection"` +} + // MonitoringConfig controls the lightweight HTTP server that exposes health and metrics endpoints. type MonitoringConfig struct { Disabled bool `koanf:"disabled"` @@ -317,6 +331,11 @@ func LoadConfig() (*Config, error) { cfg.Monitoring.HealthTimeout = 2 * time.Second } + // Set AFTN defaults + if cfg.AFTN.MessageGapThreshold == 0 { + cfg.AFTN.MessageGapThreshold = 2 * time.Minute + } + // Validate configuration if err := cfg.Validate(); err != nil { return nil, fmt.Errorf("config validation failed: %w", err) @@ -406,6 +425,10 @@ func (c *Config) Validate() error { if c.Monitoring.HealthTimeout < 0 { return fmt.Errorf("monitoring.health_timeout must be >= 0") } + // Validate AFTN configuration + if c.AFTN.MessageGapThreshold < 0 { + return fmt.Errorf("aftn.message_gap_threshold must be >= 0") + } return nil } diff --git a/internal/infra/metrics/metrics.go b/internal/infra/metrics/metrics.go index 52e2f48..4745140 100644 --- a/internal/infra/metrics/metrics.go +++ b/internal/infra/metrics/metrics.go @@ -1,6 +1,7 @@ package metrics import ( + "context" "math" "net/http" "strings" @@ -25,10 +26,14 @@ const ( MetricJSAPICallsTotal = "caatsm_js_api_calls_total" MetricDBQueriesTotal = "caatsm_db_queries_total" MetricDBQueryLatencySeconds = "caatsm_db_query_latency_seconds" - MetricDLQMessagesTotal = "caatsm_dlq_messages_total" - MetricDLQPublishFailures = "caatsm_dlq_publish_failures_total" - MetricPublishFailuresTotal = "caatsm_publish_failures_total" - MetricNATSConsumerPending = "caatsm_nats_consumer_pending_messages" + MetricDLQMessagesTotal = "caatsm_dlq_messages_total" + MetricDLQPublishFailures = "caatsm_dlq_publish_failures_total" + MetricPublishFailuresTotal = "caatsm_publish_failures_total" + MetricNATSConsumerPending = "caatsm_nats_consumer_pending_messages" + MetricAFTNValidationErrorsTotal = "caatsm_aftn_validation_errors_total" + MetricMessageGapSeconds = "caatsm_message_gap_seconds" + MetricMessageSequenceGapTotal = "caatsm_message_sequence_gap_total" + MetricSerialReaderHealthy = "caatsm_serial_reader_healthy" // Common label keys. LabelStatus = "status" @@ -39,6 +44,7 @@ const ( LabelResult = "result" LabelReason = "reason" LabelOperation = "operation" + LabelErrorType = "error_type" // Standard result label values for caatsm_messages_total. ResultOK = "ok" @@ -78,6 +84,12 @@ var ( // NATS consumer lag metrics. natsConsumerPending *prometheus.GaugeVec + + // AFTN validation and health metrics. + aftnValidationErrorsTotal *prometheus.CounterVec + messageGapSeconds *prometheus.GaugeVec + messageSequenceGapTotal *prometheus.CounterVec + serialReaderHealthy *prometheus.GaugeVec ) func initCollectors() { @@ -154,6 +166,27 @@ func initCollectors() { Help: "Approximate number of pending messages for a JetStream consumer, labelled by stream and consumer.", }, []string{LabelStream, LabelConsumer}) + // AFTN validation and health metrics. + aftnValidationErrorsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: MetricAFTNValidationErrorsTotal, + Help: "Total number of AFTN protocol validation errors, labelled by error type.", + }, []string{LabelErrorType}) + + messageGapSeconds = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: MetricMessageGapSeconds, + Help: "Time in seconds since the last message was received from the serial reader.", + }, []string{LabelStream, LabelConsumer}) + + messageSequenceGapTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: MetricMessageSequenceGapTotal, + Help: "Total number of message sequence gaps detected (missing sequence numbers).", + }, []string{LabelStream, LabelConsumer}) + + serialReaderHealthy = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: MetricSerialReaderHealthy, + Help: "Serial reader health status: 1 = healthy (messages flowing), 0 = stalled (no messages).", + }, []string{LabelStream, LabelConsumer}) + registry.MustRegister( processedCounter, failureCounter, @@ -168,6 +201,10 @@ func initCollectors() { dbQueriesTotal, dbQueryLatency, natsConsumerPending, + aftnValidationErrorsTotal, + messageGapSeconds, + messageSequenceGapTotal, + serialReaderHealthy, ) } @@ -272,6 +309,35 @@ func RecordNATSConsumerPending(stream, consumer string, pending uint64) { natsConsumerPending.WithLabelValues(streamLabel, consumerLabel).Set(float64(pending)) } +// RecordAFTNValidationError increments the AFTN validation error counter for the given error type. +func RecordAFTNValidationError(ctx context.Context, errorType string) { + ensureCollectors() + aftnValidationErrorsTotal.WithLabelValues(labelValue(errorType)).Inc() +} + +// RecordMessageGap records the time gap (in seconds) since the last message was received. +func RecordMessageGap(stream, consumer string, gapSeconds float64) { + ensureCollectors() + messageGapSeconds.WithLabelValues(labelValue(stream), labelValue(consumer)).Set(gapSeconds) +} + +// RecordSequenceGap increments the sequence gap counter when missing sequence numbers are detected. +func RecordSequenceGap(stream, consumer string, gapSize uint64) { + ensureCollectors() + messageSequenceGapTotal.WithLabelValues(labelValue(stream), labelValue(consumer)).Add(float64(gapSize)) +} + +// RecordSerialReaderHealth sets the serial reader health status. +// healthy=1 means messages are flowing normally, healthy=0 means the reader has stalled. +func RecordSerialReaderHealth(stream, consumer string, healthy bool) { + ensureCollectors() + value := 0.0 + if healthy { + value = 1.0 + } + serialReaderHealthy.WithLabelValues(labelValue(stream), labelValue(consumer)).Set(value) +} + func labelValue(value string) string { value = strings.TrimSpace(value) if value == "" { diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index abeadc5..19e64d7 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -9,6 +9,7 @@ import ( "errors" "fmt" "strings" + "sync" "time" "github.com/nats-io/nats.go" @@ -39,6 +40,11 @@ type Consumer struct { // State consecutiveProcessErrors int + + // Message tracking for health monitoring + lastMessageTime time.Time + lastMessageSequence uint64 + messageGapMutex sync.RWMutex } // consumerConfig holds normalized consumer configuration values. @@ -352,6 +358,11 @@ func (c *Consumer) startJetStream(ctx context.Context) error { fetchErrorStreak = 0 } + // Update message tracking for health monitoring (track each message) + for _, msg := range msgs { + c.updateMessageTracking(msg) + } + // Process batch c.batchProcessor.ProcessBatch(ctx, msgs) } @@ -391,6 +402,22 @@ func (c *Consumer) emitConsumerStats(ctx context.Context) { zap.Uint64("pending", info.NumPending), ) obsmetrics.RecordNATSConsumerPending(c.config.streamName, c.config.consumerName, info.NumPending) + + // Record AFTN health metrics + gapSeconds := c.getMessageGapSeconds() + healthy := c.isSerialReaderHealthy() + + obsmetrics.RecordMessageGap(c.config.streamName, c.config.consumerName, gapSeconds) + obsmetrics.RecordSerialReaderHealth(c.config.streamName, c.config.consumerName, healthy) + + if !healthy { + c.logger.Warn("Serial reader appears stalled - no messages received recently", + zap.String("stream", c.config.streamName), + zap.String("consumer", c.config.consumerName), + zap.Float64("gap_seconds", gapSeconds), + zap.Duration("threshold", c.cfg.AFTN.MessageGapThreshold), + ) + } } } } @@ -423,3 +450,67 @@ func (c *Consumer) Shutdown(ctx context.Context) error { return fmt.Errorf("nats drain timeout: %w", closeCtx.Err()) } } + +// updateMessageTracking updates the last message time and sequence number for health monitoring. +// This should be called for every message received to track message flow and detect gaps. +func (c *Consumer) updateMessageTracking(msg *nats.Msg) { + if msg == nil { + return + } + + c.messageGapMutex.Lock() + defer c.messageGapMutex.Unlock() + + now := time.Now() + c.lastMessageTime = now + + // Extract sequence number from message metadata + if meta, err := msg.Metadata(); err == nil { + currentSeq := meta.Sequence.Stream + + // Detect sequence gaps if we have a previous sequence + if c.lastMessageSequence > 0 && c.cfg.AFTN.EnableSequenceGapDetection { + if currentSeq > c.lastMessageSequence+1 { + gapSize := currentSeq - c.lastMessageSequence - 1 + c.logger.Warn("Message sequence gap detected", + zap.String("stream", c.config.streamName), + zap.String("consumer", c.config.consumerName), + zap.Uint64("last_sequence", c.lastMessageSequence), + zap.Uint64("current_sequence", currentSeq), + zap.Uint64("gap_size", gapSize), + ) + obsmetrics.RecordSequenceGap(c.config.streamName, c.config.consumerName, gapSize) + } + } + + c.lastMessageSequence = currentSeq + } +} + +// getMessageGapSeconds returns the number of seconds since the last message was received. +// Returns 0 if no message has been received yet. +func (c *Consumer) getMessageGapSeconds() float64 { + c.messageGapMutex.RLock() + defer c.messageGapMutex.RUnlock() + + if c.lastMessageTime.IsZero() { + return 0 + } + + return time.Since(c.lastMessageTime).Seconds() +} + +// isSerialReaderHealthy returns true if messages are being received within the threshold. +// Returns false if the gap exceeds the configured message gap threshold. +func (c *Consumer) isSerialReaderHealthy() bool { + c.messageGapMutex.RLock() + defer c.messageGapMutex.RUnlock() + + // If we haven't received any messages yet, consider it healthy (initial state) + if c.lastMessageTime.IsZero() { + return true + } + + gap := time.Since(c.lastMessageTime) + return gap < c.cfg.AFTN.MessageGapThreshold +} diff --git a/internal/infra/telemetry/telemetry.go b/internal/infra/telemetry/telemetry.go index 083b0e0..9ea207b 100644 --- a/internal/infra/telemetry/telemetry.go +++ b/internal/infra/telemetry/telemetry.go @@ -42,6 +42,9 @@ type Recorder interface { // RecordJSAPICall records a JetStream API call. RecordJSAPICall(operation string) + + // RecordAFTNValidationError records an AFTN protocol validation failure. + RecordAFTNValidationError(ctx context.Context, errorType string) } // ProvideRecorder wires a composite Recorder based on configuration flags. @@ -100,6 +103,9 @@ func (n *noopRecorder) RecordDLQPublishFailure(ctx context.Context, stream, cons func (n *noopRecorder) RecordJSAPICall(operation string) { } +func (n *noopRecorder) RecordAFTNValidationError(ctx context.Context, errorType string) { +} + // compositeRecorder fans out all calls to a slice of underlying recorders. type compositeRecorder struct { recorders []Recorder @@ -167,6 +173,12 @@ func (c *compositeRecorder) RecordJSAPICall(operation string) { } } +func (c *compositeRecorder) RecordAFTNValidationError(ctx context.Context, errorType string) { + for _, r := range c.recorders { + r.RecordAFTNValidationError(ctx, errorType) + } +} + // promRecorder delegates to the Prometheus metrics helpers in the // internal/infra/metrics package. type promRecorder struct{} @@ -213,6 +225,10 @@ func (p *promRecorder) RecordJSAPICall(operation string) { obsmetrics.RecordJSAPICall(operation) } +func (p *promRecorder) RecordAFTNValidationError(ctx context.Context, errorType string) { + obsmetrics.RecordAFTNValidationError(ctx, errorType) +} + // otelRecorder creates and records OpenTelemetry metrics for the CAATSM // processor. It intentionally focuses on a small set of high-value metrics to // avoid duplicating the full Prometheus surface. @@ -305,4 +321,9 @@ func (o *otelRecorder) RecordDLQPublishFailure(ctx context.Context, stream, cons func (o *otelRecorder) RecordJSAPICall(operation string) { } +func (o *otelRecorder) RecordAFTNValidationError(ctx context.Context, errorType string) { + // AFTN validation metrics are primarily tracked via Prometheus. + // This is a no-op for OTEL recorder. +} + diff --git a/pkg/di/wire_gen.go b/pkg/di/wire_gen.go index cb53fa4..64d408e 100644 --- a/pkg/di/wire_gen.go +++ b/pkg/di/wire_gen.go @@ -51,7 +51,7 @@ func buildAppComponents() (*appComponents, error) { return nil, err } recorder := telemetry.ProvideRecorder(configConfig) - messageProcessor := app.NewMessageProcessor(parserParser, repository, publisher, recorder, logger) + messageProcessor := app.NewMessageProcessor(parserParser, repository, publisher, recorder, logger, configConfig) consumer, err := nats.ProvideConsumer(conn, jetStreamContext, messageProcessor, configConfig, recorder, logger) if err != nil { return nil, err @@ -95,7 +95,7 @@ func buildAppComponentsWithConfig(cfg *config.Config) (*appComponents, error) { return nil, err } recorder := telemetry.ProvideRecorder(cfg) - messageProcessor := app.NewMessageProcessor(parserParser, repository, publisher, recorder, logger) + messageProcessor := app.NewMessageProcessor(parserParser, repository, publisher, recorder, logger, cfg) consumer, err := nats.ProvideConsumer(conn, jetStreamContext, messageProcessor, cfg, recorder, logger) if err != nil { return nil, err