Allow aborting of on-going harvesting tasks

geonode/harvesting/api/serializers.py

@@ -40,7 +40,6 @@
 from .. import (
     models,
     tasks,
-    utils,
 )
 
 logger = logging.getLogger(__name__)
@@ -141,7 +140,7 @@ def validate(self, data):
             worker_config_field, getattr(self.instance, worker_config_field, None))
         if worker_type is not None and worker_config is not None:
             try:
-                utils.validate_worker_configuration(worker_type, worker_config)
+                models.validate_worker_configuration(worker_type, worker_config)
             except jsonschema.exceptions.ValidationError:
                 raise serializers.ValidationError(
                     f"Invalid {worker_config_field!r} configuration")
@@ -162,7 +161,7 @@ def create(self, validated_data):
                 f"value of {models.Harvester.STATUS_READY!r}"
             )
         harvester = super().create(validated_data)
-        available = utils.update_harvester_availability(harvester)
+        available = harvester.update_availability()
         if available:
             harvester.status = harvester.STATUS_UPDATING_HARVESTABLE_RESOURCES
             harvester.save()
@@ -208,12 +207,17 @@ def update(self, instance: models.Harvester, validated_data):
                 f"This status can only be set by the server, when appropriate."
             )
         elif desired_status == models.Harvester.STATUS_UPDATING_HARVESTABLE_RESOURCES:
-            post_update_task = tasks.update_harvestable_resources.signature(
-                args=(instance.id,))
+            session = models.AsynchronousHarvestingSession.objects.create(
+                harvester=instance,
+                session_type=models.AsynchronousHarvestingSession.TYPE_DISCOVER_HARVESTABLE_RESOURCES
+            )
+            post_update_task = tasks.update_harvestable_resources.signature(args=(session.pk,))
         elif desired_status == models.Harvester.STATUS_PERFORMING_HARVESTING:
-            harvesting_session = models.HarvestingSession.objects.create(harvester=instance)
-            post_update_task = tasks.harvesting_dispatcher.signature(
-                args=(instance.pk, harvesting_session.pk))
+            session = models.AsynchronousHarvestingSession.objects.create(
+                harvester=instance,
+                session_type=models.AsynchronousHarvestingSession.TYPE_HARVESTING
+            )
+            post_update_task = tasks.harvesting_dispatcher.signature(args=(session.pk,))
         elif desired_status == models.Harvester.STATUS_CHECKING_AVAILABILITY:
             post_update_task = tasks.check_harvester_available.signature(
                 args=(instance.id,))
@@ -239,15 +243,16 @@ def update(self, instance: models.Harvester, validated_data):
         return updated_instance
 
 
-class BriefHarvestingSessionSerializer(DynamicModelSerializer):
+class BriefAsynchronousHarvestingSessionSerializer(DynamicModelSerializer):
     class Meta:
-        model = models.HarvestingSession
+        model = models.AsynchronousHarvestingSession
         fields = (
             "id",
             "started",
             "updated",
             "ended",
-            "records_harvested",
+            "total_records_to_process",
+            "records_done",
         )
 
 

geonode/harvesting/api/urls.py

@@ -31,6 +31,6 @@
     basename='harvestable-resources',
     parents_query_lookups=['harvester_id']
 )
-router.register('harvesting-sessions', views.HarvestingSessionViewSet)
+router.register('harvesting-sessions', views.AsynchronousHarvestingSessionViewSet)
 
 urlpatterns = router.urls

geonode/harvesting/api/views.py

@@ -94,7 +94,7 @@ def get_serializer_context(self):
         return context
 
 
-class HarvestingSessionViewSet(WithDynamicViewSetMixin, viewsets.ReadOnlyModelViewSet):
-    queryset = models.HarvestingSession.objects.all()
-    serializer_class = serializers.BriefHarvestingSessionSerializer
+class AsynchronousHarvestingSessionViewSet(WithDynamicViewSetMixin, viewsets.ReadOnlyModelViewSet):
+    queryset = models.AsynchronousHarvestingSession.objects.all()
+    serializer_class = serializers.BriefAsynchronousHarvestingSessionSerializer
     pagination_class = GeoNodeApiPagination

geonode/harvesting/harvesters/base.py

@@ -124,7 +124,6 @@ def get_geonode_resource_type(self, remote_resource_type: str) -> ResourceBase:
     def get_resource(
             self,
             harvestable_resource: "HarvestableResource",  # noqa
-            harvesting_session_id: int
     ) -> typing.Optional[HarvestedResourceInfo]:
         """Harvest a single resource from the remote service.
 
@@ -155,7 +154,6 @@ def finalize_resource_update(
             geonode_resource: ResourceBase,
             harvested_info: HarvestedResourceInfo,
             harvestable_resource: "HarvestableResource",  # noqa
-            harvesting_session_id: int
     ) -> ResourceBase:
         """Perform additional actions just after having created/updated a local GeoNode resource.
 
