Add status logging and endpoint to access status

Author: azvoleff

gefapi/celery.py

@@ -22,6 +22,16 @@ def make_celery(app):
         broker=app.config["broker_url"],
     )
     celery.conf.update(app.config)
+
+    # Configure periodic tasks
+    celery.conf.beat_schedule = {
+        "collect-system-status": {
+            "task": "gefapi.tasks.status_monitoring.collect_system_status",
+            "schedule": 120.0,  # Every 2 minutes (120 seconds)
+        },
+    }
+    celery.conf.timezone = "UTC"
+
     TaskBase = celery.Task
 
     class ContextTask(TaskBase):

gefapi/models/__init__.py

@@ -54,6 +54,7 @@ def process_result_value(self, value, dialect):
 from gefapi.models.execution_log import ExecutionLog  # noqa: E402
 from gefapi.models.script import Script  # noqa: E402
 from gefapi.models.script_log import ScriptLog  # noqa: E402
+from gefapi.models.status_log import StatusLog  # noqa: E402
 from gefapi.models.user import User  # noqa: E402
 
-__all__ = ["Execution", "ExecutionLog", "Script", "ScriptLog", "User"]
+__all__ = ["Execution", "ExecutionLog", "Script", "ScriptLog", "StatusLog", "User"]

gefapi/models/status_log.py

@@ -0,0 +1,70 @@
+"""STATUS LOG MODEL"""
+
+from __future__ import absolute_import, division, print_function
+
+import datetime
+
+from gefapi import db
+from gefapi.models import GUID
+
+db.GUID = GUID
+
+
+class StatusLog(db.Model):
+    """StatusLog Model"""
+
+    __tablename__ = "status_log"
+    id = db.Column(db.Integer(), primary_key=True)
+    timestamp = db.Column(db.DateTime(), default=datetime.datetime.utcnow)
+
+    # Execution counts
+    executions_active = db.Column(db.Integer(), default=0)
+    executions_ready = db.Column(db.Integer(), default=0)
+    executions_running = db.Column(db.Integer(), default=0)
+    executions_finished = db.Column(db.Integer(), default=0)
+
+    # Other counts
+    users_count = db.Column(db.Integer(), default=0)
+    scripts_count = db.Column(db.Integer(), default=0)
+
+    # System metrics
+    memory_available_percent = db.Column(db.Float(), default=0.0)
+    cpu_usage_percent = db.Column(db.Float(), default=0.0)
+
+    def __init__(
+        self,
+        executions_active=0,
+        executions_ready=0,
+        executions_running=0,
+        executions_finished=0,
+        users_count=0,
+        scripts_count=0,
+        memory_available_percent=0.0,
+        cpu_usage_percent=0.0,
+    ):
+        self.executions_active = executions_active
+        self.executions_ready = executions_ready
+        self.executions_running = executions_running
+        self.executions_finished = executions_finished
+        self.users_count = users_count
+        self.scripts_count = scripts_count
+        self.memory_available_percent = memory_available_percent
+        self.cpu_usage_percent = cpu_usage_percent
+
+    def __repr__(self):
+        return "<StatusLog %r>" % self.id
+
+    def serialize(self):
+        """Return object data in easily serializeable format"""
+        return {
+            "id": self.id,
+            "timestamp": self.timestamp.isoformat(),
+            "executions_active": self.executions_active,
+            "executions_ready": self.executions_ready,
+            "executions_running": self.executions_running,
+            "executions_finished": self.executions_finished,
+            "users_count": self.users_count,
+            "scripts_count": self.scripts_count,
+            "memory_available_percent": self.memory_available_percent,
+            "cpu_usage_percent": self.cpu_usage_percent,
+        }

gefapi/routes/api/v1/gef_api_router.py

@@ -21,7 +21,7 @@
 )
 from gefapi.routes.api.v1 import endpoints, error
 from gefapi.s3 import get_script_from_s3
-from gefapi.services import ExecutionService, ScriptService, UserService
+from gefapi.services import ExecutionService, ScriptService, StatusService, UserService
 from gefapi.validators import (
     validate_execution_log_creation,
     validate_execution_update,
@@ -602,3 +602,58 @@ def delete_user(user):
         logger.error("[ROUTER]: " + str(e))
         return error(status=500, detail="Generic Error")
     return jsonify(data=user.serialize()), 200
+
+
+@endpoints.route("/status", strict_slashes=False, methods=["GET"])
+@jwt_required()
+def get_status_logs():
+    """Get system status logs (Admin only)"""
+    logger.info("[ROUTER]: Getting status logs")
+
+    # Check if user is admin
+    identity = current_user
+    if identity.role != "ADMIN" and identity.email != "gef@gef.com":
+        return error(status=403, detail="Forbidden")
+
+    # Parse date filters
+    start_date = request.args.get("start_date", None)
+    if start_date:
+        start_date = dateutil.parser.parse(start_date)
+
+    end_date = request.args.get("end_date", None)
+    if end_date:
+        end_date = dateutil.parser.parse(end_date)
+
+    # Parse sorting
+    sort = request.args.get("sort", None)
+
+    # Parse pagination
+    try:
+        page = int(request.args.get("page", 1))
+        per_page = int(request.args.get("per_page", 100))
+        page = max(page, 1)
+        per_page = min(max(per_page, 1), 1000)
+    except Exception:
+        page, per_page = 1, 100
+
+    try:
+        status_logs, total = StatusService.get_status_logs(
+            start_date=start_date,
+            end_date=end_date,
+            sort=sort,
+            page=page,
+            per_page=per_page,
+        )
+    except Exception as e:
+        logger.error("[ROUTER]: " + str(e))
+        return error(status=500, detail="Generic Error")
+
+    return (
+        jsonify(
+            data=[status_log.serialize() for status_log in status_logs],
+            page=page,
+            per_page=per_page,
+            total=total,
+        ),
+        200,
+    )

gefapi/services/__init__.py

@@ -24,6 +24,7 @@ def handle_exception(exc_type, exc_value, exc_traceback):
 )
 from gefapi.services.email_service import EmailService  # noqa:E402
 from gefapi.services.script_service import ScriptService  # noqa:E402
