From Alert Storm to Sub‑Second Insight: Building a Production‑Grade AIOps Platform with Spring Boot 3.x

Why traditional monitoring becomes an alert storm

A large e‑commerce platform processes over 3 million orders daily, generating thousands of alerts per second from Prometheus, Grafana, logs, tracing and bots. A short‑lived spike in Redis, DB or network triggers dozens of derivative alerts, and operators must manually correlate root causes across metrics, logs, call‑chains, CMDB and release systems.

The core problem is not missing monitoring but the lack of a complete event‑handling pipeline that can reliably ingest, normalize, de‑duplicate, suppress, route, store and deliver alerts in real time.

Platform goals, SLOs and non‑functional constraints

Signal unification:

original signal → standard → deduplication → suppression → risk scoring → routing → real‑time push → unread & replay → claim → close → audit

SLO examples: alert ingest availability ≥ 99.95 % (API/MQ success ≤ 500 ms), alert store P99 ≤ 500 ms, WebSocket push P99 ≤ 1 s, authentication P99 ≤ 30 ms, reconnection ≤ 10 s, event loss = 0.

Non‑functional constraints: logical tenant isolation, permission changes must propagate within seconds, WebSocket must allow duplicate delivery but enforce idempotent handling, graceful shutdown must stop accepting new connections before pod termination.

High‑level architecture

The system consists of a client layer (web/IM), an ingest API, a standardization service, a fingerprinting & aggregation layer, an outbox table, RocketMQ as the event bus, a push dispatcher, and a WebSocket/STOMP gateway that delivers PushEnvelope messages to users.

Key design decisions

REST vs WebSocket : REST APIs handle queries and commands (create, acknowledge, resolve) while WebSocket is used only for real‑time notifications. Commands that change state are never sent via STOMP to keep auditability and idempotence.

Outbox pattern : Alert state changes and the corresponding OutboxEvent are written in the same DB transaction. A separate publisher reads pending rows with FOR UPDATE SKIP LOCKED, marks them SENDING, publishes to RocketMQ, and updates the status to PUBLISHED. This guarantees exactly‑once delivery to the message bus.

Simple Broker vs Broker Relay : In‑process SimpleBroker is only for development. Production clusters must use a STOMP Broker Relay (e.g., RabbitMQ, ActiveMQ) or an external real‑time messaging layer to avoid single‑node bottlenecks.

Permission model : Permissions are resolved from a snapshot cache (Redis versioned snapshot). The cache key is iam:user:{userId}:version. On permission change the version is incremented, a PermissionChangedEvent is emitted, and all nodes invalidate their local Caffeine cache.

WebSocket authentication : During the STOMP CONNECT frame the server extracts a Bearer token, validates it with a JwtDecoder, converts it to a JwtAuthenticationToken, and stores it as the principal. Subsequent SUBSCRIBE and SEND frames are authorized against the permission snapshot.

Unified alert model (Java record)

public record AlertEvent(
    String eventId,
    String source,
    String tenantId,
    String serviceId,
    String environment,
    String alertName,
    AlertSeverity severity,
    AlertStatus status,
    String fingerprint,
    Instant occurredAt,
    Instant receivedAt,
    Map<String, String> labels,
    Map<String, Object> annotations,
    String schemaVersion) {}

Key fields: eventId (global unique, used for ingest idempotency), fingerprint (dedup key), tenantId, serviceId, environment (used for routing and authorization), and version (optimistic‑lock column).

Transactional state change example

@Transactional
public void changeAlertStatus(AlertAggregate alert, AlertChangedEvent event) {
    alertRepository.update(alert);
    outboxRepository.append(OutboxEvent.pending(
        event.eventId(),
        "ALERT",
        String.valueOf(alert.id()),
        event.getClass().getSimpleName(),
        "aiops-alert-event",
        String.valueOf(alert.id()),
        json.write(event)));
}

Outbox table (MySQL)

CREATE TABLE outbox_event (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    event_id CHAR(26) NOT NULL,
    aggregate_type VARCHAR(64) NOT NULL,
    aggregate_id VARCHAR(64) NOT NULL,
    event_type VARCHAR(128) NOT NULL,
    topic VARCHAR(128) NOT NULL,
    message_key VARCHAR(128) NOT NULL,
    payload JSON NOT NULL,
    status VARCHAR(16) NOT NULL,
    retry_count INT NOT NULL DEFAULT 0,
    next_retry_at DATETIME(3) NOT NULL,
    locked_by VARCHAR(128) NULL,
    locked_at DATETIME(3) NULL,
    published_at DATETIME(3) NULL,
    last_error VARCHAR(1000) NULL,
    created_at DATETIME(3) NOT NULL,
    updated_at DATETIME(3) NOT NULL,
    UNIQUE KEY uk_outbox_event_id (event_id),
    KEY idx_outbox_publish (status, next_retry_at, id)
);

