taskman_atf/session.inc.php

<?php
namespace ATF;

use Exception;
use taskman;
use Amp;

class SessionConfig
{
  public string $atf_host;
  public ?string $apk_path = '';
  public string $proj_branch;
  public string $proj_version;
  public string $proj_rev_hash;
  public bool $override_external_files = false;
  public array $external_files = array();
  public int $max_adb_installs_in_progress = 5;
  public bool $adb_reboot = true;
  public bool $reboot_devices = false;
  public bool $error_pause = false;
  public int $sleep_time = 5;
  public int $dead_threshold = 60; 
  public int $stuck_threshold = 80;
  public bool $share_with_qa_chan = true;
  public IMChan $im;
  public IMChan $im_qa;
  public IStatsSender $stats;
  public string $device_replay_path = 'replay.txt';
  public array $stats_event_listeners = array();

  function __construct()
  {
    $this->im = new NullMessenger();
    $this->im_qa = new NullMessenger();

    $this->stats = new NullStatsSender();
  }

  function initHost(array $settings)
  {
    $this->atf_host = 'atf';
    taskman\deploy_declare_node($this->atf_host, $settings);
  }

  function registerStatsEventListener(callable $hook)
  {
    $this->stats_event_listeners[] = $hook;
  }
}

class Session
{
  const RUN_SLEEP = 3;
  const IGNORE_DEVICE_TTL = 300;

  public string $name;
  public SessionConfig $conf;

  public int $start_time;

  public IDevicePool $device_pool;
  public ApkInstaller $apk_installer;

  public string $id;
  public string $guid;

  public string $im_thread_id = '';

  public array $plans = array();

  public array $ignored_devices = array();
  public array $bogus_device2counter = array();

  public array $shared_qa_errors = array();

  static function getId() : string 
  {
    $id = getenv('ATF_SESSION_ID');
    if($id)
      return $id;
    //sort of graceful fallback
    return ''.getmypid();
  }

  function __construct(
    string $name, 
    SessionConfig $conf, 
    IDevicePool $device_pool
  )
  {
    $this->id = self::getId();

    $this->name = $name;
    $this->conf = $conf;

    $this->guid = uniqid();

    $this->device_pool = $device_pool;
  }

  function getTitle() : string
  {              
    $finished_plans = $this->countFinishedPlans();
    return "**Session ({$this->id}) '{$this->name}' (plans:$finished_plans/".sizeof($this->plans)." devices:".sizeof($this->getDevices())." ?:".sizeof($this->ignored_devices) . 
      ') (' . round($this->getDuration()/60, 1) . ' min)**';
  }

  function getDuration() : int
  {
    return time() - $this->start_time;
  }

  function getDevices() : array
  {
    $this->_checkIgnoredDevices();

    $connected_devices = $this->device_pool->get();
    $actual_devices = array_diff($connected_devices, array_keys($this->ignored_devices));

    return $actual_devices;
  }

  function _checkIgnoredDevices()
  {
    //let's remove from ignored devices devices which were ignored for some time 
    $devices = array_keys($this->ignored_devices);
    foreach($devices as $device)
    {
      if(time() - $this->ignored_devices[$device] > self::IGNORE_DEVICE_TTL)
        unset($this->ignored_devices[$device]);
    }
  }

  function ignoreDevice(string $device, string $reason)
  {
    err("Ignoring device *$device*: $reason");

    $this->ignored_devices[$device] = time();
  }

  function incBogusDevice(string $device)
  {
    if(!isset($this->bogus_device2counter[$device]))
      $this->bogus_device2counter[$device] = 0;
    return ++$this->bogus_device2counter[$device];
  }

  function resetBogusDeviceCount(string $device)
  {
    $this->bogus_device2counter[$device] = 0;
  }

  function resetBogusDevices()
  {
    $this->bogus_device2counter = array();
  }

  function addPlan(Plan $plan)
  {
    $plan->session = $this;
    $this->plans[] = $plan;
  }

  function resetPlans()
  {
    $this->plans = array();
  }

