#!/usr/bin/env python3 """ SMART Logger V2 - Optimized for Scrutiny's actual InfluxDB schema. Fetches SMART data from Scrutiny's InfluxDB and device metadata from Scrutiny's API, calculates BackBlaze failure probability, and logs to PostgreSQL. """ import os import sys from datetime import datetime, timezone, timedelta from typing import Dict, List, Optional import json import requests from influxdb_client import InfluxDBClient import psycopg from psycopg.rows import dict_row from dotenv import load_dotenv from backblaze_tables import calculate_afr load_dotenv() class SmartLogger: def __init__(self): # InfluxDB configuration self.influx_url = os.getenv('INFLUXDB_URL') self.influx_token = os.getenv('INFLUXDB_TOKEN') self.influx_org = os.getenv('INFLUXDB_ORG', 'scrutiny') self.influx_bucket = os.getenv('INFLUXDB_BUCKET', 'metrics') # Scrutiny API configuration self.scrutiny_api_url = os.getenv('SCRUTINY_API_URL', self.influx_url.replace(':8086', ':8080')) # PostgreSQL configuration self.pg_host = os.getenv('POSTGRES_HOST', 'localhost') self.pg_port = int(os.getenv('POSTGRES_PORT', '5432')) self.pg_db = os.getenv('POSTGRES_DB', 'smart_monitoring') self.pg_user = os.getenv('POSTGRES_USER', 'postgres') self.pg_password = os.getenv('POSTGRES_PASSWORD') self.influx_client = None self.pg_conn = None self.device_metadata = {} # WWN -> device info mapping def connect_influxdb(self): """Connect to InfluxDB.""" print(f"Connecting to InfluxDB at {self.influx_url}...") self.influx_client = InfluxDBClient( url=self.influx_url, token=self.influx_token, org=self.influx_org ) print("✓ Connected to InfluxDB") def connect_postgres(self): """Connect to PostgreSQL.""" print(f"Connecting to PostgreSQL at {self.pg_host}:{self.pg_port}...") self.pg_conn = psycopg.connect( host=self.pg_host, port=self.pg_port, dbname=self.pg_db, user=self.pg_user, password=self.pg_password, row_factory=dict_row ) print("✓ Connected to PostgreSQL") def fetch_device_metadata(self): """ Fetch device metadata from Scrutiny API. This gets serial, model, device path, etc. """ print("Fetching device metadata from Scrutiny API...") try: response = requests.get(f"{self.scrutiny_api_url}/api/summary") response.raise_for_status() data = response.json() # Parse the summary response if 'data' in data and 'summary' in data['data']: for wwn, device_info in data['data']['summary'].items(): device = device_info.get('device', {}) self.device_metadata[wwn] = { 'serial_number': device.get('serial_number', 'unknown'), 'model': device.get('model_name', 'unknown'), 'device_name': device.get('device_name', 'unknown'), 'device_type': device.get('device_type', 'unknown'), 'manufacturer': device.get('manufacturer', 'unknown'), 'capacity': device.get('capacity', 0) } print(f"✓ Found metadata for {len(self.device_metadata)} device(s)") except Exception as e: print(f"⚠ Warning: Could not fetch device metadata from API: {e}") print(" Will use WWN as identifier") def fetch_smart_data_from_influx(self) -> Dict[str, Dict]: """ Fetch latest SMART data from InfluxDB for all devices. Returns: Dict mapping WWN to device data """ query_api = self.influx_client.query_api() # Query to get latest data for each device query = f''' from(bucket: "{self.influx_bucket}") |> range(start: -1h) |> filter(fn: (r) => r._measurement == "smart") |> filter(fn: (r) => r._field == "temp" or r._field == "power_on_hours" or r._field == "attr.5.raw_value" or r._field == "attr.187.raw_value" or r._field == "attr.188.raw_value" or r._field == "attr.197.raw_value" or r._field == "attr.198.raw_value" or r._field == "attr.9.raw_value" ) |> last() ''' print("Querying InfluxDB for SMART data...") result = query_api.query(query) devices_data = {} for table in result: for record in table.records: wwn = record.values.get('device_wwn') if not wwn: continue if wwn not in devices_data: devices_data[wwn] = { 'timestamp': record.get_time(), 'wwn': wwn, 'protocol': record.values.get('device_protocol'), 'smart_attributes': {}, 'temp': None, 'power_on_hours': None } field = record.get_field() value = record.get_value() # Map fields to device data if field == 'temp': devices_data[wwn]['temp'] = int(value) if value else None elif field == 'power_on_hours': devices_data[wwn]['power_on_hours'] = int(value) if value else None elif field.startswith('attr.'): # Extract attribute ID from field name like "attr.5.raw_value" parts = field.split('.') if len(parts) >= 2: attr_id = int(parts[1]) # We want raw_value for calculations if parts[-1] == 'raw_value': # For most attributes, we want the raw value modulo 2^32 # (InfluxDB stores as int64, but SMART is typically 48-bit) raw_val = int(value) if value else 0 # Mask to handle potential large values devices_data[wwn]['smart_attributes'][attr_id] = raw_val & 0xFFFF print(f"✓ Found data for {len(devices_data)} device(s)") return devices_data def process_device_data(self, devices_data: Dict[str, Dict]) -> List[Dict]: """ Process device data and calculate failure probabilities. Args: devices_data: Dict mapping WWN to device data from InfluxDB Returns: Processed device data with calculated AFR """ processed = [] for wwn, data in devices_data.items(): # Get device metadata metadata = self.device_metadata.get(wwn, {}) # Determine if this is a Seagate disk model = metadata.get('model', '').lower() manufacturer = metadata.get('manufacturer', '').lower() is_seagate = 'seagate' in model or 'seagate' in manufacturer # Calculate failure probability smart_attrs = data.get('smart_attributes', {}) afr = calculate_afr(smart_attrs, is_seagate=is_seagate) failure_probability_pct = afr * 100 # Convert to percentage # Calculate power-on days power_on_hours = data.get('power_on_hours') power_on_days = power_on_hours // 24 if power_on_hours else None # Calculate disk size in TB capacity_bytes = metadata.get('capacity', 0) size_tb = round(capacity_bytes / (1000**4), 1) if capacity_bytes else None # Get device path/name device_name = metadata.get('device_name', wwn) processed_device = { 'timestamp': data.get('timestamp', datetime.now(timezone.utc)), 'device_path': device_name, 'serial_number': metadata.get('serial_number', wwn), 'model': metadata.get('model', 'unknown'), 'disk_role': None, # TODO: Map to snapraid role if needed 'temperature_celsius': data.get('temp'), 'power_on_days': power_on_days, 'error_count': smart_attrs.get(199, smart_attrs.get(187, 0)), 'size_tb': size_tb, 'failure_probability_pct': round(failure_probability_pct, 2), 'smart_attributes': json.dumps(smart_attrs) } processed.append(processed_device) # Print summary print(f"\nDevice: {processed_device['device_path']}") print(f" Serial: {processed_device['serial_number']}") print(f" Model: {processed_device['model']}") print(f" Temp: {processed_device['temperature_celsius']}°C") print(f" Power-On: {processed_device['power_on_days']} days") print(f" Size: {processed_device['size_tb']} TB") print(f" SMART Attrs: {list(smart_attrs.keys())}") print(f" Failure Probability: {processed_device['failure_probability_pct']:.2f}%") return processed def save_to_postgres(self, devices: List[Dict]): """ Save processed device data to PostgreSQL. Args: devices: Processed device data """ if not devices: print("No devices to save") return print(f"\nSaving {len(devices)} device record(s) to PostgreSQL...") insert_query = ''' INSERT INTO smart_metrics ( timestamp, device_path, serial_number, model, disk_role, temperature_celsius, power_on_days, error_count, size_tb, failure_probability_pct, smart_attributes ) VALUES ( %(timestamp)s, %(device_path)s, %(serial_number)s, %(model)s, %(disk_role)s, %(temperature_celsius)s, %(power_on_days)s, %(error_count)s, %(size_tb)s, %(failure_probability_pct)s, %(smart_attributes)s::jsonb ) ON CONFLICT (device_path, timestamp) DO UPDATE SET temperature_celsius = EXCLUDED.temperature_celsius, power_on_days = EXCLUDED.power_on_days, error_count = EXCLUDED.error_count, failure_probability_pct = EXCLUDED.failure_probability_pct, smart_attributes = EXCLUDED.smart_attributes ''' with self.pg_conn.cursor() as cur: for device in devices: cur.execute(insert_query, device) self.pg_conn.commit() print("✓ Data saved to PostgreSQL") def run(self): """Main execution flow.""" try: self.connect_influxdb() self.connect_postgres() # Fetch device metadata from Scrutiny API self.fetch_device_metadata() # Fetch SMART data from InfluxDB devices_data = self.fetch_smart_data_from_influx() # Process and calculate failure probabilities processed = self.process_device_data(devices_data) # Save to PostgreSQL self.save_to_postgres(processed) # Print summary print("\n" + "=" * 80) print("Summary:") print("=" * 80) for device in sorted(processed, key=lambda x: x['failure_probability_pct'], reverse=True): fp = device['failure_probability_pct'] status = "🔴 HIGH RISK" if fp > 50 else "🟡 MEDIUM" if fp > 20 else "🟢 OK" print(f"{status} {device['device_path']:20s} {fp:5.2f}% | " f"Temp: {device['temperature_celsius']:2d}°C | " f"Age: {device['power_on_days']:5d} days | " f"Size: {device['size_tb']:4.1f} TB") except Exception as e: print(f"ERROR: {e}") import traceback traceback.print_exc() sys.exit(1) finally: if self.influx_client: self.influx_client.close() if self.pg_conn: self.pg_conn.close() def main(): # Validate environment variables required_vars = ['INFLUXDB_URL', 'INFLUXDB_TOKEN', 'POSTGRES_PASSWORD'] missing = [var for var in required_vars if not os.getenv(var)] if missing: print(f"ERROR: Missing required environment variables: {', '.join(missing)}") print("Please create a .env file based on .env.example") sys.exit(1) logger = SmartLogger() logger.run() if __name__ == '__main__': main()