taskman_atf/deploy.inc.php

<?php
namespace taskman;
use Amp;  

require_once('composer/vendor/autoload.php');

$GLOBALS['DEPLOY_NODES'] = array();
$GLOBALS['DEPLOY_SSH_CONNS'] = array();

define('DEPLOY_OPT_ONE_HOST', 1);
define('DEPLOY_OPT_ERR_OK',   2);
define('DEPLOY_OPT_SILENT',   4);
define('DEPLOY_OPT_IS_FILE',  8);
define('DEPLOY_OPT_NO_CACHE', 16);

use phpseclib\Crypt\RSA;
use phpseclib\Net\SSH2;
use phpseclib\Net\SCP;

class DeployNode
{
  public $name;
  public $props = array();

  function __construct($name, $props = array())
  {
    $this->name = $name;
    $this->_setupProps($props);
  }

  function get($name)
  {
    if(!isset($this->props[$name]))
      throw new Exception("No such property '$name' in declaration '{$this->name}'");
    return $this->props[$name];
  }

  function set($name, $val)
  {
    $this->props[$name] = $val;
  }

  function has($name)
  {
    return isset($this->props[$name]);
  }

  private function _setupProps(array $props)
  {
    foreach($props as $k => $v)
      $this->props[$k] = $v;
  }
}

function deploy_declare_node($name, array $props)
{
  global $DEPLOY_NODES;

  if(isset($DEPLOY_NODES[$name]))
    throw new Exception("Declaration '$name' already exists");

  $decl = new DeployNode($name, $props);
  $deploy_dir = '/home/' . $decl->get('user') . '/' . $decl->get('dir');
  //for convenience
  $decl->set("deploy_dir", $deploy_dir);

  //checking deploy_dir conflicts
  foreach($DEPLOY_NODES as $other_name => $other_decl)
  {
    foreach($other_decl->get('hosts') as $other_host)
    {
      foreach($decl->get('hosts') as $host)
        if($host == $other_host && $other_decl->get('deploy_dir') === $deploy_dir)
          throw new Exception("Deploy directory '$deploy_dir' conflicts on nodes '$other_name' and '$name' for host '$host'");
    }
  }

  $DEPLOY_NODES[$name] = $decl;
}

function deploy_get_node($name)
{
  global $DEPLOY_NODES;

  if(!isset($DEPLOY_NODES[$name]))
    throw new Exception("Deploy node '{$name}' not found");

  return $DEPLOY_NODES[$name];
}

function deploy_find_node($name)
{
  global $DEPLOY_NODES;

  if(isset($DEPLOY_NODES[$name]))
    return $DEPLOY_NODES[$name];
  return null;
}

function deploy_get_node_names()
{
  global $DEPLOY_NODES;
  return array_keys($DEPLOY_NODES);
}

function deploy_get($name, $key)
{
  $decl = deploy_get_node($name);
  return $decl->get($key);
}

function deploy_set($name, $key, $val)
{
  $decl = deploy_get_node($name);
  $decl->set($key, $val);
}

function deploy_exec($name, $cmd, $opts = 0, $func = null)
{
  $decl = deploy_get_node($name);

  $outs = array();

  foreach($decl->get('hosts') as $host)
  {
    $ssh = deploy_get_ssh($decl, $host, $opts);
    $out = deploy_node_host_exec($decl, $host, $ssh, $cmd, $status, $opts);
    $outs[$host] = array($status, $out);

    if($func !== null)
      $func($decl, $host, $ssh, $out, $status);
    if(($opts & DEPLOY_OPT_ONE_HOST) != 0)
      break;
  }

  return $outs;
}

function deploy_node_host_exec($decl, $host, SSH2 $ssh, $cmd, &$status, $opts = 0)
{
  $cmd = deploy_str($decl, $cmd);
  if(($opts & DEPLOY_OPT_SILENT) == 0)
    echo "[EXE] $host: $cmd\n";
  $out = '';
  $res = $ssh->exec($cmd, function($str) use(&$out, $opts) {
    $out .= $str;
    if(($opts & DEPLOY_OPT_SILENT) == 0)
      echo $str;
    return false;
  });
  $status = $ssh->getExitStatus();
  if($res === false)
    throw new Exception("Fatal error executing($status): $cmd");
  if(($opts & DEPLOY_OPT_ERR_OK) == 0 && $status !== 0)
    throw new Exception("Invalid exit status($status) '$cmd': {$ssh->stdErrorLog}");
  return $out;
}