  function runAsync() : Amp\Promise
  {
    return Amp\call(function() {

    $this->start_time = time();

    $apk_name = !$this->conf->apk_path ? null : basename($this->conf->apk_path);
    $this->apk_installer = new ApkInstaller(
      $this->conf->atf_host,
      $this->conf->max_adb_installs_in_progress,
      $apk_name, 
      $this->conf->override_external_files, 
      $this->conf->external_files
    );

    $this->_createMessengerThread();

    //if it's not set the last one will be re-used
    if($this->conf->apk_path)
      $this->_deployApkToHost($apk_name);

    if($this->conf->reboot_devices)
      reboot_devices($this->conf->atf_host, $this->getDevices());

    if($this->conf->adb_reboot)
      adb_reboot($this->conf->atf_host);

    //TODO: should not be here
    $this->trySendApkStatsEvent($this->apk_installer->getSize());

    $all_tasks = $this->_getAllTasks();

    return yield $this->_runTasksAsync($all_tasks);

    });
  }

  function _createMessengerThread()
  {
    for($i=0;$i<5;++$i)
    {
      $resp = $this->conf->im->post("", $this->_getMessengerThreadProps()); 
      if(isset($resp['ok']))
      {
        $this->im_thread_id = $resp['id'];
        return;
      }
      sleep(1);
    }
    throw new Exception("Could not create IM thread");
  }

  function _updateMessengerThread()
  {
    $this->conf->im->updatePost(
      $this->im_thread_id,
      $this->_getMessengerThreadProps() 
    );
  }

  function _getMessengerThreadProps() : array
  {
    return ['props' => ['attachments' => [
      [
        'text' => $this->getTitle(), 
        'color' => $this->_getThreadColor()
      ]
    ]]];
  }

  function _getThreadColor() : string 
  {
    if(!$this->getDevices())
      return "#000000";

    $stats = new Stats();
    foreach($this->plans as $plan)
      $stats->add($plan->getStats());

    return $stats->getColor($this->isOver());
  }

  function _deployApkToHost($apk_name)
  {
    if(!is_file($this->conf->apk_path))
      throw new Exception("No such file '{$this->conf->apk_path}'");

    host_put_file($this->conf->atf_host, $this->conf->apk_path, "%{dir}%/$apk_name");
  }

  function _getAllTasks()
  {
    $all_tasks = array();
    foreach($this->plans as $plan)
    {
      foreach($plan->tasks as $task)
        $all_tasks[] = $task;
    }
    return $all_tasks;
  }

  function _runTasksAsync(array $all_tasks) : Amp\Promise
  {
    return Amp\call(function() use($all_tasks) {

    $active_tasks = array();

    while(!$this->isOver() && (sizeof($all_tasks) > 0 || sizeof($active_tasks) > 0))
    {
      $free_devices = $this->_findFreeDevices($active_tasks);

      $this->_assignTasksToDevices($all_tasks, $free_devices, $active_tasks);

      yield Amp\delay(self::RUN_SLEEP*1000);

      $this->_processDoneTasks($all_tasks, $active_tasks);

      $this->_updateMessengerThread();
    }

    return sizeof($all_tasks) == 0 && sizeof($active_tasks) == 0;

    });
  }

  function _assignTasksToDevices(array &$tasks, array $devices, array &$active_tasks)
  {
    static $next_run_id = 0;

    //create new coroutines and assign them to free devices
    foreach($devices as $device)
    {
      if(!$device)
        continue;

      if(sizeof($tasks) == 0)
        break;

      $task = array_shift($tasks);

      //explicitly assigning a device
      $task->device = $device;

      $task->run_id = ++$next_run_id;
      $task->run_status = false;

      $active_tasks[$task->run_id] = $task;

      $task->plan->runTaskAsync($task)->onResolve(
        function($error, $value) use($task)
        {
          $task->run_status = true;
          $task->run_error = $error;
        }
      );
    }
  }

