Improve harvesting scheduler

geonode/harvesting/admin.py

@@ -24,6 +24,7 @@
     admin,
     messages,
 )
+from django.contrib.humanize.templatetags import humanize
 from django.urls import reverse
 from django.utils.html import (
     format_html,
@@ -50,11 +51,13 @@ class HarvesterAdmin(admin.ModelAdmin):
         "scheduling_enabled",
         "remote_available",
         "get_num_harvestable_resources",
-        "get_num_harvestable_resources_selected",
         "show_link_to_selected_harvestable_resources",
         "show_link_to_latest_harvesting_session",
         "show_link_to_latest_refresh_session",
         "get_worker_specific_configuration",
+        "get_time_until_next_availability_update",
+        "get_time_until_next_refresh",
+        "get_time_until_next_harvesting",
     )
     list_filter = (
         "status",
@@ -82,6 +85,7 @@ class HarvesterAdmin(admin.ModelAdmin):
         "initiate_abort_update_harvestable_resources",
         "initiate_perform_harvesting",
         "initiate_abort_perform_harvesting",
+        "reset_harvester_status",
     ]
 
     def save_model(self, request, harvester: models.Harvester, form, change):
@@ -208,15 +212,34 @@ def initiate_abort_perform_harvesting(self, request, queryset):
             message = _("No active harvesting sessions have been found for the selected harvesters. Skipping...")
         self.message_user(request, message)
 
-    @admin.display(description="Number of selected resources to harvest")
-    def get_num_harvestable_resources_selected(self, harvester: models.Harvester):
-        return harvester.harvestable_resources.filter(should_be_harvested=True).count()
+    @admin.action(description="Reset harvester status")
+    def reset_harvester_status(self, request, queryset):
+        for harvester in queryset:
+            if harvester.status != models.Harvester.STATUS_READY:
+                harvester.status = models.Harvester.STATUS_READY
+                harvester.save()
+                self.message_user(
+                    request,
+                    _("Resetting status for harvester %(name)s...") % {"name": harvester.name}
+                )
+
+    @admin.display(description="Updating availability in")
+    def get_time_until_next_availability_update(self, harvester: models.Harvester):
+        return humanize.naturaltime(harvester.get_next_check_availability_dispatch_time())
+
+    @admin.display(description="Refreshing harvestable resources in")
+    def get_time_until_next_refresh(self, harvester: models.Harvester):
+        return humanize.naturaltime(harvester.get_next_refresh_session_dispatch_time())
+
+    @admin.display(description="Performing harvesting in")
+    def get_time_until_next_harvesting(self, harvester: models.Harvester):
+        return humanize.naturaltime(harvester.get_next_harvesting_session_dispatch_time())
 
-    @admin.display(description="Number of existing harvestable resources")
+    @admin.display(description=_("Harvestable resources"))
     def get_num_harvestable_resources(self, harvester: models.Harvester):
         return harvester.num_harvestable_resources
 
-    @admin.display(description="current worker-specific configuration")
+    @admin.display(description="Worker-specific configuration")
     def get_worker_specific_configuration(self, harvester: models.Harvester):
         worker = harvester.get_harvester_worker()
         worker_config = worker.get_current_config()
@@ -226,44 +249,48 @@ def get_worker_specific_configuration(self, harvester: models.Harvester):
         result += "</ul>"
         return mark_safe(result)
 
-    @admin.display(description="Go to selected harvestable resources")
+    @admin.display(description="Selected harvestable resources")
     def show_link_to_selected_harvestable_resources(self, harvester: models.Harvester):
         num_selected = models.HarvestableResource.objects.filter(
             harvester=harvester, should_be_harvested=True).count()
         if num_selected > 0:
             changelist_uri = reverse("admin:harvesting_harvestableresource_changelist")
             result = mark_safe(
                 format_html(
-                    f'<a class="button grp-button" href="{changelist_uri}?harvester__id__exact={harvester.id}&should_be_harvested__exact=1">Go</a>'
+                    f'<a class="button grp-button" href="{changelist_uri}?harvester__id__exact={harvester.id}&should_be_harvested__exact=1">({num_selected}) Go</a>'
                 )
             )
         else:
             result = None
         return result
 
