clusterer_controller: clusterer's side of the controller, as plain code

The changes PR OpenSIPS#4074 makes to modules/clusterer behind
CLUSTERER_CTRL_SUPPORT, with that switch resolved as defined: the
controller API (clusterer_ctrl.c/h), per-cluster node ids, the managed
sharing tags and BIN messages that do not override the controller.
Log and MI strings are ASCII where the PR had em dashes.

Author: Lt-Flash

modules-contrib/clusterer_controller/clusterer.c

@@ -0,0 +1,79 @@
+/*
+ * clusterer_ctrl.h — Controller API for the clusterer module
+ *
+ * Allows an external module (clusterer_controller) to drive the clusterer
+ * topology at runtime without any DB or static configuration.
+ *
+ * Copyright (C) 2026 Yury Kirsanov
+ *                          VoIPLine Telecom
+ *
+ * This file is part of opensips, a free SIP server.
+ *
+ * opensips is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ */
+
+#ifndef CLUSTERER_CTRL_H
+#define CLUSTERER_CTRL_H
+
+#include "../../str.h"
+
+/* clusterer's side of the controller: what the controller calls in it */
+typedef struct clusterer_ctrl_binds {
+    /* register this node in a cluster (created if new) as its current_node;
+     * call before add_node(), @bin_url is "bin:IP:PORT" */
+    int (*set_my_identity)(int cluster_id, int node_id, str *bin_url);
+
+    /* add a peer at runtime (0, no-op, if present); the ping timer brings up
+     * its BIN link within one ping interval */
+    int (*add_node)(int cluster_id, int node_id, str *bin_url);
+
+    /* remove a peer at runtime and close its BIN connection;
+     * -1 if the cluster or node is unknown */
+    int (*remove_node)(int cluster_id, int node_id);
+
+    /* close a node's BIN connection but keep it in the topology */
+    int (*close_node_conn)(int cluster_id, int node_id);
+    /* replace this node's optimistic node_id with the master-assigned one
+     * (0, no-op, if unchanged) */
+    int (*update_identity)(int cluster_id, int new_node_id, str *bin_url);
+
+    /* call in child_init() of every process: re-reads the process-local
+     * current_id from shm so BIN packets carry the right source node_id */
+    int (*sync_current_id)(void);
+
+	/* activate all BACKUP sharing tags: master only, when a peer departs or
+	 * this node becomes master */
+	int (*activate_backup_shtags)(int cluster_id);
+
+	/* force all local sharing tags to BACKUP regardless of =active,
+	 * at startup when the controller manages tags */
+	int (*force_backup_shtags)(int cluster_id);
+
+	/* block MI clusterer_set_tag_active and $shtag() writes for the cluster so
+	 * they cannot fight controller failover (manage_shtags=1) */
+	int (*set_shtag_managed)(int cluster_id);
+
+	/* re-allow MI/script shtag changes, on adopting a running cluster's
+	 * manage_shtags=0 (on_config_mismatch=adopt) */
+	int (*unset_shtag_managed)(int cluster_id);
+
+	/* cluster_ids with cluster_options use_controller=1: the count, and
+	 * clusterer's own array (valid for the process lifetime) */
+	int  managed_count;
+	int *managed_ids;
+} clusterer_ctrl_binds_t;
+
+/* Set to 1 by load_clusterer_ctrl_binds() when a controller module binds the
+ * API, so clusterer can warn if use_controller=1 yet no controller registered. */
+extern int clusterer_ctrl_bound;
+
+/* called from the controller's mod_init(); -1 if clusterer is not loaded
+ * or lacks use_controller support */
+typedef int (*load_clusterer_ctrl_binds_f)(clusterer_ctrl_binds_t *binds);
+
+int load_clusterer_ctrl_binds(clusterer_ctrl_binds_t *binds);
+
+#endif /* CLUSTERER_CTRL_H */

modules-contrib/clusterer_controller/clusterer_ctrl.c

@@ -39,6 +39,7 @@
 #include "clusterer.h"
 #include "sync.h"
 #include "sharing_tags.h"
+#include "clusterer_ctrl.h"
 #include "clusterer_evi.h"
 
 int ping_interval = DEFAULT_PING_INTERVAL;
@@ -50,6 +51,105 @@ int current_id = -1;
 int db_mode = 1;
 int clusterer_enable_rerouting = 1;
 
