| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include "rrdcollector-internals.h" |
| 4 | #include "rrdfunctions-internals.h" |
| 5 | #include "rrdfunctions-inflight.h" |
| 6 | |
| 7 | struct rrd_function_inflight { |
| 8 | bool used; |
| 9 | |
| 10 | RRDHOST *host; |
| 11 | nd_uuid_t transaction_uuid; |
| 12 | const char *transaction; |
| 13 | const char *cmd; |
| 14 | const char *sanitized_cmd; |
| 15 | const char *source; |
| 16 | size_t sanitized_cmd_length; |
| 17 | int timeout; |
| 18 | bool cancelled; |
| 19 | usec_t stop_monotonic_ut; |
| 20 | |
| 21 | HTTP_ACCESS user_access; |
| 22 | |
| 23 | BUFFER *payload; |
| 24 | |
| 25 | const DICTIONARY_ITEM *host_function_acquired; |
| 26 | |
| 27 | // the collector |
| 28 | // we acquire this structure at the beginning, |
| 29 | // and we release it at the end |
| 30 | struct rrd_host_function *rdcf; |
| 31 | |
| 32 | struct { |
| 33 | BUFFER *wb; |
| 34 | |
| 35 | // in async mode, |
| 36 | // the function to call to send the result back |
| 37 | rrd_function_result_callback_t cb; |
| 38 | void *data; |
| 39 | } result; |
| 40 | |
| 41 | struct { |
| 42 | // to be called in sync mode |
| 43 | // while the function is running |
| 44 | // to check if the function has been canceled |
| 45 | rrd_function_is_cancelled_cb_t cb; |
| 46 | void *data; |
| 47 | } is_cancelled; |
| 48 | |
| 49 | struct { |
| 50 | // to be registered by the function itself |
| 51 | // used to signal the function to cancel |
| 52 | rrd_function_cancel_cb_t cb; |
| 53 | void *data; |
| 54 | } canceller; |
| 55 | |
| 56 | struct { |
| 57 | // callback to receive progress reports from function |
| 58 | rrd_function_progress_cb_t cb; |
| 59 | void *data; |
| 60 | } progress; |
| 61 | |
| 62 | struct { |
| 63 | // to be registered by the function itself |
| 64 | // used to send progress requests to function |
| 65 | rrd_function_progresser_cb_t cb; |
| 66 | void *data; |
| 67 | } progresser; |
| 68 | }; |
| 69 | |
| 70 | static DICTIONARY *rrd_functions_inflight_requests = NULL; |
| 71 | |
| 72 | static void rrd_function_cancel_inflight(struct rrd_function_inflight *r); |
| 73 | |
| 74 | // ---------------------------------------------------------------------------- |
| 75 | |
| 76 | static void rrd_functions_inflight_cleanup(struct rrd_function_inflight *r) { |
| 77 | buffer_free(r->payload); |
| 78 | freez((void *)r->transaction); |
| 79 | freez((void *)r->cmd); |
| 80 | freez((void *)r->sanitized_cmd); |
| 81 | freez((void *)r->source); |
| 82 | |
| 83 | r->payload = NULL; |
| 84 | r->transaction = NULL; |
| 85 | r->cmd = NULL; |
| 86 | r->sanitized_cmd = NULL; |
| 87 | } |
| 88 | |
| 89 | static void rrd_functions_inflight_delete_cb(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *data __maybe_unused) { |
| 90 | struct rrd_function_inflight *r = value; |
| 91 | |
| 92 | // internal_error(true, "FUNCTIONS: transaction '%s' finished", r->transaction); |
| 93 | |
| 94 | rrd_functions_inflight_cleanup(r); |
| 95 | dictionary_acquired_item_release(r->host->functions, r->host_function_acquired); |
| 96 | } |
| 97 | |
| 98 | void rrd_functions_inflight_init(void) { |
| 99 | if(rrd_functions_inflight_requests) |
| 100 | return; |
| 101 | |
| 102 | rrd_functions_inflight_requests = dictionary_create_advanced(DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_FIXED_SIZE, NULL, sizeof(struct rrd_function_inflight)); |
| 103 | |
| 104 | dictionary_register_delete_callback(rrd_functions_inflight_requests, rrd_functions_inflight_delete_cb, NULL); |
| 105 | } |
| 106 | |
| 107 | void rrd_functions_inflight_destroy(void) { |
| 108 | if(!rrd_functions_inflight_requests) |
| 109 | return; |
| 110 | |
| 111 | dictionary_destroy(rrd_functions_inflight_requests); |
| 112 | rrd_functions_inflight_requests = NULL; |
| 113 | } |
| 114 | |
| 115 | static void rrd_inflight_async_function_register_canceller_cb(void *register_canceller_cb_data, rrd_function_cancel_cb_t canceller_cb, void *canceller_cb_data) { |
| 116 | struct rrd_function_inflight *r = register_canceller_cb_data; |
| 117 | r->canceller.cb = canceller_cb; |
| 118 | r->canceller.data = canceller_cb_data; |
| 119 | } |
| 120 | |
| 121 | static void rrd_inflight_async_function_register_progresser_cb(void *register_progresser_cb_data, rrd_function_progresser_cb_t progresser_cb, void *progresser_cb_data) { |
| 122 | struct rrd_function_inflight *r = register_progresser_cb_data; |
| 123 | r->progresser.cb = progresser_cb; |
| 124 | r->progresser.data = progresser_cb_data; |
| 125 | } |
| 126 | |
| 127 | // ---------------------------------------------------------------------------- |
| 128 | // waiting for async function completion |
| 129 | |
| 130 | struct rrd_function_call_wait { |
| 131 | RRDHOST *host; |
| 132 | const DICTIONARY_ITEM *host_function_acquired; |
| 133 | char *transaction; |
| 134 | |
| 135 | bool free_with_signal; |
| 136 | bool data_are_ready; |
| 137 | netdata_mutex_t mutex; |
| 138 | netdata_cond_t cond; |
| 139 | int code; |
| 140 | }; |
| 141 | |
| 142 | static void rrd_inflight_function_cleanup(RRDHOST *host __maybe_unused, const char *transaction) { |
| 143 | dictionary_del(rrd_functions_inflight_requests, transaction); |
| 144 | dictionary_garbage_collect(rrd_functions_inflight_requests); |
| 145 | } |
| 146 | |
| 147 | static void rrd_function_call_wait_free(struct rrd_function_call_wait *tmp) { |
| 148 | rrd_inflight_function_cleanup(tmp->host, tmp->transaction); |
| 149 | freez(tmp->transaction); |
| 150 | |
| 151 | netdata_cond_destroy(&tmp->cond); |
| 152 | netdata_mutex_destroy(&tmp->mutex); |
| 153 | freez(tmp); |
| 154 | } |
| 155 | |
| 156 | static void rrd_async_function_signal_when_ready(BUFFER *temp_wb __maybe_unused, int code, void *callback_data) { |
| 157 | struct rrd_function_call_wait *tmp = callback_data; |
| 158 | bool we_should_free = false; |
| 159 | |
| 160 | netdata_mutex_lock(&tmp->mutex); |
| 161 | |
| 162 | // since we got the mutex, |
| 163 | // the waiting thread is either in cond_timedwait() |
| 164 | // or gave up and left. |
| 165 | |
| 166 | tmp->code = code; |
| 167 | tmp->data_are_ready = true; |
| 168 | |
| 169 | if(tmp->free_with_signal) |
| 170 | we_should_free = true; |
| 171 | |
| 172 | netdata_cond_signal(&tmp->cond); |
| 173 | |
| 174 | netdata_mutex_unlock(&tmp->mutex); |
| 175 | |
| 176 | if(we_should_free) { |
| 177 | buffer_free(temp_wb); |
| 178 | rrd_function_call_wait_free(tmp); |
| 179 | } |
| 180 | } |
| 181 | |
| 182 | static void rrd_inflight_async_function_nowait_finished(BUFFER *wb, int code, void *data) { |
| 183 | struct rrd_function_inflight *r = data; |
| 184 | |
| 185 | if(r->result.cb) |
| 186 | r->result.cb(wb, code, r->result.data); |
| 187 | |
| 188 | rrd_inflight_function_cleanup(r->host, r->transaction); |
| 189 | } |
| 190 | |
| 191 | static bool rrd_inflight_async_function_is_cancelled(void *data) { |
| 192 | struct rrd_function_inflight *r = data; |
| 193 | return __atomic_load_n(&r->cancelled, __ATOMIC_RELAXED); |
| 194 | } |
| 195 | |
| 196 | static inline int rrd_call_function_async_and_dont_wait(struct rrd_function_inflight *r) { |
| 197 | struct rrd_function_execute rfe = { |
| 198 | .transaction = &r->transaction_uuid, |
| 199 | .function = r->sanitized_cmd, |
| 200 | .payload = r->payload, |
| 201 | .user_access = r->user_access, |
| 202 | .source = r->source, |
| 203 | .stop_monotonic_ut = &r->stop_monotonic_ut, |
| 204 | .result = { |
| 205 | .wb = r->result.wb, |
| 206 | .cb = rrd_inflight_async_function_nowait_finished, |
| 207 | .data = r, |
| 208 | }, |
| 209 | .progress = { |
| 210 | .cb = r->progress.cb, |
| 211 | .data = r->progress.data, |
| 212 | }, |
| 213 | .is_cancelled = { |
| 214 | .cb = rrd_inflight_async_function_is_cancelled, |
| 215 | .data = r, |
| 216 | }, |
| 217 | .register_canceller = { |
| 218 | .cb = rrd_inflight_async_function_register_canceller_cb, |
| 219 | .data = r, |
| 220 | }, |
| 221 | .register_progresser = { |
| 222 | .cb = rrd_inflight_async_function_register_progresser_cb, |
| 223 | .data = r, |
| 224 | }, |
| 225 | }; |
| 226 | int code = r->rdcf->execute_cb(&rfe, r->rdcf->execute_cb_data); |
| 227 | |
| 228 | return code; |
| 229 | } |
| 230 | |
| 231 | static int rrd_call_function_async_and_wait(struct rrd_function_inflight *r) { |
| 232 | struct rrd_function_call_wait *tmp = mallocz(sizeof(struct rrd_function_call_wait)); |
| 233 | tmp->free_with_signal = false; |
| 234 | tmp->data_are_ready = false; |
| 235 | tmp->host = r->host; |
| 236 | tmp->host_function_acquired = r->host_function_acquired; |
| 237 | tmp->transaction = strdupz(r->transaction); |
| 238 | netdata_mutex_init(&tmp->mutex); |
| 239 | netdata_cond_init(&tmp->cond); |
| 240 | |
| 241 | // we need a temporary BUFFER, because we may time out and the caller supplied one may vanish, |
| 242 | // so we create a new one we guarantee will survive until the collector finishes... |
| 243 | |
| 244 | bool we_should_free = false; |
| 245 | BUFFER *temp_wb = buffer_create(1024, &netdata_buffers_statistics.buffers_functions); // we need it because we may give up on it |
| 246 | temp_wb->content_type = r->result.wb->content_type; |
| 247 | |
| 248 | struct rrd_function_execute rfe = { |
| 249 | .transaction = &r->transaction_uuid, |
| 250 | .function = r->sanitized_cmd, |
| 251 | .payload = r->payload, |
| 252 | .user_access = r->user_access, |
| 253 | .source = r->source, |
| 254 | .stop_monotonic_ut = &r->stop_monotonic_ut, |
| 255 | .result = { |
| 256 | .wb = temp_wb, |
| 257 | |
| 258 | // we overwrite the result callbacks, |
| 259 | // so that we can clean up the allocations made |
| 260 | .cb = rrd_async_function_signal_when_ready, |
| 261 | .data = tmp, |
| 262 | }, |
| 263 | .progress = { |
| 264 | .cb = r->progress.cb, |
| 265 | .data = r->progress.data, |
| 266 | }, |
| 267 | .is_cancelled = { |
| 268 | .cb = rrd_inflight_async_function_is_cancelled, |
| 269 | .data = r, |
| 270 | }, |
| 271 | .register_canceller = { |
| 272 | .cb = rrd_inflight_async_function_register_canceller_cb, |
| 273 | .data = r, |
| 274 | }, |
| 275 | .register_progresser = { |
| 276 | .cb = rrd_inflight_async_function_register_progresser_cb, |
| 277 | .data = r, |
| 278 | }, |
| 279 | }; |
| 280 | int code = r->rdcf->execute_cb(&rfe, r->rdcf->execute_cb_data); |
| 281 | |
| 282 | // this has to happen after we execute the callback |
| 283 | // because if an async call is responded in sync mode, there will be a deadlock. |
| 284 | netdata_mutex_lock(&tmp->mutex); |
| 285 | |
| 286 | if (code == HTTP_RESP_OK || tmp->data_are_ready) { |
| 287 | bool cancelled = false; |
| 288 | int rc = 0; |
| 289 | while (rc == 0 && !cancelled && !tmp->data_are_ready) { |
| 290 | usec_t now_mono_ut = now_monotonic_usec(); |
| 291 | usec_t stop_mono_ut = __atomic_load_n(&r->stop_monotonic_ut, __ATOMIC_RELAXED) + RRDFUNCTIONS_TIMEOUT_EXTENSION_UT; |
| 292 | if(now_mono_ut > stop_mono_ut) { |
| 293 | rc = UV_ETIMEDOUT; |
| 294 | break; |
| 295 | } |
| 296 | |
| 297 | // wait for 10ms, and loop again... |
| 298 | // the mutex is unlocked within cond_timedwait() |
| 299 | rc = netdata_cond_timedwait(&tmp->cond, &tmp->mutex, 10 * NSEC_PER_MSEC); |
| 300 | // the mutex is again ours |
| 301 | |
| 302 | if(rc == UV_ETIMEDOUT) { |
| 303 | // 10ms have passed |
| 304 | |
| 305 | rc = 0; |
| 306 | if (!tmp->data_are_ready && r->is_cancelled.cb && |
| 307 | r->is_cancelled.cb(r->is_cancelled.data)) { |
| 308 | // internal_error(true, "FUNCTIONS: transaction '%s' is cancelled while waiting for response", |
| 309 | // r->transaction); |
| 310 | cancelled = true; |
| 311 | rrd_function_cancel_inflight(r); |
| 312 | break; |
| 313 | } |
| 314 | } |
| 315 | } |
| 316 | |
| 317 | if (tmp->data_are_ready) { |
| 318 | // we have a response |
| 319 | |
| 320 | buffer_contents_replace(r->result.wb, buffer_tostring(temp_wb), buffer_strlen(temp_wb)); |
| 321 | r->result.wb->content_type = temp_wb->content_type; |
| 322 | r->result.wb->expires = temp_wb->expires; |
| 323 | |
| 324 | if(r->result.wb->expires) |
| 325 | buffer_cacheable(r->result.wb); |
| 326 | else |
| 327 | buffer_no_cacheable(r->result.wb); |
| 328 | |
| 329 | code = tmp->code; |
| 330 | |
| 331 | tmp->free_with_signal = false; |
| 332 | we_should_free = true; |
| 333 | } |
| 334 | else if (rc == UV_ETIMEDOUT || cancelled) { |
| 335 | // timeout |
| 336 | // we will go away and let the callback free the structure |
| 337 | |
| 338 | if(cancelled) |
| 339 | code = rrd_call_function_error(r->result.wb, |
| 340 | "Request cancelled", |
| 341 | HTTP_RESP_CLIENT_CLOSED_REQUEST); |
| 342 | else |
| 343 | code = rrd_call_function_error(r->result.wb, |
| 344 | "Timeout while waiting for a response from the plugin that serves this features", |
| 345 | HTTP_RESP_GATEWAY_TIMEOUT); |
| 346 | |
| 347 | tmp->free_with_signal = true; |
| 348 | we_should_free = false; |
| 349 | } |
| 350 | else { |
| 351 | code = rrd_call_function_error( |
| 352 | r->result.wb, "Internal error while communicating with the plugin that serves this feature.", |
| 353 | HTTP_RESP_INTERNAL_SERVER_ERROR); |
| 354 | |
| 355 | tmp->free_with_signal = true; |
| 356 | we_should_free = false; |
| 357 | } |
| 358 | } |
| 359 | else { |
| 360 | // the response is not ok, and we don't have the data |
| 361 | tmp->free_with_signal = true; |
| 362 | we_should_free = false; |
| 363 | } |
| 364 | |
| 365 | netdata_mutex_unlock(&tmp->mutex); |
| 366 | |
| 367 | if (we_should_free) { |
| 368 | rrd_function_call_wait_free(tmp); |
| 369 | buffer_free(temp_wb); |
| 370 | } |
| 371 | |
| 372 | return code; |
| 373 | } |
| 374 | |
| 375 | static inline int rrd_call_function_async(struct rrd_function_inflight *r, bool wait) { |
| 376 | if(wait) |
| 377 | return rrd_call_function_async_and_wait(r); |
| 378 | else |
| 379 | return rrd_call_function_async_and_dont_wait(r); |
| 380 | } |
| 381 | |
| 382 | |
| 383 | // ---------------------------------------------------------------------------- |
| 384 | |
| 385 | int rrd_function_run(RRDHOST *host, BUFFER *result_wb, int timeout_s, |
| 386 | HTTP_ACCESS user_access, const char *cmd, |
| 387 | bool wait, const char *transaction, |
| 388 | rrd_function_result_callback_t result_cb, void *result_cb_data, |
| 389 | rrd_function_progress_cb_t progress_cb, void *progress_cb_data, |
| 390 | rrd_function_is_cancelled_cb_t is_cancelled_cb, void *is_cancelled_cb_data, |
| 391 | BUFFER *payload, const char *source, bool allow_restricted) { |
| 392 | |
| 393 | int code; |
| 394 | char sanitized_cmd[PLUGINSD_LINE_MAX + 1]; |
| 395 | const DICTIONARY_ITEM *host_function_acquired = NULL; |
| 396 | |
| 397 | char sanitized_source[(source ? strlen(source) : 0) + 1]; |
| 398 | rrd_functions_sanitize(sanitized_source, source ? source : "", sizeof(sanitized_source)); |
| 399 | |
| 400 | // ------------------------------------------------------------------------ |
| 401 | // check for the host |
| 402 | if(!host) { |
| 403 | code = HTTP_RESP_INTERNAL_SERVER_ERROR; |
| 404 | |
| 405 | rrd_call_function_error(result_wb, "No host given for routing this request to.", code); |
| 406 | |
| 407 | if(result_cb) |
| 408 | result_cb(result_wb, code, result_cb_data); |
| 409 | |
| 410 | return code; |
| 411 | } |
| 412 | |
| 413 | // ------------------------------------------------------------------------ |
| 414 | // find the function |
| 415 | |
| 416 | size_t sanitized_cmd_length = rrd_functions_sanitize(sanitized_cmd, cmd, sizeof(sanitized_cmd)); |
| 417 | |
| 418 | code = rrd_functions_find_by_name(host, result_wb, sanitized_cmd, sanitized_cmd_length, &host_function_acquired); |
| 419 | if(code != HTTP_RESP_OK) { |
| 420 | |
| 421 | if(result_cb) |
| 422 | result_cb(result_wb, code, result_cb_data); |
| 423 | |
| 424 | return code; |
| 425 | } |
| 426 | |
| 427 | struct rrd_host_function *rdcf = dictionary_acquired_item_value(host_function_acquired); |
| 428 | |
| 429 | if((rdcf->options & RRD_FUNCTION_RESTRICTED) && !allow_restricted) { |
| 430 | code = rrd_call_function_error(result_wb, |
| 431 | "This feature is not available via this API.", |
| 432 | HTTP_ACCESS_PERMISSION_DENIED_HTTP_CODE(user_access)); |
| 433 | dictionary_acquired_item_release(host->functions, host_function_acquired); |
| 434 | |
| 435 | if(result_cb) |
| 436 | result_cb(result_wb, code, result_cb_data); |
| 437 | |
| 438 | return code; |
| 439 | } |
| 440 | |
| 441 | if(!http_access_user_has_enough_access_level_for_endpoint(user_access, rdcf->access)) { |
| 442 | |
| 443 | if((rdcf->access & HTTP_ACCESS_SIGNED_ID) && !(user_access & HTTP_ACCESS_SIGNED_ID)) |
| 444 | code = rrd_call_function_error(result_wb, |
| 445 | "You need to be authenticated via Netdata Cloud Single-Sign-On (SSO) " |
| 446 | "to access this feature. Sign-in on this dashboard, " |
| 447 | "or access your Netdata via https://app.netdata.cloud.", |
| 448 | HTTP_ACCESS_PERMISSION_DENIED_HTTP_CODE(user_access)); |
| 449 | |
| 450 | else if((rdcf->access & HTTP_ACCESS_SAME_SPACE) && !(user_access & HTTP_ACCESS_SAME_SPACE)) |
| 451 | code = rrd_call_function_error(result_wb, |
| 452 | "You need to login to the Netdata Cloud space this agent is claimed to, " |
| 453 | "to access this feature.", |
| 454 | HTTP_ACCESS_PERMISSION_DENIED_HTTP_CODE(user_access)); |
| 455 | |
| 456 | else if((rdcf->access & HTTP_ACCESS_COMMERCIAL_SPACE) && !(user_access & HTTP_ACCESS_COMMERCIAL_SPACE)) |
| 457 | code = rrd_call_function_error(result_wb, |
| 458 | "This feature is only available for commercial users and supporters " |
| 459 | "of Netdata. To use it, please upgrade your space. " |
| 460 | "Thank you for supporting Netdata.", |
| 461 | HTTP_ACCESS_PERMISSION_DENIED_HTTP_CODE(user_access)); |
| 462 | |
| 463 | else { |
| 464 | HTTP_ACCESS missing_access = (~user_access) & rdcf->access; |
| 465 | char perms_str[1024]; |
| 466 | http_access2txt(perms_str, sizeof(perms_str), ", ", missing_access); |
| 467 | |
| 468 | char msg[2048]; |
| 469 | snprintfz(msg, sizeof(msg), "This feature requires additional permissions: %s.", perms_str); |
| 470 | |
| 471 | code = rrd_call_function_error(result_wb, msg, |
| 472 | HTTP_ACCESS_PERMISSION_DENIED_HTTP_CODE(user_access)); |
| 473 | } |
| 474 | |
| 475 | dictionary_acquired_item_release(host->functions, host_function_acquired); |
| 476 | |
| 477 | if(result_cb) |
| 478 | result_cb(result_wb, code, result_cb_data); |
| 479 | |
| 480 | return code; |
| 481 | } |
| 482 | |
| 483 | if(timeout_s <= 0) |
| 484 | timeout_s = rdcf->timeout; |
| 485 | |
| 486 | // ------------------------------------------------------------------------ |
| 487 | // validate and parse the transaction, or generate a new transaction id |
| 488 | |
| 489 | char uuid_str[UUID_COMPACT_STR_LEN]; |
| 490 | nd_uuid_t uuid; |
| 491 | |
| 492 | if(!transaction || !*transaction || uuid_parse_flexi(transaction, uuid) != 0) |
| 493 | uuid_generate_random(uuid); |
| 494 | |
| 495 | uuid_unparse_lower_compact(uuid, uuid_str); |
| 496 | transaction = uuid_str; |
| 497 | |
| 498 | // ------------------------------------------------------------------------ |
| 499 | // the function can only be executed in async mode |
| 500 | // put the function into the inflight requests |
| 501 | |
| 502 | struct rrd_function_inflight t = { |
| 503 | .used = false, |
| 504 | .host = host, |
| 505 | .cmd = strdupz(cmd), |
| 506 | .sanitized_cmd = strdupz(sanitized_cmd), |
| 507 | .sanitized_cmd_length = sanitized_cmd_length, |
| 508 | .transaction = strdupz(transaction), |
| 509 | .user_access = user_access, |
| 510 | .source = strdupz(sanitized_source), |
| 511 | .payload = buffer_dup(payload), |
| 512 | .timeout = timeout_s, |
| 513 | .cancelled = false, |
| 514 | .stop_monotonic_ut = now_monotonic_usec() + timeout_s * USEC_PER_SEC, |
| 515 | .host_function_acquired = host_function_acquired, |
| 516 | .rdcf = rdcf, |
| 517 | .result = { |
| 518 | .wb = result_wb, |
| 519 | .cb = result_cb, |
| 520 | .data = result_cb_data, |
| 521 | }, |
| 522 | .is_cancelled = { |
| 523 | .cb = is_cancelled_cb, |
| 524 | .data = is_cancelled_cb_data, |
| 525 | }, |
| 526 | .progress = { |
| 527 | .cb = progress_cb, |
| 528 | .data = progress_cb_data, |
| 529 | }, |
| 530 | }; |
| 531 | uuid_copy(t.transaction_uuid, uuid); |
| 532 | |
| 533 | struct rrd_function_inflight *r = dictionary_set(rrd_functions_inflight_requests, transaction, &t, sizeof(t)); |
| 534 | if(!r) { |
| 535 | // dictionary_set() returns NULL when the dictionary is destroyed (shutdown in progress) |
| 536 | code = rrd_call_function_error(result_wb, "Service is shutting down.", HTTP_RESP_SERVICE_UNAVAILABLE); |
| 537 | |
| 538 | rrd_functions_inflight_cleanup(&t); |
| 539 | dictionary_acquired_item_release(host->functions, t.host_function_acquired); |
| 540 | |
| 541 | if(result_cb) |
| 542 | result_cb(result_wb, code, result_cb_data); |
| 543 | |
| 544 | return code; |
| 545 | } |
| 546 | |
| 547 | if(r->used) { |
| 548 | nd_log(NDLS_DAEMON, NDLP_NOTICE, |
| 549 | "FUNCTIONS: duplicate transaction '%s', function: '%s'", |
| 550 | t.transaction, t.cmd); |
| 551 | |
| 552 | code = rrd_call_function_error(result_wb, "Duplicate transaction.", HTTP_RESP_BAD_REQUEST); |
| 553 | |
| 554 | rrd_functions_inflight_cleanup(&t); |
| 555 | dictionary_acquired_item_release(r->host->functions, t.host_function_acquired); |
| 556 | |
| 557 | if(result_cb) |
| 558 | result_cb(result_wb, code, result_cb_data); |
| 559 | |
| 560 | return code; |
| 561 | } |
| 562 | r->used = true; |
| 563 | // internal_error(true, "FUNCTIONS: transaction '%s' started", r->transaction); |
| 564 | |
| 565 | if(r->rdcf->sync) { |
| 566 | // the caller has to wait |
| 567 | |
| 568 | struct rrd_function_execute rfe = { |
| 569 | .transaction = &r->transaction_uuid, |
| 570 | .function = r->sanitized_cmd, |
| 571 | .payload = r->payload, |
| 572 | .user_access = r->user_access, |
| 573 | .source = r->source, |
| 574 | .stop_monotonic_ut = &r->stop_monotonic_ut, |
| 575 | .result = { |
| 576 | .wb = r->result.wb, |
| 577 | |
| 578 | // we overwrite the result callbacks, |
| 579 | // so that we can clean up the allocations made |
| 580 | .cb = r->result.cb, |
| 581 | .data = r->result.data, |
| 582 | }, |
| 583 | .progress = { |
| 584 | .cb = r->progress.cb, |
| 585 | .data = r->progress.data, |
| 586 | }, |
| 587 | .is_cancelled = { |
| 588 | .cb = r->is_cancelled.cb, |
| 589 | .data = r->is_cancelled.data, |
| 590 | }, |
| 591 | .register_canceller = { |
| 592 | .cb = NULL, |
| 593 | .data = NULL, |
| 594 | }, |
| 595 | .register_progresser = { |
| 596 | .cb = NULL, |
| 597 | .data = NULL, |
| 598 | }, |
| 599 | }; |
| 600 | code = r->rdcf->execute_cb(&rfe, r->rdcf->execute_cb_data); |
| 601 | |
| 602 | rrd_inflight_function_cleanup(host, r->transaction); |
| 603 | return code; |
| 604 | } |
| 605 | |
| 606 | return rrd_call_function_async(r, wait); |
| 607 | } |
| 608 | |
| 609 | bool rrd_function_has_this_original_result_callback(nd_uuid_t *transaction, rrd_function_result_callback_t cb) { |
| 610 | bool ret = false; |
| 611 | char str[UUID_COMPACT_STR_LEN]; |
| 612 | uuid_unparse_lower_compact(*transaction, str); |
| 613 | const DICTIONARY_ITEM *item = dictionary_get_and_acquire_item(rrd_functions_inflight_requests, str); |
| 614 | if(item) { |
| 615 | struct rrd_function_inflight *r = dictionary_acquired_item_value(item); |
| 616 | if(r->result.cb == cb) |
| 617 | ret = true; |
| 618 | |
| 619 | dictionary_acquired_item_release(rrd_functions_inflight_requests, item); |
| 620 | } |
| 621 | return ret; |
| 622 | } |
| 623 | |
| 624 | static void rrd_function_cancel_inflight(struct rrd_function_inflight *r) { |
| 625 | if(!r) |
| 626 | return; |
| 627 | |
| 628 | bool cancelled = __atomic_load_n(&r->cancelled, __ATOMIC_RELAXED); |
| 629 | if(cancelled) { |
| 630 | nd_log(NDLS_DAEMON, NDLP_DEBUG, |
| 631 | "FUNCTIONS: received a CANCEL request for transaction '%s', but it is already cancelled.", |
| 632 | r->transaction); |
| 633 | return; |
| 634 | } |
| 635 | |
| 636 | __atomic_store_n(&r->cancelled, true, __ATOMIC_RELAXED); |
| 637 | |
| 638 | if(!rrd_collector_dispatcher_acquire(r->rdcf->collector)) { |
| 639 | nd_log(NDLS_DAEMON, NDLP_DEBUG, |
| 640 | "FUNCTIONS: received a CANCEL request for transaction '%s', but the collector is not running.", |
| 641 | r->transaction); |
| 642 | return; |
| 643 | } |
| 644 | |
| 645 | if(r->canceller.cb) |
| 646 | r->canceller.cb(r->canceller.data); |
| 647 | |
| 648 | rrd_collector_dispatcher_release(r->rdcf->collector); |
| 649 | } |
| 650 | |
| 651 | void rrd_function_cancel(const char *transaction) { |
| 652 | // internal_error(true, "FUNCTIONS: request to cancel transaction '%s'", transaction); |
| 653 | |
| 654 | const DICTIONARY_ITEM *item = dictionary_get_and_acquire_item(rrd_functions_inflight_requests, transaction); |
| 655 | if(!item) { |
| 656 | nd_log(NDLS_DAEMON, NDLP_DEBUG, |
| 657 | "FUNCTIONS: received a CANCEL request for transaction '%s', but the transaction is not running.", |
| 658 | transaction); |
| 659 | return; |
| 660 | } |
| 661 | |
| 662 | struct rrd_function_inflight *r = dictionary_acquired_item_value(item); |
| 663 | rrd_function_cancel_inflight(r); |
| 664 | dictionary_acquired_item_release(rrd_functions_inflight_requests, item); |
| 665 | } |
| 666 | |
| 667 | void rrd_function_progress(const char *transaction) { |
| 668 | const DICTIONARY_ITEM *item = dictionary_get_and_acquire_item(rrd_functions_inflight_requests, transaction); |
| 669 | if(!item) { |
| 670 | nd_log(NDLS_DAEMON, NDLP_DEBUG, |
| 671 | "FUNCTIONS: received a PROGRESS request for transaction '%s', but the transaction is not running.", |
| 672 | transaction); |
| 673 | return; |
| 674 | } |
| 675 | |
| 676 | struct rrd_function_inflight *r = dictionary_acquired_item_value(item); |
| 677 | |
| 678 | if(!rrd_collector_dispatcher_acquire(r->rdcf->collector)) { |
| 679 | nd_log(NDLS_DAEMON, NDLP_DEBUG, |
| 680 | "FUNCTIONS: received a PROGRESS request for transaction '%s', but the collector is not running.", |
| 681 | transaction); |
| 682 | goto cleanup; |
| 683 | } |
| 684 | |
| 685 | functions_stop_monotonic_update_on_progress(&r->stop_monotonic_ut); |
| 686 | |
| 687 | if(r->progresser.cb) |
| 688 | r->progresser.cb(transaction, r->progresser.data); |
| 689 | |
| 690 | rrd_collector_dispatcher_release(r->rdcf->collector); |
| 691 | |
| 692 | cleanup: |
| 693 | dictionary_acquired_item_release(rrd_functions_inflight_requests, item); |
| 694 | } |
| 695 | |
| 696 | void rrd_function_call_progresser(nd_uuid_t *transaction) { |
| 697 | if(uuid_is_null(*transaction)) |
| 698 | return; |
| 699 | |
| 700 | char str[UUID_COMPACT_STR_LEN]; |
| 701 | uuid_unparse_lower_compact(*transaction, str); |
| 702 | rrd_function_progress(str); |
| 703 | } |