APISIX pipeline-request plugin demo

APISIX-plugin-pipeline-request.md

Apache APISIX's pipeline-request plugin demo

Logic

When handling client requests, the plugin will iterate through the nodes section of the configuration, requesting them in turn.

In the first request, it will send the complete method, query, header, and body of the client request, while in subsequent requests it will only send the last response with the POST method.

The successful response will be recorded in a temporary variable in last_resp, and each request in the loop will get the response body of the previous request from last_resp.body and overwrite the response body of the current request to last_resp.

When the loop ends, the response body in the last_resp will be sent to the client as the final response.

Usage

1. Install the plugin to APISIX and add it to the plugin list.

2. Create some routes for testing

First is a route for demonstrating response body changes, I simply set it to convert all response bodies to uppercase and it is only used for demonstration purposes.

curl -X PUT 'http://127.0.0.1:9180/apisix/admin/routes/1' \
--header 'X-API-KEY: edd1c9f034335f136f87ad84b625c8f1' \
--header 'Content-Type: application/json' \
--data-raw '{
    "uri": "/transform",
    "plugins": {
        "serverless-pre-function":{
            "phase": "access",
            "functions": ["return function() local core = require(\"apisix.core\"); local body = core.request.get_body(); core.response.set_header(\"X-Test\", \"text\"); core.response.exit(200, string.upper(body)); end"]
        }
    },
    "priority": 100
}'

The second one is the route that actually configures the pipeline-request plugin, which first gets a random uuid upstream from an httpbin, and later sends the response to the transform endpoint we just defined.

curl -XPUT 'http://127.0.0.1:9180/apisix/admin/routes/2' \
--header 'X-API-KEY: edd1c9f034335f136f87ad84b625c8f1' \
--header 'Content-Type: application/json' \
--data-raw '{
    "uri": "/test",
    "plugins": {
        "pipeline-request": {
            "nodes": [
                {
                    "uri": "https://httpbin.org/uuid",
                    "timeout": 2000
                },
                {
                    "uri": "http://127.0.0.1:9080/transform",
                    "timeout": 1000
                }
            ]
        }
    },
    "priority": 200
}'

3. let's try it

Request #0:

curl http://httpbin.org/uuid

Response #0:

{
  "uuid": "a0e8be56-2c2d-4b36-9755-86b100d3ee6b"
}

Request #1:

curl http://127.0.0.1:9080/test -i

Response #1: the uuid changes because it is random and we only need to focus on its characters

HTTP/1.1 200 OK
Content-Type: text/plain; charset=utf-8
Transfer-Encoding: chunked
Connection: keep-alive
Date: Fri, 16 Sep 2022 07:50:03 GMT
Server: APISIX/2.15.0
X-Test: text
Transfer-Encoding: chunked
Connection: keep-alive

{
  "UUID": "8886D6D7-0778-4750-9292-42C4198CE79A"
}

Look at it, it's been rewritten in capital letters.

pipeline-request.lua

--
-- Licensed to the Apache Software Foundation (ASF) under one or more
-- contributor license agreements.  See the NOTICE file distributed with
-- this work for additional information regarding copyright ownership.
-- The ASF licenses this file to You under the Apache License, Version 2.0
-- (the "License"); you may not use this file except in compliance with
-- the License.  You may obtain a copy of the License at
--
--     http://www.apache.org/licenses/LICENSE-2.0
--
-- Unless required by applicable law or agreed to in writing, software
-- distributed under the License is distributed on an "AS IS" BASIS,
-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-- See the License for the specific language governing permissions and
-- limitations under the License.
--
local core   = require("apisix.core")
local http   = require("resty.http")
local pairs       = pairs


local schema = {
    type = "object",
    properties = {
        nodes = {
            type = "array",
            items = {
                type = "object",
                properties = {
                    uri = {
                        type = "string",
                        minLength = 1
                    },
                    timeout = {
                        type = "integer",
                        minimum = 1,
                        maximum = 60000,
                        default = 3000,
                        description = "timeout in milliseconds",
                    },
                },
                required = {"uri"},
            }
        },
    },
}

local plugin_name = "pipeline-request"

local _M = {
    version = 0.1,
    priority = 1000,
    name = plugin_name,
    schema = schema,
}


function _M.check_schema(conf)
    local ok, err = core.schema.check(schema, conf)
    if not ok then
        return false, err
    end

    return true
end


function _M.access(conf, ctx)
    if #conf.nodes <= 0 then
        return 500, "empty nodes"
    end

    local last_resp, err
    for _, value in ipairs(conf.nodes) do
        local httpc = http.new()
        httpc:set_timeout(value.timeout)

        local params = {
            method = "POST",
            ssl_verify = false,
        }

        if last_resp ~= nil then
            -- setup body from last success response
            params.method = "POST"
            params.body = last_resp.body
        else
            -- setup header, query and body for first request (upstream)
            params.method = core.request.get_method()
            params.headers = core.request.headers()
            params.query = core.request.get_uri_args()
            local body, err = core.request.get_body()
            if err then
                return 503
            end
            if body then
                params.body = body
            end
        end

        -- send request to each node and temporary store response
        last_resp, err = httpc:request_uri(value.uri, params)
        if not last_resp then
            return 500, "request failed" .. err
        end
    end

    -- send all headers from last_resp to client
    for key, value in pairs(last_resp.headers) do
        core.response.set_header(key, value)
    end

    return 200, last_resp.body
end


return _M