1import hashlib
2from typing import TYPE_CHECKING, ClassVar, Dict, List, Optional, Type, Union
3
4from ..config import Config
5from .irbasemapping import IRBaseMapping, normalize_service_name
6from .irbasemappinggroup import IRBaseMappingGroup
7from .irtcpmappinggroup import IRTCPMappingGroup
8
9if TYPE_CHECKING:
10 from .ir import IR # pragma: no cover
11
12
13class IRTCPMapping(IRBaseMapping):
14 binding: str
15 service: str
16 group_id: str
17 route_weight: List[Union[str, int]]
18
19 AllowedKeys: ClassVar[Dict[str, bool]] = {
20 "address": True,
21 "circuit_breakers": False,
22 "enable_ipv4": True,
23 "enable_ipv6": True,
24 "host": True,
25 "idle_timeout_ms": True,
26 "metadata_labels": True,
27 "port": True,
28 "service": True,
29 "tls": True,
30 "weight": True,
31 "resolver": True,
32 # Include the serialization, too.
33 "serialization": True,
34 }
35
36 def __init__(
37 self,
38 ir: "IR",
39 aconf: Config,
40 rkey: str, # REQUIRED
41 name: str, # REQUIRED
42 location: str, # REQUIRED
43 service: str, # REQUIRED
44 namespace: Optional[str] = None,
45 metadata_labels: Optional[Dict[str, str]] = None,
46 kind: str = "IRTCPMapping",
47 apiVersion: str = "getambassador.io/v3alpha1", # Not a typo! See below.
48 precedence: int = 0,
49 cluster_tag: Optional[str] = None,
50 **kwargs,
51 ) -> None:
52 # OK, this is a bit of a pain. We want to preserve the name and rkey and
53 # such here, unlike most kinds of IRResource. So. Shallow copy the keys
54 # we're going to allow from the incoming kwargs...
55
56 new_args = {x: kwargs[x] for x in kwargs.keys() if x in IRTCPMapping.AllowedKeys}
57
58 # XXX The resolver lookup code is duplicated from IRBaseMapping.setup --
59 # needs to be fixed after 1.6.1.
60 resolver_name = kwargs.get("resolver") or ir.ambassador_module.get(
61 "resolver", "kubernetes-service"
62 )
63
64 assert resolver_name # for mypy -- resolver_name cannot be None at this point
65 resolver = ir.get_resolver(resolver_name)
66
67 if resolver:
68 resolver_kind = resolver.kind
69 else:
70 # In IRBaseMapping.setup, we post an error if the resolver is unknown.
71 # Here, we just don't bother; we're only using it for service
72 # qualification.
73 resolver_kind = "KubernetesBogusResolver"
74
75 service = normalize_service_name(ir, service, namespace, resolver_kind, rkey=rkey)
76 ir.logger.debug(f"TCPMapping {name} service normalized to {repr(service)}")
77
78 # ...and then init the superclass.
79 super().__init__(
80 ir=ir,
81 aconf=aconf,
82 rkey=rkey,
83 location=location,
84 service=service,
85 kind=kind,
86 name=name,
87 namespace=namespace,
88 metadata_labels=metadata_labels,
89 apiVersion=apiVersion,
90 precedence=precedence,
91 cluster_tag=cluster_tag,
92 **new_args,
93 )
94
95 ir.logger.debug("IRTCPMapping %s: self.host = %s", name, self.get("host") or "i'*'")
96
97 @staticmethod
98 def group_class() -> Type[IRBaseMappingGroup]:
99 return IRTCPMappingGroup
100
101 def bind_to(self) -> str:
102 bind_addr = self.get("address") or "0.0.0.0"
103 return f"tcp-{bind_addr}-{self.port}"
104
105 def _group_id(self) -> str:
106 # Yes, we're using a cryptographic hash here. Cope. [ :) ]
107
108 h = hashlib.new("sha1")
109
110 # This is a TCP mapping.
111 h.update("TCP-".encode("utf-8"))
112
113 address = self.get("address") or "*"
114 h.update(address.encode("utf-8"))
115
116 port = str(self.port)
117 h.update(port.encode("utf-8"))
118
119 host = self.get("host") or "*"
120 h.update(host.encode("utf-8"))
121
122 return h.hexdigest()
123
124 def _route_weight(self) -> List[Union[str, int]]:
125 # These aren't order-dependent? or are they?
126 return [0]
View as plain text