Трансфер — поставка access-логов NGINX в таблицу

$remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent "$http_referer" "$http_user_agent"

::1 - - [01/Sep/2025:15:02:47 +0500] "GET /favicon.ico HTTP/1.1" 404 181 "-" "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 YaBrowser/25.6.0.0 Safari/537.36"
::1 - - [01/Sep/2025:15:02:51 +0500] "GET / HTTP/1.1" 200 409 "-" "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 YaBrowser/25.6.0.0 Safari/537.36"
::1 - - [01/Sep/2025:15:02:51 +0500] "GET /favicon.ico HTTP/1.1" 404 181 "http://localhost/" "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 YaBrowser/25.6.0.0 Safari/537.36"

CREATE TABLE `transfer_recipe/access_log` (
  partition Uint32 NOT NULL,
  offset Uint64 NOT NULL,
  line Uint64 NOT NULL,
  remote_addr String,
  remote_user String,
  time_local Timestamp,
  request_method String,
  request_path String,
  request_protocol String,
  status Uint32,
  body_bytes_sent Uint64,
  http_referer String,
  http_user_agent Utf8,
  PRIMARY KEY (partition, offset, line)
);

ALTER TABLE `transfer_recipe/access_log` SET (TTL = Interval("PT24H") ON time_local);

$transformation_lambda = ($msg) -> {
    -- Функция преобразования строки лога в строку таблицы
    $line_lambda = ($line) -> {
        -- Сначала разбиваем строку по символу " (двойные кавычки), чтобы выделить строки, которые могут содержать пробел.
        -- Сами строки символ " (двойные кавычки) содержать не могут — он будет заменён последовательностью символов \x22.
        $parts = String::SplitToList($line.1, '"');
        -- Каждую полученную часть, которая не соответствует экранированной строке, разбиваем по пробелу.
        $info_parts = String::SplitToList($parts[0], " ");
        $request_parts = String::SplitToList($parts[1], " ");
        $response_parts = String::SplitToList($parts[2], " ");
        -- Преобразуем дату в тип Datetime
        $dateParser = DateTime::Parse("%d/%b/%Y:%H:%M:%S");
        $date = $dateParser(SUBSTRING($info_parts[3], 1));

        -- Возвращаем структуру, каждое именованное поле которой соответствует столбцу таблицы.
        -- Важно: типы значений именованных полей должны соответствовать типам столбцов таблицы. Например, если столбец имеет тип Uint32,
        -- значение именованного поля должно быть типа Uint32. В противном случае требуется явное преобразование с помощью CAST.
        -- Значения колонок NOT NULL должны быть извлечены из опционального типа с помощью функции Unwrap.
        return <|
            partition: $msg._partition,
            offset: $msg._offset,
            line: $line.0,
            remote_addr: $info_parts[0],
            remote_user: $info_parts[2],
            time_local: DateTime::MakeTimestamp($date),
            request_method: $request_parts[0],
            request_path: $request_parts[1],
            request_protocol: $request_parts[2],
            status: CAST($response_parts[1] AS Uint32),
            body_bytes_sent: CAST($response_parts[2] AS Uint64),
            http_referer: $parts[3],
            http_user_agent: CAST(String::CgiUnescape($parts[5]) AS Utf8) -- Явно преобразуем в Utf8, так как столбец http_user_agent имеет тип Utf8, а не String
        |>;
    };


    $split = String::SplitToList($msg._data, "\n"); -- Если одно сообщение содержит несколько строк лога, разделяем его на отдельные строки
    $lines = ListFilter($split, ($line) -> { -- Фильтруем пустые строки, которые, например, могут появиться после последнего символа \n 
        return LENGTH($line) > 0;
    });

    -- Преобразуем каждую строку access лога в строку таблицы
    return ListMap(ListEnumerate($lines), $line_lambda);
};

CREATE TRANSFER `transfer_recipe/access_log_transfer`
  FROM `transfer_recipe/access_log_topic` TO `transfer_recipe/access_log`
  USING $transformation_lambda;

SELECT *
FROM `transfer_recipe/access_log`;