diff --git a/consul/check.py b/consul/check.py index 55d3b2e..924f4c9 100644 --- a/consul/check.py +++ b/consul/check.py @@ -60,6 +60,35 @@ def tcp(cls, host: str, port: int, interval, timeout=None, deregister=None) -> d ret["DeregisterCriticalServiceAfter"] = deregister return ret + @classmethod + def grpc( + cls, url: str, interval, use_tls=None, tls_skip_verify=None, timeout=None, deregister=None + ) -> dict[str, Any]: + """ + Configures a gRPC health check that periodically assesses a service's health status. + + This check performs a gRPC health check against the specified *url* at each *interval* + (e.g., "10s"). An optional *timeout* sets the maximum duration for each check attempt. + The *deregister* parameter specifies the time after which a consistently failing service + is automatically deregistered from Consul. + + See also: consul grpc protocol + - https://github.com/grpc/grpc/blob/master/doc/health-checking.md + """ + ret = { + "grpc": url, + "interval": interval, + } + if use_tls is not None: + ret["grpc_use_tls"] = use_tls + if tls_skip_verify is not None: + ret["tls_skip_verify"] = tls_skip_verify + if timeout: + ret["timeout"] = timeout + if deregister: + ret["DeregisterCriticalServiceAfter"] = deregister + return ret + @classmethod def ttl(cls, ttl: str) -> dict[str, Any]: """ diff --git a/tests/test_base.py b/tests/test_base.py index c194886..8a76be6 100644 --- a/tests/test_base.py +++ b/tests/test_base.py @@ -386,6 +386,133 @@ def test_tcp_check(self, host: str, port: int, interval, timeout, deregister, wa ch = consul.check.Check.tcp(host, port, interval, timeout=timeout, deregister=deregister) assert ch == want + @pytest.mark.parametrize( + ("url", "interval", "use_tls", "tls_skip_verify", "timeout", "deregister", "want"), + [ + ( + "localhost:10055", + "10s", + None, + None, + None, + None, + { + "grpc": "localhost:10055", + "interval": "10s", + }, + ), + ( + "localhost:10055/:example.Service", + "10s", + None, + None, + None, + None, + { + "grpc": "localhost:10055/:example.Service", + "interval": "10s", + }, + ), + ( + "localhost:10055", + "10s", + True, + None, + None, + None, + { + "grpc": "localhost:10055", + "interval": "10s", + "grpc_use_tls": True, + }, + ), + ( + "localhost:10055", + "10s", + False, + None, + None, + None, + { + "grpc": "localhost:10055", + "interval": "10s", + "grpc_use_tls": False, + }, + ), + ( + "localhost:10055", + "10s", + None, + True, + None, + None, + { + "grpc": "localhost:10055", + "interval": "10s", + "tls_skip_verify": True, + }, + ), + ( + "localhost:10055", + "10s", + None, + False, + None, + None, + { + "grpc": "localhost:10055", + "interval": "10s", + "tls_skip_verify": False, + }, + ), + ( + "localhost:10055", + "10s", + None, + None, + "1s", + None, + { + "grpc": "localhost:10055", + "interval": "10s", + "timeout": "1s", + }, + ), + ( + "localhost:10055", + "10s", + None, + None, + None, + "1m", + { + "grpc": "localhost:10055", + "interval": "10s", + "DeregisterCriticalServiceAfter": "1m", + }, + ), + ( + "localhost:10055", + "10s", + True, + False, + "1s", + "1m", + { + "grpc": "localhost:10055", + "interval": "10s", + "grpc_use_tls": True, + "tls_skip_verify": False, + "timeout": "1s", + "DeregisterCriticalServiceAfter": "1m", + }, + ), + ], + ) + def test_grpc_check(self, url, interval, use_tls, tls_skip_verify, timeout, deregister, want) -> None: + ch = consul.check.Check.grpc(url, interval, use_tls, tls_skip_verify, timeout=timeout, deregister=deregister) + assert ch == want + @pytest.mark.parametrize( ("container_id", "shell", "script", "interval", "deregister", "want"), [