Permission snapshot record

public record UserAccessSnapshot(
    String userId,
    String tenantId,
    long version,
    Set<String> permissions,
    Set<String> allowedServiceIds,
    Set<String> allowedEnvironments,
    Instant expiresAt) {}

WebSocket/STOMP configuration (Spring)

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
    @Bean
    public TaskScheduler messageBrokerTaskScheduler() {
        ThreadPoolTaskScheduler s = new ThreadPoolTaskScheduler();
        s.setPoolSize(2);
        s.setThreadNamePrefix("stomp-heartbeat-");
        s.initialize();
        return s;
    }
    @Override
    public void registerStompEndpoints(StompEndpointRegistry r) {
        r.addEndpoint("/ws").setAllowedOrigins("https://ops.example.com");
        r.addEndpoint("/ws-sockjs").setAllowedOrigins("https://ops.example.com").withSockJS();
    }
    @Override
    public void configureMessageBroker(MessageBrokerRegistry r) {
        r.setApplicationDestinationPrefixes("/app");
        r.setUserDestinationPrefix("/user");
        r.enableSimpleBroker("/topic", "/queue")
         .setTaskScheduler(messageBrokerTaskScheduler())
         .setHeartbeatValue(new long[]{15000,15000});
    }
    @Override
    public void configureClientInboundChannel(ChannelRegistration r) {
        r.interceptors(authenticationInterceptor, authorizationInterceptor);
        r.taskExecutor().corePoolSize(8).maxPoolSize(32);
    }
    @Override
    public void configureWebSocketTransport(WebSocketTransportRegistration r) {
        r.setMessageSizeLimit(65536)
         .setSendTimeLimit(10000)
         .setSendBufferSizeLimit(1048576);
    }
}

STOMP CONNECT authentication interceptor

@Component
@RequiredArgsConstructor
public class StompAuthenticationInterceptor implements ChannelInterceptor {
    private final JwtDecoder jwtDecoder;
    private final Converter<Jwt, ? extends AbstractAuthenticationToken> converter;
    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
        if (accessor == null || accessor.getCommand() == null) return message;
        if (StompCommand.CONNECT.equals(accessor.getCommand())) {
            String auth = accessor.getFirstNativeHeader(HttpHeaders.AUTHORIZATION);
            String token = resolveBearerToken(auth);
            Jwt jwt = jwtDecoder.decode(token);
            AbstractAuthenticationToken authToken = converter.convert(jwt);
            if (authToken == null) throw new BadCredentialsException("cannot convert jwt authentication");
            accessor.setUser(authToken);
        }
        return message;
    }
    private static String resolveBearerToken(String auth) {
        if (auth == null || !auth.startsWith("Bearer ")) throw new BadCredentialsException("missing bearer token");
        String token = auth.substring(7).trim();
        if (token.isEmpty()) throw new BadCredentialsException("empty bearer token");
        return token;
    }
}

STOMP subscription authorization interceptor

@Component
@RequiredArgsConstructor
public class StompAuthorizationInterceptor implements ChannelInterceptor {
    private final PermissionSnapshotService permissionSnapshotService;
    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
        if (accessor == null || accessor.getCommand() == null) return message;
        if (StompCommand.SUBSCRIBE.equals(accessor.getCommand())) {
            Principal p = requirePrincipal(accessor);
            String dest = Objects.requireNonNull(accessor.getDestination(), "destination");
            authorizeSubscribe(p.getName(), dest);
        }
        if (StompCommand.SEND.equals(accessor.getCommand())) {
            Principal p = requirePrincipal(accessor);
            String dest = Objects.requireNonNull(accessor.getDestination(), "destination");
            authorizeSend(p.getName(), dest);
        }
        return message;
    }
    private void authorizeSubscribe(String userId, String destination) {
        UserAccessSnapshot snap = permissionSnapshotService.load(userId);
        boolean allowed = switch (destination) {
            case "/user/queue/alerts" -> snap.permissions().contains("alert:read");
            case "/user/queue/system" -> true;
            case "/topic/public/platform-status" -> true;
            default -> false;
        };
        if (!allowed) throw new AccessDeniedException("subscription denied: " + destination);
    }
    private void authorizeSend(String userId, String destination) {
        if (!destination.startsWith("/app/client-ack")) {
            throw new AccessDeniedException("client send destination denied");
        }
    }
    private static Principal requirePrincipal(StompHeaderAccessor accessor) {
        if (accessor.getUser() == null) throw new BadCredentialsException("unauthenticated stomp session");
        return accessor.getUser();
    }
}

Push envelope and metrics

