Обработка трафика на уровне Layer 4 и реализация сервера Memcached с помощью OpenResty

API7.ai

November 10, 2022

OpenResty (NGINX + Lua)

В нескольких предыдущих статьях мы представили некоторые Lua API для обработки запросов, которые связаны с уровнем 7. Кроме того, OpenResty предоставляет модуль stream-lua-nginx-module для обработки трафика на уровне 4. Он предоставляет инструкции и API, которые практически идентичны модулю lua-nginx-module.

Сегодня мы поговорим о реализации сервера Memcached с использованием OpenResty, для чего потребуется всего около 100 строк кода. В этом небольшом практическом задании мы используем многое из того, что изучили ранее, а также затронем некоторые темы из глав, посвященных тестированию и оптимизации производительности.

И мы должны понимать, что цель этой статьи не в том, чтобы разобраться в функциях каждой строки кода, а в том, чтобы понять полную картину того, как OpenResty разрабатывает проект с нуля, с точки зрения требований, тестирования, разработки и т.д.

Исходные требования и технические решения

Мы знаем, что HTTPS-трафик становится основным, но некоторые старые браузеры не поддерживают session tickets, поэтому нам нужно хранить идентификатор сессии на стороне сервера. Если локальное хранилище недостаточно, нам нужен кластер для хранения данных, и данные могут быть удалены, поэтому Memcached подходит больше.

На этом этапе внедрение Memcached должно быть самым простым решением. Однако в этой статье мы выберем использование OpenResty для создания собственного решения по следующим причинам.

  • Во-первых, прямое внедрение Memcached добавит дополнительный процесс, увеличивая затраты на развертывание и обслуживание.
  • Во-вторых, требования достаточно просты: нужны только операции get и set, а также поддержка срока действия.
  • В-третьих, OpenResty имеет модуль stream, который может быстро реализовать это требование.

Поскольку мы хотим реализовать сервер Memcached, нам сначала нужно понять его протокол. Протокол Memcached поддерживает TCP и UDP. Здесь мы используем TCP. Ниже приведен конкретный протокол для команд get и set.

Get
get value with key
Telnet command: get <key>*\r\n

Пример:
get key
VALUE key 0 4 data END

Set
Сохранить ключ-значение в memcached
Telnet command: set <key> <flags> <exptime> <bytes> [noreply]\r\n<value>\r\n

Пример:
set key 0 900 4 data
STORED

Нам также нужно знать, как реализована "обработка ошибок" в протоколе Memcached, помимо get и set. "Обработка ошибок" очень важна для серверных программ, и нам нужно писать программы, которые обрабатывают не только нормальные запросы, но и исключения. Например, в сценарии, подобном следующему:

  • Memcached отправляет запрос, отличный от get или set, как мне его обработать?
  • Какой ответ я должен дать клиенту Memcached, если на стороне сервера произошла ошибка?

Кроме того, мы хотим написать клиентское приложение, совместимое с Memcached. Таким образом, пользователям не нужно будет различать официальную версию Memcached и реализацию OpenResty.

Следующий рисунок из документации Memcached описывает, что должно возвращаться в случае ошибки и точный формат, который вы можете использовать в качестве справочного материала.

формат ошибки

Теперь давайте определим техническое решение. Мы знаем, что shared dict OpenResty может использоваться между workerами, и что размещение данных в shared dict очень похоже на размещение их в Memcached. Оба поддерживают операции get и set, и данные теряются при перезапуске процесса. Поэтому использование shared dict для эмуляции Memcached подходит, так как их принципы и поведение одинаковы.

Разработка через тестирование

Следующий шаг — начать работу. Однако, основываясь на идее разработки через тестирование, давайте создадим самый простой тестовый случай, прежде чем писать конкретный код. Вместо использования фреймворка test::nginx, который известен своей сложностью для начала, давайте начнем с ручного тестирования с использованием resty.

$ resty -e 'local memcached = require "resty.memcached"
    local memc, err = memcached:new()

    memc:set_timeout(1000) -- 1 секунда
    local ok, err = memc:connect("127.0.0.1", 11212)
    local ok, err = memc:set("dog", 32)
    if not ok then
        ngx.say("failed to set dog: ", err)
        return
    end

    local res, flags, err = memc:get("dog")
    ngx.say("dog: ", res)'

Этот тестовый код использует клиентскую библиотеку lua-rety-memcached для инициирования операций connect и set и предполагает, что сервер Memcached слушает порт 11212 на локальной машине.

Кажется, что он должен работать нормально. Вы можете запустить этот код на своей машине, и, что неудивительно, он вернет ошибку, например failed to set dog: closed, так как служба на этом этапе не запущена.

На этом этапе ваше техническое решение ясно: используйте модуль stream для приема и отправки данных и используйте shared dict для их хранения.

Метрика для измерения завершения требования ясна: запустите приведенный выше код и выведите фактическое значение dog.

Построение каркаса

Так чего же вы ждете? Начинайте писать код!

Моя привычка — сначала создать минимальный рабочий каркас кода, а затем постепенно заполнять его кодом. Преимущество этого подхода в том, что вы можете ставить множество небольших целей в процессе написания кода, и тестовые случаи будут давать вам положительную обратную связь, когда вы достигнете небольшой цели.

Давайте начнем с настройки конфигурационного файла NGINX, так как stream и shared dict должны быть предварительно настроены в нем. Вот конфигурационный файл, который я настроил.

