Android Device Farm 系统设计 (2)：MVP 到完整架构

# Device 状态模型
class DeviceState(Enum):
    OFFLINE = "offline"           # 物理故障或网络断线
    IDLE = "idle"                 # 在线且可用
    RESERVED = "reserved"         # 被预留，测试进行中
    RECOVERING = "recovering"     # 故障恢复中

@dataclass
class Device:
    serial: str                    # adb serial
    model: str                     # Pixel 6, OnePlus 10, ...
    android_version: str           # 13, 14, ...
    state: DeviceState             # 当前状态
    reserved_by: Optional[str]     # 谁预留的（测试名称）
    last_ping_time: datetime       # 最后健康检查时间
    fail_count: int = 0            # 连续失败次数

# 目录结构
device_farm/
├── data/
│  ├── devices.json       # 设备定义（静态）
│  ├── state.json         # 设备运行时状态（动态）
│  └── logs/
│     └── device_farm.log # 操作日志
└── scripts/
   ├── device_farm.py     # 核心逻辑
   └── monitor.py         # 健康检查

from threading import Lock, RLock

class DevicePool:
    def __init__(self):
        self.devices = {}
        self.lock = RLock()  # 可重入锁（同个线程可以多次获取）
        self.device_locks = defaultdict(Lock)  # 每个设备一个锁
    
    def reserve_device(self, criteria=None) -> Optional[str]:
        with self.lock:
            # 查找匹配设备
            candidate = self._find_idle_device(criteria)
            if not candidate:
                return None
            
            # 获取该设备的锁
            with self.device_locks[candidate['serial']]:
                # 再次检查状态（double-check pattern）
                if candidate['state'] != 'idle':
                    return None  # 被其他线程抢先预留了
                
                # 原子更新
                candidate['state'] = 'reserved'
                candidate['reserved_by'] = threading.current_thread().name
                self._save_state()  # 持久化
                
                return candidate['serial']

def health_check_loop(self):
    """后台线程：定期检查设备健康状态"""
    while True:
        time.sleep(60)  # 每 60 秒检查一次
        
        for serial, device in self.devices.items():
            if device['state'] == 'offline':
                continue  # 已离线的设备，跳过
            
            # 1. Ping 设备
            alive = self._ping_device(serial)
            
            if alive:
                device['fail_count'] = 0  # 恢复计数重置
                continue
            
            # 2. 失败计数
            device['fail_count'] += 1
            print(f"[HEALTH] {serial} ping failed (count={device['fail_count']})")
            
            # 3. 超过阈值则标记离线
            if device['fail_count'] >= 3:
                device['state'] = 'offline'
                device['reserved_by'] = None  # 强制释放
                self._alert(f"Device {serial} marked OFFLINE")
            
            # 4. 尝试恢复（可选，MVP 中简单实现）
            elif device['fail_count'] == 1:
                self._try_recovery(serial)
    
    def _ping_device(self, serial: str) -> bool:
        """用 adb 检查设备是否在线"""
        try:
            result = subprocess.run(
                ["adb", "-s", serial, "shell", "echo", "OK"],
                capture_output=True,
                timeout=10
            )
            return result.returncode == 0
        except subprocess.TimeoutExpired:
            return False
    
    def _try_recovery(self, serial: str):
        """尝试恢复离线设备"""
        print(f"[RECOVERY] Attempting to recover {serial}...")
        try:
            # 1. detach
            subprocess.run(
                ["usbip", "detach", "-p", self._get_port(serial)],
                timeout=10
            )
            time.sleep(5)
            
            # 2. attach
            subprocess.run(
                ["usbip", "attach", "-r", self.server_ip, "-b", 
                 self.devices[serial]['busid']],
                timeout=10
            )
            
            # 3. 等待 adb 识别
            for _ in range(20):
                if self._ping_device(serial):
                    print(f"[RECOVERY] {serial} recovered!")
                    return
                time.sleep(1)
            
            print(f"[RECOVERY] {serial} recovery failed")
        except Exception as e:
            print(f"[RECOVERY] Error: {e}")

from flask import Flask, request, jsonify

app = Flask(__name__)
pool = DevicePool(...)

@app.route('/devices', methods=['GET'])
def list_devices():
    """列出所有设备及其状态"""
    return jsonify({
        'devices': list(pool.devices.values()),
        'summary': {
            'total': len(pool.devices),
            'idle': sum(1 for d in pool.devices.values() if d['state'] == 'idle'),
            'reserved': sum(1 for d in pool.devices.values() if d['state'] == 'reserved'),
            'offline': sum(1 for d in pool.devices.values() if d['state'] == 'offline'),
        }
    })