+int *_current_id_shm = NULL;
+int use_controller = 0;
+
+int cl_ctr_stub_ids[64];
+int cl_ctr_stub_count = 0;
+
+/* extract "<name>=<int>" from a "key=value, key=value" cluster_options string.
+ * Returns 0 and sets *out on success, 1 if the key is absent, -1 if malformed. */
+static int cl_ctr_opt_int(str *descr, str *name, int *out)
+{
+	char *p, *pe;
+	str aux;
+
+	p = str_strstr(descr, name);
+	if (!p)
+		return 1;
+	p += name->len;
+	p = q_memchr(p, '=', descr->s + descr->len - p);
+	if (!p) {
+		LM_ERR("clusterer: expected '=' after '%.*s' in cluster_options\n",
+			name->len, name->s);
+		return -1;
+	}
+	p++;
+	pe = q_memchr(p, ',', descr->s + descr->len - p);
+	aux.s = p;
+	aux.len = pe ? pe - p : descr->s + descr->len - p;
+	str_trim_spaces_lr(aux);
+	if (aux.len == 0 || str2int(&aux, (unsigned int *)out)) {
+		LM_ERR("clusterer: bad value for '%.*s' in cluster_options\n",
+			name->len, name->s);
+		return -1;
+	}
+	return 0;
+}
+
+/* "cluster_id=N, use_controller=0|1" (default 0 = native); only use_controller=1
+ * registers the cluster, whose identity and peers the controller then drives */
+static int cl_ctr_add_cluster_options(modparam_t type, void *val)
+{
+	static str cid_prop = str_init("cluster_id");
+	static str uc_prop  = str_init("use_controller");
+	str descr;
+	int cid = -1, uc = 0, rc, i;
+
+	if (!val || !*(char *)val) {
+		LM_ERR("clusterer: empty 'cluster_options'\n");
+		return -1;
+	}
+	descr.s = (char *)val;
+	descr.len = strlen(descr.s);
+
+	rc = cl_ctr_opt_int(&descr, &cid_prop, &cid);
+	if (rc < 0)
+		return -1;
+	if (rc == 1 || cid < 1) {
+		LM_ERR("clusterer: 'cluster_options' requires cluster_id >= 1, "
+			"e.g. \"cluster_id=1, use_controller=1\"\n");
+		return -1;
+	}
+
+	rc = cl_ctr_opt_int(&descr, &uc_prop, &uc);   /* absent (rc==1) -> uc stays 0 */
+	if (rc < 0)
+		return -1;
+	if (uc != 0 && uc != 1) {
+		LM_ERR("clusterer: use_controller in 'cluster_options' must be 0 or 1 "
+			"(cluster_id %d)\n", cid);
+		return -1;
+	}
+
+	if (uc == 0)
+		return 0;   /* native (the default): nothing to register */
+
+	for (i = 0; i < cl_ctr_stub_count; i++)
+		if (cl_ctr_stub_ids[i] == cid) {
+			LM_ERR("clusterer: cluster_options declares cluster_id %d as "
+				"controller-managed more than once\n", cid);
+			return -1;
+		}
+	if (cl_ctr_stub_count >= (int)(sizeof cl_ctr_stub_ids / sizeof cl_ctr_stub_ids[0])) {
+		LM_ERR("clusterer: too many controller-managed clusters\n");
+		return -1;
+	}
+	cl_ctr_stub_ids[cl_ctr_stub_count++] = cid;
+	use_controller = 1;
+	return 0;
+}
+
+/* retired interim modparams (never released): kept only to fail with a
+ * migration hint instead of "unknown parameter" */
+static int cl_ctr_retired_param(modparam_t type, void *val)
+{
+	LM_ERR("clusterer: the 'use_controller'/'cluster_id' modparams have been "
+		"replaced; declare each controller-managed cluster with "
+		"modparam(\"clusterer\", \"cluster_options\", "
+		"\"cluster_id=N, use_controller=1\")\n");
+	return -1;
+}
+
 str clusterer_db_url = {NULL, 0};
 
 extern db_con_t *db_hdl;