-    @admin.display(description="Go to latest harvesting session")
+    @admin.display(description="Latest harvesting session")
     def show_link_to_latest_harvesting_session(self, harvester: models.Harvester):
-        changelist_uri = reverse(
-            "admin:harvesting_asynchronousharvestingsession_change",
-            args=(harvester.latest_harvesting_session.id,)
-        )
-        return mark_safe(
-            format_html(
-                f'<a class="button grp-button" href="{changelist_uri}">Go</a>'
+        session = harvester.latest_harvesting_session
+        if session is not None:
+            changelist_uri = reverse(
+                "admin:harvesting_asynchronousharvestingsession_change",
+                args=(harvester.latest_harvesting_session.id,)
+            )
+            return mark_safe(
+                format_html(
+                    f'<a class="button grp-button" href="{changelist_uri}">Go</a>'
+                )
             )
-        )
 
-    @admin.display(description="Go to latest refresh session")
+    @admin.display(description="Latest refresh session")
     def show_link_to_latest_refresh_session(self, harvester: models.Harvester):
-        changelist_uri = reverse(
-            "admin:harvesting_asynchronousharvestingsession_change",
-            args=(harvester.latest_refresh_session.id,)
-        )
-        return mark_safe(
-            format_html(
-                f'<a class="button grp-button" href="{changelist_uri}">Go</a>'
+        session = harvester.latest_refresh_session
+        if session is not None:
+            changelist_uri = reverse(
+                "admin:harvesting_asynchronousharvestingsession_change",
+                args=(harvester.latest_refresh_session.id,)
+            )
+            return mark_safe(
+                format_html(
+                    f'<a class="button grp-button" href="{changelist_uri}">Go</a>'
+                )
             )
-        )
 
 
 @admin.register(models.AsynchronousHarvestingSession)

geonode/harvesting/api/serializers.py

@@ -62,7 +62,9 @@ class Meta:
             "remote_url",
             "remote_available",
             "scheduling_enabled",
-            "update_frequency",
+            "harvesting_session_update_frequency",
+            "refresh_harvestable_resources_update_frequency",
+            "check_availability_frequency",
             "default_owner",
             "links",
         )
@@ -102,12 +104,12 @@ class Meta:
             "remote_url",
             "remote_available",
             "scheduling_enabled",
-            "update_frequency",
+            "check_availability_frequency",
+            "harvesting_session_update_frequency",
+            "refresh_harvestable_resources_update_frequency",
             "default_owner",
             "harvester_type",
             "harvester_type_specific_configuration",
-            "update_frequency",
-            "check_availability_frequency",
             "last_checked_availability",
             "last_checked_harvestable_resources",
             "last_check_harvestable_resources_message",

geonode/harvesting/apps.py

@@ -18,17 +18,57 @@
 #########################################################################
 
 from django.apps import AppConfig
+from django.conf import settings
 from django.conf.urls import url, include
+from django.db.models.signals import post_migrate
+
+from . import config
+
+
+def run_setup_hooks(sender, **kwargs):
+    from django.utils import timezone
+
+    # Initialize periodic tasks
+    if 'django_celery_beat' in settings.INSTALLED_APPS and \
+            getattr(settings, 'CELERY_BEAT_SCHEDULER', None) == 'django_celery_beat.schedulers:DatabaseScheduler':
+        from django_celery_beat.models import (
+            IntervalSchedule,
+            PeriodicTask,
+        )
+
+        secs = config.get_setting("HARVESTER_SCHEDULER_FREQUENCY_MINUTES") * 60
+        check_intervals = IntervalSchedule.objects.filter(every=secs, period="seconds")
+        if not check_intervals.exists():
+            check_interval, _ = IntervalSchedule.objects.get_or_create(
+                every=secs,
+                period="seconds"
+            )
+        else:
+            check_interval = check_intervals.first()
+
+        PeriodicTask.objects.update_or_create(
+            name="harvesting-scheduler",
+            defaults=dict(
+                task="geonode.harvesting.tasks.harvesting_scheduler",
+                interval=check_interval,
+                args='',
+                start_time=timezone.now()
+            )
+        )
 
 
 class HarvestingAppConfig(AppConfig):
 
     name = "geonode.harvesting"
 
     def ready(self):
-        from . import signals  # noqa
         from geonode.urls import urlpatterns
 
         urlpatterns += [
             url(r'^api/v2/', include('geonode.harvesting.api.urls'))
         ]
+        post_migrate.connect(run_setup_hooks, sender=self)
+        settings.CELERY_BEAT_SCHEDULE['harvesting-scheduler'] = {
+            "task": "geonode.harvesting.tasks.harvesting_scheduler",
+            "schedule": config.get_setting("HARVESTER_SCHEDULER_FREQUENCY_MINUTES") * 60,
+        }

geonode/harvesting/config.py

@@ -49,6 +49,7 @@ def get_setting(setting_key: str) -> typing.Any:
             getattr(settings, "HARVESTER_CLASSES", [])
         ),
         "HARVESTED_RESOURCE_FILE_MAX_MEMORY_SIZE": getattr(
-            settings, "HARVESTED_RESOURCE_MAX_MEMORY_SIZE", settings.FILE_UPLOAD_MAX_MEMORY_SIZE)
+            settings, "HARVESTED_RESOURCE_MAX_MEMORY_SIZE", settings.FILE_UPLOAD_MAX_MEMORY_SIZE),
+        "HARVESTER_SCHEDULER_FREQUENCY_MINUTES": getattr(settings, "HARVESTER_SCHEDULER_FREQUENCY_MINUTES", 0.5)
     }.get(setting_key, getattr(settings, setting_key, None))
     return result

