Merge pull request #25 from CoreyLeath-code/feature/aws-firehose-s3

feat: add Amazon Data Firehose + S3 telemetry mirror

Author: CoreyLeath-code

.env.example

@@ -27,6 +27,17 @@ SNOWFLAKE_DATABASE=
 SNOWFLAKE_SCHEMA=
 SNOWFLAKE_WAREHOUSE=
 
+# ---------------------------------------------------------------------------
+# AWS telemetry mirror (optional — disabled for local development)
+# ---------------------------------------------------------------------------
+# When enabled, accepted inference logs are queued and mirrored to Amazon Data
+# Firehose. Use the AWS SDK default credential chain; prefer workload roles/IRSA
+# in AWS instead of committing long-lived access keys.
+FIREHOSE_ENABLED=false
+FIREHOSE_DELIVERY_STREAM=sentinelai-telemetry
+FIREHOSE_QUEUE_SIZE=1000
+AWS_REGION=us-east-1
+
 # ---------------------------------------------------------------------------
 # LLM / Ollama (optional — llm-guard falls back to stub if unreachable)
 # ---------------------------------------------------------------------------

.github/workflows/aws-telemetry.yml

@@ -0,0 +1,75 @@
+name: AWS Telemetry Validation
+
+on:
+  pull_request:
+    branches: [main]
+    paths:
+      - "ingestion-service/**"
+      - "terraform/**"
+      - ".github/workflows/aws-telemetry.yml"
+  push:
+    branches: [main]
+    paths:
+      - "ingestion-service/**"
+      - "terraform/**"
+      - ".github/workflows/aws-telemetry.yml"
+
+permissions:
+  contents: read
+
+jobs:
+  firehose-producer:
+    name: Firehose producer unit tests
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+
+    defaults:
+      run:
+        working-directory: ingestion-service
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: "1.24.x"
+          cache-dependency-path: ingestion-service/go.sum
+
+      - name: Verify module files are tidy
+        run: |
+          go mod tidy
+          git diff --exit-code -- go.mod go.sum
+
+      - name: Run ingestion tests
+        run: go test ./...
+
+  terraform:
+    name: Terraform fmt and validate
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+
+    defaults:
+      run:
+        working-directory: terraform
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Terraform
+        uses: hashicorp/setup-terraform@v3
+        with:
+          terraform_version: "1.13.3"
+
+      - name: Verify Terraform formatting
+        run: |
+          terraform fmt -recursive
+          git diff --exit-code -- .
+
+      - name: Initialize providers without backend
+        run: terraform init -backend=false -input=false
+
+      - name: Validate configuration
+        run: terraform validate

README.md

