xref: /aosp_15_r20/external/angle/third_party/logdog/logdog/bootstrap.py (revision 8975f5c5ed3d1c378011245431ada316dfb6f244) 
1# Copyright 2016 The LUCI Authors. All rights reserved.
2# Use of this source code is governed under the Apache License, Version 2.0
3# that can be found in the LICENSE file.
4
5import collections
6import os
7
8from . import stream, streamname
9
10
11class NotBootstrappedError(RuntimeError):
12    """Raised when the current environment is missing Butler bootstrap variables.
13  """
14
15
16_ButlerBootstrapBase = collections.namedtuple(
17    '_ButlerBootstrapBase',
18    ('project', 'prefix', 'streamserver_uri', 'coordinator_host', 'namespace'))
19
20
21class ButlerBootstrap(_ButlerBootstrapBase):
22    """Loads LogDog Butler bootstrap parameters from the environment.
23
24  LogDog Butler adds variables describing the LogDog stream parameters to the
25  environment when it bootstraps an application. This class probes the
26  environment and identifies those parameters.
27  """
28
29    # TODO(iannucci): move all of these to LUCI_CONTEXT
30    _ENV_PROJECT = 'LOGDOG_STREAM_PROJECT'
31    _ENV_PREFIX = 'LOGDOG_STREAM_PREFIX'
32    _ENV_STREAM_SERVER_PATH = 'LOGDOG_STREAM_SERVER_PATH'
33    _ENV_COORDINATOR_HOST = 'LOGDOG_COORDINATOR_HOST'
34    _ENV_NAMESPACE = 'LOGDOG_NAMESPACE'
35
36    @classmethod
37    def probe(cls, env=None):
38        """Returns (ButlerBootstrap): The probed bootstrap environment.
39
40    Args:
41      env (dict): The environment to probe. If None, `os.getenv` will be used.
42
43    Raises:
44      NotBootstrappedError if the current environment is not bootstrapped.
45    """
46        if env is None:
47            env = os.environ
48
49        def _check(kind, val):
50            if not val:
51                return val
52            try:
53                streamname.validate_stream_name(val)
54                return val
55            except ValueError as exp:
56                raise NotBootstrappedError('%s (%s) is invalid: %s' % (kind, val, exp))
57
58        streamserver_uri = env.get(cls._ENV_STREAM_SERVER_PATH)
59        if not streamserver_uri:
60            raise NotBootstrappedError('No streamserver in bootstrap environment.')
61
62        return cls(
63            project=env.get(cls._ENV_PROJECT, ''),
64            prefix=_check("Prefix", env.get(cls._ENV_PREFIX, '')),
65            streamserver_uri=streamserver_uri,
66            coordinator_host=env.get(cls._ENV_COORDINATOR_HOST, ''),
67            namespace=_check("Namespace", env.get(cls._ENV_NAMESPACE, '')))
68
69    def stream_client(self, reg=None):
70        """Returns: (StreamClient) stream client for the bootstrap streamserver URI.
71
72    If the Butler accepts external stream connections, it will export a
73    streamserver URI in the environment. This will create a StreamClient
74    instance to operate on the streamserver if one is defined.
75
76    Args:
77      reg (stream.StreamProtocolRegistry or None): The stream protocol registry
78          to use to create the stream. If None, the default global registry will
79          be used (recommended).
80
81    Raises:
82      ValueError: If no streamserver URI is present in the environment.
83    """
84        reg = reg or stream._default_registry
85        return reg.create(
86            self.streamserver_uri,
87            project=self.project,
88            prefix=self.prefix,
89            coordinator_host=self.coordinator_host,
90            namespace=self.namespace)
91