moonchen commented on code in PR #13737:
URL: https://github.com/apache/trafficserver/pull/13737#discussion_r4340555199
##########
src/iocore/net/ConnectionTracker.cc:
##########
@@ -410,21 +410,21 @@ ConnectionTracker::clear_client_exempt_list()
_global_config->client_exempt_list.clear();
}
+bool
+ConnectionTracker::is_client_exempt(IpEndpoint const &addr)
+{
+ if (_global_config == nullptr) {
+ return false;
+ }
+
+ ts::bravo::shared_lock<ts::bravo::shared_mutex>
lock(_global_config->client_exempt_list_mutex);
+ return _global_config->client_exempt_list.contains(swoc::IPAddr{addr});
Review Comment:
`clear_client_exempt_list()` takes the exclusive lock,
`std::lock_guard<ts::bravo::shared_mutex>` at line 409, as do the other writers
and the records callback. This PR only moves the existing shared-lock lookup
into `is_client_exempt()`.
##########
tests/uranium_tests/client_connection/test_connections_throttle_exempt.py:
##########
@@ -0,0 +1,316 @@
+"""Verify that proxy.config.http.per_client.connection.exempt_list exempts
client connections from
+proxy.config.net.connections_throttle."""
+# 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.
+
+from collections.abc import Iterable, Mapping
+import re
+import time
+from typing import Any
+
+import pytest
+
+from tools.uranium.services import ATS, ATSFactory, ServiceFactory,
wait_for_file_lines
+
+from .throttle_client import attempt, hold, request
+
+EXEMPT_LIST_VAR = "proxy.config.http.per_client.connection.exempt_list"
+OPEN = "proxy.process.net.connections_currently_open"
+EXEMPT_OPEN = "proxy.process.net.per_client.connections_exempt_currently_open"
+THROTTLED = "proxy.process.net.connections_throttled_in"
+EXEMPT = "proxy.process.net.per_client.connections_exempt_in"
+
+# An IPv4 and an IPv6 prefix longer than the address. If either were read as
/0, it would exempt the first address, the
+# loopback address of its family. The second address is the loopback address
of the other family.
+INVALID_PREFIXES = [
+ pytest.param("192.0.2.1/256", "127.0.0.1", "::1", id="ipv4"),
+ pytest.param("2001:db8::1/256", "::1", "127.0.0.1", id="ipv6"),
+]
+
+
+def invalid_entry_warning(entry: str) -> str:
+ """Return a regular expression for the warning that rejects the exempt
list entry @a entry."""
+
+ return re.escape(f"WARNING: {EXEMPT_LIST_VAR}: '{entry}' is not a valid IP
range")
+
+
+def create_ats(ats_factory: ATSFactory, name: str, **options: Any) -> ATS:
+ """Create a Traffic Server that serves the health check with the statichit
plugin, so that it needs no origin.
+
+ :param name: The process name.
+ :param options: The options of ATSFactory.create().
+ """
+
+ ats = ats_factory.create(name, **options)
+ if not ats.plugin_exists("statichit.so"):
+ pytest.skip("statichit.so is required")
+ ats.write_config_file("healthcheck.txt", "OK\n")
+ healthcheck = ats.config_directory / "healthcheck.txt"
+ ats.remap_config.add_line(f"map / http://127.0.0.1/ @plugin=statichit.so
@pparam=--file-path={healthcheck}")
+ return ats
+
+
+def metrics(ats: ATS, names: Iterable[str]) -> dict[str, int]:
+ """Return the values of the metrics @a names, read in one traffic_ctl
call."""
+
+ result = ats.traffic_ctl("metric", "get", *names)
+ assert result.returncode == 0, result.output
+ return {name: int(value) for name, value in (line.split() for line in
result.stdout.splitlines())}
+
+
+def wait_for_metrics(ats: ATS, expected: Mapping[str, int], timeout: float =
10) -> None:
+ """Wait until one read of the metrics shows each metric in @a expected at
its value.
+
+ :param timeout: How many seconds to wait.
+ """
+
+ deadline = time.monotonic() + timeout
+ values = metrics(ats, expected)
+ while values != expected and time.monotonic() < deadline:
+ time.sleep(0.1)
+ values = metrics(ats, expected)
+ assert values == expected, f"Expected the metrics {expected}, found
{values}"
+
+
+def stop_and_check_diags(ats: ATS) -> None:
+ """Stop @a ats and require that diags.log has no errors. Teardown checks
only for FATAL."""
+
+ ats.stop()
+ diags = ats.diags_log.read_text(errors="replace")
+ assert "ERROR:" not in diags, f"diags.log should contain no
errors:\n{diags}"
+
+
+class ThrottledTrafficServer:
+ """A Traffic Server whose connections_throttle is filled by connections
from one address."""
+
+ # An accept is refused once int(connections_currently_open * 1.1) >=
connections_throttle,
+ # so for a throttle below 11 the held connections fill it exactly.
+ THROTTLE = 5
+
+ def __init__(
+ self,
+ ats_factory: ATSFactory,
+ services: ServiceFactory,
+ exempt_list: str,
+ held: str,
+ other: str,
+ accept_threads: int = 1) -> None:
+ """Start Traffic Server and fill connections_throttle.
+
+ :param exempt_list: The value of
proxy.config.http.per_client.connection.exempt_list.
+ :param held: The address, 127.0.0.1 or ::1, of the connections that
fill the throttle.
+ :param other: The other address.
+ :param accept_threads: The value of proxy.config.accept_threads, which
selects the accept path.
+ """
+
+ self.held = held
+ self.other = other
+ # The expected counter values. None when a retry makes the count
unknown.
+ self._throttled = 0
+ self._exempt_count: int | None = 0
+ self.ats = self._configure(ats_factory, services, exempt_list,
accept_threads)
+ self.ats.start()
+ self._held = hold(held, self._port(held, "plain"), self.THROTTLE)
+ self._wait_for_counters()
+
+ def _configure(self, ats_factory: ATSFactory, services: ServiceFactory,
exempt_list: str, accept_threads: int) -> ATS:
+ """Configure Traffic Server with plain, TLS, allow-plain, and PROXY
protocol ports in both families."""
+
+ ats = create_ats(ats_factory, "ts", enable_tls=True)
+ self._ports = {
+ "plain": (ats.http_port, ats.ipv6_port),
+ "tls": (ats.https_port, ats.ipv6_https_port),
+ "allow-plain": (services.allocate_port(),
services.allocate_port()),
+ "proxy": (ats.proxy_protocol_port, services.allocate_port()),
+ }
+ flags = {"plain": "", "tls": ":ssl", "allow-plain":
":ssl:allow-plain", "proxy": ":pp"}
+ server_ports = " ".join(f"{ipv4}{flags[kind]}
{ipv6}{flags[kind]}:ipv6" for kind, (ipv4, ipv6) in self._ports.items())
+ ats.records.update(
+ {
+ "proxy.config.http.server_ports": server_ports,
+ "proxy.config.http.proxy_protocol_allowlist": "127.0.0.1,::1",
+ "proxy.config.diags.debug.enabled": 1,
+ "proxy.config.diags.debug.tags": "iocore_net_accept|statichit",
+ "proxy.config.accept_threads": accept_threads,
+ "proxy.config.net.connections_throttle": self.THROTTLE,
+ EXEMPT_LIST_VAR: exempt_list,
+ # Hold the incomplete requests for the whole test.
+ "proxy.config.http.accept_no_activity_timeout": 300,
+ "proxy.config.http.transaction_no_activity_timeout_in": 300,
+ })
+ return ats
+
+ def _port(self, host: str, kind: str) -> int:
+ """Return the port of @a kind ('plain', 'tls', 'allow-plain' or
'proxy') in the family of @a host."""
+
+ return self._ports[kind][1 if ":" in host else 0]
+
+ def _wait_for_counters(self) -> None:
+ """Wait for the counters to reach their expected values, with no
exempt connection open."""
+
+ expected = {OPEN: self.THROTTLE, EXEMPT_OPEN: 0, THROTTLED:
self._throttled}
+ if self._exempt_count is not None:
+ expected[EXEMPT] = self._exempt_count
+ wait_for_metrics(self.ats, expected)
+
+ def refused(self, host: str, kind: str = "plain", proxy_source: str = "",
retry: float = 0) -> None:
+ """Check that a connection from @a host is refused and counted in
connections_throttled_in.
+
+ :param proxy_source: Claim this source address in a PROXY protocol
header.
+ :param retry: Retry for this many seconds, for a reload to take effect.
+ """
+
+ self._throttled += 1
+ if retry:
+ # The attempts that are served before the reload takes effect are
counted as exempt.
+ self._exempt_count = None
+ request(host, self._port(host, kind), "refused", retry=retry, tls=kind
== "tls", proxy_source=proxy_source)
+ self._wait_for_counters()
+
+ def served(self, kind: str = "plain", proxy_source: str = "") -> None:
+ """Check that a connection from the other address is served, and
counted as an exempt open connection.
+
+ :param proxy_source: Claim this source address in a PROXY protocol
header.
+ """
+
+ assert self._exempt_count is not None, "A served check needs a known
exempt count."
+ self._exempt_count += 1
+ open_counts: dict[str, int] = {}
+
+ def read_open_counts() -> None:
+ open_counts.update(metrics(self.ats, (OPEN, EXEMPT_OPEN)))
+
+ outcome = attempt(
+ self.other, self._port(self.other, kind), tls=kind == "tls",
proxy_source=proxy_source, while_open=read_open_counts)
+ assert outcome == "served", f"A {kind} connection from {self.other}
was refused"
+ assert open_counts == {OPEN: self.THROTTLE + 1, EXEMPT_OPEN: 1}, "The
open exempt connection is counted as exempt."
Review Comment:
The accept path increments both gauges in `count_client_connection()` before
ATS reads the request, so they are already up when the response arrives, and
`traffic_ctl metric get` reads the live atomic value. Only the decrements on
close happen later, which is why `_wait_for_counters()` polls after the
connection closes.
##########
tests/uranium_tests/client_connection/test_connections_throttle_exempt.py:
##########
@@ -0,0 +1,316 @@
+"""Verify that proxy.config.http.per_client.connection.exempt_list exempts
client connections from
+proxy.config.net.connections_throttle."""
+# 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.
+
+from collections.abc import Iterable, Mapping
+import re
+import time
+from typing import Any
+
+import pytest
+
+from tools.uranium.services import ATS, ATSFactory, ServiceFactory,
wait_for_file_lines
+
+from .throttle_client import attempt, hold, request
+
+EXEMPT_LIST_VAR = "proxy.config.http.per_client.connection.exempt_list"
+OPEN = "proxy.process.net.connections_currently_open"
+EXEMPT_OPEN = "proxy.process.net.per_client.connections_exempt_currently_open"
+THROTTLED = "proxy.process.net.connections_throttled_in"
+EXEMPT = "proxy.process.net.per_client.connections_exempt_in"
+
+# An IPv4 and an IPv6 prefix longer than the address. If either were read as
/0, it would exempt the first address, the
+# loopback address of its family. The second address is the loopback address
of the other family.
+INVALID_PREFIXES = [
+ pytest.param("192.0.2.1/256", "127.0.0.1", "::1", id="ipv4"),
+ pytest.param("2001:db8::1/256", "::1", "127.0.0.1", id="ipv6"),
+]
+
+
+def invalid_entry_warning(entry: str) -> str:
+ """Return a regular expression for the warning that rejects the exempt
list entry @a entry."""
+
+ return re.escape(f"WARNING: {EXEMPT_LIST_VAR}: '{entry}' is not a valid IP
range")
+
+
+def create_ats(ats_factory: ATSFactory, name: str, **options: Any) -> ATS:
+ """Create a Traffic Server that serves the health check with the statichit
plugin, so that it needs no origin.
+
+ :param name: The process name.
+ :param options: The options of ATSFactory.create().
+ """
+
+ ats = ats_factory.create(name, **options)
+ if not ats.plugin_exists("statichit.so"):
+ pytest.skip("statichit.so is required")
+ ats.write_config_file("healthcheck.txt", "OK\n")
+ healthcheck = ats.config_directory / "healthcheck.txt"
+ ats.remap_config.add_line(f"map / http://127.0.0.1/ @plugin=statichit.so
@pparam=--file-path={healthcheck}")
+ return ats
+
+
+def metrics(ats: ATS, names: Iterable[str]) -> dict[str, int]:
+ """Return the values of the metrics @a names, read in one traffic_ctl
call."""
+
+ result = ats.traffic_ctl("metric", "get", *names)
+ assert result.returncode == 0, result.output
+ return {name: int(value) for name, value in (line.split() for line in
result.stdout.splitlines())}
+
+
+def wait_for_metrics(ats: ATS, expected: Mapping[str, int], timeout: float =
10) -> None:
+ """Wait until one read of the metrics shows each metric in @a expected at
its value.
+
+ :param timeout: How many seconds to wait.
+ """
+
+ deadline = time.monotonic() + timeout
+ values = metrics(ats, expected)
+ while values != expected and time.monotonic() < deadline:
+ time.sleep(0.1)
+ values = metrics(ats, expected)
+ assert values == expected, f"Expected the metrics {expected}, found
{values}"
+
+
+def stop_and_check_diags(ats: ATS) -> None:
+ """Stop @a ats and require that diags.log has no errors. Teardown checks
only for FATAL."""
+
+ ats.stop()
+ diags = ats.diags_log.read_text(errors="replace")
+ assert "ERROR:" not in diags, f"diags.log should contain no
errors:\n{diags}"
+
+
+class ThrottledTrafficServer:
+ """A Traffic Server whose connections_throttle is filled by connections
from one address."""
+
+ # An accept is refused once int(connections_currently_open * 1.1) >=
connections_throttle,
+ # so for a throttle below 11 the held connections fill it exactly.
+ THROTTLE = 5
+
+ def __init__(
+ self,
+ ats_factory: ATSFactory,
+ services: ServiceFactory,
+ exempt_list: str,
+ held: str,
+ other: str,
+ accept_threads: int = 1) -> None:
+ """Start Traffic Server and fill connections_throttle.
+
+ :param exempt_list: The value of
proxy.config.http.per_client.connection.exempt_list.
+ :param held: The address, 127.0.0.1 or ::1, of the connections that
fill the throttle.
+ :param other: The other address.
+ :param accept_threads: The value of proxy.config.accept_threads, which
selects the accept path.
+ """
+
+ self.held = held
+ self.other = other
+ # The expected counter values. None when a retry makes the count
unknown.
+ self._throttled = 0
+ self._exempt_count: int | None = 0
+ self.ats = self._configure(ats_factory, services, exempt_list,
accept_threads)
+ self.ats.start()
+ self._held = hold(held, self._port(held, "plain"), self.THROTTLE)
+ self._wait_for_counters()
+
+ def _configure(self, ats_factory: ATSFactory, services: ServiceFactory,
exempt_list: str, accept_threads: int) -> ATS:
+ """Configure Traffic Server with plain, TLS, allow-plain, and PROXY
protocol ports in both families."""
+
+ ats = create_ats(ats_factory, "ts", enable_tls=True)
+ self._ports = {
+ "plain": (ats.http_port, ats.ipv6_port),
+ "tls": (ats.https_port, ats.ipv6_https_port),
+ "allow-plain": (services.allocate_port(),
services.allocate_port()),
+ "proxy": (ats.proxy_protocol_port, services.allocate_port()),
+ }
+ flags = {"plain": "", "tls": ":ssl", "allow-plain":
":ssl:allow-plain", "proxy": ":pp"}
+ server_ports = " ".join(f"{ipv4}{flags[kind]}
{ipv6}{flags[kind]}:ipv6" for kind, (ipv4, ipv6) in self._ports.items())
+ ats.records.update(
+ {
+ "proxy.config.http.server_ports": server_ports,
+ "proxy.config.http.proxy_protocol_allowlist": "127.0.0.1,::1",
+ "proxy.config.diags.debug.enabled": 1,
+ "proxy.config.diags.debug.tags": "iocore_net_accept|statichit",
+ "proxy.config.accept_threads": accept_threads,
+ "proxy.config.net.connections_throttle": self.THROTTLE,
+ EXEMPT_LIST_VAR: exempt_list,
+ # Hold the incomplete requests for the whole test.
+ "proxy.config.http.accept_no_activity_timeout": 300,
+ "proxy.config.http.transaction_no_activity_timeout_in": 300,
+ })
+ return ats
+
+ def _port(self, host: str, kind: str) -> int:
+ """Return the port of @a kind ('plain', 'tls', 'allow-plain' or
'proxy') in the family of @a host."""
+
+ return self._ports[kind][1 if ":" in host else 0]
+
+ def _wait_for_counters(self) -> None:
+ """Wait for the counters to reach their expected values, with no
exempt connection open."""
+
+ expected = {OPEN: self.THROTTLE, EXEMPT_OPEN: 0, THROTTLED:
self._throttled}
+ if self._exempt_count is not None:
+ expected[EXEMPT] = self._exempt_count
+ wait_for_metrics(self.ats, expected)
+
+ def refused(self, host: str, kind: str = "plain", proxy_source: str = "",
retry: float = 0) -> None:
+ """Check that a connection from @a host is refused and counted in
connections_throttled_in.
+
+ :param proxy_source: Claim this source address in a PROXY protocol
header.
+ :param retry: Retry for this many seconds, for a reload to take effect.
+ """
+
+ self._throttled += 1
+ if retry:
+ # The attempts that are served before the reload takes effect are
counted as exempt.
+ self._exempt_count = None
+ request(host, self._port(host, kind), "refused", retry=retry, tls=kind
== "tls", proxy_source=proxy_source)
+ self._wait_for_counters()
+
+ def served(self, kind: str = "plain", proxy_source: str = "") -> None:
+ """Check that a connection from the other address is served, and
counted as an exempt open connection.
+
+ :param proxy_source: Claim this source address in a PROXY protocol
header.
+ """
+
+ assert self._exempt_count is not None, "A served check needs a known
exempt count."
+ self._exempt_count += 1
+ open_counts: dict[str, int] = {}
+
+ def read_open_counts() -> None:
+ open_counts.update(metrics(self.ats, (OPEN, EXEMPT_OPEN)))
+
+ outcome = attempt(
+ self.other, self._port(self.other, kind), tls=kind == "tls",
proxy_source=proxy_source, while_open=read_open_counts)
+ assert outcome == "served", f"A {kind} connection from {self.other}
was refused"
+ assert open_counts == {OPEN: self.THROTTLE + 1, EXEMPT_OPEN: 1}, "The
open exempt connection is counted as exempt."
+ self._wait_for_counters()
+
+ def reload(self, exempt_list: str, warning: str = "") -> None:
+ """Set the exempt list with traffic_ctl and reload.
+
+ :param warning: Wait for this regular expression in diags.log after
the reload.
+ """
+
+ for command in (("config", "set", EXEMPT_LIST_VAR, exempt_list),
("config", "reload")):
+ result = self.ats.traffic_ctl(*command)
+ assert result.returncode == 0, result.output
+ if warning:
+ wait_for_file_lines(self.ats.diags_log, warning, 1)
+
+ def finish(self) -> None:
+ """Close the held connections, stop Traffic Server, and check
diags.log."""
+
+ for sock in self._held:
+ sock.close()
+ stop_and_check_diags(self.ats)
+
+
+def test_memory_limit_applies(ats_factory: ATSFactory) -> None:
+ """Check that the memory limit still refuses a client in the list."""
+
+ ats = create_ats(ats_factory, "ts")
+ ats.records.update(
+ {
+ "proxy.config.http.server_ports": f"{ats.http_port}
{ats.ipv6_port}:ipv6",
+ EXEMPT_LIST_VAR: "::1",
+ # Less than any resident set size, so every new client connection
is refused once memory use is checked.
+ "proxy.config.memory.max_usage": 1,
+ })
+ ats.start()
+ # Traffic Server checks memory use every 10 seconds, and serves the health
check until then.
+ request("::1", ats.ipv6_port, "refused", retry=30)
+ wait_for_metrics(ats, {THROTTLED: 1})
+ stop_and_check_diags(ats)
+
+
+# accept_threads 1 accepts in a dedicated thread; 0 accepts on each ET_NET
thread.
[email protected]("accept_threads", [1, 0], ids=["accept_threads_1",
"accept_threads_0"])
+def test_exempt_ipv6(ats_factory: ATSFactory, services: ServiceFactory,
accept_threads: int) -> None:
+ """Fill the throttle from 127.0.0.1, and check ::1 over each kind of port,
and a reload of the list."""
+
+ # A network, a range, and an empty entry that match no test client, and
::1.
+ t = ThrottledTrafficServer(
+ ats_factory,
+ services,
+ "192.0.2.0/24, 198.51.100.1-198.51.100.9, ::1,",
+ held="127.0.0.1",
+ other="::1",
+ accept_threads=accept_threads)
+ # IPv4 is refused at the limit.
+ t.refused(t.held)
+ # ::1 is served over TCP, TLS, plain HTTP on an allow-plain TLS port, and
a PROXY protocol port that claims an
+ # address not in the list.
+ t.served()
+ t.served(kind="tls")
+ t.served(kind="allow-plain")
+ t.served(kind="proxy", proxy_source="2001:db8::1")
+ # IPv4 is refused on a PROXY protocol port that claims ::1.
+ t.refused(t.held, kind="proxy", proxy_source="::1")
Review Comment:
The throttle refuses this connection at accept, before ATS reads the PROXY
header, so the family of the header does not affect the outcome. The counter
check after it requires `connections_throttled_in` to rise by exactly one, so a
connection closed for a bad header would fail the test. Claiming `::1`, the
address in the list, is the most direct case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