+from gefapi.services.status_service import StatusService  # noqa:E402
 from gefapi.services.user_service import UserService  # noqa:E402
 
 # Import last to avoid circular dependency
@@ -35,6 +36,7 @@ def handle_exception(exc_type, exc_value, exc_traceback):
     "docker_run",
     "EmailService",
     "ScriptService",
+    "StatusService",
     "UserService",
     "ExecutionService",
 ]

gefapi/services/status_service.py

@@ -0,0 +1,58 @@
+"""STATUS SERVICE"""
+
+from __future__ import absolute_import, division, print_function
+
+import logging
+
+import rollbar
+
+from gefapi import db
+from gefapi.models import StatusLog
+
+logger = logging.getLogger()
+
+
+class StatusService(object):
+    """Status Service Class"""
+
+    @staticmethod
+    def get_status_logs(
+        start_date=None,
+        end_date=None,
+        sort=None,
+        page=1,
+        per_page=100,
+    ):
+        """Get status logs with optional filtering and pagination"""
+        logger.info("[SERVICE]: Getting status logs")
+        logger.info("[DB]: QUERY")
+
+        if page < 1:
+            page = 1
+        if per_page < 1:
+            per_page = 100
+
+        query = db.session.query(StatusLog)
+
+        # Apply date filters
+        if start_date:
+            query = query.filter(StatusLog.timestamp >= start_date)
+        if end_date:
+            query = query.filter(StatusLog.timestamp <= end_date)
+
+        # Apply sorting
+        if sort:
+            sort_field = sort[1:] if sort.startswith("-") else sort
+            sort_direction = "desc" if sort.startswith("-") else "asc"
+            if hasattr(StatusLog, sort_field):
+                query = query.order_by(
+                    getattr(getattr(StatusLog, sort_field), sort_direction)()
+                )
+        else:
+            # Default to sorting by timestamp descending
+            query = query.order_by(StatusLog.timestamp.desc())
+
+        total = query.count()
+        status_logs = query.offset((page - 1) * per_page).limit(per_page).all()
+
+        return status_logs, total

gefapi/tasks/__init__.py

@@ -0,0 +1,3 @@
+"""TASKS MODULE"""
+
+from __future__ import absolute_import, division, print_function

gefapi/tasks/status_monitoring.py

@@ -0,0 +1,77 @@
+"""STATUS MONITORING TASKS"""
+
+from __future__ import absolute_import, division, print_function
+
+import logging
+
+import psutil
+import rollbar
+from celery import Task
+from sqlalchemy import func
+
+from gefapi import celery, db
+from gefapi.models import Execution, Script, StatusLog, User
+
+logger = logging.getLogger()
+
+
+class StatusMonitoringTask(Task):
+    """Base task for status monitoring"""
+
+    def on_failure(self, exc, task_id, args, kwargs, einfo):
+        logger.error(f"Status monitoring task failed: {exc}")
+        rollbar.report_exc_info()
+
+
+@celery.task(base=StatusMonitoringTask, bind=True)
+def collect_system_status(self):
+    """Collect system status and save to status_log table"""
+    logger.info("[TASK]: Collecting system status")
+
+    try:
+        # Count executions by status
+        execution_counts = (
+            db.session.query(Execution.status, func.count(Execution.id))
+            .group_by(Execution.status)
+            .all()
+        )
+
+        execution_status_map = dict(execution_counts)
+        executions_active = execution_status_map.get("ACTIVE", 0)
+        executions_ready = execution_status_map.get("READY", 0)
+        executions_running = execution_status_map.get("RUNNING", 0)
+        executions_finished = execution_status_map.get("FINISHED", 0)
+
+        # Count users and scripts
+        users_count = db.session.query(func.count(User.id)).scalar()
+        scripts_count = db.session.query(func.count(Script.id)).scalar()
+
+        # Get system metrics
+        memory = psutil.virtual_memory()
+        memory_available_percent = memory.available / memory.total * 100
+        cpu_usage_percent = psutil.cpu_percent(interval=1)
+
+        # Create status log entry
+        status_log = StatusLog(
+            executions_active=executions_active,
+            executions_ready=executions_ready,
+            executions_running=executions_running,
+            executions_finished=executions_finished,
+            users_count=users_count,
+            scripts_count=scripts_count,
+            memory_available_percent=memory_available_percent,
+            cpu_usage_percent=cpu_usage_percent,
+        )
+
+        logger.info("[DB]: ADD")
+        db.session.add(status_log)
+        db.session.commit()
+
+        logger.info(f"[TASK]: Status log created with ID {status_log.id}")
+        return status_log.serialize()
+
+    except Exception as error:
+        logger.error(f"[TASK]: Error collecting system status: {str(error)}")
+        rollbar.report_exc_info()
+        db.session.rollback()
+        raise error

requirements.txt

@@ -17,3 +17,4 @@ redis==5.0.8
 sparkpost==1.3.10
 rollbar==1.1.2
 blinker==1.8.2
+psutil>=5.8.0