path: root/bbot/worker/worker.cxx

// file      : bbot/worker.cxx -*- C++ -*-
// copyright : Copyright (c) 2014-2017 Code Synthesis Ltd
// license   : TBC; see accompanying LICENSE file

#ifndef _WIN32
#  include <signal.h> // signal()
#else
#  include <stdlib.h> // getenv(), _putenv()
#endif

#include <regex>
#include <iostream>

#include <libbutl/pager.hxx>
#include <libbutl/filesystem.hxx>

#include <libbbot/manifest.hxx>

#include <bbot/types.hxx>
#include <bbot/utility.hxx>

#include <bbot/diagnostics.hxx>
#include <bbot/bootstrap-manifest.hxx>

#include <bbot/worker/worker-options.hxx>

using namespace std;
using namespace butl;
using namespace bbot;

namespace bbot
{
  process_path argv0;
  worker_options ops;

  dir_path env_dir;

  const size_t tftp_timeout (10); // 10 seconds.
  const size_t tftp_retries (3);  // Task request retries (see startup()).
}

static dir_path
change_wd (const dir_path& d, bool create = false)
try
{
  if (create)
    try_mkdir_p (d);

  dir_path r (dir_path::current_directory ());

  dir_path::current_directory (d);

  return r;
}
catch (const system_error& e)
{
  fail << "unable to change current directory to " << d << ": " << e << endf;
}

using regexes = vector<regex>;

// Match lines read from the command's stderr against the regular expressions
// and return the warning result status (instead of success) in case of a
// match.
//
template <typename... A>
static result_status
run_bpkg (tracer& t,
          string& log, const regexes& warn_detect,
          const string& cmd, A&&... a)
{
  try
  {
    // Trace and log the command line.
    //
    auto cmdc = [&t, &log] (const char* c[], size_t n)
    {
      t (c, n);

      ostringstream os;
      process::print (os, c, n);
      log += os.str ();
      log += '\n';
    };

    fdpipe pipe (fdopen_pipe ()); // Text mode seems appropriate.

    process pr (
      process_start (cmdc,
                     fdnull (), // Never reads from stdout.
                     2,         // 1>&2
                     pipe.out,
                     "bpkg",
                     "-v",
                     cmd,
                     forward<A> (a)...));

    pipe.out.close ();

    result_status r (result_status::success);

    // Log the diagnostics.
    //
    {
      ifdstream is (move (pipe.in), fdstream_mode::skip); // Skip on exception.

      for (string l; is.peek () != ifdstream::traits_type::eof (); )
      {
        getline (is, l);
        log += l;
        log += '\n';

        // Match the log line with the warning-detecting regular expressions
        // until the first match.
        //
        if (r != result_status::warning)
        {
          for (const auto& re: warn_detect)
          {
            // Only examine the first 512 bytes. Long lines (e.g., linker
            // command lines) could trigger implementation-specific limitations
            // (like stack overflow). Plus, it is a performance concern.
            //
            if (regex_search (l.begin (),
                              l.size () < 512 ? l.end () : l.begin () + 512,
                              re))
            {
              r = result_status::warning;
              break;
            }
          }
        }
      }
    }

    if (pr.wait ())
      return r;

    log += "bpkg " + cmd;
    const process_exit& e (*pr.exit);

    if (e.normal ())
    {
      log += " exited with code " + to_string (e.code ()) + "\n";
      return result_status::error;
    }
    else
    {
      log += " terminated abnormally: " + e.description () +
        (e.core () ? " (core dumped)" : "") + "\n";
      return result_status::abnormal;
    }
  }
  catch (const process_error& e)
  {
    fail << "unable to execute bpkg " << cmd << ": " << e << endf;
  }
  catch (const io_error& e)
  {
    fail << "unable to read bpkg " << cmd << " diagnostics: " << e << endf;
  }
}