function deploy_ssh_exec($name, $cmd, $opts = 0)
{
  $decl = deploy_get_node($name);
  $cmd = deploy_str($decl, $cmd);
  $user = $decl->get('user');

  $tmp_key_file = tempnam("/tmp", "ssh_");
  $key_str = $decl->get('ssh_key_str');

  $outs = array();

  try
  {
    file_put_contents($tmp_key_file, $key_str);

    foreach($decl->get('hosts') as $host)
    {
      if(($opts & DEPLOY_OPT_SILENT) == 0)
        echo "[SSH] $host: $cmd\n";

      $host_only = $host;
      $port = 22;
      if(strpos($host, ":") !== false)
        list($host_only, $port) = explode(":", $host);

      $proc_cmd = "ssh -p $port -o StrictHostKeyChecking=no -o ConnectTimeout=90 -o ConnectionAttempts=30 -i $tmp_key_file $user@$host_only ".escapeshellarg($cmd);
      if($host === "localhost")
        $proc_cmd = $cmd;

      //echo "proc_cmd: $proc_cmd\n";
      $out = array();
      exec($proc_cmd, $out, $status);

      if(($opts & DEPLOY_OPT_SILENT) == 0 && $out)
        echo implode("\n", $out) . "\n";

      if(($opts & DEPLOY_OPT_ERR_OK) == 0 && $status !== 0)
        throw new Exception("Invalid exit status($status) '$cmd': " . implode("\n", $out));

      $outs[$host] = array($status, $out);

      if(($opts & DEPLOY_OPT_ONE_HOST) != 0)
        break;
    }
  }
  finally
  {
    unlink($tmp_key_file);
  }

  return $outs;
}

function deploy_ssh_exec_async($name, $cmd, $opts = 0)
{
  return Amp\call(function() use($name, $cmd, $opts) {

  $decl = deploy_get_node($name);
  $cmd = deploy_str($decl, $cmd);
  $user = $decl->get('user');

  $tmp_key_file = tempnam("/tmp", "ssh_");
  $key_str = $decl->get('ssh_key_str');

  $outs = array();

  try
  {
    file_put_contents($tmp_key_file, $key_str);

    foreach($decl->get('hosts') as $host)
    {
      if(($opts & DEPLOY_OPT_SILENT) == 0)
        echo "[SSH] $host: $cmd\n";

      $proc_cmd = "ssh -o StrictHostKeyChecking=no -o ConnectTimeout=90 -o ConnectionAttempts=30 -i $tmp_key_file $user@$host ".escapeshellarg($cmd);

      if($host === "localhost")
        $proc_cmd = $cmd;

      $proc = new Amp\Process\Process($proc_cmd);

      yield $proc->start();
      $out = yield Amp\ByteStream\buffer($proc->getStdout());
      $status = yield $proc->join();

      if(($opts & DEPLOY_OPT_ERR_OK) == 0 && $status !== 0)
        throw new Exception("Invalid exit status($status) '$cmd': $out");

      $outs[$host] = array($status, explode("\n", $out));

      if(($opts & DEPLOY_OPT_ONE_HOST) != 0)
        break;
    }
  }
  finally
  {
    unlink($tmp_key_file);
  }

  return $outs;

  });
}

function deploy_put_file($name, $path, $contents, $opts = 0)
{
  $decl = deploy_get_node($name);

  foreach($decl->get('hosts') as $host)
  {
    $ssh = deploy_get_ssh($decl, $host, $opts);
    $path = deploy_str($decl, $path);
    $scp = new SCP($ssh);
    if(($opts & DEPLOY_OPT_SILENT) == 0)
      echo "[PUT] $host: $path\n";

    $fails = 0;
    while($scp->put($path, $contents, ($opts & DEPLOY_OPT_IS_FILE) == 0 ? SCP::SOURCE_STRING : SCP::SOURCE_LOCAL_FILE) === false)
    {
      ++$fails;
      echo "Retrying a file put: $path...\n";
      sleep(1);
      if($fails > 5)
        throw new Exception("Could not put file: $path");
    }
  }
}

function deploy_scp_put_file($name, $local_path, $remote_path, $opts = 0)
{
  $decl = deploy_get_node($name);

  $user = $decl->get('user');

  $tmp_key_file = tempnam("/tmp", "ssh_");
  $key_str = $decl->get('ssh_key_str');

  try
  {
    file_put_contents($tmp_key_file, $key_str);

    foreach($decl->get('hosts') as $host)
    {
      if(($opts & DEPLOY_OPT_SILENT) == 0)
        echo "[PUT] $host: $local_path -> $remote_path\n";

      $host_only = $host;
      $port = 22;
      if(strpos($host, ":") !== false)
        list($host_only, $port) = explode(":", $host);

      $cmd = "scp -P $port -o StrictHostKeyChecking=no -o ConnectTimeout=90 -o ConnectionAttempts=30 -i $tmp_key_file $local_path $user@$host_only:$remote_path";
      if($host === "localhost")
        $cmd = "cp $local_path $remote_path";

      system($cmd, $ret);

      if($ret != 0)
        throw new Exception("Could not scp local $local_path to remote $remote_path: $ret");
    }
  }
  finally
  {
    unlink($tmp_key_file);
  }
}