public record PushEnvelope<T>(
    String messageId,
    long sequence,
    String eventType,
    String schemaVersion,
    Instant occurredAt,
    T payload) {}

Clients must deduplicate by messageId, ignore out‑of‑order sequence, and reject older version values.

Back‑pressure and merging

Ingress throttles per‑source QPS, payload size and concurrent connections.

STOMP limits frame size and per‑session send buffer.

P0 alerts are sent immediately; lower‑severity alerts are batched in a 200 ms window.

Slow consumers are disconnected after exceeding sendTimeLimit or buffer size, forcing replay via the REST /api/v1/notification-events endpoint.

Database schema highlights

IAM tables ( iam_user, iam_role, iam_permission, iam_user_role, iam_role_permission, iam_role_data_scope) are versioned for cache invalidation.

Alert table includes tenant_id, service_id, environment, fingerprint, severity, status, optimistic‑lock version, and JSON labels / annotations.

Optimistic‑lock update example uses WHERE version = :expectedVersion and returns 409 on conflict.

Audit log records immutable events with actor, action, resource, result and timestamps.

Kubernetes deployment and graceful shutdown

Pod termination grace period is set to 45 s. preStop sleeps 10 s to let the readiness probe reject new traffic, then the application stops pulling new push tasks, notifies clients with a SERVER_DRAINING message, and finally closes remaining sessions.

Observability and SLO metrics (Prometheus)

WebSocket: aiops_ws_sessions_active, aiops_ws_connect_total{result}, aiops_ws_message_send_total{event_type,result}, aiops_ws_message_send_duration_seconds.

Alert pipeline: aiops_alert_ingest_total{source,result}, aiops_alert_dedup_total{result}, aiops_alert_processing_duration_seconds, aiops_outbox_pending, aiops_outbox_publish_duration_seconds.

Permission: aiops_permission_check_total{result}, aiops_permission_cache_hit_ratio, aiops_access_denied_total{permission}.

Low‑cardinality label rule: never use userId, alertId, messageId or full destination as Prometheus labels; put them into structured logs instead.

Testing strategy

Unit tests for state machine, fingerprint, permission expressions.

Slice tests for controllers, security, MyBatis mappers.

Integration tests with Testcontainers for MySQL, Redis, RocketMQ, Outbox.

WebSocket tests covering CONNECT, SUBSCRIBE, permission denial, and reconnection.

Contract tests for event schema compatibility.

Load tests (k6/Gatling) for stable connections, burst pushes, reconnection storms, and rolling upgrades.

Chaos tests (Chaos Mesh) for pod restarts, Redis/MQ latency spikes, network partitions.

Post‑mortem examples

Full GC & heartbeat timeout : caused by unbounded per‑session send buffers and slow consumers. Fixed by setting sendTimeLimit, sendBufferSizeLimit, and rejecting oversized messages.

Permission cache stampede : simultaneous login caused DB thundering herd. Resolved with versioned snapshots, short‑lived locks, and exponential back‑off on cache miss.

Cross‑service over‑privilege : API accepted serviceName from client without server‑side verification. Fixed by loading the alert’s real service from DB before permission check.

Redis Pub/Sub loss : relied on Pub/Sub for durability. Re‑architected to use Outbox + RocketMQ as the reliable backbone; Pub/Sub now only carries cache‑invalidation signals.

Rolling‑upgrade connection storm : pods terminated before readiness turned false, causing 1006 errors. Added proper readiness probe, graceful shutdown, and client exponential back‑off.

Evolution roadmap

Stage 1 – modular monolith (Spring Boot, MySQL, Redis, simple broker).

Stage 2 – reliable event pipeline (Outbox, RocketMQ, external broker).

Stage 3 – multi‑tenant platform with OIDC, data‑domain enforcement, temporary grants.

Stage 4 – AIOps enhancements: streaming aggregation, topology correlation, root‑cause ranking, automated remediation.

Stage 5 – multi‑region active‑active deployment with global IDs, cross‑region routing, disaster‑recovery SLAs.

Release checklist

Security: JWT validation, method‑level security, origin whitelist, permission‑based STOMP destinations.

Reliability: transactional Outbox, idempotent consumers, replay API, back‑pressure limits.

Operations: readiness/liveness/startup probes, graceful termination, HPA based on connections & queue depth, alerts on Outbox backlog, MQ lag, WebSocket errors.

Observability: full metric set, structured JSON logs with traceId, requestId, tenantId, actorId, action, resourceId, result, durationMs.

Testing: unit, slice, integration, contract, load, chaos – all passing.

When these foundations are solid, higher‑level features such as alert deduplication, AI‑driven root‑cause suggestions and automated remediation become reliable extensions rather than fragile add‑ons.