gcpdiag.models

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

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

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

244class Resource(abc.ABC):
245  """Represents a single resource in GCP."""
246  _project_id: str
247
248  def __init__(self, project_id):
249    self._project_id = project_id
250
251  def __str__(self):
252    return self.full_path
253
254  def __hash__(self):
255    return self.full_path.__hash__()
256
257  def __lt__(self, other):
258    return self.full_path < other.full_path
259
260  def __eq__(self, other):
261    if self.__class__ == other.__class__:
262      return self.full_path == other.full_path
263    else:
264      return False
265
266  @property
267  def project_id(self) -> str:
268    """Project id (not project number)."""
269    return self._project_id
270
271  @property
272  @abc.abstractmethod
273  def full_path(self) -> str:
274    """Returns the full path of this resource.
275
276    Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'
277    """
278    pass
279
280  @property
281  def short_path(self) -> str:
282    """Returns the short name for this resource.
283
284    Note that it isn't clear from this name what kind of resource it is.
285
286    Example: 'gke1'
287    """
288    return self.full_path