DDoS Protection Demo - Quick Start Guide

-- VoltDB Schema for DDoS Detection with ML Classification

-- Create dummy table for time window calculations
CREATE TABLE dummy (x VARCHAR(1) NOT NULL, PRIMARY KEY (x));
INSERT INTO dummy VALUES ('X');

-- Traffic classification aggregates with time windows
CREATE TABLE traffic_aggregates (
    traffic_class VARCHAR(20) NOT NULL,
    tickdate TIMESTAMP NOT NULL,
    timescale VARCHAR(20) NOT NULL,
    flow_count BIGINT NOT NULL,
    cumulative_flows BIGINT NOT NULL,
    created_at TIMESTAMP DEFAULT NOW,
    PRIMARY KEY (traffic_class, tickdate, timescale)
);

-- State table for cumulative flows per traffic class
CREATE TABLE traffic_state (
    traffic_class VARCHAR(20) NOT NULL,
    window_start TIMESTAMP NOT NULL,
    cumulative_flows BIGINT NOT NULL,
    last_update TIMESTAMP NOT NULL,
    PRIMARY KEY (traffic_class, window_start)
);

-- Router connection events with 5-minute TTL
CREATE TABLE router_connection_stats (
    event_id BIGINT NOT NULL,
    connection_key VARCHAR(20) NOT NULL,
    router_a INTEGER NOT NULL,
    router_b INTEGER NOT NULL,
    traffic_class VARCHAR(20) NOT NULL,
    created_at TIMESTAMP DEFAULT NOW NOT NULL,
    PRIMARY KEY (event_id, connection_key, traffic_class)
) USING TTL 5 MINUTES ON COLUMN created_at;

-- Create indexes for efficient queries
CREATE INDEX idx_tickdate ON traffic_aggregates (tickdate);
CREATE INDEX idx_router_connection_key ON router_connection_stats (connection_key);
CREATE INDEX idx_router_created_at ON router_connection_stats (created_at);

-- Partition tables for parallel processing
PARTITION TABLE traffic_aggregates ON COLUMN traffic_class;
PARTITION TABLE traffic_state ON COLUMN traffic_class;
PARTITION TABLE router_connection_stats ON COLUMN traffic_class;

-- View for traffic class percentages
CREATE VIEW percentagesView AS
SELECT traffic_class, SUM(flow_count) as total_flows
FROM traffic_aggregates
WHERE timescale = 'MINUTES_15'
GROUP BY traffic_class;

-- View for current router connections (last 5 min)
CREATE VIEW current_router_connections AS
SELECT connection_key, router_a, router_b, traffic_class,
       COUNT(*) as event_count, MAX(created_at) as last_activity
FROM router_connection_stats
GROUP BY connection_key, router_a, router_b, traffic_class;

-- Register stored procedures
CREATE PROCEDURE PARTITION ON TABLE traffic_aggregates COLUMN traffic_class
  FROM CLASS com.ddos.detection.ProcessTrafficAggregates;
CREATE PROCEDURE PARTITION ON TABLE router_connection_stats
  COLUMN traffic_class PARAMETER 2
  FROM CLASS com.ddos.detection.ProcessRouterConnectionStats;

/*
 * VoltDB Stored Procedure for DDoS Traffic Classification Aggregates
 * Uses time windowing for 15-minute aggregations with ML inference results
 */
package com.ddos.detection;

import java.util.Map;
import java.util.HashMap;
import org.voltdb.*;
import org.voltdb.types.TimestampType;

public class ProcessTrafficAggregates extends VoltProcedure {

    private static final String TRAFFIC_COLUMNS =
        "(traffic_class, tickdate, timescale, flow_count, cumulative_flows) ";
    private static final String MINUTES_15 = "MINUTES_15";
    private static final String INSERT_TRAFFIC_START =
        "INSERT INTO traffic_aggregates " + TRAFFIC_COLUMNS + "VALUES (?,TIME_WINDOW(";
    private static final String INSERT_TRAFFIC_END = ",?,?,?);";
    private static final String MINUTES_15_INSERT = "MINUTE,15,?)";

    // Valid traffic classes from ML model
    private static final String[] VALID_TRAFFIC_CLASSES = {
        "DDOS", "DOS", "MIRAI", "BRUTEFORCE",
        "RECON", "SPOOFING", "WEB", "BENIGN"
    };

