Merge pull request #264 from ricardobranco777/paginate_gce

gce: Centralize deletion & pagination logic

ocw/lib/db.py

@@ -1,11 +1,11 @@
 import json
 import traceback
 import logging
-from datetime import datetime, timedelta
+from os.path import basename
+from datetime import datetime, timedelta, timezone
 import dateutil.parser as dateparser
 from django.db import transaction
 from django.db.models import F
-from django.utils import timezone
 from ocw.apps import getScheduler
 from webui.PCWConfig import PCWConfig
 from ..models import Instance, StateChoice, ProviderChoice, CspInfo
@@ -74,10 +74,10 @@ def ec2_extract_data(csp_instance, namespace: str, region: str, default_ttl: int
 def azure_extract_data(csp_instance, namespace: str, default_ttl: int) -> dict:
     if csp_instance.tags:
         tags = csp_instance.tags
-        first_seen = dateparser.parse(tags.get('openqa_created_date', str(timezone.now())))
+        first_seen = dateparser.parse(tags.get('openqa_created_date', str(datetime.now(tz=timezone.utc))))
     else:
         tags = {}
-        first_seen = dateparser.parse(str(timezone.now()))
+        first_seen = dateparser.parse(str(datetime.now(tz=timezone.utc)))
     return {
         'tags': tags,
         'id': csp_instance.name,
@@ -100,9 +100,9 @@ def gce_extract_data(csp_instance, namespace: str, default_ttl: int) -> dict:
         'id': csp_instance['id'],
         'first_seen': first_seen,
         'namespace': namespace,
-        'region': GCE.url_to_name(csp_instance['zone']),
+        'region': basename(csp_instance['zone']),
         'provider': ProviderChoice.GCE,
-        'type': GCE.url_to_name(csp_instance['machineType']),
+        'type': basename(csp_instance['machineType']),
         'default_ttl': default_ttl
     }
 

ocw/lib/gce.py

@@ -1,4 +1,5 @@
 import json
+from os.path import basename
 from datetime import timezone
 from dateutil.parser import parse
 import googleapiclient.discovery
@@ -21,6 +22,39 @@ def __init__(self, namespace):
         self.private_key_data = self.get_data()
         self.project = self.private_key_data["project_id"]
 
+    def _paginated(self, api_call, **kwargs) -> list:
+        results = []
+        request = api_call().list(**kwargs)
+        while request is not None:
+            response = request.execute()
+            if "items" in response:
+                results.extend(response["items"])
+            else:
+                self.log_dbg(f"response has no items. id={response.get('id')}")
+            request = api_call().list_next(previous_request=request, previous_response=response)
+        return results
+
+    def _delete_resource(self, api_call, resource_name, *args, **kwargs) -> None:
+        resource_type = {
+            self.compute_client().instances: "instance",
+            self.compute_client().images: "image",
+            self.compute_client().disks: "disk",
+        }.get(api_call, "resource")
+        if self.dry_run:
+            self.log_info(f"Deletion of {resource_type} {resource_name} skipped due to dry run mode")
+            return
+        request = api_call().delete(**kwargs)
+        try:
+            self.log_info(f"Delete {resource_type.title()} '{resource_name}'")
+            response = request.execute()
+            self.log_dbg(f"Deletion response: {response}")
+            self.log_info(f"{resource_type.title()} '{resource_name}' deleted")
+        except HttpError as err:
+            if GCE.get_error_reason(err) == 'resourceInUseByAnotherResource':
+                self.log_dbg(f"{resource_type.title()} '{resource_name}' can not be deleted because in use")
+            else:
+                raise err
+
     def compute_client(self):
         if self.__compute_client is None:
             credentials = service_account.Credentials.from_service_account_info(self.private_key_data)
@@ -29,72 +63,38 @@ def compute_client(self):
             )
         return self.__compute_client
 
-    def list_instances(self, zone):
+    def list_instances(self, zone) -> list:
         """ List all instances by zone."""
-        self.log_dbg("Call list_instances for {}", zone)
-        result = []
-        request = (
-            self.compute_client().instances().list(project=self.project, zone=zone)
-        )
-        while request is not None:
-            response = request.execute()
-            if "items" in response:
-                result += response["items"]
-            request = (
-                self.compute_client()
-                .instances()
-                .list_next(previous_request=request, previous_response=response)
-            )
-        return result
+        self.log_dbg(f"Call list_instances for {zone}")
+        return self._paginated(self.compute_client().instances, project=self.project, zone=zone)
 
-    def list_all_instances(self):
+    def list_all_instances(self) -> list:
         result = []
         self.log_dbg("Call list_all_instances")
         for region in self.list_regions():
             for zone in self.list_zones(region):
-                result += self.list_instances(zone=zone)
+                result.extend(self.list_instances(zone=zone))
         return result
 
