cluster.c (14695B)
1 #include "cluster.h" 2 3 #include <math.h> 4 #include <stdio.h> 5 #include <stdlib.h> 6 #include <string.h> 7 8 #include "common/resp.h" 9 #include "common/scheduler.h" 10 #include "common/url_utils.h" 11 #include "domain/config.h" 12 #include "finwo/url-parser.h" 13 #include "rxi/log.h" 14 15 cluster_state_t *cluster_state = NULL; 16 17 static int is_session_not_found(resp_object *resp) { 18 if (!resp) return 0; 19 if (resp->type == RESPT_ERROR && resp->u.s) { 20 return (strstr(resp->u.s, "session") != NULL && strstr(resp->u.s, "not found") != NULL) || 21 (strstr(resp->u.s, "not found") != NULL); 22 } 23 return 0; 24 } 25 26 static resp_object *cluster_forward_to_nodes(const char *cmd_str) { 27 size_t available_count = 0; 28 cluster_node_t **available = cluster_nodes_get_available(cluster_state->nodes, &available_count); 29 30 if (!available || available_count == 0) { 31 free(available); 32 return resp_error_init("ERR no available nodes"); 33 } 34 35 for (size_t i = 0; i < available_count; i++) { 36 cluster_node_t *node = available[i]; 37 38 resp_object *resp = NULL; 39 cluster_node_send_command(node, cmd_str, &resp); 40 41 if (!resp) { 42 node->available = 0; 43 continue; 44 } 45 46 if (is_session_not_found(resp)) { 47 resp_free(resp); 48 continue; 49 } 50 51 if (resp->type == RESPT_ERROR) { 52 node->available = 0; 53 resp_free(resp); 54 continue; 55 } 56 57 free(available); 58 return resp; 59 } 60 61 free(available); 62 return resp_error_init("ERR session not found"); 63 } 64 65 void cluster_init(void) { 66 if (cluster_state) { 67 cluster_shutdown(); 68 } 69 70 cluster_state = calloc(1, sizeof(cluster_state_t)); 71 cluster_state->nodes = cluster_nodes_create(); 72 73 if (!domain_cfg) return; 74 75 resp_object *cluster_nodes = domain_config_get_cluster_nodes(); 76 if (!cluster_nodes || cluster_nodes->type != RESPT_ARRAY) { 77 log_error("cluster: no cluster nodes configured"); 78 return; 79 } 80 81 log_info("cluster: found %zu cluster nodes", cluster_nodes->u.arr.n); 82 for (size_t i = 0; i < cluster_nodes->u.arr.n; i++) { 83 resp_object *elem = &cluster_nodes->u.arr.elem[i]; 84 if (elem->type != RESPT_BULK || !elem->u.s) continue; 85 86 const char *address = elem->u.s; 87 88 struct parsed_url *purl = NULL; 89 if (parse_address_url(address, &purl) != 0) { 90 log_error("cluster: failed to parse address '%s'", address); 91 continue; 92 } 93 94 char *node_name = NULL; 95 if (purl->host && purl->port) { 96 asprintf(&node_name, "%s-%s", purl->host, purl->port); 97 } else if (purl->scheme && strcmp(purl->scheme, "unix") == 0 && purl->path) { 98 const char *basename = strrchr(purl->path, '/'); 99 basename = basename ? basename + 1 : purl->path; 100 asprintf(&node_name, "unix-%s", basename); 101 } else { 102 log_error("cluster: cannot generate node name for address '%s'", address); 103 parsed_url_free(purl); 104 continue; 105 } 106 107 cluster_node_t *node = calloc(1, sizeof(cluster_node_t)); 108 if (cluster_node_init(node, node_name, address, purl->username, purl->password) == 0) { 109 cluster_nodes_add(cluster_state->nodes, node); 110 sched_create(cluster_node_healthcheck_pt, node); 111 log_info("cluster: added node '%s' at %s", node->name, node->address); 112 } else { 113 cluster_node_free(node); 114 free(node); 115 } 116 free(node_name); 117 parsed_url_free(purl); 118 } 119 120 cluster_state->initialized = 1; 121 } 122 123 void cluster_reload(void) { 124 cluster_shutdown(); 125 cluster_init(); 126 } 127 128 void cluster_shutdown(void) { 129 if (!cluster_state) return; 130 if (cluster_state->nodes) { 131 cluster_nodes_free(cluster_state->nodes); 132 } 133 free(cluster_state); 134 cluster_state = NULL; 135 } 136 137 static char *serialize_args(resp_object *args) { 138 char *cmd_str = NULL; 139 size_t cmd_len = 0; 140 resp_serialize(args, &cmd_str, &cmd_len); 141 return cmd_str; 142 } 143 144 resp_object *cluster_session_create(const char *cmd, resp_object *args) { 145 (void)cmd; 146 if (!cluster_state || !cluster_state->nodes) { 147 return resp_error_init("ERR cluster not initialized"); 148 } 149 150 char *cmd_str = serialize_args(args); 151 if (!cmd_str) { 152 return resp_error_init("ERR failed to serialize command"); 153 } 154 155 cluster_node_t *selected_node = NULL; 156 double selected_ratio = -1.0; 157 resp_object *resp = NULL; 158 159 for (int attempt = 0; attempt < 10; attempt++) { 160 size_t available_count = 0; 161 cluster_node_t **available = cluster_nodes_get_available(cluster_state->nodes, &available_count); 162 163 if (!available || available_count == 0) { 164 free(available); 165 break; 166 } 167 168 selected_node = NULL; 169 selected_ratio = -1.0; 170 171 for (size_t i = 0; i < available_count; i++) { 172 cluster_node_t *node = available[i]; 173 174 resp_object *count_args = resp_array_init(); 175 resp_array_append_bulk(count_args, "session.count"); 176 char *count_str = serialize_args(count_args); 177 resp_free(count_args); 178 179 size_t node_session_count = 0; 180 if (count_str) { 181 resp_object *count_resp = NULL; 182 if (cluster_node_send_command(node, count_str, &count_resp) == 0 && count_resp && 183 count_resp->type == RESPT_INT) { 184 node_session_count = (size_t)count_resp->u.i; 185 } 186 if (count_resp) resp_free(count_resp); 187 free(count_str); 188 } 189 190 double ratio = (double)node_session_count / (double)node->weight; 191 if (selected_node == NULL || ratio < selected_ratio) { 192 selected_node = node; 193 selected_ratio = ratio; 194 } 195 } 196 197 free(available); 198 199 if (!selected_node) { 200 break; 201 } 202 203 if (cluster_node_send_command(selected_node, cmd_str, &resp) == 0) { 204 free(cmd_str); 205 return resp; 206 } 207 208 log_debug("cluster: session.create failed on node '%s', marking unavailable", selected_node->name); 209 selected_node->available = 0; 210 if (resp) { 211 resp_free(resp); 212 resp = NULL; 213 } 214 } 215 216 free(cmd_str); 217 if (resp) resp_free(resp); 218 return resp_error_init("ERR all nodes failed"); 219 } 220 221 resp_object *cluster_session_list(const char *cmd, resp_object *args) { 222 (void)cmd; 223 (void)args; 224 if (!cluster_state || !cluster_state->nodes) { 225 return resp_error_init("ERR cluster not initialized"); 226 } 227 228 size_t available_count = 0; 229 cluster_node_t **available = cluster_nodes_get_available(cluster_state->nodes, &available_count); 230 231 if (!available || available_count == 0) { 232 free(available); 233 resp_object *res = resp_array_init(); 234 return res; 235 } 236 237 resp_object *result = resp_array_init(); 238 239 for (size_t i = 0; i < available_count; i++) { 240 cluster_node_t *node = available[i]; 241 242 resp_object *cmd_args = resp_array_init(); 243 resp_array_append_bulk(cmd_args, "session.list"); 244 char *cmd_str = serialize_args(cmd_args); 245 resp_free(cmd_args); 246 247 if (!cmd_str) continue; 248 249 resp_object *resp = NULL; 250 if (cluster_node_send_command(node, cmd_str, &resp) == 0 && resp && resp->type == RESPT_ARRAY) { 251 for (size_t j = 0; j < resp->u.arr.n; j++) { 252 resp_object *elem = &resp->u.arr.elem[j]; 253 resp_object *copy = resp_deep_copy(elem); 254 if (copy) resp_array_append_obj(result, copy); 255 } 256 } 257 if (resp) resp_free(resp); 258 free(cmd_str); 259 } 260 261 free(available); 262 return result; 263 } 264 265 resp_object *cluster_session_info(const char *cmd, resp_object *args) { 266 (void)cmd; 267 if (!cluster_state || !cluster_state->nodes) { 268 return resp_error_init("ERR cluster not initialized"); 269 } 270 271 size_t available_count = 0; 272 cluster_node_t **available = cluster_nodes_get_available(cluster_state->nodes, &available_count); 273 274 if (!available || available_count == 0) { 275 free(available); 276 return resp_error_init("ERR no available nodes"); 277 } 278 279 resp_object *cmd_args = resp_array_init(); 280 for (size_t i = 0; i < args->u.arr.n && i < 4; i++) { 281 resp_array_append_obj(cmd_args, resp_deep_copy(&args->u.arr.elem[i])); 282 } 283 char *cmd_str = serialize_args(cmd_args); 284 resp_free(cmd_args); 285 286 if (!cmd_str) { 287 free(available); 288 return resp_error_init("ERR failed to serialize command"); 289 } 290 291 for (size_t i = 0; i < available_count; i++) { 292 cluster_node_t *node = available[i]; 293 294 resp_object *resp = NULL; 295 if (cluster_node_send_command(node, cmd_str, &resp) == 0) { 296 if (resp && resp->type != RESPT_ERROR) { 297 free(cmd_str); 298 free(available); 299 return resp; 300 } 301 if (resp && !is_session_not_found(resp)) { 302 free(cmd_str); 303 free(available); 304 return resp; 305 } 306 if (resp) resp_free(resp); 307 } 308 } 309 310 free(cmd_str); 311 free(available); 312 return resp_error_init("ERR session not found"); 313 } 314 315 resp_object *cluster_session_destroy(const char *cmd, resp_object *args) { 316 (void)cmd; 317 if (!cluster_state || !cluster_state->nodes) { 318 return resp_error_init("ERR cluster not initialized"); 319 } 320 321 resp_object *cmd_args = resp_array_init(); 322 for (size_t i = 0; i < args->u.arr.n && i < 4; i++) { 323 resp_array_append_obj(cmd_args, resp_deep_copy(&args->u.arr.elem[i])); 324 } 325 char *cmd_str = serialize_args(cmd_args); 326 resp_free(cmd_args); 327 328 if (!cmd_str) { 329 return resp_error_init("ERR failed to serialize command"); 330 } 331 332 return cluster_forward_to_nodes(cmd_str); 333 } 334 335 resp_object *cluster_socket_create_listen(const char *cmd, resp_object *args) { 336 (void)cmd; 337 if (!cluster_state || !cluster_state->nodes) { 338 return resp_error_init("ERR cluster not initialized"); 339 } 340 341 resp_object *cmd_args = resp_array_init(); 342 for (size_t i = 0; i < args->u.arr.n && i < 8; i++) { 343 resp_array_append_obj(cmd_args, resp_deep_copy(&args->u.arr.elem[i])); 344 } 345 char *cmd_str = serialize_args(cmd_args); 346 resp_free(cmd_args); 347 348 if (!cmd_str) { 349 return resp_error_init("ERR failed to serialize command"); 350 } 351 352 return cluster_forward_to_nodes(cmd_str); 353 } 354 355 resp_object *cluster_socket_create_connect(const char *cmd, resp_object *args) { 356 (void)cmd; 357 if (!cluster_state || !cluster_state->nodes) { 358 return resp_error_init("ERR cluster not initialized"); 359 } 360 361 resp_object *cmd_args = resp_array_init(); 362 for (size_t i = 0; i < args->u.arr.n && i < 8; i++) { 363 resp_array_append_obj(cmd_args, resp_deep_copy(&args->u.arr.elem[i])); 364 } 365 char *cmd_str = serialize_args(cmd_args); 366 resp_free(cmd_args); 367 368 if (!cmd_str) { 369 return resp_error_init("ERR failed to serialize command"); 370 } 371 372 return cluster_forward_to_nodes(cmd_str); 373 } 374 375 resp_object *cluster_socket_destroy(const char *cmd, resp_object *args) { 376 (void)cmd; 377 if (!cluster_state || !cluster_state->nodes) { 378 return resp_error_init("ERR cluster not initialized"); 379 } 380 381 resp_object *cmd_args = resp_array_init(); 382 for (size_t i = 0; i < args->u.arr.n && i < 8; i++) { 383 resp_array_append_obj(cmd_args, resp_deep_copy(&args->u.arr.elem[i])); 384 } 385 char *cmd_str = serialize_args(cmd_args); 386 resp_free(cmd_args); 387 388 if (!cmd_str) { 389 return resp_error_init("ERR failed to serialize command"); 390 } 391 392 return cluster_forward_to_nodes(cmd_str); 393 } 394 395 resp_object *cluster_forward_list(const char *cmd, resp_object *args) { 396 (void)cmd; 397 if (!cluster_state || !cluster_state->nodes) { 398 return resp_error_init("ERR cluster not initialized"); 399 } 400 401 resp_object *cmd_args = resp_array_init(); 402 for (size_t i = 0; i < args->u.arr.n && i < 4; i++) { 403 resp_array_append_obj(cmd_args, resp_deep_copy(&args->u.arr.elem[i])); 404 } 405 char *cmd_str = serialize_args(cmd_args); 406 resp_free(cmd_args); 407 408 if (!cmd_str) { 409 return resp_error_init("ERR failed to serialize command"); 410 } 411 412 return cluster_forward_to_nodes(cmd_str); 413 } 414 415 resp_object *cluster_forward_create(const char *cmd, resp_object *args) { 416 (void)cmd; 417 if (!cluster_state || !cluster_state->nodes) { 418 return resp_error_init("ERR cluster not initialized"); 419 } 420 421 resp_object *cmd_args = resp_array_init(); 422 for (size_t i = 0; i < args->u.arr.n && i < 8; i++) { 423 resp_array_append_obj(cmd_args, resp_deep_copy(&args->u.arr.elem[i])); 424 } 425 char *cmd_str = serialize_args(cmd_args); 426 resp_free(cmd_args); 427 428 if (!cmd_str) { 429 return resp_error_init("ERR failed to serialize command"); 430 } 431 432 return cluster_forward_to_nodes(cmd_str); 433 } 434 435 resp_object *cluster_forward_destroy(const char *cmd, resp_object *args) { 436 (void)cmd; 437 if (!cluster_state || !cluster_state->nodes) { 438 return resp_error_init("ERR cluster not initialized"); 439 } 440 441 resp_object *cmd_args = resp_array_init(); 442 for (size_t i = 0; i < args->u.arr.n && i < 8; i++) { 443 resp_array_append_obj(cmd_args, resp_deep_copy(&args->u.arr.elem[i])); 444 } 445 char *cmd_str = serialize_args(cmd_args); 446 resp_free(cmd_args); 447 448 if (!cmd_str) { 449 return resp_error_init("ERR failed to serialize command"); 450 } 451 452 return cluster_forward_to_nodes(cmd_str); 453 } 454 455 resp_object *cluster_session_count(const char *cmd, resp_object *args) { 456 (void)cmd; 457 (void)args; 458 if (!cluster_state || !cluster_state->nodes) { 459 return resp_error_init("ERR cluster not initialized"); 460 } 461 462 size_t available_count = 0; 463 cluster_node_t **available = cluster_nodes_get_available(cluster_state->nodes, &available_count); 464 465 if (!available || available_count == 0) { 466 free(available); 467 resp_object *res = malloc(sizeof(resp_object)); 468 if (!res) return NULL; 469 res->type = RESPT_INT; 470 res->u.i = 0; 471 return res; 472 } 473 474 size_t total_count = 0; 475 476 for (size_t i = 0; i < available_count; i++) { 477 cluster_node_t *node = available[i]; 478 479 resp_object *cmd_args = resp_array_init(); 480 resp_array_append_bulk(cmd_args, "session.count"); 481 char *cmd_str = serialize_args(cmd_args); 482 resp_free(cmd_args); 483 484 if (!cmd_str) continue; 485 486 resp_object *resp = NULL; 487 if (cluster_node_send_command(node, cmd_str, &resp) == 0 && resp && resp->type == RESPT_INT) { 488 total_count += (size_t)resp->u.i; 489 } 490 if (resp) resp_free(resp); 491 free(cmd_str); 492 } 493 494 free(available); 495 496 resp_object *res = malloc(sizeof(resp_object)); 497 if (!res) return NULL; 498 res->type = RESPT_INT; 499 res->u.i = (long long)total_count; 500 return res; 501 } 502 503 resp_object *cluster_system_load(const char *cmd, resp_object *args) { 504 (void)cmd; 505 (void)args; 506 507 double loadavg[3]; 508 if (getloadavg(loadavg, 3) != 3) { 509 return resp_error_init("ERR failed to get load average"); 510 } 511 512 resp_object *res = resp_array_init(); 513 char buf[64]; 514 515 resp_array_append_bulk(res, "1min"); 516 snprintf(buf, sizeof(buf), "%.2f", loadavg[0]); 517 resp_array_append_bulk(res, buf); 518 519 resp_array_append_bulk(res, "5min"); 520 snprintf(buf, sizeof(buf), "%.2f", loadavg[1]); 521 resp_array_append_bulk(res, buf); 522 523 resp_array_append_bulk(res, "15min"); 524 snprintf(buf, sizeof(buf), "%.2f", loadavg[2]); 525 resp_array_append_bulk(res, buf); 526 527 return res; 528 }