cdr / cel: Use event time at event creation instead of processing.

When updating times on CDR or CEL records using the time at which
it is done can result in times being incorrect if the system is
heavily loaded and stasis message processing is delayed.

This change instead makes it so CDR and CEL use the time at which
the stasis messages that drive the systems are created. This allows
them to be backed up while still producing correct records.

ASTERISK-28498

Change-Id: I6829227e67aefa318efe5e183a94d4a1b4e8500a

diff --git a/include/asterisk/cel.h b/include/asterisk/cel.h
index fed2085f4e6f4e946e5f2339f0482f64a3830f84..c0b8848dee98958cd6d0cad960a2ab956c0ab838 100644 (file)
--- a/include/asterisk/cel.h
+++ b/include/asterisk/cel.h
@@ -263,6 +263,28 @@ struct ast_event *ast_cel_create_event(struct ast_channel_snapshot *snapshot,
                enum ast_cel_event_type event_type, const char *userdefevname,
                struct ast_json *extra, const char *peer_str);
 
+/*!
+ * \brief Allocate and populate a CEL event structure
+ *
+ * \param snapshot An ast_channel_snapshot of the primary channel associated
+ *        with this channel event.
+ * \param event_type The type of call event being reported.
+ * \param event_time The time at which the event occurred.
+ * \param userdefevname Custom name for the call event. (optional)
+ * \param extra An event-specific opaque JSON blob to be rendered and placed
+ *        in the "CEL_EXTRA" information element of the call event. (optional)
+ * \param peer_str A list of comma-separated peer channel names. (optional)
+ *
+ * \since 13.29.0
+ * \since 16.6.0
+ *
+ * \retval The created ast_event structure
+ * \retval NULL on failure
+ */
+struct ast_event *ast_cel_create_event_with_time(struct ast_channel_snapshot *snapshot,
+               enum ast_cel_event_type event_type, const struct timeval *event_time,
+               const char *userdefevname, struct ast_json *extra, const char *peer_str);
+
 /*!
  * \brief CEL backend callback
  */