@@ -262,7 +260,6 @@ def update_geonode_resource(
             self,
             harvested_info: HarvestedResourceInfo,
             harvestable_resource: "HarvestableResource",  # noqa
-            harvesting_session_id: int,
     ):
         """Create or update a local GeoNode resource with the input harvested information.
 
@@ -297,7 +294,6 @@ def update_geonode_resource(
             geonode_resource,
             harvested_info,
             harvestable_resource,
-            harvesting_session_id
         )
 
     def _create_new_geonode_resource(self, geonode_resource_type, defaults: typing.Dict):

geonode/harvesting/harvesters/geonodeharvester.py

@@ -279,7 +279,6 @@ def get_geonode_resource_type(self, remote_resource_type: str) -> typing.Type[ty
     def get_resource(
             self,
             harvestable_resource: models.HarvestableResource,
-            harvesting_session_id: int
     ) -> typing.Optional[base.HarvestedResourceInfo]:
         resource_unique_identifier = harvestable_resource.unique_identifier
         local_resource_type = self.get_geonode_resource_type(harvestable_resource.remote_resource_type)

geonode/harvesting/harvesters/wms.py

@@ -161,7 +161,6 @@ def get_geonode_resource_type(self, remote_resource_type: str) -> ResourceBase:
     def get_resource(
             self,
             harvestable_resource: "HarvestableResource",  # noqa
-            harvesting_session_id: int
     ) -> typing.Optional[base.HarvestedResourceInfo]:
         resource_unique_identifier = harvestable_resource.unique_identifier
         data = self._get_data()

geonode/harvesting/migrations/0037_alter_harvester_status.py

@@ -0,0 +1,29 @@
+# Generated by Django 3.2.4 on 2021-09-24 09:42
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('harvesting', '0036_alter_harvester_harvester_type'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='harvester',
+            name='status',
+            field=models.CharField(
+                choices=[
+                    ('ready', 'ready'),
+                    ('updating-harvestable-resources', 'updating-harvestable-resources'),
+                    ('aborting-update-harvestable-resources', 'aborting-update-harvestable-resources'),
+                    ('harvesting-resources', 'harvesting-resources'),
+                    ('aborting-harvesting-resources', 'aborting-harvesting-resources'),
+                    ('checking-availability', 'checking-availability')
+                ],
+                default='ready',
+                max_length=50
+            ),
+        ),
+    ]

geonode/harvesting/migrations/0038_auto_20210927_1455.py

@@ -0,0 +1,34 @@
+# Generated by Django 3.2.4 on 2021-09-27 14:55
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('harvesting', '0037_alter_harvester_status'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='harvestingsession',
+            name='status',
+            field=models.CharField(choices=[('pending', 'pending'), ('on-going', 'on-going'), ('finished-all-ok', 'finished-all-ok'), ('finished-all-failed', 'finished-all-failed'), ('finished-some-failed', 'finished-some-failed'), ('aborting', 'aborting'), ('aborted', 'aborted')], default='pending', editable=False, max_length=50),
+        ),
+        migrations.CreateModel(
+            name='AsynchronousHarvestingSession',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('session_type', models.CharField(choices=[('harvesting', 'harvesting'), ('discover-harvestable-resources', 'discover-harvestable-resources')], editable=False, max_length=50)),
+                ('started', models.DateTimeField(auto_now_add=True)),
+                ('updated', models.DateTimeField(auto_now=True)),
+                ('ended', models.DateTimeField(blank=True, null=True)),
+                ('status', models.CharField(choices=[('pending', 'pending'), ('on-going', 'on-going'), ('finished-all-ok', 'finished-all-ok'), ('finished-all-failed', 'finished-all-failed'), ('finished-some-failed', 'finished-some-failed'), ('aborting', 'aborting'), ('aborted', 'aborted')], default='pending', editable=False, max_length=50)),
+                ('details', models.TextField(blank=True)),
+                ('total_records_to_process', models.IntegerField(default=0, editable=False, help_text='Number of records being processed in this session')),
+                ('records_done', models.IntegerField(default=0, help_text='Number of records that have already been processed')),
+                ('harvester', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='sessions', to='harvesting.harvester')),
+            ],
+        ),
+    ]

geonode/harvesting/migrations/0039_delete_harvestingsession.py

@@ -0,0 +1,16 @@
+# Generated by Django 3.2.4 on 2021-09-27 15:01
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('harvesting', '0038_auto_20210927_1455'),
+    ]
+
+    operations = [
+        migrations.DeleteModel(
+            name='HarvestingSession',
+        ),
+    ]