-    def list_regions(self):
+    def list_regions(self) -> list:
         """Walk through all regions->zones and collect all instances to return them as list.
         @see https://cloud.google.com/compute/docs/reference/rest/v1/instances/list#examples"""
-        result = []
-        request = self.compute_client().regions().list(project=self.project)
-        while request is not None:
-            response = request.execute()
-
-            for region in response["items"]:
-                result.append(region["name"])
-            request = (
-                self.compute_client()
-                .regions()
-                .list_next(previous_request=request, previous_response=response)
-            )
-        return result
+        regions = self._paginated(self.compute_client().regions, project=self.project)
+        return [region["name"] for region in regions]
 
-    def list_zones(self, region):
+    def list_zones(self, region) -> list:
         region = (
             self.compute_client()
             .regions()
             .get(project=self.project, region=region)
             .execute()
         )
-        return [GCE.url_to_name(z) for z in region["zones"]]
+        return [basename(z) for z in region["zones"]]
 
-    def delete_instance(self, instance_id, zone):
-        if self.dry_run:
-            self.log_info(
-                "Deletion of instance {} skipped due to dry run mode", instance_id
-            )
-        else:
-            self.log_info("Delete instance {}".format(instance_id))
-            self.compute_client().instances().delete(
-                project=self.project, zone=zone, instance=instance_id
-            ).execute()
-
-    @staticmethod
-    def url_to_name(url):
-        return url[url.rindex("/")+1:]
+    def delete_instance(self, instance_id, zone) -> None:
+        self._delete_resource(
+            self.compute_client().instances, instance_id, project=self.project, zone=zone, instance=instance_id
+        )
 
     @staticmethod
     def get_error_reason(error: "googleapiclient.errors.HttpError") -> str:
@@ -106,77 +106,30 @@ def get_error_reason(error: "googleapiclient.errors.HttpError") -> str:
             pass
         return reason
 
-    def cleanup_all(self):
+    def cleanup_all(self) -> None:
         self.log_info("Call cleanup_all")
+        self.cleanup_disks()
+        self.cleanup_images()
 
+    def cleanup_disks(self) -> None:
         self.log_dbg("Disks cleanup")
         for region in self.list_regions():
             for zone in self.list_zones(region):
-                self.log_dbg("Searching for disks in {}", zone)
-                request = self.compute_client().disks().list(project=self.project, zone=zone)
-                while request is not None:
-                    response = request.execute()
-                    if "items" not in response:
-                        self.log_dbg("response has no items. id={}", response["id"])
-                        break
-                    self.log_dbg("response has {} items. id={}", len(response["items"]), response["id"])
-                    for disk in response["items"]:
-                        if self.is_outdated(parse(disk["creationTimestamp"]).astimezone(timezone.utc)):
-                            if self.dry_run:
-                                self.log_info("Deletion of disk {} created on {} skipped due to dry run mode",
-                                              disk["name"], disk["creationTimestamp"])
-                            else:
-                                delete_request = (
-                                    self.compute_client()
-                                    .disks()
-                                    .delete(project=self.project, zone=zone, disk=disk["name"])
-                                )
-                                try:
-                                    delete_response = delete_request.execute()
-                                    self.log_dbg("Deletion response: {}", delete_response)
-                                    self.log_info("Disk '{}' deleted", disk["name"])
-                                except HttpError as err:
-                                    if GCE.get_error_reason(err) == 'resourceInUseByAnotherResource':
-                                        self.log_dbg("Disk {} can not be deleted because in use", disk["name"])
-                                    else:
-                                        raise err
-                    request = (
-                        self.compute_client()
-                        .disks()
-                        .list_next(previous_request=request, previous_response=response)
-                    )
+                self.log_dbg(f"Searching for disks in {zone}")
+                disks = self._paginated(self.compute_client().disks, project=self.project, zone=zone)
+                self.log_dbg(f"{len(disks)} disks found")
+                for disk in disks:
+                    if self.is_outdated(parse(disk["creationTimestamp"]).astimezone(timezone.utc)):
+                        self._delete_resource(
+                            self.compute_client().disks, disk["name"], project=self.project, zone=zone, disk=disk["name"]
+                        )
 
+    def cleanup_images(self) -> None:
         self.log_dbg("Images cleanup")