@@ -106,6 +206,7 @@ int cmd_check_addr(struct sip_msg *msg, int *cluster_id, str *ip_str,
  */
 
 static const cmd_export_t cmds[] = {
+	{"load_clusterer_ctrl_binds", (cmd_function)load_clusterer_ctrl_binds, {{0,0,0}}, 0},
 	{"load_clusterer",  (cmd_function)load_clusterer, {{0,0,0}}, 0},
 	{"cluster_broadcast_req", (cmd_function)cmd_broadcast_req, {
 		{CMD_PARAM_INT,0,0},
@@ -158,6 +259,9 @@ static const param_export_t params[] = {
 	{"flags_col",			STR_PARAM,	&flags_col.s		},
 	{"description_col",		STR_PARAM,	&description_col.s	},
 	{"db_mode",				INT_PARAM,	&db_mode			},
+	{"cluster_options",			STR_PARAM|USE_FUNC_PARAM, (void*)cl_ctr_add_cluster_options},
+	{"use_controller",			INT_PARAM|USE_FUNC_PARAM, (void*)cl_ctr_retired_param},
+	{"cluster_id",				INT_PARAM|USE_FUNC_PARAM, (void*)cl_ctr_retired_param},
 	{"neighbor_node_info",	STR_PARAM|USE_FUNC_PARAM,
 		(void*)&provision_neighbor},
 	{"my_node_info",		STR_PARAM|USE_FUNC_PARAM,
@@ -389,11 +493,18 @@ static int mod_init(void)
 	flags_col.len = strlen(flags_col.s);
 	description_col.len = strlen(description_col.s);
 
+	/* no DB URL means no DB-native clusters: db_mode 0, so controller-only and
+	 * static (my_node_info) setups need no explicit db_mode=0 */
+	if (!clusterer_db_url.s && !db_default_url)
+		db_mode = 0;
+
 	/* only allow the DB URL to be skipped in "P2P discovery" mode */
 	init_db_url(clusterer_db_url, db_mode == 0);
 
-	if (current_id < 1) {
-		LM_CRIT("Invalid 'my_node_id' parameter\n");
+	/* controller clusters get their id at runtime: my_node_id is required only
+	 * for native (DB or static) clusters */
+	if (current_id < 1 && (db_mode != 0 || (cluster_list && *cluster_list != NULL))) {
+		LM_CRIT("Invalid 'my_node_id' parameter (required for DB or static native clusters)\n");
 		return -1;
 	}
 	if (ping_interval <= 0) {
@@ -421,6 +532,17 @@ static int mod_init(void)
 		LM_CRIT("Failed to init lock\n");
 		return -1;
 	}
+	{
+		int *_shm_id = shm_malloc(sizeof(int));
+		if (!_shm_id) {
+			LM_CRIT("No shm memory for GET_CURRENT_ID\n");
+			return -1;
+		}
+		/* my_node_id for native clusters; -1 (matches no node) only on a
+		 * controller-only node, whose id comes at runtime */
+		*_shm_id = (current_id >= 1) ? current_id : -1;
+		_current_id_shm = _shm_id;
+	}
 
 	/* if statistics are disabled, prevent their registration to core */
 	if (clusterer_enable_stats==0)
@@ -445,12 +567,14 @@ static int mod_init(void)
 
 			if (cl->current_node->node_id != current_id) {
 				LM_ERR("Bad 'my_node_id' parameter, value: %d different than"
-					" the node_id property in the 'my_node_info' parameter\n", current_id);
+					" the node_id property in the 'my_node_info' parameter\n", GET_CURRENT_ID);
 				goto error;
 			}
 		}
 	}
 
+	/* DB-native clusters load even with the controller active (hybrid); the DB
+	 * never holds controller clusters */
 	if (db_mode) {
 		/* bind to the mysql module */
 		if (db_bind_mod(&clusterer_db_url, &dr_dbf)) {
@@ -471,11 +595,51 @@ static int mod_init(void)
 			LM_ERR("Failed to load info from DB\n");
 			goto error;
 		}
-
 		dr_dbf.close(db_hdl);
 		db_hdl = NULL;
 	}
 
+	/* a stub per controller-managed cluster_id, so tm/dialog cl_register_cap()
+	 * works before the controller sets identity; a native overlap fails here */
+	if (use_controller) {
+		int _ci;
+		for (_ci = 0; _ci < cl_ctr_stub_count; _ci++) {
+			cluster_info_t *_ex;
+			for (_ex = *cluster_list; _ex; _ex = _ex->next)
+				if (_ex->cluster_id == cl_ctr_stub_ids[_ci]) {
+					LM_ERR("clusterer: cluster_id %d is registered as "
+						"controller-managed (cluster_options use_controller=1) but is "
+						"already defined via native config "
+						"(my_node_info/neighbor_node_info/DB) or listed "
+						"twice - a cluster_id must be unique and either "
+						"controller-managed or native, not both\n",
+						cl_ctr_stub_ids[_ci]);
+					return -1;
+				}
+
+			cluster_info_t *_cl = shm_malloc(sizeof *_cl);
+			if (!_cl) {
+				LM_ERR("clusterer: no shm for cluster %d stub\n",
+				       cl_ctr_stub_ids[_ci]);
+				return -1;
+			}
+			memset(_cl, 0, sizeof *_cl);
+			_cl->cluster_id = cl_ctr_stub_ids[_ci];
+			_cl->controller_managed = 1;
+			_cl->lock = lock_alloc();
+			if (!_cl->lock || lock_init(_cl->lock) == NULL) {
+				LM_ERR("clusterer: lock_alloc failed for cluster %d\n",
+				       cl_ctr_stub_ids[_ci]);
+				shm_free(_cl);
+				return -1;
+			}
+			_cl->next = *cluster_list;
+			*cluster_list = _cl;
+			LM_INFO("clusterer: pre-created controller-managed cluster %d stub\n",
+			        cl_ctr_stub_ids[_ci]);
+		}
+	}
+
 	/* register timer */
 	heartbeats_timer_interval = gcd(ping_interval*1000, ping_timeout);
 	heartbeats_timer_interval = gcd(heartbeats_timer_interval, node_timeout*1000);
@@ -536,7 +700,8 @@ static int mod_init(void)
 	/* check if the cluster IDs in the the sharing tag list are valid */
 	shtag_init_list();
 	shtag_init_reporting();
-	shtag_validate_list();
+	if (!use_controller)
+		shtag_validate_list();
 
 	return 0;
 error:
@@ -559,6 +724,16 @@ static int child_init(int rank)
 			return -1;
 		}
 	}
+
+	/* rank 1 runs once, after every mod_init: a clusterer_controller would
+	 * have bound the controller API by now */
+	if (rank == 1 && use_controller && !clusterer_ctrl_bound)
+		LM_ERR("clusterer: cluster_options declares controller-managed cluster(s) "
+		       "but no clusterer_controller module registered the controller API - "
+		       "they will never obtain a node identity or form. Load the "
+		       "clusterer_controller module, or drop use_controller from "
+		       "cluster_options.\n");
+
 	return 0;
 }
 
@@ -893,7 +1068,7 @@ static mi_response_t *clusterer_list_topology(const mi_params_t *params,
 		if (!node_item)
 			goto error;
 
-		if (add_mi_number(node_item, MI_SSTR("node_id"), current_id) < 0)
+		if (add_mi_number(node_item, MI_SSTR("node_id"), GET_CURRENT_ID) < 0)
 			goto error;
 
 		neigh_arr = add_mi_array(node_item, MI_SSTR("Neighbours"));
@@ -925,7 +1100,7 @@ static mi_response_t *clusterer_list_topology(const mi_params_t *params,
 				}
 
 			if (n_info->link_state == LS_UP)
-				if (add_mi_number(neigh_arr, 0,0, current_id) < 0) {
+				if (add_mi_number(neigh_arr, 0,0, GET_CURRENT_ID) < 0) {
 					lock_release(n_info->lock);
 					goto error;
 				}
@@ -1054,7 +1229,7 @@ static mi_response_t *cluster_send_mi(const mi_params_t *params,
 		return init_mi_param_error();
 	if (node_id < 1)
 		return init_mi_error(400, MI_SSTR("Bad value for 'destination'"));
-	if (node_id == current_id)
+	if (node_id == GET_CURRENT_ID)
 		return init_mi_error(400, MI_SSTR("Local node specified as destination"));
 
 	if (get_mi_string_param(params, "cmd_name", &cmd_name.s, &cmd_name.len) < 0)
@@ -1250,7 +1425,7 @@ static inline void generate_msg_tag(pv_value_t *tag_val, int cluster_id)
 	memcpy(tag_val->rs.s, tmp, len);
 	tag_val->rs.s[len] = '-';
 	tag_val->rs.len = len + 1;
-	tmp = int2str(current_id, &len);
+	tmp = int2str(GET_CURRENT_ID, &len);
 	memcpy(tag_val->rs.s + tag_val->rs.len, tmp, len);
 	tag_val->rs.s[tag_val->rs.len + len] = '-';
 	tag_val->rs.len += len + 1;