path: root/monitor/monitor.cxx

// file      : monitor/monitor.cxx -*- C++ -*-
// license   : MIT; see accompanying LICENSE file

#include <map>
#include <set>
#include <chrono>
#include <iostream>
#include <algorithm> // find_if()

#include <odb/database.hxx>
#include <odb/transaction.hxx>
#include <odb/schema-catalog.hxx>

#include <odb/pgsql/database.hxx>

#include <libbutl/pager.mxx>
#include <libbutl/utility.mxx> // compare_c_string

#include <libbbot/build-config.hxx>

#include <libbrep/build.hxx>
#include <libbrep/common.hxx>
#include <libbrep/build-odb.hxx>
#include <libbrep/build-package.hxx>
#include <libbrep/build-package-odb.hxx>
#include <libbrep/database-lock.hxx>

#include <mod/build-config.hxx>

#include <monitor/module-options.hxx>
#include <monitor/monitor-options.hxx>

using namespace std;
using namespace butl;
using namespace bbot;
using namespace odb::core;

namespace brep
{
  // Operation failed, diagnostics has already been issued.
  //
  struct failed {};

  static const char* help_info (
    "  info: run 'brep-monitor --help' for more information");

  static int
  main (int argc, char* argv[])
  try
  {
    cli::argv_scanner scan (argc, argv);
    options::monitor ops (scan);

    // Version.
    //
    if (ops.version ())
    {
      cout << "brep-monitor " << BREP_VERSION_ID << endl
           << "libbrep " << LIBBREP_VERSION_ID << endl
           << "libbbot " << LIBBBOT_VERSION_ID << endl
           << "libbpkg " << LIBBPKG_VERSION_ID << endl
           << "libbutl " << LIBBUTL_VERSION_ID << endl
           << "Copyright (c) " << BREP_COPYRIGHT << "." << endl
           << "This is free software released under the MIT license." << endl;

      return 0;
    }

    // Help.
    //
    if (ops.help ())
    {
      pager p ("brep-monitor help",
               false,
               ops.pager_specified () ? &ops.pager () : nullptr,
               &ops.pager_option ());

      print_usage (p.stream ());

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

    // Parse the brep module configuration.
    //
    options::module mod_ops;
    {
      if (!scan.more ())
      {
        cerr << "error: brep module configuration file is expected" << endl
             << help_info << endl;
        return 1;
      }

      string f (scan.next ());

      try
      {
        cli::argv_file_scanner scan (f, "" /* option */);

        // Parse the brep module options skipping those we don't recognize.
        //
        while (scan.more ())
        {
          // Parse until an unknown option is encountered.
          //
          mod_ops.parse (scan,
                         cli::unknown_mode::stop,
                         cli::unknown_mode::stop);

          // Skip the unknown option, unless we are done.
          //
          if (scan.more ())
          {
            // Skip the option name.
            //
            size_t l (scan.peek_line ());
            scan.skip ();

            // Skip the option value, if present.
            //
            // Note that here we rely on the configuration file having both
            // the option name and its value on the same line.
            //
            if (scan.more () && scan.peek_line () == l)
              scan.skip ();
          }
        }
      }
      catch (const cli::file_io_failure& e)
      {
        cerr << "error: unable to parse brep module configuration: " << e
             << endl;
        return 1;
      }
      catch (const cli::exception& e)
      {
        cerr << "error: unable to parse brep module configuration file '" << f
             << "': " << e << endl;
        return 1;
      }
    }

    if (!mod_ops.build_config_specified ())
    {
      cerr << "warning: package building functionality is disabled" << endl;
      return 0;
    }

    // Parse the toolchains suppressing duplicates.
    //
    // Note that specifying a toolchain both with and without version doesn't
    // make sense, so we fail if that's the case.
    //
    vector<pair<string, version>> toolchains;

    if (!scan.more ())
    {
      cerr << "error: toolchain is expected" << endl << help_info << endl;
      return 1;
    }

    while (scan.more ())
    {
      string s (scan.next ());

      string tn;
      version tv;

      try
      {
        size_t p (s.find ('/'));

        if (p == string::npos)
          tn = move (s);
        else
        {
          tn.assign (s, 0, p);
          tv = version (string (s, p + 1));
        }

        bool dup (false);
        for (const pair<string, version>& t: toolchains)
        {
          if (tn == t.first)
          {
            if (tv == t.second)
            {
              dup = true;
              break;
            }

            if (tv.empty () != t.second.empty ())
            {
              cerr << "error: toolchain '" << tn << "' is specified both "
                   << "with and without version" << endl;
              return 1;
            }
          }
        }

        if (!dup)
          toolchains.emplace_back (move (tn), move (tv));
      }
      catch (const invalid_argument& e)
      {
        cerr << "error: invalid toolchain '" << s << "': " << e << endl;
        return 1;
      }
    }

    // Parse buildtab.
    //
    build_configs configs;

    try
    {
      configs = parse_buildtab (mod_ops.build_config ());
    }
    catch (const tab_parsing& e)
    {
      cerr << "error: unable to parse buildtab: " << e << endl;
      return 1;
    }
    catch (const io_error& e)
    {
      cerr << "error: unable to read '" << mod_ops.build_config () << "': "
           << e << endl;
      return 1;
    }

    // Create the database instance.
    //
    odb::pgsql::database db (
      ops.build_db_user (),
      ops.build_db_password (),
      (ops.build_db_name_specified ()
       ? ops.build_db_name ()
       : mod_ops.build_db_name ()),
      (ops.build_db_host_specified ()
       ? ops.build_db_host ()
       : mod_ops.build_db_host ()),
      (ops.build_db_port_specified ()
       ? ops.build_db_port ()
       : mod_ops.build_db_port ()),
      "options='-c default_transaction_isolation=serializable'");

    // Prevent several brep utility instances from updating the build database
    // simultaneously.
    //
    database_lock l (db);

    // Check that the database schema matches the current one.
    //
    const string ds ("build");
    if (schema_catalog::current_version (db, ds) != db.schema_version (ds))
    {
      cerr << "error: build database schema differs from the current one"
           << endl
           << "  info: use brep-migrate to migrate the database" << endl;
      return 1;
    }

    // If requested, cleanup delays for package builds that are not expected
    // anymore (build configuration is not present, etc).
    //
    if (ops.clean ())
    {
      using config_map = map<const char*,
                             const build_config*,
                             compare_c_string>;

      config_map conf_map;
      for (const build_config& c: configs)
        conf_map[c.name.c_str ()] = &c;

      // Prepare the build delay prepared query.
      //
      // Query package build delays in chunks in order not to hold locks for
      // too long. Sort the result by package version as a first priority to
      // minimize number of queries to the package database. Note that we
      // still need to sort by configuration and toolchain to make sure that
      // build delays are sorted consistently across queries and we don't miss
      // any of them.
      //
      using query = query<build_delay>;
      using prep_query = prepared_query<build_delay>;

      // Specify the portion.
      //
      size_t offset (0);

      query q ("ORDER BY" +
               query::id.package.tenant + "," +
               query::id.package.name +
               order_by_version (query::id.package.version,
                                 false /* first */) + "," +
               query::id.configuration + "," +
               query::id.toolchain_name +
               order_by_version (query::id.toolchain_version,
                                 false /* first */) +
               "OFFSET" + query::_ref (offset) + "LIMIT 100");

      connection_ptr conn (db.connection ());

      prep_query pq (
        conn->prepare_query<build_delay> ("build-delay-query", q));

      // Cache the delayed build package object to reuse it in case the next
      // delay refers to the same package (which is often the case due to the
      // query result sorting criteria we use).
      //
      package_id pid;
      shared_ptr<build_package> p;

      for (bool ne (true); ne; )
      {
        transaction t (conn->begin ());

        // Query delays.
        //
        auto delays (pq.execute ());

        if ((ne = !delays.empty ()))
        {
          // Iterate over the build delays and cleanup the outdated ones.
          //
          for (const build_delay& d: delays)
          {
            config_map::const_iterator ci;

            bool cleanup (
              // Check that the toolchain is still used.
              //
              find_if (toolchains.begin (), toolchains.end (),
                       [&d] (const pair<string, version>& t)
                       {
                         return t.first == d.toolchain_name &&
                                t.second == d.toolchain_version;
                       }) == toolchains.end () ||
              //
              // Check that the build configuration is still present.
              //
              (ci = conf_map.find (d.configuration.c_str ())) ==
              conf_map.end ());

            // Check that the package still present, is buildable and doesn't
            // exclude the build configuration.
            //
            if (!cleanup)
            {
              if (d.id.package != pid)
              {
                pid = d.id.package;
                p = db.find<build_package> (pid);
              }

              cleanup = (p == nullptr  ||
                         !p->buildable ||
                         exclude (p->builds,
                                  p->constraints,
                                  *ci->second,
                                  configs.class_inheritance_map));
            }

            if (cleanup)
              db.erase (d);
            else
              ++offset;
          }
        }

        t.commit ();
      }
    }

    // Collect and report delays as separate steps not to hold database locks
    // while printing to stderr. Also we need to properly order delays for
    // printing.
    //
    // Iterate through all possible package builds creating the list of delays
    // with the following sort priority:
    //
    // 1: toolchain name
    // 2: toolchain version (descending)
    // 3: configuration name
    // 4: tenant
    // 5: package name
    // 6: package version (descending)
    //
    // Such ordering will allow us to group build delays by toolchain and
    // configuration while printing the report.
    //
    struct compare_delay
    {
      bool
      operator() (const shared_ptr<const build_delay>& x,
                  const shared_ptr<const build_delay>& y) const
      {
        if (int r = x->toolchain_name.compare (y->toolchain_name))
          return r < 0;

        if (int r = x->toolchain_version.compare (y->toolchain_version))
          return r > 0;

        if (int r = x->configuration.compare (y->configuration))
          return r < 0;

        if (int r = x->tenant.compare (y->tenant))
          return r < 0;

        if (int r = x->package_name.compare (y->package_name))
          return r < 0;

        return x->package_version.compare (y->package_version) > 0;
      }
    };

    set<shared_ptr<const build_delay>, compare_delay> delays;
    {
      connection_ptr conn (db.connection ());

      // Prepare the buildable package prepared query.
      //
      // Query buildable packages in chunks in order not to hold locks for
      // too long.
      //
      using pquery = query<buildable_package>;
      using prep_pquery = prepared_query<buildable_package>;

      // Specify the portion.
      //
      size_t offset (0);

      pquery pq ("ORDER BY"                            +
                 pquery::build_package::id.tenant + "," +
                 pquery::build_package::id.name         +
                 order_by_version (pquery::build_package::id.version,
                                   false /* first */)  +
                 "OFFSET" + pquery::_ref (offset) + "LIMIT 50");

      prep_pquery ppq (
        conn->prepare_query<buildable_package> ("buildable-package-query",
                                                pq));

      // Prepare the package build prepared query.
      //
      // This query will only be used for toolchains that have no version
      // specified on the command line to obtain the latest completed build
      // across all toolchain versions, if present, and the latest incomplete
      // build otherwise.
      //
      using bquery = query<package_build>;
      using prep_bquery = prepared_query<package_build>;

      build_id id;
      const auto& bid (bquery::build::id);

      bquery bq ((equal<package_build> (bid.package, id.package)       &&
                  bid.configuration == bquery::_ref (id.configuration) &&
                  bid.toolchain_name == bquery::_ref (id.toolchain_name)) +
                 "ORDER BY"                                               +
                 bquery::build::completion_timestamp + "DESC, "           +
                 bquery::build::timestamp + "DESC"                        +
                 "LIMIT 1");

      prep_bquery pbq (
        conn->prepare_query<package_build> ("package-build-query", bq));

      timestamp::duration build_timeout (
        ops.build_timeout_specified ()
        ? chrono::seconds (ops.build_timeout ())
        : chrono::seconds (mod_ops.build_normal_rebuild_timeout () +
                           mod_ops.build_result_timeout ()));

      timestamp now (system_clock::now ());

      timestamp build_expiration (now - build_timeout);

      timestamp report_expiration (
        now - chrono::seconds (ops.report_timeout ()));

      for (bool ne (true); ne; )
      {
        transaction t (conn->begin ());

        // Query buildable packages (and cache the result).
        //
        auto bps (ppq.execute ());

        if ((ne = !bps.empty ()))
        {
          offset += bps.size ();

          for (auto& bp: bps)
          {
            shared_ptr<build_package> p (db.load<build_package> (bp.id));

            for (const build_config& c: configs)
            {
              if (exclude (p->builds,
                           p->constraints,
                           c,
                           configs.class_inheritance_map))
                continue;

              for (const pair<string, version>& t: toolchains)
              {
                id = build_id (p->id, c.name, t.first, t.second);

                // If the toolchain version is not specified then search for
                // the latest build across all toolchain versions and search
                // for a specific build otherwise.
                //
                shared_ptr<build> b;

                if (id.toolchain_version.empty ())
                {
                  auto pbs (pbq.execute ());

                  if (!pbs.empty ())
                    b = move (pbs.begin ()->build);
                }
                else
                  b = db.find<build> (id);

                // Note that we consider a build as delayed if it is not
                // completed in the expected timeframe. So even if the build
                // task have been issued recently we may still consider the
                // build as delayed.
                //
                timestamp bct (b != nullptr
                               ? b->completion_timestamp
                               : timestamp_nonexistent);

                // Create the delay object to record a timestamp when the
                // package build could have potentially been started, unless
                // it already exists.
                //
                shared_ptr<build_delay> d (db.find<build_delay> (id));

                if (d == nullptr)
                {
                  // If the archived package has no build nor build delay
                  // for this configuration, then we assume that the
                  // configuration was added after the package tenant has
                  // been archived and so the package could have never been
                  // built for this configuration. Thus, we don't consider
                  // this build as delayed and so skip it.
                  //
                  if (bp.archived && b == nullptr)
                    continue;

                  // Use the build completion or build status change
                  // timestamp, whichever is earlier, as the build delay
                  // tracking starting point and fallback to the current time
                  // if there is no build yet.
                  //
                  timestamp pts (
                    b == nullptr                                       ? now :
                    bct != timestamp_nonexistent && bct < b->timestamp ? bct :
                    b->timestamp);

                  d = make_shared<build_delay> (move (id.package.tenant),
                                                move (id.package.name),
                                                p->version,
                                                move (id.configuration),
                                                move (id.toolchain_name),
                                                t.second,
                                                pts);
                  db.persist (d);
                }

                // Handle package builds differently based on their tenant's
                // archive status.
                //
                // If the package is not archived then consider it as delayed
                // if it is not (re-)built by the expiration time. Otherwise,
                // consider it as delayed if it is unbuilt.
                //
                bool delayed;

                if (!bp.archived)
                {
                  timestamp bts (bct != timestamp_nonexistent
                                 ? bct
                                 : d->package_timestamp);

                  delayed = (bts <= build_expiration);
                }
                else
                  delayed = (bct == timestamp_nonexistent);

                // If the report timeout is not specified then report the
                // delay unconditionally. Otherwise, report the active package
                // build delay if the report timeout is expired and the
                // archived package build delay if it was never reported. Note
                // that fixing the building infrastructure won't help building
                // an archived package, so reporting its build delays
                // repeatedly is meaningless.
                //
                if (delayed &&
                    (!ops.report_timeout_specified () ||
                     (!bp.archived
                      ? d->report_timestamp <= report_expiration
                      : d->report_timestamp == timestamp_nonexistent)))
                {
                  // Note that we update the delay objects persistent state
                  // later, after we successfully print the report.
                  //
                  d->report_timestamp = now;
                  delays.insert (move (d));
                }
              }
            }
          }
        }

        t.commit ();
      }
    }

    // Report package build delays, if any.
    //
    if (!delays.empty ())
    try
    {
      // Print the report.
      //
      cerr.exceptions (ostream::badbit | ostream::failbit);

      cerr << "Package build delays (" << delays.size () << "):" << endl;

      // Group the printed delays by toolchain and configuration.
      //
      const string*  toolchain_name    (nullptr);
      const version* toolchain_version (nullptr);
      const string*  configuration     (nullptr);

      // Print the delayed package build number per configuration rather than
      // the packages themselves in the brief report mode.
      //
      size_t config_build_count (0);

      auto brief_config = [&configuration, &config_build_count] ()
      {
        if (configuration != nullptr)
        {
          cerr << "    " << *configuration << " (" << config_build_count
               << ")" << endl;

          config_build_count = 0;
        }
      };

      for (shared_ptr<const build_delay> d: delays)
      {
        // Print the toolchain, if changed.
        //
        if (toolchain_name == nullptr            ||
            d->toolchain_name != *toolchain_name ||
            d->toolchain_version != *toolchain_version)
        {
          if (!ops.full_report ())
            brief_config ();

          if (toolchain_name != nullptr)
            cerr << endl;

          cerr << "  " << d->toolchain_name;

          if (!d->toolchain_version.empty ())
            cerr << "/" << d->toolchain_version;

          cerr << endl;

          toolchain_name    = &d->toolchain_name;
          toolchain_version = &d->toolchain_version;
          configuration     = nullptr;
        }

        // Print the configuration, if changed.
        //
        if (configuration == nullptr || d->configuration != *configuration)
        {
          if (ops.full_report ())
          {
            if (configuration != nullptr)
              cerr << endl;

            cerr << "    " << d->configuration << endl;
          }
          else
            brief_config ();

          configuration = &d->configuration;
        }

        // Print the delayed build package in the full report mode and count
        // configuration builds otherwise.
        //
        if (ops.full_report ())
        {
          // We can potentially extend this information with the archived flag
          // or the delay duration.
          //
          cerr << "      " << d->package_name << "/" << d->package_version;

          if (!d->tenant.empty ())
            cerr << " " << d->tenant;

          cerr << endl;
        }
        else
          ++config_build_count;
      }

      if (!ops.full_report ())
        brief_config ();

      // Persist the delay report timestamps.
      //
      transaction t (db.begin ());

      for (shared_ptr<const build_delay> d: delays)
        db.update (d);

      t.commit ();
    }
    catch (const io_error&)
    {
      return 1; // Not much we can do on stderr writing failure.
    }

    return 0;
  }
  catch (const database_locked&)
  {
    cerr << "brep-monitor or some other brep utility is running" << endl;
    return 2;
  }
  catch (const recoverable& e)
  {
    cerr << "recoverable database error: " << e << endl;
    return 3;
  }
  catch (const cli::exception& e)
  {
    cerr << "error: " << e << endl << help_info << endl;
    return 1;
  }
  catch (const failed&)
  {
    return 1; // Diagnostics has already been issued.
  }
  // Fully qualified to avoid ambiguity with odb exception.
  //
  catch (const std::exception& e)
  {
    cerr << "error: " << e << endl;
    return 1;
  }
}

int
main (int argc, char* argv[])
{
  return brep::main (argc, argv);
}