workshop: implement stickiness

Like commit 777d5b4 but for classic
Workshop jobs.

Author: MaxKellermann

debian/changelog

@@ -1,6 +1,6 @@
 cm4all-workshop (7.7) unstable; urgency=low
 
-  * 
+  * workshop: implement stickiness
 
  --   
 

doc/index.rst

@@ -78,6 +78,25 @@ The following settings are recognized:
   * ``journal``: set to :samp:`yes` to send structured log
     messages to the systemd journal
 
+  * ``sticky``: if ``yes``, then jobs with the same ``sticky_id``
+    value are always executed on the same server.  This requires that
+    all Workshop processes on all servers know each others via
+    Zeroconf (i.e. at least ``zeroconf_service`` must be configured).
+    See :ref:`cron.sticky` for more information.
+
+  * ``zeroconf_service``: discover other Workshop instances with this
+    Zeroconf service name (for ``sticky``).
+  * ``zeroconf_domain``: The name of the Zeroconf domain.
+  * ``zeroconf_interface``: publish the Zeroconf service only on the
+    given interface.
+  * ``zeroconf_protocol``: Publish only protocol ``inet`` or
+    ``inet6``.
+  * ``zeroconf_weight``: publish the Zeroconf service with the
+    specified "weight", i.e. accept less or more jobs on this host.
+    The value is a decimal number; the implied default value is
+    :samp:`1.0`.  For example, if you specify :samp:`0.5`, you expect
+    this node to get only half as many jobs as others.
+
 .. _workshop_translation_server:
 
   * ``translation_server``: address the translation server is
@@ -535,6 +554,8 @@ The `jobs` table
 ----------------
 
 * ``id``: The primary key.
+* ``sticky_id``: An opaque string which is used to calculate
+  :ref:`stickiness <cron.sticky>`.
 * ``name``: An optional name assigned by the job creator.  Not
   used by Workshop.
 * ``description``: Human readable description.  Not used by

sql/jobs.sql

@@ -28,6 +28,11 @@ CREATE TABLE jobs (
     -- the management application might not know where/how exactly
     account_id varchar(64) NULL,
 
+    -- if not NULL, then use this string to calculate stickiness for
+    -- this job (i.e. execute all jobs with the same sticky_id on the
+    -- same cluster node)
+    sticky_id varchar(256) NULL,
+
     --------------------------------
     -- Execution parameters
     --------------------------------

src/Config.cxx

@@ -155,6 +155,16 @@ WorkshopConfigParser::Partition::ParseLine(FileLineParser &line)
 	} else if (StringIsEqual(word, "journal")) {
 		config.enable_journal = line.NextBool();
 		line.ExpectEnd();
+#ifdef HAVE_AVAHI
+	} else if (StringIsEqual(word, "sticky")) {
+		config.sticky = line.NextBool();
+		line.ExpectEnd();
+	} else if (config.zeroconf.ParseLine(word, line)) {
+#else
+	} else if (StringIsEqual(word, "sticky") ||
+		   StringStartsWith(word, "zeroconf_"sv)) {
+		throw LineParser::Error{"Zeroconf support is disabled at compile time"};
+#endif // HAVE_AVAHI
 	} else
 		throw LineParser::Error("Unknown option");
 }

src/Config.hxx

@@ -48,8 +48,10 @@ struct Config {
 #ifdef HAVE_AVAHI
 	[[gnu::pure]]
 	bool UsesZeroconf() const noexcept {
-		return std::any_of(cron_partitions.begin(), cron_partitions.end(),
-				   [](const auto &i) { return i.UsesZeroconf(); });
+		return std::any_of(partitions.begin(), partitions.end(),
+				   [](const auto &i) { return i.UsesZeroconf(); }) ||
+			 std::any_of(cron_partitions.begin(), cron_partitions.end(),
+				     [](const auto &i) { return i.UsesZeroconf(); });
 	}
 #endif
 };

src/Instance.cxx

@@ -53,6 +53,11 @@ Instance::Instance(const Config &config,
 
 	for (const auto &i : config.partitions)
 		partitions.emplace_front(*this, *library, *spawn_service,
+#ifdef HAVE_AVAHI
+					 avahi_client.get(),
+					 avahi_publisher.get(),
+					 avahi_error_handler,
+#endif // HAVE_AVAHI
 					 config, i,
 					 BIND_THIS_METHOD(OnPartitionIdle));
 

src/Migrate.cxx

@@ -58,6 +58,9 @@ MigrateWorkshopDatabase(Pg::Connection &c, const char *schema)
 	// since Workshop 7.4
 	c.Execute("ALTER TABLE jobs ADD COLUMN IF NOT EXISTS account_id varchar(32) NULL");
 	c.Execute("CREATE INDEX IF NOT EXISTS jobs_account_modified ON jobs(account_id, plan_name, time_modified)");
+
+	// since Workshop 7.7
+	c.Execute("ALTER TABLE jobs ADD COLUMN IF NOT EXISTS sticky_id varchar(256) NULL");
 }
 
 static void

src/workshop/Config.cxx

@@ -11,4 +11,9 @@ WorkshopPartitionConfig::Check() const
 {
 	if (database.connect.empty())
 		throw std::runtime_error("Missing 'database' setting");
+
+#ifdef HAVE_AVAHI
+	if (sticky && !zeroconf.IsEnabled())
+		throw std::runtime_error{"Must configure Zeroconf if 'sticky' is enabled"};
+#endif
 }

src/workshop/Config.hxx

@@ -6,6 +6,11 @@
 
 #include "pg/Config.hxx"
 #include "net/LocalSocketAddress.hxx"
+#include "config.h"
+
+#ifdef HAVE_AVAHI
+#include "lib/avahi/ServiceConfig.hxx"
+#endif
 
 #include <string>
 
@@ -15,6 +20,10 @@ struct WorkshopPartitionConfig {
 	 */
 	std::string name;
 
+#ifdef HAVE_AVAHI
+	Avahi::ServiceConfig zeroconf;
+#endif
+
 	Pg::Config database;
 
 	LocalSocketAddress translation_socket;
@@ -29,11 +38,22 @@ struct WorkshopPartitionConfig {
 
 	bool enable_journal = false;
 
+#ifdef HAVE_AVAHI
+	bool sticky = false;
+#endif
+
 	explicit WorkshopPartitionConfig(std::string &&_name) noexcept
 		:name(std::move(_name))
 	{
 		translation_socket.Clear();
 	}
 
 	void Check() const;
+
+#ifdef HAVE_AVAHI
+	[[gnu::pure]]
+	bool UsesZeroconf() const noexcept {
+		return sticky;
+	}
+#endif
 };

src/workshop/Job.hxx

@@ -18,6 +18,12 @@ struct WorkshopJob {
 
 	std::string id, plan_name;
 
+	/**
+	 * An opaque string for rendezvous-hashing which determines
+	 * the Workshop node that shall execute this job.
+	 */
+	std::string sticky_id;
+
 	std::forward_list<std::string> args;
 
 	std::forward_list<std::string> env;