1import logging
2import os
3
4import httpretty
5import pytest
6
7logging.basicConfig(
8 level=logging.INFO,
9 format="%(asctime)s test %(levelname)s: %(message)s",
10 datefmt="%Y-%m-%d %H:%M:%S",
11)
12
13logger = logging.getLogger("ambassador")
14
15from ambassador import IR, Config, EnvoyConfig
16from ambassador.fetch import ResourceFetcher
17from ambassador.utils import NullSecretHandler
18
19
20def _get_envoy_config(yaml, version="V3"):
21 aconf = Config()
22 fetcher = ResourceFetcher(logger, aconf)
23 fetcher.parse_yaml(yaml)
24
25 aconf.load_all(fetcher.sorted())
26
27 secret_handler = NullSecretHandler(logger, None, None, "0")
28
29 ir = IR(aconf, file_checker=lambda path: True, secret_handler=secret_handler)
30
31 assert ir
32 return EnvoyConfig.generate(ir, version)
33
34
35def del_env(env_key):
36 if env_key in os.environ:
37 del os.environ[env_key]
38
39
40def teardown_function(function):
41 del_env("STATSD_ENABLED")
42 del_env("DOGSTATSD")
43 del_env("STATSD_HOST")
44
45
46@pytest.mark.compilertest
47@httpretty.activate
48def test_statsd_default():
49 httpretty.register_uri(httpretty.GET, "statsd-sink", body='{"origin": "127.0.0.1"}')
50 yaml = """
51apiVersion: getambassador.io/v3alpha1
52kind: Mapping
53name: thing-rest
54case_sensitive: false
55hostname: "*"
56prefix: /reset/
57rewrite: /RESET/
58service: beepboop
59
60"""
61 expected_stats_sinks = {
62 "name": "envoy.stats_sinks.statsd",
63 "typed_config": {
64 "@type": "type.googleapis.com/envoy.config.metrics.v3.StatsdSink",
65 "address": {
66 "socket_address": {"protocol": "UDP", "address": "127.0.0.1", "port_value": 8125}
67 },
68 },
69 }
70
71 os.environ["STATSD_ENABLED"] = "true"
72 econf = _get_envoy_config(yaml, version="V3")
73
74 assert econf
75
76 econf_dict = econf.as_dict()
77
78 assert "stats_sinks" in econf_dict["bootstrap"]
79 assert len(econf_dict["bootstrap"]["stats_sinks"]) > 0
80 assert econf_dict["bootstrap"]["stats_sinks"][0] == expected_stats_sinks
81 assert "stats_flush_interval" in econf_dict["bootstrap"]
82 assert econf_dict["bootstrap"]["stats_flush_interval"]["seconds"] == "1"
83
84
85@pytest.mark.compilertest
86@httpretty.activate
87def test_statsd_other():
88 httpretty.register_uri(httpretty.GET, "other-statsd-sink", body='{"origin": "127.0.0.1"}')
89 yaml = """
90apiVersion: getambassador.io/v3alpha1
91kind: Mapping
92name: thing-rest
93case_sensitive: false
94hostname: "*"
95prefix: /reset/
96rewrite: /RESET/
97service: beepboop
98
99"""
100 expected_stats_sinks = {
101 "name": "envoy.stats_sinks.statsd",
102 "typed_config": {
103 "@type": "type.googleapis.com/envoy.config.metrics.v3.StatsdSink",
104 "address": {
105 "socket_address": {"protocol": "UDP", "address": "127.0.0.1", "port_value": 8125}
106 },
107 },
108 }
109
110 os.environ["STATSD_ENABLED"] = "true"
111 os.environ["STATSD_HOST"] = "other-statsd-sink"
112 econf = _get_envoy_config(yaml, version="V3")
113
114 assert econf
115
116 econf_dict = econf.as_dict()
117
118 assert "stats_sinks" in econf_dict["bootstrap"]
119 assert len(econf_dict["bootstrap"]["stats_sinks"]) > 0
120 assert econf_dict["bootstrap"]["stats_sinks"][0] == expected_stats_sinks
121 assert "stats_flush_interval" in econf_dict["bootstrap"]
122 assert econf_dict["bootstrap"]["stats_flush_interval"]["seconds"] == "1"
123
124
125@pytest.mark.compilertest
126@httpretty.activate
127def test_dogstatsd():
128 httpretty.register_uri(httpretty.GET, "statsd-sink", body='{"origin": "127.0.0.1"}')
129 yaml = """
130apiVersion: getambassador.io/v3alpha1
131kind: Mapping
132name: thing-rest
133case_sensitive: false
134hostname: "*"
135prefix: /reset/
136rewrite: /RESET/
137service: beepboop
138
139"""
140 expected_stats_sinks = {
141 "name": "envoy.stat_sinks.dog_statsd",
142 "typed_config": {
143 "@type": "type.googleapis.com/envoy.config.metrics.v3.DogStatsdSink",
144 "address": {
145 "socket_address": {"protocol": "UDP", "address": "127.0.0.1", "port_value": 8125}
146 },
147 },
148 }
149
150 os.environ["STATSD_ENABLED"] = "true"
151 os.environ["DOGSTATSD"] = "true"
152 econf = _get_envoy_config(yaml, version="V3")
153
154 assert econf
155
156 econf_dict = econf.as_dict()
157
158 assert "stats_sinks" in econf_dict["bootstrap"]
159 assert len(econf_dict["bootstrap"]["stats_sinks"]) > 0
160 assert econf_dict["bootstrap"]["stats_sinks"][0] == expected_stats_sinks
161 assert "stats_flush_interval" in econf_dict["bootstrap"]
162 assert econf_dict["bootstrap"]["stats_flush_interval"]["seconds"] == "1"
View as plain text