analyze_storage

Instructions

Implementation Reference

• src/arch_ops_server/system.py:310-328 (handler)

  The main handler function for the 'analyze_storage' tool. Dispatches to check_disk_space() or get_pacman_cache_stats() based on the action parameter ('disk_usage' or 'cache_stats'). Returns an error response for invalid actions.

  async def analyze_storage(action: str) -> Dict[str, Any]:
      """
      Unified storage analysis tool.
      
      Args:
          action: "disk_usage" or "cache_stats"
      
      Returns:
          Analysis results based on action
      """
      if action == "disk_usage":
          return await check_disk_space()
      elif action == "cache_stats":
          return await get_pacman_cache_stats()
      else:
          return create_error_response(
              "InvalidAction",
              f"Unknown action: {action}. Use 'disk_usage' or 'cache_stats'"
          )

• src/arch_ops_server/system.py:89-149 (helper)

  Helper: check_disk_space() - Checks disk usage for critical paths (/, /home, /var, /var/cache/pacman/pkg) using 'df -h'. Returns disk usage info with warnings if usage exceeds 80% or 90%.

  async def check_disk_space() -> Dict[str, Any]:
      """
      Check disk space for critical paths.
  
      Returns:
          Dict with disk usage for /, /home, /var, /var/cache/pacman/pkg
      """
      logger.info("Checking disk space")
  
      paths_to_check = ["/", "/home", "/var"]
  
      if IS_ARCH:
          paths_to_check.append("/var/cache/pacman/pkg")
  
      disk_info = {}
  
      try:
          for path in paths_to_check:
              if not Path(path).exists():
                  continue
  
              exit_code, stdout, _ = await run_command(
                  ["df", "-h", path],
                  timeout=5,
                  check=False
              )
  
              if exit_code == 0:
                  lines = stdout.strip().split('\n')
                  if len(lines) >= 2:
                      # Parse df output
                      parts = lines[1].split()
                      if len(parts) >= 5:
                          disk_info[path] = {
                              "size": parts[1],
                              "used": parts[2],
                              "available": parts[3],
                              "use_percent": parts[4],
                              "mounted_on": parts[5] if len(parts) > 5 else path
                          }
  
                          # Check if space is critically low
                          use_pct = int(parts[4].rstrip('%'))
                          if use_pct > 90:
                              disk_info[path]["warning"] = "Critical: Less than 10% free"
                          elif use_pct > 80:
                              disk_info[path]["warning"] = "Low: Less than 20% free"
  
          logger.info(f"Checked disk space for {len(disk_info)} paths")
  
          return {
              "disk_usage": disk_info,
              "paths_checked": len(disk_info)
          }
  
      except Exception as e:
          logger.error(f"Failed to check disk space: {e}")
          return create_error_response(
              "DiskCheckError",
              f"Failed to check disk space: {str(e)}"
          )

• src/arch_ops_server/system.py:152-200 (helper)

  Helper: get_pacman_cache_stats() - Analyzes pacman cache in /var/cache/pacman/pkg. Counts packages and calculates total cache size.

  async def get_pacman_cache_stats() -> Dict[str, Any]:
      """
      Analyze pacman package cache.
  
      Returns:
          Dict with cache size, package count, statistics
      """
      if not IS_ARCH:
          return create_error_response(
              "NotSupported",
              "Pacman cache analysis is only available on Arch Linux"
          )
  
      logger.info("Analyzing pacman cache")
  
      cache_dir = Path("/var/cache/pacman/pkg")
  
      try:
          if not cache_dir.exists():
              return create_error_response(
                  "NotFound",
                  "Pacman cache directory not found"
              )
  
          # Count packages
          pkg_files = list(cache_dir.glob("*.pkg.tar.*"))
          pkg_count = len(pkg_files)
  
          # Calculate total size
          total_size = sum(f.stat().st_size for f in pkg_files)
          total_size_mb = total_size / (1024 * 1024)
          total_size_gb = total_size_mb / 1024
  
          logger.info(f"Cache: {pkg_count} packages, {total_size_gb:.2f} GB")
  
          return {
              "cache_dir": str(cache_dir),
              "package_count": pkg_count,
              "total_size_bytes": total_size,
              "total_size_mb": round(total_size_mb, 2),
              "total_size_gb": round(total_size_gb, 2)
          }
  
      except Exception as e:
          logger.error(f"Failed to analyze cache: {e}")
          return create_error_response(
              "CacheAnalysisError",
              f"Failed to analyze pacman cache: {str(e)}"
          )

• src/arch_ops_server/__init__.py:42-50 (registration)

  Tool is exported from the package via __init__.py: imported from .system and included in __all__.

  from .system import (
      get_system_info,
      check_disk_space,
      get_pacman_cache_stats,
      analyze_storage,
      check_failed_services,
      get_boot_logs,
      diagnose_system,
  )

• src/arch_ops_server/tool_metadata.py:185-193 (schema)

  Tool metadata registration: defines analyze_storage as a 'monitoring' tool with 'read' permission in the 'diagnose' workflow.

  "analyze_storage": ToolMetadata(
      name="analyze_storage",
      category="monitoring",
      platform="any",
      permission="read",
      workflow="diagnose",
      related_tools=["check_failed_services"],
      prerequisite_tools=[]
  ),