Added RebalanceService to rebalance tasks from unavailable to available hosts (#43)

* Added RebalanceService to rebalance tasks from unavailable to available hosts

Author: ppavlovic

src/Model/Domain/TaskerPool/DiscoveryService.php

@@ -32,6 +32,14 @@ public function notify()
      */
     public function getAvailableHostnames()
     {
-        return  implode(Identifier::DELIMITER, $this->repo->getAvailableHostnames());
+        return  implode(Identifier::DELIMITER, $this->getAvailableHostnamesAsArray());
+    }
+
+    /**
+     * @return array
+     */
+    public function getAvailableHostnamesAsArray()
+    {
+        return $this->repo->getAvailableHostnames();
     }
 }

src/Model/Repository/Mysql/TaskRepository.php

@@ -15,6 +15,9 @@ class TaskRepository implements TaskRepositoryInterface
     const RESET_TASKS_AFTER_SECONDS  = 60;  // seconds to retry failed tasks
     const RESET_TASKS_LIMIT = 20;           // how many tasks to reset at once
 
+    const REBALANCE_LIMIT = 1000;           // how many tasks to rebalance at once
+    const REBALANCE_TIME_IN_FUTURE = 3600;  // seconds in future tasks to rebalance
+
     /**
      * @var \PDO
      */
@@ -100,6 +103,30 @@ private function fetchTasks($status, $olderThanSeconds, $limit)
         }, $stmt->fetchAll());
     }
 
+    /**
+     * @param array $availableHostnames
+     * @return array|int[]
+     */
+    public function findTasksForRebalance(array $availableHostnames)
+    {
+        $query = sprintf('SELECT task_id FROM %s
+               WHERE
+                   identifier NOT IN (:availableHostnames) AND status=:status AND ts_created <= :ts_created
+               ORDER BY ts_created ASC LIMIT :limit',
+            Consts::TASKS_TABLE_NAME
+        );
+
+        $stmt = $this->pdo->prepare($query);
+        $stmt->bindValue(':availableHostnames', implode(',', $availableHostnames));
+        $stmt->bindValue(':status', Consts::STATUS_PENDING, \PDO::PARAM_INT);
+        $stmt->bindValue(':limit', self::REBALANCE_LIMIT, \PDO::PARAM_INT);
+        $stmt->bindValue(':ts_created', time() + self::REBALANCE_TIME_IN_FUTURE, \PDO::PARAM_INT);
+
+        $stmt->execute();
+
+        return $stmt->fetchAll(\PDO::FETCH_COLUMN);
+    }
+
     private function getIdentifier()
     {
         if ($this->identifier === null) {
@@ -157,7 +184,6 @@ public function addBulk($tasks)
         $sql .= implode(', ', $insertQuery);
 
         $stmt = $this->pdo->prepare($sql);
-        $stmt->bindValue(':mudo', Consts::TASKS_TABLE_NAME);
         $this->execute($stmt,$insertData);
     }
 
@@ -193,6 +219,27 @@ public function updateStatus($status, Task ...$tasks)
         );
     }
 
+    /**
+     * @param string$identifier
+     * @param array $taskIds
+     * @return void
+     */
+    public function updateIdentifier($identifier, array $taskIds)
+    {
+        if (count($taskIds) === 0) {
+            return;
+        }
+
+        $query= sprintf(
+            'UPDATE %s SET identifier="%s" WHERE task_id IN (%s)',
+            Consts::TASKS_TABLE_NAME,
+            $identifier,
+            implode(',', $taskIds)
+        );
+
+        $this->pdo->exec($query);
+    }
+
     /**
      * @param \PDOStatement $stmt
      * @param Task $task
@@ -228,4 +275,4 @@ private function execute(\PDOStatement $stmt, $data = null)
         }
         return $res;
     }
-}
+}

src/Tasker2/Exception/NoAvailableHostsException.php

@@ -0,0 +1,7 @@
+<?php
+
+namespace G4\Tasker\Tasker2\Exception;
+
+class NoAvailableHostsException extends \RuntimeException
+{
+}

src/Tasker2/RebalanceService.php

@@ -0,0 +1,74 @@
+<?php
+
+namespace G4\Tasker\Tasker2;
+
+use G4\Tasker\Model\Domain\TaskerPool\DiscoveryService;
+use G4\Tasker\Model\Repository\Mysql\TaskRepository;
+use G4\Tasker\Tasker2\Exception\NoAvailableHostsException;
+
+class RebalanceService
+{
+    /**
+     * @var DiscoveryService
+     */
+    private $discoveryService;
+
+    /**
+     * @var TaskRepository
+     */
+    private $taskRepository;
+
+    /**
+     * @var \G4\Log\ErrorLogger
+     */
+    private $errorLogger;
+
+    public function __construct(DiscoveryService $discoveryService, TaskRepository $taskRepository)
+    {
+        $this->discoveryService = $discoveryService;
+        $this->taskRepository = $taskRepository;
+    }
+
+    public function setErrorLogger(\G4\Log\ErrorLogger $errorLogger)
+    {
+        $this->errorLogger = $errorLogger;
+        return $this;
+    }
+
+    /**
+     * @return void
+     */
+    public function rebalance()
+    {
+        $availableHostnames = $this->discoveryService->getAvailableHostnamesAsArray();
+
+        if (count($availableHostnames) === 0) {
+            $this->errorLogger !== null && $this->errorLogger->log(new NoAvailableHostsException(
+                'No available hosts found for ' . __CLASS__ . __METHOD__
+            ));
+            return;
+        }
+
+        $tasksForRebalance = $this->taskRepository->findTasksForRebalance($availableHostnames);
+        if (count($tasksForRebalance) === 0) {
+            return;
+        }
+
+        foreach ($tasksForRebalance as $taskId) {
+            $index = array_rand($availableHostnames);
+            $tasks[$availableHostnames[$index]][] = (int) $taskId;
+        }
+
+        print("Tasks rebalance results:\n");
+        foreach ($tasks as $identifier => $taskIds) {
+            $this->taskRepository->updateIdentifier($identifier, $taskIds);
+            printf(
+                "[%s] - Hostname %s rebalanced %d tasks to %s\n",
+                date('Y-m-d H:i:s'),
+                gethostname(),
+                count($taskIds),
+                $identifier
+            );
+        }
+    }
+}