feat: Implement background analysis and control endpoints for AI Analytics Engine (2399b315) · Commits · TFS / controller

src/tests/mwc26-f5ga/AI_analytics_engine/api/api_blueprint.py

+287 −46

Original line number	Diff line number	Diff line
		@@ -19,10 +19,13 @@ Defines the REST API endpoints for the AI Analytics Engine.
		"""

		import logging
		import threading
		import time
		from datetime import datetime, UTC


		from flask import Blueprint, jsonify, request
		import requests

		from ..config import Config
		from ..ai_model.ai_processor import AIModelProcessor
		@@ -33,6 +36,15 @@ from ..ai_model.sla_policy import SLAPolicyConfig

		LOGGER = logging.getLogger(__name__)

		# Background analysis state - track multiple analyses by simap_id
		_analysis_threads = {} # {simap_id: {'thread': Thread, 'stop_event': Event}}
		_threads_lock = threading.Lock()

		# OSM endpoint configuration
		END_HOST = '10.0.58.25'
		END_PORT = 8084
		BASE_URL = f'http://{END_HOST}:{END_PORT}/osm/aiAnalyticsEvent/v1'


		def create_ai_analytics_blueprint(
		simap_fetcher: SimapDataFetcher,
		@@ -54,24 +66,102 @@ def create_ai_analytics_blueprint(

		Returns:
		Configured Flask Blueprint with routes:
		- POST /api/v1/analyze: Run SLA policy analysis
		- POST /api/v1/analyze: Start background SLA policy analysis
		- POST /api/v1/analyze/stop: Stop analysis for specific SIMAP ID
		- POST /api/v1/analyze/stop-all: Stop all running analyses
		- GET /api/v1/status: Get status of all running analyses
		- GET /api/v1/health: Health check endpoint
		- GET /api/v1/config: Get current configuration
		- POST /api/v1/notify: Handle telemetry update notifications
		"""
		blueprint = Blueprint('ai_analytics', __name__, url_prefix='/api/v1')

		def _background_analysis_task(sla_policy: SLAPolicyConfig, duration_minutes: int, stop_event: threading.Event):
		"""
		Background task that periodically analyzes data and posts results.
		Args:
		sla_policy: SLA policy configuration for analysis.
		duration_minutes: How long to run the analysis (in minutes).
		stop_event: Threading event to signal task termination.
		"""
		simap_id = sla_policy.simap_id

		try:
		LOGGER.info(f"[{simap_id}] Starting background analysis for {duration_minutes} minutes")
		start_time = time.time()
		end_time = start_time + (duration_minutes * 60)
		iteration = 0

		while time.time() < end_time and not stop_event.is_set():
		iteration += 1
		iteration_start = time.time()

		try:
		LOGGER.debug(f"[{simap_id}] Analysis iteration {iteration} - Fetching performance data")

		performance_data = influxdb_fetcher.fetch_performance_data(sla_policy)

		LOGGER.debug(f"[{simap_id}] Analysis iteration {iteration} - Processing with AI models")
		results = ai_processor.process_data(performance_data)

		results['simap_id'] = simap_id
		results['timestamp'] = datetime.now(UTC).isoformat()
		results['iteration'] = iteration

		LOGGER.debug(f"[{simap_id}] Analysis iteration {iteration} - Posting results to {BASE_URL}")
		response = requests.post(
		BASE_URL,
		json = results,
		timeout = 10,
		headers = {'Content-Type': 'application/json'}
		)

		if response.status_code in (200, 201, 202):
		LOGGER.info(f"[{simap_id}] Iteration {iteration}: Results posted successfully (status {response.status_code})")
		else:
		LOGGER.warning(f"[{simap_id}] Iteration {iteration}: POST returned status {response.status_code}: {response.text}")

		except Exception as e:
		LOGGER.error(f"[{simap_id}] Error in analysis iteration {iteration}: {e}")

		# Wait for 30 seconds (accounting for processing time per iteration)
		elapsed = time.time() - iteration_start
		sleep_time = max(0, 30 - elapsed)

		if sleep_time > 0 and time.time() + sleep_time < end_time and not stop_event.is_set():
		LOGGER.debug(f"[{simap_id}] Sleeping for {sleep_time:.1f} seconds until next iteration")
		stop_event.wait(timeout=sleep_time) # Use wait instead of sleep for immediate response
		elif time.time() < end_time:
		# Not enough time for another full iteration cycle, exit gracefully
		LOGGER.debug(f"[{simap_id}] Insufficient time remaining for next iteration, terminating")
		break

		if stop_event.is_set():
		LOGGER.info(f"[{simap_id}] Background analysis stopped after {iteration} iterations")
		else:
		LOGGER.info(f"[{simap_id}] Background analysis completed after {iteration} iterations. Time limit reached.")

		except Exception as e:
		LOGGER.exception(f"[{simap_id}] Fatal error in background analysis task: {e}")

		finally:
		# Clean up thread tracking
		with _threads_lock:
		if simap_id in _analysis_threads:
		del _analysis_threads[simap_id]
		LOGGER.info(f"[{simap_id}] Background analysis task terminated")

		@blueprint.route('/analyze', methods=['POST'])
		def analyze():
		"""
		Run SLA policy analysis.
		Start SLA policy analysis in background.

		Expects JSON payload with SLA policy configuration.
		Orchestrates the full analysis workflow:
		fetch metrics from InfluxDB, process through AI models, and
		send results to Decision Engine.
		Expects JSON payload with SLA policy configuration including duration_minutes.
		Validates input and immediately returns 202 Accepted.
		Analysis runs in background, posting results every 30 seconds for the specified duration.

		Returns:
		JSON response with analysis results or error message.
		JSON response with acceptance confirmation or error message.
		"""
		LOGGER.info("Received analysis request")

		@@ -108,57 +198,208 @@ def create_ai_analytics_blueprint(
		'message': f'Invalid field value: {str(e)}'
		}), 400

		# Execute analysis workflow
		# Extract duration from request
		try:
		# Step 1: Fetch device data from SIMAP
		# (At the moment, leaving it as it is. No more needed, to be removed in future)
		# LOGGER.debug("Step 1: Fetching device data from SIMAP")
		# device_data = simap_fetcher.fetch_device_data(sla_policy)

		# Step 2: Fetch performance data from InfluxDB
		LOGGER.debug(">>> Step 2: Fetching performance data from InfluxDB")
		performance_data = influxdb_fetcher.fetch_performance_data(
		sla_policy
		)
		duration_minutes = int(data.get('duration_minutes', 0))
		if duration_minutes <= 0:
		raise ValueError("duration_minutes must be positive")
		except (TypeError, ValueError) as e:
		LOGGER.error(f"Invalid duration_minutes: {e}")
		return jsonify({
		'status': 'error',
		'message': f'Invalid duration_minutes: {str(e)}'
		}), 400

		# >>> Step 3: Process data through AI models
		LOGGER.debug(">>> Step 3: Processing data through AI models")
		results = ai_processor.process_data(
		performance_data
		# Check if analysis is already running for this simap_id
		with _threads_lock:
		if sla_policy.simap_id in _analysis_threads:
		thread_info = _analysis_threads[sla_policy.simap_id]
		if thread_info['thread'].is_alive():
		LOGGER.warning(f"Analysis request rejected: analysis for SIMAP ID {sla_policy.simap_id} is already running")
		return jsonify({
		'status': 'error',
		'message': f'Analysis for SIMAP ID {sla_policy.simap_id} is already running. Stop it first.'
		}), 409 # Conflict

		# Start background analysis task
		try:
		stop_event = threading.Event()
		analysis_thread = threading.Thread(
		target=_background_analysis_task,
		args=(sla_policy, duration_minutes, stop_event),
		daemon=True,
		name=f"AI-Analysis-Thread-{sla_policy.simap_id}"
		)

		# >>> Step 4: Send results to Decision Engine
		results['simap_id'] = sla_policy.simap_id # Include SIMAP ID in results
		LOGGER.debug(">>> Step 4: Sending results to Decision Engine")
		if not decision_client.send_results(results):
		LOGGER.error("Failed to send results to Decision Engine")
		# Register thread before starting
		with _threads_lock:
		_analysis_threads[sla_policy.simap_id] = {
		'thread': analysis_thread,
		'stop_event': stop_event,
		'start_time': datetime.now(UTC).isoformat(),
		'duration_minutes': duration_minutes
		}

		analysis_thread.start()

		LOGGER.info(f"Background analysis started for SIMAP ID {sla_policy.simap_id}, duration {duration_minutes} minutes")

		# Return immediate confirmation
		return jsonify({
		'status': 'accepted',
		'message': f'Analysis started successfully. Results will be posted every 30 seconds for {duration_minutes} minutes.',
		'simap_id': sla_policy.simap_id,
		'duration_minutes': duration_minutes,
		'endpoint': BASE_URL
		}), 202 # Accepted

		except Exception as e:
		# Clean up on failure
		with _threads_lock:
		if sla_policy.simap_id in _analysis_threads:
		del _analysis_threads[sla_policy.simap_id]
		LOGGER.exception(f"Failed to start background analysis: {e}")
		return jsonify({
		'status': 'error',
		'message': 'Failed to send results to Decision Engine'
		'message': f'Failed to start analysis: {str(e)}'
		}), 500

		LOGGER.info("Analysis completed successfully")
		@blueprint.route('/analyze/stop', methods=['POST'])
		def stop_analyze():
		"""
		Stop running analysis for a specific SIMAP ID.

		Expects JSON payload with simap_id.
		Signals the background thread to stop gracefully.

		Returns:
		JSON response with stop confirmation or error message.
		"""
		LOGGER.info("Received stop analysis request")

		# Parse and validate request JSON
		try:
		data = request.get_json()
		if data is None:
		LOGGER.error("Request body is empty or not valid JSON")
		return jsonify({
		'status': 'error',
		'message': 'Request body must be valid JSON'
		}), 400
		except Exception as e:
		LOGGER.error(f"Failed to parse request JSON: {e}")
		return jsonify({
		'status': 'error',
		'message': f'Invalid JSON: {str(e)}'
		}), 400

		# Extract simap_id
		simap_id = data.get('simap_id')
		if not simap_id:
		LOGGER.error("Missing simap_id in request")
		return jsonify({
		'status': 'error',
		'message': 'Missing required field: simap_id'
		}), 400

		# Find and stop the thread
		with _threads_lock:
		if simap_id not in _analysis_threads:
		LOGGER.warning(f"No running analysis found for SIMAP ID {simap_id}")
		return jsonify({
		'status': 'error',
		'message': f'No running analysis found for SIMAP ID {simap_id}'
		}), 404

		thread_info = _analysis_threads[simap_id]
		if not thread_info['thread'].is_alive():
		# Clean up dead thread
		del _analysis_threads[simap_id]
		LOGGER.warning(f"Analysis thread for SIMAP ID {simap_id} is not alive")
		return jsonify({
		'status': 'error',
		'message': f'Analysis for SIMAP ID {simap_id} is not running'
		}), 404

		# Signal thread to stop
		thread_info['stop_event'].set()
		LOGGER.info(f"Stop signal sent to analysis thread for SIMAP ID {simap_id}")

		return jsonify({
		'status': 'success',
		'data': results,
		'message': 'Analysis completed successfully'
		'message': f'Stop signal sent to analysis for SIMAP ID {simap_id}',
		'simap_id': simap_id
		}), 200

		except Exception as e:
		# Check if this is a retry failure (service unavailable)
		error_msg = str(e)
		if 'Giving up' in error_msg or 'unavailable' in error_msg.lower():
		LOGGER.error(f"External service unavailable: {e}")
		@blueprint.route('/analyze/stop-all', methods=['POST'])
		def stop_all_analyses():
		"""
		Stop all running analyses.

		Signals all background threads to stop gracefully.

		Returns:
		JSON response with summary of stopped analyses.
		"""
		LOGGER.info("Received stop all analyses request")

		stopped_ids = []
		skipped_ids = []

		with _threads_lock:
		if not _analysis_threads:
		LOGGER.info("No running analyses to stop")
		return jsonify({
		'status': 'error',
		'message': f'External service unavailable: {error_msg}'
		}), 503
		'status': 'success',
		'message': 'No running analyses to stop',
		'stopped_count': 0,
		'stopped_ids': []
		}), 200

		# Signal all threads to stop
		for simap_id, thread_info in list(_analysis_threads.items()):
		if thread_info['thread'].is_alive():
		thread_info['stop_event'].set()
		stopped_ids.append(simap_id)
		LOGGER.info(f"Stop signal sent to analysis thread for SIMAP ID {simap_id}")
		else:
		LOGGER.exception(f"Unexpected error during analysis: {e}")
		skipped_ids.append(simap_id)
		LOGGER.warning(f"Analysis thread for SIMAP ID {simap_id} is not alive, skipping")

		return jsonify({
		'status': 'error',
		'message': f'Internal server error: {error_msg}'
		}), 500
		'status': 'success',
		'message': f'Stop signal sent to {len(stopped_ids)} running analyses',
		'stopped_count': len(stopped_ids),
		}), 200

		@blueprint.route('/status', methods=['GET'])
		def status():
		"""
		Get status of all running analyses.

		Returns:
		JSON response with list of running analyses.
		"""
		LOGGER.debug("Status check requested")

		with _threads_lock:
		running_analyses = [
		{
		'simap_id': simap_id,
		'is_alive': info['thread'].is_alive(),
		'start_time': info['start_time'],
		'duration_minutes': info['duration_minutes']
		}
		for simap_id, info in _analysis_threads.items()
		]

		return jsonify({
		'running_count': len(running_analyses),
		'analyses': running_analyses,
		'timestamp': datetime.now(UTC).isoformat()
		}), 200



		@blueprint.route('/health', methods=['GET'])
		def health():

src/tests/mwc26-f5ga/AI_analytics_engine/tests/run_test.sh

+1 −1

Original line number	Diff line number	Diff line
		@@ -37,7 +37,7 @@ LOG_FILE="${PWD}/test_api_docker.log"
		TEST_FILE="${PWD}/test_api_docker.py"

		# Run the test with logging enabled and capture output
		pytest $TEST_FILE \
		pytest $TEST_FILE::test_analyze_endpoint \
		-v -s \
		--log-cli-level=DEBUG \
		--log-file="${LOG_FILE}" \

src/tests/mwc26-f5ga/AI_analytics_engine/tests/test_api_docker.py

+204 −14

File changed.

Preview size limit exceeded, changes collapsed.