gcpdiag.models

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

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(default, str) else default
71      self[key] = converted_default
72    return super().setdefault(key, self[key])
73
74  def __str__(self):
75    return _mapping_str(self)

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

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