1import re
2from typing import Generator, Tuple, Union
3
4import pytest
5
6from abstract_tests import HTTP, AmbassadorTest, Node, ServiceType
7from kat.harness import EDGE_STACK
8from kat.utils import ShellCommand
9
10
11class EnvoyLogTest(AmbassadorTest):
12 target: ServiceType
13 log_path: str
14
15 def init(self):
16 if EDGE_STACK:
17 self.xfail = "Not yet supported in Edge Stack"
18
19 self.target = HTTP()
20 self.log_path = "/tmp/ambassador/ambassador.log"
21 self.log_format = 'MY_REQUEST %RESPONSE_CODE% "%REQ(:AUTHORITY)%" "%REQ(USER-AGENT)%" "%REQ(X-REQUEST-ID)%" "%UPSTREAM_HOST%"'
22
23 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
24 yield self, self.format(
25 """
26---
27apiVersion: getambassador.io/v3alpha1
28kind: Module
29name: ambassador
30ambassador_id: [{self.ambassador_id}]
31config:
32 envoy_log_path: {self.log_path}
33 envoy_log_format: {self.log_format}
34"""
35 )
36
37 def check(self):
38 access_log_entry_regex = re.compile("^MY_REQUEST 200 .*")
39
40 cmd = ShellCommand("tools/bin/kubectl", "exec", self.path.k8s, "cat", self.log_path)
41 if not cmd.check("check envoy access log"):
42 pytest.exit("envoy access log does not exist")
43
44 for line in cmd.stdout.splitlines():
45 assert access_log_entry_regex.match(
46 line
47 ), f"{line} does not match {access_log_entry_regex}"
48
49
50class EnvoyLogJSONTest(AmbassadorTest):
51 target: ServiceType
52 log_path: str
53
54 def init(self):
55 self.target = HTTP()
56 self.log_path = "/tmp/ambassador/ambassador.log"
57
58 def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
59 yield self, self.format(
60 """
61---
62apiVersion: getambassador.io/v3alpha1
63kind: Module
64name: ambassador
65ambassador_id: [{self.ambassador_id}]
66config:
67 envoy_log_path: {self.log_path}
68 envoy_log_format:
69 protocol: "%PROTOCOL%"
70 duration: "%DURATION%"
71 envoy_log_type: json
72"""
73 )
74
75 def check(self):
76 access_log_entry_regex = re.compile('^({"duration":|{"protocol":)')
77
78 cmd = ShellCommand("tools/bin/kubectl", "exec", self.path.k8s, "cat", self.log_path)
79 if not cmd.check("check envoy access log"):
80 pytest.exit("envoy access log does not exist")
81
82 for line in cmd.stdout.splitlines():
83 assert access_log_entry_regex.match(
84 line
85 ), f"{line} does not match {access_log_entry_regex}"
View as plain text