1# Copyright 2018 Datawire. All rights reserved.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License
14
15from abc import abstractmethod
16from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple
17
18from ..cache import Cache
19from ..utils import dump_json
20
21if TYPE_CHECKING:
22 from ..ir import IR, IRResource # pragma: no cover
23 from ..ir.irhttpmappinggroup import IRHTTPMappingGroup # pragma: no cover
24 from ..ir.irserviceresolver import ClustermapEntry # pragma: no cover
25
26
27def sanitize_pre_json(input):
28 # Removes all potential null values
29 if isinstance(input, dict):
30 for key, value in list(input.items()):
31 if value is None:
32 del input[key]
33 else:
34 sanitize_pre_json(value)
35 elif isinstance(input, list):
36 for item in input:
37 sanitize_pre_json(item)
38 return input
39
40
41class EnvoyConfig:
42 """
43 Base class for Envoy configuration that permits fetching configuration
44 for various elements to show in diagnostics.
45 """
46
47 ir: "IR"
48 elements: Dict[str, Dict[str, Any]]
49
50 def __init__(self, ir: "IR") -> None:
51 self.ir = ir
52 self.elements = {}
53
54 def add_element(self, kind: str, key: str, obj: Any) -> None:
55 eldict = self.elements.setdefault(kind, {})
56 eldict[key] = obj
57
58 def get_element(self, kind: str, key: str, default: Any) -> Optional[Any]:
59 eldict = self.elements.get(kind, {})
60 return eldict.get(key, default)
61
62 def pop_element(self, kind: str, key: str, default: Any) -> Optional[Any]:
63 eldict = self.elements.get(kind, {})
64 return eldict.pop(key, default)
65
66 def save_element(self, kind: str, resource: "IRResource", obj: Any):
67 self.add_element(kind, resource.rkey, obj)
68 self.add_element(kind, resource.location, obj)
69 return obj
70
71 @abstractmethod
72 def has_listeners(self) -> bool:
73 pass
74
75 @abstractmethod
76 def split_config(self) -> Tuple[Dict[str, Any], Dict[str, Any], Dict[str, "ClustermapEntry"]]:
77 pass
78
79 @abstractmethod
80 def as_dict(self) -> Dict[str, Any]:
81 pass
82
83 def as_json(self):
84 return dump_json(sanitize_pre_json(self.as_dict()), pretty=True)
85
86 @classmethod
87 def generate(cls, ir: "IR", cache: Optional[Cache] = None) -> "EnvoyConfig":
88 from . import V3Config
89
90 return V3Config(ir, cache=cache)
91
92
93class EnvoyRoute:
94 def __init__(self, group: "IRHTTPMappingGroup"):
95 self.prefix = "prefix"
96 self.path = "path"
97 self.regex = "regex"
98 self.envoy_route = self._get_envoy_route(group)
99
100 def _get_envoy_route(self, group: "IRHTTPMappingGroup") -> str:
101 if group.get("prefix_regex", False):
102 return self.regex
103 if group.get("prefix_exact", False):
104 return self.path
105 else:
106 return self.prefix
View as plain text