gcpdiag.queries.osconfig

29class Inventory(models.Resource):
30  """Represents OS Inventory data of a GCE VM instance"""
31
32  _resource_data: dict
33
34  def __init__(self, project_id, resource_data):
35    super().__init__(project_id=project_id)
36    self._resource_data = resource_data
37
38  # e.g: projects/{project_number}/locations/{location}/instances/{instance_id}/inventory
39  @property
40  def full_path(self) -> str:
41    return self._resource_data['name']
42
43  # e.g: {project_number}/{location}/{instance_id}/inventory
44  @property
45  def short_path(self) -> str:
46    path = self.full_path
47    path = re.sub(r'^projects/', '', path)
48    path = re.sub(r'/locations/', '/', path)
49    path = re.sub(r'/instances/', '/', path)
50    return path
51
52  # e.g: '5221437597918447050'
53  @property
54  def instance_id(self) -> str:
55    return self._resource_data['name'].split('/')[-2]
56
57  # e.g: debian, windows.
58  @property
59  def os_shortname(self) -> str:
60    if 'osInfo' in self._resource_data:
61      return self._resource_data['osInfo'].get('shortName', '')
62    return ''
63
64  @property
65  def os_version(self) -> str:
66    if 'osInfo' in self._resource_data:
67      return self._resource_data['osInfo'].get('version', '')
68    return ''
69
70  # <key: installed package name, value: installed version>
71  @property
72  def installed_packages(self) -> Mapping[str, str]:
73    installed_packages: Dict[str, str] = {}
74    if 'items' in self._resource_data:
75      installed_items = [
76        i for i in self._resource_data['items'].values() if i.get('type', '') == 'INSTALLED_PACKAGE'
77      ]
78      for item in installed_items:
79        if 'installedPackage' not in item:
80          continue
81        pkg = item['installedPackage']
82        if 'yumPackage' in pkg:
83          p = pkg['yumPackage']
84          installed_packages[p.get('packageName', '')] = p.get('version', '')
85        elif 'aptPackage' in pkg:
86          p = pkg['aptPackage']
87          installed_packages[p.get('packageName', '')] = p.get('version', '')
88        elif 'googetPackage' in pkg:
89          p = pkg['googetPackage']
90          installed_packages[p.get('packageName', '')] = p.get('version', '')
91        elif 'windowsApplication' in pkg:
92          p = pkg['windowsApplication']
93          installed_packages[p.get('displayName', '')] = p.get('displayVersion', '')
94    return installed_packages

 97@caching.cached_api_call(in_memory=True)
 98def list_inventories(
 99  context: models.Context,
100  location: str,
101) -> Mapping[str, Inventory]:
102  inventories: Dict[str, Inventory] = {}
103  if not apis.is_enabled(context.project_id, 'osconfig'):
104    return inventories
105  osconfig_api = apis.get_api('osconfig', 'v1', context.project_id)
106  logging.debug(
107    'fetching inventory data for all VMs under zone %s in project %s',
108    location,
109    context.project_id,
110  )
111  query = osconfig_api.projects().locations().instances().inventories()
112
113  try:
114    resp = apis_utils.list_all(
115      query.list(
116        parent=(f'projects/{context.project_id}/locations/{location}/instances/-'),
117        view='FULL',
118      ),
119      query.list_next,
120      'inventories',
121    )
122  except googleapiclient.errors.HttpError as err:
123    if err.resp.status in [404]:
124      return inventories
125    raise utils.GcpApiError(err) from err
126
127  for i in resp:
128    inventory = Inventory(context.project_id, resource_data=i)
129    inventories[inventory.instance_id] = inventory
130  return inventories

133@caching.cached_api_call(in_memory=True)
134def get_inventory(
135  context: models.Context, location: str, instance_name: str
136) -> Optional[Inventory]:
137  if not apis.is_enabled(context.project_id, 'osconfig'):
138    return None
139  osconfig_api = apis.get_api('osconfig', 'v1', context.project_id)
140  logging.debug(
141    'fetching inventory data for VM %s in zone %s in project %s',
142    instance_name,
143    location,
144    context.project_id,
145  )
146  query = (
147    osconfig_api.projects()
148    .locations()
149    .instances()
150    .inventories()
151    .get(
152      name=f'projects/{context.project_id}/locations/{location}/instances/{instance_name}/inventory',
153      view='FULL',
154    )
155  )
156  try:
157    resp = query.execute(num_retries=config.API_RETRIES)
158  except googleapiclient.errors.HttpError as err:
159    if err.resp.status in [404]:
160      return None
161    raise utils.GcpApiError(err) from err
162  return Inventory(context.project_id, resource_data=resp)