Как гарантировать выполнение каждой Jenkins Jobs из базы данных только одним Worker?

Обеспечение уникального выполнения Jenkins Job одним Worker

Эта задача относится к области распределенных систем и требует решения проблемы гоночных условий (race conditions) при обработке задач из базы данных несколькими параллельными воркерами.

Основные подходы и стратегии

1. Блокировки на уровне базы данных

Самый надежный подход — использование транзакционных блокировок в СУБД.

-- Пример с PostgreSQL и advisory locks
BEGIN;
SELECT pg_advisory_lock(job_id) FROM jobs WHERE status = 'pending' 
ORDER BY created_at LIMIT 1 FOR UPDATE SKIP LOCKED;

-- Работа с полученной задачей
COMMIT; -- блокировка автоматически снимается

Преимущества:

• Гарантированная атомарность
• Автоматическое освобождение при отказе воркера
• Поддержка в большинстве СУБД

2. Механизм аренды (Lease) с временными метками

Каждый воркер "арендует" задачу на определенное время.

class JobProcessor {
    private $leaseTimeout = 300; // 5 минут
    
    public function acquireJob() {
        $now = time();
        $connection = $this->getDatabaseConnection();
        
        $connection->beginTransaction();
        
        // Атомарное получение и блокировка задачи
        $job = $connection->executeQuery(
            "SELECT * FROM jobs 
             WHERE status = 'pending' 
             AND (leased_by IS NULL OR lease_expires < :now)
             ORDER BY priority DESC, created_at 
             LIMIT 1 FOR UPDATE",
            ['now' => $now]
        )->fetch();
        
        if ($job) {
            // Обновление с блокировкой
            $connection->executeUpdate(
                "UPDATE jobs SET 
                 status = 'processing',
                 leased_by = :workerId,
                 lease_expires = :expires,
                 started_at = :now
                 WHERE id = :id AND status = 'pending'",
                [
                    'workerId' => $this->workerId,
                    'expires' => $now + $this->leaseTimeout,
                    'now' => $now,
                    'id' => $job['id']
                ]
            );
            
            if ($connection->commit()) {
                return $job;
            }
        }
        
        $connection->rollBack();
        return null;
    }
}

3. Использование Redis для распределенных блокировок

Redis идеально подходит для координации распределенных процессов.

class DistributedJobLock {
    private $redis;
    private $lockTimeout = 30;
    
    public function tryAcquireJob($jobId) {
        $lockKey = "job_lock:{$jobId}";
        $workerId = uniqid('worker_', true);
        
        // SET с NX и EX для атомарной установки блокировки
        $acquired = $this->redis->set(
            $lockKey,
            $workerId,
            ['nx', 'ex' => $this->lockTimeout]
        );
        
        if ($acquired) {
            // Обновляем статус задачи в БД
            $this->markJobAsProcessing($jobId, $workerId);
            return true;
        }
        
        return false;
    }
    
    public function releaseJob($jobId, $workerId) {
        $lockKey = "job_lock:{$jobId}";
        
        // Lua-скрипт для атомарной проверки и удаления
        $script = "
            if redis.call('GET', KEYS[1]) == ARGV[1] then
                return redis.call('DEL', KEYS[1])
            else
                return 0
            end
        ";
        
        return $this->redis->eval($script, [$lockKey, $workerId], 1);
    }
}

4. Паттерн "Очередь сообщений"

Использование специализированных систем очередей вместо прямой работы с БД.

# Конфигурация Jenkins Pipeline
pipeline {
    agent any
    
    parameters {
        choice(
            name: 'EXECUTION_GROUP',
            choices: ['group1', 'group2', 'group3'],
            description: 'Группа для уникального выполнения'
        )
    }
    
    options {
        disableConcurrentBuilds()
        lock(resource: "job-${params.EXECUTION_GROUP}")
    }
    
    stages {
        stage('Process') {
            steps {
                script {
                    // Уникальная обработка гарантирована блокировкой
                    processDatabaseJob(env.JOB_NAME)
                }
            }
        }
    }
}

5. Комбинированная стратегия с идемпотентностью

Для максимальной надежности комбинируем несколько подходов.

class JobOrchestrator {
    public function processJobSafely($jobId) {
        // 1. Проверка идемпотентности
        if ($this->isAlreadyProcessed($jobId)) {
            return 'already_processed';
        }
        
        // 2. Попытка получить блокировку в Redis
        if (!$this->acquireDistributedLock($jobId)) {
            return 'locked_by_other';
        }
        
        try {
            // 3. Блокировка в БД
            $job = $this->acquireDatabaseLock($jobId);
            
            if (!$job) {
                throw new Exception('Failed to acquire DB lock');
            }
            
            // 4. Выполнение задачи
            $result = $this->executeJenkinsJob($job);
            
            // 5. Отметка как выполненной
            $this->markAsCompleted($jobId, $result);
            
            return 'success';
            
        } finally {
            // 6. Гарантированное освобождение блокировок
            $this->releaseDistributedLock($jobId);
        }
    }
}

Критические аспекты реализации:

1. Таймауты и heartbeat — воркеры должны периодически обновлять время аренды
2. Обработка сбоев — автоматическое освобождение блокировок при падении воркера
3. Мониторинг — отслеживание зависших задач и deadlock'ов
4. Ретри логика — повторные попытки при временных сбоях
5. Приоритизация — механизм для обработки срочных задач

Рекомендации для Jenkins:

• Используйте Lockable Resources Plugin для координации выполнения
• Настройте Throttle Concurrent Builds Plugin для контроля параллелизма
• Реализуйте здоровые checks для воркеров
• Внедрите механизм heartbeat для отслеживания активности

Наиболее надежным решением является комбинация Redis для распределенных блокировок и транзакционных блокировок в БД, что обеспечивает как высокую производительность, так и гарантированную уникальность выполнения каждой задачи даже в распределенной среде с десятками воркеров.