function deploy_scp_put_file_async($name, $local_path, $remote_path, $opts = 0)
{
  return Amp\call(function() use($name, $local_path, $remote_path, $opts) {

  $decl = deploy_get_node($name);

  $user = $decl->get('user');

  $tmp_key_file = tempnam("/tmp", "ssh_");
  $key_str = $decl->get('ssh_key_str');

  try
  {
    file_put_contents($tmp_key_file, $key_str);

    foreach($decl->get('hosts') as $host)
    {
      if(($opts & DEPLOY_OPT_SILENT) == 0)
        echo "[PUT] $host: $local_path -> $remote_path\n";

      $proc_cmd = "scp -o StrictHostKeyChecking=no -o ConnectTimeout=90 -o ConnectionAttempts=30 -i $tmp_key_file $local_path $user@$host:$remote_path";
      if($host === "localhost")
        $proc_cmd = "cp $local_path $remote_path";

      $proc = new Amp\Process\Process($proc_cmd);

      yield $proc->start();

      $err_stream = $proc->getStderr();
      while(null !== $chunk = yield $err_stream->read())
        echo $chunk;

      $status = yield $proc->join();

      if($status !== 0)
        throw new Exception("Could not scp file $local_path to remote $remote_path: $status");
    }
  }
  finally
  {
    unlink($tmp_key_file);
  }
  });
}

function deploy_rsync_async($name, $src_dir, $dst_dir, $rsync_opts = '', $opts = 0)
{
  return Amp\call(function() use($name, $src_dir, $dst_dir, $rsync_opts, $opts) {

  $decl = deploy_get_node($name);

  $tmp_key_file = tempnam("/tmp", "ssh_");
  $key_str = $decl->get('ssh_key_str');

  try
  {
    file_put_contents($tmp_key_file, $key_str);

    $ssh_transport = "ssh -A -o StrictHostKeyChecking=no -o ConnectTimeout=90 -o ConnectionAttempts=30 -i $tmp_key_file";

    foreach($decl->get('hosts') as $host)
    {
      list($ssh_host, $ssh_port) = deploy_ssh_host_port($host);

      $proc_cmd = "rsync -e '" . $ssh_transport . " -p $ssh_port' -a $rsync_opts $src_dir/ " . $decl->get('user') . '@' . $ssh_host . ":$dst_dir/";

      if(($opts & DEPLOY_OPT_SILENT) == 0)
        echo "[RSN] $host: $proc_cmd\n";

      $proc = new Amp\Process\Process($proc_cmd);

      yield $proc->start();

      $err_stream = $proc->getStderr();
      while(null !== $chunk = yield $err_stream->read())
        echo $chunk;

      $status = yield $proc->join();

      if($status !== 0)
        throw new Exception("Could not rsync $src_dir to remote $dst_dir: $status");
    }
  }
  finally
  {
    unlink($tmp_key_file);
  }
  });
}

function deploy_get_file($name, $path, $opts = 0)
{
  $files = array();
  $decl = deploy_get_node($name);
  foreach($decl->get('hosts') as $host)
  {
    if(($opts & DEPLOY_OPT_SILENT) == 0)
      echo "[GET] $host: $path\n";

    if($host !== "localhost")
    {
      $ssh = deploy_get_ssh($decl, $host);
      $path = deploy_str($decl, $path);
      $scp = new SCP($ssh);
      $files[$host] = $scp->get($path);
    }
    else
      $files[$host] = file_get_contents($path);
  }
  return $files;
}

function deploy_ssh_host_port($host)
{
  $port = '22';

  $host_parts = explode(':', $host);
  if(isset($host_parts[1]))
    $port = $host_parts[1];
  $host = $host_parts[0];

  return array($host, $port);
}

function deploy_get_ssh(DeployNode $decl, $host, $opts = 0)
{
  global $DEPLOY_SSH_CONNS;

  $conn_id = $decl->name.'_'.$host;

  if(!isset($DEPLOY_SSH_CONNS[$conn_id]) || ($opts & DEPLOY_OPT_NO_CACHE) != 0)
  {
    $ssh = _deploy_make_ssh($decl, $host);
    $DEPLOY_SSH_CONNS[$conn_id] = $ssh;
  }
  else
  {
    $ssh = $DEPLOY_SSH_CONNS[$conn_id];
    //let's check the cached connection
    try
    {
      $ok = $ssh->ping();
    }
    catch(Exception $e)
    {
      $ok = false;
    }
    //if it's broken for some reason let's just fetch the new one
    if(!$ok)
    {
      $ssh = _deploy_make_ssh($decl, $host);
      $DEPLOY_SSH_CONNS[$conn_id] = $ssh;
    }
  }

  return $DEPLOY_SSH_CONNS[$conn_id];
}

function _deploy_make_ssh($decl, $host)
{
  list($ssh_host, $ssh_port) = deploy_ssh_host_port($host);
  $key = new RSA();
  $key->loadKey($decl->get('ssh_key_str'));

  $ssh = new SSH2($ssh_host, $ssh_port);
  if(!$ssh->login($decl->get('user'), $key))
    throw new Exception("Login failed");
  return $ssh;
}

function deploy_str($decl, $str)
{
  global $DEPLOY_NODES;

  $res = preg_replace_callback(
    '~%\{([^\}]+)\}%~',
    function($m) use($decl)
    {
      return $decl->get($m[1]);
    },
    $str
  );

  return $res;
}