  function _processDoneTasks(array &$tasks, array &$active_tasks)
  {
    $tmp_tasks = array();
    foreach($active_tasks as $run_id => $task)
    {
      //task not in done ids, let's keep it
      if(!$task->run_status)
      {
        $tmp_tasks[$run_id] = $task;
      }
      else if($task->run_error)
      {
        throw $task->run_error;
      }
      //task is done, let's check if it had a problem and if yes
      //let's put it back into unscheduled tasks
      else if($task->need_reschedule)
      {
        array_unshift($tasks, $task);
        $task->need_reschedule = false;
      }
    }

    $active_tasks = $tmp_tasks;
  }

  function _findFreeDevices(array $active_tasks)
  {
    $free = array();

    foreach($this->getDevices() as $device)
    {
      //let's skip invalid devices
      if(!$device)
        continue;

      $busy = false;
      foreach($active_tasks as $task)
      {
        if($task->device == $device)
        {
          $busy = true;
          break;
        }
      }
      if(!$busy)
        $free[] = $device;
    }

    return $free;
  }

  function isOver() : bool
  {
    if(!$this->getDevices())
      return true;

    foreach($this->plans as $plan)
      if(!$plan->isOver())
        return false;

    return true;
  }

  function countFinishedPlans() : int
  {
    $n = 0;
    foreach($this->plans as $plan)
      if($plan->isOver())
        $n++;
    return $n;
  }

  function trySendStatsFromJzonAsync(Task $task, string $jzon) : Amp\Promise
  {
    return Amp\call(function() use($task, $jzon) {

    try
    {
      $data = jzon_parse(trim(str_replace('\"', '"', $jzon)));
      $table = $data['table'];
      unset($data['table']);

      $data = yield $this->applyStatsEventsListeners($task, $table, $data);

      if($data !== null)
        $this->trySendStats($task, $table, $data);
    }
    catch(Exception $e)
    {
      echo $e->getMessage() . "\n";
    }

    });
  }

  function applyStatsEventsListeners(Task $task, string $table, array $data) : Amp\Promise
  {
    return Amp\call(function() use($task, $table, $data) {

      foreach($this->conf->stats_event_listeners as $hook)
        $data = yield $hook($this, $task, $table, $data);

      return $data;
    });
  }

  function trySendApkStatsEvent(int $apk_size)
  {
    try
    {
      $data = array();
      $data['guid'] = $this->guid; 
      $data['time'] = time(); 
      $data['version'] = $this->conf->proj_version; 
      $data['revHash'] = $this->conf->proj_rev_hash; 
      $data['branch'] = $this->conf->proj_branch; 
      $data['event'] = 'apk_size';
      $data['value'] = $apk_size;
      $this->conf->stats->send('event_value', $data);
    }
    catch(Exception $e)
    {
      echo $e->getMessage() . "\n";
    }
  }

  function trySendStats(Task $task, string $table, array $data)
  {
    try
    {
      $data['guid'] = $this->guid; 
      $data['time'] = time();
      $data['deviceId'] = $task->device;
      $this->conf->stats->send($table, $data);
    }
    catch(Exception $e)
    {
      echo $e->getMessage() . "\n";
    }
  }

  function trySendStatsEvent(Task $task, string $event, int $value = 1)
  {
    $this->trySendStats(
      $task,
      'event_value', 
      array(
        'event' => $event,
        'value' => $value, 
        'branch' => $this->conf->proj_branch, 
        'version' => $this->conf->proj_version,
        'revHash' => $this->conf->proj_rev_hash
      )
    );
  }

  function tryShareToQAChannel($msg_id, $error)
  {
    if(!$this->conf->share_with_qa_chan)
      return;

    //let's skip similar already shared errors 
    if($this->_calcSharedErrorSimilarity($error) > 80)
      return;
    
    $this->shared_qa_errors[] = $error;

    //let's share the exception message to the regency-qa channel
    $permalink = $this->conf->im->getPermalink($msg_id);
    $this->conf->im_qa->post($permalink);
  }

  function _calcSharedErrorSimilarity($error)
  {
    $max_perc = 0;
    foreach($this->shared_qa_errors as $shared)
    {
      $perc = 0;
      similar_text($shared, $error, $perc);
      if($perc > $max_perc)
        $max_perc = $perc;
    }
    return $max_perc;
  }
}