| # |
| # 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. |
| # |
| use t::APISIX; |
| |
| my $nginx_binary = $ENV{'TEST_NGINX_BINARY'} || 'nginx'; |
| my $version = eval { `$nginx_binary -V 2>&1` }; |
| |
| if ($version !~ m/\/apisix-nginx-module/) { |
| plan(skip_all => "apisix-nginx-module not installed"); |
| } else { |
| plan('no_plan'); |
| } |
| |
| $ENV{TEST_NGINX_REDIS_PORT} ||= 1985; |
| |
| add_block_preprocessor(sub { |
| my ($block) = @_; |
| |
| if (!$block->extra_yaml_config) { |
| my $extra_yaml_config = <<_EOC_; |
| xrpc: |
| protocols: |
| - name: redis |
| _EOC_ |
| $block->set_value("extra_yaml_config", $extra_yaml_config); |
| } |
| |
| if ((!defined $block->error_log) && (!defined $block->no_error_log)) { |
| $block->set_value("no_error_log", "[error]\nRPC is not finished"); |
| } |
| |
| if (!defined $block->request) { |
| $block->set_value("request", "GET /t"); |
| } |
| |
| $block; |
| }); |
| |
| worker_connections(1024); |
| run_tests; |
| |
| __DATA__ |
| |
| === TEST 1: init |
| --- config |
| location /t { |
| content_by_lua_block { |
| local t = require("lib.test_admin").test |
| local code, body = t('/apisix/admin/stream_routes/1', |
| ngx.HTTP_PUT, |
| { |
| protocol = { |
| name = "redis" |
| }, |
| upstream = { |
| nodes = { |
| ["127.0.0.1:6379"] = 1 |
| }, |
| type = "roundrobin" |
| } |
| } |
| ) |
| if code >= 300 then |
| ngx.status = code |
| end |
| ngx.say(body) |
| } |
| } |
| --- response_body |
| passed |
| |
| |
| |
| === TEST 2: sanity |
| --- config |
| location /t { |
| content_by_lua_block { |
| local redis = require "resty.redis" |
| local red = redis:new() |
| |
| local ok, err = red:connect("127.0.0.1", $TEST_NGINX_REDIS_PORT) |
| if not ok then |
| ngx.say("failed to connect: ", err) |
| return |
| end |
| |
| local res, err = red:hmset("animals", "dog", "bark", "cat", "meow") |
| if not res then |
| ngx.say("failed to set animals: ", err) |
| return |
| end |
| ngx.say("hmset animals: ", res) |
| |
| local res, err = red:hmget("animals", "dog", "cat") |
| if not res then |
| ngx.say("failed to get animals: ", err) |
| return |
| end |
| |
| ngx.say("hmget animals: ", res) |
| |
| local res, err = red:hget("animals", "dog") |
| if not res then |
| ngx.say("failed to get animals: ", err) |
| return |
| end |
| |
| ngx.say("hget animals: ", res) |
| |
| local res, err = red:hget("animals", "not_found") |
| if not res then |
| ngx.say("failed to get animals: ", err) |
| return |
| end |
| |
| ngx.say("hget animals: ", res) |
| } |
| } |
| --- response_body |
| hmset animals: OK |
| hmget animals: barkmeow |
| hget animals: bark |
| hget animals: null |
| --- stream_conf_enable |
| |
| |
| |
| === TEST 3: error |
| --- config |
| location /t { |
| content_by_lua_block { |
| local redis = require "resty.redis" |
| local red = redis:new() |
| |
| local ok, err = red:connect("127.0.0.1", $TEST_NGINX_REDIS_PORT) |
| if not ok then |
| ngx.say("failed to connect: ", err) |
| return |
| end |
| |
| local res, err = red:get("animals") |
| if not res then |
| ngx.say("failed to set animals: ", err) |
| end |
| |
| local res, err = red:hget("animals", "dog") |
| if not res then |
| ngx.say("failed to get animals: ", err) |
| return |
| end |
| |
| ngx.say("hget animals: ", res) |
| } |
| } |
| --- response_body |
| failed to set animals: WRONGTYPE Operation against a key holding the wrong kind of value |
| hget animals: bark |
| --- stream_conf_enable |
| |
| |
| |
| === TEST 4: big value |
| --- config |
| location /t { |
| content_by_lua_block { |
| local redis = require "resty.redis" |
| local red = redis:new() |
| |
| local ok, err = red:connect("127.0.0.1", $TEST_NGINX_REDIS_PORT) |
| if not ok then |
| ngx.say("failed to connect: ", err) |
| return |
| end |
| |
| local res, err = red:set("big-key", ("\r\n"):rep(1024 * 1024 * 16)) |
| if not res then |
| ngx.say("failed to set: ", err) |
| return |
| end |
| |
| local res, err = red:get("big-key") |
| if not res then |
| ngx.say("failed to get: ", err) |
| return |
| end |
| |
| ngx.print(res) |
| } |
| } |
| --- response_body eval |
| "\r\n" x 16777216 |
| --- stream_conf_enable |
| |
| |
| |
| === TEST 5: pipeline |
| --- config |
| location /t { |
| content_by_lua_block { |
| local cjson = require("cjson") |
| local redis = require "resty.redis" |
| |
| local t = {} |
| for i = 1, 180 do |
| local th = assert(ngx.thread.spawn(function(i) |
| local red = redis:new() |
| local ok, err = red:connect("127.0.0.1", $TEST_NGINX_REDIS_PORT) |
| if not ok then |
| ngx.say("failed to connect: ", err) |
| return |
| end |
| |
| red:init_pipeline() |
| |
| red:set("mark_" .. i, i) |
| red:get("mark_" .. i) |
| red:get("counter") |
| for j = 1, 4 do |
| red:incr("counter") |
| end |
| |
| local results, err = red:commit_pipeline() |
| if not results then |
| ngx.say("failed to commit: ", err) |
| return |
| end |
| |
| local begin = tonumber(results[3]) |
| for j = 1, 4 do |
| local incred = results[3 + j] |
| if incred ~= results[2 + j] + 1 then |
| ngx.log(ngx.ERR, cjson.encode(results)) |
| end |
| end |
| end, i)) |
| table.insert(t, th) |
| end |
| for i, th in ipairs(t) do |
| ngx.thread.wait(th) |
| end |
| } |
| } |
| --- response_body |
| --- stream_conf_enable |
| |
| |
| |
| === TEST 6: delay |
| --- config |
| location /t { |
| content_by_lua_block { |
| local t = require("lib.test_admin").test |
| local code, body = t('/apisix/admin/stream_routes/1', |
| ngx.HTTP_PUT, |
| { |
| protocol = { |
| name = "redis", |
| conf = { |
| faults = { |
| {delay = 0.01, key = "ignored", commands = {"Ping", "time"}} |
| } |
| } |
| }, |
| upstream = { |
| nodes = { |
| ["127.0.0.1:6379"] = 1 |
| }, |
| type = "roundrobin" |
| } |
| } |
| ) |
| if code >= 300 then |
| ngx.status = code |
| end |
| ngx.say(body) |
| } |
| } |
| --- response_body |
| passed |
| |
| |
| |
| === TEST 7: hit |
| --- config |
| location /t { |
| content_by_lua_block { |
| local redis = require "resty.redis" |
| local red = redis:new() |
| |
| local ok, err = red:connect("127.0.0.1", $TEST_NGINX_REDIS_PORT) |
| if not ok then |
| ngx.say("failed to connect: ", err) |
| return |
| end |
| |
| local start = ngx.now() |
| local res, err = red:ping() |
| if not res then |
| ngx.say(err) |
| return |
| end |
| local now = ngx.now() |
| -- use integer to bypass float point number precision problem |
| if math.ceil((now - start) * 1000) < 10 then |
| ngx.say(now, " ", start) |
| return |
| end |
| start = now |
| |
| local res, err = red:time() |
| if not res then |
| ngx.say(err) |
| return |
| end |
| local now = ngx.now() |
| if math.ceil((now - start) * 1000) < 10 then |
| ngx.say(now, " ", start) |
| return |
| end |
| start = now |
| |
| red:init_pipeline() |
| red:time() |
| red:time() |
| red:get("A") |
| |
| local results, err = red:commit_pipeline() |
| if not results then |
| ngx.say("failed to commit: ", err) |
| return |
| end |
| local now = ngx.now() |
| if math.ceil((now - start) * 1000) < 20 or math.ceil((now - start) * 1000) > 30 then |
| ngx.say(now, " ", start) |
| return |
| end |
| |
| ngx.say("ok") |
| } |
| } |
| --- response_body |
| ok |
| --- stream_conf_enable |
| |
| |
| |
| === TEST 8: DFS match |
| --- config |
| location /t { |
| content_by_lua_block { |
| local t = require("lib.test_admin").test |
| local code, body = t('/apisix/admin/stream_routes/1', |
| ngx.HTTP_PUT, |
| { |
| protocol = { |
| name = "redis", |
| conf = { |
| faults = { |
| {delay = 0.02, key = "a", commands = {"get"}}, |
| {delay = 0.01, commands = {"get", "set"}}, |
| } |
| } |
| }, |
| upstream = { |
| nodes = { |
| ["127.0.0.1:6379"] = 1 |
| }, |
| type = "roundrobin" |
| } |
| } |
| ) |
| if code >= 300 then |
| ngx.status = code |
| end |
| ngx.say(body) |
| } |
| } |
| --- response_body |
| passed |
| |
| |
| |
| === TEST 9: hit |
| --- config |
| location /t { |
| content_by_lua_block { |
| local redis = require "resty.redis" |
| local red = redis:new() |
| |
| local ok, err = red:connect("127.0.0.1", $TEST_NGINX_REDIS_PORT) |
| if not ok then |
| ngx.say("failed to connect: ", err) |
| return |
| end |
| |
| local start = ngx.now() |
| local res, err = red:get("a") |
| if not res then |
| ngx.say(err) |
| return |
| end |
| local now = ngx.now() |
| if math.ceil((now - start) * 1000) < 20 then |
| ngx.say(now, " ", start) |
| return |
| end |
| start = now |
| |
| local res, err = red:set("a", "a") |
| if not res then |
| ngx.say(err) |
| return |
| end |
| local now = ngx.now() |
| if math.ceil((now - start) * 1000) < 10 then |
| ngx.say(now, " ", start) |
| return |
| end |
| start = now |
| |
| red:init_pipeline() |
| red:get("b") |
| red:set("A", "a") |
| |
| local results, err = red:commit_pipeline() |
| if not results then |
| ngx.say("failed to commit: ", err) |
| return |
| end |
| local now = ngx.now() |
| if math.ceil((now - start) * 1000) < 20 or math.ceil((now - start) * 1000) > 30 then |
| ngx.say(now, " ", start) |
| return |
| end |
| |
| ngx.say("ok") |
| } |
| } |
| --- response_body |
| ok |
| --- stream_conf_enable |
| |
| |
| |
| === TEST 10: multi keys |
| --- config |
| location /t { |
| content_by_lua_block { |
| local t = require("lib.test_admin").test |
| local code, body = t('/apisix/admin/stream_routes/1', |
| ngx.HTTP_PUT, |
| { |
| protocol = { |
| name = "redis", |
| conf = { |
| faults = { |
| {delay = 0.06, key = "b", commands = {"del"}}, |
| {delay = 0.04, key = "a", commands = {"mset"}}, |
| {delay = 0.02, key = "b", commands = {"mset"}}, |
| } |
| } |
| }, |
| upstream = { |
| nodes = { |
| ["127.0.0.1:6379"] = 1 |
| }, |
| type = "roundrobin" |
| } |
| } |
| ) |
| if code >= 300 then |
| ngx.status = code |
| end |
| ngx.say(body) |
| } |
| } |
| --- response_body |
| passed |
| |
| |
| |
| === TEST 11: hit |
| --- config |
| location /t { |
| content_by_lua_block { |
| local redis = require "resty.redis" |
| local red = redis:new() |
| |
| local ok, err = red:connect("127.0.0.1", $TEST_NGINX_REDIS_PORT) |
| if not ok then |
| ngx.say("failed to connect: ", err) |
| return |
| end |
| |
| local start = ngx.now() |
| local res, err = red:mset("c", 1, "a", 2, "b", 3) |
| if not res then |
| ngx.say(err) |
| return |
| end |
| local now = ngx.now() |
| if math.ceil((now - start) * 1000) < 40 then |
| ngx.say("mset a ", now, " ", start) |
| return |
| end |
| start = now |
| |
| local res, err = red:mset("b", 2, "a", 3) |
| if not res then |
| ngx.say(err) |
| return |
| end |
| local now = ngx.now() |
| if math.ceil((now - start) * 1000) < 20 or math.ceil((now - start) * 1000) > 35 then |
| ngx.say("mset b ", now, " ", start) |
| return |
| end |
| start = now |
| |
| local res, err = red:mset("c", "a") |
| if not res then |
| ngx.say(err) |
| return |
| end |
| local now = ngx.now() |
| if math.ceil((now - start) * 1000) > 20 then |
| ngx.say("mset mismatch ", now, " ", start) |
| return |
| end |
| start = now |
| |
| local res, err = red:del("a", "b") |
| if not res then |
| ngx.say(err) |
| return |
| end |
| local now = ngx.now() |
| if math.ceil((now - start) * 1000) < 60 then |
| ngx.say("del b ", now, " ", start) |
| return |
| end |
| start = now |
| |
| ngx.say("ok") |
| } |
| } |
| --- response_body |
| ok |
| --- stream_conf_enable |
| |
| |
| |
| === TEST 12: publish & subscribe |
| --- stream_extra_init_by_lua |
| local cjson = require "cjson" |
| local redis_proto = require("apisix.stream.xrpc.protocols.redis") |
| redis_proto.log = function(sess, ctx) |
| ngx.log(ngx.WARN, "log redis request ", cjson.encode(ctx.cmd_line)) |
| end |
| |
| --- config |
| location /t { |
| content_by_lua_block { |
| local cjson = require "cjson" |
| local redis = require "resty.redis" |
| |
| local red = redis:new() |
| local red2 = redis:new() |
| |
| red:set_timeout(1000) -- 1 sec |
| red2:set_timeout(1000) -- 1 sec |
| |
| local ok, err = red:connect("127.0.0.1", $TEST_NGINX_REDIS_PORT) |
| if not ok then |
| ngx.say("1: failed to connect: ", err) |
| return |
| end |
| |
| ok, err = red2:connect("127.0.0.1", $TEST_NGINX_REDIS_PORT) |
| if not ok then |
| ngx.say("2: failed to connect: ", err) |
| return |
| end |
| |
| local res, err = red:subscribe("dog") |
| if not res then |
| ngx.say("1: failed to subscribe: ", err) |
| return |
| end |
| |
| ngx.say("1: subscribe dog: ", cjson.encode(res)) |
| |
| res, err = red:subscribe("cat") |
| if not res then |
| ngx.say("1: failed to subscribe: ", err) |
| return |
| end |
| |
| ngx.say("1: subscribe cat: ", cjson.encode(res)) |
| |
| res, err = red2:publish("dog", "Hello") |
| if not res then |
| ngx.say("2: failed to publish: ", err) |
| return |
| end |
| |
| ngx.say("2: publish: ", cjson.encode(res)) |
| |
| res, err = red:read_reply() |
| if not res then |
| ngx.say("1: failed to read reply: ", err) |
| else |
| ngx.say("1: receive: ", cjson.encode(res)) |
| end |
| |
| red:set_timeout(10) -- 10ms |
| res, err = red:read_reply() |
| if not res then |
| ngx.say("1: failed to read reply: ", err) |
| else |
| ngx.say("1: receive: ", cjson.encode(res)) |
| end |
| red:set_timeout(1000) -- 1s |
| |
| res, err = red:unsubscribe() |
| if not res then |
| ngx.say("1: failed to unscribe: ", err) |
| else |
| ngx.say("1: unsubscribe: ", cjson.encode(res)) |
| end |
| |
| res, err = red:read_reply() |
| if not res then |
| ngx.say("1: failed to read reply: ", err) |
| else |
| ngx.say("1: receive: ", cjson.encode(res)) |
| end |
| |
| red:set_timeout(10) -- 10ms |
| res, err = red:read_reply() |
| if not res then |
| ngx.say("1: failed to read reply: ", err) |
| else |
| ngx.say("1: receive: ", cjson.encode(res)) |
| end |
| red:set_timeout(1000) -- 1s |
| |
| res, err = red:set("dog", 1) |
| if not res then |
| ngx.say("1: failed to set: ", err) |
| else |
| ngx.say("1: receive: ", cjson.encode(res)) |
| end |
| |
| red:close() |
| red2:close() |
| } |
| } |
| --- response_body_like chop |
| ^1: subscribe dog: \["subscribe","dog",1\] |
| 1: subscribe cat: \["subscribe","cat",2\] |
| 2: publish: 1 |
| 1: receive: \["message","dog","Hello"\] |
| 1: failed to read reply: timeout |
| 1: unsubscribe: \[\["unsubscribe","(?:dog|cat)",1\],\["unsubscribe","(?:dog|cat)",0\]\] |
| 1: failed to read reply: not subscribed |
| 1: failed to read reply: not subscribed |
| 1: receive: "OK" |
| $ |
| --- stream_conf_enable |
| --- grep_error_log eval |
| qr/log redis request \[[^]]+\]/ |
| --- grep_error_log_out |
| log redis request ["subscribe","dog"] |
| log redis request ["subscribe","cat"] |
| log redis request ["publish","dog","Hello"] |
| log redis request ["unsubscribe"] |
| log redis request ["set","dog","1"] |
| |
| |
| |
| === TEST 13: psubscribe & punsubscribe |
| --- stream_extra_init_by_lua |
| local cjson = require "cjson" |
| local redis_proto = require("apisix.stream.xrpc.protocols.redis") |
| redis_proto.log = function(sess, ctx) |
| ngx.log(ngx.WARN, "log redis request ", cjson.encode(ctx.cmd_line)) |
| end |
| |
| --- config |
| location /t { |
| content_by_lua_block { |
| local cjson = require "cjson" |
| local redis = require "resty.redis" |
| |
| local red = redis:new() |
| local red2 = redis:new() |
| |
| red:set_timeout(1000) -- 1 sec |
| red2:set_timeout(1000) -- 1 sec |
| |
| local ok, err = red:connect("127.0.0.1", $TEST_NGINX_REDIS_PORT) |
| if not ok then |
| ngx.say("1: failed to connect: ", err) |
| return |
| end |
| |
| ok, err = red2:connect("127.0.0.1", $TEST_NGINX_REDIS_PORT) |
| if not ok then |
| ngx.say("2: failed to connect: ", err) |
| return |
| end |
| |
| local res, err = red:psubscribe("dog*", "cat*") |
| if not res then |
| ngx.say("1: failed to subscribe: ", err) |
| return |
| end |
| |
| ngx.say("1: psubscribe: ", cjson.encode(res)) |
| |
| res, err = red2:publish("dog1", "Hello") |
| if not res then |
| ngx.say("2: failed to publish: ", err) |
| return |
| end |
| |
| ngx.say("2: publish: ", cjson.encode(res)) |
| |
| res, err = red:read_reply() |
| if not res then |
| ngx.say("1: failed to read reply: ", err) |
| else |
| ngx.say("1: receive: ", cjson.encode(res)) |
| end |
| |
| res, err = red:punsubscribe("cat*", "dog*") |
| if not res then |
| ngx.say("1: failed to unscribe: ", err) |
| else |
| ngx.say("1: punsubscribe: ", cjson.encode(res)) |
| end |
| |
| res, err = red:set("dog", 1) |
| if not res then |
| ngx.say("1: failed to set: ", err) |
| else |
| ngx.say("1: receive: ", cjson.encode(res)) |
| end |
| |
| red:close() |
| red2:close() |
| } |
| } |
| --- response_body_like chop |
| ^1: psubscribe: \[\["psubscribe","dog\*",1\],\["psubscribe","cat\*",2\]\] |
| 2: publish: 1 |
| 1: receive: \["pmessage","dog\*","dog1","Hello"\] |
| 1: punsubscribe: \[\["punsubscribe","cat\*",1\],\["punsubscribe","dog\*",0\]\] |
| 1: receive: "OK" |
| $ |
| --- stream_conf_enable |
| --- grep_error_log eval |
| qr/log redis request \[[^]]+\]/ |
| --- grep_error_log_out |
| log redis request ["psubscribe","dog*","cat*"] |
| log redis request ["publish","dog1","Hello"] |
| log redis request ["punsubscribe","cat*","dog*"] |
| log redis request ["set","dog","1"] |