@@ -27,6 +27,10 @@ services:
       WAREHOUSE_MODE: ${WAREHOUSE_MODE:-postgres}
       DATABASE_URL:   ${DATABASE_URL:-postgres://sentinel:sentinel@postgres:5432/sentinel?sslmode=disable}
       PORT:           "8080"
+      FIREHOSE_ENABLED: ${FIREHOSE_ENABLED:-false}
+      FIREHOSE_DELIVERY_STREAM: ${FIREHOSE_DELIVERY_STREAM:-sentinelai-telemetry}
+      FIREHOSE_QUEUE_SIZE: ${FIREHOSE_QUEUE_SIZE:-1000}
+      AWS_REGION: ${AWS_REGION:-us-east-1}
       # Disabled by default; CI enables this to verify multiple replicas route through NGINX.
       EXPOSE_INSTANCE_ID: ${EXPOSE_INSTANCE_ID:-false}
     expose:

docker-compose.yml

@@ -0,0 +1,97 @@
+# Amazon Data Firehose + S3 telemetry path
+
+SentinelAI can optionally mirror accepted inference telemetry from the Go ingestion service to Amazon Data Firehose. Firehose buffers the records and delivers GZIP-compressed newline-delimited JSON (NDJSON) objects to a private S3 telemetry bucket.
+
+This AWS path is **optional**. Local Docker Compose keeps `FIREHOSE_ENABLED=false`, so PostgreSQL remains the default local persistence path and no AWS account is required for development.
+
+## Architecture
+
+```mermaid
+flowchart LR
+    Client[Model / application] --> Gateway[NGINX ingestion gateway]
+    Gateway --> Go[Go ingestion replicas]
+    Go --> DB[(PostgreSQL / Snowflake path)]
+    Go -. bounded fail-open mirror .-> Queue[In-memory Firehose queue]
+    Queue --> Firehose[Amazon Data Firehose]
+    Firehose --> S3[(Amazon S3 telemetry lake)]
+    Firehose --> CW[CloudWatch delivery logs]
+```
+
+The Firehose mirror intentionally does not participate in `/ready`. A temporary AWS failure increments `ingestion_firehose_records_total{status="error"}` and is logged, while the primary ingestion response continues to reflect the primary warehouse write. If the bounded queue fills, records are dropped from the mirror and counted with `status="dropped"` rather than allowing telemetry backpressure to take down ingestion.
+
+## Provision the AWS resources
+
+Prerequisites:
+
+- Terraform 1.6+
+- AWS credentials available through the standard AWS credential chain
+- Permission to create S3, Firehose, CloudWatch Logs, IAM policy/role, and ECR resources
+
+```bash
+cd terraform
+terraform init
+terraform fmt -check
+terraform validate
+terraform plan
+terraform apply
+```
+
+The default configuration creates:
+
+- a private, versioned S3 bucket with SSE-S3 encryption;
+- a 30-day telemetry lifecycle policy;
+- an Amazon Data Firehose delivery stream named `sentinelai-telemetry`;
+- 60-second / 5-MiB buffering with GZIP compression;
+- time-partitioned S3 keys under `inference/year=.../month=.../day=.../hour=.../`;
+- a CloudWatch log group for Firehose delivery errors;
+- a Firehose service role with only the S3 and CloudWatch permissions it needs;
+- a separate `sentinelai-firehose-writer` IAM policy granting `PutRecord` and `PutRecordBatch` to the SentinelAI workload;
+- the existing SentinelAI ECR repository, now defined in a valid Terraform `.tf` file.
+
+Use `terraform output` after apply to retrieve the generated S3 bucket name, stream ARN, stream name, and writer-policy ARN.
+
+## Give the ingestion workload permission
+
+Do not put long-lived AWS access keys in the repository. Attach the Terraform output `firehose_writer_policy_arn` to the workload identity used by SentinelAI. On EKS, the intended production pattern is an IAM role associated with the ingestion service account (IRSA / EKS workload identity).
+
+For local development against a real AWS account, the AWS SDK for Go v2 uses its normal credential provider chain. Keep credentials outside the repo.
+
+## Enable the mirror
+
+```bash
+export FIREHOSE_ENABLED=true
+export FIREHOSE_DELIVERY_STREAM=sentinelai-telemetry
+export FIREHOSE_QUEUE_SIZE=1000
+export AWS_REGION=us-east-1
+```
+
+Then start SentinelAI and send a normal inference log:
+
+```bash
+curl -X POST http://localhost:8080/log \
+  -H "Content-Type: application/json" \
+  -d '{"model_id":"demo","model_version":"v1","latency_ms":120,"tokens_in":32,"tokens_out":64,"status":"ok"}'
+```
+
+The producer appends a newline to every JSON record before `PutRecord`. That keeps individual events parseable after Firehose concatenates buffered records into S3 objects.
+
+## Observability
+
+The ingestion service exports:
+
+```text
+ingestion_firehose_records_total{status="queued"}
+ingestion_firehose_records_total{status="delivered"}
+ingestion_firehose_records_total{status="error"}
+ingestion_firehose_records_total{status="dropped"}
+```
+
+CloudWatch delivery logs cover the managed Firehose-to-S3 leg. Application metrics cover the producer-side queue and `PutRecord` result.
+
+## Failure semantics
+
+This feature is a telemetry mirror, not a transactional dual-write guarantee. The in-memory queue is intentionally bounded and is not persisted across process termination. AWS SDK retries may also produce duplicate Firehose records in some failure scenarios. Consumers should therefore treat the S3 telemetry dataset as at-least-once/best-effort observability data and use stable event identifiers if strict de-duplication is later required.
+
+## Cost control
+
+Firehose, S3, CloudWatch Logs, and related data transfer can incur AWS charges. The default 30-day S3 expiration and 14-day CloudWatch log retention are intended to keep a portfolio/dev deployment bounded. Review the Terraform plan and AWS pricing before leaving the stack running.

docs/aws-firehose-s3.md

@@ -1,5 +1,5 @@
 # ingestion-service — multi-stage Go build
-FROM golang:1.21-alpine AS builder
+FROM golang:1.24-alpine AS builder
 WORKDIR /app
 COPY go.mod go.sum* ./
 RUN go mod download

ingestion-service/Dockerfile

@@ -0,0 +1,144 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/service/firehose"
+	"github.com/aws/aws-sdk-go-v2/service/firehose/types"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	defaultFirehoseQueueSize = 1000
+	firehosePublishTimeout   = 5 * time.Second
+)
+
+var firehoseRecordsTotal = prometheus.NewCounterVec(
+	prometheus.CounterOpts{
+		Name: "ingestion_firehose_records_total",
+		Help: "Inference telemetry records handled by the optional Amazon Data Firehose mirror.",
+	},
+	[]string{"status"},
+)
+
+func init() {
+	prometheus.MustRegister(firehoseRecordsTotal)
+}
+
+type firehoseAPI interface {
+	PutRecord(context.Context, *firehose.PutRecordInput, ...func(*firehose.Options)) (*firehose.PutRecordOutput, error)
+}
+
+type firehosePublisher struct {
+	client     firehoseAPI
+	streamName string
+}
+
+type firehoseDispatcher struct {
+	publisher *firehosePublisher
+	queue     chan InferenceLog
+}
+
+func newFirehoseDispatcher() (*firehoseDispatcher, error) {
+	if !strings.EqualFold(strings.TrimSpace(os.Getenv("FIREHOSE_ENABLED")), "true") {
+		return nil, nil
+	}
+
+	streamName := strings.TrimSpace(os.Getenv("FIREHOSE_DELIVERY_STREAM"))
+	if streamName == "" {
+		return nil, errors.New("FIREHOSE_DELIVERY_STREAM is required when FIREHOSE_ENABLED=true")
+	}
+
+	region := strings.TrimSpace(os.Getenv("AWS_REGION"))
+	if region == "" {
+		region = "us-east-1"
+	}
+
+	queueSize := defaultFirehoseQueueSize
+	if raw := strings.TrimSpace(os.Getenv("FIREHOSE_QUEUE_SIZE")); raw != "" {
+		parsed, err := strconv.Atoi(raw)
+		if err != nil || parsed < 1 {
+			return nil, fmt.Errorf("FIREHOSE_QUEUE_SIZE must be a positive integer: %q", raw)
+		}
+		queueSize = parsed
+	}
+
+	cfg, err := config.LoadDefaultConfig(context.Background(), config.WithRegion(region))
+	if err != nil {
+		return nil, fmt.Errorf("load AWS configuration: %w", err)
+	}
+
+	dispatcher := &firehoseDispatcher{
+		publisher: &firehosePublisher{
+			client:     firehose.NewFromConfig(cfg),
+			streamName: streamName,
+		},
+		queue: make(chan InferenceLog, queueSize),
+	}
+	go dispatcher.run()
+
+	return dispatcher, nil
+}
+
+func (d *firehoseDispatcher) enqueue(entry InferenceLog) {
+	select {
+	case d.queue <- entry:
+		firehoseRecordsTotal.WithLabelValues("queued").Inc()
+	default:
+		firehoseRecordsTotal.WithLabelValues("dropped").Inc()
+		log.Printf("firehose mirror queue full; dropping telemetry record for model=%s", entry.ModelID)
+	}
+}
+
+func (d *firehoseDispatcher) run() {
+	for entry := range d.queue {
+		ctx, cancel := context.WithTimeout(context.Background(), firehosePublishTimeout)
+		err := d.publisher.publish(ctx, entry)
+		cancel()
+
+		if err != nil {
+			firehoseRecordsTotal.WithLabelValues("error").Inc()
+			log.Printf("firehose PutRecord failed for model=%s: %v", entry.ModelID, err)
+			continue
+		}
+		firehoseRecordsTotal.WithLabelValues("delivered").Inc()
+	}
+}
+
+func (p *firehosePublisher) publish(ctx context.Context, entry InferenceLog) error {
+	payload, err := encodeFirehoseRecord(entry)
+	if err != nil {
+		return err
+	}
+
+	_, err = p.client.PutRecord(ctx, &firehose.PutRecordInput{
+		DeliveryStreamName: &p.streamName,
+		Record: &types.Record{
+			Data: payload,
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("put record: %w", err)
+	}
+	return nil
+}
+
+func encodeFirehoseRecord(entry InferenceLog) ([]byte, error) {
+	payload, err := json.Marshal(entry)
+	if err != nil {
+		return nil, fmt.Errorf("marshal inference log: %w", err)
+	}
+
+	// Firehose concatenates records inside delivered objects. NDJSON keeps each
+	// inference event independently parseable after buffering and compression.
+	return append(payload, '\n'), nil
+}