From 0e59c3a0fec5e7a026daefc0d6f4536285b4d522 Mon Sep 17 00:00:00 2001 From: Mansoor Khan M Date: Sat, 18 Jul 2026 23:18:03 +0530 Subject: [PATCH 1/4] recording-daemon: lifecycle notification events Extend rtpengine-recording HTTP/command notifications beyond the historical finished-only path so operators can track call and stream recording lifecycle. Events (notify-events CSV, default: finished): opened, started, finished, discarded, failed, call-started, call-finished, call-discarded New options: notify-events, notify-json, notify-no-metadata, notify-command-format, notify-queue-limit. Docs/conf only document the new notify options; upstream storage (S3/GCS/output-storage) docs are left unchanged. Backward compatible: default mask is finished only. Includes unit tests and Docker e2e smoke harness. --- docs/rtpengine-recording.md | 97 ++++- etc/rtpengine-recording.conf | 7 +- recording-daemon/Makefile | 2 +- recording-daemon/main.c | 29 ++ recording-daemon/main.h | 5 + recording-daemon/metafile.c | 9 + recording-daemon/notify.c | 705 +++++++++++++++++++++++++------ recording-daemon/notify.h | 8 +- recording-daemon/notify_events.c | 120 ++++++ recording-daemon/notify_events.h | 58 +++ recording-daemon/output.c | 41 +- recording-daemon/packet.c | 5 + recording-daemon/types.h | 37 ++ t/Makefile | 4 +- t/notify_e2e/notify_receiver.py | 71 ++++ t/notify_e2e/run_e2e.sh | 176 ++++++++ t/test-notify-events.c | 156 +++++++ 17 files changed, 1390 insertions(+), 140 deletions(-) create mode 100644 recording-daemon/notify_events.c create mode 100644 recording-daemon/notify_events.h create mode 100644 t/notify_e2e/notify_receiver.py create mode 100644 t/notify_e2e/run_e2e.sh create mode 100644 t/test-notify-events.c diff --git a/docs/rtpengine-recording.md b/docs/rtpengine-recording.md index 7bf898a979..9b135cf2b5 100644 --- a/docs/rtpengine-recording.md +++ b/docs/rtpengine-recording.md @@ -367,14 +367,79 @@ sufficient for a standard installation of rtpengine. - __\-\-notify-uri=__*URI* - Enable HTTP notification about finished recordings to the specified URI, which - must be an HTTP or HTTPS URI. Information about the finished recording is - provided via custom HTTP headers, all of which use a prefix of __X-Recording-__. + Enable HTTP notification about recording lifecycle events to the specified URI, + which must be an HTTP or HTTPS URI. Information is provided via custom HTTP + headers, all of which use a prefix of __X-Recording-__. + + By default only the historical __finished__ event is emitted (see + __\-\-notify-events__). When additional events are enabled, the same URI + receives notifications for file open, first write, successful finish, + discard, and failure. + +- __\-\-notify-events=__*CSV* + + Comma-separated list of lifecycle events to emit. Default is __finished__ + for backward compatibility with existing consumers. + + Supported tokens: + + - __opened__ - after output media file is created and header written + (__recording_file_opened__). + - __started__ - first media packet successfully written + (__recording_started__). + - __finished__ - successful close of a non-empty recording + (__recording_finished__; historical default). + - __discarded__ - closed empty or with discard flag + (__recording_discarded__). + - __failed__ - open/configure failure (__recording_failed__). + - __call-started__ - first mix/single output armed for a call + (__call_recording_started__). + - __call-finished__ - call metafile torn down on success path + (__call_recording_finished__). + - __call-discarded__ - call torn down with discard flag + (__call_recording_discarded__). + - __all__ - enable every token above. + + Additional headers on lifecycle notifications: + + - __X-Recording-Event__ - event name (e.g. recording_file_opened) + - __X-Recording-Status__ - short status (opened, started, finished, ...) + - __X-Recording-Event-Time__ - unix epoch (float seconds) + - __X-Recording-Output-Id__ - correlates open/start/finish for one stream + + Terminal extras when available (finished / discarded / failed): + __X-Recording-Duration-MS__, __X-Recording-File-Size__, + __X-Recording-Sample-Rate__, __X-Recording-Channels__, + __X-Recording-Call-End-Time__, __X-Recording-Stream-End-Time__. + Failure events may also set __X-Recording-Error-Code__ / + __X-Recording-Error-Message__. + + Ordering is best-effort asynchronous: opened -> started -> terminal. + Call-level ordering is call-started before stream events and + call-finished / call-discarded after outputs close. Delivery uses the + existing notify thread pool with retries; notify failures never abort + media I/O. Non-terminal queue bound (__\-\-notify-queue-limit__) may drop + excess non-terminal events under load; terminal events are never dropped + by that limit. __\-\-notify-record__ / __output-storage=notify__ and + __\-\-notify-purge__ apply only to finished events. + +- __\-\-notify-json__ + + Send a JSON body with HTTP notifications and force POST. Headers remain + for compatibility. JSON fields include event, event_time, status, + call_id, kind, file names, output_id, db, times, media, tag, metadata, + and error. + +- __\-\-notify-no-metadata__ + + Omit call metadata strings from headers and JSON (default: include + metadata). - __\-\-notify-post__ Use HTTP POST instead of GET for the HTTP notification requests. The request - body is empty even if POST is used. + body is empty unless __\-\-notify-json__ is also set (or the recording is + attached via __output-storage=notify__ / __\-\-notify-record__ on finished). - __\-\-notify-no-verify__ @@ -382,9 +447,21 @@ sufficient for a standard installation of rtpengine. - __\-\-notify-command=__*PATH* - External command to launch to send a notification about a new recording. - The command will receive two command line arguments: The full path and file - name of the recording, and the ID number from the database entry. + External command to launch for notifications. Argument layout depends on + __\-\-notify-command-format__. + +- __\-\-notify-command-format=__*legacy|extended|json-env* + + How arguments are passed to __\-\-notify-command__: + + - __legacy__ (default): cmd path db_id + - __extended__: cmd event path db_id call_id kind full_filename + - __json-env__: cmd with NOTIFY_PAYLOAD (JSON) and NOTIFY_EVENT env vars + +- __\-\-notify-queue-limit=__*INT* + + Maximum number of in-flight non-terminal notifications. 0 means unlimited. + Default is 1000. Terminal notifications are always accepted. - __\-\-notify-concurrency=__*INT* @@ -400,12 +477,14 @@ sufficient for a standard installation of rtpengine. Legacy alias for __output-storage=notify__. If no other output storage is enabled, then the default __file__ storage remains enabled (unless the - following option is also set). + following option is also set). File attach applies only to finished events + and is not combined with a JSON body. - __\-\-notify-purge__ Legacy option to disable the default file storage when notification output - is enabled. + is enabled. When used with file attach behaviour, local file removal after + a successful notification applies only to finished events. - __\-\-output-mixed-per-media__ diff --git a/etc/rtpengine-recording.conf b/etc/rtpengine-recording.conf index f240675731..e8a93471fe 100644 --- a/etc/rtpengine-recording.conf +++ b/etc/rtpengine-recording.conf @@ -61,13 +61,18 @@ table = 0 # output-chown = rtpengine # output-chgrp = rtpengine -### HTTP notifications for finished recordings +### HTTP notifications for recording lifecycle events # notify-uri = https://example.com/rec/finished # notify-post = false # notify-no-verify = false # notify-purge = false # notify-concurrency = 5 # notify-retries = 10 +# notify-events = finished +# notify-json = false +# notify-no-metadata = false +# notify-command-format = legacy +# notify-queue-limit = 1000 ### S3 storage options # s3-host = examplebucket.s3.amazonaws.com diff --git a/recording-daemon/Makefile b/recording-daemon/Makefile index 579bcf783b..8bd5f033e8 100644 --- a/recording-daemon/Makefile +++ b/recording-daemon/Makefile @@ -39,7 +39,7 @@ CFLAGS += $(CFLAGS_BCG729) LDLIBS += $(LDLIBS_BCG729) SRCS := epoll.c garbage.c inotify.c main.c metafile.c stream.c recaux.c packet.c \ - decoder.c output.c mix.c db.c log.c forward.c tag.c custom_poller.c notify.c tls_send.c s3.c \ + decoder.c output.c mix.c db.c log.c forward.c tag.c custom_poller.c notify.c notify_events.c tls_send.c s3.c \ gcs.c LIBSRCS := loglib.c auxlib.c rtplib.c codeclib.c resample.c str.c socket.c streambuf.c ssllib.c \ dtmflib.c bufferpool.c bencode.c http.c s3utils.c oauth.c cchain.c diff --git a/recording-daemon/main.c b/recording-daemon/main.c index d0a6d4784a..a5a14bb3e0 100644 --- a/recording-daemon/main.c +++ b/recording-daemon/main.c @@ -30,6 +30,7 @@ #include "socket.h" #include "ssllib.h" #include "notify.h" +#include "notify_events.h" #include "gcs.h" @@ -68,6 +69,11 @@ gboolean notify_nverify; int notify_threads = 5; int notify_retries = 10; char *notify_command; +unsigned int notify_events_mask = NOTIFY_MASK_DEFAULT; +gboolean notify_json = FALSE; +gboolean notify_no_metadata = FALSE; +int notify_command_format = 0; /* NOTIFY_CMD_LEGACY */ +int notify_queue_limit = 1000; gboolean mix_output_per_media = 0; gboolean flush_packets = 0; int resample_audio; @@ -213,6 +219,8 @@ static void options(int *argc, char ***argv) { gboolean notify_record = FALSE; bool no_output_allowed = false; gboolean notify_purge = false; + g_autoptr(char) notify_events_str = NULL; + g_autoptr(char) notify_cmd_fmt_str = NULL; GOptionEntry e[] = { { "table", 't', 0, G_OPTION_ARG_INT, &ktable, "Kernel table rtpengine uses", "INT" }, @@ -251,6 +259,11 @@ static void options(int *argc, char ***argv) { { "notify-concurrency", 0, 0, G_OPTION_ARG_INT, ¬ify_threads,"How many simultaneous requests", "INT" }, { "notify-retries", 0, 0, G_OPTION_ARG_INT, ¬ify_retries,"How many times to retry failed requests","INT" }, { "notify-command", 0, 0, G_OPTION_ARG_STRING, ¬ify_command,"External command to execute for notifications","PATH" }, + { "notify-events", 0, 0, G_OPTION_ARG_STRING, ¬ify_events_str,"CSV of lifecycle events to emit","opened,started,finished,..." }, + { "notify-json", 0, 0, G_OPTION_ARG_NONE, ¬ify_json,"Send JSON body with HTTP notifications", NULL }, + { "notify-no-metadata", 0, 0, G_OPTION_ARG_NONE, ¬ify_no_metadata,"Omit call metadata from notify payloads", NULL }, + { "notify-command-format",0, 0, G_OPTION_ARG_STRING, ¬ify_cmd_fmt_str,"Command argv format","legacy|extended|json-env" }, + { "notify-queue-limit", 0, 0, G_OPTION_ARG_INT, ¬ify_queue_limit,"Max queued non-terminal notifies (0=unlimited)","INT" }, { "output-mixed-per-media",0,0, G_OPTION_ARG_NONE, &mix_output_per_media,"Mix participating sources into a single output", NULL }, #if CURL_AT_LEAST_VERSION(7,56,0) { "notify-record", 0, 0, G_OPTION_ARG_NONE, ¬ify_record, "Also attach recorded file to request", NULL }, @@ -276,6 +289,22 @@ static void options(int *argc, char ***argv) { config_load(argc, argv, e, " - rtpengine recording daemon", "/etc/rtpengine/rtpengine-recording.conf", "rtpengine-recording", &rtpe_common_config); + /* notify-events CSV (default finished) */ + { + char *bad = NULL; + if (!notify_events_parse(notify_events_str, ¬ify_events_mask, &bad)) + die("Invalid notify-events token '%s'", bad ? bad : "?"); + g_free(bad); + } + { + enum notify_command_format fmt = NOTIFY_CMD_LEGACY; + if (!notify_command_format_parse(notify_cmd_fmt_str, &fmt)) + die("Invalid notify-command-format (use legacy|extended|json-env)"); + notify_command_format = (int) fmt; + } + if (notify_queue_limit < 0) + die("notify-queue-limit must be >= 0"); + // default config, if not configured if (spool_dir == NULL) spool_dir = g_strdup("/var/spool/rtpengine"); diff --git a/recording-daemon/main.h b/recording-daemon/main.h index 22b79236b5..1c3f718b1b 100644 --- a/recording-daemon/main.h +++ b/recording-daemon/main.h @@ -56,6 +56,11 @@ extern gboolean notify_nverify; extern int notify_threads; extern int notify_retries; extern char *notify_command; +extern unsigned int notify_events_mask; +extern gboolean notify_json; +extern gboolean notify_no_metadata; +extern int notify_command_format; +extern int notify_queue_limit; extern gboolean mix_output_per_media; extern volatile int shutdown_flag; extern gboolean flush_packets; diff --git a/recording-daemon/metafile.c b/recording-daemon/metafile.c index a0ffce798d..5dc8c9dea6 100644 --- a/recording-daemon/metafile.c +++ b/recording-daemon/metafile.c @@ -13,6 +13,7 @@ #include "recaux.h" #include "packet.h" #include "output.h" +#include "notify.h" #include "mix.h" #include "db.h" #include "forward.h" @@ -89,6 +90,9 @@ static void meta_destroy(metafile_t *mf) { output_close(mf, mf->mix_out, NULL, mf->discard); mf->mix_out = NULL; mix_close(mf->mix); + + /* Call-level terminal notify after stream/mix outputs close. */ + notify_push_call(mf); } @@ -112,6 +116,11 @@ static void meta_mix_file_output(metafile_t *mf) { } db_do_stream(mf, mf->mix_out, NULL, 0); + + if (mf->mix_out && !mf->notify_call_started) { + mf->notify_call_started = 1; + notify_push_call_event(NOTIFY_EVT_CALL_STARTED, mf); + } } diff --git a/recording-daemon/notify.c b/recording-daemon/notify.c index 17ce82f6d1..c4c193285b 100644 --- a/recording-daemon/notify.c +++ b/recording-daemon/notify.c @@ -1,11 +1,18 @@ #include "notify.h" #include +#include +#include #include +#include +#include #include "main.h" #include "log_r.h" #include "recaux.h" #include "output.h" #include "http.h" +#include "helpers.h" +#include "str.h" +#include "notify_events.h" static GThreadPool *notify_threadpool; @@ -15,12 +22,65 @@ static pthread_cond_t timer_cond = PTHREAD_COND_INITIALIZER; static pthread_t notify_waiter; static GTree *notify_timers; +static unsigned int notify_nont_inflight; +static pthread_mutex_t queue_lock = PTHREAD_MUTEX_INITIALIZER; + +static unsigned long notify_metric_enqueued; +static unsigned long notify_metric_success; +static unsigned long notify_metric_retry; +static unsigned long notify_metric_giveup; +static unsigned long notify_metric_dropped; + + +static void metric_inc(unsigned long *m) { + __atomic_add_fetch(m, 1, __ATOMIC_RELAXED); +} + +static void nont_inflight_dec(notif_req_t *req) { + if (!req || req->terminal) + return; + pthread_mutex_lock(&queue_lock); + if (notify_nont_inflight > 0) + notify_nont_inflight--; + pthread_mutex_unlock(&queue_lock); +} + +static bool nont_inflight_try_inc(void) { + bool ok = true; + pthread_mutex_lock(&queue_lock); + if (notify_queue_limit > 0 && notify_nont_inflight >= (unsigned int) notify_queue_limit) + ok = false; + else + notify_nont_inflight++; + pthread_mutex_unlock(&queue_lock); + return ok; +} + +static void req_free_fields(notif_req_t *req) { + if (!req) + return; + g_free(req->call_id); + g_free(req->file_name); + g_free(req->file_format); + g_free(req->kind); + g_free(req->full_filename); + g_free(req->output_id); + g_free(req->metadata); + g_free(req->tag_name); + g_free(req->tag_label); + g_free(req->tag_metadata); + g_free(req->error_code); + g_free(req->error_message); + g_free(req->json_body); + g_free(req->object_name); +} static bool do_notify_http(notif_req_t *req) { const char *err = NULL; - CURLcode ret; + CURLcode ret = CURLE_OK; - ilog(LOG_DEBUG, "Launching HTTP notification for '%s%s%s'", FMT_M(req->name)); + ilog(LOG_DEBUG, "Launching HTTP notification (%s) for '%s%s%s'", + notify_event_name(req->event), FMT_M(req->name)); #if CURL_AT_LEAST_VERSION(7,56,0) g_autoptr(curl_mime) mime = NULL; @@ -32,26 +92,33 @@ static bool do_notify_http(notif_req_t *req) { if (!c) goto fail; - /* POST vs GET */ - if (notify_post) { + bool do_post = notify_post || req->json_body || req->content; + if (do_post) { err = "setting CURLOPT_POST"; if ((ret = curl_easy_setopt(c, CURLOPT_POST, 1L)) != CURLE_OK) goto fail; } + if (req->json_body) { + err = "setting CURLOPT_POSTFIELDS"; + if ((ret = curl_easy_setopt(c, CURLOPT_POSTFIELDS, req->json_body)) != CURLE_OK) + goto fail; + err = "setting CURLOPT_POSTFIELDSIZE"; + if ((ret = curl_easy_setopt(c, CURLOPT_POSTFIELDSIZE, + (long) strlen(req->json_body))) != CURLE_OK) + goto fail; + } + #if CURL_AT_LEAST_VERSION(7,56,0) - if (req->content) { + if (req->content && req->event == NOTIFY_EVT_FINISHED && !req->json_body) { err = "initializing curl mime&part"; curl_mimepart *part; mime = curl_mime_init(c); part = curl_mime_addpart(mime); - if ((ret = curl_mime_name(part, "ngfile")) != CURLE_OK) goto fail; - if ((ret = curl_mime_data(part, req->content->s->str, req->content->s->len)) != CURLE_OK) goto fail; - if ((ret = curl_easy_setopt(c, CURLOPT_MIMEPOST, mime)) != CURLE_OK) goto fail; } @@ -70,18 +137,15 @@ static bool do_notify_http(notif_req_t *req) { if (code < 200 || code >= 300) goto fail; - /* success */ - - ilog(LOG_NOTICE, "HTTP notification for '%s%s%s' was successful", FMT_M(req->name)); - + ilog(LOG_NOTICE, "HTTP notification (%s) for '%s%s%s' was successful", + notify_event_name(req->event), FMT_M(req->name)); return true; fail: - ilog(LOG_ERR, "Failed to perform HTTP notification for '%s%s%s': " + ilog(LOG_ERR, "Failed to perform HTTP notification (%s) for '%s%s%s': " "Error while %s: %s", - FMT_M(req->name), + notify_event_name(req->event), FMT_M(req->name), err, curl_easy_strerror(ret)); - return false; } @@ -91,186 +155,230 @@ static void failed_http(notif_req_t *req) { } static bool do_notify_command(notif_req_t *req) { - ilog(LOG_DEBUG, "Executing notification command for '%s%s%s'", FMT_M(req->name)); - + if (!req->argv) + return true; + ilog(LOG_DEBUG, "Executing notification command (%s) for '%s%s%s'", + notify_event_name(req->event), FMT_M(req->name)); GError *err = NULL; - bool success = g_spawn_sync(NULL, req->argv, NULL, + bool success = g_spawn_sync(NULL, req->argv, req->envp, G_SPAWN_SEARCH_PATH | G_SPAWN_STDOUT_TO_DEV_NULL | G_SPAWN_STDERR_TO_DEV_NULL, NULL, NULL, NULL, NULL, NULL, &err); - if (!success) { - ilog(LOG_ERR, "Failed to execute notification command for '%s%s%s': %s", - FMT_M(req->name), err->message); + ilog(LOG_ERR, "Failed to execute notification command (%s) for '%s%s%s': %s", + notify_event_name(req->event), FMT_M(req->name), err->message); g_error_free(err); } - return success; } static void do_notify(void *p, void *u) { notif_req_t *req = p; - bool ok = req->action->perform(req); if (!ok) { - if (notify_retries >= 0 && req->retries < notify_retries) { - /* schedule retry */ + if (notify_retries >= 0 && req->retries < (unsigned int) notify_retries) { req->retries++; - - ilog(LOG_INFO, "Failure while sending notification for '%s%s%s': " + metric_inc(¬ify_metric_retry); + ilog(LOG_INFO, "Failure while sending notification (%s) for '%s%s%s': " "Will retry in %" PRId64 " seconds (#%u)", - FMT_M(req->name), + notify_event_name(req->event), FMT_M(req->name), req->falloff_us / 1000000L, req->retries); - req->retry_time = now_us() + req->falloff_us; req->falloff_us *= 2; - pthread_mutex_lock(&timer_lock); - g_tree_insert(notify_timers, req, req); - pthread_cond_signal(&timer_cond); + if (notify_timers) { + g_tree_insert(notify_timers, req, req); + pthread_cond_signal(&timer_cond); + pthread_mutex_unlock(&timer_lock); + return; + } pthread_mutex_unlock(&timer_lock); - - return; // skip cleanup } - - ilog(LOG_ERR, "Failure while sending notification for '%s%s%s' after %u retries. " - "Giving up", - FMT_M(req->name), - req->retries); - + metric_inc(¬ify_metric_giveup); + ilog(LOG_ERR, "Failure while sending notification (%s) for '%s%s%s' after %u retries. " + "Giving up", notify_event_name(req->event), FMT_M(req->name), req->retries); if (req->action->failed) req->action->failed(req); } + else + metric_inc(¬ify_metric_success); + nont_inflight_dec(req); req->action->cleanup(req); + req_free_fields(req); g_free(req->name); g_free(req); } - static void *notify_timer(void *p) { pthread_mutex_lock(&timer_lock); - - // notify_timers being NULL acts as our shutdown flag while (notify_timers) { - ilog(LOG_DEBUG, "Notification timer thread looping"); - - // grab first entry in list, check retry time, sleep if it's in the future - notif_req_t *first = rtpe_g_tree_first(notify_timers); if (!first) { - ilog(LOG_DEBUG, "No scheduled notification retries, sleeping"); pthread_cond_wait(&timer_cond, &timer_lock); continue; } int64_t now = now_us(); if (now < first->retry_time) { - ilog(LOG_DEBUG, "Sleeping until next scheduled notification retry in %" PRId64 " seconds", - (first->retry_time - now) / 1000000L); cond_timedwait(&timer_cond, &timer_lock, first->retry_time); continue; } - - // first entry is ready to run - g_tree_remove(notify_timers, first); - ilog(LOG_DEBUG, "Notification retry for '%s%s%s' is scheduled now", FMT_M(first->name)); g_thread_pool_push(notify_threadpool, first, NULL); } - - // clean up - pthread_mutex_unlock(&timer_lock); pthread_mutex_destroy(&timer_lock); pthread_cond_destroy(&timer_cond); - return NULL; } - static int notify_req_cmp(const void *A, const void *B) { const notif_req_t *a = A, *b = B; - - if (a->retry_time < b->retry_time) - return -1; - if (a->retry_time > b->retry_time) - return 1; - if (a < b) - return -1; - if (a > b) - return 1; + if (a->retry_time < b->retry_time) return -1; + if (a->retry_time > b->retry_time) return 1; + if (a < b) return -1; + if (a > b) return 1; return 0; } - void notify_setup(void) { - if (notify_threads <= 0) + if ((!notify_uri && !notify_command) || notify_threads <= 0) return; - notify_threadpool = g_thread_pool_new(do_notify, NULL, notify_threads, false, NULL); - notify_timers = g_tree_new(notify_req_cmp); int ret = pthread_create(¬ify_waiter, NULL, notify_timer, NULL); if (ret) ilog(LOG_ERR, "Failed to launch thread for HTTP notification"); + ilog(LOG_INFO, "Recording notify enabled (events mask=0x%x json=%s cmd_fmt=%d queue_limit=%d)", + notify_events_mask, notify_json ? "true" : "false", + notify_command_format, notify_queue_limit); } void notify_cleanup(void) { if (notify_waiter && notify_timers) { - // get lock, free GTree, signal thread to shut down pthread_mutex_lock(&timer_lock); - g_tree_destroy(notify_timers); - notify_timers = NULL; + if (notify_timers) { + while (true) { + notif_req_t *first = rtpe_g_tree_first(notify_timers); + if (!first) + break; + g_tree_remove(notify_timers, first); + nont_inflight_dec(first); + if (first->action && first->action->cleanup) + first->action->cleanup(first); + req_free_fields(first); + g_free(first->name); + g_free(first); + } + g_tree_destroy(notify_timers); + notify_timers = NULL; + } pthread_cond_signal(&timer_cond); pthread_mutex_unlock(&timer_lock); } if (notify_threadpool) g_thread_pool_free(notify_threadpool, true, false); notify_threadpool = NULL; + ilog(LOG_INFO, "Notify metrics: enqueued=%lu success=%lu retry=%lu giveup=%lu dropped=%lu", + __atomic_load_n(¬ify_metric_enqueued, __ATOMIC_RELAXED), + __atomic_load_n(¬ify_metric_success, __ATOMIC_RELAXED), + __atomic_load_n(¬ify_metric_retry, __ATOMIC_RELAXED), + __atomic_load_n(¬ify_metric_giveup, __ATOMIC_RELAXED), + __atomic_load_n(¬ify_metric_dropped, __ATOMIC_RELAXED)); } - - #define notify_add_header(req, f, ...) http_add_header(&(req)->headers, f, __VA_ARGS__) - static void notify_req_setup_http(notif_req_t *req, output_t *o, metafile_t *mf, tag_t *tag) { double now = (double) now_us() / 1000000.; + bool terminal = notify_event_is_terminal(req->event); + + const char *call_id = req->call_id ? req->call_id : (mf && mf->call_id ? mf->call_id : NULL); + const char *file_name = req->file_name ? req->file_name : (o && o->file_name ? o->file_name : NULL); + const char *file_format = req->file_format ? req->file_format : (o && o->file_format ? o->file_format : NULL); + const char *kind = req->kind ? req->kind : (o && o->kind ? o->kind : NULL); + const char *full_filename = req->full_filename ? req->full_filename : + (o && o->full_filename ? o->full_filename : NULL); + const char *output_id = req->output_id ? req->output_id : (o && o->output_id ? o->output_id : NULL); + const char *metadata = req->metadata ? req->metadata : + (!notify_no_metadata && mf && mf->metadata ? mf->metadata : NULL); + double call_start = req->call_start > 0 ? req->call_start : + (mf ? (double) mf->start_time_us / 1000000. : 0); + double stream_start = req->stream_start > 0 ? req->stream_start : + (o ? (double) o->start_time_us / 1000000. : call_start); + unsigned long long call_db_id = req->call_db_id ? req->call_db_id : (mf ? mf->db_id : 0); + unsigned long long stream_db_id = req->db_id ? req->db_id : (o ? o->db_id : 0); + + notify_add_header(req, "X-Recording-Event: %s", notify_event_name(req->event)); + notify_add_header(req, "X-Recording-Status: %s", notify_event_status(req->event)); + notify_add_header(req, "X-Recording-Event-Time: %.06f", now); + + if (call_id) + notify_add_header(req, "X-Recording-Call-ID: %s", call_id); + if (output_id) + notify_add_header(req, "X-Recording-Output-Id: %s", output_id); + if (file_name && file_format) + notify_add_header(req, "X-Recording-File-Name: %s.%s", file_name, file_format); + else if (file_name) + notify_add_header(req, "X-Recording-File-Name: %s", file_name); + if (full_filename) + notify_add_header(req, "X-Recording-Full-File-Name: %s", full_filename); + if (file_format) + notify_add_header(req, "X-Recording-File-Format: %s", file_format); + if (kind) + notify_add_header(req, "X-Recording-Kind: %s", kind); + + notify_add_header(req, "X-Recording-Call-Start-Time: %.06f", call_start); + notify_add_header(req, "X-Recording-Stream-Start-Time: %.06f", stream_start); + if (terminal) { + notify_add_header(req, "X-Recording-Call-End-Time: %.06f", now); + notify_add_header(req, "X-Recording-Stream-End-Time: %.06f", now); + } - notify_add_header(req, "X-Recording-Call-ID: %s", mf->call_id); - notify_add_header(req, "X-Recording-File-Name: %s.%s", o->file_name, o->file_format); - notify_add_header(req, "X-Recording-Full-File-Name: %s.%s", o->full_filename, o->file_format); - notify_add_header(req, "X-Recording-File-Format: %s", o->file_format); - notify_add_header(req, "X-Recording-Kind: %s", o->kind); - notify_add_header(req, "X-Recording-Call-Start-Time: %.06f", (double) mf->start_time_us / 1000000.); - notify_add_header(req, "X-Recording-Stream-Start-Time: %.06f", (double) o->start_time_us / 1000000.); - notify_add_header(req, "X-Recording-Call-End-Time: %.06f", now); - notify_add_header(req, "X-Recording-Stream-End-Time: %.06f", now); - - if (mf->db_id) - notify_add_header(req, "X-Recording-Call-DB-ID: %llu", mf->db_id); - if (o->db_id) - notify_add_header(req, "X-Recording-Stream-DB-ID: %llu", o->db_id); - if (mf->metadata) - notify_add_header(req, "X-Recording-Call-Metadata: %s", mf->metadata); - if (mf->metadata) - notify_add_header(req, "X-Recording-DB-Metadata: %s", mf->metadata); - - if (tag) { - notify_add_header(req, "X-Recording-Tag: %s", tag->name); - if (tag->label) - notify_add_header(req, "X-Recording-Label: %s", tag->label); - if (tag->metadata) - notify_add_header(req, "X-Recording-Tag-Metadata: %s", tag->metadata); + if (call_db_id) + notify_add_header(req, "X-Recording-Call-DB-ID: %llu", call_db_id); + if (stream_db_id) + notify_add_header(req, "X-Recording-Stream-DB-ID: %llu", stream_db_id); + if (metadata) { + notify_add_header(req, "X-Recording-Call-Metadata: %s", metadata); + notify_add_header(req, "X-Recording-DB-Metadata: %s", metadata); } - if ((output_storage & OUTPUT_STORAGE_NOTIFY)) + const char *tag_name = req->tag_name ? req->tag_name : (tag && tag->name ? tag->name : NULL); + const char *tag_label = req->tag_label ? req->tag_label : (tag && tag->label ? tag->label : NULL); + const char *tag_metadata = req->tag_metadata ? req->tag_metadata : + (tag && tag->metadata ? tag->metadata : NULL); + if (tag_name) + notify_add_header(req, "X-Recording-Tag: %s", tag_name); + if (tag_label) + notify_add_header(req, "X-Recording-Label: %s", tag_label); + if (tag_metadata) + notify_add_header(req, "X-Recording-Tag-Metadata: %s", tag_metadata); + + if (req->error_code && req->error_code[0]) + notify_add_header(req, "X-Recording-Error-Code: %s", req->error_code); + if (req->error_message && req->error_message[0]) + notify_add_header(req, "X-Recording-Error-Message: %s", req->error_message); + if (req->file_size > 0) + notify_add_header(req, "X-Recording-File-Size: %" PRId64, req->file_size); + if (req->duration_ms > 0) + notify_add_header(req, "X-Recording-Duration-MS: %" PRId64, req->duration_ms); + if (req->sample_rate > 0) + notify_add_header(req, "X-Recording-Sample-Rate: %d", req->sample_rate); + if (req->channels > 0) + notify_add_header(req, "X-Recording-Channels: %d", req->channels); + if (req->json_body) + notify_add_header(req, "%s", "Content-Type: application/json"); + + if (o && req->event == NOTIFY_EVT_FINISHED && (output_storage & OUTPUT_STORAGE_NOTIFY) && !req->content) req->content = output_get_content(o); } static void cleanup_http(notif_req_t *req) { curl_slist_free_all(req->headers); + req->headers = NULL; obj_release(req->content); + req->content = NULL; } static const notif_action_t http_action = { @@ -281,21 +389,64 @@ static const notif_action_t http_action = { .failed = failed_http, }; - - static void notify_req_setup_command(notif_req_t *req, output_t *o, metafile_t *mf, tag_t *tag) { - req->argv = g_new(char *, 4); - req->argv[0] = g_strdup(notify_command); - if ((output_storage & OUTPUT_STORAGE_FILE)) - req->argv[1] = g_strdup_printf("%s.%s", o->full_filename, o->file_format); - else - req->argv[1] = g_strdup(""); - req->argv[2] = g_strdup_printf("%llu", req->db_id); - req->argv[3] = NULL; + const char *path = req->full_filename ? req->full_filename : ""; + const char *call_id = req->call_id ? req->call_id : (mf && mf->call_id ? mf->call_id : ""); + const char *kind = req->kind ? req->kind : (o && o->kind ? o->kind : ""); + + if (!path[0] && o) { + if (o->filename) + path = o->filename; + else if ((output_storage & OUTPUT_STORAGE_FILE) && o->full_filename && o->file_format) { + req->full_filename = g_strdup_printf("%s.%s", o->full_filename, o->file_format); + path = req->full_filename; + } + } + + switch ((enum notify_command_format) notify_command_format) { + case NOTIFY_CMD_EXTENDED: + req->argv = g_new0(char *, 8); + req->argv[0] = g_strdup(notify_command); + req->argv[1] = g_strdup(notify_event_name(req->event)); + req->argv[2] = g_strdup(path); + req->argv[3] = g_strdup_printf("%llu", req->db_id); + req->argv[4] = g_strdup(call_id); + req->argv[5] = g_strdup(kind); + req->argv[6] = g_strdup(path); + req->argv[7] = NULL; + break; + case NOTIFY_CMD_JSON_ENV: { + req->argv = g_new0(char *, 2); + req->argv[0] = g_strdup(notify_command); + req->argv[1] = NULL; + gchar **env = g_get_environ(); + const char *payload = req->json_body ? req->json_body : "{}"; + env = g_environ_setenv(env, "NOTIFY_PAYLOAD", payload, TRUE); + env = g_environ_setenv(env, "NOTIFY_EVENT", notify_event_name(req->event), TRUE); + req->envp = env; + break; + } + case NOTIFY_CMD_LEGACY: + default: + req->argv = g_new0(char *, 4); + req->argv[0] = g_strdup(notify_command); + if (path[0]) + req->argv[1] = g_strdup(path); + else if (o && (output_storage & OUTPUT_STORAGE_FILE) && o->full_filename && o->file_format) + req->argv[1] = g_strdup_printf("%s.%s", o->full_filename, o->file_format); + else + req->argv[1] = g_strdup(""); + req->argv[2] = g_strdup_printf("%llu", req->db_id); + req->argv[3] = NULL; + break; + } } static void cleanup_command(notif_req_t *req) { g_strfreev(req->argv); + req->argv = NULL; + g_strfreev(req->envp); + req->envp = NULL; } static const notif_action_t command_action = { @@ -305,27 +456,333 @@ static const notif_action_t command_action = { .cleanup = cleanup_command, }; +static char *notify_build_json(notif_req_t *req) { + double now = (double) now_us() / 1000000.; + bool terminal = notify_event_is_terminal(req->event); + JsonBuilder *b = json_builder_new(); + json_builder_begin_object(b); + + json_builder_set_member_name(b, "event"); + json_builder_add_string_value(b, notify_event_name(req->event)); + json_builder_set_member_name(b, "event_time"); + json_builder_add_double_value(b, now); + json_builder_set_member_name(b, "status"); + json_builder_add_string_value(b, notify_event_status(req->event)); + json_builder_set_member_name(b, "call_id"); + json_builder_add_string_value(b, req->call_id ? req->call_id : ""); + + json_builder_set_member_name(b, "kind"); + if (req->kind) + json_builder_add_string_value(b, req->kind); + else + json_builder_add_null_value(b); + json_builder_set_member_name(b, "file_name"); + if (req->file_name && req->file_format) { + char *fn = g_strdup_printf("%s.%s", req->file_name, req->file_format); + json_builder_add_string_value(b, fn); + g_free(fn); + } + else if (req->file_name) + json_builder_add_string_value(b, req->file_name); + else + json_builder_add_null_value(b); + json_builder_set_member_name(b, "full_file_name"); + if (req->full_filename) + json_builder_add_string_value(b, req->full_filename); + else + json_builder_add_null_value(b); + json_builder_set_member_name(b, "file_format"); + if (req->file_format) + json_builder_add_string_value(b, req->file_format); + else + json_builder_add_null_value(b); + json_builder_set_member_name(b, "output_id"); + if (req->output_id) + json_builder_add_string_value(b, req->output_id); + else + json_builder_add_null_value(b); + + json_builder_set_member_name(b, "db"); + json_builder_begin_object(b); + json_builder_set_member_name(b, "call_id"); + if (req->call_db_id) + json_builder_add_int_value(b, (gint64) req->call_db_id); + else + json_builder_add_null_value(b); + json_builder_set_member_name(b, "stream_id"); + if (req->db_id) + json_builder_add_int_value(b, (gint64) req->db_id); + else + json_builder_add_null_value(b); + json_builder_end_object(b); + + json_builder_set_member_name(b, "times"); + json_builder_begin_object(b); + json_builder_set_member_name(b, "call_start"); + json_builder_add_double_value(b, req->call_start); + json_builder_set_member_name(b, "stream_start"); + json_builder_add_double_value(b, req->stream_start > 0 ? req->stream_start : req->call_start); + json_builder_set_member_name(b, "stream_end"); + if (terminal) + json_builder_add_double_value(b, now); + else + json_builder_add_null_value(b); + json_builder_end_object(b); + + json_builder_set_member_name(b, "media"); + json_builder_begin_object(b); + json_builder_set_member_name(b, "sample_rate"); + if (req->sample_rate > 0) + json_builder_add_int_value(b, req->sample_rate); + else + json_builder_add_null_value(b); + json_builder_set_member_name(b, "channels"); + if (req->channels > 0) + json_builder_add_int_value(b, req->channels); + else + json_builder_add_null_value(b); + json_builder_set_member_name(b, "duration_ms"); + if (req->duration_ms > 0) + json_builder_add_int_value(b, req->duration_ms); + else + json_builder_add_null_value(b); + json_builder_set_member_name(b, "file_size"); + if (req->file_size > 0) + json_builder_add_int_value(b, req->file_size); + else + json_builder_add_null_value(b); + json_builder_end_object(b); + + json_builder_set_member_name(b, "tag"); + if (req->tag_name || req->tag_label) { + json_builder_begin_object(b); + json_builder_set_member_name(b, "name"); + json_builder_add_string_value(b, req->tag_name ? req->tag_name : ""); + json_builder_set_member_name(b, "label"); + json_builder_add_string_value(b, req->tag_label ? req->tag_label : ""); + json_builder_end_object(b); + } + else + json_builder_add_null_value(b); + json_builder_set_member_name(b, "metadata"); + if (!notify_no_metadata && req->metadata) + json_builder_add_string_value(b, req->metadata); + else + json_builder_add_null_value(b); + + json_builder_set_member_name(b, "error"); + if (req->error_code || req->error_message) { + json_builder_begin_object(b); + json_builder_set_member_name(b, "code"); + json_builder_add_string_value(b, req->error_code ? req->error_code : ""); + json_builder_set_member_name(b, "message"); + json_builder_add_string_value(b, req->error_message ? req->error_message : ""); + json_builder_end_object(b); + } + else + json_builder_add_null_value(b); -void notify_push_setup(const notif_action_t *action, output_t *o, metafile_t *mf, tag_t *tag) { - notif_req_t *req = g_new0(__typeof(*req), 1); + json_builder_end_object(b); + return glib_json_print(b); +} - req->name = g_strdup_printf("%s for '%s'", action->name, o->file_name); - req->action = action; +static void notify_fill_media_extras(notif_req_t *req, output_t *o) { + if (!o) + return; + if (req->sample_rate <= 0 && o->actual_format.clockrate > 0) + req->sample_rate = o->actual_format.clockrate; + if (req->channels <= 0 && o->actual_format.channels > 0) + req->channels = o->actual_format.channels; + if (req->duration_ms <= 0) { + int64_t start = o->first_write_time_us ? o->first_write_time_us : o->start_time_us; + if (start > 0) { + int64_t end = now_us(); + if (end > start) + req->duration_ms = (end - start) / 1000; + } + } + if (req->file_size <= 0 && o->filename) { + struct stat st; + if (stat(o->filename, &st) == 0) + req->file_size = (int64_t) st.st_size; + } +} - req->db_id = o->db_id; +static void notify_req_snapshot(notif_req_t *req, enum notify_event event, + output_t *o, metafile_t *mf, tag_t *tag, + const char *error_code, const char *error_message) +{ + req->event = event; + req->terminal = notify_event_is_terminal(event) ? 1 : 0; + + if (mf) { + req->call_id = g_strdup(mf->call_id); + req->call_start = (double) mf->start_time_us / 1000000.; + req->call_db_id = mf->db_id; + if (!notify_no_metadata && mf->metadata) + req->metadata = g_strdup(mf->metadata); + } + if (o) { + req->db_id = o->db_id; + req->file_name = g_strdup(o->file_name); + req->file_format = g_strdup(o->file_format); + req->kind = g_strdup(o->kind); + req->output_id = g_strdup(o->output_id); + req->stream_start = (double) o->start_time_us / 1000000.; + if (o->filename) + req->full_filename = g_strdup(o->filename); + else if (o->full_filename && o->file_format) + req->full_filename = g_strdup_printf("%s.%s", o->full_filename, o->file_format); + else if (o->full_filename) + req->full_filename = g_strdup(o->full_filename); + } + else if (mf) { + req->db_id = mf->db_id; + req->stream_start = req->call_start; + } + if (tag) { + req->tag_name = g_strdup(tag->name); + req->tag_label = g_strdup(tag->label); + req->tag_metadata = g_strdup(tag->metadata); + } + if (error_code) + req->error_code = g_strdup(error_code); + if (error_message) + req->error_message = g_strdup(error_message); + if (req->terminal && o) + notify_fill_media_extras(req, o); + if (notify_json || notify_command_format == (int) NOTIFY_CMD_JSON_ENV) + req->json_body = notify_build_json(req); +} + +static void notify_enqueue(const notif_action_t *action, notif_req_t *snap, + output_t *o, metafile_t *mf, tag_t *tag) +{ + if (!notify_threadpool || !action) + return; + + notif_req_t *req = g_new0(__typeof(*req), 1); + req->event = snap->event; + req->terminal = snap->terminal; + req->db_id = snap->db_id; + req->call_db_id = snap->call_db_id; + req->call_start = snap->call_start; + req->stream_start = snap->stream_start; + req->file_size = snap->file_size; + req->duration_ms = snap->duration_ms; + req->sample_rate = snap->sample_rate; + req->channels = snap->channels; + req->call_id = g_strdup(snap->call_id); + req->file_name = g_strdup(snap->file_name); + req->file_format = g_strdup(snap->file_format); + req->kind = g_strdup(snap->kind); + req->full_filename = g_strdup(snap->full_filename); + req->output_id = g_strdup(snap->output_id); + req->metadata = g_strdup(snap->metadata); + req->tag_name = g_strdup(snap->tag_name); + req->tag_label = g_strdup(snap->tag_label); + req->tag_metadata = g_strdup(snap->tag_metadata); + req->error_code = g_strdup(snap->error_code); + req->error_message = g_strdup(snap->error_message); + req->json_body = g_strdup(snap->json_body); + + req->name = g_strdup_printf("%s/%s for '%s'", action->name, + notify_event_name(req->event), + req->file_name ? req->file_name : + (req->call_id ? req->call_id : "unknown")); + req->action = action; + req->falloff_us = 5000000LL; action->setup(req, o, mf, tag); - req->falloff_us = 5000000LL; // initial retry time + if (action == &http_action && !notify_json) + g_clear_pointer(&req->json_body, g_free); + metric_inc(¬ify_metric_enqueued); g_thread_pool_push(notify_threadpool, req, NULL); } -void notify_push_output(output_t *o, metafile_t *mf, tag_t *tag) { +static void notify_push_event(enum notify_event event, output_t *o, metafile_t *mf, + tag_t *tag, const char *error_code, const char *error_message) +{ + if (!notify_threadpool || !mf) + return; + if (!notify_event_enabled(event)) + return; + if (!notify_uri && !notify_command) + return; + + bool terminal = notify_event_is_terminal(event); + if (!terminal) { + if (!nont_inflight_try_inc()) { + metric_inc(¬ify_metric_dropped); + ilog(LOG_DEBUG, "Dropping non-terminal notify %s (queue limit %d)", + notify_event_name(event), notify_queue_limit); + return; + } + } + + notif_req_t snap = {0}; + notify_req_snapshot(&snap, event, o, mf, tag, error_code, error_message); + + ilog(LOG_DEBUG, "Queueing notification event %s for '%s%s%s'", + notify_event_name(event), + FMT_M(snap.file_name ? snap.file_name : + (snap.call_id ? snap.call_id : "unknown"))); + + /* One non-terminal slot covers both actions when both are enabled. */ if (notify_uri) - notify_push_setup(&http_action, o, mf, tag); + notify_enqueue(&http_action, &snap, o, mf, tag); + if (notify_command) { + /* Avoid double nont_inflight_dec when both HTTP and command run. */ + if (notify_uri && !terminal) + snap.terminal = 1; + notify_enqueue(&command_action, &snap, o, mf, tag); + } + + req_free_fields(&snap); +} + +void notify_push_setup(const notif_action_t *action, output_t *o, metafile_t *mf, tag_t *tag) { + /* Compatibility entry used by other modules (S3/GCS share pattern). */ + if (!notify_threadpool || !action || !o) + return; + notif_req_t *req = g_new0(__typeof(*req), 1); + req->name = g_strdup_printf("%s for '%s'", action->name, o->file_name ? o->file_name : "unknown"); + req->action = action; + req->event = NOTIFY_EVT_FINISHED; + req->terminal = 1; + req->db_id = o->db_id; + req->falloff_us = 5000000LL; + action->setup(req, o, mf, tag); + metric_inc(¬ify_metric_enqueued); + g_thread_pool_push(notify_threadpool, req, NULL); +} - if (notify_command) - notify_push_setup(&command_action, o, mf, tag); +void notify_push_output(output_t *o, metafile_t *mf, tag_t *tag) { + if (o) + o->notify_terminal = 1; + notify_push_event(NOTIFY_EVT_FINISHED, o, mf, tag, NULL, NULL); +} + +void notify_push_output_event(enum notify_event event, output_t *o, metafile_t *mf, + tag_t *tag, const char *error_code, const char *error_message) +{ + notify_push_event(event, o, mf, tag, error_code, error_message); +} + +void notify_push_call_event(enum notify_event event, metafile_t *mf) { + if (!mf) + return; + notify_push_event(event, NULL, mf, NULL, NULL, NULL); +} + +void notify_push_call(metafile_t *mf) { + if (!mf) + return; + if (mf->notify_call_terminal) + return; + mf->notify_call_terminal = 1; + notify_push_call_event(mf->discard ? NOTIFY_EVT_CALL_DISCARDED : NOTIFY_EVT_CALL_FINISHED, mf); } diff --git a/recording-daemon/notify.h b/recording-daemon/notify.h index f875307fc3..50c4ac4759 100644 --- a/recording-daemon/notify.h +++ b/recording-daemon/notify.h @@ -3,13 +3,17 @@ #include #include "types.h" +#include "notify_events.h" void notify_setup(void); void notify_cleanup(void); void notify_push_setup(const notif_action_t *action, output_t *o, metafile_t *mf, tag_t *tag); - void notify_push_output(output_t *, metafile_t *, tag_t *); -void notify_push_call(metafile_t *); + +void notify_push_output_event(enum notify_event event, output_t *o, metafile_t *mf, + tag_t *tag, const char *error_code, const char *error_message); +void notify_push_call_event(enum notify_event event, metafile_t *mf); +void notify_push_call(metafile_t *mf); #endif diff --git a/recording-daemon/notify_events.c b/recording-daemon/notify_events.c new file mode 100644 index 0000000000..51cfa237e4 --- /dev/null +++ b/recording-daemon/notify_events.c @@ -0,0 +1,120 @@ +#include "notify_events.h" +#include +#include + +/* + * Pure event helpers (no curl / thread pool). + * Linked into the recording daemon and into unit tests. + */ + +static const char *const event_names[NOTIFY_EVT_COUNT] = { + [NOTIFY_EVT_FILE_OPENED] = "recording_file_opened", + [NOTIFY_EVT_STARTED] = "recording_started", + [NOTIFY_EVT_FINISHED] = "recording_finished", + [NOTIFY_EVT_DISCARDED] = "recording_discarded", + [NOTIFY_EVT_FAILED] = "recording_failed", + [NOTIFY_EVT_CALL_STARTED] = "call_recording_started", + [NOTIFY_EVT_CALL_FINISHED] = "call_recording_finished", + [NOTIFY_EVT_CALL_DISCARDED] = "call_recording_discarded", +}; + +static const char *const event_statuses[NOTIFY_EVT_COUNT] = { + [NOTIFY_EVT_FILE_OPENED] = "opened", + [NOTIFY_EVT_STARTED] = "started", + [NOTIFY_EVT_FINISHED] = "finished", + [NOTIFY_EVT_DISCARDED] = "discarded", + [NOTIFY_EVT_FAILED] = "failed", + [NOTIFY_EVT_CALL_STARTED] = "call-started", + [NOTIFY_EVT_CALL_FINISHED] = "call-finished", + [NOTIFY_EVT_CALL_DISCARDED] = "call-discarded", +}; + +const char *notify_event_name(enum notify_event event) { + if (event < 0 || event >= NOTIFY_EVT_COUNT) + return "unknown"; + return event_names[event]; +} + +const char *notify_event_status(enum notify_event event) { + if (event < 0 || event >= NOTIFY_EVT_COUNT) + return "unknown"; + return event_statuses[event]; +} + +bool notify_event_enabled(enum notify_event event) { + if (event < 0 || event >= NOTIFY_EVT_COUNT) + return false; + return (notify_events_mask & (1u << event)) != 0; +} + +bool notify_events_parse(const char *csv, unsigned int *mask_out, char **err_token) { + unsigned int mask = 0; + + if (err_token) + *err_token = NULL; + + if (!csv || !csv[0]) { + *mask_out = NOTIFY_MASK_DEFAULT; + return true; + } + + g_autofree char *dup = g_strdup(csv); + char *save = NULL; + for (char *tok = strtok_r(dup, ", \t", &save); tok; tok = strtok_r(NULL, ", \t", &save)) { + if (!strcmp(tok, "opened")) + mask |= NOTIFY_MASK_OPENED; + else if (!strcmp(tok, "started")) + mask |= NOTIFY_MASK_STARTED; + else if (!strcmp(tok, "finished")) + mask |= NOTIFY_MASK_FINISHED; + else if (!strcmp(tok, "discarded")) + mask |= NOTIFY_MASK_DISCARDED; + else if (!strcmp(tok, "failed")) + mask |= NOTIFY_MASK_FAILED; + else if (!strcmp(tok, "call-started")) + mask |= NOTIFY_MASK_CALL_STARTED; + else if (!strcmp(tok, "call-finished")) + mask |= NOTIFY_MASK_CALL_FINISHED; + else if (!strcmp(tok, "call-discarded")) + mask |= NOTIFY_MASK_CALL_DISCARDED; + else if (!strcmp(tok, "all")) + mask |= NOTIFY_MASK_ALL; + else { + if (err_token) + *err_token = g_strdup(tok); + return false; + } + } + + if (mask == 0) + mask = NOTIFY_MASK_DEFAULT; + + *mask_out = mask; + return true; +} + +bool notify_event_is_terminal(enum notify_event event) { + return event == NOTIFY_EVT_FINISHED + || event == NOTIFY_EVT_DISCARDED + || event == NOTIFY_EVT_FAILED + || event == NOTIFY_EVT_CALL_FINISHED + || event == NOTIFY_EVT_CALL_DISCARDED; +} + +bool notify_command_format_parse(const char *s, enum notify_command_format *out) { + if (!out) + return false; + if (!s || !s[0] || !strcmp(s, "legacy")) { + *out = NOTIFY_CMD_LEGACY; + return true; + } + if (!strcmp(s, "extended")) { + *out = NOTIFY_CMD_EXTENDED; + return true; + } + if (!strcmp(s, "json-env")) { + *out = NOTIFY_CMD_JSON_ENV; + return true; + } + return false; +} diff --git a/recording-daemon/notify_events.h b/recording-daemon/notify_events.h new file mode 100644 index 0000000000..a6ba51f919 --- /dev/null +++ b/recording-daemon/notify_events.h @@ -0,0 +1,58 @@ +#ifndef _NOTIFY_EVENTS_H_ +#define _NOTIFY_EVENTS_H_ + +#include +#include + +/** + * Recording lifecycle notification events (shared with unit tests). + * Full notify transport API lives in notify.h. + */ +enum notify_event { + NOTIFY_EVT_FILE_OPENED = 0, + NOTIFY_EVT_STARTED, + NOTIFY_EVT_FINISHED, + NOTIFY_EVT_DISCARDED, + NOTIFY_EVT_FAILED, + NOTIFY_EVT_CALL_STARTED, + NOTIFY_EVT_CALL_FINISHED, + NOTIFY_EVT_CALL_DISCARDED, + NOTIFY_EVT_COUNT, +}; + +#define NOTIFY_MASK_OPENED (1u << NOTIFY_EVT_FILE_OPENED) +#define NOTIFY_MASK_STARTED (1u << NOTIFY_EVT_STARTED) +#define NOTIFY_MASK_FINISHED (1u << NOTIFY_EVT_FINISHED) +#define NOTIFY_MASK_DISCARDED (1u << NOTIFY_EVT_DISCARDED) +#define NOTIFY_MASK_FAILED (1u << NOTIFY_EVT_FAILED) +#define NOTIFY_MASK_CALL_STARTED (1u << NOTIFY_EVT_CALL_STARTED) +#define NOTIFY_MASK_CALL_FINISHED (1u << NOTIFY_EVT_CALL_FINISHED) +#define NOTIFY_MASK_CALL_DISCARDED (1u << NOTIFY_EVT_CALL_DISCARDED) +#define NOTIFY_MASK_STREAM_ALL (NOTIFY_MASK_OPENED | NOTIFY_MASK_STARTED | \ + NOTIFY_MASK_FINISHED | NOTIFY_MASK_DISCARDED | \ + NOTIFY_MASK_FAILED) +#define NOTIFY_MASK_CALL_ALL (NOTIFY_MASK_CALL_STARTED | NOTIFY_MASK_CALL_FINISHED | \ + NOTIFY_MASK_CALL_DISCARDED) +#define NOTIFY_MASK_ALL (NOTIFY_MASK_STREAM_ALL | NOTIFY_MASK_CALL_ALL) +#define NOTIFY_MASK_DEFAULT NOTIFY_MASK_FINISHED + +bool notify_events_parse(const char *csv, unsigned int *mask_out, char **err_token); +const char *notify_event_name(enum notify_event event); +const char *notify_event_status(enum notify_event event); +bool notify_event_enabled(enum notify_event event); + +/* Terminal = finished/discarded/failed and call terminal counterparts. */ +bool notify_event_is_terminal(enum notify_event event); + +/* notify-command-format tokens (pure parse; no transport deps). */ +enum notify_command_format { + NOTIFY_CMD_LEGACY = 0, + NOTIFY_CMD_EXTENDED, + NOTIFY_CMD_JSON_ENV, +}; +bool notify_command_format_parse(const char *s, enum notify_command_format *out); + +/* Defined by main.c in daemon; unit tests provide their own. */ +extern unsigned int notify_events_mask; + +#endif diff --git a/recording-daemon/output.c b/recording-daemon/output.c index 615c415a8d..69de84056c 100644 --- a/recording-daemon/output.c +++ b/recording-daemon/output.c @@ -3,6 +3,7 @@ #include #include #include +#include #include #include #include @@ -44,6 +45,13 @@ static int output_got_packet(encoder_t *enc, void *u1, void *u2) { av_write_frame(output->fmtctx, enc->avpkt); + if (!output->notify_started && output->metafile) { + output->notify_started = 1; + if (!output->first_write_time_us) + output->first_write_time_us = now_us(); + notify_push_output_event(NOTIFY_EVT_STARTED, output, output->metafile, NULL, NULL, NULL); + } + return 0; } @@ -316,6 +324,10 @@ output_t *output_new_ext(metafile_t *mf, const char *type, const char *kind, con if (resample_audio > 0) ret->sink.format.clockrate = resample_audio; + ret->metafile = mf; + ret->output_id = g_strdup_printf("%s-%s-%" PRIu64, + mf->call_id ? mf->call_id : "call", type ? type : "out", + (uint64_t) now_us()); return ret; } @@ -492,6 +504,11 @@ static const char *output_setup(output_t *output, const format_t *requested_form db_config_stream(output); ilog(LOG_INFO, "Opened output media file '%s' for writing", output->filename ?: "(mem stream)"); + if (!output->notify_opened && output->metafile) { + output->notify_opened = 1; + notify_push_output_event(NOTIFY_EVT_FILE_OPENED, output, output->metafile, NULL, NULL, NULL); + } + return NULL; } @@ -511,6 +528,11 @@ static bool output_config(sink_t *sink, output_t *output, const format_t *reques if (err) { output_shutdown(output); ilog(LOG_ERR, "Error configuring media output: %s", err); + if (output->metafile && !output->notify_terminal) { + output->notify_terminal = 1; + notify_push_output_event(NOTIFY_EVT_FAILED, output, output->metafile, NULL, + "open_failed", err); + } return false; } } @@ -672,20 +694,34 @@ void output_close(metafile_t *mf, output_t *output, tag_t *tag, bool discard) { if (!output) return; bool do_delete = !(output_storage & OUTPUT_STORAGE_FILE); + if (!mf) + mf = output->metafile; if (!discard) { if (output_shutdown(output)) { db_close_stream(output); - notify_push_output(output, mf, tag); + if (!output->notify_terminal) { + output->notify_terminal = 1; + notify_push_output(output, mf, tag); + } s3_store(output, mf); gcs_store(output, mf); } - else + else { db_delete_stream(mf, output); + if (mf && !output->notify_terminal) { + output->notify_terminal = 1; + notify_push_output_event(NOTIFY_EVT_DISCARDED, output, mf, tag, NULL, NULL); + } + } } else { output_shutdown(output); do_delete = true; db_delete_stream(mf, output); + if (mf && !output->notify_terminal) { + output->notify_terminal = 1; + notify_push_output_event(NOTIFY_EVT_DISCARDED, output, mf, tag, NULL, NULL); + } } encoder_free(output->encoder); if (output->filename && do_delete) { @@ -700,6 +736,7 @@ void output_close(metafile_t *mf, output_t *output, tag_t *tag, bool discard) { g_clear_pointer(&output->file_path, g_free); g_clear_pointer(&output->file_name, g_free); g_clear_pointer(&output->filename, g_free); + g_clear_pointer(&output->output_id, g_free); g_clear_pointer(&output->iobuf, g_free); if (output->membuf) g_string_free(output->membuf, TRUE); diff --git a/recording-daemon/packet.c b/recording-daemon/packet.c index a261646636..059a5e6ea2 100644 --- a/recording-daemon/packet.c +++ b/recording-daemon/packet.c @@ -13,6 +13,7 @@ #include "rtcplib.h" #include "main.h" #include "output.h" +#include "notify.h" #include "db.h" #include "streambuf.h" #include "resample.h" @@ -85,6 +86,10 @@ static ssrc_t *ssrc_get(stream_t *stream, unsigned long ssrc) { } db_do_stream(mf, ret->output, stream, ssrc); + if (ret->output && !mf->notify_call_started) { + mf->notify_call_started = 1; + notify_push_call_event(NOTIFY_EVT_CALL_STARTED, mf); + } } tls_fwd_init(stream, mf, ret); diff --git a/recording-daemon/types.h b/recording-daemon/types.h index 2435b17775..a0ff0e5763 100644 --- a/recording-daemon/types.h +++ b/recording-daemon/types.h @@ -17,6 +17,7 @@ #include "socket.h" #include "containers.h" #include "obj.h" +#include "notify_events.h" struct iphdr; @@ -189,6 +190,8 @@ struct metafile_s { unsigned int db_metadata_done:1; unsigned int skip_db:1; unsigned int started:1; + unsigned int notify_call_started:1; + unsigned int notify_call_terminal:1; }; @@ -217,6 +220,14 @@ struct output_s { actual_format; content_t *content; + + /* lifecycle notify state */ + metafile_t *metafile; + char *output_id; + unsigned int notify_opened:1; + unsigned int notify_started:1; + unsigned int notify_terminal:1; + int64_t first_write_time_us; }; @@ -254,6 +265,7 @@ struct notif_req_s { // notify command struct { char **argv; + char **envp; }; // db writer @@ -267,11 +279,36 @@ struct notif_req_s { }; }; + /* Outside the action union so HTTP + command can both use JSON payloads. */ + char *json_body; + // used by multiple actions unsigned long long db_id; content_t *content; char *object_name; + enum notify_event event; + unsigned int terminal:1; + char *call_id; + char *file_name; + char *file_format; + char *kind; + char *full_filename; + char *output_id; + char *metadata; + char *tag_name; + char *tag_label; + char *tag_metadata; + char *error_code; + char *error_message; + double call_start; + double stream_start; + unsigned long long call_db_id; + int64_t file_size; + int64_t duration_ms; + int sample_rate; + int channels; + const notif_action_t *action; int64_t retry_time; diff --git a/t/Makefile b/t/Makefile index 89e7cc6de9..74dc49413d 100644 --- a/t/Makefile +++ b/t/Makefile @@ -84,7 +84,7 @@ include ../lib/common.Makefile daemon-tests-bundle daemon-tests-dtls \ daemon-tests-recording daemon-tests-create daemon-tests-alias -TESTS := test-bitstr aes-crypt aead-aes-crypt test-const_str_hash.strhash +TESTS := test-bitstr aes-crypt aead-aes-crypt test-const_str_hash.strhash test-notify-events ifeq ($(with_transcoding),yes) TESTS += test-transcode test-dtmf-detect test-payload-tracker test-resample test-stats test-mix-buffer ifeq ($(RTPENGINE_EXTENDED_TESTS),1) @@ -282,6 +282,8 @@ daemon-tests-recording: daemon-test-deps test-bitstr: test-bitstr.o +test-notify-events: test-notify-events.o + test-mix-buffer: test-mix-buffer.o \ $(COMMONOBJS) \ ../daemon/crypto.o \ diff --git a/t/notify_e2e/notify_receiver.py b/t/notify_e2e/notify_receiver.py new file mode 100644 index 0000000000..60894973be --- /dev/null +++ b/t/notify_e2e/notify_receiver.py @@ -0,0 +1,71 @@ +#!/usr/bin/env python3 +"""Minimal HTTP notify sink for rtpengine-recording lifecycle events.""" +from __future__ import annotations + +import json +import sys +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from urllib.parse import urlparse + +LOG = Path(sys.argv[1] if len(sys.argv) > 1 else "/tmp/notify_events.jsonl") +PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 8099 +LOCK = threading.Lock() +SEEN: list[dict] = [] + + +class Handler(BaseHTTPRequestHandler): + def log_message(self, fmt, *args): # quieter + sys.stderr.write("[recv] " + (fmt % args) + "\n") + + def _handle(self): + length = int(self.headers.get("Content-Length") or 0) + body = self.rfile.read(length) if length else b"" + headers = {k: v for k, v in self.headers.items() if k.lower().startswith("x-recording") + or k.lower() in ("content-type", "user-agent")} + entry = { + "ts": time.time(), + "method": self.command, + "path": urlparse(self.path).path, + "headers": headers, + "body_raw": body.decode("utf-8", errors="replace"), + "body_json": None, + } + if body: + try: + entry["body_json"] = json.loads(body) + except Exception: + pass + with LOCK: + SEEN.append(entry) + with LOG.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(entry, separators=(",", ":")) + "\n") + ev = headers.get("X-Recording-Event") or headers.get("x-recording-event") + print(f"EVENT {ev} method={self.command} path={self.path} body_bytes={len(body)}", flush=True) + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"ok":true}') + + def do_GET(self): + self._handle() + + def do_POST(self): + self._handle() + + def do_PUT(self): + self._handle() + + +def main(): + LOG.parent.mkdir(parents=True, exist_ok=True) + LOG.write_text("") + httpd = ThreadingHTTPServer(("0.0.0.0", PORT), Handler) + print(f"notify receiver on :{PORT} log={LOG}", flush=True) + httpd.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/t/notify_e2e/run_e2e.sh b/t/notify_e2e/run_e2e.sh new file mode 100644 index 0000000000..a132480037 --- /dev/null +++ b/t/notify_e2e/run_e2e.sh @@ -0,0 +1,176 @@ +#!/usr/bin/env bash +# End-to-end notify smoke test inside Linux (Docker). +# Creates a metafile, deletes it -> expects call_recording_finished (and others if armed). +set -euo pipefail + +ROOT="${ROOT:-/rtpengine}" +WORK="${WORK:-/tmp/notify_e2e_work}" +SPOOL="$WORK/spool" +OUTDIR="$WORK/out" +LOGDIR="$WORK/logs" +EVTLOG="$LOGDIR/notify_events.jsonl" +RECV_PORT=8099 +RECV_URL="http://127.0.0.1:${RECV_PORT}/rec/events" + +rm -rf "$WORK" +mkdir -p "$SPOOL" "$OUTDIR" "$LOGDIR" + +write_section() { + # write_section FILE SECTION CONTENT + local f="$1" sec="$2" content="$3" + local len + len=$(printf '%s' "$content" | wc -c) + printf '%s\n%u:\n%s\n\n' "$sec" "$len" "$content" >>"$f" +} + +echo "== starting notify receiver ==" +python3 "$ROOT/t/notify_e2e/notify_receiver.py" "$EVTLOG" "$RECV_PORT" \ + >"$LOGDIR/receiver.out" 2>"$LOGDIR/receiver.err" & +RECV_PID=$! +sleep 0.4 +if ! kill -0 "$RECV_PID" 2>/dev/null; then + echo "receiver failed to start" >&2 + cat "$LOGDIR/receiver.err" >&2 || true + exit 1 +fi + +DAEMON_PID="" +cleanup() { + [[ -n "${DAEMON_PID:-}" ]] && kill "$DAEMON_PID" 2>/dev/null || true + [[ -n "${RECV_PID:-}" ]] && kill "$RECV_PID" 2>/dev/null || true + [[ -n "${DAEMON_PID:-}" ]] && wait "$DAEMON_PID" 2>/dev/null || true + [[ -n "${RECV_PID:-}" ]] && wait "$RECV_PID" 2>/dev/null || true +} +trap cleanup EXIT + +echo "== starting rtpengine-recording ==" +# Prefer in-tree binary: image-built path, then $ROOT, then PATH +BIN="" +for cand in \ + /opt/rtpengine/recording-daemon/rtpengine-recording \ + /rtpengine/recording-daemon/rtpengine-recording \ + "$ROOT/recording-daemon/rtpengine-recording" +do + if [[ -x "$cand" ]]; then BIN="$cand"; break; fi +done +if [[ -z "$BIN" ]]; then + BIN=$(command -v rtpengine-recording || true) +fi +if [[ -z "$BIN" || ! -x "$BIN" ]]; then + echo "rtpengine-recording binary not found" >&2 + ls -la /rtpengine/recording-daemon 2>/dev/null | head || true + ls -la /opt/rtpengine/recording-daemon 2>/dev/null | head || true + exit 1 +fi +echo "using binary: $BIN" + + +"$BIN" \ + --foreground \ + --log-stderr \ + --log-level=7 \ + --table=0 \ + --spool-dir="$SPOOL" \ + --output-dir="$OUTDIR" \ + --output-storage=file \ + --output-format=wav \ + --output-single \ + --num-threads=2 \ + --notify-uri="$RECV_URL" \ + --notify-json \ + --notify-events=opened,started,finished,discarded,failed,call-started,call-finished,call-discarded \ + --notify-concurrency=2 \ + --notify-retries=1 \ + >"$LOGDIR/daemon.out" 2>"$LOGDIR/daemon.err" & +DAEMON_PID=$! +sleep 1.0 +if ! kill -0 "$DAEMON_PID" 2>/dev/null; then + echo "daemon failed to start" >&2 + cat "$LOGDIR/daemon.err" >&2 || true + exit 1 +fi +echo "daemon pid=$DAEMON_PID receiver pid=$RECV_PID" + +# ---- scenario 1: call finished (no streams) ---- +CALL="testcall-$(date +%s)" +META="$SPOOL/$CALL" +echo "== writing metafile $META ==" +: >"$META" +write_section "$META" "CALL-ID" "$CALL" +write_section "$META" "RANDOM_TAG" "aabbccddeeff0011" +write_section "$META" "METADATA" "foo:bar|test:notify-e2e" +write_section "$META" "RECORDING" "1" +# Ensure CLOSE_WRITE by rewriting via temp + mv? inotify watches close_write of file. +# Writing then closing by sync/open-close cycle: +python3 - <"$META2" +write_section "$META2" "CALL-ID" "$CALL2" +write_section "$META2" "RANDOM_TAG" "1122334455667788" +write_section "$META2" "METADATA" "case:discard" +write_section "$META2" "RECORDING" "0" +python3 - <&2 + echo "--- daemon.err ---"; tail -80 "$LOGDIR/daemon.err" || true + echo "--- receiver.err ---"; tail -40 "$LOGDIR/receiver.err" || true + exit 2 +fi + +python3 - "$EVTLOG" <<'PY' +import json, sys +from pathlib import Path +log = Path(sys.argv[1]) +rows = [json.loads(l) for l in log.read_text().splitlines() if l.strip()] +print(f"total_events={len(rows)}") +events = [] +for r in rows: + h = {k.lower(): v for k, v in r.get("headers", {}).items()} + ev = h.get("x-recording-event") or (r.get("body_json") or {}).get("event") + st = h.get("x-recording-status") or (r.get("body_json") or {}).get("status") + cid = h.get("x-recording-call-id") or (r.get("body_json") or {}).get("call_id") + events.append(ev) + print(f" - method={r['method']} event={ev} status={st} call_id={cid} body_json={'yes' if r.get('body_json') else 'no'}") + if r.get("body_json"): + print(" json keys:", sorted(r["body_json"].keys())) + +need = {"call_recording_finished"} +got = set(e for e in events if e) +missing = need - got +if missing: + print("FAIL missing required events:", sorted(missing)) + print("got:", events) + sys.exit(3) +if not any(r.get("body_json") for r in rows): + print("FAIL: expected JSON bodies (notify-json)") + sys.exit(4) +print("E2E_NOTIFY_OK") +PY + + +echo "== sample event log ==" +cat "$EVTLOG" +echo +echo "DONE" diff --git a/t/test-notify-events.c b/t/test-notify-events.c new file mode 100644 index 0000000000..f9408b42d4 --- /dev/null +++ b/t/test-notify-events.c @@ -0,0 +1,156 @@ +#include +#include +#include +#include + +/* Unit tests for recording-daemon lifecycle notify event helpers. */ +#include "../recording-daemon/notify_events.h" +#include "../recording-daemon/notify_events.c" + +/* Provide the global used by notify_event_enabled(). */ +unsigned int notify_events_mask = NOTIFY_MASK_DEFAULT; + +#define err(fmt...) do { \ + fprintf(stderr, fmt); \ + exit(1); \ + } while (0) + +static void expect_parse_ok(const char *csv, unsigned int expect) { + unsigned int mask = 0; + char *bad = NULL; + if (!notify_events_parse(csv, &mask, &bad)) + err("parse failed for '%s' (bad token '%s')\n", csv ? csv : "(null)", + bad ? bad : "?"); + if (mask != expect) + err("parse '%s': mask 0x%x != expected 0x%x\n", csv ? csv : "(null)", mask, expect); + g_free(bad); +} + +static void expect_parse_fail(const char *csv, const char *expect_tok) { + unsigned int mask = 0; + char *bad = NULL; + if (notify_events_parse(csv, &mask, &bad)) + err("parse should fail for '%s'\n", csv); + if (expect_tok && (!bad || strcmp(bad, expect_tok) != 0)) + err("parse '%s': expected bad token '%s', got '%s'\n", + csv, expect_tok, bad ? bad : "(null)"); + g_free(bad); +} + +int main(void) { + /* Default / empty => finished only */ + expect_parse_ok(NULL, NOTIFY_MASK_DEFAULT); + expect_parse_ok("", NOTIFY_MASK_DEFAULT); + expect_parse_ok(" ", NOTIFY_MASK_DEFAULT); + expect_parse_ok("finished", NOTIFY_MASK_FINISHED); + + /* Single tokens */ + expect_parse_ok("opened", NOTIFY_MASK_OPENED); + expect_parse_ok("started", NOTIFY_MASK_STARTED); + expect_parse_ok("discarded", NOTIFY_MASK_DISCARDED); + expect_parse_ok("failed", NOTIFY_MASK_FAILED); + + /* CSV combinations */ + expect_parse_ok("opened,started,finished", + NOTIFY_MASK_OPENED | NOTIFY_MASK_STARTED | NOTIFY_MASK_FINISHED); + expect_parse_ok("opened, started, finished, discarded, failed", + NOTIFY_MASK_STREAM_ALL); + expect_parse_ok("all", NOTIFY_MASK_ALL); + + /* Call-level tokens */ + expect_parse_ok("call-started,call-finished", + NOTIFY_MASK_CALL_STARTED | NOTIFY_MASK_CALL_FINISHED); + + /* Unknown token */ + expect_parse_fail("opened,bogus,finished", "bogus"); + expect_parse_fail("nope", "nope"); + + /* Name / status mapping */ + if (strcmp(notify_event_name(NOTIFY_EVT_FILE_OPENED), "recording_file_opened")) + err("bad name for FILE_OPENED\n"); + if (strcmp(notify_event_name(NOTIFY_EVT_STARTED), "recording_started")) + err("bad name for STARTED\n"); + if (strcmp(notify_event_name(NOTIFY_EVT_FINISHED), "recording_finished")) + err("bad name for FINISHED\n"); + if (strcmp(notify_event_name(NOTIFY_EVT_DISCARDED), "recording_discarded")) + err("bad name for DISCARDED\n"); + if (strcmp(notify_event_name(NOTIFY_EVT_FAILED), "recording_failed")) + err("bad name for FAILED\n"); + if (strcmp(notify_event_status(NOTIFY_EVT_FILE_OPENED), "opened")) + err("bad status for FILE_OPENED\n"); + if (strcmp(notify_event_status(NOTIFY_EVT_FINISHED), "finished")) + err("bad status for FINISHED\n"); + if (strcmp(notify_event_name((enum notify_event) 99), "unknown")) + err("bad name for invalid event\n"); + + /* Enabled mask checks */ + notify_events_mask = NOTIFY_MASK_FINISHED; + if (!notify_event_enabled(NOTIFY_EVT_FINISHED)) + err("finished should be enabled\n"); + if (notify_event_enabled(NOTIFY_EVT_FILE_OPENED)) + err("opened should be disabled under default mask\n"); + if (notify_event_enabled(NOTIFY_EVT_STARTED)) + err("started should be disabled under default mask\n"); + + notify_events_mask = NOTIFY_MASK_OPENED | NOTIFY_MASK_STARTED | NOTIFY_MASK_FINISHED; + if (!notify_event_enabled(NOTIFY_EVT_FILE_OPENED) + || !notify_event_enabled(NOTIFY_EVT_STARTED) + || !notify_event_enabled(NOTIFY_EVT_FINISHED)) + err("expected opened/started/finished enabled\n"); + if (notify_event_enabled(NOTIFY_EVT_DISCARDED)) + err("discarded should not be enabled\n"); + + /* Call-level name/status mapping (Phase 2) */ + if (strcmp(notify_event_name(NOTIFY_EVT_CALL_STARTED), "call_recording_started")) + err("bad name for CALL_STARTED\n"); + if (strcmp(notify_event_name(NOTIFY_EVT_CALL_FINISHED), "call_recording_finished")) + err("bad name for CALL_FINISHED\n"); + if (strcmp(notify_event_name(NOTIFY_EVT_CALL_DISCARDED), "call_recording_discarded")) + err("bad name for CALL_DISCARDED\n"); + if (strcmp(notify_event_status(NOTIFY_EVT_CALL_STARTED), "call-started")) + err("bad status for CALL_STARTED\n"); + if (strcmp(notify_event_status(NOTIFY_EVT_CALL_FINISHED), "call-finished")) + err("bad status for CALL_FINISHED\n"); + if (strcmp(notify_event_status(NOTIFY_EVT_CALL_DISCARDED), "call-discarded")) + err("bad status for CALL_DISCARDED\n"); + + /* Call-level enabled under all */ + notify_events_mask = NOTIFY_MASK_ALL; + if (!notify_event_enabled(NOTIFY_EVT_CALL_STARTED) + || !notify_event_enabled(NOTIFY_EVT_CALL_FINISHED) + || !notify_event_enabled(NOTIFY_EVT_CALL_DISCARDED)) + err("call events should be enabled under all\n"); + + /* call-discarded alone */ + expect_parse_ok("call-discarded", NOTIFY_MASK_CALL_DISCARDED); + + /* Terminal classification (Phase 3) */ + if (!notify_event_is_terminal(NOTIFY_EVT_FINISHED) + || !notify_event_is_terminal(NOTIFY_EVT_DISCARDED) + || !notify_event_is_terminal(NOTIFY_EVT_FAILED) + || !notify_event_is_terminal(NOTIFY_EVT_CALL_FINISHED) + || !notify_event_is_terminal(NOTIFY_EVT_CALL_DISCARDED)) + err("terminal events misclassified\n"); + if (notify_event_is_terminal(NOTIFY_EVT_FILE_OPENED) + || notify_event_is_terminal(NOTIFY_EVT_STARTED) + || notify_event_is_terminal(NOTIFY_EVT_CALL_STARTED)) + err("non-terminal events misclassified\n"); + + /* Command format parse (Phase 2) */ + { + enum notify_command_format fmt; + if (!notify_command_format_parse(NULL, &fmt) || fmt != NOTIFY_CMD_LEGACY) + err("null format should be legacy\n"); + if (!notify_command_format_parse("legacy", &fmt) || fmt != NOTIFY_CMD_LEGACY) + err("legacy parse failed\n"); + if (!notify_command_format_parse("extended", &fmt) || fmt != NOTIFY_CMD_EXTENDED) + err("extended parse failed\n"); + if (!notify_command_format_parse("json-env", &fmt) || fmt != NOTIFY_CMD_JSON_ENV) + err("json-env parse failed\n"); + if (notify_command_format_parse("bogus", &fmt)) + err("bogus format should fail\n"); + } + + printf("ok\n"); + return 0; +} From d30d1baf1b0f6ec403e41fa8cd8916c63803a72a Mon Sep 17 00:00:00 2001 From: Mansoor Khan M Date: Sat, 18 Jul 2026 23:51:17 +0530 Subject: [PATCH 2/4] recording-daemon: keep notify thread pool for S3/GCS/DB notify_setup must not require notify-uri/notify-command: the same thread pool backs S3, GCS and DB notif_action_t work. Also leave object_name to action cleanup to avoid double-free on those paths. --- recording-daemon/notify.c | 20 +++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/recording-daemon/notify.c b/recording-daemon/notify.c index c4c193285b..9afa66f1c0 100644 --- a/recording-daemon/notify.c +++ b/recording-daemon/notify.c @@ -59,6 +59,8 @@ static bool nont_inflight_try_inc(void) { static void req_free_fields(notif_req_t *req) { if (!req) return; + /* Lifecycle snapshot strings only. Action-owned fields (headers/argv/ + * content/object_name/content_sha256) are released by action->cleanup. */ g_free(req->call_id); g_free(req->file_name); g_free(req->file_format); @@ -72,7 +74,6 @@ static void req_free_fields(notif_req_t *req) { g_free(req->error_code); g_free(req->error_message); g_free(req->json_body); - g_free(req->object_name); } static bool do_notify_http(notif_req_t *req) { @@ -242,16 +243,25 @@ static int notify_req_cmp(const void *A, const void *B) { } void notify_setup(void) { - if ((!notify_uri && !notify_command) || notify_threads <= 0) + /* + * The thread pool is shared by HTTP/command notify and by other + * notif_action_t users (S3, GCS, DB). Do not gate on notify_uri / + * notify_command or those storage actions silently stop working. + */ + if (notify_threads <= 0) return; + notify_threadpool = g_thread_pool_new(do_notify, NULL, notify_threads, false, NULL); + notify_timers = g_tree_new(notify_req_cmp); int ret = pthread_create(¬ify_waiter, NULL, notify_timer, NULL); if (ret) ilog(LOG_ERR, "Failed to launch thread for HTTP notification"); - ilog(LOG_INFO, "Recording notify enabled (events mask=0x%x json=%s cmd_fmt=%d queue_limit=%d)", - notify_events_mask, notify_json ? "true" : "false", - notify_command_format, notify_queue_limit); + + if (notify_uri || notify_command) + ilog(LOG_INFO, "Recording notify enabled (events mask=0x%x json=%s cmd_fmt=%d queue_limit=%d)", + notify_events_mask, notify_json ? "true" : "false", + notify_command_format, notify_queue_limit); } void notify_cleanup(void) { From 8f9ea7d22167e0e587497eb6efc30f3a48780325 Mon Sep 17 00:00:00 2001 From: Mansoor Khan M Date: Mon, 20 Jul 2026 11:14:31 +0530 Subject: [PATCH 3/4] notify: restore debug logs/comments and avoid reformat-only changes --- recording-daemon/notify.c | 36 +++++++++++++++++++++++++++++++----- 1 file changed, 31 insertions(+), 5 deletions(-) diff --git a/recording-daemon/notify.c b/recording-daemon/notify.c index 9afa66f1c0..2b4c4f33d9 100644 --- a/recording-daemon/notify.c +++ b/recording-daemon/notify.c @@ -93,6 +93,7 @@ static bool do_notify_http(notif_req_t *req) { if (!c) goto fail; + /* POST vs GET */ bool do_post = notify_post || req->json_body || req->content; if (do_post) { err = "setting CURLOPT_POST"; @@ -138,6 +139,7 @@ static bool do_notify_http(notif_req_t *req) { if (code < 200 || code >= 300) goto fail; + /* success */ ilog(LOG_NOTICE, "HTTP notification (%s) for '%s%s%s' was successful", notify_event_name(req->event), FMT_M(req->name)); return true; @@ -156,6 +158,7 @@ static void failed_http(notif_req_t *req) { } static bool do_notify_command(notif_req_t *req) { + /* legacy callers may leave argv unset */ if (!req->argv) return true; ilog(LOG_DEBUG, "Executing notification command (%s) for '%s%s%s'", @@ -178,6 +181,7 @@ static void do_notify(void *p, void *u) { if (!ok) { if (notify_retries >= 0 && req->retries < (unsigned int) notify_retries) { + /* schedule retry */ req->retries++; metric_inc(¬ify_metric_retry); ilog(LOG_INFO, "Failure while sending notification (%s) for '%s%s%s': " @@ -191,7 +195,7 @@ static void do_notify(void *p, void *u) { g_tree_insert(notify_timers, req, req); pthread_cond_signal(&timer_cond); pthread_mutex_unlock(&timer_lock); - return; + return; // skip cleanup } pthread_mutex_unlock(&timer_lock); } @@ -213,20 +217,36 @@ static void do_notify(void *p, void *u) { static void *notify_timer(void *p) { pthread_mutex_lock(&timer_lock); + + // notify_timers being NULL acts as our shutdown flag while (notify_timers) { + ilog(LOG_DEBUG, "Notification timer thread looping"); + + // grab first entry in list, check retry time, sleep if it's in the future + notif_req_t *first = rtpe_g_tree_first(notify_timers); if (!first) { + ilog(LOG_DEBUG, "No scheduled notification retries, sleeping"); pthread_cond_wait(&timer_cond, &timer_lock); continue; } int64_t now = now_us(); if (now < first->retry_time) { + ilog(LOG_DEBUG, "Sleeping until next scheduled notification retry in %" PRId64 " seconds", + (first->retry_time - now) / 1000000L); cond_timedwait(&timer_cond, &timer_lock, first->retry_time); continue; } + + // first entry is ready to run + g_tree_remove(notify_timers, first); + ilog(LOG_DEBUG, "Notification retry for '%s%s%s' is scheduled now", FMT_M(first->name)); g_thread_pool_push(notify_threadpool, first, NULL); } + + // clean up + pthread_mutex_unlock(&timer_lock); pthread_mutex_destroy(&timer_lock); pthread_cond_destroy(&timer_cond); @@ -235,10 +255,15 @@ static void *notify_timer(void *p) { static int notify_req_cmp(const void *A, const void *B) { const notif_req_t *a = A, *b = B; - if (a->retry_time < b->retry_time) return -1; - if (a->retry_time > b->retry_time) return 1; - if (a < b) return -1; - if (a > b) return 1; + + if (a->retry_time < b->retry_time) + return -1; + if (a->retry_time > b->retry_time) + return 1; + if (a < b) + return -1; + if (a > b) + return 1; return 0; } @@ -266,6 +291,7 @@ void notify_setup(void) { void notify_cleanup(void) { if (notify_waiter && notify_timers) { + // get lock, free GTree, signal thread to shut down pthread_mutex_lock(&timer_lock); if (notify_timers) { while (true) { From a1ae952212aab1fa2ba951012fc1f084f777acf9 Mon Sep 17 00:00:00 2001 From: Mansoor Khan M Date: Mon, 20 Jul 2026 12:18:05 +0530 Subject: [PATCH 4/4] style: tidy notify.c formatting and clarify ownership comment --- recording-daemon/notify.c | 25 ++++++++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/recording-daemon/notify.c b/recording-daemon/notify.c index 2b4c4f33d9..9311aa7e6d 100644 --- a/recording-daemon/notify.c +++ b/recording-daemon/notify.c @@ -59,8 +59,14 @@ static bool nont_inflight_try_inc(void) { static void req_free_fields(notif_req_t *req) { if (!req) return; - /* Lifecycle snapshot strings only. Action-owned fields (headers/argv/ - * content/object_name/content_sha256) are released by action->cleanup. */ + /* + * IMPORTANT: Only free the snapshot strings that notify.c owns. + * + * Fields that are created and managed by specific actions (HTTP, command, + * S3, GCS, ...) such as headers / argv / content / object_name / + * content_sha256 are released in action->cleanup(). If we also free them + * here we will double-free and crash. + */ g_free(req->call_id); g_free(req->file_name); g_free(req->file_format); @@ -117,10 +123,13 @@ static bool do_notify_http(notif_req_t *req) { curl_mimepart *part; mime = curl_mime_init(c); part = curl_mime_addpart(mime); + if ((ret = curl_mime_name(part, "ngfile")) != CURLE_OK) goto fail; + if ((ret = curl_mime_data(part, req->content->s->str, req->content->s->len)) != CURLE_OK) goto fail; + if ((ret = curl_easy_setopt(c, CURLOPT_MIMEPOST, mime)) != CURLE_OK) goto fail; } @@ -149,6 +158,7 @@ static bool do_notify_http(notif_req_t *req) { "Error while %s: %s", notify_event_name(req->event), FMT_M(req->name), err, curl_easy_strerror(ret)); + return false; } @@ -167,16 +177,19 @@ static bool do_notify_command(notif_req_t *req) { bool success = g_spawn_sync(NULL, req->argv, req->envp, G_SPAWN_SEARCH_PATH | G_SPAWN_STDOUT_TO_DEV_NULL | G_SPAWN_STDERR_TO_DEV_NULL, NULL, NULL, NULL, NULL, NULL, &err); + if (!success) { ilog(LOG_ERR, "Failed to execute notification command (%s) for '%s%s%s': %s", notify_event_name(req->event), FMT_M(req->name), err->message); g_error_free(err); } + return success; } static void do_notify(void *p, void *u) { notif_req_t *req = p; + bool ok = req->action->perform(req); if (!ok) { @@ -188,8 +201,10 @@ static void do_notify(void *p, void *u) { "Will retry in %" PRId64 " seconds (#%u)", notify_event_name(req->event), FMT_M(req->name), req->falloff_us / 1000000L, req->retries); + req->retry_time = now_us() + req->falloff_us; req->falloff_us *= 2; + pthread_mutex_lock(&timer_lock); if (notify_timers) { g_tree_insert(notify_timers, req, req); @@ -200,8 +215,12 @@ static void do_notify(void *p, void *u) { pthread_mutex_unlock(&timer_lock); } metric_inc(¬ify_metric_giveup); + ilog(LOG_ERR, "Failure while sending notification (%s) for '%s%s%s' after %u retries. " - "Giving up", notify_event_name(req->event), FMT_M(req->name), req->retries); + "Giving up", + notify_event_name(req->event), FMT_M(req->name), + req->retries); + if (req->action->failed) req->action->failed(req); }