geonode/harvesting/migrations/0001_initial.py

@@ -10,7 +10,6 @@ class Migration(migrations.Migration):
     initial = True
 
     dependencies = [
-        ('django_celery_beat', '0015_edit_solarschedule_events_choices'),
         migrations.swappable_dependency(settings.AUTH_USER_MODEL),
     ]
 
@@ -27,7 +26,6 @@ class Migration(migrations.Migration):
                 ('harvester_type', models.CharField(choices=[('geonode.harvesting.harvesters.geonode.GeonodeHarvester', 'geonode.harvesting.harvesters.geonode.GeonodeHarvester')], default='geonode.harvesting.harvesters.geonode.GeonodeHarvester', help_text='Harvester class used to perform harvesting sessions', max_length=255)),
                 ('harvester_type_specific_configuration', models.JSONField(default=dict, help_text='Configuration specific to each harvester type. Please consult GeoNode documentation on harvesting for more info.')),
                 ('default_owner', models.ForeignKey(help_text='Default owner of harvested resources', on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
-                ('periodic_task', models.OneToOneField(blank=True, help_text='Periodic task used to configure harvest scheduling', null=True, on_delete=django.db.models.deletion.CASCADE, to='django_celery_beat.PeriodicTask')),
             ],
         ),
         migrations.CreateModel(

geonode/harvesting/migrations/0002_auto_20210415_1719.py

@@ -16,11 +16,6 @@ class Migration(migrations.Migration):
             name='harvester_type',
             field=models.CharField(choices=[('geonode.harvesting.harvesters.geonode.GeonodeHarvester', 'geonode.harvesting.harvesters.geonode.GeonodeHarvester')], default='geonode.harvesting.harvesters.geonode.GeonodeHarvester', help_text='Harvester class used to perform harvesting sessions. New harvester types can be added by an admin by changing the main GeoNode `settings.py` file', max_length=255),
         ),
