2a9423dcfafee2933b44a3b8947fd4dbc364ece6
[lttng-tools.git] / src / bin / lttng-sessiond / notification-thread-events.c
1 /*
2 * Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
3 *
4 * SPDX-License-Identifier: GPL-2.0-only
5 *
6 */
7
8 #include "lttng/action/action.h"
9 #include "lttng/trigger/trigger-internal.h"
10 #define _LGPL_SOURCE
11 #include <urcu.h>
12 #include <urcu/rculfhash.h>
13
14 #include <common/defaults.h>
15 #include <common/error.h>
16 #include <common/futex.h>
17 #include <common/unix.h>
18 #include <common/dynamic-buffer.h>
19 #include <common/hashtable/utils.h>
20 #include <common/sessiond-comm/sessiond-comm.h>
21 #include <common/macros.h>
22 #include <lttng/condition/condition.h>
23 #include <lttng/action/action-internal.h>
24 #include <lttng/notification/notification-internal.h>
25 #include <lttng/condition/condition-internal.h>
26 #include <lttng/condition/buffer-usage-internal.h>
27 #include <lttng/condition/session-consumed-size-internal.h>
28 #include <lttng/condition/session-rotation-internal.h>
29 #include <lttng/notification/channel-internal.h>
30
31 #include <time.h>
32 #include <unistd.h>
33 #include <assert.h>
34 #include <inttypes.h>
35 #include <fcntl.h>
36
37 #include "notification-thread.h"
38 #include "notification-thread-events.h"
39 #include "notification-thread-commands.h"
40 #include "lttng-sessiond.h"
41 #include "kernel.h"
42
43 #define CLIENT_POLL_MASK_IN (LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP)
44 #define CLIENT_POLL_MASK_IN_OUT (CLIENT_POLL_MASK_IN | LPOLLOUT)
45
46 enum lttng_object_type {
47 LTTNG_OBJECT_TYPE_UNKNOWN,
48 LTTNG_OBJECT_TYPE_NONE,
49 LTTNG_OBJECT_TYPE_CHANNEL,
50 LTTNG_OBJECT_TYPE_SESSION,
51 };
52
53 struct lttng_trigger_list_element {
54 /* No ownership of the trigger object is assumed. */
55 struct lttng_trigger *trigger;
56 struct cds_list_head node;
57 };
58
59 struct lttng_channel_trigger_list {
60 struct channel_key channel_key;
61 /* List of struct lttng_trigger_list_element. */
62 struct cds_list_head list;
63 /* Node in the channel_triggers_ht */
64 struct cds_lfht_node channel_triggers_ht_node;
65 /* call_rcu delayed reclaim. */
66 struct rcu_head rcu_node;
67 };
68
69 /*
70 * List of triggers applying to a given session.
71 *
72 * See:
73 * - lttng_session_trigger_list_create()
74 * - lttng_session_trigger_list_build()
75 * - lttng_session_trigger_list_destroy()
76 * - lttng_session_trigger_list_add()
77 */
78 struct lttng_session_trigger_list {
79 /*
80 * Not owned by this; points to the session_info structure's
81 * session name.
82 */
83 const char *session_name;
84 /* List of struct lttng_trigger_list_element. */
85 struct cds_list_head list;
86 /* Node in the session_triggers_ht */
87 struct cds_lfht_node session_triggers_ht_node;
88 /*
89 * Weak reference to the notification system's session triggers
90 * hashtable.
91 *
92 * The session trigger list structure structure is owned by
93 * the session's session_info.
94 *
95 * The session_info is kept alive the the channel_infos holding a
96 * reference to it (reference counting). When those channels are
97 * destroyed (at runtime or on teardown), the reference they hold
98 * to the session_info are released. On destruction of session_info,
99 * session_info_destroy() will remove the list of triggers applying
100 * to this session from the notification system's state.
101 *
102 * This implies that the session_triggers_ht must be destroyed
103 * after the channels.
104 */
105 struct cds_lfht *session_triggers_ht;
106 /* Used for delayed RCU reclaim. */
107 struct rcu_head rcu_node;
108 };
109
110 struct lttng_trigger_ht_element {
111 struct lttng_trigger *trigger;
112 struct cds_lfht_node node;
113 /* call_rcu delayed reclaim. */
114 struct rcu_head rcu_node;
115 };
116
117 struct lttng_condition_list_element {
118 struct lttng_condition *condition;
119 struct cds_list_head node;
120 };
121
122 struct channel_state_sample {
123 struct channel_key key;
124 struct cds_lfht_node channel_state_ht_node;
125 uint64_t highest_usage;
126 uint64_t lowest_usage;
127 uint64_t channel_total_consumed;
128 /* call_rcu delayed reclaim. */
129 struct rcu_head rcu_node;
130 };
131
132 static unsigned long hash_channel_key(struct channel_key *key);
133 static int evaluate_buffer_condition(const struct lttng_condition *condition,
134 struct lttng_evaluation **evaluation,
135 const struct notification_thread_state *state,
136 const struct channel_state_sample *previous_sample,
137 const struct channel_state_sample *latest_sample,
138 uint64_t previous_session_consumed_total,
139 uint64_t latest_session_consumed_total,
140 struct channel_info *channel_info);
141 static
142 int send_evaluation_to_clients(const struct lttng_trigger *trigger,
143 const struct lttng_evaluation *evaluation,
144 struct notification_client_list *client_list,
145 struct notification_thread_state *state,
146 uid_t channel_uid, gid_t channel_gid);
147
148
149 /* session_info API */
150 static
151 void session_info_destroy(void *_data);
152 static
153 void session_info_get(struct session_info *session_info);
154 static
155 void session_info_put(struct session_info *session_info);
156 static
157 struct session_info *session_info_create(const char *name,
158 uid_t uid, gid_t gid,
159 struct lttng_session_trigger_list *trigger_list,
160 struct cds_lfht *sessions_ht);
161 static
162 void session_info_add_channel(struct session_info *session_info,
163 struct channel_info *channel_info);
164 static
165 void session_info_remove_channel(struct session_info *session_info,
166 struct channel_info *channel_info);
167
168 /* lttng_session_trigger_list API */
169 static
170 struct lttng_session_trigger_list *lttng_session_trigger_list_create(
171 const char *session_name,
172 struct cds_lfht *session_triggers_ht);
173 static
174 struct lttng_session_trigger_list *lttng_session_trigger_list_build(
175 const struct notification_thread_state *state,
176 const char *session_name);
177 static
178 void lttng_session_trigger_list_destroy(
179 struct lttng_session_trigger_list *list);
180 static
181 int lttng_session_trigger_list_add(struct lttng_session_trigger_list *list,
182 struct lttng_trigger *trigger);
183
184 static
185 int client_handle_transmission_status(
186 struct notification_client *client,
187 enum client_transmission_status transmission_status,
188 struct notification_thread_state *state);
189
190 static
191 int match_client_socket(struct cds_lfht_node *node, const void *key)
192 {
193 /* This double-cast is intended to supress pointer-to-cast warning. */
194 const int socket = (int) (intptr_t) key;
195 const struct notification_client *client = caa_container_of(node,
196 struct notification_client, client_socket_ht_node);
197
198 return client->socket == socket;
199 }
200
201 static
202 int match_client_id(struct cds_lfht_node *node, const void *key)
203 {
204 /* This double-cast is intended to supress pointer-to-cast warning. */
205 const notification_client_id id = *((notification_client_id *) key);
206 const struct notification_client *client = caa_container_of(
207 node, struct notification_client, client_id_ht_node);
208
209 return client->id == id;
210 }
211
212 static
213 int match_channel_trigger_list(struct cds_lfht_node *node, const void *key)
214 {
215 struct channel_key *channel_key = (struct channel_key *) key;
216 struct lttng_channel_trigger_list *trigger_list;
217
218 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
219 channel_triggers_ht_node);
220
221 return !!((channel_key->key == trigger_list->channel_key.key) &&
222 (channel_key->domain == trigger_list->channel_key.domain));
223 }
224
225 static
226 int match_session_trigger_list(struct cds_lfht_node *node, const void *key)
227 {
228 const char *session_name = (const char *) key;
229 struct lttng_session_trigger_list *trigger_list;
230
231 trigger_list = caa_container_of(node, struct lttng_session_trigger_list,
232 session_triggers_ht_node);
233
234 return !!(strcmp(trigger_list->session_name, session_name) == 0);
235 }
236
237 static
238 int match_channel_state_sample(struct cds_lfht_node *node, const void *key)
239 {
240 struct channel_key *channel_key = (struct channel_key *) key;
241 struct channel_state_sample *sample;
242
243 sample = caa_container_of(node, struct channel_state_sample,
244 channel_state_ht_node);
245
246 return !!((channel_key->key == sample->key.key) &&
247 (channel_key->domain == sample->key.domain));
248 }
249
250 static
251 int match_channel_info(struct cds_lfht_node *node, const void *key)
252 {
253 struct channel_key *channel_key = (struct channel_key *) key;
254 struct channel_info *channel_info;
255
256 channel_info = caa_container_of(node, struct channel_info,
257 channels_ht_node);
258
259 return !!((channel_key->key == channel_info->key.key) &&
260 (channel_key->domain == channel_info->key.domain));
261 }
262
263 static
264 int match_condition(struct cds_lfht_node *node, const void *key)
265 {
266 struct lttng_condition *condition_key = (struct lttng_condition *) key;
267 struct lttng_trigger_ht_element *trigger;
268 struct lttng_condition *condition;
269
270 trigger = caa_container_of(node, struct lttng_trigger_ht_element,
271 node);
272 condition = lttng_trigger_get_condition(trigger->trigger);
273 assert(condition);
274
275 return !!lttng_condition_is_equal(condition_key, condition);
276 }
277
278 static
279 int match_client_list_condition(struct cds_lfht_node *node, const void *key)
280 {
281 struct lttng_condition *condition_key = (struct lttng_condition *) key;
282 struct notification_client_list *client_list;
283 const struct lttng_condition *condition;
284
285 assert(condition_key);
286
287 client_list = caa_container_of(node, struct notification_client_list,
288 notification_trigger_clients_ht_node);
289 condition = lttng_trigger_get_const_condition(client_list->trigger);
290
291 return !!lttng_condition_is_equal(condition_key, condition);
292 }
293
294 static
295 int match_session(struct cds_lfht_node *node, const void *key)
296 {
297 const char *name = key;
298 struct session_info *session_info = caa_container_of(
299 node, struct session_info, sessions_ht_node);
300
301 return !strcmp(session_info->name, name);
302 }
303
304 static
305 unsigned long lttng_condition_buffer_usage_hash(
306 const struct lttng_condition *_condition)
307 {
308 unsigned long hash;
309 unsigned long condition_type;
310 struct lttng_condition_buffer_usage *condition;
311
312 condition = container_of(_condition,
313 struct lttng_condition_buffer_usage, parent);
314
315 condition_type = (unsigned long) condition->parent.type;
316 hash = hash_key_ulong((void *) condition_type, lttng_ht_seed);
317 if (condition->session_name) {
318 hash ^= hash_key_str(condition->session_name, lttng_ht_seed);
319 }
320 if (condition->channel_name) {
321 hash ^= hash_key_str(condition->channel_name, lttng_ht_seed);
322 }
323 if (condition->domain.set) {
324 hash ^= hash_key_ulong(
325 (void *) condition->domain.type,
326 lttng_ht_seed);
327 }
328 if (condition->threshold_ratio.set) {
329 uint64_t val;
330
331 val = condition->threshold_ratio.value * (double) UINT32_MAX;
332 hash ^= hash_key_u64(&val, lttng_ht_seed);
333 } else if (condition->threshold_bytes.set) {
334 uint64_t val;
335
336 val = condition->threshold_bytes.value;
337 hash ^= hash_key_u64(&val, lttng_ht_seed);
338 }
339 return hash;
340 }
341
342 static
343 unsigned long lttng_condition_session_consumed_size_hash(
344 const struct lttng_condition *_condition)
345 {
346 unsigned long hash;
347 unsigned long condition_type =
348 (unsigned long) LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE;
349 struct lttng_condition_session_consumed_size *condition;
350 uint64_t val;
351
352 condition = container_of(_condition,
353 struct lttng_condition_session_consumed_size, parent);
354
355 hash = hash_key_ulong((void *) condition_type, lttng_ht_seed);
356 if (condition->session_name) {
357 hash ^= hash_key_str(condition->session_name, lttng_ht_seed);
358 }
359 val = condition->consumed_threshold_bytes.value;
360 hash ^= hash_key_u64(&val, lttng_ht_seed);
361 return hash;
362 }
363
364 static
365 unsigned long lttng_condition_session_rotation_hash(
366 const struct lttng_condition *_condition)
367 {
368 unsigned long hash, condition_type;
369 struct lttng_condition_session_rotation *condition;
370
371 condition = container_of(_condition,
372 struct lttng_condition_session_rotation, parent);
373 condition_type = (unsigned long) condition->parent.type;
374 hash = hash_key_ulong((void *) condition_type, lttng_ht_seed);
375 assert(condition->session_name);
376 hash ^= hash_key_str(condition->session_name, lttng_ht_seed);
377 return hash;
378 }
379
380 /*
381 * The lttng_condition hashing code is kept in this file (rather than
382 * condition.c) since it makes use of GPLv2 code (hashtable utils), which we
383 * don't want to link in liblttng-ctl.
384 */
385 static
386 unsigned long lttng_condition_hash(const struct lttng_condition *condition)
387 {
388 switch (condition->type) {
389 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
390 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
391 return lttng_condition_buffer_usage_hash(condition);
392 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
393 return lttng_condition_session_consumed_size_hash(condition);
394 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
395 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
396 return lttng_condition_session_rotation_hash(condition);
397 default:
398 ERR("[notification-thread] Unexpected condition type caught");
399 abort();
400 }
401 }
402
403 static
404 unsigned long hash_channel_key(struct channel_key *key)
405 {
406 unsigned long key_hash = hash_key_u64(&key->key, lttng_ht_seed);
407 unsigned long domain_hash = hash_key_ulong(
408 (void *) (unsigned long) key->domain, lttng_ht_seed);
409
410 return key_hash ^ domain_hash;
411 }
412
413 static
414 unsigned long hash_client_socket(int socket)
415 {
416 return hash_key_ulong((void *) (unsigned long) socket, lttng_ht_seed);
417 }
418
419 static
420 unsigned long hash_client_id(notification_client_id id)
421 {
422 return hash_key_u64(&id, lttng_ht_seed);
423 }
424
425 /*
426 * Get the type of object to which a given condition applies. Bindings let
427 * the notification system evaluate a trigger's condition when a given
428 * object's state is updated.
429 *
430 * For instance, a condition bound to a channel will be evaluated everytime
431 * the channel's state is changed by a channel monitoring sample.
432 */
433 static
434 enum lttng_object_type get_condition_binding_object(
435 const struct lttng_condition *condition)
436 {
437 switch (lttng_condition_get_type(condition)) {
438 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
439 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
440 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
441 return LTTNG_OBJECT_TYPE_CHANNEL;
442 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
443 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
444 return LTTNG_OBJECT_TYPE_SESSION;
445 default:
446 return LTTNG_OBJECT_TYPE_UNKNOWN;
447 }
448 }
449
450 static
451 void free_channel_info_rcu(struct rcu_head *node)
452 {
453 free(caa_container_of(node, struct channel_info, rcu_node));
454 }
455
456 static
457 void channel_info_destroy(struct channel_info *channel_info)
458 {
459 if (!channel_info) {
460 return;
461 }
462
463 if (channel_info->session_info) {
464 session_info_remove_channel(channel_info->session_info,
465 channel_info);
466 session_info_put(channel_info->session_info);
467 }
468 if (channel_info->name) {
469 free(channel_info->name);
470 }
471 call_rcu(&channel_info->rcu_node, free_channel_info_rcu);
472 }
473
474 static
475 void free_session_info_rcu(struct rcu_head *node)
476 {
477 free(caa_container_of(node, struct session_info, rcu_node));
478 }
479
480 /* Don't call directly, use the ref-counting mechanism. */
481 static
482 void session_info_destroy(void *_data)
483 {
484 struct session_info *session_info = _data;
485 int ret;
486
487 assert(session_info);
488 if (session_info->channel_infos_ht) {
489 ret = cds_lfht_destroy(session_info->channel_infos_ht, NULL);
490 if (ret) {
491 ERR("[notification-thread] Failed to destroy channel information hash table");
492 }
493 }
494 lttng_session_trigger_list_destroy(session_info->trigger_list);
495
496 rcu_read_lock();
497 cds_lfht_del(session_info->sessions_ht,
498 &session_info->sessions_ht_node);
499 rcu_read_unlock();
500 free(session_info->name);
501 call_rcu(&session_info->rcu_node, free_session_info_rcu);
502 }
503
504 static
505 void session_info_get(struct session_info *session_info)
506 {
507 if (!session_info) {
508 return;
509 }
510 lttng_ref_get(&session_info->ref);
511 }
512
513 static
514 void session_info_put(struct session_info *session_info)
515 {
516 if (!session_info) {
517 return;
518 }
519 lttng_ref_put(&session_info->ref);
520 }
521
522 static
523 struct session_info *session_info_create(const char *name, uid_t uid, gid_t gid,
524 struct lttng_session_trigger_list *trigger_list,
525 struct cds_lfht *sessions_ht)
526 {
527 struct session_info *session_info;
528
529 assert(name);
530
531 session_info = zmalloc(sizeof(*session_info));
532 if (!session_info) {
533 goto end;
534 }
535 lttng_ref_init(&session_info->ref, session_info_destroy);
536
537 session_info->channel_infos_ht = cds_lfht_new(DEFAULT_HT_SIZE,
538 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
539 if (!session_info->channel_infos_ht) {
540 goto error;
541 }
542
543 cds_lfht_node_init(&session_info->sessions_ht_node);
544 session_info->name = strdup(name);
545 if (!session_info->name) {
546 goto error;
547 }
548 session_info->uid = uid;
549 session_info->gid = gid;
550 session_info->trigger_list = trigger_list;
551 session_info->sessions_ht = sessions_ht;
552 end:
553 return session_info;
554 error:
555 session_info_put(session_info);
556 return NULL;
557 }
558
559 static
560 void session_info_add_channel(struct session_info *session_info,
561 struct channel_info *channel_info)
562 {
563 rcu_read_lock();
564 cds_lfht_add(session_info->channel_infos_ht,
565 hash_channel_key(&channel_info->key),
566 &channel_info->session_info_channels_ht_node);
567 rcu_read_unlock();
568 }
569
570 static
571 void session_info_remove_channel(struct session_info *session_info,
572 struct channel_info *channel_info)
573 {
574 rcu_read_lock();
575 cds_lfht_del(session_info->channel_infos_ht,
576 &channel_info->session_info_channels_ht_node);
577 rcu_read_unlock();
578 }
579
580 static
581 struct channel_info *channel_info_create(const char *channel_name,
582 struct channel_key *channel_key, uint64_t channel_capacity,
583 struct session_info *session_info)
584 {
585 struct channel_info *channel_info = zmalloc(sizeof(*channel_info));
586
587 if (!channel_info) {
588 goto end;
589 }
590
591 cds_lfht_node_init(&channel_info->channels_ht_node);
592 cds_lfht_node_init(&channel_info->session_info_channels_ht_node);
593 memcpy(&channel_info->key, channel_key, sizeof(*channel_key));
594 channel_info->capacity = channel_capacity;
595
596 channel_info->name = strdup(channel_name);
597 if (!channel_info->name) {
598 goto error;
599 }
600
601 /*
602 * Set the references between session and channel infos:
603 * - channel_info holds a strong reference to session_info
604 * - session_info holds a weak reference to channel_info
605 */
606 session_info_get(session_info);
607 session_info_add_channel(session_info, channel_info);
608 channel_info->session_info = session_info;
609 end:
610 return channel_info;
611 error:
612 channel_info_destroy(channel_info);
613 return NULL;
614 }
615
616 LTTNG_HIDDEN
617 bool notification_client_list_get(struct notification_client_list *list)
618 {
619 return urcu_ref_get_unless_zero(&list->ref);
620 }
621
622 static
623 void free_notification_client_list_rcu(struct rcu_head *node)
624 {
625 free(caa_container_of(node, struct notification_client_list,
626 rcu_node));
627 }
628
629 static
630 void notification_client_list_release(struct urcu_ref *list_ref)
631 {
632 struct notification_client_list *list =
633 container_of(list_ref, typeof(*list), ref);
634 struct notification_client_list_element *client_list_element, *tmp;
635
636 if (list->notification_trigger_clients_ht) {
637 rcu_read_lock();
638 cds_lfht_del(list->notification_trigger_clients_ht,
639 &list->notification_trigger_clients_ht_node);
640 rcu_read_unlock();
641 list->notification_trigger_clients_ht = NULL;
642 }
643 cds_list_for_each_entry_safe(client_list_element, tmp,
644 &list->list, node) {
645 free(client_list_element);
646 }
647 pthread_mutex_destroy(&list->lock);
648 call_rcu(&list->rcu_node, free_notification_client_list_rcu);
649 }
650
651 static
652 struct notification_client_list *notification_client_list_create(
653 const struct lttng_trigger *trigger)
654 {
655 struct notification_client_list *client_list =
656 zmalloc(sizeof(*client_list));
657
658 if (!client_list) {
659 goto error;
660 }
661 pthread_mutex_init(&client_list->lock, NULL);
662 urcu_ref_init(&client_list->ref);
663 cds_lfht_node_init(&client_list->notification_trigger_clients_ht_node);
664 CDS_INIT_LIST_HEAD(&client_list->list);
665 client_list->trigger = trigger;
666 error:
667 return client_list;
668 }
669
670 static
671 void publish_notification_client_list(
672 struct notification_thread_state *state,
673 struct notification_client_list *list)
674 {
675 const struct lttng_condition *condition =
676 lttng_trigger_get_const_condition(list->trigger);
677
678 assert(!list->notification_trigger_clients_ht);
679 notification_client_list_get(list);
680
681 list->notification_trigger_clients_ht =
682 state->notification_trigger_clients_ht;
683
684 rcu_read_lock();
685 cds_lfht_add(state->notification_trigger_clients_ht,
686 lttng_condition_hash(condition),
687 &list->notification_trigger_clients_ht_node);
688 rcu_read_unlock();
689 }
690
691 LTTNG_HIDDEN
692 void notification_client_list_put(struct notification_client_list *list)
693 {
694 if (!list) {
695 return;
696 }
697 return urcu_ref_put(&list->ref, notification_client_list_release);
698 }
699
700 /* Provides a reference to the returned list. */
701 static
702 struct notification_client_list *get_client_list_from_condition(
703 struct notification_thread_state *state,
704 const struct lttng_condition *condition)
705 {
706 struct cds_lfht_node *node;
707 struct cds_lfht_iter iter;
708 struct notification_client_list *list = NULL;
709
710 rcu_read_lock();
711 cds_lfht_lookup(state->notification_trigger_clients_ht,
712 lttng_condition_hash(condition),
713 match_client_list_condition,
714 condition,
715 &iter);
716 node = cds_lfht_iter_get_node(&iter);
717 if (node) {
718 list = container_of(node, struct notification_client_list,
719 notification_trigger_clients_ht_node);
720 list = notification_client_list_get(list) ? list : NULL;
721 }
722
723 rcu_read_unlock();
724 return list;
725 }
726
727 static
728 int evaluate_channel_condition_for_client(
729 const struct lttng_condition *condition,
730 struct notification_thread_state *state,
731 struct lttng_evaluation **evaluation,
732 uid_t *session_uid, gid_t *session_gid)
733 {
734 int ret;
735 struct cds_lfht_iter iter;
736 struct cds_lfht_node *node;
737 struct channel_info *channel_info = NULL;
738 struct channel_key *channel_key = NULL;
739 struct channel_state_sample *last_sample = NULL;
740 struct lttng_channel_trigger_list *channel_trigger_list = NULL;
741
742 rcu_read_lock();
743
744 /* Find the channel associated with the condition. */
745 cds_lfht_for_each_entry(state->channel_triggers_ht, &iter,
746 channel_trigger_list, channel_triggers_ht_node) {
747 struct lttng_trigger_list_element *element;
748
749 cds_list_for_each_entry(element, &channel_trigger_list->list, node) {
750 const struct lttng_condition *current_condition =
751 lttng_trigger_get_const_condition(
752 element->trigger);
753
754 assert(current_condition);
755 if (!lttng_condition_is_equal(condition,
756 current_condition)) {
757 continue;
758 }
759
760 /* Found the trigger, save the channel key. */
761 channel_key = &channel_trigger_list->channel_key;
762 break;
763 }
764 if (channel_key) {
765 /* The channel key was found stop iteration. */
766 break;
767 }
768 }
769
770 if (!channel_key){
771 /* No channel found; normal exit. */
772 DBG("[notification-thread] No known channel associated with newly subscribed-to condition");
773 ret = 0;
774 goto end;
775 }
776
777 /* Fetch channel info for the matching channel. */
778 cds_lfht_lookup(state->channels_ht,
779 hash_channel_key(channel_key),
780 match_channel_info,
781 channel_key,
782 &iter);
783 node = cds_lfht_iter_get_node(&iter);
784 assert(node);
785 channel_info = caa_container_of(node, struct channel_info,
786 channels_ht_node);
787
788 /* Retrieve the channel's last sample, if it exists. */
789 cds_lfht_lookup(state->channel_state_ht,
790 hash_channel_key(channel_key),
791 match_channel_state_sample,
792 channel_key,
793 &iter);
794 node = cds_lfht_iter_get_node(&iter);
795 if (node) {
796 last_sample = caa_container_of(node,
797 struct channel_state_sample,
798 channel_state_ht_node);
799 } else {
800 /* Nothing to evaluate, no sample was ever taken. Normal exit */
801 DBG("[notification-thread] No channel sample associated with newly subscribed-to condition");
802 ret = 0;
803 goto end;
804 }
805
806 ret = evaluate_buffer_condition(condition, evaluation, state,
807 NULL, last_sample,
808 0, channel_info->session_info->consumed_data_size,
809 channel_info);
810 if (ret) {
811 WARN("[notification-thread] Fatal error occurred while evaluating a newly subscribed-to condition");
812 goto end;
813 }
814
815 *session_uid = channel_info->session_info->uid;
816 *session_gid = channel_info->session_info->gid;
817 end:
818 rcu_read_unlock();
819 return ret;
820 }
821
822 static
823 const char *get_condition_session_name(const struct lttng_condition *condition)
824 {
825 const char *session_name = NULL;
826 enum lttng_condition_status status;
827
828 switch (lttng_condition_get_type(condition)) {
829 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
830 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
831 status = lttng_condition_buffer_usage_get_session_name(
832 condition, &session_name);
833 break;
834 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
835 status = lttng_condition_session_consumed_size_get_session_name(
836 condition, &session_name);
837 break;
838 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
839 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
840 status = lttng_condition_session_rotation_get_session_name(
841 condition, &session_name);
842 break;
843 default:
844 abort();
845 }
846 if (status != LTTNG_CONDITION_STATUS_OK) {
847 ERR("[notification-thread] Failed to retrieve session rotation condition's session name");
848 goto end;
849 }
850 end:
851 return session_name;
852 }
853
854 static
855 int evaluate_session_condition_for_client(
856 const struct lttng_condition *condition,
857 struct notification_thread_state *state,
858 struct lttng_evaluation **evaluation,
859 uid_t *session_uid, gid_t *session_gid)
860 {
861 int ret;
862 struct cds_lfht_iter iter;
863 struct cds_lfht_node *node;
864 const char *session_name;
865 struct session_info *session_info = NULL;
866
867 rcu_read_lock();
868 session_name = get_condition_session_name(condition);
869
870 /* Find the session associated with the trigger. */
871 cds_lfht_lookup(state->sessions_ht,
872 hash_key_str(session_name, lttng_ht_seed),
873 match_session,
874 session_name,
875 &iter);
876 node = cds_lfht_iter_get_node(&iter);
877 if (!node) {
878 DBG("[notification-thread] No known session matching name \"%s\"",
879 session_name);
880 ret = 0;
881 goto end;
882 }
883
884 session_info = caa_container_of(node, struct session_info,
885 sessions_ht_node);
886 session_info_get(session_info);
887
888 /*
889 * Evaluation is performed in-line here since only one type of
890 * session-bound condition is handled for the moment.
891 */
892 switch (lttng_condition_get_type(condition)) {
893 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
894 if (!session_info->rotation.ongoing) {
895 ret = 0;
896 goto end_session_put;
897 }
898
899 *evaluation = lttng_evaluation_session_rotation_ongoing_create(
900 session_info->rotation.id);
901 if (!*evaluation) {
902 /* Fatal error. */
903 ERR("[notification-thread] Failed to create session rotation ongoing evaluation for session \"%s\"",
904 session_info->name);
905 ret = -1;
906 goto end_session_put;
907 }
908 ret = 0;
909 break;
910 default:
911 ret = 0;
912 goto end_session_put;
913 }
914
915 *session_uid = session_info->uid;
916 *session_gid = session_info->gid;
917
918 end_session_put:
919 session_info_put(session_info);
920 end:
921 rcu_read_unlock();
922 return ret;
923 }
924
925 static
926 int evaluate_condition_for_client(const struct lttng_trigger *trigger,
927 const struct lttng_condition *condition,
928 struct notification_client *client,
929 struct notification_thread_state *state)
930 {
931 int ret;
932 struct lttng_evaluation *evaluation = NULL;
933 struct notification_client_list client_list = {
934 .lock = PTHREAD_MUTEX_INITIALIZER,
935 };
936 struct notification_client_list_element client_list_element = { 0 };
937 uid_t object_uid = 0;
938 gid_t object_gid = 0;
939
940 assert(trigger);
941 assert(condition);
942 assert(client);
943 assert(state);
944
945 switch (get_condition_binding_object(condition)) {
946 case LTTNG_OBJECT_TYPE_SESSION:
947 ret = evaluate_session_condition_for_client(condition, state,
948 &evaluation, &object_uid, &object_gid);
949 break;
950 case LTTNG_OBJECT_TYPE_CHANNEL:
951 ret = evaluate_channel_condition_for_client(condition, state,
952 &evaluation, &object_uid, &object_gid);
953 break;
954 case LTTNG_OBJECT_TYPE_NONE:
955 ret = 0;
956 goto end;
957 case LTTNG_OBJECT_TYPE_UNKNOWN:
958 default:
959 ret = -1;
960 goto end;
961 }
962 if (ret) {
963 /* Fatal error. */
964 goto end;
965 }
966 if (!evaluation) {
967 /* Evaluation yielded nothing. Normal exit. */
968 DBG("[notification-thread] Newly subscribed-to condition evaluated to false, nothing to report to client");
969 ret = 0;
970 goto end;
971 }
972
973 /*
974 * Create a temporary client list with the client currently
975 * subscribing.
976 */
977 cds_lfht_node_init(&client_list.notification_trigger_clients_ht_node);
978 CDS_INIT_LIST_HEAD(&client_list.list);
979 client_list.trigger = trigger;
980
981 CDS_INIT_LIST_HEAD(&client_list_element.node);
982 client_list_element.client = client;
983 cds_list_add(&client_list_element.node, &client_list.list);
984
985 /* Send evaluation result to the newly-subscribed client. */
986 DBG("[notification-thread] Newly subscribed-to condition evaluated to true, notifying client");
987 ret = send_evaluation_to_clients(trigger, evaluation, &client_list,
988 state, object_uid, object_gid);
989
990 end:
991 return ret;
992 }
993
994 static
995 int notification_thread_client_subscribe(struct notification_client *client,
996 struct lttng_condition *condition,
997 struct notification_thread_state *state,
998 enum lttng_notification_channel_status *_status)
999 {
1000 int ret = 0;
1001 struct notification_client_list *client_list = NULL;
1002 struct lttng_condition_list_element *condition_list_element = NULL;
1003 struct notification_client_list_element *client_list_element = NULL;
1004 enum lttng_notification_channel_status status =
1005 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
1006
1007 /*
1008 * Ensure that the client has not already subscribed to this condition
1009 * before.
1010 */
1011 cds_list_for_each_entry(condition_list_element, &client->condition_list, node) {
1012 if (lttng_condition_is_equal(condition_list_element->condition,
1013 condition)) {
1014 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ALREADY_SUBSCRIBED;
1015 goto end;
1016 }
1017 }
1018
1019 condition_list_element = zmalloc(sizeof(*condition_list_element));
1020 if (!condition_list_element) {
1021 ret = -1;
1022 goto error;
1023 }
1024 client_list_element = zmalloc(sizeof(*client_list_element));
1025 if (!client_list_element) {
1026 ret = -1;
1027 goto error;
1028 }
1029
1030 /*
1031 * Add the newly-subscribed condition to the client's subscription list.
1032 */
1033 CDS_INIT_LIST_HEAD(&condition_list_element->node);
1034 condition_list_element->condition = condition;
1035 cds_list_add(&condition_list_element->node, &client->condition_list);
1036
1037 client_list = get_client_list_from_condition(state, condition);
1038 if (!client_list) {
1039 /*
1040 * No notification-emiting trigger registered with this
1041 * condition. We don't evaluate the condition right away
1042 * since this trigger is not registered yet.
1043 */
1044 free(client_list_element);
1045 goto end;
1046 }
1047
1048 /*
1049 * The condition to which the client just subscribed is evaluated
1050 * at this point so that conditions that are already TRUE result
1051 * in a notification being sent out.
1052 *
1053 * The client_list's trigger is used without locking the list itself.
1054 * This is correct since the list doesn't own the trigger and the
1055 * object is immutable.
1056 */
1057 if (evaluate_condition_for_client(client_list->trigger, condition,
1058 client, state)) {
1059 WARN("[notification-thread] Evaluation of a condition on client subscription failed, aborting.");
1060 ret = -1;
1061 free(client_list_element);
1062 goto end;
1063 }
1064
1065 /*
1066 * Add the client to the list of clients interested in a given trigger
1067 * if a "notification" trigger with a corresponding condition was
1068 * added prior.
1069 */
1070 client_list_element->client = client;
1071 CDS_INIT_LIST_HEAD(&client_list_element->node);
1072
1073 pthread_mutex_lock(&client_list->lock);
1074 cds_list_add(&client_list_element->node, &client_list->list);
1075 pthread_mutex_unlock(&client_list->lock);
1076 end:
1077 if (_status) {
1078 *_status = status;
1079 }
1080 if (client_list) {
1081 notification_client_list_put(client_list);
1082 }
1083 return ret;
1084 error:
1085 free(condition_list_element);
1086 free(client_list_element);
1087 return ret;
1088 }
1089
1090 static
1091 int notification_thread_client_unsubscribe(
1092 struct notification_client *client,
1093 struct lttng_condition *condition,
1094 struct notification_thread_state *state,
1095 enum lttng_notification_channel_status *_status)
1096 {
1097 struct notification_client_list *client_list;
1098 struct lttng_condition_list_element *condition_list_element,
1099 *condition_tmp;
1100 struct notification_client_list_element *client_list_element,
1101 *client_tmp;
1102 bool condition_found = false;
1103 enum lttng_notification_channel_status status =
1104 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
1105
1106 /* Remove the condition from the client's condition list. */
1107 cds_list_for_each_entry_safe(condition_list_element, condition_tmp,
1108 &client->condition_list, node) {
1109 if (!lttng_condition_is_equal(condition_list_element->condition,
1110 condition)) {
1111 continue;
1112 }
1113
1114 cds_list_del(&condition_list_element->node);
1115 /*
1116 * The caller may be iterating on the client's conditions to
1117 * tear down a client's connection. In this case, the condition
1118 * will be destroyed at the end.
1119 */
1120 if (condition != condition_list_element->condition) {
1121 lttng_condition_destroy(
1122 condition_list_element->condition);
1123 }
1124 free(condition_list_element);
1125 condition_found = true;
1126 break;
1127 }
1128
1129 if (!condition_found) {
1130 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNKNOWN_CONDITION;
1131 goto end;
1132 }
1133
1134 /*
1135 * Remove the client from the list of clients interested the trigger
1136 * matching the condition.
1137 */
1138 client_list = get_client_list_from_condition(state, condition);
1139 if (!client_list) {
1140 goto end;
1141 }
1142
1143 pthread_mutex_lock(&client_list->lock);
1144 cds_list_for_each_entry_safe(client_list_element, client_tmp,
1145 &client_list->list, node) {
1146 if (client_list_element->client->id != client->id) {
1147 continue;
1148 }
1149 cds_list_del(&client_list_element->node);
1150 free(client_list_element);
1151 break;
1152 }
1153 pthread_mutex_unlock(&client_list->lock);
1154 notification_client_list_put(client_list);
1155 client_list = NULL;
1156 end:
1157 lttng_condition_destroy(condition);
1158 if (_status) {
1159 *_status = status;
1160 }
1161 return 0;
1162 }
1163
1164 static
1165 void free_notification_client_rcu(struct rcu_head *node)
1166 {
1167 free(caa_container_of(node, struct notification_client, rcu_node));
1168 }
1169
1170 static
1171 void notification_client_destroy(struct notification_client *client,
1172 struct notification_thread_state *state)
1173 {
1174 if (!client) {
1175 return;
1176 }
1177
1178 /*
1179 * The client object is not reachable by other threads, no need to lock
1180 * the client here.
1181 */
1182 if (client->socket >= 0) {
1183 (void) lttcomm_close_unix_sock(client->socket);
1184 client->socket = -1;
1185 }
1186 client->communication.active = false;
1187 lttng_dynamic_buffer_reset(&client->communication.inbound.buffer);
1188 lttng_dynamic_buffer_reset(&client->communication.outbound.buffer);
1189 pthread_mutex_destroy(&client->lock);
1190 call_rcu(&client->rcu_node, free_notification_client_rcu);
1191 }
1192
1193 /*
1194 * Call with rcu_read_lock held (and hold for the lifetime of the returned
1195 * client pointer).
1196 */
1197 static
1198 struct notification_client *get_client_from_socket(int socket,
1199 struct notification_thread_state *state)
1200 {
1201 struct cds_lfht_iter iter;
1202 struct cds_lfht_node *node;
1203 struct notification_client *client = NULL;
1204
1205 cds_lfht_lookup(state->client_socket_ht,
1206 hash_client_socket(socket),
1207 match_client_socket,
1208 (void *) (unsigned long) socket,
1209 &iter);
1210 node = cds_lfht_iter_get_node(&iter);
1211 if (!node) {
1212 goto end;
1213 }
1214
1215 client = caa_container_of(node, struct notification_client,
1216 client_socket_ht_node);
1217 end:
1218 return client;
1219 }
1220
1221 /*
1222 * Call with rcu_read_lock held (and hold for the lifetime of the returned
1223 * client pointer).
1224 */
1225 static
1226 struct notification_client *get_client_from_id(notification_client_id id,
1227 struct notification_thread_state *state)
1228 {
1229 struct cds_lfht_iter iter;
1230 struct cds_lfht_node *node;
1231 struct notification_client *client = NULL;
1232
1233 cds_lfht_lookup(state->client_id_ht,
1234 hash_client_id(id),
1235 match_client_id,
1236 &id,
1237 &iter);
1238 node = cds_lfht_iter_get_node(&iter);
1239 if (!node) {
1240 goto end;
1241 }
1242
1243 client = caa_container_of(node, struct notification_client,
1244 client_id_ht_node);
1245 end:
1246 return client;
1247 }
1248
1249 static
1250 bool buffer_usage_condition_applies_to_channel(
1251 const struct lttng_condition *condition,
1252 const struct channel_info *channel_info)
1253 {
1254 enum lttng_condition_status status;
1255 enum lttng_domain_type condition_domain;
1256 const char *condition_session_name = NULL;
1257 const char *condition_channel_name = NULL;
1258
1259 status = lttng_condition_buffer_usage_get_domain_type(condition,
1260 &condition_domain);
1261 assert(status == LTTNG_CONDITION_STATUS_OK);
1262 if (channel_info->key.domain != condition_domain) {
1263 goto fail;
1264 }
1265
1266 status = lttng_condition_buffer_usage_get_session_name(
1267 condition, &condition_session_name);
1268 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_session_name);
1269
1270 status = lttng_condition_buffer_usage_get_channel_name(
1271 condition, &condition_channel_name);
1272 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_channel_name);
1273
1274 if (strcmp(channel_info->session_info->name, condition_session_name)) {
1275 goto fail;
1276 }
1277 if (strcmp(channel_info->name, condition_channel_name)) {
1278 goto fail;
1279 }
1280
1281 return true;
1282 fail:
1283 return false;
1284 }
1285
1286 static
1287 bool session_consumed_size_condition_applies_to_channel(
1288 const struct lttng_condition *condition,
1289 const struct channel_info *channel_info)
1290 {
1291 enum lttng_condition_status status;
1292 const char *condition_session_name = NULL;
1293
1294 status = lttng_condition_session_consumed_size_get_session_name(
1295 condition, &condition_session_name);
1296 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_session_name);
1297
1298 if (strcmp(channel_info->session_info->name, condition_session_name)) {
1299 goto fail;
1300 }
1301
1302 return true;
1303 fail:
1304 return false;
1305 }
1306
1307 static
1308 bool trigger_applies_to_channel(const struct lttng_trigger *trigger,
1309 const struct channel_info *channel_info)
1310 {
1311 const struct lttng_condition *condition;
1312 bool trigger_applies;
1313
1314 condition = lttng_trigger_get_const_condition(trigger);
1315 if (!condition) {
1316 goto fail;
1317 }
1318
1319 switch (lttng_condition_get_type(condition)) {
1320 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
1321 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
1322 trigger_applies = buffer_usage_condition_applies_to_channel(
1323 condition, channel_info);
1324 break;
1325 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
1326 trigger_applies = session_consumed_size_condition_applies_to_channel(
1327 condition, channel_info);
1328 break;
1329 default:
1330 goto fail;
1331 }
1332
1333 return trigger_applies;
1334 fail:
1335 return false;
1336 }
1337
1338 static
1339 bool trigger_applies_to_client(struct lttng_trigger *trigger,
1340 struct notification_client *client)
1341 {
1342 bool applies = false;
1343 struct lttng_condition_list_element *condition_list_element;
1344
1345 cds_list_for_each_entry(condition_list_element, &client->condition_list,
1346 node) {
1347 applies = lttng_condition_is_equal(
1348 condition_list_element->condition,
1349 lttng_trigger_get_condition(trigger));
1350 if (applies) {
1351 break;
1352 }
1353 }
1354 return applies;
1355 }
1356
1357 /* Must be called with RCU read lock held. */
1358 static
1359 struct lttng_session_trigger_list *get_session_trigger_list(
1360 struct notification_thread_state *state,
1361 const char *session_name)
1362 {
1363 struct lttng_session_trigger_list *list = NULL;
1364 struct cds_lfht_node *node;
1365 struct cds_lfht_iter iter;
1366
1367 cds_lfht_lookup(state->session_triggers_ht,
1368 hash_key_str(session_name, lttng_ht_seed),
1369 match_session_trigger_list,
1370 session_name,
1371 &iter);
1372 node = cds_lfht_iter_get_node(&iter);
1373 if (!node) {
1374 /*
1375 * Not an error, the list of triggers applying to that session
1376 * will be initialized when the session is created.
1377 */
1378 DBG("[notification-thread] No trigger list found for session \"%s\" as it is not yet known to the notification system",
1379 session_name);
1380 goto end;
1381 }
1382
1383 list = caa_container_of(node,
1384 struct lttng_session_trigger_list,
1385 session_triggers_ht_node);
1386 end:
1387 return list;
1388 }
1389
1390 /*
1391 * Allocate an empty lttng_session_trigger_list for the session named
1392 * 'session_name'.
1393 *
1394 * No ownership of 'session_name' is assumed by the session trigger list.
1395 * It is the caller's responsability to ensure the session name is alive
1396 * for as long as this list is.
1397 */
1398 static
1399 struct lttng_session_trigger_list *lttng_session_trigger_list_create(
1400 const char *session_name,
1401 struct cds_lfht *session_triggers_ht)
1402 {
1403 struct lttng_session_trigger_list *list;
1404
1405 list = zmalloc(sizeof(*list));
1406 if (!list) {
1407 goto end;
1408 }
1409 list->session_name = session_name;
1410 CDS_INIT_LIST_HEAD(&list->list);
1411 cds_lfht_node_init(&list->session_triggers_ht_node);
1412 list->session_triggers_ht = session_triggers_ht;
1413
1414 rcu_read_lock();
1415 /* Publish the list through the session_triggers_ht. */
1416 cds_lfht_add(session_triggers_ht,
1417 hash_key_str(session_name, lttng_ht_seed),
1418 &list->session_triggers_ht_node);
1419 rcu_read_unlock();
1420 end:
1421 return list;
1422 }
1423
1424 static
1425 void free_session_trigger_list_rcu(struct rcu_head *node)
1426 {
1427 free(caa_container_of(node, struct lttng_session_trigger_list,
1428 rcu_node));
1429 }
1430
1431 static
1432 void lttng_session_trigger_list_destroy(struct lttng_session_trigger_list *list)
1433 {
1434 struct lttng_trigger_list_element *trigger_list_element, *tmp;
1435
1436 /* Empty the list element by element, and then free the list itself. */
1437 cds_list_for_each_entry_safe(trigger_list_element, tmp,
1438 &list->list, node) {
1439 cds_list_del(&trigger_list_element->node);
1440 free(trigger_list_element);
1441 }
1442 rcu_read_lock();
1443 /* Unpublish the list from the session_triggers_ht. */
1444 cds_lfht_del(list->session_triggers_ht,
1445 &list->session_triggers_ht_node);
1446 rcu_read_unlock();
1447 call_rcu(&list->rcu_node, free_session_trigger_list_rcu);
1448 }
1449
1450 static
1451 int lttng_session_trigger_list_add(struct lttng_session_trigger_list *list,
1452 struct lttng_trigger *trigger)
1453 {
1454 int ret = 0;
1455 struct lttng_trigger_list_element *new_element =
1456 zmalloc(sizeof(*new_element));
1457
1458 if (!new_element) {
1459 ret = -1;
1460 goto end;
1461 }
1462 CDS_INIT_LIST_HEAD(&new_element->node);
1463 new_element->trigger = trigger;
1464 cds_list_add(&new_element->node, &list->list);
1465 end:
1466 return ret;
1467 }
1468
1469 static
1470 bool trigger_applies_to_session(const struct lttng_trigger *trigger,
1471 const char *session_name)
1472 {
1473 bool applies = false;
1474 const struct lttng_condition *condition;
1475
1476 condition = lttng_trigger_get_const_condition(trigger);
1477 switch (lttng_condition_get_type(condition)) {
1478 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
1479 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
1480 {
1481 enum lttng_condition_status condition_status;
1482 const char *condition_session_name;
1483
1484 condition_status = lttng_condition_session_rotation_get_session_name(
1485 condition, &condition_session_name);
1486 if (condition_status != LTTNG_CONDITION_STATUS_OK) {
1487 ERR("[notification-thread] Failed to retrieve session rotation condition's session name");
1488 goto end;
1489 }
1490
1491 assert(condition_session_name);
1492 applies = !strcmp(condition_session_name, session_name);
1493 break;
1494 }
1495 default:
1496 goto end;
1497 }
1498 end:
1499 return applies;
1500 }
1501
1502 /*
1503 * Allocate and initialize an lttng_session_trigger_list which contains
1504 * all triggers that apply to the session named 'session_name'.
1505 *
1506 * No ownership of 'session_name' is assumed by the session trigger list.
1507 * It is the caller's responsability to ensure the session name is alive
1508 * for as long as this list is.
1509 */
1510 static
1511 struct lttng_session_trigger_list *lttng_session_trigger_list_build(
1512 const struct notification_thread_state *state,
1513 const char *session_name)
1514 {
1515 int trigger_count = 0;
1516 struct lttng_session_trigger_list *session_trigger_list = NULL;
1517 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
1518 struct cds_lfht_iter iter;
1519
1520 session_trigger_list = lttng_session_trigger_list_create(session_name,
1521 state->session_triggers_ht);
1522
1523 /* Add all triggers applying to the session named 'session_name'. */
1524 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1525 node) {
1526 int ret;
1527
1528 if (!trigger_applies_to_session(trigger_ht_element->trigger,
1529 session_name)) {
1530 continue;
1531 }
1532
1533 ret = lttng_session_trigger_list_add(session_trigger_list,
1534 trigger_ht_element->trigger);
1535 if (ret) {
1536 goto error;
1537 }
1538
1539 trigger_count++;
1540 }
1541
1542 DBG("[notification-thread] Found %i triggers that apply to newly created session",
1543 trigger_count);
1544 return session_trigger_list;
1545 error:
1546 lttng_session_trigger_list_destroy(session_trigger_list);
1547 return NULL;
1548 }
1549
1550 static
1551 struct session_info *find_or_create_session_info(
1552 struct notification_thread_state *state,
1553 const char *name, uid_t uid, gid_t gid)
1554 {
1555 struct session_info *session = NULL;
1556 struct cds_lfht_node *node;
1557 struct cds_lfht_iter iter;
1558 struct lttng_session_trigger_list *trigger_list;
1559
1560 rcu_read_lock();
1561 cds_lfht_lookup(state->sessions_ht,
1562 hash_key_str(name, lttng_ht_seed),
1563 match_session,
1564 name,
1565 &iter);
1566 node = cds_lfht_iter_get_node(&iter);
1567 if (node) {
1568 DBG("[notification-thread] Found session info of session \"%s\" (uid = %i, gid = %i)",
1569 name, uid, gid);
1570 session = caa_container_of(node, struct session_info,
1571 sessions_ht_node);
1572 assert(session->uid == uid);
1573 assert(session->gid == gid);
1574 session_info_get(session);
1575 goto end;
1576 }
1577
1578 trigger_list = lttng_session_trigger_list_build(state, name);
1579 if (!trigger_list) {
1580 goto error;
1581 }
1582
1583 session = session_info_create(name, uid, gid, trigger_list,
1584 state->sessions_ht);
1585 if (!session) {
1586 ERR("[notification-thread] Failed to allocation session info for session \"%s\" (uid = %i, gid = %i)",
1587 name, uid, gid);
1588 lttng_session_trigger_list_destroy(trigger_list);
1589 goto error;
1590 }
1591 trigger_list = NULL;
1592
1593 cds_lfht_add(state->sessions_ht, hash_key_str(name, lttng_ht_seed),
1594 &session->sessions_ht_node);
1595 end:
1596 rcu_read_unlock();
1597 return session;
1598 error:
1599 rcu_read_unlock();
1600 session_info_put(session);
1601 return NULL;
1602 }
1603
1604 static
1605 int handle_notification_thread_command_add_channel(
1606 struct notification_thread_state *state,
1607 const char *session_name, uid_t session_uid, gid_t session_gid,
1608 const char *channel_name, enum lttng_domain_type channel_domain,
1609 uint64_t channel_key_int, uint64_t channel_capacity,
1610 enum lttng_error_code *cmd_result)
1611 {
1612 struct cds_list_head trigger_list;
1613 struct channel_info *new_channel_info = NULL;
1614 struct channel_key channel_key = {
1615 .key = channel_key_int,
1616 .domain = channel_domain,
1617 };
1618 struct lttng_channel_trigger_list *channel_trigger_list = NULL;
1619 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
1620 int trigger_count = 0;
1621 struct cds_lfht_iter iter;
1622 struct session_info *session_info = NULL;
1623
1624 DBG("[notification-thread] Adding channel %s from session %s, channel key = %" PRIu64 " in %s domain",
1625 channel_name, session_name, channel_key_int,
1626 channel_domain == LTTNG_DOMAIN_KERNEL ? "kernel" : "user space");
1627
1628 CDS_INIT_LIST_HEAD(&trigger_list);
1629
1630 session_info = find_or_create_session_info(state, session_name,
1631 session_uid, session_gid);
1632 if (!session_info) {
1633 /* Allocation error or an internal error occurred. */
1634 goto error;
1635 }
1636
1637 new_channel_info = channel_info_create(channel_name, &channel_key,
1638 channel_capacity, session_info);
1639 if (!new_channel_info) {
1640 goto error;
1641 }
1642
1643 rcu_read_lock();
1644 /* Build a list of all triggers applying to the new channel. */
1645 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1646 node) {
1647 struct lttng_trigger_list_element *new_element;
1648
1649 if (!trigger_applies_to_channel(trigger_ht_element->trigger,
1650 new_channel_info)) {
1651 continue;
1652 }
1653
1654 new_element = zmalloc(sizeof(*new_element));
1655 if (!new_element) {
1656 rcu_read_unlock();
1657 goto error;
1658 }
1659 CDS_INIT_LIST_HEAD(&new_element->node);
1660 new_element->trigger = trigger_ht_element->trigger;
1661 cds_list_add(&new_element->node, &trigger_list);
1662 trigger_count++;
1663 }
1664 rcu_read_unlock();
1665
1666 DBG("[notification-thread] Found %i triggers that apply to newly added channel",
1667 trigger_count);
1668 channel_trigger_list = zmalloc(sizeof(*channel_trigger_list));
1669 if (!channel_trigger_list) {
1670 goto error;
1671 }
1672 channel_trigger_list->channel_key = new_channel_info->key;
1673 CDS_INIT_LIST_HEAD(&channel_trigger_list->list);
1674 cds_lfht_node_init(&channel_trigger_list->channel_triggers_ht_node);
1675 cds_list_splice(&trigger_list, &channel_trigger_list->list);
1676
1677 rcu_read_lock();
1678 /* Add channel to the channel_ht which owns the channel_infos. */
1679 cds_lfht_add(state->channels_ht,
1680 hash_channel_key(&new_channel_info->key),
1681 &new_channel_info->channels_ht_node);
1682 /*
1683 * Add the list of triggers associated with this channel to the
1684 * channel_triggers_ht.
1685 */
1686 cds_lfht_add(state->channel_triggers_ht,
1687 hash_channel_key(&new_channel_info->key),
1688 &channel_trigger_list->channel_triggers_ht_node);
1689 rcu_read_unlock();
1690 session_info_put(session_info);
1691 *cmd_result = LTTNG_OK;
1692 return 0;
1693 error:
1694 channel_info_destroy(new_channel_info);
1695 session_info_put(session_info);
1696 return 1;
1697 }
1698
1699 static
1700 void free_channel_trigger_list_rcu(struct rcu_head *node)
1701 {
1702 free(caa_container_of(node, struct lttng_channel_trigger_list,
1703 rcu_node));
1704 }
1705
1706 static
1707 void free_channel_state_sample_rcu(struct rcu_head *node)
1708 {
1709 free(caa_container_of(node, struct channel_state_sample,
1710 rcu_node));
1711 }
1712
1713 static
1714 int handle_notification_thread_command_remove_channel(
1715 struct notification_thread_state *state,
1716 uint64_t channel_key, enum lttng_domain_type domain,
1717 enum lttng_error_code *cmd_result)
1718 {
1719 struct cds_lfht_node *node;
1720 struct cds_lfht_iter iter;
1721 struct lttng_channel_trigger_list *trigger_list;
1722 struct lttng_trigger_list_element *trigger_list_element, *tmp;
1723 struct channel_key key = { .key = channel_key, .domain = domain };
1724 struct channel_info *channel_info;
1725
1726 DBG("[notification-thread] Removing channel key = %" PRIu64 " in %s domain",
1727 channel_key, domain == LTTNG_DOMAIN_KERNEL ? "kernel" : "user space");
1728
1729 rcu_read_lock();
1730
1731 cds_lfht_lookup(state->channel_triggers_ht,
1732 hash_channel_key(&key),
1733 match_channel_trigger_list,
1734 &key,
1735 &iter);
1736 node = cds_lfht_iter_get_node(&iter);
1737 /*
1738 * There is a severe internal error if we are being asked to remove a
1739 * channel that doesn't exist.
1740 */
1741 if (!node) {
1742 ERR("[notification-thread] Channel being removed is unknown to the notification thread");
1743 goto end;
1744 }
1745
1746 /* Free the list of triggers associated with this channel. */
1747 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
1748 channel_triggers_ht_node);
1749 cds_list_for_each_entry_safe(trigger_list_element, tmp,
1750 &trigger_list->list, node) {
1751 cds_list_del(&trigger_list_element->node);
1752 free(trigger_list_element);
1753 }
1754 cds_lfht_del(state->channel_triggers_ht, node);
1755 call_rcu(&trigger_list->rcu_node, free_channel_trigger_list_rcu);
1756
1757 /* Free sampled channel state. */
1758 cds_lfht_lookup(state->channel_state_ht,
1759 hash_channel_key(&key),
1760 match_channel_state_sample,
1761 &key,
1762 &iter);
1763 node = cds_lfht_iter_get_node(&iter);
1764 /*
1765 * This is expected to be NULL if the channel is destroyed before we
1766 * received a sample.
1767 */
1768 if (node) {
1769 struct channel_state_sample *sample = caa_container_of(node,
1770 struct channel_state_sample,
1771 channel_state_ht_node);
1772
1773 cds_lfht_del(state->channel_state_ht, node);
1774 call_rcu(&sample->rcu_node, free_channel_state_sample_rcu);
1775 }
1776
1777 /* Remove the channel from the channels_ht and free it. */
1778 cds_lfht_lookup(state->channels_ht,
1779 hash_channel_key(&key),
1780 match_channel_info,
1781 &key,
1782 &iter);
1783 node = cds_lfht_iter_get_node(&iter);
1784 assert(node);
1785 channel_info = caa_container_of(node, struct channel_info,
1786 channels_ht_node);
1787 cds_lfht_del(state->channels_ht, node);
1788 channel_info_destroy(channel_info);
1789 end:
1790 rcu_read_unlock();
1791 *cmd_result = LTTNG_OK;
1792 return 0;
1793 }
1794
1795 static
1796 int handle_notification_thread_command_session_rotation(
1797 struct notification_thread_state *state,
1798 enum notification_thread_command_type cmd_type,
1799 const char *session_name, uid_t session_uid, gid_t session_gid,
1800 uint64_t trace_archive_chunk_id,
1801 struct lttng_trace_archive_location *location,
1802 enum lttng_error_code *_cmd_result)
1803 {
1804 int ret = 0;
1805 enum lttng_error_code cmd_result = LTTNG_OK;
1806 struct lttng_session_trigger_list *trigger_list;
1807 struct lttng_trigger_list_element *trigger_list_element;
1808 struct session_info *session_info;
1809 const struct lttng_credentials session_creds = {
1810 .uid = session_uid,
1811 .gid = session_gid,
1812 };
1813
1814 rcu_read_lock();
1815
1816 session_info = find_or_create_session_info(state, session_name,
1817 session_uid, session_gid);
1818 if (!session_info) {
1819 /* Allocation error or an internal error occurred. */
1820 ret = -1;
1821 cmd_result = LTTNG_ERR_NOMEM;
1822 goto end;
1823 }
1824
1825 session_info->rotation.ongoing =
1826 cmd_type == NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING;
1827 session_info->rotation.id = trace_archive_chunk_id;
1828 trigger_list = get_session_trigger_list(state, session_name);
1829 if (!trigger_list) {
1830 DBG("[notification-thread] No triggers applying to session \"%s\" found",
1831 session_name);
1832 goto end;
1833 }
1834
1835 cds_list_for_each_entry(trigger_list_element, &trigger_list->list,
1836 node) {
1837 const struct lttng_condition *condition;
1838 struct lttng_trigger *trigger;
1839 struct notification_client_list *client_list;
1840 struct lttng_evaluation *evaluation = NULL;
1841 enum lttng_condition_type condition_type;
1842 enum action_executor_status executor_status;
1843
1844 trigger = trigger_list_element->trigger;
1845 condition = lttng_trigger_get_const_condition(trigger);
1846 assert(condition);
1847 condition_type = lttng_condition_get_type(condition);
1848
1849 if (condition_type == LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING &&
1850 cmd_type != NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING) {
1851 continue;
1852 } else if (condition_type == LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED &&
1853 cmd_type != NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED) {
1854 continue;
1855 }
1856
1857 client_list = get_client_list_from_condition(state, condition);
1858 if (cmd_type == NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING) {
1859 evaluation = lttng_evaluation_session_rotation_ongoing_create(
1860 trace_archive_chunk_id);
1861 } else {
1862 evaluation = lttng_evaluation_session_rotation_completed_create(
1863 trace_archive_chunk_id, location);
1864 }
1865
1866 if (!evaluation) {
1867 /* Internal error */
1868 ret = -1;
1869 cmd_result = LTTNG_ERR_UNK;
1870 goto put_list;
1871 }
1872
1873 /*
1874 * Ownership of `evaluation` transferred to the action executor
1875 * no matter the result.
1876 */
1877 executor_status = action_executor_enqueue(state->executor,
1878 trigger, evaluation, &session_creds,
1879 client_list);
1880 evaluation = NULL;
1881 switch (executor_status) {
1882 case ACTION_EXECUTOR_STATUS_OK:
1883 break;
1884 case ACTION_EXECUTOR_STATUS_ERROR:
1885 case ACTION_EXECUTOR_STATUS_INVALID:
1886 /*
1887 * TODO Add trigger identification (name/id) when
1888 * it is added to the API.
1889 */
1890 ERR("Fatal error occurred while enqueuing action associated with session rotation trigger");
1891 ret = -1;
1892 goto put_list;
1893 case ACTION_EXECUTOR_STATUS_OVERFLOW:
1894 /*
1895 * TODO Add trigger identification (name/id) when
1896 * it is added to the API.
1897 *
1898 * Not a fatal error.
1899 */
1900 WARN("No space left when enqueuing action associated with session rotation trigger");
1901 ret = 0;
1902 goto put_list;
1903 default:
1904 abort();
1905 }
1906
1907 put_list:
1908 notification_client_list_put(client_list);
1909 if (caa_unlikely(ret)) {
1910 break;
1911 }
1912 }
1913 end:
1914 session_info_put(session_info);
1915 *_cmd_result = cmd_result;
1916 rcu_read_unlock();
1917 return ret;
1918 }
1919
1920 static
1921 int condition_is_supported(struct lttng_condition *condition)
1922 {
1923 int ret;
1924
1925 switch (lttng_condition_get_type(condition)) {
1926 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
1927 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
1928 {
1929 enum lttng_domain_type domain;
1930
1931 ret = lttng_condition_buffer_usage_get_domain_type(condition,
1932 &domain);
1933 if (ret) {
1934 ret = -1;
1935 goto end;
1936 }
1937
1938 if (domain != LTTNG_DOMAIN_KERNEL) {
1939 ret = 1;
1940 goto end;
1941 }
1942
1943 /*
1944 * Older kernel tracers don't expose the API to monitor their
1945 * buffers. Therefore, we reject triggers that require that
1946 * mechanism to be available to be evaluated.
1947 */
1948 ret = kernel_supports_ring_buffer_snapshot_sample_positions();
1949 break;
1950 }
1951 default:
1952 ret = 1;
1953 }
1954 end:
1955 return ret;
1956 }
1957
1958 /* Must be called with RCU read lock held. */
1959 static
1960 int bind_trigger_to_matching_session(struct lttng_trigger *trigger,
1961 struct notification_thread_state *state)
1962 {
1963 int ret = 0;
1964 const struct lttng_condition *condition;
1965 const char *session_name;
1966 struct lttng_session_trigger_list *trigger_list;
1967
1968 condition = lttng_trigger_get_const_condition(trigger);
1969 switch (lttng_condition_get_type(condition)) {
1970 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
1971 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
1972 {
1973 enum lttng_condition_status status;
1974
1975 status = lttng_condition_session_rotation_get_session_name(
1976 condition, &session_name);
1977 if (status != LTTNG_CONDITION_STATUS_OK) {
1978 ERR("[notification-thread] Failed to bind trigger to session: unable to get 'session_rotation' condition's session name");
1979 ret = -1;
1980 goto end;
1981 }
1982 break;
1983 }
1984 default:
1985 ret = -1;
1986 goto end;
1987 }
1988
1989 trigger_list = get_session_trigger_list(state, session_name);
1990 if (!trigger_list) {
1991 DBG("[notification-thread] Unable to bind trigger applying to session \"%s\" as it is not yet known to the notification system",
1992 session_name);
1993 goto end;
1994
1995 }
1996
1997 DBG("[notification-thread] Newly registered trigger bound to session \"%s\"",
1998 session_name);
1999 ret = lttng_session_trigger_list_add(trigger_list, trigger);
2000 end:
2001 return ret;
2002 }
2003
2004 /* Must be called with RCU read lock held. */
2005 static
2006 int bind_trigger_to_matching_channels(struct lttng_trigger *trigger,
2007 struct notification_thread_state *state)
2008 {
2009 int ret = 0;
2010 struct cds_lfht_node *node;
2011 struct cds_lfht_iter iter;
2012 struct channel_info *channel;
2013
2014 cds_lfht_for_each_entry(state->channels_ht, &iter, channel,
2015 channels_ht_node) {
2016 struct lttng_trigger_list_element *trigger_list_element;
2017 struct lttng_channel_trigger_list *trigger_list;
2018 struct cds_lfht_iter lookup_iter;
2019
2020 if (!trigger_applies_to_channel(trigger, channel)) {
2021 continue;
2022 }
2023
2024 cds_lfht_lookup(state->channel_triggers_ht,
2025 hash_channel_key(&channel->key),
2026 match_channel_trigger_list,
2027 &channel->key,
2028 &lookup_iter);
2029 node = cds_lfht_iter_get_node(&lookup_iter);
2030 assert(node);
2031 trigger_list = caa_container_of(node,
2032 struct lttng_channel_trigger_list,
2033 channel_triggers_ht_node);
2034
2035 trigger_list_element = zmalloc(sizeof(*trigger_list_element));
2036 if (!trigger_list_element) {
2037 ret = -1;
2038 goto end;
2039 }
2040 CDS_INIT_LIST_HEAD(&trigger_list_element->node);
2041 trigger_list_element->trigger = trigger;
2042 cds_list_add(&trigger_list_element->node, &trigger_list->list);
2043 DBG("[notification-thread] Newly registered trigger bound to channel \"%s\"",
2044 channel->name);
2045 }
2046 end:
2047 return ret;
2048 }
2049
2050 static
2051 bool is_trigger_action_notify(const struct lttng_trigger *trigger)
2052 {
2053 bool is_notify = false;
2054 unsigned int i, count;
2055 enum lttng_action_status action_status;
2056 const struct lttng_action *action =
2057 lttng_trigger_get_const_action(trigger);
2058 enum lttng_action_type action_type;
2059
2060 assert(action);
2061 action_type = lttng_action_get_type_const(action);
2062 if (action_type == LTTNG_ACTION_TYPE_NOTIFY) {
2063 is_notify = true;
2064 goto end;
2065 } else if (action_type != LTTNG_ACTION_TYPE_GROUP) {
2066 goto end;
2067 }
2068
2069 action_status = lttng_action_group_get_count(action, &count);
2070 assert(action_status == LTTNG_ACTION_STATUS_OK);
2071
2072 for (i = 0; i < count; i++) {
2073 const struct lttng_action *inner_action =
2074 lttng_action_group_get_at_index(
2075 action, i);
2076
2077 action_type = lttng_action_get_type_const(inner_action);
2078 if (action_type == LTTNG_ACTION_TYPE_NOTIFY) {
2079 is_notify = true;
2080 goto end;
2081 }
2082 }
2083
2084 end:
2085 return is_notify;
2086 }
2087
2088 /*
2089 * FIXME A client's credentials are not checked when registering a trigger.
2090 *
2091 * The effects of this are benign since:
2092 * - The client will succeed in registering the trigger, as it is valid,
2093 * - The trigger will, internally, be bound to the channel/session,
2094 * - The notifications will not be sent since the client's credentials
2095 * are checked against the channel at that moment.
2096 *
2097 * If this function returns a non-zero value, it means something is
2098 * fundamentally broken and the whole subsystem/thread will be torn down.
2099 *
2100 * If a non-fatal error occurs, just set the cmd_result to the appropriate
2101 * error code.
2102 */
2103 static
2104 int handle_notification_thread_command_register_trigger(
2105 struct notification_thread_state *state,
2106 struct lttng_trigger *trigger,
2107 enum lttng_error_code *cmd_result)
2108 {
2109 int ret = 0;
2110 struct lttng_condition *condition;
2111 struct notification_client *client;
2112 struct notification_client_list *client_list = NULL;
2113 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
2114 struct notification_client_list_element *client_list_element;
2115 struct cds_lfht_node *node;
2116 struct cds_lfht_iter iter;
2117 bool free_trigger = true;
2118 struct lttng_evaluation *evaluation = NULL;
2119 struct lttng_credentials object_creds;
2120 enum action_executor_status executor_status;
2121
2122 rcu_read_lock();
2123
2124 condition = lttng_trigger_get_condition(trigger);
2125 assert(condition);
2126
2127 ret = condition_is_supported(condition);
2128 if (ret < 0) {
2129 goto error;
2130 } else if (ret == 0) {
2131 *cmd_result = LTTNG_ERR_NOT_SUPPORTED;
2132 goto error;
2133 } else {
2134 /* Feature is supported, continue. */
2135 ret = 0;
2136 }
2137
2138 trigger_ht_element = zmalloc(sizeof(*trigger_ht_element));
2139 if (!trigger_ht_element) {
2140 ret = -1;
2141 goto error;
2142 }
2143
2144 /* Add trigger to the trigger_ht. */
2145 cds_lfht_node_init(&trigger_ht_element->node);
2146 trigger_ht_element->trigger = trigger;
2147
2148 node = cds_lfht_add_unique(state->triggers_ht,
2149 lttng_condition_hash(condition),
2150 match_condition,
2151 condition,
2152 &trigger_ht_element->node);
2153 if (node != &trigger_ht_element->node) {
2154 /* Not a fatal error, simply report it to the client. */
2155 *cmd_result = LTTNG_ERR_TRIGGER_EXISTS;
2156 goto error_free_ht_element;
2157 }
2158
2159 /*
2160 * Ownership of the trigger and of its wrapper was transfered to
2161 * the triggers_ht.
2162 */
2163 trigger_ht_element = NULL;
2164 free_trigger = false;
2165
2166 /*
2167 * The rest only applies to triggers that have a "notify" action.
2168 * It is not skipped as this is the only action type currently
2169 * supported.
2170 */
2171 if (is_trigger_action_notify(trigger)) {
2172 client_list = notification_client_list_create(trigger);
2173 if (!client_list) {
2174 ret = -1;
2175 goto error_free_ht_element;
2176 }
2177
2178 /* Build a list of clients to which this new trigger applies. */
2179 cds_lfht_for_each_entry (state->client_socket_ht, &iter, client,
2180 client_socket_ht_node) {
2181 if (!trigger_applies_to_client(trigger, client)) {
2182 continue;
2183 }
2184
2185 client_list_element =
2186 zmalloc(sizeof(*client_list_element));
2187 if (!client_list_element) {
2188 ret = -1;
2189 goto error_put_client_list;
2190 }
2191
2192 CDS_INIT_LIST_HEAD(&client_list_element->node);
2193 client_list_element->client = client;
2194 cds_list_add(&client_list_element->node,
2195 &client_list->list);
2196 }
2197
2198 /*
2199 * Client list ownership transferred to the
2200 * notification_trigger_clients_ht.
2201 */
2202 publish_notification_client_list(state, client_list);
2203 }
2204
2205 switch (get_condition_binding_object(condition)) {
2206 case LTTNG_OBJECT_TYPE_SESSION:
2207 /* Add the trigger to the list if it matches a known session. */
2208 ret = bind_trigger_to_matching_session(trigger, state);
2209 if (ret) {
2210 goto error_put_client_list;
2211 }
2212 break;
2213 case LTTNG_OBJECT_TYPE_CHANNEL:
2214 /*
2215 * Add the trigger to list of triggers bound to the channels
2216 * currently known.
2217 */
2218 ret = bind_trigger_to_matching_channels(trigger, state);
2219 if (ret) {
2220 goto error_put_client_list;
2221 }
2222 break;
2223 case LTTNG_OBJECT_TYPE_NONE:
2224 break;
2225 default:
2226 ERR("Unknown object type on which to bind a newly registered trigger was encountered");
2227 ret = -1;
2228 goto error_put_client_list;
2229 }
2230
2231 /*
2232 * The new trigger's condition must be evaluated against the current
2233 * state.
2234 *
2235 * In the case of `notify` action, nothing preventing clients from
2236 * subscribing to a condition before the corresponding trigger is
2237 * registered, we have to evaluate this new condition right away.
2238 *
2239 * At some point, we were waiting for the next "evaluation" (e.g. on
2240 * reception of a channel sample) to evaluate this new condition, but
2241 * that was broken.
2242 *
2243 * The reason it was broken is that waiting for the next sample
2244 * does not allow us to properly handle transitions for edge-triggered
2245 * conditions.
2246 *
2247 * Consider this example: when we handle a new channel sample, we
2248 * evaluate each conditions twice: once with the previous state, and
2249 * again with the newest state. We then use those two results to
2250 * determine whether a state change happened: a condition was false and
2251 * became true. If a state change happened, we have to notify clients.
2252 *
2253 * Now, if a client subscribes to a given notification and registers
2254 * a trigger *after* that subscription, we have to make sure the
2255 * condition is evaluated at this point while considering only the
2256 * current state. Otherwise, the next evaluation cycle may only see
2257 * that the evaluations remain the same (true for samples n-1 and n) and
2258 * the client will never know that the condition has been met.
2259 */
2260 switch (get_condition_binding_object(condition)) {
2261 case LTTNG_OBJECT_TYPE_SESSION:
2262 ret = evaluate_session_condition_for_client(condition, state,
2263 &evaluation, &object_creds.uid,
2264 &object_creds.gid);
2265 break;
2266 case LTTNG_OBJECT_TYPE_CHANNEL:
2267 ret = evaluate_channel_condition_for_client(condition, state,
2268 &evaluation, &object_creds.uid,
2269 &object_creds.gid);
2270 break;
2271 case LTTNG_OBJECT_TYPE_NONE:
2272 ret = 0;
2273 goto error_put_client_list;
2274 case LTTNG_OBJECT_TYPE_UNKNOWN:
2275 default:
2276 ret = -1;
2277 goto error_put_client_list;
2278 }
2279
2280 if (ret) {
2281 /* Fatal error. */
2282 goto error_put_client_list;
2283 }
2284
2285 DBG("Newly registered trigger's condition evaluated to %s",
2286 evaluation ? "true" : "false");
2287 if (!evaluation) {
2288 /* Evaluation yielded nothing. Normal exit. */
2289 ret = 0;
2290 goto error_put_client_list;
2291 }
2292
2293 /*
2294 * Ownership of `evaluation` transferred to the action executor
2295 * no matter the result.
2296 */
2297 executor_status = action_executor_enqueue(state->executor, trigger,
2298 evaluation, &object_creds, client_list);
2299 evaluation = NULL;
2300 switch (executor_status) {
2301 case ACTION_EXECUTOR_STATUS_OK:
2302 break;
2303 case ACTION_EXECUTOR_STATUS_ERROR:
2304 case ACTION_EXECUTOR_STATUS_INVALID:
2305 /*
2306 * TODO Add trigger identification (name/id) when
2307 * it is added to the API.
2308 */
2309 ERR("Fatal error occurred while enqueuing action associated to newly registered trigger");
2310 ret = -1;
2311 goto error_put_client_list;
2312 case ACTION_EXECUTOR_STATUS_OVERFLOW:
2313 /*
2314 * TODO Add trigger identification (name/id) when
2315 * it is added to the API.
2316 *
2317 * Not a fatal error.
2318 */
2319 WARN("No space left when enqueuing action associated to newly registered trigger");
2320 ret = 0;
2321 goto error_put_client_list;
2322 default:
2323 abort();
2324 }
2325
2326 *cmd_result = LTTNG_OK;
2327
2328 error_put_client_list:
2329 notification_client_list_put(client_list);
2330
2331 error_free_ht_element:
2332 free(trigger_ht_element);
2333 error:
2334 if (free_trigger) {
2335 lttng_trigger_destroy(trigger);
2336 }
2337 rcu_read_unlock();
2338 return ret;
2339 }
2340
2341 static
2342 void free_lttng_trigger_ht_element_rcu(struct rcu_head *node)
2343 {
2344 free(caa_container_of(node, struct lttng_trigger_ht_element,
2345 rcu_node));
2346 }
2347
2348 static
2349 int handle_notification_thread_command_unregister_trigger(
2350 struct notification_thread_state *state,
2351 struct lttng_trigger *trigger,
2352 enum lttng_error_code *_cmd_reply)
2353 {
2354 struct cds_lfht_iter iter;
2355 struct cds_lfht_node *triggers_ht_node;
2356 struct lttng_channel_trigger_list *trigger_list;
2357 struct notification_client_list *client_list;
2358 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
2359 struct lttng_condition *condition = lttng_trigger_get_condition(
2360 trigger);
2361 enum lttng_error_code cmd_reply;
2362
2363 rcu_read_lock();
2364
2365 cds_lfht_lookup(state->triggers_ht,
2366 lttng_condition_hash(condition),
2367 match_condition,
2368 condition,
2369 &iter);
2370 triggers_ht_node = cds_lfht_iter_get_node(&iter);
2371 if (!triggers_ht_node) {
2372 cmd_reply = LTTNG_ERR_TRIGGER_NOT_FOUND;
2373 goto end;
2374 } else {
2375 cmd_reply = LTTNG_OK;
2376 }
2377
2378 /* Remove trigger from channel_triggers_ht. */
2379 cds_lfht_for_each_entry(state->channel_triggers_ht, &iter, trigger_list,
2380 channel_triggers_ht_node) {
2381 struct lttng_trigger_list_element *trigger_element, *tmp;
2382
2383 cds_list_for_each_entry_safe(trigger_element, tmp,
2384 &trigger_list->list, node) {
2385 const struct lttng_condition *current_condition =
2386 lttng_trigger_get_const_condition(
2387 trigger_element->trigger);
2388
2389 assert(current_condition);
2390 if (!lttng_condition_is_equal(condition,
2391 current_condition)) {
2392 continue;
2393 }
2394
2395 DBG("[notification-thread] Removed trigger from channel_triggers_ht");
2396 cds_list_del(&trigger_element->node);
2397 /* A trigger can only appear once per channel */
2398 break;
2399 }
2400 }
2401
2402 /*
2403 * Remove and release the client list from
2404 * notification_trigger_clients_ht.
2405 */
2406 client_list = get_client_list_from_condition(state, condition);
2407 assert(client_list);
2408
2409 /* Put new reference and the hashtable's reference. */
2410 notification_client_list_put(client_list);
2411 notification_client_list_put(client_list);
2412 client_list = NULL;
2413
2414 /* Remove trigger from triggers_ht. */
2415 trigger_ht_element = caa_container_of(triggers_ht_node,
2416 struct lttng_trigger_ht_element, node);
2417 cds_lfht_del(state->triggers_ht, triggers_ht_node);
2418
2419 /* Release the ownership of the trigger. */
2420 lttng_trigger_destroy(trigger_ht_element->trigger);
2421 call_rcu(&trigger_ht_element->rcu_node, free_lttng_trigger_ht_element_rcu);
2422 end:
2423 rcu_read_unlock();
2424 if (_cmd_reply) {
2425 *_cmd_reply = cmd_reply;
2426 }
2427 return 0;
2428 }
2429
2430 /* Returns 0 on success, 1 on exit requested, negative value on error. */
2431 int handle_notification_thread_command(
2432 struct notification_thread_handle *handle,
2433 struct notification_thread_state *state)
2434 {
2435 int ret;
2436 uint64_t counter;
2437 struct notification_thread_command *cmd;
2438
2439 /* Read the event pipe to put it back into a quiescent state. */
2440 ret = lttng_read(lttng_pipe_get_readfd(handle->cmd_queue.event_pipe), &counter,
2441 sizeof(counter));
2442 if (ret != sizeof(counter)) {
2443 goto error;
2444 }
2445
2446 pthread_mutex_lock(&handle->cmd_queue.lock);
2447 cmd = cds_list_first_entry(&handle->cmd_queue.list,
2448 struct notification_thread_command, cmd_list_node);
2449 cds_list_del(&cmd->cmd_list_node);
2450 pthread_mutex_unlock(&handle->cmd_queue.lock);
2451 switch (cmd->type) {
2452 case NOTIFICATION_COMMAND_TYPE_REGISTER_TRIGGER:
2453 DBG("[notification-thread] Received register trigger command");
2454 ret = handle_notification_thread_command_register_trigger(
2455 state, cmd->parameters.trigger,
2456 &cmd->reply_code);
2457 break;
2458 case NOTIFICATION_COMMAND_TYPE_UNREGISTER_TRIGGER:
2459 DBG("[notification-thread] Received unregister trigger command");
2460 ret = handle_notification_thread_command_unregister_trigger(
2461 state, cmd->parameters.trigger,
2462 &cmd->reply_code);
2463 break;
2464 case NOTIFICATION_COMMAND_TYPE_ADD_CHANNEL:
2465 DBG("[notification-thread] Received add channel command");
2466 ret = handle_notification_thread_command_add_channel(
2467 state,
2468 cmd->parameters.add_channel.session.name,
2469 cmd->parameters.add_channel.session.uid,
2470 cmd->parameters.add_channel.session.gid,
2471 cmd->parameters.add_channel.channel.name,
2472 cmd->parameters.add_channel.channel.domain,
2473 cmd->parameters.add_channel.channel.key,
2474 cmd->parameters.add_channel.channel.capacity,
2475 &cmd->reply_code);
2476 break;
2477 case NOTIFICATION_COMMAND_TYPE_REMOVE_CHANNEL:
2478 DBG("[notification-thread] Received remove channel command");
2479 ret = handle_notification_thread_command_remove_channel(
2480 state, cmd->parameters.remove_channel.key,
2481 cmd->parameters.remove_channel.domain,
2482 &cmd->reply_code);
2483 break;
2484 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING:
2485 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED:
2486 DBG("[notification-thread] Received session rotation %s command",
2487 cmd->type == NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING ?
2488 "ongoing" : "completed");
2489 ret = handle_notification_thread_command_session_rotation(
2490 state,
2491 cmd->type,
2492 cmd->parameters.session_rotation.session_name,
2493 cmd->parameters.session_rotation.uid,
2494 cmd->parameters.session_rotation.gid,
2495 cmd->parameters.session_rotation.trace_archive_chunk_id,
2496 cmd->parameters.session_rotation.location,
2497 &cmd->reply_code);
2498 break;
2499 case NOTIFICATION_COMMAND_TYPE_QUIT:
2500 DBG("[notification-thread] Received quit command");
2501 cmd->reply_code = LTTNG_OK;
2502 ret = 1;
2503 goto end;
2504 case NOTIFICATION_COMMAND_TYPE_CLIENT_COMMUNICATION_UPDATE:
2505 {
2506 const enum client_transmission_status client_status =
2507 cmd->parameters.client_communication_update
2508 .status;
2509 const notification_client_id client_id =
2510 cmd->parameters.client_communication_update.id;
2511 struct notification_client *client;
2512
2513 rcu_read_lock();
2514 client = get_client_from_id(client_id, state);
2515
2516 if (!client) {
2517 /*
2518 * Client error was probably already picked-up by the
2519 * notification thread or it has disconnected
2520 * gracefully while this command was queued.
2521 */
2522 DBG("Failed to find notification client to update communication status, client id = %" PRIu64,
2523 client_id);
2524 ret = 0;
2525 } else {
2526 pthread_mutex_lock(&client->lock);
2527 ret = client_handle_transmission_status(
2528 client, client_status, state);
2529 pthread_mutex_unlock(&client->lock);
2530 }
2531 rcu_read_unlock();
2532 break;
2533 }
2534 default:
2535 ERR("[notification-thread] Unknown internal command received");
2536 goto error_unlock;
2537 }
2538
2539 if (ret) {
2540 goto error_unlock;
2541 }
2542 end:
2543 if (cmd->is_async) {
2544 free(cmd);
2545 cmd = NULL;
2546 } else {
2547 lttng_waiter_wake_up(&cmd->reply_waiter);
2548 }
2549 return ret;
2550 error_unlock:
2551 /* Wake-up and return a fatal error to the calling thread. */
2552 lttng_waiter_wake_up(&cmd->reply_waiter);
2553 cmd->reply_code = LTTNG_ERR_FATAL;
2554 error:
2555 /* Indicate a fatal error to the caller. */
2556 return -1;
2557 }
2558
2559 static
2560 int socket_set_non_blocking(int socket)
2561 {
2562 int ret, flags;
2563
2564 /* Set the pipe as non-blocking. */
2565 ret = fcntl(socket, F_GETFL, 0);
2566 if (ret == -1) {
2567 PERROR("fcntl get socket flags");
2568 goto end;
2569 }
2570 flags = ret;
2571
2572 ret = fcntl(socket, F_SETFL, flags | O_NONBLOCK);
2573 if (ret == -1) {
2574 PERROR("fcntl set O_NONBLOCK socket flag");
2575 goto end;
2576 }
2577 DBG("Client socket (fd = %i) set as non-blocking", socket);
2578 end:
2579 return ret;
2580 }
2581
2582 /* Client lock must be acquired by caller. */
2583 static
2584 int client_reset_inbound_state(struct notification_client *client)
2585 {
2586 int ret;
2587
2588 ASSERT_LOCKED(client->lock);
2589
2590 ret = lttng_dynamic_buffer_set_size(
2591 &client->communication.inbound.buffer, 0);
2592 assert(!ret);
2593
2594 client->communication.inbound.bytes_to_receive =
2595 sizeof(struct lttng_notification_channel_message);
2596 client->communication.inbound.msg_type =
2597 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN;
2598 LTTNG_SOCK_SET_UID_CRED(&client->communication.inbound.creds, -1);
2599 LTTNG_SOCK_SET_GID_CRED(&client->communication.inbound.creds, -1);
2600 ret = lttng_dynamic_buffer_set_size(
2601 &client->communication.inbound.buffer,
2602 client->communication.inbound.bytes_to_receive);
2603 return ret;
2604 }
2605
2606 int handle_notification_thread_client_connect(
2607 struct notification_thread_state *state)
2608 {
2609 int ret;
2610 struct notification_client *client;
2611
2612 DBG("[notification-thread] Handling new notification channel client connection");
2613
2614 client = zmalloc(sizeof(*client));
2615 if (!client) {
2616 /* Fatal error. */
2617 ret = -1;
2618 goto error;
2619 }
2620 pthread_mutex_init(&client->lock, NULL);
2621 client->id = state->next_notification_client_id++;
2622 CDS_INIT_LIST_HEAD(&client->condition_list);
2623 lttng_dynamic_buffer_init(&client->communication.inbound.buffer);
2624 lttng_dynamic_buffer_init(&client->communication.outbound.buffer);
2625 client->communication.inbound.expect_creds = true;
2626
2627 pthread_mutex_lock(&client->lock);
2628 ret = client_reset_inbound_state(client);
2629 pthread_mutex_unlock(&client->lock);
2630 if (ret) {
2631 ERR("[notification-thread] Failed to reset client communication's inbound state");
2632 ret = 0;
2633 goto error;
2634 }
2635
2636 ret = lttcomm_accept_unix_sock(state->notification_channel_socket);
2637 if (ret < 0) {
2638 ERR("[notification-thread] Failed to accept new notification channel client connection");
2639 ret = 0;
2640 goto error;
2641 }
2642
2643 client->socket = ret;
2644
2645 ret = socket_set_non_blocking(client->socket);
2646 if (ret) {
2647 ERR("[notification-thread] Failed to set new notification channel client connection socket as non-blocking");
2648 goto error;
2649 }
2650
2651 ret = lttcomm_setsockopt_creds_unix_sock(client->socket);
2652 if (ret < 0) {
2653 ERR("[notification-thread] Failed to set socket options on new notification channel client socket");
2654 ret = 0;
2655 goto error;
2656 }
2657
2658 ret = lttng_poll_add(&state->events, client->socket,
2659 LPOLLIN | LPOLLERR |
2660 LPOLLHUP | LPOLLRDHUP);
2661 if (ret < 0) {
2662 ERR("[notification-thread] Failed to add notification channel client socket to poll set");
2663 ret = 0;
2664 goto error;
2665 }
2666 DBG("[notification-thread] Added new notification channel client socket (%i) to poll set",
2667 client->socket);
2668
2669 rcu_read_lock();
2670 cds_lfht_add(state->client_socket_ht,
2671 hash_client_socket(client->socket),
2672 &client->client_socket_ht_node);
2673 cds_lfht_add(state->client_id_ht,
2674 hash_client_id(client->id),
2675 &client->client_id_ht_node);
2676 rcu_read_unlock();
2677
2678 return ret;
2679 error:
2680 notification_client_destroy(client, state);
2681 return ret;
2682 }
2683
2684 /* RCU read-lock must be held by the caller. */
2685 /* Client lock must be held by the caller */
2686 static
2687 int notification_thread_client_disconnect(
2688 struct notification_client *client,
2689 struct notification_thread_state *state)
2690 {
2691 int ret;
2692 struct lttng_condition_list_element *condition_list_element, *tmp;
2693
2694 /* Acquire the client lock to disable its communication atomically. */
2695 client->communication.active = false;
2696 ret = lttng_poll_del(&state->events, client->socket);
2697 if (ret) {
2698 ERR("[notification-thread] Failed to remove client socket %d from poll set",
2699 client->socket);
2700 }
2701
2702 cds_lfht_del(state->client_socket_ht, &client->client_socket_ht_node);
2703 cds_lfht_del(state->client_id_ht, &client->client_id_ht_node);
2704
2705 /* Release all conditions to which the client was subscribed. */
2706 cds_list_for_each_entry_safe(condition_list_element, tmp,
2707 &client->condition_list, node) {
2708 (void) notification_thread_client_unsubscribe(client,
2709 condition_list_element->condition, state, NULL);
2710 }
2711
2712 /*
2713 * Client no longer accessible to other threads (through the
2714 * client lists).
2715 */
2716 notification_client_destroy(client, state);
2717 return ret;
2718 }
2719
2720 int handle_notification_thread_client_disconnect(
2721 int client_socket, struct notification_thread_state *state)
2722 {
2723 int ret = 0;
2724 struct notification_client *client;
2725
2726 rcu_read_lock();
2727 DBG("[notification-thread] Closing client connection (socket fd = %i)",
2728 client_socket);
2729 client = get_client_from_socket(client_socket, state);
2730 if (!client) {
2731 /* Internal state corruption, fatal error. */
2732 ERR("[notification-thread] Unable to find client (socket fd = %i)",
2733 client_socket);
2734 ret = -1;
2735 goto end;
2736 }
2737
2738 pthread_mutex_lock(&client->lock);
2739 ret = notification_thread_client_disconnect(client, state);
2740 pthread_mutex_unlock(&client->lock);
2741 end:
2742 rcu_read_unlock();
2743 return ret;
2744 }
2745
2746 int handle_notification_thread_client_disconnect_all(
2747 struct notification_thread_state *state)
2748 {
2749 struct cds_lfht_iter iter;
2750 struct notification_client *client;
2751 bool error_encoutered = false;
2752
2753 rcu_read_lock();
2754 DBG("[notification-thread] Closing all client connections");
2755 cds_lfht_for_each_entry(state->client_socket_ht, &iter, client,
2756 client_socket_ht_node) {
2757 int ret;
2758
2759 pthread_mutex_lock(&client->lock);
2760 ret = notification_thread_client_disconnect(
2761 client, state);
2762 pthread_mutex_unlock(&client->lock);
2763 if (ret) {
2764 error_encoutered = true;
2765 }
2766 }
2767 rcu_read_unlock();
2768 return error_encoutered ? 1 : 0;
2769 }
2770
2771 int handle_notification_thread_trigger_unregister_all(
2772 struct notification_thread_state *state)
2773 {
2774 bool error_occurred = false;
2775 struct cds_lfht_iter iter;
2776 struct lttng_trigger_ht_element *trigger_ht_element;
2777
2778 rcu_read_lock();
2779 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
2780 node) {
2781 int ret = handle_notification_thread_command_unregister_trigger(
2782 state, trigger_ht_element->trigger, NULL);
2783 if (ret) {
2784 error_occurred = true;
2785 }
2786 }
2787 rcu_read_unlock();
2788 return error_occurred ? -1 : 0;
2789 }
2790
2791 static
2792 int client_handle_transmission_status(
2793 struct notification_client *client,
2794 enum client_transmission_status transmission_status,
2795 struct notification_thread_state *state)
2796 {
2797 int ret = 0;
2798
2799 switch (transmission_status) {
2800 case CLIENT_TRANSMISSION_STATUS_COMPLETE:
2801 ret = lttng_poll_mod(&state->events, client->socket,
2802 CLIENT_POLL_MASK_IN);
2803 if (ret) {
2804 goto end;
2805 }
2806
2807 client->communication.outbound.queued_command_reply = false;
2808 client->communication.outbound.dropped_notification = false;
2809 break;
2810 case CLIENT_TRANSMISSION_STATUS_QUEUED:
2811 /*
2812 * We want to be notified whenever there is buffer space
2813 * available to send the rest of the payload.
2814 */
2815 ret = lttng_poll_mod(&state->events, client->socket,
2816 CLIENT_POLL_MASK_IN_OUT);
2817 if (ret) {
2818 goto end;
2819 }
2820 break;
2821 case CLIENT_TRANSMISSION_STATUS_FAIL:
2822 ret = notification_thread_client_disconnect(client, state);
2823 if (ret) {
2824 goto end;
2825 }
2826 break;
2827 case CLIENT_TRANSMISSION_STATUS_ERROR:
2828 ret = -1;
2829 goto end;
2830 default:
2831 abort();
2832 }
2833 end:
2834 return ret;
2835 }
2836
2837 /* Client lock must be acquired by caller. */
2838 static
2839 enum client_transmission_status client_flush_outgoing_queue(
2840 struct notification_client *client)
2841 {
2842 ssize_t ret;
2843 size_t to_send_count;
2844 enum client_transmission_status status;
2845
2846 ASSERT_LOCKED(client->lock);
2847
2848 if (!client->communication.active) {
2849 status = CLIENT_TRANSMISSION_STATUS_FAIL;
2850 goto end;
2851 }
2852
2853 assert(client->communication.outbound.buffer.size != 0);
2854 to_send_count = client->communication.outbound.buffer.size;
2855 DBG("[notification-thread] Flushing client (socket fd = %i) outgoing queue",
2856 client->socket);
2857
2858 ret = lttcomm_send_unix_sock_non_block(client->socket,
2859 client->communication.outbound.buffer.data,
2860 to_send_count);
2861 if ((ret >= 0 && ret < to_send_count)) {
2862 DBG("[notification-thread] Client (socket fd = %i) outgoing queue could not be completely flushed",
2863 client->socket);
2864 to_send_count -= max(ret, 0);
2865
2866 memcpy(client->communication.outbound.buffer.data,
2867 client->communication.outbound.buffer.data +
2868 client->communication.outbound.buffer.size - to_send_count,
2869 to_send_count);
2870 ret = lttng_dynamic_buffer_set_size(
2871 &client->communication.outbound.buffer,
2872 to_send_count);
2873 if (ret) {
2874 goto error;
2875 }
2876 status = CLIENT_TRANSMISSION_STATUS_QUEUED;
2877 } else if (ret < 0) {
2878 /* Generic error, disconnect the client. */
2879 ERR("[notification-thread] Failed to flush outgoing queue, disconnecting client (socket fd = %i)",
2880 client->socket);
2881 status = CLIENT_TRANSMISSION_STATUS_FAIL;
2882 } else {
2883 /* No error and flushed the queue completely. */
2884 ret = lttng_dynamic_buffer_set_size(
2885 &client->communication.outbound.buffer, 0);
2886 if (ret) {
2887 goto error;
2888 }
2889 status = CLIENT_TRANSMISSION_STATUS_COMPLETE;
2890 }
2891 end:
2892 return status;
2893 error:
2894 return CLIENT_TRANSMISSION_STATUS_ERROR;
2895 }
2896
2897 /* Client lock must be acquired by caller. */
2898 static
2899 int client_send_command_reply(struct notification_client *client,
2900 struct notification_thread_state *state,
2901 enum lttng_notification_channel_status status)
2902 {
2903 int ret;
2904 struct lttng_notification_channel_command_reply reply = {
2905 .status = (int8_t) status,
2906 };
2907 struct lttng_notification_channel_message msg = {
2908 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY,
2909 .size = sizeof(reply),
2910 };
2911 char buffer[sizeof(msg) + sizeof(reply)];
2912 enum client_transmission_status transmission_status;
2913
2914 ASSERT_LOCKED(client->lock);
2915
2916 if (client->communication.outbound.queued_command_reply) {
2917 /* Protocol error. */
2918 goto error;
2919 }
2920
2921 memcpy(buffer, &msg, sizeof(msg));
2922 memcpy(buffer + sizeof(msg), &reply, sizeof(reply));
2923 DBG("[notification-thread] Send command reply (%i)", (int) status);
2924
2925 /* Enqueue buffer to outgoing queue and flush it. */
2926 ret = lttng_dynamic_buffer_append(
2927 &client->communication.outbound.buffer,
2928 buffer, sizeof(buffer));
2929 if (ret) {
2930 goto error;
2931 }
2932
2933 transmission_status = client_flush_outgoing_queue(client);
2934 ret = client_handle_transmission_status(
2935 client, transmission_status, state);
2936 if (ret) {
2937 goto error;
2938 }
2939
2940 if (client->communication.outbound.buffer.size != 0) {
2941 /* Queue could not be emptied. */
2942 client->communication.outbound.queued_command_reply = true;
2943 }
2944
2945 return 0;
2946 error:
2947 return -1;
2948 }
2949
2950 static
2951 int client_handle_message_unknown(struct notification_client *client,
2952 struct notification_thread_state *state)
2953 {
2954 int ret;
2955
2956 pthread_mutex_lock(&client->lock);
2957
2958 /*
2959 * Receiving message header. The function will be called again
2960 * once the rest of the message as been received and can be
2961 * interpreted.
2962 */
2963 const struct lttng_notification_channel_message *msg;
2964
2965 assert(sizeof(*msg) == client->communication.inbound.buffer.size);
2966 msg = (const struct lttng_notification_channel_message *)
2967 client->communication.inbound.buffer.data;
2968
2969 if (msg->size == 0 ||
2970 msg->size > DEFAULT_MAX_NOTIFICATION_CLIENT_MESSAGE_PAYLOAD_SIZE) {
2971 ERR("[notification-thread] Invalid notification channel message: length = %u",
2972 msg->size);
2973 ret = -1;
2974 goto end;
2975 }
2976
2977 switch (msg->type) {
2978 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE:
2979 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE:
2980 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
2981 break;
2982 default:
2983 ret = -1;
2984 ERR("[notification-thread] Invalid notification channel message: unexpected message type");
2985 goto end;
2986 }
2987
2988 client->communication.inbound.bytes_to_receive = msg->size;
2989 client->communication.inbound.msg_type =
2990 (enum lttng_notification_channel_message_type) msg->type;
2991 ret = lttng_dynamic_buffer_set_size(
2992 &client->communication.inbound.buffer, msg->size);
2993 end:
2994 pthread_mutex_unlock(&client->lock);
2995 return ret;
2996 }
2997
2998 static
2999 int client_handle_message_handshake(struct notification_client *client,
3000 struct notification_thread_state *state)
3001 {
3002 int ret;
3003 struct lttng_notification_channel_command_handshake *handshake_client;
3004 const struct lttng_notification_channel_command_handshake handshake_reply = {
3005 .major = LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR,
3006 .minor = LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR,
3007 };
3008 const struct lttng_notification_channel_message msg_header = {
3009 .type = LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE,
3010 .size = sizeof(handshake_reply),
3011 };
3012 enum lttng_notification_channel_status status =
3013 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
3014 char send_buffer[sizeof(msg_header) + sizeof(handshake_reply)];
3015 enum client_transmission_status transmission_status;
3016
3017 pthread_mutex_lock(&client->lock);
3018
3019 memcpy(send_buffer, &msg_header, sizeof(msg_header));
3020 memcpy(send_buffer + sizeof(msg_header), &handshake_reply,
3021 sizeof(handshake_reply));
3022
3023 handshake_client =
3024 (struct lttng_notification_channel_command_handshake *)
3025 client->communication.inbound.buffer
3026 .data;
3027 client->major = handshake_client->major;
3028 client->minor = handshake_client->minor;
3029 if (!client->communication.inbound.creds_received) {
3030 ERR("[notification-thread] No credentials received from client");
3031 ret = -1;
3032 goto end;
3033 }
3034
3035 client->uid = LTTNG_SOCK_GET_UID_CRED(
3036 &client->communication.inbound.creds);
3037 client->gid = LTTNG_SOCK_GET_GID_CRED(
3038 &client->communication.inbound.creds);
3039 DBG("[notification-thread] Received handshake from client (uid = %u, gid = %u) with version %i.%i",
3040 client->uid, client->gid, (int) client->major,
3041 (int) client->minor);
3042
3043 if (handshake_client->major !=
3044 LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR) {
3045 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNSUPPORTED_VERSION;
3046 }
3047
3048 ret = lttng_dynamic_buffer_append(
3049 &client->communication.outbound.buffer, send_buffer,
3050 sizeof(send_buffer));
3051 if (ret) {
3052 ERR("[notification-thread] Failed to send protocol version to notification channel client");
3053 goto end;
3054 }
3055
3056 client->validated = true;
3057 client->communication.active = true;
3058
3059 transmission_status = client_flush_outgoing_queue(client);
3060 ret = client_handle_transmission_status(
3061 client, transmission_status, state);
3062 if (ret) {
3063 goto end;
3064 }
3065
3066 ret = client_send_command_reply(client, state, status);
3067 if (ret) {
3068 ERR("[notification-thread] Failed to send reply to notification channel client");
3069 goto end;
3070 }
3071
3072 /* Set reception state to receive the next message header. */
3073 ret = client_reset_inbound_state(client);
3074 if (ret) {
3075 ERR("[notification-thread] Failed to reset client communication's inbound state");
3076 goto end;
3077 }
3078
3079 end:
3080 pthread_mutex_unlock(&client->lock);
3081 return ret;
3082 }
3083
3084 static
3085 int client_handle_message_subscription(
3086 struct notification_client *client,
3087 enum lttng_notification_channel_message_type msg_type,
3088 struct notification_thread_state *state)
3089 {
3090 int ret;
3091 struct lttng_condition *condition;
3092 enum lttng_notification_channel_status status =
3093 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
3094 struct lttng_payload_view condition_view =
3095 lttng_payload_view_from_dynamic_buffer(
3096 &client->communication.inbound.buffer,
3097 0, -1);
3098 size_t expected_condition_size;
3099
3100 pthread_mutex_lock(&client->lock);
3101 expected_condition_size = client->communication.inbound.buffer.size;
3102 pthread_mutex_unlock(&client->lock);
3103
3104 ret = lttng_condition_create_from_payload(&condition_view, &condition);
3105 if (ret != expected_condition_size) {
3106 ERR("[notification-thread] Malformed condition received from client");
3107 goto end;
3108 }
3109
3110 if (msg_type == LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE) {
3111 ret = notification_thread_client_subscribe(
3112 client, condition, state, &status);
3113 } else {
3114 ret = notification_thread_client_unsubscribe(
3115 client, condition, state, &status);
3116 }
3117 if (ret) {
3118 goto end;
3119 }
3120
3121 pthread_mutex_lock(&client->lock);
3122 ret = client_send_command_reply(client, state, status);
3123 if (ret) {
3124 ERR("[notification-thread] Failed to send reply to notification channel client");
3125 goto end_unlock;
3126 }
3127
3128 /* Set reception state to receive the next message header. */
3129 ret = client_reset_inbound_state(client);
3130 if (ret) {
3131 ERR("[notification-thread] Failed to reset client communication's inbound state");
3132 goto end_unlock;
3133 }
3134
3135 end_unlock:
3136 pthread_mutex_unlock(&client->lock);
3137 end:
3138 return ret;
3139 }
3140
3141 static
3142 int client_dispatch_message(struct notification_client *client,
3143 struct notification_thread_state *state)
3144 {
3145 int ret = 0;
3146
3147 if (client->communication.inbound.msg_type !=
3148 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE &&
3149 client->communication.inbound.msg_type !=
3150 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN &&
3151 !client->validated) {
3152 WARN("[notification-thread] client attempted a command before handshake");
3153 ret = -1;
3154 goto end;
3155 }
3156
3157 switch (client->communication.inbound.msg_type) {
3158 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN:
3159 {
3160 ret = client_handle_message_unknown(client, state);
3161 break;
3162 }
3163 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
3164 {
3165 ret = client_handle_message_handshake(client, state);
3166 break;
3167 }
3168 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE:
3169 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE:
3170 {
3171 ret = client_handle_message_subscription(client,
3172 client->communication.inbound.msg_type, state);
3173 break;
3174 }
3175 default:
3176 abort();
3177 }
3178 end:
3179 return ret;
3180 }
3181
3182 /* Incoming data from client. */
3183 int handle_notification_thread_client_in(
3184 struct notification_thread_state *state, int socket)
3185 {
3186 int ret = 0;
3187 struct notification_client *client;
3188 ssize_t recv_ret;
3189 size_t offset;
3190 bool message_is_complete = false;
3191
3192 rcu_read_lock();
3193 client = get_client_from_socket(socket, state);
3194 if (!client) {
3195 /* Internal error, abort. */
3196 ret = -1;
3197 goto end;
3198 }
3199
3200 pthread_mutex_lock(&client->lock);
3201 offset = client->communication.inbound.buffer.size -
3202 client->communication.inbound.bytes_to_receive;
3203 if (client->communication.inbound.expect_creds) {
3204 recv_ret = lttcomm_recv_creds_unix_sock(socket,
3205 client->communication.inbound.buffer.data + offset,
3206 client->communication.inbound.bytes_to_receive,
3207 &client->communication.inbound.creds);
3208 if (recv_ret > 0) {
3209 client->communication.inbound.expect_creds = false;
3210 client->communication.inbound.creds_received = true;
3211 }
3212 } else {
3213 recv_ret = lttcomm_recv_unix_sock_non_block(socket,
3214 client->communication.inbound.buffer.data + offset,
3215 client->communication.inbound.bytes_to_receive);
3216 }
3217 if (recv_ret >= 0) {
3218 client->communication.inbound.bytes_to_receive -= recv_ret;
3219 message_is_complete = client->communication.inbound
3220 .bytes_to_receive == 0;
3221 }
3222 pthread_mutex_unlock(&client->lock);
3223 if (recv_ret < 0) {
3224 goto error_disconnect_client;
3225 }
3226
3227 if (message_is_complete) {
3228 ret = client_dispatch_message(client, state);
3229 if (ret) {
3230 /*
3231 * Only returns an error if this client must be
3232 * disconnected.
3233 */
3234 goto error_disconnect_client;
3235 }
3236 }
3237 end:
3238 rcu_read_unlock();
3239 return ret;
3240 error_disconnect_client:
3241 pthread_mutex_lock(&client->lock);
3242 ret = notification_thread_client_disconnect(client, state);
3243 pthread_mutex_unlock(&client->lock);
3244 goto end;
3245 }
3246
3247 /* Client ready to receive outgoing data. */
3248 int handle_notification_thread_client_out(
3249 struct notification_thread_state *state, int socket)
3250 {
3251 int ret;
3252 struct notification_client *client;
3253 enum client_transmission_status transmission_status;
3254
3255 rcu_read_lock();
3256 client = get_client_from_socket(socket, state);
3257 if (!client) {
3258 /* Internal error, abort. */
3259 ret = -1;
3260 goto end;
3261 }
3262
3263 pthread_mutex_lock(&client->lock);
3264 transmission_status = client_flush_outgoing_queue(client);
3265 ret = client_handle_transmission_status(
3266 client, transmission_status, state);
3267 pthread_mutex_unlock(&client->lock);
3268 if (ret) {
3269 goto end;
3270 }
3271 end:
3272 rcu_read_unlock();
3273 return ret;
3274 }
3275
3276 static
3277 bool evaluate_buffer_usage_condition(const struct lttng_condition *condition,
3278 const struct channel_state_sample *sample,
3279 uint64_t buffer_capacity)
3280 {
3281 bool result = false;
3282 uint64_t threshold;
3283 enum lttng_condition_type condition_type;
3284 const struct lttng_condition_buffer_usage *use_condition = container_of(
3285 condition, struct lttng_condition_buffer_usage,
3286 parent);
3287
3288 if (use_condition->threshold_bytes.set) {
3289 threshold = use_condition->threshold_bytes.value;
3290 } else {
3291 /*
3292 * Threshold was expressed as a ratio.
3293 *
3294 * TODO the threshold (in bytes) of conditions expressed
3295 * as a ratio of total buffer size could be cached to
3296 * forego this double-multiplication or it could be performed
3297 * as fixed-point math.
3298 *
3299 * Note that caching should accommodates the case where the
3300 * condition applies to multiple channels (i.e. don't assume
3301 * that all channels matching my_chann* have the same size...)
3302 */
3303 threshold = (uint64_t) (use_condition->threshold_ratio.value *
3304 (double) buffer_capacity);
3305 }
3306
3307 condition_type = lttng_condition_get_type(condition);
3308 if (condition_type == LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW) {
3309 DBG("[notification-thread] Low buffer usage condition being evaluated: threshold = %" PRIu64 ", highest usage = %" PRIu64,
3310 threshold, sample->highest_usage);
3311
3312 /*
3313 * The low condition should only be triggered once _all_ of the
3314 * streams in a channel have gone below the "low" threshold.
3315 */
3316 if (sample->highest_usage <= threshold) {
3317 result = true;
3318 }
3319 } else {
3320 DBG("[notification-thread] High buffer usage condition being evaluated: threshold = %" PRIu64 ", highest usage = %" PRIu64,
3321 threshold, sample->highest_usage);
3322
3323 /*
3324 * For high buffer usage scenarios, we want to trigger whenever
3325 * _any_ of the streams has reached the "high" threshold.
3326 */
3327 if (sample->highest_usage >= threshold) {
3328 result = true;
3329 }
3330 }
3331
3332 return result;
3333 }
3334
3335 static
3336 bool evaluate_session_consumed_size_condition(
3337 const struct lttng_condition *condition,
3338 uint64_t session_consumed_size)
3339 {
3340 uint64_t threshold;
3341 const struct lttng_condition_session_consumed_size *size_condition =
3342 container_of(condition,
3343 struct lttng_condition_session_consumed_size,
3344 parent);
3345
3346 threshold = size_condition->consumed_threshold_bytes.value;
3347 DBG("[notification-thread] Session consumed size condition being evaluated: threshold = %" PRIu64 ", current size = %" PRIu64,
3348 threshold, session_consumed_size);
3349 return session_consumed_size >= threshold;
3350 }
3351
3352 static
3353 int evaluate_buffer_condition(const struct lttng_condition *condition,
3354 struct lttng_evaluation **evaluation,
3355 const struct notification_thread_state *state,
3356 const struct channel_state_sample *previous_sample,
3357 const struct channel_state_sample *latest_sample,
3358 uint64_t previous_session_consumed_total,
3359 uint64_t latest_session_consumed_total,
3360 struct channel_info *channel_info)
3361 {
3362 int ret = 0;
3363 enum lttng_condition_type condition_type;
3364 const bool previous_sample_available = !!previous_sample;
3365 bool previous_sample_result = false;
3366 bool latest_sample_result;
3367
3368 condition_type = lttng_condition_get_type(condition);
3369
3370 switch (condition_type) {
3371 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
3372 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
3373 if (caa_likely(previous_sample_available)) {
3374 previous_sample_result =
3375 evaluate_buffer_usage_condition(condition,
3376 previous_sample, channel_info->capacity);
3377 }
3378 latest_sample_result = evaluate_buffer_usage_condition(
3379 condition, latest_sample,
3380 channel_info->capacity);
3381 break;
3382 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
3383 if (caa_likely(previous_sample_available)) {
3384 previous_sample_result =
3385 evaluate_session_consumed_size_condition(
3386 condition,
3387 previous_session_consumed_total);
3388 }
3389 latest_sample_result =
3390 evaluate_session_consumed_size_condition(
3391 condition,
3392 latest_session_consumed_total);
3393 break;
3394 default:
3395 /* Unknown condition type; internal error. */
3396 abort();
3397 }
3398
3399 if (!latest_sample_result ||
3400 (previous_sample_result == latest_sample_result)) {
3401 /*
3402 * Only trigger on a condition evaluation transition.
3403 *
3404 * NOTE: This edge-triggered logic may not be appropriate for
3405 * future condition types.
3406 */
3407 goto end;
3408 }
3409
3410 if (!evaluation || !latest_sample_result) {
3411 goto end;
3412 }
3413
3414 switch (condition_type) {
3415 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
3416 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
3417 *evaluation = lttng_evaluation_buffer_usage_create(
3418 condition_type,
3419 latest_sample->highest_usage,
3420 channel_info->capacity);
3421 break;
3422 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
3423 *evaluation = lttng_evaluation_session_consumed_size_create(
3424 latest_session_consumed_total);
3425 break;
3426 default:
3427 abort();
3428 }
3429
3430 if (!*evaluation) {
3431 ret = -1;
3432 goto end;
3433 }
3434 end:
3435 return ret;
3436 }
3437
3438 static
3439 int client_notification_overflow(struct notification_client *client)
3440 {
3441 int ret = 0;
3442 const struct lttng_notification_channel_message msg = {
3443 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED,
3444 };
3445
3446 ASSERT_LOCKED(client->lock);
3447
3448 DBG("Dropping notification addressed to client (socket fd = %i)",
3449 client->socket);
3450 if (client->communication.outbound.dropped_notification) {
3451 /*
3452 * The client already has a "notification dropped" message
3453 * in its outgoing queue. Nothing to do since all
3454 * of those messages are coalesced.
3455 */
3456 goto end;
3457 }
3458
3459 client->communication.outbound.dropped_notification = true;
3460 ret = lttng_dynamic_buffer_append(
3461 &client->communication.outbound.buffer, &msg,
3462 sizeof(msg));
3463 if (ret) {
3464 PERROR("Failed to enqueue \"dropped notification\" message in client's (socket fd = %i) outgoing queue",
3465 client->socket);
3466 }
3467 end:
3468 return ret;
3469 }
3470
3471 static int client_handle_transmission_status_wrapper(
3472 struct notification_client *client,
3473 enum client_transmission_status status,
3474 void *user_data)
3475 {
3476 return client_handle_transmission_status(client, status,
3477 (struct notification_thread_state *) user_data);
3478 }
3479
3480 static
3481 int send_evaluation_to_clients(const struct lttng_trigger *trigger,
3482 const struct lttng_evaluation *evaluation,
3483 struct notification_client_list* client_list,
3484 struct notification_thread_state *state,
3485 uid_t object_uid, gid_t object_gid)
3486 {
3487 return notification_client_list_send_evaluation(client_list,
3488 lttng_trigger_get_const_condition(trigger), evaluation,
3489 lttng_trigger_get_credentials(trigger),
3490 &(struct lttng_credentials){
3491 .uid = object_uid, .gid = object_gid},
3492 client_handle_transmission_status_wrapper, state);
3493 }
3494
3495 /*
3496 * Permission checks relative to notification channel clients are performed
3497 * here. Notice how object, client, and trigger credentials are involved in
3498 * this check.
3499 *
3500 * The `object` credentials are the credentials associated with the "subject"
3501 * of a condition. For instance, a `rotation completed` condition applies
3502 * to a session. When that condition is met, it will produce an evaluation
3503 * against a session. Hence, in this case, the `object` credentials are the
3504 * credentials of the "subject" session.
3505 *
3506 * The `trigger` credentials are the credentials of the user that registered the
3507 * trigger.
3508 *
3509 * The `client` credentials are the credentials of the user that created a given
3510 * notification channel.
3511 *
3512 * In terms of visibility, it is expected that non-privilieged users can only
3513 * register triggers against "their" objects (their own sessions and
3514 * applications they are allowed to interact with). They can then open a
3515 * notification channel and subscribe to notifications associated with those
3516 * triggers.
3517 *
3518 * As for privilieged users, they can register triggers against the objects of
3519 * other users. They can then subscribe to the notifications associated to their
3520 * triggers. Privilieged users _can't_ subscribe to the notifications of
3521 * triggers owned by other users; they must create their own triggers.
3522 *
3523 * This is more a concern of usability than security. It would be difficult for
3524 * a root user reliably subscribe to a specific set of conditions without
3525 * interference from external users (those could, for instance, unregister
3526 * their triggers).
3527 */
3528 LTTNG_HIDDEN
3529 int notification_client_list_send_evaluation(
3530 struct notification_client_list *client_list,
3531 const struct lttng_condition *condition,
3532 const struct lttng_evaluation *evaluation,
3533 const struct lttng_credentials *trigger_creds,
3534 const struct lttng_credentials *source_object_creds,
3535 report_client_transmission_result_cb client_report,
3536 void *user_data)
3537 {
3538 int ret = 0;
3539 struct lttng_payload msg_payload;
3540 struct notification_client_list_element *client_list_element, *tmp;
3541 const struct lttng_notification notification = {
3542 .condition = (struct lttng_condition *) condition,
3543 .evaluation = (struct lttng_evaluation *) evaluation,
3544 };
3545 struct lttng_notification_channel_message msg_header = {
3546 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION,
3547 };
3548
3549 lttng_payload_init(&msg_payload);
3550
3551 ret = lttng_dynamic_buffer_append(&msg_payload.buffer, &msg_header,
3552 sizeof(msg_header));
3553 if (ret) {
3554 goto end;
3555 }
3556
3557 ret = lttng_notification_serialize(&notification, &msg_payload);
3558 if (ret) {
3559 ERR("[notification-thread] Failed to serialize notification");
3560 ret = -1;
3561 goto end;
3562 }
3563
3564 /* Update payload size. */
3565 ((struct lttng_notification_channel_message *) msg_payload.buffer.data)
3566 ->size = (uint32_t)(
3567 msg_payload.buffer.size - sizeof(msg_header));
3568
3569 pthread_mutex_lock(&client_list->lock);
3570 cds_list_for_each_entry_safe(client_list_element, tmp,
3571 &client_list->list, node) {
3572 enum client_transmission_status transmission_status;
3573 struct notification_client *client =
3574 client_list_element->client;
3575
3576 ret = 0;
3577 pthread_mutex_lock(&client->lock);
3578 if (source_object_creds) {
3579 if (client->uid != source_object_creds->uid &&
3580 client->gid != source_object_creds->gid &&
3581 client->uid != 0) {
3582 /*
3583 * Client is not allowed to monitor this
3584 * object.
3585 */
3586 DBG("[notification-thread] Skipping client at it does not have the object permission to receive notification for this trigger");
3587 goto unlock_client;
3588 }
3589 }
3590
3591 if (client->uid != trigger_creds->uid && client->gid != trigger_creds->gid) {
3592 DBG("[notification-thread] Skipping client at it does not have the permission to receive notification for this trigger");
3593 goto unlock_client;
3594 }
3595
3596 DBG("[notification-thread] Sending notification to client (fd = %i, %zu bytes)",
3597 client->socket, msg_payload.buffer.size);
3598 if (client->communication.outbound.buffer.size) {
3599 /*
3600 * Outgoing data is already buffered for this client;
3601 * drop the notification and enqueue a "dropped
3602 * notification" message if this is the first dropped
3603 * notification since the socket spilled-over to the
3604 * queue.
3605 */
3606 ret = client_notification_overflow(client);
3607 if (ret) {
3608 goto unlock_client;
3609 }
3610 }
3611
3612 ret = lttng_dynamic_buffer_append_buffer(
3613 &client->communication.outbound.buffer,
3614 &msg_payload.buffer);
3615 if (ret) {
3616 goto unlock_client;
3617 }
3618
3619 transmission_status = client_flush_outgoing_queue(client);
3620 ret = client_report(client, transmission_status, user_data);
3621 if (ret) {
3622 goto unlock_client;
3623 }
3624 unlock_client:
3625 pthread_mutex_unlock(&client->lock);
3626 if (ret) {
3627 goto end_unlock_list;
3628 }
3629 }
3630 ret = 0;
3631
3632 end_unlock_list:
3633 pthread_mutex_unlock(&client_list->lock);
3634 end:
3635 lttng_payload_reset(&msg_payload);
3636 return ret;
3637 }
3638
3639 int handle_notification_thread_channel_sample(
3640 struct notification_thread_state *state, int pipe,
3641 enum lttng_domain_type domain)
3642 {
3643 int ret = 0;
3644 struct lttcomm_consumer_channel_monitor_msg sample_msg;
3645 struct channel_info *channel_info;
3646 struct cds_lfht_node *node;
3647 struct cds_lfht_iter iter;
3648 struct lttng_channel_trigger_list *trigger_list;
3649 struct lttng_trigger_list_element *trigger_list_element;
3650 bool previous_sample_available = false;
3651 struct channel_state_sample previous_sample, latest_sample;
3652 uint64_t previous_session_consumed_total, latest_session_consumed_total;
3653 struct lttng_credentials channel_creds;
3654
3655 /*
3656 * The monitoring pipe only holds messages smaller than PIPE_BUF,
3657 * ensuring that read/write of sampling messages are atomic.
3658 */
3659 ret = lttng_read(pipe, &sample_msg, sizeof(sample_msg));
3660 if (ret != sizeof(sample_msg)) {
3661 ERR("[notification-thread] Failed to read from monitoring pipe (fd = %i)",
3662 pipe);
3663 ret = -1;
3664 goto end;
3665 }
3666
3667 ret = 0;
3668 latest_sample.key.key = sample_msg.key;
3669 latest_sample.key.domain = domain;
3670 latest_sample.highest_usage = sample_msg.highest;
3671 latest_sample.lowest_usage = sample_msg.lowest;
3672 latest_sample.channel_total_consumed = sample_msg.total_consumed;
3673
3674 rcu_read_lock();
3675
3676 /* Retrieve the channel's informations */
3677 cds_lfht_lookup(state->channels_ht,
3678 hash_channel_key(&latest_sample.key),
3679 match_channel_info,
3680 &latest_sample.key,
3681 &iter);
3682 node = cds_lfht_iter_get_node(&iter);
3683 if (caa_unlikely(!node)) {
3684 /*
3685 * Not an error since the consumer can push a sample to the pipe
3686 * and the rest of the session daemon could notify us of the
3687 * channel's destruction before we get a chance to process that
3688 * sample.
3689 */
3690 DBG("[notification-thread] Received a sample for an unknown channel from consumerd, key = %" PRIu64 " in %s domain",
3691 latest_sample.key.key,
3692 domain == LTTNG_DOMAIN_KERNEL ? "kernel" :
3693 "user space");
3694 goto end_unlock;
3695 }
3696 channel_info = caa_container_of(node, struct channel_info,
3697 channels_ht_node);
3698 DBG("[notification-thread] Handling channel sample for channel %s (key = %" PRIu64 ") in session %s (highest usage = %" PRIu64 ", lowest usage = %" PRIu64", total consumed = %" PRIu64")",
3699 channel_info->name,
3700 latest_sample.key.key,
3701 channel_info->session_info->name,
3702 latest_sample.highest_usage,
3703 latest_sample.lowest_usage,
3704 latest_sample.channel_total_consumed);
3705
3706 previous_session_consumed_total =
3707 channel_info->session_info->consumed_data_size;
3708
3709 /* Retrieve the channel's last sample, if it exists, and update it. */
3710 cds_lfht_lookup(state->channel_state_ht,
3711 hash_channel_key(&latest_sample.key),
3712 match_channel_state_sample,
3713 &latest_sample.key,
3714 &iter);
3715 node = cds_lfht_iter_get_node(&iter);
3716 if (caa_likely(node)) {
3717 struct channel_state_sample *stored_sample;
3718
3719 /* Update the sample stored. */
3720 stored_sample = caa_container_of(node,
3721 struct channel_state_sample,
3722 channel_state_ht_node);
3723
3724 memcpy(&previous_sample, stored_sample,
3725 sizeof(previous_sample));
3726 stored_sample->highest_usage = latest_sample.highest_usage;
3727 stored_sample->lowest_usage = latest_sample.lowest_usage;
3728 stored_sample->channel_total_consumed = latest_sample.channel_total_consumed;
3729 previous_sample_available = true;
3730
3731 latest_session_consumed_total =
3732 previous_session_consumed_total +
3733 (latest_sample.channel_total_consumed - previous_sample.channel_total_consumed);
3734 } else {
3735 /*
3736 * This is the channel's first sample, allocate space for and
3737 * store the new sample.
3738 */
3739 struct channel_state_sample *stored_sample;
3740
3741 stored_sample = zmalloc(sizeof(*stored_sample));
3742 if (!stored_sample) {
3743 ret = -1;
3744 goto end_unlock;
3745 }
3746
3747 memcpy(stored_sample, &latest_sample, sizeof(*stored_sample));
3748 cds_lfht_node_init(&stored_sample->channel_state_ht_node);
3749 cds_lfht_add(state->channel_state_ht,
3750 hash_channel_key(&stored_sample->key),
3751 &stored_sample->channel_state_ht_node);
3752
3753 latest_session_consumed_total =
3754 previous_session_consumed_total +
3755 latest_sample.channel_total_consumed;
3756 }
3757
3758 channel_info->session_info->consumed_data_size =
3759 latest_session_consumed_total;
3760
3761 /* Find triggers associated with this channel. */
3762 cds_lfht_lookup(state->channel_triggers_ht,
3763 hash_channel_key(&latest_sample.key),
3764 match_channel_trigger_list,
3765 &latest_sample.key,
3766 &iter);
3767 node = cds_lfht_iter_get_node(&iter);
3768 if (caa_likely(!node)) {
3769 goto end_unlock;
3770 }
3771
3772 channel_creds = (typeof(channel_creds)) {
3773 .uid = channel_info->session_info->uid,
3774 .gid = channel_info->session_info->gid,
3775 };
3776
3777 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
3778 channel_triggers_ht_node);
3779 cds_list_for_each_entry(trigger_list_element, &trigger_list->list,
3780 node) {
3781 const struct lttng_condition *condition;
3782 struct lttng_trigger *trigger;
3783 struct notification_client_list *client_list = NULL;
3784 struct lttng_evaluation *evaluation = NULL;
3785 enum action_executor_status executor_status;
3786
3787 ret = 0;
3788 trigger = trigger_list_element->trigger;
3789 condition = lttng_trigger_get_const_condition(trigger);
3790 assert(condition);
3791
3792 /*
3793 * Check if any client is subscribed to the result of this
3794 * evaluation.
3795 */
3796 client_list = get_client_list_from_condition(state, condition);
3797
3798 ret = evaluate_buffer_condition(condition, &evaluation, state,
3799 previous_sample_available ? &previous_sample : NULL,
3800 &latest_sample,
3801 previous_session_consumed_total,
3802 latest_session_consumed_total,
3803 channel_info);
3804 if (caa_unlikely(ret)) {
3805 goto put_list;
3806 }
3807
3808 if (caa_likely(!evaluation)) {
3809 goto put_list;
3810 }
3811
3812 /*
3813 * Ownership of `evaluation` transferred to the action executor
3814 * no matter the result.
3815 */
3816 executor_status = action_executor_enqueue(state->executor,
3817 trigger, evaluation, &channel_creds,
3818 client_list);
3819 evaluation = NULL;
3820 switch (executor_status) {
3821 case ACTION_EXECUTOR_STATUS_OK:
3822 break;
3823 case ACTION_EXECUTOR_STATUS_ERROR:
3824 case ACTION_EXECUTOR_STATUS_INVALID:
3825 /*
3826 * TODO Add trigger identification (name/id) when
3827 * it is added to the API.
3828 */
3829 ERR("Fatal error occurred while enqueuing action associated with buffer-condition trigger");
3830 ret = -1;
3831 goto put_list;
3832 case ACTION_EXECUTOR_STATUS_OVERFLOW:
3833 /*
3834 * TODO Add trigger identification (name/id) when
3835 * it is added to the API.
3836 *
3837 * Not a fatal error.
3838 */
3839 WARN("No space left when enqueuing action associated with buffer-condition trigger");
3840 ret = 0;
3841 goto put_list;
3842 default:
3843 abort();
3844 }
3845
3846 put_list:
3847 notification_client_list_put(client_list);
3848 if (caa_unlikely(ret)) {
3849 break;
3850 }
3851 }
3852 end_unlock:
3853 rcu_read_unlock();
3854 end:
3855 return ret;
3856 }
This page took 0.112075 seconds and 4 git commands to generate.