-        request = self.compute_client().images().list(project=self.project)
-
-        while request is not None:
-            response = request.execute()
-            if "items" not in response:
-                self.log_dbg("response has no items. id={}", response["id"])
-                break
-            self.log_dbg("response has {} items. id={}", len(response["items"]), response["id"])
-            for image in response["items"]:
-                if self.is_outdated(parse(image["creationTimestamp"]).astimezone(timezone.utc)):
-                    if self.dry_run:
-                        self.log_info("Deletion of image {} skipped due to dry run mode", image["name"])
-                    else:
-                        delete_request = (
-                            self.compute_client()
-                            .images()
-                            .delete(project=self.project, image=image["name"])
-                        )
-                        try:
-                            delete_response = delete_request.execute()
-                            self.log_dbg("Deletion response: {}", delete_response)
-                            self.log_info("Delete image '{}'", image["name"])
-                        except HttpError as err:
-                            if GCE.get_error_reason(err) == 'resourceInUseByAnotherResource':
-                                self.log_dbg("Image {} can not be deleted because in use", image["name"])
-                            else:
-                                raise err
-
-            request = (
-                self.compute_client()
-                .images()
-                .list_next(previous_request=request, previous_response=response)
-            )
+        images = self._paginated(self.compute_client().images, project=self.project)
+        self.log_dbg(f"{len(images)} images found")
+        for image in images:
+            if self.is_outdated(parse(image["creationTimestamp"]).astimezone(timezone.utc)):
+                self._delete_resource(
+                    self.compute_client().images, image["name"], project=self.project, image=image["name"]
+                )

ocw/models.py

@@ -1,6 +1,5 @@
 from django.db import models
-from django.utils import timezone
-from datetime import timedelta
+from datetime import datetime, timedelta, timezone
 from webui.PCWConfig import PCWConfig
 import json
 from .enums import ProviderChoice, StateChoice
@@ -50,7 +49,7 @@ def all_time_fields(self):
         return all_time_pattern.format(self.age_formated(), first_fmt, last_fmt, self.ttl_formated())
 
     def set_alive(self):
-        self.last_seen = timezone.now()
+        self.last_seen = datetime.now(tz=timezone.utc)
         self.active = True
         self.age = self.last_seen - self.first_seen
         if self.state != StateChoice.DELETING:

requirements.txt

@@ -6,16 +6,16 @@ azure-storage-blob==12.13.0
 azure-identity==1.10.0
 msrestazure==0.6.4
 uwsgi==2.0.21
-requests==2.28.1
-django~=4.0.10
-django-tables2==2.4.1
-django-filter==22.1
-django-bootstrap4==22.1
+requests==2.31.0
+Django~=4.2.2
+django-tables2==2.5.3
+django-filter==23.2
+django-bootstrap4==23.1
 texttable
 oauth2client
 google-api-python-client==2.55.0
 google-cloud-storage==2.4.0
 openstacksdk~=1.2.0
 python-dateutil
 apscheduler
-kubernetes
+kubernetes

tests/test_db.py

@@ -1,3 +1,4 @@
+from os.path import basename
 from ocw.lib.db import update_run, ec2_extract_data, gce_extract_data, azure_extract_data, delete_instance
 from webui.PCWConfig import PCWConfig
 from faker import Faker
@@ -8,8 +9,7 @@
 from ocw.lib.EC2 import EC2
 import pytest
 import dateutil.parser as dateparser
-from django.utils import timezone
-from datetime import datetime
+from datetime import datetime, timezone
 
 fake = Faker()
 
@@ -106,9 +106,9 @@ def test_gce_extract_data(extract_data):
     assert rez['id'] == csp_instance['id']
     assert rez['first_seen'] == csp_instance['creationTimestamp']
     assert rez['namespace'] == extract_data['namespace']
-    assert rez['region'] == GCE.url_to_name(csp_instance['zone'])
+    assert rez['region'] == basename(csp_instance['zone'])
     assert rez['provider'] == ProviderChoice.GCE
-    assert rez['type'] == GCE.url_to_name(csp_instance['machineType'])
+    assert rez['type'] == basename(csp_instance['machineType'])
     assert rez['default_ttl'] == extract_data['default_ttl']
 
 

tests/test_gce.py

@@ -73,7 +73,9 @@ def delete(self, *args, **kwargs):
             self.deleted_disks.append(kwargs['disk'])
         else:
             raise ValueError("Unexpected delete request")
-        return self.responses.pop(0)
+        if len(self.responses) > 0:
+            return self.responses.pop(0)
+        return FakeRequest(None)
 
 
 def test_cleanup_all(monkeypatch):
@@ -116,6 +118,7 @@ def test_cleanup_all(monkeypatch):
 
     def mocked_compute_client():
         pass
+    mocked_compute_client.instances = lambda *args, **kwargs: []
     mocked_compute_client.images = lambda *args, **kwargs: fmi
     mocked_compute_client.disks = lambda *args, **kwargs: fmd
     monkeypatch.setattr(GCE, 'compute_client', lambda self: mocked_compute_client)

webui/settings.py

@@ -124,10 +124,12 @@
 
 LANGUAGE_CODE = 'en-us'
 
-TIME_ZONE = 'UTC'
-
 USE_I18N = True
 
+# NOTE:
+# If you change these values you would have to change every instance of
+# `datetime.now(tz=timezone.utc)` accordingly
+TIME_ZONE = 'UTC'
 USE_TZ = True