|
| 1 | +use Test::Nginx::Socket::Lua; |
| 2 | + |
| 3 | +log_level('info'); |
| 4 | +no_long_string(); |
| 5 | +repeat_each(1); |
| 6 | +plan 'no_plan'; |
| 7 | + |
| 8 | +# etcd coalesces watch responses into a single body read whenever they are |
| 9 | +# written close enough together (very common once progress_notify is on), and a |
| 10 | +# real etcd cannot be driven into that state reliably -- so each test below uses |
| 11 | +# a fake endpoint that emits a fixed, hand-crafted coalesced read. The point of |
| 12 | +# the fix under test is that the response holding the events is NOT necessarily |
| 13 | +# the last one in the read, so we exercise both orderings and an interleaving. |
| 14 | + |
| 15 | +# [event, progress]: event (rev 7) followed by a progress notification (rev 9). |
| 16 | +our $HttpConfig = <<'_EOC_'; |
| 17 | + lua_socket_log_errors off; |
| 18 | + lua_package_path 'lib/?.lua;/usr/local/share/lua/5.3/?.lua;/usr/share/lua/5.1/?.lua;;'; |
| 19 | + server { |
| 20 | + listen 1985; |
| 21 | + location /v3/watch { |
| 22 | + content_by_lua_block { |
| 23 | + ngx.print('{"result":{"header":{"revision":"7"},"events":' .. |
| 24 | + '[{"type":"PUT","kv":{"key":"L3Rlc3Q=","value":"ImFiYyI=",' .. |
| 25 | + '"mod_revision":"7"}}]}}\n' .. |
| 26 | + '{"result":{"header":{"revision":"9"}}}\n') |
| 27 | + } |
| 28 | + } |
| 29 | + } |
| 30 | +_EOC_ |
| 31 | +
|
| 32 | +# [progress, event]: progress notification (rev 7) followed by an event (rev 9). |
| 33 | +our $HttpConfigProgressFirst = <<'_EOC_'; |
| 34 | + lua_socket_log_errors off; |
| 35 | + lua_package_path 'lib/?.lua;/usr/local/share/lua/5.3/?.lua;/usr/share/lua/5.1/?.lua;;'; |
| 36 | + server { |
| 37 | + listen 1985; |
| 38 | + location /v3/watch { |
| 39 | + content_by_lua_block { |
| 40 | + ngx.print('{"result":{"header":{"revision":"7"}}}\n' .. |
| 41 | + '{"result":{"header":{"revision":"9"},"events":' .. |
| 42 | + '[{"type":"PUT","kv":{"key":"L3Rlc3Q=","value":"ImFiYyI=",' .. |
| 43 | + '"mod_revision":"9"}}]}}\n') |
| 44 | + } |
| 45 | + } |
| 46 | + } |
| 47 | +_EOC_ |
| 48 | +
|
| 49 | +# [event, progress, event]: an event (rev 7), a progress notification (rev 8), |
| 50 | +# and a second event (rev 9) all in one read. |
| 51 | +our $HttpConfigInterleaved = <<'_EOC_'; |
| 52 | + lua_socket_log_errors off; |
| 53 | + lua_package_path 'lib/?.lua;/usr/local/share/lua/5.3/?.lua;/usr/share/lua/5.1/?.lua;;'; |
| 54 | + server { |
| 55 | + listen 1985; |
| 56 | + location /v3/watch { |
| 57 | + content_by_lua_block { |
| 58 | + ngx.print('{"result":{"header":{"revision":"7"},"events":' .. |
| 59 | + '[{"type":"PUT","kv":{"key":"L3Rlc3Q=","value":"ImFiYyI=",' .. |
| 60 | + '"mod_revision":"7"}}]}}\n' .. |
| 61 | + '{"result":{"header":{"revision":"8"}}}\n' .. |
| 62 | + '{"result":{"header":{"revision":"9"},"events":' .. |
| 63 | + '[{"type":"PUT","kv":{"key":"L3Rlc3Qy","value":"ImJjZCI=",' .. |
| 64 | + '"mod_revision":"9"}}]}}\n') |
| 65 | + } |
| 66 | + } |
| 67 | + } |
| 68 | +_EOC_ |
| 69 | +
|
| 70 | +run_tests(); |
| 71 | +
|
| 72 | +__DATA__ |
| 73 | +
|
| 74 | +=== TEST 1: a coalesced read keeps the events of every response it contains |
| 75 | +--- http_config eval: $::HttpConfig |
| 76 | +--- config |
| 77 | + location /t { |
| 78 | + content_by_lua_block { |
| 79 | + local etcd, err = require "resty.etcd" .new({ |
| 80 | + protocol = "v3", |
| 81 | + http_host = "http://127.0.0.1:1985", |
| 82 | + }) |
| 83 | + if not etcd then |
| 84 | + ngx.say("failed to new: ", err) |
| 85 | + return |
| 86 | + end |
| 87 | +
|
| 88 | + local res_func, err = etcd:watchdir("/test", {timeout = 5}) |
| 89 | + if not res_func then |
| 90 | + ngx.say("failed to watchdir: ", err) |
| 91 | + return |
| 92 | + end |
| 93 | +
|
| 94 | + local res, err = res_func() |
| 95 | + if not res then |
| 96 | + ngx.say("failed to read watch: ", err) |
| 97 | + return |
| 98 | + end |
| 99 | +
|
| 100 | + -- The read is [event(rev 7), progress(rev 9)]. We report the |
| 101 | + -- progress notification's revision (9, the last response in the |
| 102 | + -- read) while still delivering the rev-7 event: a progress |
| 103 | + -- notification guarantees nothing happened between the event and |
| 104 | + -- it, so the caller can safely advance start_revision past 9. |
| 105 | + ngx.say("revision: ", res.result.header.revision) |
| 106 | + local events = res.result.events |
| 107 | + ngx.say("events: ", events and #events or 0) |
| 108 | + if events and events[1] then |
| 109 | + ngx.say("key: ", events[1].kv.key) |
| 110 | + ngx.say("value: ", events[1].kv.value) |
| 111 | + end |
| 112 | + } |
| 113 | + } |
| 114 | +--- request |
| 115 | +GET /t |
| 116 | +--- response_body |
| 117 | +revision: 9 |
| 118 | +events: 1 |
| 119 | +key: /test |
| 120 | +value: abc |
| 121 | +--- no_error_log |
| 122 | +[error] |
| 123 | +
|
| 124 | +
|
| 125 | +
|
| 126 | +=== TEST 2: a coalesced read keeps the event when it is the last response |
| 127 | +--- http_config eval: $::HttpConfigProgressFirst |
| 128 | +--- config |
| 129 | + location /t { |
| 130 | + content_by_lua_block { |
| 131 | + local etcd, err = require "resty.etcd" .new({ |
| 132 | + protocol = "v3", |
| 133 | + http_host = "http://127.0.0.1:1985", |
| 134 | + }) |
| 135 | + if not etcd then |
| 136 | + ngx.say("failed to new: ", err) |
| 137 | + return |
| 138 | + end |
| 139 | +
|
| 140 | + local res_func, err = etcd:watchdir("/test", {timeout = 5}) |
| 141 | + if not res_func then |
| 142 | + ngx.say("failed to watchdir: ", err) |
| 143 | + return |
| 144 | + end |
| 145 | +
|
| 146 | + local res, err = res_func() |
| 147 | + if not res then |
| 148 | + ngx.say("failed to read watch: ", err) |
| 149 | + return |
| 150 | + end |
| 151 | +
|
| 152 | + -- The read is [progress(rev 7), event(rev 9)]. The event is the |
| 153 | + -- last response, so its revision (9) is reported and its event |
| 154 | + -- delivered. |
| 155 | + ngx.say("revision: ", res.result.header.revision) |
| 156 | + local events = res.result.events |
| 157 | + ngx.say("events: ", events and #events or 0) |
| 158 | + if events and events[1] then |
| 159 | + ngx.say("key: ", events[1].kv.key) |
| 160 | + ngx.say("value: ", events[1].kv.value) |
| 161 | + end |
| 162 | + } |
| 163 | + } |
| 164 | +--- request |
| 165 | +GET /t |
| 166 | +--- response_body |
| 167 | +revision: 9 |
| 168 | +events: 1 |
| 169 | +key: /test |
| 170 | +value: abc |
| 171 | +--- no_error_log |
| 172 | +[error] |
| 173 | +
|
| 174 | +
|
| 175 | +
|
| 176 | +=== TEST 3: a coalesced read keeps events from every response when interleaved |
| 177 | +--- http_config eval: $::HttpConfigInterleaved |
| 178 | +--- config |
| 179 | + location /t { |
| 180 | + content_by_lua_block { |
| 181 | + local etcd, err = require "resty.etcd" .new({ |
| 182 | + protocol = "v3", |
| 183 | + http_host = "http://127.0.0.1:1985", |
| 184 | + }) |
| 185 | + if not etcd then |
| 186 | + ngx.say("failed to new: ", err) |
| 187 | + return |
| 188 | + end |
| 189 | +
|
| 190 | + local res_func, err = etcd:watchdir("/test", {timeout = 5}) |
| 191 | + if not res_func then |
| 192 | + ngx.say("failed to watchdir: ", err) |
| 193 | + return |
| 194 | + end |
| 195 | +
|
| 196 | + local res, err = res_func() |
| 197 | + if not res then |
| 198 | + ngx.say("failed to read watch: ", err) |
| 199 | + return |
| 200 | + end |
| 201 | +
|
| 202 | + -- The read is [event(rev 7), progress(rev 8), event(rev 9)]. Both |
| 203 | + -- events must survive, in order, under the highest revision (9). |
| 204 | + ngx.say("revision: ", res.result.header.revision) |
| 205 | + local events = res.result.events |
| 206 | + ngx.say("events: ", events and #events or 0) |
| 207 | + if events then |
| 208 | + for i = 1, #events do |
| 209 | + ngx.say("key", i, ": ", events[i].kv.key, |
| 210 | + " value", i, ": ", events[i].kv.value) |
| 211 | + end |
| 212 | + end |
| 213 | + } |
| 214 | + } |
| 215 | +--- request |
| 216 | +GET /t |
| 217 | +--- response_body |
| 218 | +revision: 9 |
| 219 | +events: 2 |
| 220 | +key1: /test value1: abc |
| 221 | +key2: /test2 value2: bcd |
| 222 | +--- no_error_log |
| 223 | +[error] |
0 commit comments