diff --git a/main/cdr.c b/main/cdr.c
index f8f038c138b386c3f10c2eb7ee396d470364f8b3..2c3ca9762e8013be5a0d6abd7fe431807af9e19b 100644 (file)
--- a/main/cdr.c
+++ b/main/cdr.c
@@ -718,6 +718,7 @@ struct cdr_object {
        struct timeval start;                   /*!< When this CDR was created */
        struct timeval answer;                  /*!< Either when the channel was answered, or when the path between channels was established */
        struct timeval end;                     /*!< When this CDR was finalized */
+       struct timeval lastevent;               /*!< The time at which the last event was created regarding this CDR */
        unsigned int sequence;                  /*!< A monotonically increasing number for each CDR */
        struct ast_flags flags;                 /*!< Flags on the CDR */
        AST_DECLARE_STRING_FIELDS(
@@ -1035,7 +1036,7 @@ static void cdr_object_dtor(void *obj)
  * This implicitly sets the state of the newly created CDR to the Single state
  * (\ref single_state_fn_table)
  */
-static struct cdr_object *cdr_object_alloc(struct ast_channel_snapshot *chan)
+static struct cdr_object *cdr_object_alloc(struct ast_channel_snapshot *chan, const struct timeval *event_time)
 {
        struct cdr_object *cdr;
 
@@ -1055,6 +1056,7 @@ static struct cdr_object *cdr_object_alloc(struct ast_channel_snapshot *chan)
        ast_string_field_set(cdr, linkedid, chan->peer->linkedid);
        cdr->disposition = AST_CDR_NULL;
        cdr->sequence = ast_atomic_fetchadd_int(&global_cdr_sequence, +1);
+       cdr->lastevent = *event_time;
 
        cdr->party_a.snapshot = chan;
        ao2_t_ref(cdr->party_a.snapshot, +1, "bump snapshot during CDR creation");
@@ -1070,14 +1072,14 @@ static struct cdr_object *cdr_object_alloc(struct ast_channel_snapshot *chan)
  * \brief Create a new \ref cdr_object and append it to an existing chain
  * \param cdr The \ref cdr_object to append to
  */
-static struct cdr_object *cdr_object_create_and_append(struct cdr_object *cdr)
+static struct cdr_object *cdr_object_create_and_append(struct cdr_object *cdr, const struct timeval *event_time)
 {
        struct cdr_object *new_cdr;
        struct cdr_object *it_cdr;
        struct cdr_object *cdr_last;
 
        cdr_last = cdr->last;
-       new_cdr = cdr_object_alloc(cdr_last->party_a.snapshot);
+       new_cdr = cdr_object_alloc(cdr_last->party_a.snapshot, event_time);
        if (!new_cdr) {
                return NULL;
        }
@@ -1440,7 +1442,7 @@ static void cdr_object_finalize(struct cdr_object *cdr)
        if (!ast_tvzero(cdr->end)) {
                return;
        }
-       cdr->end = ast_tvnow();
+       cdr->end = cdr->lastevent;
 
        if (cdr->disposition == AST_CDR_NULL) {
                if (!ast_tvzero(cdr->answer)) {
@@ -1490,7 +1492,7 @@ static void cdr_object_check_party_a_hangup(struct cdr_object *cdr)
 static void cdr_object_check_party_a_answer(struct cdr_object *cdr)
 {
        if (cdr->party_a.snapshot->state == AST_STATE_UP && ast_tvzero(cdr->answer)) {
-               cdr->answer = ast_tvnow();
+               cdr->answer = cdr->lastevent;
                /* tv_usec is suseconds_t, which could be int or long */
                CDR_DEBUG("%p - Set answered time to %ld.%06ld\n", cdr,
                        (long)cdr->answer.tv_sec,
@@ -1634,7 +1636,7 @@ static int base_process_parked_channel(struct cdr_object *cdr, struct ast_parked
 
 static void single_state_init_function(struct cdr_object *cdr)
 {
-       cdr->start = ast_tvnow();
+       cdr->start = cdr->lastevent;
        cdr_object_check_party_a_answer(cdr);
 }
 
@@ -2149,6 +2151,7 @@ static void handle_dial_message(void *data, struct stasis_subscription *sub, str
 
        ao2_lock(cdr);
        for (it_cdr = cdr; it_cdr; it_cdr = it_cdr->next) {
+               it_cdr->lastevent = *stasis_message_timestamp(message);
                if (ast_strlen_zero(dial_status)) {
                        if (!it_cdr->fn_table->process_dial_begin) {
                                continue;
@@ -2179,7 +2182,7 @@ static void handle_dial_message(void *data, struct stasis_subscription *sub, str
        if (res && ast_strlen_zero(dial_status)) {
                struct cdr_object *new_cdr;
 
-               new_cdr = cdr_object_create_and_append(cdr);
+               new_cdr = cdr_object_create_and_append(cdr, stasis_message_timestamp(message));
                if (new_cdr) {
                        new_cdr->fn_table->process_dial_begin(new_cdr, caller, peer);
                }
@@ -2281,7 +2284,7 @@ static void handle_channel_snapshot_update_message(void *data, struct stasis_sub
        }
 
        if (update->new_snapshot && !update->old_snapshot) {
-               cdr = cdr_object_alloc(update->new_snapshot);
+               cdr = cdr_object_alloc(update->new_snapshot, stasis_message_timestamp(message));
                if (!cdr) {
                        return;
                }
@@ -2300,6 +2303,7 @@ static void handle_channel_snapshot_update_message(void *data, struct stasis_sub
 
                ao2_lock(cdr);
                for (it_cdr = cdr; it_cdr; it_cdr = it_cdr->next) {
+                       it_cdr->lastevent = *stasis_message_timestamp(message);
                        if (!it_cdr->fn_table->process_party_a) {
                                continue;
                        }
@@ -2309,7 +2313,7 @@ static void handle_channel_snapshot_update_message(void *data, struct stasis_sub
                        /* We're not hung up and we have a new snapshot - we need a new CDR */
                        struct cdr_object *new_cdr;
 
-                       new_cdr = cdr_object_create_and_append(cdr);
+                       new_cdr = cdr_object_create_and_append(cdr, stasis_message_timestamp(message));
                        if (new_cdr) {
                                new_cdr->fn_table->process_party_a(new_cdr, update->new_snapshot);
                        }
@@ -2321,6 +2325,7 @@ static void handle_channel_snapshot_update_message(void *data, struct stasis_sub
                ao2_lock(cdr);
                CDR_DEBUG("%p - Beginning finalize/dispatch for %s\n", cdr, update->old_snapshot->base->name);
                for (it_cdr = cdr; it_cdr; it_cdr = it_cdr->next) {
+                       it_cdr->lastevent = *stasis_message_timestamp(message);
                        cdr_object_finalize(it_cdr);
                }
                cdr_object_dispatch(cdr);
@@ -2429,6 +2434,7 @@ static void handle_bridge_leave_message(void *data, struct stasis_subscription *
        /* Party A */
        ao2_lock(cdr);
        for (it_cdr = cdr; it_cdr; it_cdr = it_cdr->next) {
+               it_cdr->lastevent = *stasis_message_timestamp(message);
                if (!it_cdr->fn_table->process_bridge_leave) {
                        continue;
                }
@@ -2463,7 +2469,7 @@ static void bridge_candidate_add_to_cdr(struct cdr_object *cdr,
 {
        struct cdr_object *new_cdr;
 
-       new_cdr = cdr_object_create_and_append(cdr);
+       new_cdr = cdr_object_create_and_append(cdr, &cdr->lastevent);
        if (!new_cdr) {
                return;
        }
@@ -2574,7 +2580,8 @@ static void handle_bridge_pairings(struct cdr_object *cdr, struct ast_bridge_sna
  */
 static void handle_parking_bridge_enter_message(struct cdr_object *cdr,
                struct ast_bridge_snapshot *bridge,
-               struct ast_channel_snapshot *channel)
+               struct ast_channel_snapshot *channel,
+               const struct timeval *event_time)
 {
        int res = 1;
        struct cdr_object *it_cdr;
@@ -2583,6 +2590,8 @@ static void handle_parking_bridge_enter_message(struct cdr_object *cdr,
        ao2_lock(cdr);
 
        for (it_cdr = cdr; it_cdr; it_cdr = it_cdr->next) {
+               it_cdr->lastevent = *event_time;
+
                if (it_cdr->fn_table->process_parking_bridge_enter) {
                        res &= it_cdr->fn_table->process_parking_bridge_enter(it_cdr, bridge, channel);
                }
@@ -2595,7 +2604,7 @@ static void handle_parking_bridge_enter_message(struct cdr_object *cdr,
 
        if (res) {
                /* No one handled it - we need a new one! */
-               new_cdr = cdr_object_create_and_append(cdr);
+               new_cdr = cdr_object_create_and_append(cdr, event_time);
                if (new_cdr) {
                        /* Let the single state transition us to Parked */
                        cdr_object_transition_state(new_cdr, &single_state_fn_table);
@@ -2612,7 +2621,8 @@ static void handle_parking_bridge_enter_message(struct cdr_object *cdr,
  */
 static void handle_standard_bridge_enter_message(struct cdr_object *cdr,
                struct ast_bridge_snapshot *bridge,
-               struct ast_channel_snapshot *channel)
+               struct ast_channel_snapshot *channel,
+               const struct timeval *event_time)
 {
        enum process_bridge_enter_results result;
        struct cdr_object *it_cdr;
@@ -2623,6 +2633,8 @@ static void handle_standard_bridge_enter_message(struct cdr_object *cdr,
 
 try_again:
        for (it_cdr = cdr; it_cdr; it_cdr = it_cdr->next) {
+               it_cdr->lastevent = *event_time;
+
                if (it_cdr->fn_table->process_party_a) {
                        CDR_DEBUG("%p - Updating Party A %s snapshot\n", it_cdr,
                                channel->base->name);
@@ -2669,7 +2681,7 @@ try_again:
                handle_bridge_pairings(handled_cdr, bridge);
        } else {
                /* Nothing handled it - we need a new one! */
-               new_cdr = cdr_object_create_and_append(cdr);
+               new_cdr = cdr_object_create_and_append(cdr, event_time);
                if (new_cdr) {
                        /* This is guaranteed to succeed: the new CDR is created in the single state
                         * and will be able to handle the bridge enter message
@@ -2717,9 +2729,9 @@ static void handle_bridge_enter_message(void *data, struct stasis_subscription *
        }
 
        if (!strcmp(bridge->subclass, "parking")) {
-               handle_parking_bridge_enter_message(cdr, bridge, channel);
+               handle_parking_bridge_enter_message(cdr, bridge, channel, stasis_message_timestamp(message));
        } else {
-               handle_standard_bridge_enter_message(cdr, bridge, channel);
+               handle_standard_bridge_enter_message(cdr, bridge, channel, stasis_message_timestamp(message));
        }
        ao2_cleanup(cdr);
 }
@@ -2769,6 +2781,7 @@ static void handle_parked_call_message(void *data, struct stasis_subscription *s
        ao2_lock(cdr);
 
        for (it_cdr = cdr; it_cdr; it_cdr = it_cdr->next) {
+               it_cdr->lastevent = *stasis_message_timestamp(message);
                if (it_cdr->fn_table->process_parked_channel) {
                        unhandled &= it_cdr->fn_table->process_parked_channel(it_cdr, payload);
                }
@@ -2778,7 +2791,7 @@ static void handle_parked_call_message(void *data, struct stasis_subscription *s
                /* Nothing handled the messgae - we need a new one! */
                struct cdr_object *new_cdr;
 
-               new_cdr = cdr_object_create_and_append(cdr);
+               new_cdr = cdr_object_create_and_append(cdr, stasis_message_timestamp(message));
                if (new_cdr) {
                        /* As the new CDR is created in the single state, it is guaranteed
                         * to have a function for the parked call message and will handle
@@ -3607,6 +3620,7 @@ int ast_cdr_reset(const char *channel_name, int keep_variables)
                memset(&it_cdr->end, 0, sizeof(it_cdr->end));
                memset(&it_cdr->answer, 0, sizeof(it_cdr->answer));
                it_cdr->start = ast_tvnow();
+               it_cdr->lastevent = it_cdr->start;
                cdr_object_check_party_a_answer(it_cdr);
        }
        ao2_unlock(cdr);
@@ -3628,6 +3642,7 @@ int ast_cdr_fork(const char *channel_name, struct ast_flags *options)
 
        {
                SCOPED_AO2LOCK(lock, cdr);
+               struct timeval now = ast_tvnow();
 
                cdr_obj = cdr->last;
                if (cdr_obj->fn_table == &finalized_state_fn_table) {
@@ -3641,7 +3656,7 @@ int ast_cdr_fork(const char *channel_name, struct ast_flags *options)
                 * copied over automatically as part of the append
                 */
                ast_debug(1, "Forking CDR for channel %s\n", cdr->party_a.snapshot->base->name);
-               new_cdr = cdr_object_create_and_append(cdr);
+               new_cdr = cdr_object_create_and_append(cdr, &now);
                if (!new_cdr) {
                        return -1;
                }
@@ -3671,6 +3686,7 @@ int ast_cdr_fork(const char *channel_name, struct ast_flags *options)
                }
                new_cdr->start = cdr_obj->start;
                new_cdr->answer = cdr_obj->answer;
+               new_cdr->lastevent = ast_tvnow();
 
                /* Modify the times based on the flags passed in */
                if (ast_test_flag(options, AST_CDR_FLAG_SET_ANSWER)

diff --git a/main/cel.c b/main/cel.c
index 1e77d2589558b4fd033b3f6eb725637ec7f45e77..00020ea13e1ac172a80d68550ff1c14eae70acd5 100644 (file)
--- a/main/cel.c
+++ b/main/cel.c
@@ -519,14 +519,23 @@ struct ast_event *ast_cel_create_event(struct ast_channel_snapshot *snapshot,
                struct ast_json *extra, const char *peer)
 {
        struct timeval eventtime = ast_tvnow();
+
+       return ast_cel_create_event_with_time(snapshot, event_type, &eventtime,
+               userdefevname, extra, peer);
+}
+
+struct ast_event *ast_cel_create_event_with_time(struct ast_channel_snapshot *snapshot,
+               enum ast_cel_event_type event_type, const struct timeval *event_time,
+               const char *userdefevname, struct ast_json *extra, const char *peer)
+{
        RAII_VAR(char *, extra_txt, NULL, ast_json_free);
        if (extra) {
                extra_txt = ast_json_dump_string(extra);
        }
        return ast_event_new(AST_EVENT_CEL,
                AST_EVENT_IE_CEL_EVENT_TYPE, AST_EVENT_IE_PLTYPE_UINT, event_type,
-               AST_EVENT_IE_CEL_EVENT_TIME, AST_EVENT_IE_PLTYPE_UINT, eventtime.tv_sec,
-               AST_EVENT_IE_CEL_EVENT_TIME_USEC, AST_EVENT_IE_PLTYPE_UINT, eventtime.tv_usec,
+               AST_EVENT_IE_CEL_EVENT_TIME, AST_EVENT_IE_PLTYPE_UINT, event_time->tv_sec,
+               AST_EVENT_IE_CEL_EVENT_TIME_USEC, AST_EVENT_IE_PLTYPE_UINT, event_time->tv_usec,
                AST_EVENT_IE_CEL_USEREVENT_NAME, AST_EVENT_IE_PLTYPE_STR, S_OR(userdefevname, ""),
                AST_EVENT_IE_CEL_CIDNAME, AST_EVENT_IE_PLTYPE_STR, snapshot->caller->name,
                AST_EVENT_IE_CEL_CIDNUM, AST_EVENT_IE_PLTYPE_STR, snapshot->caller->number,
@@ -558,8 +567,9 @@ static int cel_backend_send_cb(void *obj, void *arg, int flags)
 }
 
 static int cel_report_event(struct ast_channel_snapshot *snapshot,
-               enum ast_cel_event_type event_type, const char *userdefevname,
-               struct ast_json *extra, const char *peer_str)
+               enum ast_cel_event_type event_type, const struct timeval *event_time,
+               const char *userdefevname, struct ast_json *extra,
+               const char *peer_str)
 {
        struct ast_event *ev;
        RAII_VAR(struct cel_config *, cfg, ao2_global_obj_ref(cel_configs), ao2_cleanup);
@@ -587,7 +597,7 @@ static int cel_report_event(struct ast_channel_snapshot *snapshot,
                return 0;
        }
 
-       ev = ast_cel_create_event(snapshot, event_type, userdefevname, extra, peer_str);
+       ev = ast_cel_create_event_with_time(snapshot, event_type, event_time, userdefevname, extra, peer_str);
        if (!ev) {
                return -1;
        }
@@ -601,7 +611,7 @@ static int cel_report_event(struct ast_channel_snapshot *snapshot,
 
 /* called whenever a channel is destroyed or a linkedid is changed to
  * potentially emit a CEL_LINKEDID_END event */
-static void check_retire_linkedid(struct ast_channel_snapshot *snapshot)
+static void check_retire_linkedid(struct ast_channel_snapshot *snapshot, const struct timeval *event_time)
 {
        RAII_VAR(struct ao2_container *, linkedids, ao2_global_obj_ref(cel_linkedids), ao2_cleanup);
        struct cel_linkedid *lid;
@@ -632,7 +642,7 @@ static void check_retire_linkedid(struct ast_channel_snapshot *snapshot)
                ao2_unlink(linkedids, lid);
                ao2_unlock(linkedids);
 
-               cel_report_event(snapshot, AST_CEL_LINKEDID_END, NULL, NULL, NULL);
+               cel_report_event(snapshot, AST_CEL_LINKEDID_END, event_time, NULL, NULL, NULL);
        } else {
                ao2_unlock(linkedids);
        }
@@ -852,7 +862,8 @@ int ast_cel_fill_record(const struct ast_event *e, struct ast_cel_event_record *
 /*! \brief Typedef for callbacks that get called on channel snapshot updates */
 typedef void (*cel_channel_snapshot_monitor)(
        struct ast_channel_snapshot *old_snapshot,
-       struct ast_channel_snapshot *new_snapshot);
+       struct ast_channel_snapshot *new_snapshot,
+       const struct timeval *event_time);
 
 static struct cel_dialstatus *get_dialstatus(const char *uniqueid)
 {
@@ -884,12 +895,13 @@ static const char *get_blob_variable(struct ast_multi_channel_blob *blob, const
 /*! \brief Handle channel state changes */
 static void cel_channel_state_change(
        struct ast_channel_snapshot *old_snapshot,
-       struct ast_channel_snapshot *new_snapshot)
+       struct ast_channel_snapshot *new_snapshot,
+       const struct timeval *event_time)
 {
        int is_hungup, was_hungup;
 
        if (!old_snapshot) {
-               cel_report_event(new_snapshot, AST_CEL_CHANNEL_START, NULL, NULL, NULL);
+               cel_report_event(new_snapshot, AST_CEL_CHANNEL_START, event_time, NULL, NULL, NULL);
                return;
        }
 
@@ -904,26 +916,27 @@ static void cel_channel_state_change(
                        "hangupcause", new_snapshot->hangup->cause,
                        "hangupsource", new_snapshot->hangup->source,
                        "dialstatus", dialstatus ? dialstatus->dialstatus : "");
-               cel_report_event(new_snapshot, AST_CEL_HANGUP, NULL, extra, NULL);
+               cel_report_event(new_snapshot, AST_CEL_HANGUP, event_time, NULL, extra, NULL);
                ast_json_unref(extra);
                ao2_cleanup(dialstatus);
 
-               cel_report_event(new_snapshot, AST_CEL_CHANNEL_END, NULL, NULL, NULL);
+               cel_report_event(new_snapshot, AST_CEL_CHANNEL_END, event_time, NULL, NULL, NULL);
                if (ast_cel_track_event(AST_CEL_LINKEDID_END)) {
-                       check_retire_linkedid(new_snapshot);
+                       check_retire_linkedid(new_snapshot, event_time);
                }
                return;
        }
 
        if (old_snapshot->state != new_snapshot->state && new_snapshot->state == AST_STATE_UP) {
-               cel_report_event(new_snapshot, AST_CEL_ANSWER, NULL, NULL, NULL);
+               cel_report_event(new_snapshot, AST_CEL_ANSWER, event_time, NULL, NULL, NULL);
                return;
        }
 }
 
 static void cel_channel_linkedid_change(
        struct ast_channel_snapshot *old_snapshot,
-       struct ast_channel_snapshot *new_snapshot)
+       struct ast_channel_snapshot *new_snapshot,
+       const struct timeval *event_time)
 {
        if (!old_snapshot) {
                return;
@@ -935,13 +948,14 @@ static void cel_channel_linkedid_change(
        if (ast_cel_track_event(AST_CEL_LINKEDID_END)
                && strcmp(old_snapshot->peer->linkedid, new_snapshot->peer->linkedid)) {
                cel_linkedid_ref(new_snapshot->peer->linkedid);
-               check_retire_linkedid(old_snapshot);
+               check_retire_linkedid(old_snapshot, event_time);
        }
 }
 
 static void cel_channel_app_change(
        struct ast_channel_snapshot *old_snapshot,
-       struct ast_channel_snapshot *new_snapshot)
+       struct ast_channel_snapshot *new_snapshot,
+       const struct timeval *event_time)
 {
        if (old_snapshot && !strcmp(old_snapshot->dialplan->appl, new_snapshot->dialplan->appl)) {
                return;
@@ -949,12 +963,12 @@ static void cel_channel_app_change(
 
        /* old snapshot has an application, end it */
        if (old_snapshot && !ast_strlen_zero(old_snapshot->dialplan->appl)) {
-               cel_report_event(old_snapshot, AST_CEL_APP_END, NULL, NULL, NULL);
+               cel_report_event(old_snapshot, AST_CEL_APP_END, event_time, NULL, NULL, NULL);
        }
 
        /* new snapshot has an application, start it */
        if (!ast_strlen_zero(new_snapshot->dialplan->appl)) {
-               cel_report_event(new_snapshot, AST_CEL_APP_START, NULL, NULL, NULL);
+               cel_report_event(new_snapshot, AST_CEL_APP_START, event_time, NULL, NULL, NULL);
        }
 }
 
@@ -988,7 +1002,7 @@ static void cel_snapshot_update_cb(void *data, struct stasis_subscription *sub,
        }
 
        for (i = 0; i < ARRAY_LEN(cel_channel_monitors); ++i) {
-               cel_channel_monitors[i](update->old_snapshot, update->new_snapshot);
+               cel_channel_monitors[i](update->old_snapshot, update->new_snapshot, stasis_message_timestamp(message));
        }
 }
 
@@ -1056,7 +1070,8 @@ static void cel_bridge_enter_cb(
                return;
        }
 
-       cel_report_event(chan_snapshot, AST_CEL_BRIDGE_ENTER, NULL, extra, ast_str_buffer(peer_str));
+       cel_report_event(chan_snapshot, AST_CEL_BRIDGE_ENTER, stasis_message_timestamp(message),
+               NULL, extra, ast_str_buffer(peer_str));
 }
 
 static void cel_bridge_leave_cb(
@@ -1085,7 +1100,8 @@ static void cel_bridge_leave_cb(
                return;
        }
 
-       cel_report_event(chan_snapshot, AST_CEL_BRIDGE_EXIT, NULL, extra, ast_str_buffer(peer_str));
+       cel_report_event(chan_snapshot, AST_CEL_BRIDGE_EXIT, stasis_message_timestamp(message),
+               NULL, extra, ast_str_buffer(peer_str));
 }
 
 static void cel_parking_cb(
@@ -1102,7 +1118,8 @@ static void cel_parking_cb(
                        "parker_dial_string", parked_payload->parker_dial_string,
                        "parking_lot", parked_payload->parkinglot);
                if (extra) {
-                       cel_report_event(parked_payload->parkee, AST_CEL_PARK_START, NULL, extra, NULL);
+                       cel_report_event(parked_payload->parkee, AST_CEL_PARK_START, stasis_message_timestamp(message),
+                               NULL, extra, NULL);
                }
                return;
        case PARKED_CALL_TIMEOUT:
@@ -1131,7 +1148,8 @@ static void cel_parking_cb(
        }
 
        if (extra) {
-               cel_report_event(parked_payload->parkee, AST_CEL_PARK_END, NULL, extra, NULL);
+               cel_report_event(parked_payload->parkee, AST_CEL_PARK_END, stasis_message_timestamp(message),
+                       NULL, extra, NULL);
        }
 }
 
@@ -1224,7 +1242,8 @@ static void cel_dial_cb(void *data, struct stasis_subscription *sub,
 
                extra = ast_json_pack("{s: s}", "forward", get_blob_variable(blob, "forward"));
                if (extra) {
-                       cel_report_event(snapshot, AST_CEL_FORWARD, NULL, extra, NULL);
+                       cel_report_event(snapshot, AST_CEL_FORWARD, stasis_message_timestamp(message),
+                               NULL, extra, NULL);
                        ast_json_unref(extra);
                }
        }
@@ -1247,7 +1266,8 @@ static void cel_generic_cb(
                {
                        const char *event = ast_json_string_get(ast_json_object_get(event_details, "event"));
                        struct ast_json *extra = ast_json_object_get(event_details, "extra");
-                       cel_report_event(obj->snapshot, event_type, event, extra, NULL);
+                       cel_report_event(obj->snapshot, event_type, stasis_message_timestamp(message),
+                               event, extra, NULL);
                        break;
                }
        default:
@@ -1276,7 +1296,8 @@ static void cel_blind_transfer_cb(
                "transferee_channel_name", transfer_msg->transferee ? transfer_msg->transferee->base->name : "N/A",
                "transferee_channel_uniqueid", transfer_msg->transferee ? transfer_msg->transferee->base->uniqueid  : "N/A");
        if (extra) {
-               cel_report_event(chan_snapshot, AST_CEL_BLINDTRANSFER, NULL, extra, NULL);
+               cel_report_event(chan_snapshot, AST_CEL_BLINDTRANSFER, stasis_message_timestamp(message),
+                       NULL, extra, NULL);
                ast_json_unref(extra);
        }
 }
@@ -1339,7 +1360,8 @@ static void cel_attended_transfer_cb(
                }
                break;
        }
-       cel_report_event(channel1, AST_CEL_ATTENDEDTRANSFER, NULL, extra, NULL);
+       cel_report_event(channel1, AST_CEL_ATTENDEDTRANSFER, stasis_message_timestamp(message),
+               NULL, extra, NULL);
        ast_json_unref(extra);
 }
 
@@ -1363,7 +1385,7 @@ static void cel_pickup_cb(
                return;
        }
 
-       cel_report_event(target, AST_CEL_PICKUP, NULL, extra, NULL);
+       cel_report_event(target, AST_CEL_PICKUP, stasis_message_timestamp(message), NULL, extra, NULL);
        ast_json_unref(extra);
 }
 
@@ -1387,7 +1409,7 @@ static void cel_local_cb(
                return;
        }
 
-       cel_report_event(localone, AST_CEL_LOCAL_OPTIMIZE, NULL, extra, NULL);
+       cel_report_event(localone, AST_CEL_LOCAL_OPTIMIZE, stasis_message_timestamp(message), NULL, extra, NULL);
        ast_json_unref(extra);
 }