stream {
    lua_shared_dict memcached 100m;
    lua_package_path 'lib/?.lua;;';
    server {
        listen 11212;
        content_by_lua_block {
            local m = require("resty.memcached.server")
            m.run()
        }
    }
}

Как видите, в этом конфигурационном файле содержится несколько ключевых элементов информации.

  • Во-первых, код выполняется в контексте stream NGINX, а не в контексте HTTP, и слушает порт 11212.
  • Во-вторых, имя shared dict — memcached, а размер — 100M, который нельзя изменить во время выполнения.
  • Кроме того, код находится в каталоге lib/resty/memcached, имя файла — server.lua, а входная функция — run(), что можно понять из lua_package_path и content_by_lua_block.

Далее пришло время построить каркас кода. Вы можете попробовать это сделать самостоятельно, а затем давайте посмотрим на мой каркасный код вместе.

local new_tab = require "table.new"
local str_sub = string.sub
local re_find = ngx.re.find
local mc_shdict = ngx.shared.memcached

local _M = { _VERSION = '0.01' }

local function parse_args(s, start)
end

function _M.get(tcpsock, keys)
end

function _M.set(tcpsock, res)
end

function _M.run()
    local tcpsock = assert(ngx.req.socket(true))

    while true do
        tcpsock:settimeout(60000) -- 60 секунд
        local data, err = tcpsock:receive("*l")

        local command, args
        if data then
            local from, to, err = re_find(data, [[(\S+)]], "jo")
            if from then
                command = str_sub(data, from, to)
                args = parse_args(data, to + 1)
            end
        end

        if args then
            local args_len = #args
            if command == 'get' and args_len > 0 then
                _M.get(tcpsock, args)
            elseif command == "set" and args_len == 4 then
                _M.set(tcpsock, args)
            end
        end
    end
end

return _M

Этот фрагмент кода реализует основную логику входной функции run(). Хотя я не сделал никакой обработки исключений, а зависимости parse_args, get и set — это пустые функции, этот каркас уже полностью выражает логику сервера Memcached.

Заполнение кода

Далее давайте реализуем эти пустые функции в порядке выполнения кода.

Сначала мы можем разобрать параметры команды Memcached в соответствии с документацией по протоколу Memcached.

local function parse_args(s, start)
    local arr = {}

    while true do
        local from, to = re_find(s, [[\S+]], "jo", {pos = start})
        if not from then
            break
        end

        table.insert(arr, str_sub(s, from, to))

        start = to + 1
    end

    return arr
end

Мой совет — сначала реализовать версию наиболее интуитивно, не думая о какой-либо оптимизации производительности. В конце концов, завершение всегда важнее совершенства, и инкрементальная оптимизация на основе завершения — это единственный способ приблизиться к совершенству.

Далее давайте реализуем функцию get. Она может запрашивать несколько ключей одновременно, поэтому я использую цикл for в следующем коде.

function _M.get(tcpsock, keys)
    local reply = ""

    for i = 1, #keys do
        local key = keys[i]
        local value, flags = mc_shdict:get(key)
        if value then
            local flags  = flags or 0
            reply = reply .. "VALUE" .. key .. " " .. flags .. " " .. #value .. "\r\n" .. value .. "\r\n"
        end
    end
    reply = reply ..  "END\r\n"

    tcpsock:settimeout(1000)  -- таймаут в одну секунду
    local bytes, err = tcpsock:send(reply)
end

Здесь только одна строка основного кода: local value, flags = mc_shdict:get(key), то есть запрос данных из shared dict; остальной код следует протоколу Memcached для формирования строки и, наконец, отправки ее клиенту.

Наконец, давайте посмотрим на функцию set. Она преобразует полученные параметры в формат API shared dict, сохраняет данные и в случае ошибок обрабатывает их в соответствии с протоколом Memcached.

function _M.set(tcpsock, res)
    local reply =  ""

    local key = res[1]
    local flags = res[2]
    local exptime = res[3]
    local bytes = res[4]

    local value, err = tcpsock:receive(tonumber(bytes) + 2)

    if str_sub(value, -2, -1) == "\r\n" then
        local succ, err, forcible = mc_shdict:set(key, str_sub(value, 1, bytes), exptime, flags)
        if succ then
            reply = reply .. “STORED\r\n"
        else
            reply = reply .. "SERVER_ERROR " .. err .. “\r\n”
        end
    else
        reply = reply .. "ERROR\r\n"
    end

    tcpsock:settimeout(1000)  -- таймаут в одну секунду
    local bytes, err = tcpsock:send(reply)
end

Кроме того, вы можете использовать тестовые случаи для проверки и отладки с помощью ngx.log при заполнении вышеуказанных функций. К сожалению, мы используем ngx.say и ngx.log для отладки, так как в OpenResty нет отладчика с точками останова, что все еще находится на этапе "подсечно-огневого земледелия", ожидая дальнейшего исследования.

Итог

Это практическое задание подходит к концу, и, наконец, я хотел бы оставить вопрос: Можете ли вы взять приведенный выше код реализации сервера Memcached, полностью запустить его и пройти тестовый случай?

Сегодняшний вопрос, вероятно, потребует много усилий, но это все еще примитивная версия. Здесь нет обработки ошибок, оптимизации производительности и автоматизированного тестирования, что будет улучшено позже.

Если у вас есть какие-либо сомнения относительно сегодняшнего объяснения или вашей практики, вы можете оставить комментарий и обсудить это с нами. Вы также можете поделиться этой статьей с вашими коллегами и друзьями, чтобы мы могли практиковаться и прогрессировать вместе.