app/Services/ProcessManager/SupervisorService.php - master - itp_practice/itp_backend

app/Services/ProcessManager/SupervisorService.php
501 строка15 KB
vivanenko
profile edit
29 май 2025, 19:57
29 май 2025, 19:572ec716b
<?php

namespace App\Services\ProcessManager;

use GuzzleHttp\Client;
use Illuminate\Support\Facades\Log;

class SupervisorService
{
    private Client $client;
    private string $baseUrl;
    private array $auth;
    private string $configPath;

    public function __construct()
    {
        $this->baseUrl = config('services.supervisor.url', 'http://taps:9001/RPC2');
        $this->auth = [
            config('services.supervisor.username', 'admin'),
            config('services.supervisor.password', 'secret')
        ];
        
        $this->client = new Client([
            'base_uri' => $this->baseUrl,
            'auth' => $this->auth,
            'headers' => [
                'Content-Type' => 'text/xml',
            ],
        ]);
        
        // Base directory for supervisor configs
        $this->configPath = storage_path('supervisor/conf.d');
    }

    /**
     * Get config file path for specific tap process
     */
    private function getProcessConfigPath(int $tapId): string 
    {
        return $this->configPath . "/tap_process_{$tapId}.conf";
    }

    /**
     * Create and start a tap process with dynamic tap_id
     */
    public function startTapProcess(int $tapId): bool
    {
        try {
            $processName = "tap_process_{$tapId}";
            $configPath = $this->getProcessConfigPath($tapId);
            
            // Check if process already exists
            $processInfo = $this->getProcessInfo($processName);
            Log::info(__METHOD__, $processInfo ?? []);
            $force = (bool) $processInfo;

            if ($processInfo && in_array($processInfo['statename'], ['RUNNING', 'STARTING'])) {
                Log::info("Process {$processName} is already running");
                return true;
            }
            
            // Check if config file already exists
            if (file_exists($configPath)) {
                Log::info("Config file already exists for process {$processName}");
            } else {
                // Add process configuration
                $config = $this->buildTapProcessConfig($tapId);
                $added = $this->addProcessGroup($tapId, $config);
                
                if (!$added) {
                    Log::error("Failed to add process configuration for tap {$tapId}");
                    return false;
                }
            }
            
            // Reload supervisor to pick up new configuration
            $this->reloadConfig(!$force);
            
            // Start the process
            return $this->startProcess($processName);
            
        } catch (\Exception $e) {
            Log::error('Failed to start tap process', [
                'tap_id' => $tapId,
                'error' => $e->getMessage()
            ]);
            return false;
        }
    }

    /**
     * Stop a tap process
     */
    public function stopTapProcess(int $tapId): bool
    {
        $processName = "tap_process_{$tapId}";
        
        try {
            // Stop the process
            $stopped = $this->stopProcess($processName);
            
            if ($stopped) {
                // Remove process from configuration
                $this->removeProcessFromGroup($processName);
                $this->reloadConfig();
            }
            
            return $stopped;
        } catch (\Exception $e) {
            Log::error('Failed to stop tap process', [
                'tap_id' => $tapId,
                'error' => $e->getMessage()
            ]);
            return false;
        }
    }

    /**
     * Get status of a tap process
     */
    public function getTapProcessStatus(int $tapId): ?array
    {
        $processName = "tap_process_{$tapId}";
        return $this->getProcessInfo($processName);
    }

    /**
     * Get all tap processes
     */
    public function getAllTapProcesses(): array
    {
        $allProcesses = $this->getAllProcessInfo();
        $tapProcesses = [];
        
        foreach ($allProcesses as $process) {
            if (strpos($process['name'], 'tap_process_') === 0) {
                // Extract tap_id from process name
                $tapId = (int) str_replace('tap_process_', '', $process['name']);
                $process['tap_id'] = $tapId;
                $tapProcesses[] = $process;
            }
        }
        
        return $tapProcesses;
    }

    /**
     * Ensure logs directory exists and is writable
     */
    private function ensureLogsDirectory(): void
    {
        $logsDir = storage_path('logs/taps');
        if (!is_dir($logsDir)) {
            mkdir($logsDir, 0755, true);
        }
    }

