Skip to content

clusterer_controller: zero-config HA clusterer control module with automatic sharing tag management via encrypted UDP multicast #4072

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Lt-Flash wants to merge 1 commit into from
Closed

clusterer_controller: zero-config HA clusterer control module with automatic sharing tag management via encrypted UDP multicast #4072

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions ipc.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -274,6 +274,7 @@ int ipc_recv_sync_reply(void **param)
return 0;
}

int ipc_running_rpc_job;
void ipc_handle_job(int fd)
{
ipc_job job;
Expand Down Expand Up @@ -301,7 +302,9 @@ void ipc_handle_job(int fd)

/* custom handling for RPC type */
if (job.handler_type==ipc_rpc_type) {
ipc_running_rpc_job = 1;
((ipc_rpc_f*)job.payload1)( job.snd_proc, job.payload2);
ipc_running_rpc_job = 0;
} else {
/* generic registered type */
ipc_handlers[job.handler_type].func( job.snd_proc, job.payload1);
Expand Down
9 changes: 9 additions & 0 deletions ipc.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,15 @@ void ipc_handle_job(int fd);
void ipc_handle_all_pending_jobs(int fd);


/*
* A way of checking whether a process is already running within a
* ipc_dispatch_rpc() context, in order to avoid double async dispatching,
* which is effectively a waste of resources.
*/
extern int ipc_running_rpc_job;
static inline int ipc_is_async_dispatch(void) { return ipc_running_rpc_job; }


/* internal functions */
int init_ipc(void);

Expand Down
1 change: 0 additions & 1 deletion modules/clusterer/api.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -327,6 +327,5 @@ static inline module_dependency_t *get_deps_clusterer(const param_export_t *para
return alloc_module_dep(MOD_TYPE_DEFAULT, "clusterer", DEP_ABORT);
}


#endif /* CLUSTERER_API_H */

147 changes: 104 additions & 43 deletions modules/clusterer/clusterer.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,7 @@ void sync_check_timer(utime_t ticks, void *param)
lock_start_read(cl_list_lock);

for (cl = *cluster_list; cl; cl = cl->next) {
if (!cl->current_node) continue;
lock_get(cl->current_node->lock);
if (!(cl->current_node->flags & NODE_STATE_ENABLED)) {
lock_release(cl->current_node->lock);
Expand All @@ -106,11 +107,21 @@ void sync_check_timer(utime_t ticks, void *param)
cap->flags &= ~(CAP_SYNC_PENDING|CAP_SYNC_STARTUP);
sr_set_status(cl_srg, STR2CI(cap->reg.sr_id), CAP_SR_SYNCED,
STR2CI(CAP_SR_STATUS_STR(CAP_SR_SYNCED)), 0);
sr_add_report_fmt(cl_srg, STR2CI(cap->reg.sr_id), 0,
"ERROR: Sync request aborted! (no donor found in due time)"
" => fallback to synced state");
LM_ERR("Sync request aborted! (no donor found in due time)"
", falling back to synced state\n");
if (cl->node_list == NULL) {
/* no peers — first/lone node, self-sync is expected */
sr_add_report_fmt(cl_srg, STR2CI(cap->reg.sr_id), 0,
"No peers present — self-synced as first node in cluster");
LM_DBG("No peers in cluster %d, capability '%.*s' "
"self-marked as synced (first/lone node)\n",
cl->cluster_id,
cap->reg.name.len, cap->reg.name.s);
} else {
sr_add_report_fmt(cl_srg, STR2CI(cap->reg.sr_id), 0,
"ERROR: Sync request aborted! (no donor found in due time)"
" => fallback to synced state");
LM_ERR("Sync request aborted! (no donor found in due time)"
", falling back to synced state\n");
}
/* send update about the state of this capability */
send_single_cap_update(cl, cap, 1);

Expand Down Expand Up @@ -152,7 +163,7 @@ int cl_set_state(int cluster_id, int node_id, enum cl_node_state state)
return -1;
}

if (node_id != current_id) {
if (node_id != cluster_self_id(cluster)) {
node = get_node_by_id(cluster, node_id);
if (!node) {
lock_stop_read(cl_list_lock);
Expand Down Expand Up @@ -187,13 +198,16 @@ int cl_set_state(int cluster_id, int node_id, enum cl_node_state state)
LM_INFO("Set state: %s for node: %d in cluster: %d\n",
state ? "enabled" : "disabled", node_id, cluster_id);

if (db_mode && update_db_state(cluster_id, node_id, state) < 0)
if (cl_db_mode(cluster) && update_db_state(cluster_id, node_id, state) < 0)
LM_ERR("Failed to update state in clusterer DB for node [%d] cluster [%d]\n",
node_id, cluster_id);

return 0;
}

if (!cluster->current_node)
return -1;

lock_get(cluster->current_node->lock);

if (state == STATE_DISABLED && cluster->current_node->flags & NODE_STATE_ENABLED)
Expand Down Expand Up @@ -223,7 +237,7 @@ int cl_set_state(int cluster_id, int node_id, enum cl_node_state state)
LM_INFO("Set state: %s for local node in cluster: %d\n",
state ? "enabled" : "disabled", cluster_id);

if (db_mode && update_db_state(cluster_id, current_id, state) < 0)
if (cl_db_mode(cluster) && update_db_state(cluster_id, current_id, state) < 0)
LM_ERR("Failed to update state in clusterer DB for cluster [%d]\n", cluster->cluster_id);

return 0;
Expand Down Expand Up @@ -524,11 +538,27 @@ clusterer_bcast_msg(bin_packet_t *packet, int dst_cid,
return CLUSTERER_SEND_ERR;
}

/* This node's id in @cluster_id, resolved without taking cl_list_lock: clusters
* persist for the module's lifetime, and callers may already hold cl_list_lock
* (read) or a different lock (shtags) whose order is cl_list_lock -> shtags_lock,
* so taking cl_list_lock here would risk a reverse-order deadlock. Matches the
* lock-free nature of the GET_CURRENT_ID this replaces. Returns -1 if the
* cluster is unknown or this node's identity in it is not yet established. */
static int trailer_self_id(int cluster_id)
{
cluster_info_t *cl;

for (cl = *cluster_list; cl; cl = cl->next)
if (cl->cluster_id == cluster_id)
return cluster_self_id(cl);
return -1;
}

int msg_add_trailer(bin_packet_t *packet, int cluster_id, int dst_id)
{
if (bin_push_int(packet, cluster_id) < 0)
return -1;
if (bin_push_int(packet, current_id) < 0)
if (bin_push_int(packet, trailer_self_id(cluster_id)) < 0)
return -1;
if (bin_push_int(packet, dst_id) < 0)
return -1;
Expand Down Expand Up @@ -810,7 +840,7 @@ static void handle_cap_update(bin_packet_t *packet, node_info_t *source)
for (i = 0; i < nr_nodes; i++) {
bin_pop_int(packet, &node_id);

if (node_id == current_id) {
if (node_id == cluster_self_id(source->cluster)) {
bin_pop_int(packet, &nr_cap);
for (j = 0; j < nr_cap; j++) {
bin_pop_str(packet, &cap);
Expand Down Expand Up @@ -986,12 +1016,15 @@ static void handle_remove_node(bin_packet_t *packet, cluster_info_t *cl)
bin_pop_int(packet, &target_node);
LM_DBG("Received remove node command for node id: [%d]\n", target_node);

if (db_mode) {
if (cl_db_mode(cl)) {
LM_DBG("We are in DB mode, ignoring received remove node command\n");
return;
}

if (target_node == current_id) {
if (target_node == cluster_self_id(cl)) {
if (!cl->current_node)
return;

lock_get(cl->current_node->lock);

if (cl->current_node->flags & NODE_STATE_ENABLED) {
Expand Down Expand Up @@ -1038,14 +1071,9 @@ void bin_rcv_cl_extra_packets(bin_packet_t *packet, int packet_type,
LM_DBG("received clusterer message from: %s:%hu with source id: %d and"
" cluster id: %d\n", ip, port, source_id, cluster_id);

if (source_id == current_id) {
LM_ERR("Received message with bad source - same node id as this instance\n");
return;
}

gettimeofday(&now, NULL);

if (!db_mode && packet_type == CLUSTERER_REMOVE_NODE)
if ((!db_mode || use_controller) && packet_type == CLUSTERER_REMOVE_NODE)
lock_start_write(cl_list_lock);
else
lock_start_read(cl_list_lock);
Expand All @@ -1057,6 +1085,11 @@ void bin_rcv_cl_extra_packets(bin_packet_t *packet, int packet_type,
goto exit;
}

if (source_id == cluster_self_id(cl)) {
LM_ERR("Received message with bad source - same node id as this instance\n");
goto exit;
}

lock_get(cl->current_node->lock);
if (!(cl->current_node->flags & NODE_STATE_ENABLED)) {
lock_release(cl->current_node->lock);
Expand Down Expand Up @@ -1094,7 +1127,7 @@ void bin_rcv_cl_extra_packets(bin_packet_t *packet, int packet_type,
} else
lock_release(node->lock);

if (dest_id != current_id) {
if (dest_id != cluster_self_id(cl)) {
/* route the message */
bin_push_int(packet, cluster_id);
bin_push_int(packet, source_id);
Expand Down Expand Up @@ -1146,7 +1179,7 @@ void bin_rcv_cl_extra_packets(bin_packet_t *packet, int packet_type,
}

exit:
if (!db_mode && packet_type == CLUSTERER_REMOVE_NODE)
if ((!db_mode || use_controller) && packet_type == CLUSTERER_REMOVE_NODE)
lock_stop_write(cl_list_lock);
else
lock_stop_read(cl_list_lock);
Expand All @@ -1172,12 +1205,7 @@ void bin_rcv_cl_packets(bin_packet_t *packet, int packet_type,
LM_DBG("received clusterer message from: %s:%hu with source id: %d and "
"cluster id: %d\n", ip, port, source_id, cl_id);

if (source_id == current_id) {
LM_ERR("Received message with bad source - same node id as this instance\n");
return;
}

if (!db_mode && (packet_type == CLUSTERER_NODE_DESCRIPTION ||
if ((!db_mode || use_controller) && (packet_type == CLUSTERER_NODE_DESCRIPTION ||
packet_type == CLUSTERER_FULL_TOP_UPDATE))
lock_start_write(cl_list_lock);
else
Expand All @@ -1189,6 +1217,22 @@ void bin_rcv_cl_packets(bin_packet_t *packet, int packet_type,
goto exit;
}

/* current_node is legitimately NULL while this node's identity is being
* (re)established for a dynamically constructed cluster (clusterer_ctrl
* update_identity: the cluster can already exist and receive BIN packets
* before current_node is assigned). Drop the packet instead of
* dereferencing NULL. */
if (!cl->current_node) {
LM_INFO("Received message for cluster [%d] before local identity is "
"established, ignoring\n", cl_id);
goto exit;
}

if (source_id == cluster_self_id(cl)) {
LM_ERR("Received message with bad source - same node id as this instance\n");
goto exit;
}

lock_get(cl->current_node->lock);
if (!(cl->current_node->flags & NODE_STATE_ENABLED)) {
lock_release(cl->current_node->lock);
Expand All @@ -1201,7 +1245,7 @@ void bin_rcv_cl_packets(bin_packet_t *packet, int packet_type,

if (!node) {
LM_INFO("Received message with unknown source id [%d]\n", source_id);
if (!db_mode)
if (!cl_db_mode(cl))
handle_internal_msg_unknown(packet, cl, packet_type, &ri->src_su,
ri->proto, source_id);
} else {
Expand All @@ -1228,7 +1272,7 @@ void bin_rcv_cl_packets(bin_packet_t *packet, int packet_type,
}

exit:
if (!db_mode && (packet_type == CLUSTERER_NODE_DESCRIPTION ||
if ((!db_mode || use_controller) && (packet_type == CLUSTERER_NODE_DESCRIPTION ||
packet_type == CLUSTERER_FULL_TOP_UPDATE))
lock_stop_write(cl_list_lock);
else
Expand Down Expand Up @@ -1317,11 +1361,6 @@ static void bin_rcv_mod_packets(bin_packet_t *packet, int packet_type,
LM_DBG("received bin packet from: %s:%hu with source id: %d and cluster id: %d\n",
ip, port, source_id, cluster_id);

if (source_id == current_id) {
LM_ERR("Received message with bad source - same node id as this instance\n");
return;
}

cap = (struct capability_reg *)ptr;
if (!cap) {
LM_ERR("Failed to get bin callback parameter\n");
Expand All @@ -1338,6 +1377,11 @@ static void bin_rcv_mod_packets(bin_packet_t *packet, int packet_type,
goto exit;
}

if (source_id == cluster_self_id(cl)) {
LM_ERR("Received message with bad source - same node id as this instance\n");
goto exit;
}

lock_get(cl->current_node->lock);
if (!(cl->current_node->flags & NODE_STATE_ENABLED)) {
lock_release(cl->current_node->lock);
Expand Down Expand Up @@ -1385,7 +1429,7 @@ static void bin_rcv_mod_packets(bin_packet_t *packet, int packet_type,
} else
lock_release(node->lock);

if (dest_id != current_id) {
if (dest_id != cluster_self_id(cl)) {
/* route the message */
bin_push_int(packet, cluster_id);
bin_push_int(packet, source_id);
Expand Down Expand Up @@ -1460,6 +1504,7 @@ int send_single_cap_update(cluster_info_t *cluster, struct local_cap *cap,

timestamp = time(NULL);

if (!cluster->current_node) return -1;
lock_get(cluster->current_node->lock);

for (neigh = cluster->current_node->neighbour_list; neigh;
Expand All @@ -1478,7 +1523,7 @@ int send_single_cap_update(cluster_info_t *cluster, struct local_cap *cap,
return -1;
}
bin_push_int(&packet, cluster->cluster_id);
bin_push_int(&packet, current_id);
bin_push_int(&packet, cluster_self_id(cluster));

bin_push_int(&packet, ++cluster->current_node->cap_seq_no);
bin_push_int(&packet, timestamp);
Expand All @@ -1487,7 +1532,7 @@ int send_single_cap_update(cluster_info_t *cluster, struct local_cap *cap,

/* only the current node */
bin_push_int(&packet, 1);
bin_push_int(&packet, current_id);
bin_push_int(&packet, cluster_self_id(cluster));

/* only a single capability */
bin_push_int(&packet, 1);
Expand All @@ -1497,7 +1542,7 @@ int send_single_cap_update(cluster_info_t *cluster, struct local_cap *cap,
bin_push_int(&packet, 0); /* don't require reply */

bin_push_int(&packet, 1); /* path length is 1, only current node at this point */
bin_push_int(&packet, current_id);
bin_push_int(&packet, cluster_self_id(cluster));
bin_get_buffer(&packet, &bin_buffer);

for (i = 0; i < no_dests; i++)
Expand Down Expand Up @@ -1545,7 +1590,7 @@ int send_cap_update(node_info_t *dest_node, int require_reply)
return -1;
}
bin_push_int(&packet, dest_node->cluster->cluster_id);
bin_push_int(&packet, current_id);
bin_push_int(&packet, cluster_self_id(dest_node->cluster));

lock_get(dest_node->cluster->current_node->lock);

Expand All @@ -1560,7 +1605,7 @@ int send_cap_update(node_info_t *dest_node, int require_reply)
for (cl_cap = dest_node->cluster->capabilities, nr_cap = 0; cl_cap;
cl_cap = cl_cap->next, nr_cap++) ;
if (nr_cap) {
bin_push_int(&packet, current_id);
bin_push_int(&packet, cluster_self_id(dest_node->cluster));
bin_push_int(&packet, nr_cap);
for (cl_cap=dest_node->cluster->capabilities;cl_cap;cl_cap=cl_cap->next) {
bin_push_str(&packet, &cl_cap->reg.name);
Expand Down Expand Up @@ -1591,7 +1636,7 @@ int send_cap_update(node_info_t *dest_node, int require_reply)
bin_push_int(&packet, require_reply);

bin_push_int(&packet, 1); /* path length is 1, only current node at this point */
bin_push_int(&packet, current_id);
bin_push_int(&packet, cluster_self_id(dest_node->cluster));
bin_get_buffer(&packet, &bin_buffer);

if (msg_send(dest_node->cluster->send_sock, dest_node->proto, &dest_node->addr,
Expand Down Expand Up @@ -1744,9 +1789,25 @@ int cl_register_cap(str *cap, cl_packet_cb_f packet_cb, cl_event_cb_f event_cb,

cluster = get_cluster_by_id(cluster_id);
if (!cluster) {
LM_ERR("cluster id %d is not defined in the %s\n", cluster_id,
db_mode ? "DB" : "script");
return -1;
if (use_controller) {
cluster = shm_malloc(sizeof *cluster);
if (!cluster) { LM_ERR("no shm\n"); return -1; }
memset(cluster, 0, sizeof *cluster);
cluster->cluster_id = cluster_id;
cluster->controller_managed = 1; /* on-demand controller stub */
if ((cluster->lock = lock_alloc()) == NULL || !lock_init(cluster->lock)) {
shm_free(cluster); return -1;
}
if (cl_list_lock) lock_start_write(cl_list_lock);
cluster->next = *cluster_list;
*cluster_list = cluster;
if (cl_list_lock) lock_stop_write(cl_list_lock);
LM_INFO("clusterer: auto-created stub for cluster %d\n", cluster_id);
} else {
LM_ERR("cluster id %d is not defined in the %s\n", cluster_id,
db_mode ? "DB" : "script");
return -1;
}
}

new_cl_cap = shm_malloc(sizeof *new_cl_cap + cap->len + CAP_SR_ID_PREFIX_LEN);
Expand Down
Loading