-        migrations.AlterField(
-            model_name='harvester',
-            name='periodic_task',
-            field=models.OneToOneField(blank=True, editable=False, help_text='Periodic task used to configure harvest scheduling', null=True, on_delete=django.db.models.deletion.CASCADE, to='django_celery_beat.PeriodicTask'),
-        ),
         migrations.AlterField(
             model_name='harvestingsession',
             name='harvester',

geonode/harvesting/migrations/0006_harvester_availability_check_task.py

@@ -7,14 +7,8 @@
 class Migration(migrations.Migration):
 
     dependencies = [
-        ('django_celery_beat', '0015_edit_solarschedule_events_choices'),
         ('harvesting', '0005_auto_20210428_1100'),
     ]
 
     operations = [
-        migrations.AddField(
-            model_name='harvester',
-            name='availability_check_task',
-            field=models.OneToOneField(blank=True, editable=False, help_text='Periodic task used to check availability of the remote', null=True, on_delete=django.db.models.deletion.CASCADE, related_name='checked_harvester', to='django_celery_beat.PeriodicTask'),
-        ),
     ]

geonode/harvesting/migrations/0037_alter_harvester_status.py → geonode/harvesting/migrations/0038_alter_harvester_status.py

@@ -6,7 +6,7 @@
 class Migration(migrations.Migration):
 
     dependencies = [
-        ('harvesting', '0036_alter_harvester_harvester_type'),
+        ('harvesting', '0037_alter_harvester_name'),
     ]
 
     operations = [

geonode/harvesting/migrations/0041_harvestableresource_abstract.py → geonode/harvesting/migrations/0039_harvestableresource_abstract.py

@@ -6,7 +6,7 @@
 class Migration(migrations.Migration):
 
     dependencies = [
-        ('harvesting', '0040_merge_20211001_0832'),
+        ('harvesting', '0038_alter_harvester_status'),
     ]
 
     operations = [

geonode/harvesting/migrations/0038_auto_20210927_1455.py → geonode/harvesting/migrations/0040_auto_20210927_1455.py

@@ -7,7 +7,7 @@
 class Migration(migrations.Migration):
 
     dependencies = [
-        ('harvesting', '0037_alter_harvester_status'),
+        ('harvesting', '0039_harvestableresource_abstract'),
     ]
 
     operations = [

geonode/harvesting/migrations/0039_delete_harvestingsession.py → geonode/harvesting/migrations/0041_delete_harvestingsession.py

@@ -6,7 +6,7 @@
 class Migration(migrations.Migration):
 
     dependencies = [
-        ('harvesting', '0038_auto_20210927_1455'),
+        ('harvesting', '0040_auto_20210927_1455'),
     ]
 
     operations = [

geonode/harvesting/migrations/0042_auto_20210930_1155.py

@@ -0,0 +1,28 @@
+# Generated by Django 3.2.4 on 2021-09-30 11:55
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('harvesting', '0041_delete_harvestingsession'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='harvester',
+            name='refresh_harvestable_resources_update_frequency',
+            field=models.PositiveIntegerField(default=30, help_text='How often (in minutes) should new refresh sessions be automatically scheduled?'),
+        ),
+        migrations.AlterField(
+            model_name='harvester',
+            name='check_availability_frequency',
+            field=models.PositiveIntegerField(default=10, help_text='How often (in minutes) should the remote service be checked for availability?'),
+        ),
+        migrations.AlterField(
+            model_name='harvester',
+            name='update_frequency',
+            field=models.PositiveIntegerField(default=60, help_text='How often (in minutes) should new harvesting sessions be automatically scheduled?'),
+        ),
+    ]

geonode/harvesting/migrations/0043_harvester_harvesting_session_update_frequency.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.2.4 on 2021-09-30 11:58
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('harvesting', '0042_auto_20210930_1155'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='harvester',
+            name='harvesting_session_update_frequency',
+            field=models.PositiveIntegerField(default=60, help_text='How often (in minutes) should new harvesting sessions be automatically scheduled?'),
+        ),
+    ]

geonode/harvesting/migrations/0044_migrate_harvester_harvesting_session_update_frequency.py

@@ -0,0 +1,21 @@
+# Generated by Django 3.2.4 on 2021-09-30 12:01
+
+from django.db import migrations
+
+
+def migrate_harvesting_session_update_frequency(apps, schema_editor):
+    harvester_model = apps.get_model("harvesting", "Harvester")
+    for harvester in harvester_model.objects.all():
+        harvester.harvesting_session_update_frequency = harvester.update_frequency
+        harvester.save()
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('harvesting', '0043_harvester_harvesting_session_update_frequency'),
+    ]
+
+    operations = [
+        migrations.RunPython(migrate_harvesting_session_update_frequency, reverse_code=migrations.RunPython.noop),
+    ]