static void
build (size_t argc, const char* argv[])
{
  tracer trace ("build");

  // Our overall plan is as follows:
  //
  // 1. Parse the task manifest (it should be in CWD).
  //
  // 2. Run bpkg to create the configuration, add the repository, and
  //    configure, build, and test the package all while saving the logs in
  //    the result manifest.
  //
  // 3. Upload the result manifest.
  //
  // Note also that we are being "watched" by the startup version of us which
  // will upload an appropriate result in case we exit with an error. So here
  // for abnormal situations (like a failure to parse the manifest), we just
  // fail.
  //
  task_manifest tm (parse_manifest<task_manifest> (path ("manifest"), "task"));

  result_manifest rm {
    tm.name,
    tm.version,
    result_status::success,
    operation_results {}
  };

  auto add_result = [&rm] (string o) -> operation_result&
  {
    rm.results.push_back (
      operation_result {move (o), result_status::success, ""});

    return rm.results.back ();
  };

  dir_path owd;

  for (;;) // The "breakout" loop.
  {
    // Regular expressions that detect different forms of build2 toolchain
    // warnings. Accidently (or not), they also cover GCC and Clang warnings
    // (for the English locale).
    //
    // The expressions will be matched multiple times, so let's make the
    // matching faster, with the potential cost of making regular expressions
    // creation slower.
    //
    regex::flag_type f (regex_constants::optimize); // ECMAScript is implied.

    regexes wre ({
      regex ("^warning: ", f),
      regex ("^.+: warning: ", f)});

    for (const auto& re: tm.unquoted_warning_regex ())
      wre.emplace_back (re, f);

    // Configure.
    //
    {
      operation_result& r (add_result ("configure"));

      // bpkg create <config-vars> <env-module> <env-config-vars>
      //
      const vector_view<const char*> env (argv + 1, argc - 1);

      // Configuration directory name.
      //
      dir_path dir ("build");

      r.status |= run_bpkg (trace, r.log, wre,
                            "create",
                            "-d", dir.string (),
                            "--wipe",
                            tm.unquoted_config (),
                            env);
      if (!r.status)
        break;

      owd = change_wd (dir);

      // bpkg add <repository-url>
      //
      r.status |= run_bpkg (trace, r.log, wre, "add", tm.repository.string ());

      if (!r.status)
        break;

      // bpkg fetch
      //
      string t ("--trust-no");

      cstrings ts;
      for (const string& fp: tm.trust)
      {
        if (fp == "yes")
          t = "--trust-yes";
        else
        {
          ts.push_back ("--trust");
          ts.push_back (fp.c_str ());
        }
      }

      r.status |= run_bpkg (trace, r.log, wre, "fetch", ts, t);

      if (!r.status)
        break;

      // bpkg build --configure-only <package-name>/<package-version>
      //
      r.status |= run_bpkg (trace, r.log, wre,
                            "build",
                            "--configure-only",
                            "--yes",
                            tm.name + '/' + tm.version.string ());

      if (!r.status)
        break;

      rm.status |= r.status;
    }

    // Update.
    //
    {
      operation_result& r (add_result ("update"));

      // bpkg update <package-name>
      //
      r.status |= run_bpkg (trace, r.log, wre, "update", tm.name);

      if (!r.status)
        break;

      rm.status |= r.status;
    }

    // Test.
    //
    {
      operation_result& r (add_result ("test"));

      // bpkg test <package-name>
      //
      r.status |= run_bpkg (trace, r.log, wre, "test", tm.name);

      if (!r.status)
        break;

      rm.status |= r.status;
    }

    break;
  }

  rm.status |= rm.results.back ().status; // Merge last in case of a break.

  if (!owd.empty ())
    change_wd (owd);

  // Upload the result.
  //
  const string url ("tftp://" + ops.tftp_host () + "/manifest");

  try
  {
    tftp_curl c (trace,
                 path ("-"),
                 nullfd,
                 curl::put,
                 url,
                 "--max-time", tftp_timeout);

    serialize_manifest (rm, c.out, url, "result");
    c.out.close ();

    if (!c.wait ())
      throw_generic_error (EIO);
  }
  catch (const system_error& e)
  {
    fail << "unable to upload result manifest to " << url << ": " << e;
  }
}

