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()
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])
74
75  def __str__(self):
76    return _mapping_str(self)

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

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