    // SQL statements for time window calculations
    public static final SQLStmt getRolling15Minutes = new SQLStmt(
        "SELECT TIME_WINDOW(MINUTE,15,?) TICKDATE FROM dummy WHERE x = 'X';"
    );

    public static final SQLStmt getWindowStart = new SQLStmt(
        "SELECT TIME_WINDOW(MINUTE,15,?) window_start FROM dummy WHERE x = 'X';"
    );

    // SQL statements for aggregates
    public static final SQLStmt get15MinuteAggregate = new SQLStmt(
        "SELECT * FROM traffic_aggregates WHERE traffic_class = ? " +
        "AND tickdate = TIME_WINDOW(MINUTE,15,?) AND timescale = '" + MINUTES_15 + "';"
    );

    public static final SQLStmt insertTraffic15Minutes = new SQLStmt(
        INSERT_TRAFFIC_START + MINUTES_15_INSERT + INSERT_TRAFFIC_END
    );

    public static final SQLStmt updateTrafficAggregate = new SQLStmt(
        "UPDATE traffic_aggregates SET flow_count = ?, cumulative_flows = ? " +
        "WHERE traffic_class = ? AND tickdate = ? AND timescale = ?;"
    );

    // State management
    public final SQLStmt SELECT_TRAFFIC_STATE = new SQLStmt(
        "SELECT traffic_class, window_start, cumulative_flows " +
        "FROM traffic_state WHERE traffic_class = ? " +
        "AND window_start = TIME_WINDOW(MINUTE,15,?);"
    );

    public final SQLStmt UPSERT_TRAFFIC_STATE = new SQLStmt(
        "UPSERT INTO traffic_state (traffic_class, window_start, " +
        "cumulative_flows, last_update) VALUES (?, TIME_WINDOW(MINUTE,15,?), ?, ?);"
    );

    private long getAndUpdateCumulativeFlows(String trafficClass,
            TimestampType tickdate, long newFlowCount, VoltTable stateResults) {
        long cumulativeFlows = 0;
        if (stateResults.getRowCount() > 0) {
            VoltTableRow row = stateResults.fetchRow(0);
            cumulativeFlows = row.getLong("cumulative_flows");
        }
        cumulativeFlows += newFlowCount;
        voltQueueSQL(UPSERT_TRAFFIC_STATE, trafficClass, tickdate,
                     cumulativeFlows, tickdate);
        return cumulativeFlows;
    }

    public VoltTable[] run(String aggregateName, Integer aggrCount,
            TimestampType tickdate, int performCleanupInt) throws VoltAbortException {

        Map<String, Integer> aggregates = Map.of(aggregateName, aggrCount);

        if (aggregates.isEmpty()) {
            throw new VoltAbortException("No valid aggregates provided");
        }

        // Get time window information
        voltQueueSQL(getRolling15Minutes, tickdate);
        voltQueueSQL(getWindowStart, tickdate);

        // Queue state queries for all traffic classes
        for (String trafficClass : aggregates.keySet()) {
            voltQueueSQL(SELECT_TRAFFIC_STATE, trafficClass, tickdate);
        }

        for (String trafficClass : aggregates.keySet()) {
            voltQueueSQL(get15MinuteAggregate, trafficClass, tickdate);
        }

        VoltTable[] initialResults = voltExecuteSQL();

        initialResults[0].advanceRow();
        final TimestampType windowTickdate =
            initialResults[0].getTimestampAsTimestamp("TICKDATE");

        // Process each traffic class
        int stateResultIndex = 2;
        int aggregateResultIndex = 2 + aggregates.size();

        for (Map.Entry<String, Integer> entry : aggregates.entrySet()) {
            String trafficClass = entry.getKey();
            Integer flowCount = entry.getValue();

            if (flowCount == null || flowCount <= 0) {
                stateResultIndex++;
                aggregateResultIndex++;
                continue;
            }

            VoltTable stateResult = initialResults[stateResultIndex++];
            long cumulativeFlows = getAndUpdateCumulativeFlows(
                trafficClass, tickdate, flowCount, stateResult);

            VoltTable existingAggregate = initialResults[aggregateResultIndex++];

            if (!existingAggregate.advanceRow()) {
                voltQueueSQL(insertTraffic15Minutes, trafficClass, tickdate,
                             MINUTES_15, flowCount, cumulativeFlows);
            } else {
                final TimestampType actualTickdate =
                    existingAggregate.getTimestampAsTimestamp("TICKDATE");
                final long existingFlowCount = existingAggregate.getLong("flow_count");
                voltQueueSQL(updateTrafficAggregate, existingFlowCount + flowCount,
                             cumulativeFlows, trafficClass, actualTickdate, MINUTES_15);
            }
        }

        return voltExecuteSQL(true);
    }
}

