gcpdiag.models

32class Messages(dict):
33
34  def get_msg(self, template: str, **kwargs):
35    return self.get(
36        template,
37        'NOTICE: No message available to parse for this step').format(**kwargs)

44class Parameter(dict[T, V], Generic[T, V]):
45  """Class to store parameters"""
46
47  def __init__(self, *args, **kwargs):
48    super().__init__()
49    for dict_arg in args:
50      for key, value in dict_arg.items():
51        self[key] = value
52    for key, value in kwargs.items():
53      self[key] = value
54
55  def _parse_value(self, value: str) -> Any:
56    """Make all values lower string and strip whitespaces."""
57    if isinstance(value, str):
58      return value.strip().lower()
59    return value
60
61  def __setitem__(self, key: T, value: V) -> None:
62    super().__setitem__(key, self._parse_value(value))
63
64  def update(self, *args, **kwargs) -> None:
65    for k, v in dict(*args, **kwargs).items():
66      self[k] = v
67
68  def setdefault(self, key: T, default: V = None) -> V:
69    if key not in self:
70      converted_default = self._parse_value(default) if isinstance(
71          default, str) else default
72      self[key] = converted_default
73    return super().setdefault(key, self[key])

 76@dataclasses.dataclass
 77class Context:
 78  """List of resource groups / scopes that should be analyzed."""
 79  # project_id of project that is being analyzed, mandatory
 80  project_id: str
 81  # a pattern of sub project resources that match
 82  resources_pattern: Optional[re.Pattern]
 83  # list of GCP all locations to use as linting scope
 84  # i.e. regions (e.g.: 'us-central1') or zone (e.g.: 'us-central1-a').
 85  # a compiled project resources provided by user
 86  locations_pattern: Optional[re.Pattern]
 87
 88  # list of "label sets" that must match.
 89  labels: Optional[Mapping[str, str]]
 90  # list of "runbook parameters sets" that must match.
 91  parameters: Parameter[str, Any]
 92
 93  # the selected resources are the intersection of project_id, locations,
 94  # and labels(i.e. all must match), but each value in locations, and
 95  # labels is a OR, so it means:
 96  # project_id AND
 97  # (region1 OR region2) AND
 98  # ({label1=value1,label2=value2} OR {label3=value3})
 99
100  def __init__(self,
101               project_id: str,
102               locations: Optional[Iterable[str]] = None,
103               labels: Optional[Mapping[str, str]] = None,
104               parameters: Optional[Parameter[str, str]] = None,
105               resources: Optional[Iterable[str]] = None):
106    """Args:
107
108      project: project_id of project that should be inspected.
109      locations: only include resources in these GCP locations.
110      labels: only include resources with these labels. Expected
111        is a dict, is a set of key=value pairs that must match.
112
113        Example: `{'key1'='bla', 'key2'='baz'}`. This
114        will match resources that either have key1=bla or key2=baz.
115      resources: only include sub project resources with this name attribute.
116    """
117
118    self.project_id = project_id
119
120    if locations:
121      if not isinstance(locations, List):
122        raise ValueError(
123            str(locations) + ' did not supply full list of locations')
124      for location in locations:
125        if not (utils.is_region(location) or utils.is_zone(location)):
126          raise ValueError(location + ' does not look like a valid region/zone')
127
128      self.locations_pattern = re.compile('|'.join(locations), re.IGNORECASE)
129    else:
130      self.locations_pattern = None
131
132    if labels:
133      if not isinstance(labels, Mapping):
134        raise ValueError('labels must be Mapping[str,str]]')
135
136      self.labels = labels
137    else:
138      self.labels = None
139
140    if resources:
141      if not isinstance(resources, List):
142        raise ValueError(
143            str(resources) + ' did not supply full list of resources')
144
145      self.resources_pattern = re.compile('|'.join(resources), re.IGNORECASE)
146
147    else:
148      self.resources_pattern = None
149
150    if parameters:
151      if not isinstance(parameters, Mapping):
152        raise ValueError('parameters must be Mapping[str,str]]')
153
154      self.parameters = Parameter(parameters)
155    else:
156      self.parameters = Parameter()
157    self.parameters['project_id'] = self.project_id
158
159  def __str__(self):
160    string = 'project: ' + self.project_id
161    if self.resources_pattern:
162      string += ', resources: ' + self.resources_pattern.pattern
163    if self.locations_pattern:
164      string += ', locations (regions/zones): ' + self.locations_pattern.pattern
165    if self.labels:
166      string += ', labels: {' + _mapping_str(self.labels) + '}'
167    if self.parameters:
168      string += ', parameters: {' + _mapping_str(self.parameters) + '}'
169    return string
170
171  def __hash__(self):
172    return self.__str__().__hash__()
173
174  IGNORELOCATION = 'IGNORELOCATION'
175  IGNORELABEL = MappingProxyType({'IGNORELABEL': 'IGNORELABEL'})
176
177  def match_project_resource(
178      self,
179      resource: Optional[str],
180      location: Optional[str] = IGNORELOCATION,
181      labels: Optional[Mapping[str, str]] = IGNORELABEL,
182  ) -> bool:
183    """Compare resource fields to the name and/or location and/or labels supplied
184    by the user and return a boolean outcome depending on the context.
185
186    Args:
187      rosource: name of the resource under analysis. Always inspected if user
188      supplied a name criteria
189
190      location: region or zone of the resource. IGNORELOCATION completely skips analysis
191      of the location even if user has supplied location criteria
192
193      labels: labels in the resource under inspection. Functions which do not
194      support labels can completely skip checks by providing the IGNORELABEL constant
195
196    Returns:
197      A boolean which indicates the outcome of the analysis
198    """
199
200    # Match resources.
201    if self.resources_pattern:
202      if not resource or not self.resources_pattern.match(resource):
203        return False
204
205    # Match location.
206    if self.locations_pattern and location is not self.IGNORELOCATION:
207      if not location or not self.locations_pattern.match(location):
208        return False
209
210    # Match labels.
211    if self.labels and labels is not self.IGNORELABEL:
212      if not labels:
213        return False
214
215      if any(labels.get(k) == v for k, v in self.labels.items()):
216        pass
217      else:
218        return False
219
220    # Everything matched.
221    return True

224class Resource(abc.ABC):
225  """Represents a single resource in GCP."""
226  _project_id: str
227
228  def __init__(self, project_id):
229    self._project_id = project_id
230
231  def __str__(self):
232    return self.full_path
233
234  def __hash__(self):
235    return self.full_path.__hash__()
236
237  def __lt__(self, other):
238    return self.full_path < other.full_path
239
240  def __eq__(self, other):
241    if self.__class__ == other.__class__:
242      return self.full_path == other.full_path
243    else:
244      return False
245
246  @property
247  def project_id(self) -> str:
248    """Project id (not project number)."""
249    return self._project_id
250
251  @property
252  @abc.abstractmethod
253  def full_path(self) -> str:
254    """Returns the full path of this resource.
255
256    Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'
257    """
258    pass
259
260  @property
261  def short_path(self) -> str:
262    """Returns the short name for this resource.
263
264    Note that it isn't clear from this name what kind of resource it is.
265
266    Example: 'gke1'
267    """
268    return self.full_path