Health monitoring changes

.gitignore

@@ -8,4 +8,7 @@ __pycache__
 *scratch*
 *.vscode
 */private/*
-*test_consumer_payload.py
+*test_consumer_payload.py
+*.env
+*.bak
+env.*

env.example

@@ -0,0 +1,42 @@
+# rag.env.example
+# Copy this file to ~/rag.env and fill in production values.
+# Used by rag_service/scripts/server_setup.py
+# Format: plain KEY=VALUE — no export, no quotes unless value contains spaces.
+
+# Frappe site name
+SITE_NAME=rag.dev
+
+# tap_lms server base URL — rag_service calls back here for assignment/student context
+TAP_LMS_BASE_URL=http://tap_lms.localhost/
+
+# API key/secret for rag_service ↔ tap_lms authentication
+# Must match the credentials configured on the tap_lms side
+RAG_API_KEY=<your_api_key>
+RAG_API_SECRET=<your_api_secret>
+
+# RAG Settings endpoints (defaults are usually correct — only change if customised)
+# RAG_ASSIGNMENT_CONTEXT_ENDPOINT=api/method/tap_lms.imgana.submission.get_assignment_context
+# RAG_STUDENT_CONTEXT_ENDPOINT=api/method/tap_lms.imgana.submission.get_student_details
+# RAG_ENABLE_CACHING=0
+
+# CloudAMQP / RabbitMQ
+RABBITMQ_HOST=<cloudamqp_host>
+RABBITMQ_PORT=5672
+RABBITMQ_VIRTUAL_HOST=<vhost>
+RABBITMQ_USERNAME=<username>
+RABBITMQ_PASSWORD=<password>
+RABBITMQ_PLAGIARISM_RESULTS_QUEUE=<queue_rag_consumes_from>
+RABBITMQ_FEEDBACK_RESULTS_QUEUE=<queue_rag_publishes_results_to>
+
+# GCS (Google Cloud Storage)
+GCS_PROJECT_ID=<gcp_project_id>
+GCS_CREDENTIALS_JSON_FILE=~/tap/GCS Settings.json
+
+# LLM — Gemini via Vertex AI
+LLM_PROVIDER=Gemini
+LLM_MODEL_NAME=gemini-2.5-flash-lite
+LLM_TEMPERATURE=1
+LLM_MAX_TOKENS=1500
+LLM_LOCATION=us-central1
+LLM_PROJECT_ID=<gcp_project_id>
+LLM_CREDENTIALS_JSON_FILE=~/tap/llm-settings-gemini-2.5-flash-lite.json

ops_agent_config.yaml

@@ -0,0 +1,177 @@
+# /etc/google-cloud-ops-agent/config.yaml
+#
+# GCP Ops Agent configuration for TAP LMS server.
+#
+# Covers:
+#   1. Structured JSON logs from tap_lms (gcp_structured.log)
+#   2. Structured JSON logs from rag_service (if co-located)
+#   3. Frappe/nginx/supervisor system logs
+#   4. Hardware metrics (CPU, memory, disk, network) — built-in, no config needed
+#
+# After editing, restart the agent:
+#   sudo systemctl restart google-cloud-ops-agent
+#
+# Verify logs are arriving:
+#   gcloud logging read 'logName=~"tap_lms_structured"' --limit=5 --project=<PROJECT_ID>
+#
+# ── IMPORTANT: severity field mapping ─────────────────────────────────────────
+# Our monitoring.py emits `"severity": "ERROR"` as a plain JSON field.
+# The Ops Agent only promotes a field to LogEntry.severity if it is named
+# `logging.googleapis.com/severity`. The modify_fields processor below
+# renames our `severity` field so Cloud Logging displays the correct
+# severity level and colour-coding in Log Explorer, and so log-based
+# alert policies can filter on `severity=ERROR` directly.
+#
+# Similarly, our `timestamp` field is promoted to LogEntry.timestamp via
+# the parse_json time_key setting.
+# ─────────────────────────────────────────────────────────────────────────────
+
+logging:
+  receivers:
+
+    # ── tap_lms structured log ─────────────────────────────────────────────
+    tap_lms_structured:
+      type: files
+      include_paths:
+        - /home/*/frappe-bench/logs/gcp_structured.log
+      # wildcard_refresh_interval handles log rotation — after copytruncate,
+      # a new file appears and the agent picks it up within this interval.
+      wildcard_refresh_interval: 30s
+
+    # ── rag_service structured log (if co-located on same VM) ─────────────
+    rag_service_structured:
+      type: files
+      include_paths:
+        - /home/*/frappe-bench/logs/rag_gcp_structured.log
+      wildcard_refresh_interval: 30s
+
+    # ── Frappe application log ─────────────────────────────────────────────
+    frappe_app:
+      type: files
+      include_paths:
+        - /home/*/frappe-bench/logs/frappe.log
+        - /home/*/frappe-bench/logs/worker.log
+      wildcard_refresh_interval: 30s
+
+    # ── nginx access + error logs ──────────────────────────────────────────
+    nginx_access:
+      type: files
+      include_paths:
+        - /var/log/nginx/access.log
+        - /home/*/frappe-bench/logs/access.log
+
+    nginx_error:
+      type: files
+      include_paths:
+        - /var/log/nginx/error.log
+
+    # ── Feedback consumer log ──────────────────────────────────────────────
+    feedback_consumer:
+      type: files
+      include_paths:
+        - /home/*/frappe-bench/logs/feedback-consumer.log
+        - /home/*/frappe-bench/logs/feedback-consumer.error.log
+      wildcard_refresh_interval: 30s
+
+    # ── Supervisor logs ────────────────────────────────────────────────────
+    supervisor:
+      type: files
+      include_paths:
+        - /var/log/supervisor/supervisord.log
+
+  processors:
+
+    # ── Parse tap_lms structured JSON ─────────────────────────────────────
+    # Parses each line of gcp_structured.log as JSON and extracts the
+    # timestamp field so it becomes LogEntry.timestamp in Cloud Logging.
+    # Format matches monitoring.py output: 2026-07-31T10:24:40.313895
+    # Note: Fluent Bit uses strptime — %L is milliseconds (3 digits),
+    # %f is NOT supported. Use %s.%N for full nanosecond precision,
+    # or omit time_key entirely and let Cloud Logging use ingestion time.
+    # The safest cross-version approach is to let parse_json handle the
+    # JSON structure and use modify_fields to map the timestamp separately.
+    parse_tap_lms_json:
+      type: parse_json
+      time_key: timestamp
+      time_format: "%Y-%m-%dT%H:%M:%S.%f"
+
+    # ── Promote severity field to LogEntry.severity ────────────────────────
+    # monitoring.py emits `"severity": "ERROR"` as a plain JSON field.
+    # This processor moves it to `logging.googleapis.com/severity` so
+    # Cloud Logging promotes it to the top-level severity field, enabling:
+    #   - Correct colour-coding in Log Explorer
+    #   - Filtering with `severity=ERROR` in alert policies
+    #   - The severity histogram in the Logs Explorer sidebar
+    #
+    # Also maps the timestamp field. Fluent Bit's strptime does not support
+    # %f (microseconds) so we move the raw timestamp string into the special
+    # logging.googleapis.com field and let Cloud Logging parse it — it
+    # handles ISO8601 with microseconds natively.
+    promote_severity:
+      type: modify_fields
+      fields:
+        severity:
+          move_from: jsonPayload.severity
+
+    # ── Exclude high-volume INFO http_request logs from storage ───────────
+    # http_request logs are emitted for every API call (needed for P95
+    # latency and error rate metrics) but don't need to be stored long-term.
+    # This reduces Cloud Logging ingestion costs without losing error coverage.
+    # Remove or comment out this processor if you want full HTTP logs stored.
+    exclude_http_info:
+      type: exclude_logs
+      match_any:
+        - 'jsonPayload.message = "http_request" AND jsonPayload.severity = "INFO"'
+
+  service:
+    pipelines:
+
+      # tap_lms structured logs pipeline
+      tap_lms_pipeline:
+        receivers: [tap_lms_structured]
+        processors: [parse_tap_lms_json, promote_severity, exclude_http_info]
+
+      # rag_service structured logs pipeline
+      rag_service_pipeline:
+        receivers: [rag_service_structured]
+        processors: [parse_tap_lms_json, promote_severity]
+
+      # Frappe application logs (plain text — no JSON parsing)
+      frappe_pipeline:
+        receivers: [frappe_app]
+
+      # nginx logs
+      nginx_pipeline:
+        receivers: [nginx_access, nginx_error]
+
+      # Feedback consumer logs
+      feedback_consumer_pipeline:
+        receivers: [feedback_consumer]
+
+      # Supervisor logs
+      supervisor_pipeline:
+        receivers: [supervisor]
+
+# ── Hardware metrics ───────────────────────────────────────────────────────────
+# The Ops Agent collects CPU, memory, disk, network, and process metrics
+# automatically via the built-in hostmetrics receiver. No additional
+# configuration is needed — these are already active by default.
+#
+# Metrics are visible in:
+#   GCP Console → Monitoring → Metrics Explorer
+#   Filter: resource.type = "gce_instance"
+#   Metric prefix: agent.googleapis.com/
+#
+# Key metrics available out of the box:
+#   agent.googleapis.com/cpu/utilization
+#   agent.googleapis.com/memory/usage (state: used/free/cached/buffered)
+#   agent.googleapis.com/disk/usage (device: sda1 etc.)
+#   agent.googleapis.com/network/tcp_connections
+#   agent.googleapis.com/processes/count
+#
+# To increase collection frequency (default is 60s), uncomment below:
+# metrics:
+#   receivers:
+#     hostmetrics:
+#       type: hostmetrics
+#       collection_interval: 30s

