Implement brep-monitor

1 files changed, 766 insertions, 0 deletions

diff --git a/monitor/monitor.cxx b/monitor/monitor.cxx
new file mode 100644
index 0000000..e04c5e1
--- /dev/null
+++ b/monitor/monitor.cxx
@@ -0,0 +1,766 @@
+// 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 build across all
+      // toolchain versions.
+      //
+      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::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);
+}