Cache stats in Redis while jobs are running (#206)

* BITMAKER-1980: Save real-time stats in Redis (#144)

* Retrieve job stats from Redis if the job is running
* Catch Exception rather than PyMongo exceptions
* Add unless-stopped restart policy to docker services
* Save stats from Redis when job is manually stopped

Author: mgonnav

database_adapters/db_adapters.py

@@ -2,7 +2,7 @@
 
 import pymongo
 from bson.objectid import ObjectId
-from pymongo.errors import ConnectionFailure, PyMongoError
+from pymongo.errors import ConnectionFailure
 
 
 class InsertionResponse:
@@ -163,7 +163,7 @@ def insert_one_to_unique_collection(self, database_name, collection_name, item):
                 item, {"$set": item}, upsert=True
             )
             response = InsertionResponse(True)
-        except PyMongoError as ex:
+        except Exception as ex:
             response = InsertionResponse(False, ex)
         finally:
             return response
@@ -173,7 +173,7 @@ def insert_one_to_collection(self, database_name, collection_name, item):
         try:
             self.client[database_name][collection_name].insert_one(item)
             response = InsertionResponse(True)
-        except PyMongoError as ex:
+        except Exception as ex:
             response = InsertionResponse(False, ex)
         finally:
             return response
@@ -187,7 +187,7 @@ def insert_many_to_collection(
                 items, ordered=ordered
             )
             response = InsertionResponse(True)
-        except PyMongoError as ex:
+        except Exception as ex:
             response = InsertionResponse(False, ex, need_upsert=True)
         finally:
             return response

docs/estela/installation/helm-variables.md

@@ -85,6 +85,10 @@ All the queue platform variables should be written as children of the _<QUEUE\_P
 {: .highlight }
 > Refer to the [estela Queue Adapter](./../queueing.html#estela-queue-adapter) documentation to fill in any additional variables needed for the selected queue platform.
 
+#### Redis Stats
+* _<REDIS\_URL>_ (Required): The connection URL to the Redis instance.
+* _<REDIS\_STATS\_INTERVAL>_ (Required): The interval, in seconds, of how often the job stats should be updated.
+
 ### estela API variables
 
 #### Database
@@ -176,26 +180,26 @@ All the queue platform variables should be written as children of the _<QUEUE\_P
   to `"True"`.
 
 * _<WORKER\_POOL>_ (Optional): Number of worker threads per consumer, it must be an integer.
-  If the variable is left blank, the default value is 10.
+  If the variable is left blank, the default value is `10`.
 
 * _<HEARTBEAT\_TICK>_ (Optional): Number of seconds between heartbeat inspections, it must
-  be an integer. If the variable is left blank, the default value is 300.
+  be an integer. If the variable is left blank, the default value is `300`.
 
 * _<QUEUE\_BASE\_TIMEOUT>_ (Optional): Minimum number of seconds a worker thread can wait 
   for an item to be available in the internal item queue, it must be an integer. If the 
-  variable is left blank, the default value is 5.
+  variable is left blank, the default value is `5`.
 
 * _<QUEUE\_MAX\_TIMEOUT>_ (Optional): Maximum number of seconds a worker thread can wait 
   for an item to be available in the internal item queue, it must be an integer. If the 
-  variable is left blank, the default value is 300.
+  variable is left blank, the default value is `300`.
 
 * _<BATCH\_SIZE\_THRESHOLD>_ (Optional): Size threshold in bytes of the data batch to be
-  inserted, it must be an integer. If the variable is left blank, the default value is 4096.
+  inserted, it must be an integer. If the variable is left blank, the default value is `4096`.
 
 * _<INSERT\_TIME\_THRESHOLD>_ (Optional): Time threshold in seconds of the insertion of 
   consecutive items belonging to the same batch of data, it must be an integer. If the 
-  variable is left blank, the default value is 5.
+  variable is left blank, the default value is `5`.
 
 * _<ACTIVITY\_TIME\_THRESHOLD>_ (Optional): Time threshold in seconds of the activity time 
   of an Inserter object before being cleaned up, it must be an integer. If the variable is 
-  left blank, the default value is 600.
+  left blank, the default value is `600`.

estela-api/api/filters.py

@@ -1,6 +1,6 @@
 from django_filters import rest_framework as filters
 
-from core.models import SpiderJob, SpiderCronJob
+from core.models import SpiderCronJob, SpiderJob
 
 
 class SpiderJobFilter(filters.FilterSet):

estela-api/api/mixins.py

@@ -4,7 +4,7 @@
 from rest_framework.pagination import PageNumberPagination
 from rest_framework.permissions import IsAuthenticated
 
-from api.permissions import IsProjectUser, IsAdminOrReadOnly
+from api.permissions import IsAdminOrReadOnly, IsProjectUser
 from core.models import Notification, Activity
 
 

estela-api/api/permissions.py

@@ -1,7 +1,8 @@
 from rest_framework.permissions import BasePermission, SAFE_METHODS
-from core.models import Project, Permission
 from django.contrib.auth.models import User
 
+from core.models import Project, Permission
+
 
 class IsProjectUser(BasePermission):
     def has_permission(self, request, view):

estela-api/api/serializers/deploy.py

@@ -1,8 +1,8 @@
 from rest_framework import serializers
 
-from core.models import Deploy, Spider
-from api.serializers.spider import SpiderSerializer
 from api.serializers.project import UserDetailSerializer
+from api.serializers.spider import SpiderSerializer
+from core.models import Deploy, Spider
 
 
 class DeploySerializer(serializers.ModelSerializer):

estela-api/api/serializers/job.py

@@ -1,3 +1,6 @@
+from datetime import timedelta
+
+from django.conf import settings
 from rest_framework import serializers
 
 from api import errors
@@ -6,6 +9,7 @@
     SpiderJobEnvVarSerializer,
     SpiderJobTagSerializer,
 )
+from api.utils import update_stats_from_redis, delete_stats_from_redis
 from config.job_manager import job_manager
 from core.models import (
     DataStatus,
@@ -157,6 +161,13 @@ def update(self, instance, validated_data):
                         }
                     )
                 else:
+                    if instance.status == SpiderJob.RUNNING_STATUS:
+                        try:
+                            update_stats_from_redis(instance)
+                            delete_stats_from_redis(instance)
+                        except:
+                            pass
+                        instance.save()
                     job_manager.delete_job(instance.name)
         instance.status = status
 

estela-api/api/utils.py

@@ -0,0 +1,26 @@
+import redis
+
+from django.conf import settings
+
+from datetime import timedelta
+
+
+def update_stats_from_redis(job):
+    redis_conn = redis.from_url(settings.REDIS_URL)
+    job_stats = redis_conn.hgetall(f"scrapy_stats_{job.key}")
+    job.lifespan = timedelta(
+        seconds=int(float(job_stats.get(b"elapsed_time_seconds", b"0").decode()))
+    )
+    job.total_response_bytes = int(
+        job_stats.get(b"downloader/response_bytes", b"0").decode()
+    )
+    job.item_count = int(job_stats.get(b"item_scraped_count", b"0").decode())
+    job.request_count = int(job_stats.get(b"downloader/request_count", b"0").decode())
+
+
+def delete_stats_from_redis(job):
+    redis_conn = redis.from_url(settings.REDIS_URL)
+    try:
+        redis_conn.delete(f"scrapy_stats_{job.key}")
+    except:
+        pass

estela-api/api/views/deploy.py

@@ -1,20 +1,19 @@
 from django.shortcuts import get_object_or_404
 from drf_yasg.utils import swagger_auto_schema
-from rest_framework import viewsets, status
+from rest_framework import status, viewsets
+from rest_framework.exceptions import APIException, ParseError, PermissionDenied
 from rest_framework.response import Response
