-- vim: ft=sql expandtab
-- This file is a source file for Ramen (https://github.com/rixed/ramen).
-- It contains the operations to load the CSV files produced by
-- dump-transactions.

PARAMETER csv_dir DEFAULTS TO "/tmp";

DEFINE web AS
  READ AND DELETE FILE "${param.csv_dir}/web-*.csv" SEPARATOR "\t" NULL "<NULL>" (
    start float,
    device u32,
    vlan u16?,
    clt_mac eth,
    clt_ip ip,
    srv_mac eth,
    srv_ip ip,
    srv_port u16,
    method string,
    status u16,
    resp_time float,
    host string,
    url string)
  FACTORS device, vlan, method AND status
  EVENT STARTING AT start WITH DURATION resp_time;

DEFINE dns AS
  READ AND DELETE FILE "${param.csv_dir}/dns-*.csv" SEPARATOR "\t" NULL "<NULL>" (
    start float,
    device u32,
    vlan u16?,
    clt_mac eth,
    clt_ip ip,
    srv_mac eth,
    srv_ip ip,
    srv_port u16,
    status u16,
    resp_time float,
    query string)
  FACTORS device, vlan AND status
  EVENT STARTING AT start WITH DURATION resp_time;

DEFINE traffic AS
  READ AND DELETE FILE "${param.csv_dir}/traffic-*.csv" SEPARATOR "\t" NULL "<NULL>" (
    start float,
    stop float,
    device u32,
    packets u32,
    vlan u16?,
    src_mac eth,
    dst_mac eth,
    eth_proto string,
    eth_payload u32,
    eth_mtu u32,
    src_ip ip?,
    dst_ip ip?,
    ip_proto string?,
    ip_payload u32?,
    src_port u16?,
    dst_port u16?)
  FACTORS device, vlan, eth_proto AND ip_proto
  EVENT STARTING AT start AND STOPPING AT stop;

-- list the TCP connection establishment only, not the full socket
-- (for this, see traffic and flow).
DEFINE tcp AS
  READ AND DELETE FILE "${param.csv_dir}/tcp-*.csv" SEPARATOR "\t" NULL "<NULL>" (
    start float,
    device u32,
    vlan u16?,
    clt_mac eth,
    clt_ip ip,
    clt_port u16,
    srv_mac eth,
    srv_ip ip,
    srv_port u16,
    num_syns u32,
    cnx_time float)
  FACTORS device AND vlan
  EVENT STARTING AT start WITH DURATION cnx_time;

DEFINE flow AS
  READ AND DELETE FILE "${param.csv_dir}/flow-*.csv" SEPARATOR "\t" NULL "<NULL>" (
    start float,
    stop float,
    device u32,
    vlan u16?,
    clt_mac eth,
    clt_ip ip,
    srv_mac eth,
    srv_ip ip,
    ip_proto string,
    src_port u16,
    dst_port u16,
    packets u32,
    payload u32)
  FACTORS device, vlan AND ip_proto
  EVENT STARTING AT start AND STOPPING AT stop;

DEFINE x509 AS
  READ AND DELETE FILE "${param.csv_dir}/x509-*.csv" SEPARATOR "\t" NULL "<NULL>" (
    start float,
    device u32,
    vlan u16?,
    clt_mac eth,
    clt_ip ip,
    srv_mac eth,
    srv_ip ip,
    srv_port u16,
    cert_depth u16,
    num_certs u16,
    serial_number string,
    not_before string,
    not_after string,
    subject string,
    issuer string)
  FACTORS device, vlan AND issuer
  EVENT STARTING AT start;

DEFINE tls AS
  READ AND DELETE FILE "${param.csv_dir}/tls-*.csv" SEPARATOR "\t" NULL "<NULL>" (
    start float,
    device u32,
    vlan u16?,
    clt_mac eth,
    clt_ip ip,
    srv_mac eth,
    srv_ip ip,
    srv_port u16,
    cipher_suite string,
    compress_algo string,
    server_name_indication string?)
  FACTORS device, vlan, cipher_suite AND compress_algo
  EVENT STARTING AT start;

-- Now let's do something useful with this info

-- Starting with some pre-aggregated minutely volumetry:

DEFINE labelled_traffic AS
  FROM traffic
  SELECT
    *,
    "dev:" || string(device) || COALESCE(",vlan:" || string(vlan), "") || ",mac:" || string(src_mac) || "," ||
      (IF eth_proto = "IPv4" OR eth_proto = "IPv6"
         THEN COALESCE(ip_proto, "unk") || "/ip:" || COALESCE(string(src_ip), "unk")
         ELSE string(eth_proto))
      AS _src_label,
    "dev:" || string(device) || COALESCE(",vlan:" || string(vlan), "") || ",mac:" || string(dst_mac) || "," ||
      (IF eth_proto = "IPv4" OR eth_proto = "IPv6"
         THEN COALESCE(ip_proto, "unk") || "/ip:" || COALESCE(string(dst_ip), "unk")
         ELSE string(eth_proto))
      AS _dst_label,
    MIN(_src_label, _dst_label) || " - " || MAX(_src_label, _dst_label) AS _label,
    -- Pretend the lowest numbered port is the service port:
    MIN(src_port, dst_port) AS _min_port,
    _label || COALESCE(", port:" || string(_min_port), "") AS _label_long,

    IF IS _label_long IN TOP 20 THEN _label_long ELSE
    IF IS _label IN TOP 20 THEN _label ELSE "other" AS label

  FACTORS device, vlan, eth_proto, ip_proto AND label
  EVENT STARTING AT start AND STOPPING AT stop;

DEFINE volumetry AS
  FROM labelled_traffic
  GROUP BY label, start // 60
  SELECT
    (start // 60) * 60 AS start,
    MIN start AS _min_start,
    MAX stop AS _max_stop,
    vlan, device, eth_proto, ip_proto, label,
    SUM packets / 60 AS avg_packets_per_secs,
    SUM eth_payload / 60 AS avg_payload_per_secs,
    MAX eth_mtu AS eth_mtu
  --COMMIT BEFORE in.start > out.start + 60 + 30
  COMMIT AFTER in.start > out.start + 60 + 30
  FACTORS vlan, device, eth_proto, ip_proto AND label
  EVENT STARTING AT start WITH DURATION 60 SECONDS;