Implementing an API Pool with OpenResty and Redis for Dynamic Request Allocation, Queued Waiting, and Timeout Handling

To provide a complete implementation plan, we will explain in detail how to use OpenResty and Redis to build a system that can not only allocate requests dynamically according to QPS limits, but also place excess requests into a queue for waiting when the QPS limit is exceeded, with a timeout mechanism.

Complete Implementation Steps

1. Environment Preparation

Make sure you have installed the following components:

• OpenResty: an extended version of Nginx that supports Lua scripting.
• Redis: used to store API QPS limits, counters, and information about queued requests.
• lua-cjson: used for JSON serialization/deserialization.

Install OpenResty and Redis

# Install OpenResty
sudo apt-get update
sudo apt-get install -y software-properties-common
sudo add-apt-repository -y ppa:openresty/ppa
sudo apt-get update
sudo apt-get install -y openresty

# Install Redis
sudo apt-get install redis-server

# Install lua-cjson (if not already included with OpenResty)
sudo luarocks install lua-cjson

2. Configure OpenResty

Edit the OpenResty configuration file nginx.conf, usually located at /usr/local/openresty/nginx/conf/nginx.conf or /etc/openresty/nginx.conf.

http {
    lua_shared_dict api_limits 10m; # Used to store API QPS limits and counters
    lua_package_path "/path/to/lua/scripts/?.lua;;"; # Include custom Lua script paths

    upstream backend_apis {
        server api1.example.com;
        server api2.example.com;
        # Add more API instances
    }

    server {
        listen 80;

        location /api/ {
            access_by_lua_file /path/to/lua_scripts/api_limit.lua; # Handle QPS limiting and queuing logic
            
            proxy_pass http://backend_apis;
            proxy_next_upstream error timeout http_500 http_502 http_503 http_504;
        }
    }
}

3. Write the Lua Scripts

Create the directory /path/to/lua_scripts/ and create two Lua script files inside it: api_limit.lua and process_queue.lua.

api_limit.lua

local redis = require "resty.redis"
local cjson = require "cjson"

local red = redis:new()
red:set_timeout(1000) -- 1-second timeout

-- Connect to the Redis server
local ok, err = red:connect("127.0.0.1", 6379)
if not ok then
    ngx.log(ngx.ERR, "failed to connect to Redis: ", err)
    return ngx.exit(ngx.HTTP_SERVICE_UNAVAILABLE)
end

local api_key = ngx.var.uri
local limit = 100 -- Default QPS limit; adjust as needed
local current_count = tonumber(red:get(api_key)) or 0

if current_count < limit then
    red:incr(api_key)
    red:expire(api_key, 60) -- Reset the counter every 60 seconds
else
    local queue_name = "queue:" .. api_key
    local request_info = {
        uri = ngx.var.request_uri,
        timestamp = ngx.time() -- Current timestamp
    }
    local queued, err = red:rpush(queue_name, cjson.encode(request_info))
    if not queued then
        ngx.log(ngx.ERR, "failed to push to queue: ", err)
        return ngx.exit(ngx.HTTP_SERVICE_UNAVAILABLE)
    end
    return ngx.exit(ngx.HTTP_TOO_MANY_REQUESTS)
end

process_queue.lua

local redis = require "resty.redis"
local cjson = require "cjson"

local function process_queue()
    local red = redis:new()
    red:set_timeout(1000)

    local ok, err = red:connect("127.0.0.1", 6379)
    if not ok then
        ngx.log(ngx.ERR, "failed to connect to Redis: ", err)
        return
    end

    local queue_name = "queue:/api/path" -- Replace with your API path
    local timeout_seconds = 30 -- Set the timeout to 30 seconds

    while true do
        local request_json, err = red:lpop(queue_name)
        if not request_json then
            ngx.sleep(1) -- If there are no requests in the queue, sleep briefly
            goto continue
        end

        local request_info = cjson.decode(request_json)
        local request_time = request_info.timestamp
        local current_time = ngx.time()
        if (current_time - request_time) > timeout_seconds then
            ngx.log(ngx.WARN, "Request timed out and will be discarded: ", request_info.uri)
            goto continue
        end

        -- Send the request to the backend API
        local res = ngx.location.capture("/proxy_backend", { args = { uri = request_info.uri } })
        if res.status ~= ngx.HTTP_OK then
            ngx.log(ngx.ERR, "failed to process queued request: ", request_info.uri)
        end

        ::continue::
    end
end

process_queue()

4. Create a Background Task Script

Create a simple shell script, or use a cron job, to run the process_queue.lua script periodically.

Example Shell Script (run_process_queue.sh)

#!/bin/bash

/usr/local/openresty/bin/resty /path/to/lua_scripts/process_queue.lua

Grant execute permission:

chmod +x /path/to/run_process_queue.sh

Run Periodically with Cron

Edit the crontab to run the script once per minute:

crontab -e

Add the following line:

* * * * * /path/to/run_process_queue.sh

Testing and Verification

1. Start OpenResty:

1. sudo systemctl start openresty
2. Test API rate limiting: Use curl or another tool to send a large number of requests to /api/your_api_path, and observe whether rate limiting and queuing behave as expected.
3. Inspect Redis: Use the redis-cli command-line tool to inspect the data structures in Redis, ensuring that requests are correctly enqueued and that the timeout mechanism works properly.

Summary

Through the steps above, we implemented an API rate-limiting and queuing system based on OpenResty and Redis. The system can queue requests when the QPS limit is exceeded and process them asynchronously via a background task, while a reasonable timeout mechanism avoids long waits that would harm the user experience. Depending on specific needs, you can further optimize and extend this system.