portfolio/migration-validation-framework/collectors/mounts.py

"""
Mounts Data Collector

Collects filesystem mount information including mount points, devices,
filesystem types, and usage statistics.
"""

import json
import logging
import subprocess
from typing import Dict, Any, List
from pathlib import Path

logger = logging.getLogger(__name__)

class MountsCollector:
    """Collector for filesystem mount information."""

    def __init__(self):
        self.exclude_patterns = [
            "/proc/*",
            "/sys/*",
            "/dev/*",
            "/run/*"
        ]

    def collect_mounts(self, system: str) -> Dict[str, Any]:
        """Collect mount information from target system."""
        logger.info(f"Collecting mounts data from {system}")

        try:
            # Run mount command
            result = subprocess.run(
                ["ssh", system, "mount"],
                capture_output=True,
                text=True,
                timeout=30
            )

            if result.returncode != 0:
                raise RuntimeError(f"Mount command failed: {result.stderr}")

            mounts = self.parse_mount_output(result.stdout)
            filtered_mounts = self.filter_mounts(mounts)

            # Get usage statistics
            usage_stats = self.collect_usage_stats(system, filtered_mounts)

            return {
                "mounts": filtered_mounts,
                "usage": usage_stats,
                "timestamp": self.get_timestamp(system)
            }

        except subprocess.TimeoutExpired:
            logger.error(f"Timeout collecting mounts from {system}")
            raise
        except Exception as e:
            logger.error(f"Failed to collect mounts from {system}: {e}")
            raise

    def parse_mount_output(self, output: str) -> List[Dict[str, str]]:
        """Parse mount command output."""
        mounts = []

        for line in output.strip().split('\n'):
            if not line.strip():
                continue

            # Parse mount output format: device on mountpoint type fstype (options)
            parts = line.split()
            if len(parts) >= 6 and parts[1] == 'on' and parts[3] == 'type':
                mount_info = {
                    "device": parts[0],
                    "mountpoint": parts[2],
                    "fstype": parts[4],
                    "options": parts[5].strip('()')
                }
                mounts.append(mount_info)

        return mounts

    def filter_mounts(self, mounts: List[Dict[str, str]]) -> List[Dict[str, str]]:
        """Filter out unwanted mount points."""
        filtered = []

        for mount in mounts:
            mountpoint = mount["mountpoint"]
            if not any(Path(mountpoint).match(pattern.rstrip('*')) for pattern in self.exclude_patterns):
                filtered.append(mount)

        return filtered

    def collect_usage_stats(self, system: str, mounts: List[Dict[str, str]]) -> Dict[str, Any]:
        """Collect disk usage statistics for mount points."""
        usage_stats = {}

        for mount in mounts:
            mountpoint = mount["mountpoint"]

            try:
                # Run df command for usage statistics
                result = subprocess.run(
                    ["ssh", system, f"df -BG {mountpoint}"],
                    capture_output=True,
                    text=True,
                    timeout=15
                )

                if result.returncode == 0:
                    usage_stats[mountpoint] = self.parse_df_output(result.stdout)

            except subprocess.TimeoutExpired:
                logger.warning(f"Timeout getting usage for {mountpoint} on {system}")
                usage_stats[mountpoint] = {"error": "timeout"}
            except Exception as e:
                logger.warning(f"Failed to get usage for {mountpoint} on {system}: {e}")
                usage_stats[mountpoint] = {"error": str(e)}

        return usage_stats

    def parse_df_output(self, output: str) -> Dict[str, Any]:
        """Parse df command output."""
        lines = output.strip().split('\n')
        if len(lines) < 2:
            return {"error": "invalid df output"}

        # Parse header and data
        header = lines[0].split()
        data = lines[1].split()

        if len(header) != len(data):
            return {"error": "header/data mismatch"}

        stats = {}
        for i, field in enumerate(header):
            if i < len(data):
                if field in ['1G-blocks', 'Used', 'Available']:
                    # Convert to GB
                    value = data[i]
                    if value.endswith('G'):
                        stats[field.lower()] = float(value.rstrip('G'))
                    else:
                        stats[field.lower()] = float(value) / (1024**3)  # Assume bytes
                elif field == 'Use%':
                    stats['use_percent'] = int(data[i].rstrip('%'))
                else:
                    stats[field.lower()] = data[i]

        return stats

    def get_timestamp(self, system: str) -> str:
        """Get current timestamp from target system."""
        try:
            result = subprocess.run(
                ["ssh", system, "date -Iseconds"],
                capture_output=True,
                text=True,
                timeout=10
            )

            if result.returncode == 0:
                return result.stdout.strip()
            else:
                return "unknown"

        except Exception:
            return "unknown"

def collect(system: str) -> Dict[str, Any]:
    """Main collection function for mounts data."""
    collector = MountsCollector()
    return collector.collect_mounts(system)