    /**
     * Build process configuration for a tap
     */
    private function buildTapProcessConfig(int $tapId): string
    {
        $processName = "tap_process_{$tapId}";
        $this->ensureLogsDirectory();
        
        return "[program:{$processName}]
command=python3 -u /var/www/html/run_tapper.py {$tapId}
directory=/var/www/html
autostart=true
autorestart=true
stderr_logfile=/var/www/html/logs/tap_{$tapId}.err.log
stdout_logfile=/var/www/html/logs/tap_{$tapId}.out.log
stdout_logfile_maxbytes=50MB
stdout_logfile_backups=10
stdout_capture_maxbytes=1MB
redirect_stderr=true
user=root
numprocs=1
process_name=%(program_name)s
environment=TAP_ID=\"{$tapId}\"";
    }

    /**
     * Reload Supervisor configuration to recognize new programs
     */
    public function reloadConfig($force = false): bool
    {
        try {
            if (!$force) {
                $result = $this->xmlRpcCall('supervisor.reloadConfig');
            } else {
                $result = $this->xmlRpcCall('supervisor.restart');
            }

            sleep(5);
            
            // reloadConfig returns [string $status, array $changes]
            if (is_array($result) && count($result) >= 2) {
                Log::info('Supervisor configuration reloaded', [
                    'status' => $result[0],
                    'changes' => $result[1]
                ]);
                
                // If there are added groups, add them explicitly
                if (isset($result[1]['added']) && is_array($result[1]['added'])) {
                    foreach ($result[1]['added'] as $groupName) {
                        try {
                            $this->xmlRpcCall('supervisor.addProcessGroup', [$groupName]);
                            Log::info("Added process group: {$groupName}");
                        } catch (\Exception $e) {
                            Log::warning("Failed to add process group: {$groupName}", [
                                'error' => $e->getMessage()
                            ]);
                        }
                    }
                }
            }

            return true;
        } catch (\Exception $e) {
            Log::error('Failed to reload Supervisor configuration', [
                'error' => $e->getMessage()
            ]);
            throw $e;
        }
    }

    /**
     * Add process group configuration
     */
    private function addProcessGroup(int $tapId, string $config): bool
    {
        try {
            // Ensure directory exists
            if (!is_dir($this->configPath)) {
                if (!mkdir($this->configPath, 0755, true)) {
                    Log::error("Failed to create directory: {$this->configPath}");
                    return false;
                }
            }
            
            $configPath = $this->getProcessConfigPath($tapId);
            
            if (file_put_contents($configPath, $config) === false) {
                Log::error("Failed to write config file: {$configPath}");
                return false;
            }
            
            Log::info("Successfully wrote config to: {$configPath}", [
                'config' => $config
            ]);
            
            return true;
        } catch (\Exception $e) {
            Log::error('Failed to add process group', ['error' => $e->getMessage()]);
            return false;
        }
    }

    /**
     * Remove process from group
     */
    private function removeProcessFromGroup(string $processName): bool
    {
        try {
            if (preg_match('/tap_process_(\d+)/', $processName, $matches)) {
                $tapId = $matches[1];
                $configPath = $this->getProcessConfigPath($tapId);
                
                if (file_exists($configPath)) {
                    if (unlink($configPath)) {
                        Log::info("Removed config file: {$configPath}");
                        return true;
                    }
                    Log::error("Failed to remove config file: {$configPath}");
                    return false;
                }
            }
            return true;
        } catch (\Exception $e) {
            Log::error('Failed to remove process from group', ['error' => $e->getMessage()]);
            return false;
        }
    }

    /**
     * Start a process
     */
    public function startProcess(string $processName): bool
    {
        try {
            $response = $this->xmlRpcCall('supervisor.startProcess', [$processName]);
            return $response === true;
        } catch (\Exception $e) {
            // Check if process is already started
            if (str_contains($e->getMessage(), 'ALREADY_STARTED')) {
                return true;
            }
            
            Log::error('Failed to start process', [
                'process' => $processName,
                'error' => $e->getMessage()
            ]);
            return false;
        }
    }

    /**
     * Stop a process
     */
    public function stopProcess(string $processName): bool
    {
        try {
            $response = $this->xmlRpcCall('supervisor.stopProcess', [$processName]);
            return $response === true;
        } catch (\Exception $e) {
            // Check if process is not running
            if (str_contains($e->getMessage(), 'NOT_RUNNING')) {
                return true;
            }
            
            Log::error('Failed to stop process', [
                'process' => $processName,
                'error' => $e->getMessage()
            ]);
            return false;
        }
    }

    /**
     * Get process info
     */
    public function getProcessInfo(string $processName): ?array
    {
        try {
            return $this->xmlRpcCall('supervisor.getProcessInfo', [$processName]);
        } catch (\Exception $e) {
            // Process might not exist
            return null;
        }
    }