static void
startup ()
{
  tracer trace ("startup");

  // Our overall plan is as follows:
  //
  // 1. Download the task manifest into the build directory (CWD).
  //
  // 2. Parse it and get the target.
  //
  // 3. Find the environment setup executable for this target.
  //
  // 4. Execute the environment setup executable.
  //
  // 5. If the environment setup executable fails, then upload the (failed)
  //    result ourselves.
  //
  const string url ("tftp://" + ops.tftp_host () + "/manifest");
  const path mf ("manifest");

  // If we fail, try to upload the result manifest (abnormal termination). The
  // idea is that the machine gets suspended and we can investigate what's
  // going on by logging in and examining the diagnostics (e.g., via
  // journalctl, etc).
  //
  task_manifest tm;

  try
  {
    // Download the task.
    //
    // We are downloading from our host so there shouldn't normally be any
    // connectivity issues. Unless, of course, we are on Windows where all
    // kinds of flakiness is business as usual. Note that having a long enough
    // timeout is not enough: if we try to connect before the network is up,
    // we will keep waiting forever, even after it is up. So we have to
    // timeout and try again. This is also pretty bad (unlike, say during
    // bootstrap which doesn't happen very often) since we are wasting the
    // machine time. So we are going to log it as a warning and not merely a
    // trace since if this is a common occurrence, then something has to be
    // done about it.
    //
    for (size_t retry (1);; ++retry)
    {
      try
      {
        tftp_curl c (trace,
                     nullfd,
                     mf,
                     curl::get,
                     url,
                     "--max-time", tftp_timeout);

        if (!c.wait ())
          throw_generic_error (EIO);

        break;
      }
      catch (const system_error& e)
      {
        bool bail (retry > tftp_retries);
        diag_record dr (bail ? error : warn);

        dr << "unable to download task manifest from " << url << " on "
           << retry << " try: " << e;

        if (bail)
          throw failed ();
      }
    }

    // Parse it.
    //
    tm = parse_manifest<task_manifest> (mf, "task");

    // Find the environment setup executable.
    //
    // While the executable path contains a directory (so the PATH search does
    // not apply) we still use process::path_search() to automatically handle
    // appending platform-specific executable extensions (.exe/.bat, etc).
    //
    string tg (tm.target.string ());
    process_path pp (process::try_path_search (env_dir / tg, false));

    if (pp.empty ())
      pp = process::try_path_search (env_dir / "default", false);

    if (pp.empty ())
      fail << "no environment setup executable in " << env_dir << " "
           << "for target '" << tg << "'";

    // Run it.
    //
    strings os;

    if (ops.systemd_daemon ())
      os.push_back ("--systemd-daemon");

    if (ops.verbose_specified ())
    {
      os.push_back ("--verbose");
      os.push_back (to_string (ops.verbose ()));
    }

    if (ops.tftp_host_specified ())
    {
      os.push_back ("--tftp-host");
      os.push_back (ops.tftp_host ());
    }

    // Note that we use the effective (absolute) path instead of recall since
    // we may have changed the CWD.
    //
    run (trace, pp, tg, argv0.effect_string (), os);
  }
  catch (const failed&)
  {
    // If we failed before being able to parse the task manifest, use the
    // "unknown" values for the package name and version.
    //
    result_manifest rm {
      tm.name.empty ()    ? "unknown"           : tm.name,
      tm.version.empty () ? bpkg::version ("0") : tm.version,
      result_status::abnormal,
      operation_results {}
    };

    try
    {
      tftp_curl c (trace,
                   path ("-"),
                   nullfd,
                   curl::put,
                   url,
                   "--max-time", tftp_timeout);

      serialize_manifest (rm, c.out, url, "result");
      c.out.close ();

      if (!c.wait ())
        throw_generic_error (EIO);
    }
    catch (const system_error& e)
    {
      error << "unable to upload result manifest to " << url << ": " << e;
    }

    throw;
  }
}

