Modifying fastmon agent to use most recent version of DB schema in swf-monitor, and refactoring to use same workflow for continuous and message-driven mode

Author: michmx

CLAUDE.md

@@ -28,6 +28,31 @@ The project is designed to work with PostgreSQL databases and ActiveMQ messaging
 - **Code Formatter**: Black
 - **License**: Apache 2.0
 - **Environment Variable**: `SWF_HOME` automatically set to parent directory containing all swf-* repos (via swf-testbed CLI)
+- **Architecture**: Extends BaseAgent from swf-common-lib for standardized agent behavior
+
+## Recent Infrastructure Updates (2025-11)
+
+### BaseAgent Integration
+The agent now inherits from **BaseAgent** (swf-common-lib) providing:
+- Automatic environment setup and .env loading
+- REST logging to swf-monitor
+- Sequential agent ID generation
+- Enhanced heartbeat with workflow metadata
+- Automatic subscriber registration
+- Connection resilience with auto-reconnection
+
+### Workflow Tracking
+Integrated with swf-monitor's workflow tracking system:
+- Creates workflow stages via `/api/workflow-stages/`
+- Tracks statuses: `fastmon_received`, `fastmon_processing`, `fastmon_complete`
+- Records input/output messages and processing times
+- Enables end-to-end workflow visibility
+
+### MQ Communications
+Updated to use swf-common-lib's mq_comms module:
+- Requires `client_id` parameter for durable subscriptions
+- SSL support with `MQ_CAFILE` environment variable
+- Standardized error handling and reconnection logic
 
 ## Project Structure
 
@@ -115,8 +140,44 @@ This project integrates with:
 - **mypy**: Static type checking (>=1.0.0)
 - **django-stubs**: Django type stubs (>=1.13.0)
 
-### Database Environment Variables
-Django settings support standard environment variables:
+### Environment Variables Configuration
+
+The agent requires a `.env` file with the following variables:
+
+**Monitor Connection:**
+- `SWF_MONITOR_URL` - HTTPS URL for authenticated API calls (required)
+- `SWF_MONITOR_HTTP_URL` - HTTP URL for REST logging (optional)
+- `SWF_API_TOKEN` - Authentication token for swf-monitor API (required)
+
+**ActiveMQ Configuration:**
+- `ACTIVEMQ_HOST` - ActiveMQ broker host (default: localhost)
+- `ACTIVEMQ_PORT` - STOMP port (default: 61612)
+- `ACTIVEMQ_USER` - ActiveMQ username (required)
+- `ACTIVEMQ_PASSWORD` - ActiveMQ password (required)
+- `ACTIVEMQ_USE_SSL` - Enable SSL connections (true/false)
+- `ACTIVEMQ_SSL_CA_CERTS` - Path to CA certificate file
+
+**MQ Communications (swf-common-lib):**
+- `MQ_USER` - Message queue username (required)
+- `MQ_PASSWD` - Message queue password (required)
+- `MQ_HOST` - Message queue host (required)
+- `MQ_PORT` - Message queue port (required)
+- `MQ_CAFILE` - SSL CA certificate path (required for SSL)
+
+**Logging:**
+- `SWF_LOG_LEVEL` - Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
+- `SWF_STOMP_DEBUG` - Enable STOMP protocol debugging (true/false)
+- `SWF_AGENT_QUIET` - Minimal output mode (true/false)
+
+**Agent Configuration:**
+- `FASTMON_MODE` - Operation mode: `message` (default) or `continuous`
+- `FASTMON_SELECTION_FRACTION` - STF sampling fraction (0.0-1.0, default: 0.1)
+- `FASTMON_TF_FILES_PER_STF` - TF files per STF (default: 7)
+
+See `.env.example` for a complete template with all available options.
+
+### Database Environment Variables (Django Legacy)
+Legacy Django settings (if needed for local development):
 - `POSTGRES_HOST` (default: localhost)
 - `POSTGRES_PORT` (default: 5432)
 - `POSTGRES_DB` (default: epic_monitoring)