-from rest_framework.exceptions import ParseError, APIException, PermissionDenied
 
-from api.mixins import BaseViewSet, ActionHandlerMixin
+from api import errors
+from api.mixins import ActionHandlerMixin, BaseViewSet
 from api.serializers.deploy import (
-    DeploySerializer,
     DeployCreateSerializer,
+    DeploySerializer,
     DeployUpdateSerializer,
 )
-
-from api import errors
+from config.job_manager import credentials
 from core.models import Deploy, Project
 from core.views import launch_deploy_job
-from config.job_manager import credentials
 
 
 class DeployViewSet(

estela-api/api/views/job.py

@@ -1,20 +1,23 @@
+from datetime import timedelta
+
 from django.shortcuts import get_object_or_404
 from django_filters.rest_framework import DjangoFilterBackend
-from drf_yasg.utils import swagger_auto_schema
 from drf_yasg import openapi
+from drf_yasg.utils import swagger_auto_schema
 from rest_framework import mixins, status
-from rest_framework.response import Response
 from rest_framework.exceptions import ParseError
+from rest_framework.response import Response
 
 from api.filters import SpiderJobFilter
 from api.mixins import BaseViewSet, ActionHandlerMixin
 from api.serializers.job import (
-    SpiderJobSerializer,
     SpiderJobCreateSerializer,
+    SpiderJobSerializer,
     SpiderJobUpdateSerializer,
 )
+from api.utils import update_stats_from_redis
 from config.job_manager import job_manager
-from core.models import DataStatus, Spider, SpiderJob, Project
+from core.models import DataStatus, Project, Spider, SpiderJob
 
 
 class SpiderJobViewSet(
@@ -169,3 +172,13 @@ def update(self, request, *args, **kwargs):
 
         headers = self.get_success_headers(serializer.data)
         return Response(serializer.data, status=status.HTTP_200_OK, headers=headers)
+
+    def retrieve(self, request, *args, jid=None, **kwargs):
+        job = get_object_or_404(self.queryset, jid=jid)
+        if job.status == SpiderJob.RUNNING_STATUS:
+            update_stats_from_redis(job)
+
+        serializer = SpiderJobSerializer(job)
+
+        headers = self.get_success_headers(serializer.data)
+        return Response(serializer.data, status=status.HTTP_200_OK, headers=headers)

estela-api/api/views/job_data.py

@@ -1,3 +1,5 @@
+import redis
+
 from django.conf import settings
 from drf_yasg import openapi
 from drf_yasg.utils import swagger_auto_schema
@@ -25,6 +27,10 @@ class JobDataViewSet(
     DEFAULT_PAGINATION_SIZE = 50
     JOB_DATA_TYPES = ["items", "requests", "logs", "stats"]
 
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.redis_conn = redis.from_url(settings.REDIS_URL)
+
     def get_parameters(self, request):
         page = int(request.query_params.get("page", 1))
         data_type = request.query_params.get("type", "items")
@@ -98,10 +104,18 @@ def list(self, request, *args, **kwargs):
         count = spiderdata_db_client.get_estimated_document_count(
             kwargs["pid"], job_collection_name
         )
+
         if data_type == "stats":
-            result = spiderdata_db_client.get_job_stats(
-                kwargs["pid"], job_collection_name
-            )
+            if job.status == SpiderJob.RUNNING_STATUS:
+                job_stats = self.redis_conn.hgetall(f"scrapy_stats_{job.key}")
+                parsed_job_stats = {
+                    key.decode(): value.decode() for key, value in job_stats.items()
+                }
+                result = [parsed_job_stats]
+            else:
+                result = spiderdata_db_client.get_job_stats(
+                    kwargs["pid"], job_collection_name
+                )
         elif request.META["HTTP_USER_AGENT"].startswith("estela-cli/"):
             chunk_size = max(
                 1,
@@ -223,19 +237,10 @@ def delete(self, request, *args, **kwargs):
         data_type = request.query_params.get("type")
         if not spiderdata_db_client.get_connection():
             raise DataBaseError({"error": errors.UNABLE_CONNECT_DB})
-        if (
-            job.cronjob is not None
-            and job.cronjob.unique_collection
-            and data_type == "items"
-        ):
-            job_collection_name = "{}-scj{}-job_{}".format(
-                kwargs["sid"], job.cronjob.cjid, data_type
-            )
-        else:
-            job_collection_name = "{}-{}-job_{}".format(
-                kwargs["sid"], kwargs["jid"], data_type
-            )
 
+        job_collection_name = self.get_collection_name(
+            job, data_type, kwargs["sid"], kwargs["jid"]
+        )
         count = spiderdata_db_client.delete_collection_data(
             kwargs["pid"], job_collection_name
         )

estela-api/api/views/project.py

@@ -1,7 +1,7 @@
 from datetime import datetime, timedelta
-from django.shortcuts import get_object_or_404
 
 from django.core.paginator import Paginator
+from django.shortcuts import get_object_or_404
 from drf_yasg import openapi
 from drf_yasg.utils import swagger_auto_schema
 from rest_framework import status, viewsets
@@ -13,6 +13,7 @@
 from api.mixins import BaseViewSet, ActionHandlerMixin
 from api.serializers.job import ProjectJobSerializer, SpiderJobSerializer
 from api.serializers.cronjob import ProjectCronJobSerializer, SpiderCronJobSerializer
+from api.serializers.job import ProjectJobSerializer, SpiderJobSerializer
 from api.serializers.project import (
     ProjectSerializer,
     ProjectUpdateSerializer,

estela-api/build_project/build.py

@@ -1,16 +1,14 @@
-import os
-import sys
-import json
-import docker
 import base64
-import requests
+import json
 import logging
-
+import os
+import sys
 from zipfile import ZipFile
 
+import docker
+import requests
 from django.conf import settings
 
-
 sys.path.append("/home/estela/estela-api")
 os.environ["DJANGO_SETTINGS_MODULE"] = "config.settings.base"
 

estela-api/config/job_manager.py

@@ -1,7 +1,8 @@
-from engines.config import JobManager
-from credentials.config import Credentials
-from django.conf import settings
 from database_adapters.db_adapters import get_database_interface
+from django.conf import settings
+
+from credentials.config import Credentials
+from engines.config import JobManager
 
 credentials = Credentials(plataform=settings.CREDENTIALS)
 

estela-api/config/settings/base.py

@@ -10,9 +10,10 @@
 https://docs.djangoproject.com/en/3.1/ref/settings/
 """
 
+from pathlib import Path
 from urllib.parse import urlparse
+
 import environ
-from pathlib import Path
 
 from estela_queue_adapter import get_queue_env_vars
 
@@ -38,6 +39,8 @@
     DJANGO_EXTERNAL_APPS=(str, ""),
     EXTERNAL_APP_KEYS=(str, "dummy"),
     EXTERNAL_MIDDLEWARES=(str, ""),
+    REDIS_URL=(str, "redis://redis"),
+    REDIS_STATS_INTERVAL=(str, "5.0"),
     CORS_ORIGIN_WHITELIST=(str, "http://127.0.0.1:3000"),
     AWS_ACCESS_KEY_ID=(str, "dummy"),
     AWS_SECRET_ACCESS_KEY=(str, "dummy"),
@@ -232,6 +235,11 @@
 # Queue settings
 QUEUE_PARAMS = get_queue_env_vars()
 
+# Redis settings for job stats
+
+REDIS_URL = env("REDIS_URL")
+REDIS_STATS_INTERVAL = env("REDIS_STATS_INTERVAL")
+
 
 # Cluster settings
 MULTI_NODE_MODE = False

estela-api/config/storage_backends.py

@@ -1,4 +1,5 @@
 from storages.backends.s3boto3 import S3Boto3Storage
+
 from config.settings import prod