/*
 * VoltSP Java Pipeline for DDoS Detection with ML Classification
 * Uses tumbling windows for time-based aggregation (requires Java config)
 */
package com.ddos.detection;

import org.voltdb.stream.api.*;
import org.voltdb.stream.api.pipeline.*;
import org.voltdb.stream.api.pipeline.window.*;
import org.voltdb.stream.plugin.kafka.api.*;
import org.voltdb.stream.plugin.onnx.api.*;
import org.voltdb.stream.plugin.volt.api.*;
import org.voltdb.stream.plugin.window.api.tumbling.*;
import org.voltdb.types.TimestampType;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.time.Duration;

public class DDoSProtectionStream implements VoltPipeline {

    @Override
    public void define(VoltStreamBuilder stream) {
        VoltProcedureSinkConfigBuilder sinkBuilder =
            VoltProcedureSinkConfigBuilder.builder();
        VoltProcedureEmitterConfigBuilder emitterBuilder =
            VoltProcedureEmitterConfigBuilder.builder()
                .withComputeFunction((request, result) -> result);

        // Configure Kafka source with FlowEvent deserializer
        var dataStream = stream.withName("Process network flows")
            .configureResource("voltdb-database", VoltDBResourceConfigBuilder.class)
            .consumeFromSource(KafkaSourceConfigBuilder.<FlowEvent>builder()
                .withKeyDeserializer(StringDeserializer.class)
                .withValueDeserializer(FlowEventDeserializer.class)
                .withStartingOffset(KafkaStartingOffset.EARLIEST))
            .processWith(kafkaRequest -> kafkaRequest.getValue())

            // ONNX ML inference for traffic classification
            .processWith(OnnxProcessorConfigBuilder.<FlowEvent>builder()
                .withInputTensorName("float_input"))

            // Transform inference result to procedure call
            .processWith(inferenceResult -> {
                FlowEvent flowEvent = inferenceResult.request();
                int routerA = Integer.parseInt(flowEvent.getSourceRouter());
                int routerB = Integer.parseInt(flowEvent.getDestinationRouter());

                // Get ML classification from ONNX output
                String trafficClass = ((String[]) inferenceResult
                    .outputTensor().get("output_label"))[0];

                TimestampType timestamp = new TimestampType(
                    flowEvent.getEventTime().toEpochMilli() * 1000);
                long eventId = flowEvent.getEventId();

                // Serialize raw feature data as JSON string
                ObjectMapper mapper = new ObjectMapper();
                String dataRaw = mapper.writeValueAsString(
                    flowEvent.getDataRaw());

                return new VoltProcedureRequest("ProcessRouterConnectionStats",
                    new Object[]{routerA, routerB, trafficClass,
                                 timestamp, dataRaw, eventId});
            })
            .terminateWithEmitter(emitterBuilder);

        // Tumbling window aggregation (5-second windows by traffic class)
        var aggregator = dataStream
            .emit("emit-cached-data", VoltProcedureTrigger.onEachResponse)
            .processWith(table -> getFlowEvent(table))
            .aggregateWithWindow(
                TumblingTimeWindowConfiguratorBuilder.<FlowEvent>builder()
                    .withWindowSpan(Duration.ofSeconds(5))
                    .withEventTimeSupplier(FlowEvent::getEventTime)
                    .withKeyExtractor(FlowEvent::getClassification)
                    .withAggregateDefinition(builder -> {
                        builder.count()
                               .distinct("label", FlowEvent::getClassification);
                    })
            );

        // Emit aggregates at end of each window
        aggregator.emit("at-the-end", TumblingWindowTrigger.atEndOfWindow)
            .processWith((aggregate, consumer, ctx) -> {
                if (aggregate.count() == 0) return;

                String aggregateName = aggregate.getKey(String.class);
                int aggregateCount = aggregate.count();
                TimestampType emitTime = new TimestampType(
                    aggregate.emitTimestamp().toEpochMilli() * 1000);

                consumer.consume(new VoltProcedureRequest(
                    "ProcessTrafficAggregates",
                    new Object[]{aggregateName, aggregateCount, emitTime, 0}
                ));
            })
            .terminateWithSink(sinkBuilder);
    }
}