@@ -129,17 +190,71 @@ Django settings support standard environment variables:
 - Database credentials (`.pgpass`) are excluded from version control
 - Log files are excluded from commits
 
+## API Integration
+
+The agent integrates with swf-monitor REST API endpoints:
+
+### Core Endpoints Used
+- `POST /api/runs/` - Create/retrieve run records
+- `POST /api/stf-files/` - Register STF files (development mode only)
+- `POST /api/fastmon-files/` - Register TF files (primary endpoint)
+- `POST /api/workflow-stages/` - Create workflow stage tracking
+- `PATCH /api/workflow-stages/{id}/` - Update stage status and timestamps
+- `POST /api/subscribers/` - Auto-register as ActiveMQ subscriber (via BaseAgent)
+
+### FastMonFile API Schema
+```json
+{
+  "stf_file": "parent_stf_filename.stf",
+  "tf_filename": "tf_001.tf",
+  "file_size_bytes": 1234567,
+  "status": "registered",
+  "metadata": {
+    "simulation": true,
+    "created_from": "stf_filename.stf",
+    "agent_name": "swf-fastmon-agent-1"
+  }
+}
+```
+
+### Workflow Stage Tracking
+The agent creates and updates workflow stages for each STF processed:
+```python
+# Create stage
+stage_data = {
+    'workflow': workflow_id,
+    'agent_name': 'swf-fastmon-agent-1',
+    'agent_type': 'fastmon',
+    'status': 'fastmon_received',
+    'input_message': {...}
+}
+
+# Update during processing
+{'status': 'fastmon_processing', 'started_at': '2025-11-19T10:30:00Z'}
+
+# Mark complete
+{'status': 'fastmon_complete', 'completed_at': '2025-11-19T10:30:15Z', 'output_message': {...}}
+```
+
 ## Development Commands
 
 ### System Initialization
 ```bash
 cd $SWF_PARENT_DIR/swf-testbed
 source .venv/bin/activate  # or conda activate your_env_name
 pip install -e $SWF_PARENT_DIR/swf-common-lib $SWF_PARENT_DIR/swf-monitor $SWF_PARENT_DIR/swf-fastmon-agent .
-# CRITICAL: Set up Django environment
+
+# CRITICAL: Set up environment configuration
+cd $SWF_PARENT_DIR/swf-fastmon-agent
+cp .env.example .env
+# Edit .env with actual values for SWF_MONITOR_URL, SWF_API_TOKEN, ActiveMQ credentials, etc.
+
+# Set up Django environment (swf-monitor)
 cp $SWF_PARENT_DIR/swf-monitor/.env.example $SWF_PARENT_DIR/swf-monitor/.env
 # Edit .env to set DB_PASSWORD='your_db_password' and SECRET_KEY
 cd $SWF_PARENT_DIR/swf-monitor/src && python manage.py migrate
+
+# Initialize testbed
 cd $SWF_PARENT_DIR/swf-testbed && swf-testbed init
 ```
 

pyproject.toml

