1import logging
2from typing import Literal
3
4import pytest
5
6logging.basicConfig(
7 level=logging.INFO,
8 format="%(asctime)s test %(levelname)s: %(message)s",
9 datefmt="%Y-%m-%d %H:%M:%S",
10)
11
12logger = logging.getLogger("emissary-ingress")
13
14from ambassador import IR, Config, EnvoyConfig
15from ambassador.fetch import ResourceFetcher
16from ambassador.utils import NullSecretHandler
17from tests.utils import default_listener_manifests
18
19SERVICE_NAME = "cool-log-svcname"
20
21
22def _get_log_config(yaml, driver: Literal["http", "tcp"]):
23 for listener in yaml["static_resources"]["listeners"]:
24 if listener["name"].startswith("ambassador-listener-ready"):
25 # don't want to test the ready listener since it's different from the default 8080/8443
26 # listeners and is already tested in test_ready.py
27 continue
28 for filter_chain in listener["filter_chains"]:
29 for f in filter_chain["filters"]:
30 for log_filter in f["typed_config"]["access_log"]:
31 if log_filter["name"] == f"envoy.access_loggers.{driver}_grpc":
32 return log_filter
33 return False
34
35
36def _get_envoy_config(yaml):
37 aconf = Config()
38 fetcher = ResourceFetcher(logger, aconf)
39 fetcher.parse_yaml(default_listener_manifests() + yaml, k8s=True)
40
41 aconf.load_all(fetcher.sorted())
42
43 secret_handler = NullSecretHandler(logger, None, None, "0")
44
45 ir = IR(aconf, file_checker=lambda path: True, secret_handler=secret_handler)
46
47 assert ir
48 return EnvoyConfig.generate(ir)
49
50
51def _get_logfilter_http_default_conf():
52 return {
53 "@type": f"type.googleapis.com/envoy.extensions.access_loggers.grpc.v3.HttpGrpcAccessLogConfig",
54 "common_config": {
55 "transport_api_version": "V3",
56 "log_name": "logservice",
57 "grpc_service": {
58 "envoy_grpc": {"cluster_name": "cluster_logging_cool_log_svcname_default"}
59 },
60 "buffer_flush_interval": "1s",
61 "buffer_size_bytes": 16384,
62 },
63 "additional_request_headers_to_log": [],
64 "additional_response_headers_to_log": [],
65 "additional_response_trailers_to_log": [],
66 }
67
68
69def _get_logfilter_tcp_default_conf():
70 return {
71 "@type": f"type.googleapis.com/envoy.extensions.access_loggers.grpc.v3.TcpGrpcAccessLogConfig",
72 "common_config": {
73 "transport_api_version": "V3",
74 "log_name": "logservice",
75 "grpc_service": {
76 "envoy_grpc": {"cluster_name": "cluster_logging_cool_log_svcname_default"}
77 },
78 "buffer_flush_interval": "1s",
79 "buffer_size_bytes": 16384,
80 },
81 }
82
83
84###################### unit test covering http driver ###########################
85
86
87@pytest.mark.compilertest
88def test_irlogservice_http_defaults():
89 """tests defaults for log service when http driver is used and ensures that transport protocol is v3"""
90
91 yaml = (
92 """
93---
94apiVersion: getambassador.io/v3alpha1
95kind: LogService
96metadata:
97 name: myls
98 namespace: default
99spec:
100 service: """
101 + SERVICE_NAME
102 + """
103 driver: http
104 driver_config: {}
105 grpc: true
106"""
107 )
108
109 driver: Literal["http", "tcp"] = "http"
110
111 econf = _get_envoy_config(yaml)
112 conf = _get_log_config(econf.as_dict(), driver)
113
114 assert conf == False
115
116 errors = econf.ir.aconf.errors
117 assert "ir.logservice" in errors
118 assert (
119 errors["ir.logservice"][0]["error"]
120 == 'LogService: protocol_version v2 is unsupported, protocol_version must be "v3"'
121 )
122
123
124@pytest.mark.compilertest
125def test_irlogservice_http_default_overrides():
126 """tests default overrides for log service and ensures that transport protocol is v3"""
127
128 yaml = (
129 """
130---
131apiVersion: getambassador.io/v3alpha1
132kind: LogService
133metadata:
134 name: myls
135 namespace: default
136spec:
137 service: """
138 + SERVICE_NAME
139 + """
140 driver: http
141 grpc: true
142 protocol_version: "v3"
143 flush_interval_time: 33
144 flush_interval_byte_size: 9999
145 driver_config:
146 additional_log_headers:
147 - header_name: "x-dino-power"
148 - header_name: "x-dino-request-power"
149 during_request: true
150 during_response: false
151 during_trailer: false
152 - header_name: "x-dino-response-power"
153 during_request: false
154 during_response: true
155 during_trailer: false
156 - header_name: "x-dino-trailer-power"
157 during_request: false
158 during_response: false
159 during_trailer: true
160"""
161 )
162
163 driver: Literal["http", "tcp"] = "http"
164
165 econf = _get_envoy_config(yaml)
166 conf = _get_log_config(econf.as_dict(), driver)
167 assert conf
168
169 config = _get_logfilter_http_default_conf()
170 config["common_config"]["buffer_flush_interval"] = "33s"
171 config["common_config"]["buffer_size_bytes"] = 9999
172 config["additional_request_headers_to_log"] = ["x-dino-power", "x-dino-request-power"]
173 config["additional_response_headers_to_log"] = ["x-dino-power", "x-dino-response-power"]
174 config["additional_response_trailers_to_log"] = ["x-dino-power", "x-dino-trailer-power"]
175
176 assert conf.get("typed_config") == config
177
178 assert "ir.logservice" not in econf.ir.aconf.errors
179
180
181@pytest.mark.compilertest
182def test_irlogservice_http_v2():
183 """ensures that no longer supported v2 transport protocol is defaulted to v3"""
184
185 yaml = (
186 """
187---
188apiVersion: getambassador.io/v3alpha1
189kind: LogService
190metadata:
191 name: myls
192 namespace: default
193spec:
194 service: """
195 + SERVICE_NAME
196 + """
197 driver: http
198 driver_config: {}
199 grpc: true
200 protocol_version: "v2"
201"""
202 )
203
204 driver: Literal["http", "tcp"] = "http"
205
206 econf = _get_envoy_config(yaml)
207 conf = _get_log_config(econf.as_dict(), driver)
208
209 assert conf == False
210
211 errors = econf.ir.aconf.errors
212 assert "ir.logservice" in errors
213 assert (
214 errors["ir.logservice"][0]["error"]
215 == 'LogService: protocol_version v2 is unsupported, protocol_version must be "v3"'
216 )
217
218
219@pytest.mark.compilertest
220def test_irlogservice_http_v3():
221 """ensures that when transport protocol v3 is provided, nothing is logged"""
222
223 yaml = (
224 """
225---
226apiVersion: getambassador.io/v3alpha1
227kind: LogService
228metadata:
229 name: myls
230 namespace: default
231spec:
232 service: """
233 + SERVICE_NAME
234 + """
235 driver: http
236 driver_config: {}
237 grpc: true
238 protocol_version: "v3"
239"""
240 )
241
242 driver: Literal["http", "tcp"] = "http"
243
244 econf = _get_envoy_config(yaml)
245 conf = _get_log_config(econf.as_dict(), driver)
246
247 assert conf
248 assert conf.get("typed_config") == _get_logfilter_http_default_conf()
249
250 assert "ir.logservice" not in econf.ir.aconf.errors
251
252
253############### unit test covering tcp driver #######################
254@pytest.mark.compilertest
255def test_irlogservice_tcp_defaults():
256 """tests defaults for log service using tcp driver and ensures that transport protocol is v3"""
257
258 yaml = (
259 """
260---
261apiVersion: getambassador.io/v3alpha1
262kind: LogService
263metadata:
264 name: myls
265 namespace: default
266spec:
267 service: """
268 + SERVICE_NAME
269 + """
270 driver: tcp
271 driver_config: {}
272 grpc: true
273"""
274 )
275
276 driver: Literal["http", "tcp"] = "tcp"
277
278 econf = _get_envoy_config(yaml)
279 conf = _get_log_config(econf.as_dict(), driver)
280
281 assert conf == False
282
283 errors = econf.ir.aconf.errors
284 assert "ir.logservice" in errors
285 assert (
286 errors["ir.logservice"][0]["error"]
287 == 'LogService: protocol_version v2 is unsupported, protocol_version must be "v3"'
288 )
289
290
291@pytest.mark.compilertest
292def test_irlogservice_tcp_default_overrides():
293 """tests default overrides for log service with tcp driver and ensures that transport protocol is v3"""
294
295 yaml = (
296 """
297---
298apiVersion: getambassador.io/v3alpha1
299kind: LogService
300metadata:
301 name: myls
302 namespace: default
303spec:
304 service: """
305 + SERVICE_NAME
306 + """
307 driver: tcp
308 driver_config: {}
309 grpc: true
310 protocol_version: "v3"
311 flush_interval_time: 33
312 flush_interval_byte_size: 9999
313"""
314 )
315
316 driver: Literal["http", "tcp"] = "tcp"
317
318 econf = _get_envoy_config(yaml)
319 conf = _get_log_config(econf.as_dict(), driver)
320 assert conf
321
322 config = _get_logfilter_tcp_default_conf()
323 config["common_config"]["buffer_flush_interval"] = "33s"
324 config["common_config"]["buffer_size_bytes"] = 9999
325
326 assert conf.get("typed_config") == config
327
328 assert "ir.logservice" not in econf.ir.aconf.errors
329
330
331@pytest.mark.compilertest
332def test_irlogservice_tcp_v2():
333 """ensures that no longer supported v2 transport protocol is defaulted to v3"""
334
335 yaml = (
336 """
337---
338apiVersion: getambassador.io/v3alpha1
339kind: LogService
340metadata:
341 name: myls
342 namespace: default
343spec:
344 service: """
345 + SERVICE_NAME
346 + """
347 driver: tcp
348 driver_config: {}
349 grpc: true
350 protocol_version: "v2"
351"""
352 )
353
354 driver: Literal["http", "tcp"] = "tcp"
355
356 econf = _get_envoy_config(yaml)
357 conf = _get_log_config(econf.as_dict(), driver)
358
359 assert conf == False
360
361 errors = econf.ir.aconf.errors
362 assert "ir.logservice" in errors
363 assert (
364 errors["ir.logservice"][0]["error"]
365 == 'LogService: protocol_version v2 is unsupported, protocol_version must be "v3"'
366 )
367
368
369@pytest.mark.compilertest
370def test_irlogservice_tcp_v3():
371 """ensures that when transport protocol v3 is provided, nothing is logged"""
372
373 yaml = (
374 """
375---
376apiVersion: getambassador.io/v3alpha1
377kind: LogService
378metadata:
379 name: myls
380 namespace: default
381spec:
382 service: """
383 + SERVICE_NAME
384 + """
385 driver: tcp
386 driver_config: {}
387 grpc: true
388 protocol_version: "v3"
389"""
390 )
391
392 driver: Literal["http", "tcp"] = "tcp"
393
394 econf = _get_envoy_config(yaml)
395 conf = _get_log_config(econf.as_dict(), driver)
396
397 assert conf
398 assert conf.get("typed_config") == _get_logfilter_tcp_default_conf()
399
400 assert "ir.logservice" not in econf.ir.aconf.errors
View as plain text