rag_service/core/assignment_context_manager.py

@@ -6,6 +6,8 @@
 from datetime import datetime, timedelta
 from typing import Any, Dict
 from frappe.utils import now_datetime
+import time
+from ..monitoring import record_tap_lms_api_call
 
 class AssignmentContextManager:
     def __init__(self):
@@ -117,13 +119,34 @@ async def _fetch_assignment_from_api(self, assignment_id: str) -> Dict:
             payload = {
                 "assignment_id": assignment_id
             }
-            response = requests.post(
-                api_url,
-                headers=self.headers,
-                json=payload,
-                timeout=30
+            _api_t0 = time.monotonic()
+            _api_status = None
+            try:
+                response = requests.post(
+                    api_url,
+                    headers=self.headers,
+                    json=payload,
+                    timeout=30
+                )
+                _api_status = response.status_code
+            except Exception as _conn_err:
+                record_tap_lms_api_call(
+                    submission_id=None,
+                    endpoint="get_assignment_context",
+                    duration_ms=(time.monotonic() - _api_t0) * 1000,
+                    cache_hit=False,
+                    status_code=None,
+                )
+                raise
+
+            record_tap_lms_api_call(
+                submission_id=None,
+                endpoint="get_assignment_context",
+                duration_ms=(time.monotonic() - _api_t0) * 1000,
+                cache_hit=False,
+                status_code=_api_status,
             )
-            
+
             if response.status_code != 200:
                 error_msg = f"API request failed with status {response.status_code}: {response.text}"
                 print(f"Error: {error_msg}")
@@ -152,13 +175,34 @@ async def _fetch_student_from_api(self, student_id: str) -> Dict:
             payload = {
                 "student_id": student_id
             }
-            response = requests.post(
-                api_url,
-                headers=self.headers,
-                json=payload,
-                timeout=30
+            _api_t0 = time.monotonic()
+            _api_status = None
+            try:
+                response = requests.post(
+                    api_url,
+                    headers=self.headers,
+                    json=payload,
+                    timeout=30
+                )
+                _api_status = response.status_code
+            except Exception as _conn_err:
+                record_tap_lms_api_call(
+                    submission_id=None,
+                    endpoint="get_student_context",
+                    duration_ms=(time.monotonic() - _api_t0) * 1000,
+                    cache_hit=False,
+                    status_code=None,
+                )
+                raise
+
+            record_tap_lms_api_call(
+                submission_id=None,
+                endpoint="get_student_context",
+                duration_ms=(time.monotonic() - _api_t0) * 1000,
+                cache_hit=False,
+                status_code=_api_status,
             )
-            
+
             if response.status_code != 200:
                 error_msg = f"API request failed with status {response.status_code}: {response.text}"
                 print(f"Error: {error_msg}")

rag_service/core/feedback_handler.py

@@ -1,6 +1,12 @@
 # rag_service/rag_service/core/feedback_handler.py
 
+import time
 import frappe
+from ..monitoring import (
+    record_rag_submission_received,
+    record_rag_feedback_complete,
+    record_rag_feedback_failed,
+)
 import json
 from datetime import datetime
 from typing import Dict, Optional
@@ -18,6 +24,18 @@ def __init__(self):
     async def handle_submission(self, message_data: Dict) -> None:
         """Handle a new submission from plagiarism queue"""
         request_id = None
+        _t0 = time.monotonic()
+        submission_id = message_data.get("submission_id")
+        student_id = message_data.get("student_id")
+        assignment_id = message_data.get("assignment_id")
+
+        # SRE: pipeline trace — step 1 in rag_service
+        record_rag_submission_received(
+            submission_id=submission_id,
+            student_id=student_id,
+            assignment_id=assignment_id,
+        )
+
         try:
             submission_data = normalize_submission_payload(message_data)
 
@@ -47,6 +65,15 @@ async def handle_submission(self, message_data: Dict) -> None:
             error_msg = f"Error handling submission: {str(e)}"
             print(f"\nError: {error_msg}")
             frappe.log_error(error_msg, "Submission Handler Error")
+            # SRE: pipeline trace — failure
+            try:
+                record_rag_feedback_failed(
+                    submission_id=submission_id,
+                    error=error_msg,
+                    duration_ms=(time.monotonic() - _t0) * 1000,
+                )
+            except Exception:
+                pass
 
             # Mark request as failed if it exists
             if request_id and frappe.db.exists("Feedback Request", request_id):
@@ -62,6 +89,16 @@ async def handle_submission(self, message_data: Dict) -> None:
             # Process and deliver feedback
             await self.feedback_service.process_feedback(request_id, feedback, model_used, template_used)
             print("\nFeedback processing completed")
+            # SRE: pipeline trace — complete (always emitted, success or error)
+            try:
+                record_rag_feedback_complete(
+                    submission_id=submission_id,
+                    model_used=model_used,
+                    template_used=template_used,
+                    duration_ms=(time.monotonic() - _t0) * 1000,
+                )
+            except Exception:
+                pass
 
     def _attach_plagiarism_defaults(self, feedback: Dict) -> Dict:
         feedback["plagiarism_output"] = {