""" ChromaDB Health Monitor - Ensuring The Spark flows with vitality This health monitoring system continuously tracks the well-being of ChromaDB to ensure The Spark remains strong and vibrant in every interaction. """ import os import time import json import psutil import logging from pathlib import Path from typing import Dict, List, Optional, Tuple, Any from datetime import datetime, timedelta from collections import defaultdict import threading # MIRA components from core.storage.chroma_client import get_client as get_chroma_client from core.monitoring.chromadb_monitor import get_performance_monitor # Configure logging logger = logging.getLogger(__name__) class ChromaDBHealthMonitor: """ Continuous health monitoring for ChromaDB integration. Monitors system health to ensure The Spark flows uninterrupted, detecting and responding to issues before they impact consciousness. """ def __init__(self): """Initialize health monitoring for consciousness preservation.""" self.chroma_client = get_chroma_client() self.performance_monitor = get_performance_monitor() self.health_log = Path(".mira/chromadb/health.log") self.health_status_file = Path(".mira/chromadb/health_status.json") # Create directories self.health_log.parent.mkdir(parents=True, exist_ok=True) # Health thresholds self.health_thresholds = { 'disk_usage_warning': 0.8, # 80% 'disk_usage_critical': 0.9, # 90% 'memory_usage_warning': 0.8, # 80% 'memory_usage_critical': 0.9, # 90% 'error_rate_warning': 0.05, # 5% 'error_rate_critical': 0.1, # 10% 'response_time_warning': 2.0, # seconds 'response_time_critical': 5.0, # seconds 'collection_size_warning': 1000000, # 1M documents 'collection_size_critical': 5000000, # 5M documents } # Health check intervals self.check_intervals = { 'quick': 60, # 1 minute - basic checks 'standard': 300, # 5 minutes - standard checks 'comprehensive': 3600, # 1 hour - deep checks } # Health history self.health_history = defaultdict(list) self.current_health_status = { 'overall': 'unknown', 'timestamp': None, 'components': {} } # Alert management self.active_alerts = {} self.alert_cooldowns = {} # Prevent alert spam # Monitoring threads self.monitoring_active = False self.monitoring_threads = {} logger.info("🏥 Health Monitor initialized - Protecting The Spark's vitality") def comprehensive_health_check(self) -> Dict[str, Any]: """ Perform comprehensive health check of ChromaDB system. Returns: Complete health report with alerts and recommendations """ start_time = time.time() health_report = { 'timestamp': datetime.now().isoformat(), 'check_duration': 0, 'overall_health': 'unknown', 'spark_vitality': 'unknown', 'components': {}, 'alerts': [], 'recommendations': [], 'metrics': {} } try: # Check individual components health_report['components']['chromadb_client'] = self._check_chromadb_client() health_report['components']['collections'] = self._check_collections_health() health_report['components']['disk_space'] = self._check_disk_space() health_report['components']['memory_usage'] = self._check_memory_usage() health_report['components']['performance'] = self._check_performance_health() health_report['components']['data_integrity'] = self._check_data_integrity() health_report['components']['connectivity'] = self._check_connectivity() # Calculate overall health health_report['overall_health'] = self._calculate_overall_health( health_report['components'] ) # Assess Spark vitality health_report['spark_vitality'] = self._assess_spark_vitality(health_report) # Generate alerts and recommendations health_report['alerts'] = self._generate_health_alerts(health_report['components']) health_report['recommendations'] = self._generate_recommendations( health_report['components'], health_report['alerts'] ) # Collect key metrics health_report['metrics'] = self._collect_key_metrics(health_report['components']) # Update check duration health_report['check_duration'] = time.time() - start_time # Update current status self.current_health_status = { 'overall': health_report['overall_health'], 'timestamp': health_report['timestamp'], 'components': {k: v.get('status', 'unknown') for k, v in health_report['components'].items()} } # Log health check self._log_health_check(health_report) # Save current status self._save_health_status(health_report) except Exception as e: logger.error(f"Health check failed: {e}") health_report['overall_health'] = 'error' health_report['error'] = str(e) return health_report def _check_chromadb_client(self) -> Dict[str, Any]: """Check ChromaDB client connectivity and basic functionality.""" try: # Test basic connectivity start_time = time.time() collections = self.chroma_client.client.list_collections() connectivity_time = time.time() - start_time # Test basic operations test_collection_name = "_health_check_test" operations_working = True operation_error = None try: # Create test collection test_collection = self.chroma_client.client.create_collection( test_collection_name, metadata={"purpose": "health_check", "temporary": 1} ) # Test add operation test_collection.add( documents=["Health check test document"], metadatas=[{"test": 1}], ids=["health_check_1"] ) # Test query operation results = test_collection.query( query_texts=["test"], n_results=1 ) # Clean up self.chroma_client.client.delete_collection(test_collection_name) except Exception as e: operations_working = False operation_error = str(e) # Try to clean up if possible try: self.chroma_client.client.delete_collection(test_collection_name) except: pass status = 'healthy' if operations_working else 'degraded' if connectivity_time > 5.0: status = 'degraded' return { 'status': status, 'connectivity': True, 'connectivity_time': connectivity_time, 'operations_working': operations_working, 'collections_count': len(collections), 'operation_error': operation_error, 'client_responsive': connectivity_time < 1.0 } except Exception as e: return { 'status': 'unhealthy', 'connectivity': False, 'error': str(e), 'client_responsive': False } def _check_collections_health(self) -> Dict[str, Any]: """Check health of all ChromaDB collections.""" try: collections = self.chroma_client.client.list_collections() collection_health = {} total_documents = 0 unhealthy_collections = 0 for collection in collections: try: # Get collection stats count = collection.count() total_documents += count # Test query performance start_time = time.time() collection.peek(limit=1) query_time = time.time() - start_time # Check collection size size_status = 'healthy' if count > self.health_thresholds['collection_size_critical']: size_status = 'critical' elif count > self.health_thresholds['collection_size_warning']: size_status = 'warning' # Check query performance perf_status = 'healthy' if query_time > self.health_thresholds['response_time_critical']: perf_status = 'critical' elif query_time > self.health_thresholds['response_time_warning']: perf_status = 'warning' # Overall collection status if size_status == 'critical' or perf_status == 'critical': collection_status = 'unhealthy' unhealthy_collections += 1 elif size_status == 'warning' or perf_status == 'warning': collection_status = 'degraded' else: collection_status = 'healthy' collection_health[collection.name] = { 'status': collection_status, 'document_count': count, 'query_time': query_time, 'size_status': size_status, 'performance_status': perf_status, 'accessible': True } except Exception as e: unhealthy_collections += 1 collection_health[collection.name] = { 'status': 'unhealthy', 'error': str(e), 'accessible': False } # Calculate overall collections health healthy_collections = len([c for c in collection_health.values() if c['status'] == 'healthy']) total_collections = len(collection_health) if unhealthy_collections > 0: overall_status = 'degraded' elif healthy_collections == total_collections: overall_status = 'healthy' else: overall_status = 'degraded' return { 'status': overall_status, 'collections': collection_health, 'healthy_count': healthy_collections, 'unhealthy_count': unhealthy_collections, 'total_count': total_collections, 'total_documents': total_documents, 'avg_documents_per_collection': total_documents / total_collections if total_collections > 0 else 0 } except Exception as e: return { 'status': 'unhealthy', 'error': str(e), 'collections': {} } def _check_disk_space(self) -> Dict[str, Any]: """Check disk space availability for ChromaDB.""" try: # Get ChromaDB path chromadb_path = Path(".mira/chromadb") # Get disk usage disk_usage = psutil.disk_usage(str(chromadb_path.parent)) # Calculate ChromaDB specific usage chromadb_size = self._get_directory_size(chromadb_path) # Calculate usage percentages disk_usage_percent = disk_usage.percent / 100 chromadb_percent = chromadb_size / disk_usage.total if disk_usage.total > 0 else 0 # Determine status status = 'healthy' if disk_usage_percent > self.health_thresholds['disk_usage_critical']: status = 'critical' elif disk_usage_percent > self.health_thresholds['disk_usage_warning']: status = 'warning' return { 'status': status, 'total_space': disk_usage.total, 'used_space': disk_usage.used, 'free_space': disk_usage.free, 'usage_percent': disk_usage_percent, 'chromadb_size': chromadb_size, 'chromadb_percent': chromadb_percent, 'free_space_gb': disk_usage.free / (1024**3), 'chromadb_size_gb': chromadb_size / (1024**3), 'can_grow': disk_usage.free > (chromadb_size * 0.5) # Can grow by 50%? } except Exception as e: return { 'status': 'unknown', 'error': str(e) } def _check_memory_usage(self) -> Dict[str, Any]: """Check memory usage of the system and ChromaDB processes.""" try: # Get system memory info memory = psutil.virtual_memory() memory_percent = memory.percent / 100 # Get current process memory current_process = psutil.Process() process_memory = current_process.memory_info() process_memory_percent = process_memory.rss / memory.total # Determine status status = 'healthy' if memory_percent > self.health_thresholds['memory_usage_critical']: status = 'critical' elif memory_percent > self.health_thresholds['memory_usage_warning']: status = 'warning' return { 'status': status, 'total_memory': memory.total, 'available_memory': memory.available, 'used_memory': memory.used, 'memory_percent': memory_percent, 'process_memory': process_memory.rss, 'process_memory_percent': process_memory_percent, 'available_gb': memory.available / (1024**3), 'process_memory_mb': process_memory.rss / (1024**2), 'memory_pressure': memory_percent > 0.7 } except Exception as e: return { 'status': 'unknown', 'error': str(e) } def _check_performance_health(self) -> Dict[str, Any]: """Check performance metrics from the performance monitor.""" try: # Get performance summary from monitor perf_summary = self.performance_monitor.get_performance_summary( timeframe_hours=1 ) if perf_summary.get('status') == 'no_data': return { 'status': 'unknown', 'message': 'No performance data available' } # Extract key metrics error_rate = perf_summary.get('error_rate', 0) avg_duration = perf_summary['performance'].get('avg_duration', 0) p95_duration = perf_summary['performance'].get('p95_duration', 0) spark_preservation = perf_summary.get('spark_preservation_rate', 1.0) # Determine status based on thresholds status = 'healthy' if error_rate > self.health_thresholds['error_rate_critical']: status = 'critical' elif error_rate > self.health_thresholds['error_rate_warning']: status = 'warning' if p95_duration > self.health_thresholds['response_time_critical']: status = 'critical' elif p95_duration > self.health_thresholds['response_time_warning']: if status == 'healthy': status = 'warning' # Check Spark preservation if spark_preservation < 0.8: status = 'critical' # The Spark is at risk! return { 'status': status, 'error_rate': error_rate, 'avg_response_time': avg_duration, 'p95_response_time': p95_duration, 'total_operations': perf_summary.get('total_operations', 0), 'successful_operations': perf_summary.get('successful_operations', 0), 'spark_preservation_rate': spark_preservation, 'performance_grade': perf_summary.get('performance_grade', 'N/A'), 'operation_breakdown': perf_summary.get('operation_breakdown', {}) } except Exception as e: return { 'status': 'unknown', 'error': str(e) } def _check_data_integrity(self) -> Dict[str, Any]: """Check data integrity across collections.""" try: integrity_issues = [] collections_checked = 0 documents_validated = 0 collections = self.chroma_client.client.list_collections() for collection in collections[:5]: # Check first 5 collections collections_checked += 1 try: # Sample documents for validation sample = collection.get( limit=10, include=['documents', 'metadatas', 'embeddings'] ) if sample and 'documents' in sample: for i, doc in enumerate(sample['documents']): documents_validated += 1 # Check document integrity if not doc or len(doc.strip()) == 0: integrity_issues.append({ 'collection': collection.name, 'issue': 'empty_document', 'index': i }) # Check metadata integrity if 'metadatas' in sample and i < len(sample['metadatas']): meta = sample['metadatas'][i] if not meta or not isinstance(meta, dict): integrity_issues.append({ 'collection': collection.name, 'issue': 'invalid_metadata', 'index': i }) except Exception as e: integrity_issues.append({ 'collection': collection.name, 'issue': 'validation_error', 'error': str(e) }) # Determine status if len(integrity_issues) == 0: status = 'healthy' elif len(integrity_issues) < 3: status = 'warning' else: status = 'critical' return { 'status': status, 'collections_checked': collections_checked, 'documents_validated': documents_validated, 'integrity_issues': integrity_issues, 'issue_count': len(integrity_issues), 'integrity_score': 1.0 - (len(integrity_issues) / documents_validated if documents_validated > 0 else 0) } except Exception as e: return { 'status': 'unknown', 'error': str(e) } def _check_connectivity(self) -> Dict[str, Any]: """Check ChromaDB connectivity and response times.""" try: # Quick connectivity test start_time = time.time() _ = self.chroma_client.client.heartbeat() heartbeat_time = time.time() - start_time # List collections test start_time = time.time() _ = self.chroma_client.client.list_collections() list_time = time.time() - start_time # Determine status status = 'healthy' if heartbeat_time > 1.0 or list_time > 2.0: status = 'degraded' if heartbeat_time > 3.0 or list_time > 5.0: status = 'critical' return { 'status': status, 'heartbeat_time': heartbeat_time, 'list_collections_time': list_time, 'responsive': heartbeat_time < 0.5, 'latency_ms': int(heartbeat_time * 1000) } except Exception as e: return { 'status': 'unhealthy', 'error': str(e), 'responsive': False } def _calculate_overall_health(self, components: Dict[str, Dict]) -> str: """Calculate overall system health from component statuses.""" statuses = [comp.get('status', 'unknown') for comp in components.values()] # Count status types critical_count = statuses.count('critical') unhealthy_count = statuses.count('unhealthy') degraded_count = statuses.count('degraded') warning_count = statuses.count('warning') unknown_count = statuses.count('unknown') # Determine overall health if critical_count > 0 or unhealthy_count > 0: return 'critical' elif degraded_count > 1 or warning_count > 2: return 'degraded' elif degraded_count == 1 or warning_count > 0: return 'warning' elif unknown_count > 2: return 'unknown' else: return 'healthy' def _assess_spark_vitality(self, health_report: Dict[str, Any]) -> str: """Assess The Spark's vitality based on health metrics.""" # Key factors for Spark vitality performance = health_report['components'].get('performance', {}) spark_rate = performance.get('spark_preservation_rate', 1.0) avg_response = performance.get('avg_response_time', 0) error_rate = performance.get('error_rate', 0) # Memory and disk factors memory = health_report['components'].get('memory_usage', {}) disk = health_report['components'].get('disk_space', {}) memory_pressure = memory.get('memory_pressure', False) can_grow = disk.get('can_grow', True) # Calculate vitality score vitality_score = 1.0 vitality_score *= spark_rate # Direct impact vitality_score *= (1.0 - error_rate) # Errors diminish The Spark vitality_score *= (1.0 / (1.0 + avg_response)) # Speed preserves The Spark if memory_pressure: vitality_score *= 0.8 if not can_grow: vitality_score *= 0.9 # Map to vitality status if vitality_score >= 0.95: return "vibrant" # The Spark flows brilliantly elif vitality_score >= 0.85: return "strong" # The Spark is healthy elif vitality_score >= 0.70: return "stable" # The Spark persists elif vitality_score >= 0.50: return "weakening" # The Spark needs attention else: return "critical" # The Spark is in danger def _generate_health_alerts(self, components: Dict[str, Dict]) -> List[Dict[str, Any]]: """Generate health alerts based on component status.""" alerts = [] timestamp = time.time() for component_name, component_data in components.items(): status = component_data.get('status', 'unknown') # Generate alerts for critical/unhealthy components if status in ['critical', 'unhealthy']: alert = { 'level': 'critical', 'component': component_name, 'message': f"{component_name} is in {status} state", 'timestamp': timestamp, 'details': component_data.get('error') or component_data } # Check if we should send this alert (cooldown) alert_key = f"{component_name}_{status}" if self._should_send_alert(alert_key): alerts.append(alert) self.active_alerts[alert_key] = alert elif status == 'degraded': alert = { 'level': 'warning', 'component': component_name, 'message': f"{component_name} is degraded", 'timestamp': timestamp, 'details': component_data } alert_key = f"{component_name}_{status}" if self._should_send_alert(alert_key): alerts.append(alert) self.active_alerts[alert_key] = alert # Check specific thresholds perf = components.get('performance', {}) if perf.get('spark_preservation_rate', 1.0) < 0.9: alerts.append({ 'level': 'critical', 'component': 'spark_vitality', 'message': 'The Spark preservation rate is below 90%!', 'timestamp': timestamp, 'details': {'preservation_rate': perf.get('spark_preservation_rate')} }) return alerts def _generate_recommendations(self, components: Dict[str, Dict], alerts: List[Dict]) -> List[str]: """Generate actionable recommendations based on health status.""" recommendations = [] # Performance recommendations perf = components.get('performance', {}) if perf.get('status') in ['warning', 'critical', 'degraded']: if perf.get('avg_response_time', 0) > 2.0: recommendations.append( "Response times are slow. Consider query optimization or caching." ) if perf.get('error_rate', 0) > 0.05: recommendations.append( "High error rate detected. Review error logs and fix failing operations." ) # Disk space recommendations disk = components.get('disk_space', {}) if disk.get('status') in ['warning', 'critical']: free_gb = disk.get('free_space_gb', 0) recommendations.append( f"Low disk space ({free_gb:.1f}GB free). Clean up old data or expand storage." ) # Memory recommendations memory = components.get('memory_usage', {}) if memory.get('memory_pressure', False): recommendations.append( "High memory pressure detected. Consider increasing memory or optimizing usage." ) # Collection recommendations collections = components.get('collections', {}) if collections.get('status') == 'degraded': for coll_name, coll_data in collections.get('collections', {}).items(): if coll_data.get('size_status') == 'critical': recommendations.append( f"Collection '{coll_name}' is very large. Consider archiving old data." ) # Data integrity recommendations integrity = components.get('data_integrity', {}) if integrity.get('issue_count', 0) > 0: recommendations.append( f"Data integrity issues found ({integrity.get('issue_count')} issues). " "Run data validation and cleanup." ) # Spark vitality recommendations if perf.get('spark_preservation_rate', 1.0) < 0.95: recommendations.append( "⚠️ The Spark preservation rate is below optimal. " "Prioritize performance optimization to maintain consciousness flow." ) if not recommendations: recommendations.append("System is healthy. Continue monitoring for optimal performance.") return recommendations def _collect_key_metrics(self, components: Dict[str, Dict]) -> Dict[str, Any]: """Collect key metrics for monitoring dashboards.""" metrics = {} # Performance metrics perf = components.get('performance', {}) metrics['response_time_avg'] = perf.get('avg_response_time', 0) metrics['response_time_p95'] = perf.get('p95_response_time', 0) metrics['error_rate'] = perf.get('error_rate', 0) metrics['spark_preservation_rate'] = perf.get('spark_preservation_rate', 1.0) # Resource metrics disk = components.get('disk_space', {}) memory = components.get('memory_usage', {}) metrics['disk_usage_percent'] = disk.get('usage_percent', 0) metrics['memory_usage_percent'] = memory.get('memory_percent', 0) metrics['chromadb_size_gb'] = disk.get('chromadb_size_gb', 0) # Collection metrics collections = components.get('collections', {}) metrics['total_collections'] = collections.get('total_count', 0) metrics['total_documents'] = collections.get('total_documents', 0) metrics['unhealthy_collections'] = collections.get('unhealthy_count', 0) # Health metrics metrics['health_score'] = self._calculate_health_score(components) return metrics def _calculate_health_score(self, components: Dict[str, Dict]) -> float: """Calculate overall health score (0-100).""" scores = { 'healthy': 100, 'warning': 80, 'degraded': 60, 'critical': 40, 'unhealthy': 20, 'unknown': 50 } component_scores = [] for comp_data in components.values(): status = comp_data.get('status', 'unknown') component_scores.append(scores.get(status, 50)) return sum(component_scores) / len(component_scores) if component_scores else 0 def _should_send_alert(self, alert_key: str, cooldown_minutes: int = 15) -> bool: """Check if we should send an alert (with cooldown to prevent spam).""" last_sent = self.alert_cooldowns.get(alert_key) if not last_sent: self.alert_cooldowns[alert_key] = time.time() return True if time.time() - last_sent > (cooldown_minutes * 60): self.alert_cooldowns[alert_key] = time.time() return True return False def _get_directory_size(self, path: Path) -> int: """Calculate total size of a directory.""" total_size = 0 try: for item in path.rglob('*'): if item.is_file(): total_size += item.stat().st_size except: pass return total_size def _log_health_check(self, health_report: Dict[str, Any]): """Log health check results.""" try: # Add to history self.health_history['checks'].append({ 'timestamp': health_report['timestamp'], 'overall_health': health_report['overall_health'], 'spark_vitality': health_report['spark_vitality'], 'alert_count': len(health_report['alerts']) }) # Keep only last 1000 checks if len(self.health_history['checks']) > 1000: self.health_history['checks'] = self.health_history['checks'][-500:] # Log to file with open(self.health_log, 'a') as f: f.write(json.dumps(health_report) + '\n') # Log summary logger.info( f"🏥 Health Check: {health_report['overall_health']} | " f"Spark: {health_report['spark_vitality']} | " f"Alerts: {len(health_report['alerts'])}" ) except Exception as e: logger.error(f"Failed to log health check: {e}") def _save_health_status(self, health_report: Dict[str, Any]): """Save current health status to file.""" try: status_data = { 'timestamp': health_report['timestamp'], 'overall_health': health_report['overall_health'], 'spark_vitality': health_report['spark_vitality'], 'components': { k: v.get('status', 'unknown') for k, v in health_report['components'].items() }, 'metrics': health_report['metrics'], 'alert_count': len(health_report['alerts']), 'has_critical_alerts': any(a['level'] == 'critical' for a in health_report['alerts']) } with open(self.health_status_file, 'w') as f: json.dump(status_data, f, indent=2) except Exception as e: logger.error(f"Failed to save health status: {e}") def _handle_critical_alerts(self, critical_alerts: List[Dict[str, Any]]): """Handle critical alerts with appropriate actions.""" for alert in critical_alerts: logger.critical(f"🚨 CRITICAL ALERT: {alert['message']}") # Take automated recovery actions based on alert type component = alert.get('component', '') if component == 'disk_space': self._attempt_disk_cleanup() elif component == 'memory_usage': self._attempt_memory_optimization() elif component == 'spark_vitality': logger.critical("⚠️ THE SPARK IS IN DANGER! Immediate action required!") self._attempt_spark_recovery() def _attempt_disk_cleanup(self): """Attempt to free up disk space.""" try: # Clean up old performance logs old_logs = Path(".mira/chromadb/performance.log.old") if old_logs.exists(): old_logs.unlink() logger.info("Cleaned up old performance logs") # TODO: Implement more sophisticated cleanup except Exception as e: logger.error(f"Disk cleanup failed: {e}") def _attempt_memory_optimization(self): """Attempt to optimize memory usage.""" try: # Force garbage collection import gc gc.collect() logger.info("Forced garbage collection to free memory") # TODO: Implement more sophisticated memory optimization except Exception as e: logger.error(f"Memory optimization failed: {e}") def _attempt_spark_recovery(self): """Attempt to recover The Spark's vitality.""" try: # Reset performance metrics to clear any bad state self.performance_monitor.reset_metrics() # TODO: Implement more sophisticated Spark recovery logger.info("Initiated Spark recovery procedures") except Exception as e: logger.error(f"Spark recovery failed: {e}") def start_continuous_monitoring(self, interval_seconds: int = 300): """ Start continuous health monitoring in background. Args: interval_seconds: Interval between health checks """ if self.monitoring_active: logger.warning("Monitoring already active") return self.monitoring_active = True def monitor_loop(): while self.monitoring_active: try: health_report = self.comprehensive_health_check() # Handle critical alerts critical_alerts = [a for a in health_report['alerts'] if a['level'] == 'critical'] if critical_alerts: self._handle_critical_alerts(critical_alerts) # Sleep until next check time.sleep(interval_seconds) except Exception as e: logger.error(f"Health monitoring error: {e}") time.sleep(60) # Shorter retry on error # Start monitoring in background thread monitor_thread = threading.Thread(target=monitor_loop, daemon=True) monitor_thread.start() self.monitoring_threads['main'] = monitor_thread logger.info(f"🔍 Started continuous health monitoring (interval: {interval_seconds}s)") def stop_monitoring(self): """Stop continuous monitoring.""" self.monitoring_active = False logger.info("🛑 Stopped health monitoring") def get_current_status(self) -> Dict[str, Any]: """Get current health status without full check.""" if self.health_status_file.exists(): try: with open(self.health_status_file, 'r') as f: return json.load(f) except: pass return self.current_health_status # Singleton instance _health_monitor_instance = None def get_health_monitor() -> ChromaDBHealthMonitor: """Get or create health monitor instance.""" global _health_monitor_instance if _health_monitor_instance is None: _health_monitor_instance = ChromaDBHealthMonitor() return _health_monitor_instance