    /**
     * Get all processes info
     */
    public function getAllProcessInfo(): array
    {
        try {
            return $this->xmlRpcCall('supervisor.getAllProcessInfo', []) ?? [];
        } catch (\Exception $e) {
            Log::error('Failed to get all processes info', [
                'error' => $e->getMessage()
            ]);
            return [];
        }
    }

    /**
     * Restart a process
     */
    public function restartProcess(string $processName): bool
    {
        $this->stopProcess($processName);
        sleep(1); // Give it time to stop
        return $this->startProcess($processName);
    }

    /**
     * Read process stdout log
     */
    public function readProcessStdoutLog(string $processName, int $offset = 0, int $length = 1000): ?string
    {
        try {
            $result = $this->xmlRpcCall('supervisor.readProcessStdoutLog', [$processName, $offset, $length]);
            return $result['log'] ?? null;
        } catch (\Exception $e) {
            Log::error('Failed to read process stdout log', [
                'process' => $processName,
                'error' => $e->getMessage()
            ]);
            return null;
        }
    }

    /**
     * Read tap process log
     */
    public function readTapProcessLog(int $tapId, int $offset = 0, int $length = 1000): ?string
    {
        $processName = "tap_process_{$tapId}";
        return $this->readProcessStdoutLog($processName, $offset, $length);
    }

    /**
     * Make XML-RPC call
     */
    private function xmlRpcCall(string $method, array $params = [])
    {
        $xml = $this->buildXmlRpcRequest($method, $params);
        
        $response = $this->client->post('', [
            'body' => $xml,
        ]);

        return $this->parseXmlRpcResponse($response->getBody()->getContents());
    }

    /**
     * Build XML-RPC request
     */
    private function buildXmlRpcRequest(string $method, array $params): string
    {
        $xml = '<?xml version="1.0"?>';
        $xml .= '<methodCall>';
        $xml .= '<methodName>' . $method . '</methodName>';
        $xml .= '<params>';
        
        foreach ($params as $param) {
            $xml .= '<param><value>';
            $xml .= $this->xmlRpcValue($param);
            $xml .= '</value></param>';
        }
        
        $xml .= '</params>';
        $xml .= '</methodCall>';
        
        return $xml;
    }

    /**
     * Convert PHP value to XML-RPC value
     */
    private function xmlRpcValue($value): string
    {
        if (is_int($value)) {
            return '<int>' . $value . '</int>';
        } elseif (is_bool($value)) {
            return '<boolean>' . ($value ? '1' : '0') . '</boolean>';
        } elseif (is_string($value)) {
            return '<string>' . htmlspecialchars($value) . '</string>';
        } elseif (is_array($value)) {
            $xml = '<array><data>';
            foreach ($value as $item) {
                $xml .= '<value>' . $this->xmlRpcValue($item) . '</value>';
            }
            $xml .= '</data></array>';
            return $xml;
        }
        
        return '<string></string>';
    }

    /**
     * Parse XML-RPC response
     */
    private function parseXmlRpcResponse(string $response)
    {
        $xml = simplexml_load_string($response);
        
        if (isset($xml->fault)) {
            $faultCode = (int) $xml->fault->value->struct->member[0]->value->int;
            $faultString = (string) $xml->fault->value->struct->member[1]->value->string;
            throw new \Exception("XML-RPC Fault: [$faultCode] $faultString");
        }
        
        if (isset($xml->params->param->value)) {
            return $this->parseXmlRpcValue($xml->params->param->value);
        }
        
        return null;
    }

    /**
     * Parse XML-RPC value
     */
    private function parseXmlRpcValue($value)
    {
        if (isset($value->boolean)) {
            return (bool) (int) $value->boolean;
        } elseif (isset($value->int) || isset($value->i4)) {
            return (int) ($value->int ?? $value->i4);
        } elseif (isset($value->string)) {
            return (string) $value->string;
        } elseif (isset($value->array)) {
            $result = [];
            foreach ($value->array->data->value as $item) {
                $result[] = $this->parseXmlRpcValue($item);
            }
            return $result;
        } elseif (isset($value->struct)) {
            $result = [];
            foreach ($value->struct->member as $member) {
                $key = (string) $member->name;
                $result[$key] = $this->parseXmlRpcValue($member->value);
            }
            return $result;
        }
        
        return (string) $value;
    }
}