@app.route('/devices/reserve', methods=['POST'])
def reserve_device():
    """预留一个设备"""
    data = request.json
    model = data.get('model')
    android_version = data.get('android_version')
    timeout = data.get('timeout', 30)  # 秒
    
    # 轮询等待，直到找到可用设备或超时
    start_time = time.time()
    while time.time() - start_time < timeout:
        serial = pool.reserve_device({
            'model': model,
            'android_version': android_version
        })
        
        if serial:
            return jsonify({'status': 'ok', 'serial': serial})
        
        time.sleep(1)  # 等待 1 秒后重试
    
    return jsonify({'status': 'error', 'reason': 'timeout'}), 503

@app.route('/devices/<serial>/release', methods=['POST'])
def release_device(serial):
    """释放一个设备"""
    pool.release_device(serial)
    return jsonify({'status': 'ok'})

@app.route('/status', methods=['GET'])
def get_status():
    """获取系统整体状态"""
    return jsonify(pool.export_metrics())

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, threaded=True)

@dataclass
class Lease:
    id: str                    # 租赁 ID
    device_serial: str         # 设备序列号
    reserved_by: str           # 租赁人（用户/流程名）
    start_time: datetime       # 开始时间
    expected_duration: int     # 预期租赁时长（秒）
    status: str                # active / expired / released
    
class DevicePool:
    def reserve_device_with_lease(self, user_id: str, duration_sec: int):
        """预留设备并创建租赁记录"""
        serial = self.reserve_device()
        
        lease = Lease(
            id=str(uuid.uuid4()),
            device_serial=serial,
            reserved_by=user_id,
            start_time=datetime.now(),
            expected_duration=duration_sec
        )
        
        self.leases[lease.id] = lease
        return lease

    def enforce_lease_timeout(self):
        """后台线程：强制执行租赁超时"""
        while True:
            time.sleep(30)
            
            for lease_id, lease in self.leases.items():
                if lease.status != 'active':
                    continue
                
                elapsed = (datetime.now() - lease.start_time).total_seconds()
                if elapsed > lease.expected_duration * 1.5:  # 允许 50% 超期
                    # 强制释放
                    self.release_device(lease.device_serial)
                    lease.status = 'expired'
                    
                    alert(f"Lease {lease_id} expired, device released")

class SmartDevicePool(DevicePool):
    def get_similar_device(self, serial: str):
        """找到一个相同型号的设备"""
        device = self.devices[serial]
        
        # 查找其他在线的相同设备
        for d in self.devices.values():
            if (d['model'] == device['model'] and 
                d['android_version'] == device['android_version'] and
                d['state'] == 'idle' and
                d['host_ip'] != device['host_ip']):  # 不同主机
                return d
        
        return None
    
    def handle_device_failure(self, serial: str, test_context: dict):
        """
        设备故障处理：
        1. 尝试恢复原设备
        2. 如果失败，转移到备用设备
        3. 通知测试框架
        """
        device = self.devices[serial]
        
        # 1. 尝试本地恢复（2 次）
        for attempt in range(2):
            self._try_recovery(serial)
            if self._ping_device(serial):
                print(f"Recovery succeeded for {serial}")
                return None  # 恢复成功
        
        # 2. 尝试自动转移
        alternative = self.get_similar_device(serial)
        if alternative:
            print(f"Failover: {serial} -> {alternative['serial']}")
            # 强制释放故障设备
            self.devices[serial]['state'] = 'offline'
            # 预留备用设备
            self.reserve_device({'model': device['model']})
            return alternative['serial']
        
        # 3. 没有备用设备
        print(f"No similar device available for failover")
        return None

def capacity_analysis(self):
    """容量分析：预测是否需要扩展"""
    # 1. 计算利用率指标
    total = len(self.devices)
    idle = sum(1 for d in self.devices.values() if d['state'] == 'idle')
    utilization = (total - idle) / total if total > 0 else 0
    
    # 2. 分析预留等待时间（过去 1 小时）
    recent_waits = self._get_reservation_wait_times(hours=1)
    avg_wait = sum(recent_waits) / len(recent_waits) if recent_waits else 0
    
    # 3. 触发告警的条件
    if utilization > 0.9:
        alert(f"Device utilization too high: {utilization*100:.1f}%")
    
    if avg_wait > 60:  # 平均等待超过 1 分钟
        alert(f"Average reservation wait time: {avg_wait}s (recommend +3 devices)")
    
    # 4. 生成容量规划报告
    return {
        'current_devices': total,
        'utilization': utilization,
        'avg_wait_time': avg_wait,
        'recommended_devices': self._calculate_needed_capacity()
    }

def _calculate_needed_capacity(self):
    """根据历史数据和增长趋势，推荐需要多少设备"""
    # 简单方案：确保 p99 的预留延迟 < 30 秒
    peak_utilization = self._get_peak_utilization(hours=24)
    
    # Little's Law: L = λ * W
    # L: 平均占用设备数
    # λ: 预留速率（requests/sec）
    # W: 测试平均耗时（sec）
    
    needed_devices = ceil(peak_utilization / (1 - target_wait_ratio))
    return needed_devices