- on-client-session
- on-server-session
- on-frontend-tcp-request
+ - on-switching-rules
- on-backend-tcp-request
- on-tcp-response
- on-frontend-http-request
This event is only available for SPOE filters
declared in a frontend or a listen section.
+ * on-switching-rules is triggered after the evaluation of
+ "tcp-request content" rules on the frontend side,
+ just before the "use_backend" switching rules. It
+ allows an agent to influence the backend selection.
+ This event is only available for SPOE filters
+ declared in a frontend or a listen section. Note
+ that, in TCP mode, request data is only guaranteed
+ to be available if a "tcp-request content" rule
+ (combined with a "tcp-request inspect-delay") waits
+ for it; otherwise the event fires as soon as the
+ connection is established.
+
* on-backend-tcp-request is triggered just before the evaluation of
"tcp-request content" rules on the backend side.
This event is skipped for SPOE filters declared
--- /dev/null
+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# Minimal SPOA (Stream Processing Offload Agent) for the
+# on-switching-rules reg-test.
+#
+# It performs the SPOP HELLO/AGENT-HELLO handshake, then on every NOTIFY:
+# - decodes the message args (in particular "reqlen"),
+# - replies with an AGENT-ACK carrying a set-var action that sets
+# sess.<var-prefix>.backend_name = "be_spoe",
+# - appends a line to a marker file so the test can assert what the
+# agent actually received (proving the event fired *with* request data).
+#
+# This is intentionally a minimal, self-contained SPOP implementation so the
+# reg-test has no external dependencies beyond python3. SPOA reference
+# implementations are maintained out of the haproxy tree (they do not depend
+# on haproxy's version); see https://github.com/haproxy/spoa-example for a
+# full-featured agent.
+#
+# Usage: spoa.py <listen_port> <marker_file>
+import socket, struct, sys, threading
+
+def enc_varint(v):
+ if v < 240:
+ return bytes([v])
+ out = bytearray()
+ out.append(0xF0 | (v & 0x0F))
+ v = (v >> 4) - 16
+ while v >= 128:
+ out.append(0x80 | (v & 0x7F))
+ v = (v >> 7) - 1
+ out.append(v & 0x7F)
+ return bytes(out)
+
+def dec_varint(buf, off):
+ b = buf[off]; off += 1
+ if b < 240:
+ return b, off
+ val = b & 0x0F
+ shift = 4
+ while True:
+ b = buf[off]; off += 1
+ val |= (b & 0x7F) << shift
+ shift += 7
+ if b < 128:
+ break
+ return val, off
+
+def enc_str(s):
+ b = s.encode() if isinstance(s, str) else s
+ return enc_varint(len(b)) + b
+
+def read_frame(sock):
+ hdr = b''
+ while len(hdr) < 4:
+ c = sock.recv(4 - len(hdr))
+ if not c: return None
+ hdr += c
+ (length,) = struct.unpack(">I", hdr)
+ body = b''
+ while len(body) < length:
+ c = sock.recv(length - len(body))
+ if not c: return None
+ body += c
+ return body
+
+def send_frame(sock, body):
+ sock.sendall(struct.pack(">I", len(body)) + body)
+
+# frame types
+HAPROXY_HELLO = 1
+HAPROXY_DISCON = 2
+HAPROXY_NOTIFY = 3
+AGENT_HELLO = 101
+AGENT_ACK = 103
+
+# data types
+T_NULL=0; T_BOOL=1; T_INT32=2; T_UINT32=3; T_INT64=4; T_UINT64=5
+T_IPV4=6; T_IPV6=7; T_STR=8; T_BIN=9
+FL_TRUE=0x10
+
+SET_VAR = 1
+SCOPE_SESS = 1
+
+MARKER = None
+
+def parse_typed(buf, off):
+ t = buf[off]; off += 1
+ dt = t & 0x0F
+ if dt in (T_STR, T_BIN):
+ vlen, off = dec_varint(buf, off)
+ v = buf[off:off+vlen]; off += vlen; return v, off
+ if dt in (T_INT32, T_UINT32, T_INT64, T_UINT64):
+ v, off = dec_varint(buf, off); return v, off
+ if dt == T_IPV4:
+ v = buf[off:off+4]; off += 4; return v, off
+ if dt == T_IPV6:
+ v = buf[off:off+16]; off += 16; return v, off
+ if dt == T_BOOL:
+ return bool(t & FL_TRUE), off
+ return None, off
+
+def handle(sock, addr):
+ try:
+ while True:
+ body = read_frame(sock)
+ if body is None:
+ return
+ ftype = body[0]
+ off = 1
+ off += 4 # flags
+ sid, off = dec_varint(body, off)
+ fid, off = dec_varint(body, off)
+
+ if ftype == HAPROXY_HELLO:
+ # parse HELLO kv to echo back max-frame-size
+ mfs = 16380
+ o = off
+ while o < len(body):
+ klen, o = dec_varint(body, o)
+ k = body[o:o+klen].decode(); o += klen
+ v, o = parse_typed(body, o)
+ if k == "max-frame-size":
+ mfs = v
+ # flags=1 sets the SPOP FIN bit (this is a complete frame).
+ payload = bytes([AGENT_HELLO]) + struct.pack(">I", 1) + enc_varint(0) + enc_varint(0)
+ payload += enc_str("version") + bytes([T_STR]) + enc_str("2.0")
+ payload += enc_str("max-frame-size") + bytes([T_UINT32]) + enc_varint(mfs)
+ payload += enc_str("capabilities") + bytes([T_STR]) + enc_str("")
+ send_frame(sock, payload)
+
+ elif ftype == HAPROXY_NOTIFY:
+ moff = off
+ kv = {}
+ while moff < len(body):
+ nlen, moff = dec_varint(body, moff)
+ name = body[moff:moff+nlen].decode(); moff += nlen
+ nbargs = body[moff]; moff += 1
+ for _ in range(nbargs):
+ klen, moff = dec_varint(body, moff)
+ k = body[moff:moff+klen].decode(); moff += klen
+ v, moff = parse_typed(body, moff)
+ kv[k] = v
+ if MARKER:
+ with open(MARKER, "a") as f:
+ f.write("NOTIFY reqlen=%s\n" % kv.get("reqlen"))
+ f.flush()
+ # ACK: set-var sess.<prefix>.backend_name = "be_spoe"
+ # flags=1 sets the SPOP FIN bit (this is a complete frame).
+ payload = bytes([AGENT_ACK]) + struct.pack(">I", 1) + enc_varint(sid) + enc_varint(fid)
+ payload += bytes([SET_VAR, 3, SCOPE_SESS]) + enc_str("backend_name")
+ payload += bytes([T_STR]) + enc_str("be_spoe")
+ send_frame(sock, payload)
+
+ elif ftype == HAPROXY_DISCON:
+ return
+ except Exception:
+ pass
+ finally:
+ sock.close()
+
+def main():
+ port = int(sys.argv[1])
+ global MARKER
+ MARKER = sys.argv[2] if len(sys.argv) > 2 else None
+ srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ srv.bind(("127.0.0.1", port))
+ srv.listen(5)
+ # Signal readiness so the test harness can synchronize before starting
+ # HAProxy (mirrors the "READY" convention used by other reg-test fixtures).
+ print("READY", flush=True)
+ while True:
+ c, a = srv.accept()
+ threading.Thread(target=handle, args=(c, a), daemon=True).start()
+
+if __name__ == "__main__":
+ main()
--- /dev/null
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# This test checks the new "on-switching-rules" SPOE event end to
+# end. A real (minimal) SPOA agent is started as an external process. On the
+# frontend, the event fires after the "tcp-request content" rules and before
+# the "use_backend" switching rules; the agent replies with a set-var action
+# that populates sess.spoe.backend_name, which the "use_backend" rule then
+# uses to route the request to be_spoe instead of the default backend.
+#
+# Success therefore proves three things at once:
+# - the event is parsed/accepted in a frontend,
+# - it fires with request data available (the agent logs a non-zero reqlen),
+# - its result is applied *before* backend selection (traffic lands on
+# be_spoe, not be_default).
+
+varnishtest "SPOE: on-switching-rules drives backend selection"
+
+#REGTEST_TYPE=devel
+
+feature ignore_unknown_macro
+feature cmd "python3 --version"
+
+# Two distinguishable backends. be_spoe is the one the agent must select.
+server s_default {
+ rxreq
+ txresp -hdr "x-backend: default"
+} -start
+
+server s_spoe {
+ rxreq
+ txresp -hdr "x-backend: spoe"
+} -start
+
+# Reserve a random local port for the SPOA agent, then release the listener
+# so the Python agent can bind it (same trick as reg-tests/resolvers).
+server s_agent {} -start
+server s_agent -break
+
+process pagent "python3 ${testdir}/spoa_mock.py ${s_agent_port} ${tmpdir}/spoa.marker" -start
+process pagent -expect-text 0 0 "READY"
+
+shell {
+ cat > "${tmpdir}/spoe-engine.conf" << EOF
+[test-engine]
+spoe-agent test-agent
+ messages check-route
+ use-backend spoe-backend
+ option var-prefix spoe
+ timeout hello 1s
+ timeout idle 2s
+ timeout processing 1s
+
+spoe-message check-route
+ args sni=req.ssl_sni reqlen=req.len
+ event on-switching-rules
+EOF
+}
+
+haproxy h1 -conf {
+ global
+ .if feature(THREAD)
+ thread-groups 1
+ .endif
+
+ defaults
+ mode tcp
+ timeout connect "${HAPROXY_TEST_TIMEOUT-5s}"
+ timeout client "${HAPROXY_TEST_TIMEOUT-5s}"
+ timeout server "${HAPROXY_TEST_TIMEOUT-5s}"
+
+ frontend fe
+ bind "fd@${fe}"
+ tcp-request inspect-delay 1s
+ # Wait until the full request is buffered so req.len is non-zero when
+ # the event fires (proves the event sees request data).
+ acl req_rdy req.len gt 0
+ tcp-request content accept if req_rdy
+ filter spoe engine test-engine config "${tmpdir}/spoe-engine.conf"
+ use_backend be_spoe if { var(sess.spoe.backend_name) -m found }
+ default_backend be_default
+
+ backend be_default
+ server s1 ${s_default_addr}:${s_default_port}
+
+ backend be_spoe
+ server s1 ${s_spoe_addr}:${s_spoe_port}
+
+ backend spoe-backend
+ mode spop
+ timeout connect 1s
+ timeout server 2s
+ server agent 127.0.0.1:${s_agent_port}
+} -start
+
+# The agent sets sess.spoe.backend_name, so use_backend routes to be_spoe.
+client c1 -connect ${h1_fe_sock} {
+ txreq
+ rxresp
+ expect resp.status == 200
+ expect resp.http.x-backend == "spoe"
+} -run
+
+# The agent must have received a NOTIFY carrying request data (reqlen > 0),
+# which proves the event fired *after* content inspection buffered the body.
+shell "grep -Eq 'NOTIFY reqlen=[1-9][0-9]*' ${tmpdir}/spoa.marker"
/* Request events */
SPOE_EV_ON_CLIENT_SESS = 1,
SPOE_EV_ON_TCP_REQ_FE,
+ SPOE_EV_ON_SWITCHING_RULES, /* after tcp-request content, before use_backend */
SPOE_EV_ON_TCP_REQ_BE,
SPOE_EV_ON_HTTP_REQ_FE,
SPOE_EV_ON_HTTP_REQ_BE,
static const char *spoe_event_str[SPOE_EV_EVENTS] = {
[SPOE_EV_ON_CLIENT_SESS] = "on-client-session",
- [SPOE_EV_ON_TCP_REQ_FE] = "on-frontend-tcp-request",
- [SPOE_EV_ON_TCP_REQ_BE] = "on-backend-tcp-request",
+ [SPOE_EV_ON_TCP_REQ_FE] = "on-frontend-tcp-request",
+ [SPOE_EV_ON_SWITCHING_RULES] = "on-switching-rules",
+ [SPOE_EV_ON_TCP_REQ_BE] = "on-backend-tcp-request",
[SPOE_EV_ON_HTTP_REQ_FE] = "on-frontend-http-request",
[SPOE_EV_ON_HTTP_REQ_BE] = "on-backend-http-request",
if (!LIST_ISEMPTY(&ctx->events[SPOE_EV_ON_TCP_REQ_FE]))
filter->pre_analyzers |= AN_REQ_INSPECT_FE;
+ if (!LIST_ISEMPTY(&ctx->events[SPOE_EV_ON_SWITCHING_RULES]))
+ filter->pre_analyzers |= AN_REQ_SWITCHING_RULES;
+
if (!LIST_ISEMPTY(&ctx->events[SPOE_EV_ON_TCP_REQ_BE]))
filter->pre_analyzers |= AN_REQ_INSPECT_BE;
if (!(chn->flags & CF_ISRESP)) {
if (filter->pre_analyzers & AN_REQ_INSPECT_FE)
chn->analysers |= AN_REQ_INSPECT_FE;
+ if (filter->pre_analyzers & AN_REQ_SWITCHING_RULES)
+ chn->analysers |= AN_REQ_SWITCHING_RULES;
if (filter->pre_analyzers & AN_REQ_INSPECT_BE)
chn->analysers |= AN_REQ_INSPECT_BE;
case AN_REQ_INSPECT_FE:
ret = spoe_process_event(s, ctx, SPOE_EV_ON_TCP_REQ_FE);
break;
+ case AN_REQ_SWITCHING_RULES:
+ ret = spoe_process_event(s, ctx, SPOE_EV_ON_SWITCHING_RULES);
+ break;
case AN_REQ_INSPECT_BE:
ret = spoe_process_event(s, ctx, SPOE_EV_ON_TCP_REQ_BE);
break;
else if (strcmp(args[1], spoe_event_str[SPOE_EV_ON_TCP_REQ_FE]) == 0)
curmsg->event = SPOE_EV_ON_TCP_REQ_FE;
+ else if (strcmp(args[1], spoe_event_str[SPOE_EV_ON_SWITCHING_RULES]) == 0)
+ curmsg->event = SPOE_EV_ON_SWITCHING_RULES;
else if (strcmp(args[1], spoe_event_str[SPOE_EV_ON_TCP_REQ_BE]) == 0)
curmsg->event = SPOE_EV_ON_TCP_REQ_BE;
else if (strcmp(args[1], spoe_event_str[SPOE_EV_ON_TCP_RSP]) == 0)
}
if (!(px->cap & PR_CAP_FE) && (msg->event == SPOE_EV_ON_CLIENT_SESS ||
msg->event == SPOE_EV_ON_TCP_REQ_FE ||
+ msg->event == SPOE_EV_ON_SWITCHING_RULES ||
msg->event == SPOE_EV_ON_HTTP_REQ_FE)) {
ha_warning("Proxy '%s': frontend event used on a backend proxy at %s:%d.\n",
px->id, msg->conf.file, msg->conf.line);
where |= SMP_VAL_FE_REQ_CNT;
break;
+ case SPOE_EV_ON_SWITCHING_RULES:
+ where |= SMP_VAL_FE_SET_BCK;
+ break;
+
case SPOE_EV_ON_HTTP_REQ_FE:
where |= SMP_VAL_FE_HRQ_HDR;
break;
projects / haproxy.git / commitdiff