@@ -14,9 +14,9 @@ dependencies = [
     "psycopg2-binary>=2.9.0",
     "typer>=0.9.0",
     "stomp.py>=8.1.0",
-    # This is a dependency from our other local package
-    #"swf-common-lib",
-    #"supervisor"
+    "requests>=2.31.0",
+    "python-json-logger>=2.0.0",
+    "swf-common-lib",
 ]
 [project.optional-dependencies]
 dev = [

requirements.txt

@@ -5,6 +5,12 @@ Django>=4.2,<5.0
 psycopg>=3.2.0
 psycopg2-binary>=2.9.0
 
+# HTTP and API
+requests>=2.31.0
+
+# Logging
+python-json-logger>=2.0.0
+
 # Development and testing
 pytest>=7.0.0
 pytest-django>=4.5.0
@@ -13,4 +19,4 @@ flake8>=4.0.0
 
 # CLI and messaging
 typer>=0.9.0
-stomp.py>=8.1.0
+stomp.py>=8.1.0

src/swf_fastmon_agent/fastmon_utils.py

@@ -311,7 +311,8 @@ def simulate_tf_subsamples(stf_file: Dict[str, Any], config: dict, logger: loggi
                 "tf_filename": tf_filename,
                 "file_size_bytes": tf_size,
                 "sequence_number": sequence_number,
-                "stf_parent": stf_file.get("filename"),  # Use unique filename as parent identifier
+                "stf_file_id": stf_file.get("file_id"),  # UUID for foreign key reference
+                "stf_parent": stf_file.get("filename"),  # Keep filename for reference
                 "metadata": {
                     "simulation": True,
                     "created_from": stf_file.get('filename'),
@@ -349,7 +350,7 @@ def record_tf_file(tf_metadata: Dict[str, Any], config: dict, agent, logger: log
     try:
         # Prepare FastMonFile data for API
         tf_file_data = {
-            "stf_file": tf_metadata.get("stf_parent", None),  # STF filename as parent identifier
+            "stf_file": tf_metadata.get("stf_file_id"),  # UUID foreign key to StfFile
             "tf_filename": tf_metadata["tf_filename"],
             "file_size_bytes": tf_metadata["file_size_bytes"],
             "status": FileStatus.REGISTERED,

src/swf_fastmon_agent/main.py

@@ -10,11 +10,13 @@
 """
 
 import sys
+import os
+import time
 import json
 from datetime import datetime
 
 from swf_common_lib.base_agent import BaseAgent, setup_environment
-import fastmon_utils as fastmon_utils
+from swf_fastmon_agent import fastmon_utils
 
 
 class FastMonitorAgent(BaseAgent):
@@ -42,6 +44,9 @@ def __init__(self, config: dict, debug=False):
 
         self.logger.info("Fast Monitor Agent initialized successfully")
 
+        # Set destination for broadcasting TF file notifications
+        self.destination = os.getenv('ACTIVEMQ_FASTMON_TOPIC', 'epictopic')
+
         self.config = config
 
         # Validate configuration
@@ -51,7 +56,12 @@ def __init__(self, config: dict, debug=False):
         # Fast monitoring specific state
         self.stf_messages_processed = 0
         self.last_message_time = None
-        self.processing_stats = {'total_stf_messages': 0, 'total_tf_files_created': 0}
+        self.files_processed = 0  # For continuous monitoring mode
+        self.processing_stats = {
+            'total_stf_messages': 0,
+            'total_tf_files_created': 0,
+            'total_files': 0  # For continuous monitoring mode
+        }
 
 
     def _emulate_stf_registration_and_sampling(self):
@@ -88,20 +98,27 @@ def _emulate_stf_registration_and_sampling(self):
                 stf_file = fastmon_utils.record_stf_file(file_path, self.config, self, self.logger)
                 self.files_processed += 1
 
-                # Simulate TF subsamples for this STF file
-                tf_subsamples = fastmon_utils.simulate_tf_subsamples(stf_file, file_path, self.config, self.logger)
-
-                # Record each TF file in the FastMonFile table
-                tf_files_created = 0
-                for tf_metadata in tf_subsamples:
-                    tf_file = fastmon_utils.record_tf_file(stf_file, tf_metadata, self.config, self, self.logger)
-                    if tf_file:
-                        tf_files_created += 1
-                        # Send notification to clients about new TF file
-                        self.send_tf_file_notification(tf_file, stf_file)
-                    tf_files_registered.append(tf_file)
+                # Create mock stf_ready message (matching format from data agent)
+                message_data = {
+                    "msg_type": "stf_ready",
+                    "filename": stf_file.get('stf_filename'),
+                    "file_id": stf_file.get('file_id'),  # UUID for foreign key
+                    "run_id": stf_file.get('run'),
+                    "file_url": stf_file.get('metadata', {}).get('file_url', ''),
+                    "checksum": stf_file.get('checksum', ''),
+                    "size_bytes": stf_file.get('file_size_bytes'),
+                    "start": stf_file.get('metadata', {}).get('creation_time', ''),
+                    "end": stf_file.get('metadata', {}).get('modification_time', ''),
+                    "state": "physics",
+                    "substate": "running",
+                    "processed_by": self.agent_name
+                }
+
+                # Use the same sample_timeframes method as message-driven mode
+                tf_files = self.sample_timeframes(message_data)
+                tf_files_registered.extend(tf_files)
 
-                self.logger.info(f"Registered {tf_files_created} TF subsamples for STF file {stf_file['filename']}")
+                self.logger.info(f"Processed STF file {stf_file['stf_filename']} -> {len(tf_files)} TF files")
 
             # Report successful processing
             self.report_agent_status('OK', f'Emulating {len(tf_files_registered)} fast monitoring files')
@@ -173,22 +190,68 @@ def sample_timeframes(self, message_data):
             self.logger.error("No filename provided in message")
             return tf_files_registered
 
+        # Track workflow stage (optional - controlled by FASTMON_TRACK_WORKFLOW env var)
+        workflow_id = message_data.get('workflow_id')
+        stage_id = None
+        track_workflow = os.getenv('FASTMON_TRACK_WORKFLOW', 'false').lower() == 'true'
+
+        if workflow_id and track_workflow:
+            try:
+                # Create workflow stage entry for fast monitoring
+                stage_data = {
+                    'workflow': workflow_id,
+                    'agent_name': self.agent_name,
+                    'agent_type': 'fastmon',
+                    'status': 'fastmon_received',
+                    'input_message': message_data
+                }
+                stage = self.call_monitor_api('POST', '/workflow-stages/', stage_data)
+                stage_id = stage.get('id')
+                self.logger.debug(f"Created workflow stage {stage_id} for workflow {workflow_id}")
+
+                # Update to processing status
+                self.call_monitor_api('PATCH', f'/workflow-stages/{stage_id}/', {
+                    'status': 'fastmon_processing',
+                    'started_at': datetime.now().isoformat()
+                })
+            except Exception as e:
+                self.logger.warning(f"Could not create workflow stage: {e}")
+
+        # Simulate TF subsamples from STF data
         tf_subsamples = fastmon_utils.simulate_tf_subsamples(message_data, self.config, self.logger, self.agent_name)
 
-        # Record each TF file in the FastMonFile table
+        # Record each TF file in the FastMonFile table and send notifications
         # TODO: register in bulk
         tf_files_created = 0
         for tf_metadata in tf_subsamples:
             self.logger.debug(f"Processing {tf_metadata}")
             tf_file = fastmon_utils.record_tf_file(tf_metadata, self.config, self, self.logger)
             if tf_file:
                 tf_files_created += 1
+                # Send notification to clients about new TF file
+                self.send_tf_file_notification(tf_file, message_data)
             tf_files_registered.append(tf_file)
 
         # Update TF creation stats
         self.processing_stats['total_tf_files_created'] += tf_files_created
 
         self.logger.info(f"Registered {tf_files_created} TF subsamples for STF file {message_data.get('filename')}")
+
+        # Mark workflow stage as complete
+        if stage_id:
+            try:
+                output_message = {
+                    'tf_files_created': tf_files_created,
+                    'tf_filenames': [tf.get('tf_filename') for tf in tf_files_registered if tf]
+                }
+                self.call_monitor_api('PATCH', f'/workflow-stages/{stage_id}/', {
+                    'status': 'fastmon_complete',
+                    'completed_at': datetime.now().isoformat(),
+                    'output_message': output_message
+                })
+            except Exception as e:
+                self.logger.warning(f"Could not update workflow stage: {e}")
+
         return tf_files_registered
 
     def start_continuous_monitoring(self):