src/plugins/ctf/lttng-live/lttng-live.h

   1 #ifndef BABELTRACE_PLUGIN_CTF_LTTNG_LIVE_H
   2 #define BABELTRACE_PLUGIN_CTF_LTTNG_LIVE_H
   3
   4 /*
   5  * BabelTrace - LTTng-live client Component
   6  *
   7  * Copyright 2019 Francis Deslauriers <francis.deslauriers@efficios.com>
   8  * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   9  * Copyright 2016 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
  10  *
  11  * Author: Jérémie Galarneau <jeremie.galarneau@efficios.com>
  12  *
  13  * Permission is hereby granted, free of charge, to any person obtaining a copy
  14  * of this software and associated documentation files (the "Software"), to deal
  15  * in the Software without restriction, including without limitation the rights
  16  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  17  * copies of the Software, and to permit persons to whom the Software is
  18  * furnished to do so, subject to the following conditions:
  19  *
  20  * The above copyright notice and this permission notice shall be included in
  21  * all copies or substantial portions of the Software.
  22  *
  23  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  24  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  25  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  26  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  27  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  28  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  29  * SOFTWARE.
  30  */
  31
  32 #include <stdbool.h>
  33
  34 #include "common/macros.h"
  35 #include <babeltrace2/babeltrace.h>
  36
  37 #include "../common/metadata/decoder.h"
  38 #include "../common/msg-iter/msg-iter.h"
  39
  40 #include "viewer-connection.h"
  41
  42 struct lttng_live_component;
  43 struct lttng_live_session;
  44 struct lttng_live_msg_iter;
  45
  46 enum lttng_live_stream_state {
  47         /* This stream won't have data until some known time in the future. */
  48         LTTNG_LIVE_STREAM_QUIESCENT,
  49         /*
  50          * This stream won't have data until some known time in the future and
  51          * the message iterator inactivity message was already sent downstream.
  52          */
  53         LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA, /* */
  54         /* This stream has data ready to be consumed. */
  55         LTTNG_LIVE_STREAM_ACTIVE_DATA,
  56         /*
  57          * This stream has no data left to consume. We should asked the relay
  58          * for more.
  59          */
  60         LTTNG_LIVE_STREAM_ACTIVE_NO_DATA,
  61         /* This stream won't have anymore data, ever. */
  62         LTTNG_LIVE_STREAM_EOF,
  63 };
  64
  65 /* Iterator over a live stream. */
  66 struct lttng_live_stream_iterator {
  67         bt_logging_level log_level;
  68         bt_self_component *self_comp;
  69
  70         /* Owned by this. */
  71         bt_stream *stream;
  72
  73         /* Weak reference. */
  74         struct lttng_live_trace *trace;
  75
  76         /*
  77          * Since only a single iterator per viewer connection, we have
  78          * only a single message iterator per stream.
  79          */
  80         struct bt_msg_iter *msg_iter;
  81
  82         uint64_t viewer_stream_id;
  83
  84         uint64_t ctf_stream_class_id;
  85
  86         /* base offset in current index. */
  87         uint64_t base_offset;
  88         /* len to read in current index. */
  89         uint64_t len;
  90         /* offset in current index. */
  91         uint64_t offset;
  92
  93         /*
  94          * Clock Snapshot value of the last message iterator inactivity message
  95          * sent downstream.
  96          */
  97         uint64_t last_inactivity_ts;
  98
  99         /*
 100          * Clock Snapshot value of the current message iterator inactivity
 101          * message we might want to send downstream.
 102          */
 103         uint64_t current_inactivity_ts;
 104
 105         enum lttng_live_stream_state state;
 106
 107         /*
 108          * The current message produced by this live stream iterator. Owned by
 109          * this.
 110          */
 111         bt_message *current_msg;
 112
 113         /* Timestamp in nanoseconds of the current message (current_msg). */
 114         int64_t current_msg_ts_ns;
 115
 116         /* Owned by this. */
 117         uint8_t *buf;
 118         size_t buflen;
 119
 120         /* Owned by this. */
 121         GString *name;
 122 };
 123
 124 struct lttng_live_metadata {
 125         bt_logging_level log_level;
 126         bt_self_component *self_comp;
 127
 128         /* Weak reference. */
 129         struct lttng_live_trace *trace;
 130
 131         uint64_t stream_id;
 132         /* Weak reference. */
 133         struct ctf_metadata_decoder *decoder;
 134
 135         bool closed;
 136 };
 137
 138 struct lttng_live_trace {
 139         bt_logging_level log_level;
 140         bt_self_component *self_comp;
 141
 142         /* Back reference to session. */
 143         struct lttng_live_session *session;
 144
 145         /* ctf trace ID within the session. */
 146         uint64_t id;
 147
 148         /* Owned by this. */
 149         bt_trace *trace;
 150
 151         /* Weak reference. */
 152         bt_trace_class *trace_class;
 153
 154         struct lttng_live_metadata *metadata;
 155
 156         const bt_clock_class *clock_class;
 157
 158         /* Array of pointers to struct lttng_live_stream_iterator. */
 159         /* Owned by this. */
 160         GPtrArray *stream_iterators;
 161
 162         bool new_metadata_needed;
 163 };
 164
 165 struct lttng_live_session {
 166         bt_logging_level log_level;
 167         bt_self_component *self_comp;
 168
 169         /* Weak reference. */
 170         struct lttng_live_msg_iter *lttng_live_msg_iter;
 171
 172         /* Owned by this. */
 173         GString *hostname;
 174
 175         /* Owned by this. */
 176         GString *session_name;
 177
 178         uint64_t id;
 179
 180         /* Array of pointers to struct lttng_live_trace. */
 181         GPtrArray *traces;
 182
 183         bool attached;
 184         bool new_streams_needed;
 185         bool lazy_stream_msg_init;
 186         bool closed;
 187 };
 188
 189 enum session_not_found_action {
 190         SESSION_NOT_FOUND_ACTION_CONTINUE,
 191         SESSION_NOT_FOUND_ACTION_FAIL,
 192         SESSION_NOT_FOUND_ACTION_END,
 193 };
 194
 195 /*
 196  * A component instance is an iterator on a single session.
 197  */
 198 struct lttng_live_component {
 199         bt_logging_level log_level;
 200
 201         /* Weak reference. */
 202         bt_self_component *self_comp;
 203
 204         struct {
 205                 GString *url;
 206                 enum session_not_found_action sess_not_found_act;
 207         } params;
 208
 209         size_t max_query_size;
 210
 211         /*
 212          * Keeps track of whether the downstream component already has a
 213          * message iterator on this component.
 214          */
 215         bool has_msg_iter;
 216 };
 217
 218 struct lttng_live_msg_iter {
 219         bt_logging_level log_level;
 220         bt_self_component *self_comp;
 221
 222         /* Weak reference. */
 223         struct lttng_live_component *lttng_live_comp;
 224
 225         /* Weak reference. */
 226         bt_self_message_iterator *self_msg_iter;
 227
 228         /* Owned by this. */
 229         struct live_viewer_connection *viewer_connection;
 230
 231         /* Array of pointers to struct lttng_live_session. */
 232         GPtrArray *sessions;
 233
 234         /* Number of live stream iterator this message iterator has.*/
 235         uint64_t active_stream_iter;
 236
 237         /* Timestamp in nanosecond of the last message sent downstream. */
 238         int64_t last_msg_ts_ns;
 239 };
 240
 241 enum lttng_live_iterator_status {
 242         /** Iterator state has progressed. Continue iteration immediately. */
 243         LTTNG_LIVE_ITERATOR_STATUS_CONTINUE = 3,
 244         /** No message available for now. Try again later. */
 245         LTTNG_LIVE_ITERATOR_STATUS_AGAIN = 2,
 246         /** No more CTF_LTTNG_LIVEs to be delivered. */
 247         LTTNG_LIVE_ITERATOR_STATUS_END = 1,
 248         /** No error, okay. */
 249         LTTNG_LIVE_ITERATOR_STATUS_OK = 0,
 250         /** Invalid arguments. */
 251         LTTNG_LIVE_ITERATOR_STATUS_INVAL = -1,
 252         /** General error. */
 253         LTTNG_LIVE_ITERATOR_STATUS_ERROR = -2,
 254         /** Out of memory. */
 255         LTTNG_LIVE_ITERATOR_STATUS_NOMEM = -3,
 256         /** Unsupported iterator feature. */
 257         LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED = -4,
 258 };
 259
 260 bt_component_class_init_method_status lttng_live_component_init(
 261                 bt_self_component_source *self_comp,
 262                 const bt_value *params, void *init_method_data);
 263
 264 bt_component_class_query_method_status lttng_live_query(
 265                 bt_self_component_class_source *comp_class,
 266                 const bt_query_executor *query_exec,
 267                 const char *object, const bt_value *params,
 268                 bt_logging_level log_level,
 269                 const bt_value **result);
 270
 271 void lttng_live_component_finalize(bt_self_component_source *component);
 272
 273 bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next(
 274                 bt_self_message_iterator *iterator,
 275                 bt_message_array_const msgs, uint64_t capacity,
 276                 uint64_t *count);
 277
 278 bt_component_class_message_iterator_init_method_status lttng_live_msg_iter_init(
 279                 bt_self_message_iterator *self_msg_it,
 280                 bt_self_component_source *self_comp,
 281                 bt_self_component_port_output *self_port);
 282
 283 void lttng_live_msg_iter_finalize(bt_self_message_iterator *it);
 284
 285 int lttng_live_create_viewer_session(struct lttng_live_msg_iter *lttng_live_msg_iter);
 286 int lttng_live_attach_session(struct lttng_live_session *session);
 287 int lttng_live_detach_session(struct lttng_live_session *session);
 288 enum lttng_live_iterator_status lttng_live_get_new_streams(
 289                 struct lttng_live_session *session);
 290
 291 int lttng_live_add_session(struct lttng_live_msg_iter *lttng_live_msg_iter,
 292                 uint64_t session_id,
 293                 const char *hostname,
 294                 const char *session_name);
 295
 296 ssize_t lttng_live_get_one_metadata_packet(struct lttng_live_trace *trace,
 297                 FILE *fp);
 298 enum lttng_live_iterator_status lttng_live_get_next_index(
 299                 struct lttng_live_msg_iter *lttng_live_msg_iter,
 300                 struct lttng_live_stream_iterator *stream,
 301                 struct packet_index *index);
 302
 303 enum bt_msg_iter_medium_status lttng_live_get_stream_bytes(
 304                 struct lttng_live_msg_iter *lttng_live_msg_iter,
 305                 struct lttng_live_stream_iterator *stream, uint8_t *buf,
 306                 uint64_t offset, uint64_t req_len, uint64_t *recv_len);
 307 void lttng_live_add_stream_iterator(struct lttng_live_msg_iter *lttng_live_msg_iter,
 308                 struct lttng_live_stream_iterator *stream_iter);
 309 void lttng_live_remove_stream_iterator(struct lttng_live_msg_iter *lttng_live_msg_iter,
 310                 struct lttng_live_stream_iterator *stream_iter);
 311
 312 struct lttng_live_trace *lttng_live_borrow_trace(
 313                 struct lttng_live_session *session, uint64_t trace_id);
 314 void lttng_live_need_new_streams(struct lttng_live_msg_iter *lttng_live_msg_iter);
 315
 316 bool lttng_live_graph_is_canceled(struct lttng_live_msg_iter *msg_iter);
 317
 318 #endif /* BABELTRACE_PLUGIN_CTF_LTTNG_LIVE_H */