1# Copyright 2020 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
15import logging
16from typing import Optional
17
18from typing_extensions import NotRequired, TypedDict
19
20from .cache import Cache
21from .config import Config
22from .envoy import EnvoyConfig
23from .fetch import ResourceFetcher
24from .ir import IR
25from .ir.ir import IRFileChecker
26from .utils import NullSecretHandler, SecretHandler
27
28
29class _CompileResult(TypedDict):
30 ir: IR
31 xds: NotRequired[EnvoyConfig]
32
33
34def Compile(
35 logger: logging.Logger,
36 input_text: str,
37 cache: Optional[Cache] = None,
38 file_checker: Optional[IRFileChecker] = None,
39 secret_handler: Optional[SecretHandler] = None,
40 k8s: bool = False,
41) -> _CompileResult:
42 """
43 Compile is a helper function to take a bunch of YAML and compile it into an
44 IR and, optionally, an Envoy config.
45
46 The output is a dictionary:
47
48 {
49 "ir": the IR data structure
50 "xds": the Envoy config
51 }
52
53 :param input_text: The input text (WATT snapshot JSON or K8s YAML per 'k8s')
54 :param k8s: If true, input_text is K8s YAML, otherwise it's WATT snapshot JSON
55 :param ir: Generate the IR IFF True
56 """
57
58 if not file_checker:
59 file_checker = lambda path: True
60
61 if not secret_handler:
62 secret_handler = NullSecretHandler(logger, None, None, "fake")
63
64 aconf = Config()
65
66 fetcher = ResourceFetcher(logger, aconf)
67
68 if k8s:
69 fetcher.parse_yaml(input_text, k8s=True)
70 else:
71 fetcher.parse_watt(input_text)
72
73 aconf.load_all(fetcher.sorted())
74
75 ir = IR(aconf, cache=cache, file_checker=file_checker, secret_handler=secret_handler)
76
77 out: _CompileResult = {"ir": ir}
78
79 if ir:
80 out["xds"] = EnvoyConfig.generate(ir, cache=cache)
81
82 return out
View as plain text