static void
bootstrap ()
{
  bootstrap_manifest bm {
    bootstrap_manifest::versions_type {
      {"bbot",    standard_version (BBOT_VERSION_STR)},
      {"libbbot", standard_version (LIBBBOT_VERSION_STR)},
      {"libbpkg", standard_version (LIBBPKG_VERSION_STR)},
      {"libbutl", standard_version (LIBBUTL_VERSION_STR)}
    }
  };

  serialize_manifest (bm, cout, "stdout", "bootstrap");
}

int
main (int argc, char* argv[])
try
{
  // This is a little hack to make out baseutils for Windows work when called
  // with absolute path. In a nutshell, MSYS2's exec*p() doesn't search in the
  // parent's executable directory, only in PATH. And since we are running
  // without a shell (that would read /etc/profile which sets PATH to some
  // sensible values), we are only getting Win32 PATH values. And MSYS2 /bin
  // is not one of them. So what we are going to do is add /bin at the end of
  // PATH (which will be passed as is by the MSYS2 machinery). This will make
  // MSYS2 search in /bin (where our baseutils live). And for everyone else
  // this should be harmless since it is not a valid Win32 path.
  //
#ifdef _WIN32
  {
    string mp ("PATH=");
    if (const char* p = getenv ("PATH"))
    {
      mp += p;
      mp += ';';
    }
    mp += "/bin";

    _putenv (mp.c_str ());
  }
#endif

  // On POSIX ignore SIGPIPE which is signaled to a pipe-writing process if
  // the pipe reading end is closed. Note that by default this signal
  // terminates a process. Also note that there is no way to disable this
  // behavior on a file descriptor basis or for the write() function call.
  //
#ifndef _WIN32
  if (signal (SIGPIPE, SIG_IGN) == SIG_ERR)
    fail << "unable to ignore broken pipe (SIGPIPE) signal: "
         << system_error (errno, generic_category ()); // Sanitize.
#endif

  cli::argv_scanner scan (argc, argv, true);
  ops.parse (scan);

  verb = ops.verbose ();

  if (ops.systemd_daemon ())
    systemd_diagnostics (false);

  // Version.
  //
  if (ops.version ())
  {
    cout << "bbot-worker " << BBOT_VERSION_ID << endl
         << "libbbot " << LIBBBOT_VERSION_ID << endl
         << "libbpkg " << LIBBPKG_VERSION_ID << endl
         << "libbutl " << LIBBUTL_VERSION_ID << endl
         << "Copyright (c) 2014-2017 Code Synthesis Ltd" << endl
         << "TBC; All rights reserved" << endl;

    return 0;
  }

  // Help.
  //
  if (ops.help ())
  {
    pager p ("bbot-worker help", false);
    print_bbot_worker_usage (p.stream ());

    // If the pager failed, assume it has issued some diagnostics.
    //
    return p.wait () ? 0 : 1;
  }

  // Figure out our mode.
  //
  if (ops.bootstrap () && ops.startup ())
    fail << "--bootstrap and --startup are mutually exclusive";

  enum class mode {boot, start, build} m (mode::build);

  if (ops.bootstrap ()) m = mode::boot;
  if (ops.startup ())   m = mode::start;

  if (ops.systemd_daemon ())
  {
    info << "bbot worker " << BBOT_VERSION_ID;
  }

  // Figure out our path (used for re-exec).
  //
  argv0 = process::path_search (argv[0], true);

  // Sort out the build directory.
  //
  if (ops.build_specified ())
    change_wd (ops.build (), true); // Create if does not exist.

  // Sort out the environment directory.
  //
  try
  {
    env_dir = ops.environments_specified ()
      ? ops.environments ()
      : dir_path::home_directory ();

    if (!dir_exists (env_dir))
      throw_generic_error (ENOENT);
  }
  catch (const system_error& e)
  {
    fail << "invalid environment directory: " << e;
  }

  switch (m)
  {
  case mode::boot:  bootstrap ();                            break;
  case mode::start: startup ();                              break;
  case mode::build: build (static_cast<size_t> (argc),
                           const_cast<const char**> (argv)); break;
  }
}
catch (const failed&)
{
  return 1; // Diagnostics has already been issued.
}
catch (const cli::exception& e)
{
  error << e;
  return 1;
}