2 * dnstap/dtstream.h - Frame Streams thread for unbound DNSTAP
4 * Copyright (c) 2020, NLnet Labs. All rights reserved.
6 * This software is open source.
8 * Redistribution and use in source and binary forms, with or without
9 * modification, are permitted provided that the following conditions
12 * Redistributions of source code must retain the above copyright notice,
13 * this list of conditions and the following disclaimer.
15 * Redistributions in binary form must reproduce the above copyright notice,
16 * this list of conditions and the following disclaimer in the documentation
17 * and/or other materials provided with the distribution.
19 * Neither the name of the NLNET LABS nor the names of its contributors may
20 * be used to endorse or promote products derived from this software without
21 * specific prior written permission.
23 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
24 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
25 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
26 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
27 * HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
28 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
29 * TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
30 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
31 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
32 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
33 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
40 * An implementation of the Frame Streams data transport protocol for
41 * the Unbound DNSTAP message logging facility.
47 #include "util/locks.h"
49 struct dt_io_list_item;
55 * A message buffer with dnstap messages queued up. It is per-worker.
56 * It has locks to synchronize. If the buffer is full, a new message
57 * cannot be added and is discarded. A thread reads the messages and sends
61 /** lock of the buffer structure. Hold this lock to add or remove
62 * entries to the buffer. Release it so that other threads can also
63 * put messages to log, or a message can be taken out to send away
64 * by the writer thread.
67 /** the maximum size of the buffer, in bytes */
69 /** current size of the buffer, in bytes. data bytes of messages.
70 * If a new message make it more than maxsize, the buffer is full */
72 /** number of messages in the queue */
74 /** list of messages. The messages are added to the back and taken
75 * out from the front. */
76 struct dt_msg_entry* first, *last;
77 /** reference to the io thread to wakeup */
78 struct dt_io_thread* dtio;
79 /** the wakeup timer for dtio, on worker event base */
80 struct comm_timer* wakeup_timer;
84 * An entry in the dt_msg_queue. contains one DNSTAP message.
88 /** next in the list. */
89 struct dt_msg_entry* next;
90 /** the buffer with the data to send, an encoded DNSTAP message */
92 /** the length to send. */
97 * Containing buffer and counter for reading DNSTAP frames.
99 struct dt_frame_read_buf {
100 /** Buffer containing frame, except length counter(s). */
102 /** Number of bytes written to buffer. */
104 /** Capacity of the buffer. */
107 /** Frame length field. Will contain the 2nd length field for control
110 /** Number of bytes that have been written to the frame_length field. */
111 size_t frame_len_done;
113 /** Set to 1 if this is a control frame, 0 otherwise (ie data frame). */
118 * IO thread that reads from the queues and writes them.
120 struct dt_io_thread {
121 /** the thread number for the dtio thread,
122 * must be first to cast thread arg to int* in checklock code. */
124 /** event base, for event handling */
126 /** list of queues that is registered to get written */
127 struct dt_io_list_item* io_list;
128 /** iterator point in the io_list, to pick from them in a
129 * round-robin fashion, instead of only from the first when busy.
130 * if NULL it means start at the start of the list. */
131 struct dt_io_list_item* io_list_iter;
132 /** thread id, of the io thread */
134 /** if the io processing has started */
136 /** ssl context for the io thread, for tls connections. type SSL_CTX* */
138 /** if SNI will be used for TLS connections. */
141 /** file descriptor that the thread writes to */
143 /** event structure that the thread uses */
145 /** the event is added */
147 /** event added is a write event */
148 int event_added_is_write;
149 /** check for nonblocking connect errors on fd */
150 int check_nb_connect;
151 /** ssl for current connection, type SSL* */
153 /** true if the handshake for SSL is done, 0 if not */
154 int ssl_handshake_done;
155 /** true if briefly the SSL wants a read event, 0 if not.
156 * This happens during negotiation, we then do not want to write,
157 * but wait for a read event. */
159 /** true if SSL_read is waiting for a write event. Set back to 0 after
160 * single write event is handled. */
163 /** the buffer that currently getting written, or NULL if no
164 * (partial) message written now */
166 /** length of the current message */
168 /** number of bytes written for the current message */
170 /** number of bytes of the length that have been written,
171 * for the current message length that precedes the frame */
172 size_t cur_msg_len_done;
174 /** lock on wakeup_timer_enabled */
175 lock_basic_type wakeup_timer_lock;
176 /** if wakeup timer is enabled in some thread */
177 int wakeup_timer_enabled;
178 /** command pipe that stops the pipe if closed. Used to quit
179 * the program. [0] is read, [1] is written to. */
181 /** the event to listen to the commandpipe */
183 /** the io thread wants to exit */
186 /** in stop flush, this is nonNULL and references the stop_ev */
187 void* stop_flush_event;
189 /** the timer event for connection retries */
190 void* reconnect_timer;
191 /** if the reconnect timer is added to the event base */
192 int reconnect_is_added;
193 /** the current reconnection timeout, it is increased with
194 * exponential backoff, in msec */
195 int reconnect_timeout;
197 /** If the log server is connected to over unix domain sockets,
198 * eg. a file is named that is created to log onto. */
199 int upstream_is_unix;
200 /** if the log server is connected to over TCP. The ip address and
203 /** if the log server is connected to over TLS. ip address, port,
204 * and client certificates can be used for authentication. */
207 /** Perform bidirectional Frame Streams handshake before sending
209 int is_bidirectional;
210 /** Set if the READY control frame has been sent. */
211 int ready_frame_sent;
212 /** Set if valid ACCEPT frame is received. */
213 int accept_frame_received;
214 /** (partially) read frame */
215 struct dt_frame_read_buf read_frame;
217 /** the file path for unix socket (or NULL) */
219 /** the ip address and port number (or NULL) */
221 /** is the TLS upstream authenticated by name, if nonNULL,
222 * we use the same cert bundle as used by other TLS streams. */
223 char* tls_server_name;
224 /** are client certificates in use */
225 int use_client_certs;
226 /** client cert files: the .key file */
227 char* client_key_file;
228 /** client cert files: the .pem file */
229 char* client_cert_file;
233 * IO thread list of queues list item
234 * lists a worker queue that should be looked at and sent to the log server.
236 struct dt_io_list_item {
237 /** next in the list of buffers to inspect */
238 struct dt_io_list_item* next;
239 /** buffer of this worker */
240 struct dt_msg_queue* queue;
244 * Create new (empty) worker message queue. Limit set to default on max.
245 * @param base: event base for wakeup timer.
246 * @return NULL on malloc failure or a new queue (not locked).
248 struct dt_msg_queue* dt_msg_queue_create(struct comm_base* base);
251 * Delete a worker message queue. It has to be unlinked from access,
252 * so it can be deleted without lock worries. The queue is emptied (deleted).
253 * @param mq: message queue.
255 void dt_msg_queue_delete(struct dt_msg_queue* mq);
258 * Submit a message to the queue. The queue is locked by the routine,
259 * the message is inserted, and then the queue is unlocked so the
260 * message can be picked up by the writer thread.
261 * @param mq: message queue.
262 * @param buf: buffer with message (dnstap contents).
263 * The buffer must have been malloced by caller. It is linked in
264 * the queue, and is free()d after use. If the routine fails
265 * the buffer is freed as well (and nothing happens, the item
266 * could not be logged).
267 * @param len: length of buffer.
269 void dt_msg_queue_submit(struct dt_msg_queue* mq, void* buf, size_t len);
271 /** timer callback to wakeup dtio thread to process messages */
272 void mq_wakeup_cb(void* arg);
276 * @return new io thread object. not yet started. or NULL malloc failure.
278 struct dt_io_thread* dt_io_thread_create(void);
281 * Delete the IO thread structure.
282 * @param dtio: the io thread that is deleted. It must not be running.
284 void dt_io_thread_delete(struct dt_io_thread* dtio);
287 * Apply config to the dtio thread
288 * @param dtio: io thread, not yet started.
289 * @param cfg: config file struct.
290 * @return false on malloc failure.
292 int dt_io_thread_apply_cfg(struct dt_io_thread* dtio,
293 struct config_file *cfg);
296 * Register a msg queue to the io thread. It will be polled to see if
297 * there are messages and those then get removed and sent, when the thread
299 * @param dtio: the io thread.
300 * @param mq: message queue to register.
301 * @return false on failure (malloc failure).
303 int dt_io_thread_register_queue(struct dt_io_thread* dtio,
304 struct dt_msg_queue* mq);
307 * Unregister queue from io thread.
308 * @param dtio: the io thread.
309 * @param mq: message queue.
311 void dt_io_thread_unregister_queue(struct dt_io_thread* dtio,
312 struct dt_msg_queue* mq);
315 * Start the io thread
316 * @param dtio: the io thread.
317 * @param event_base_nothr: the event base to attach the events to, in case
318 * we are running without threads. With threads, this is ignored
319 * and a thread is started to process the dnstap log messages.
320 * @param numworkers: number of worker threads. The dnstap io thread is
321 * that number +1 as the threadnumber (in logs).
322 * @return false on failure.
324 int dt_io_thread_start(struct dt_io_thread* dtio, void* event_base_nothr,
329 * @param dtio: the io thread.
331 void dt_io_thread_stop(struct dt_io_thread* dtio);
333 /** callback for the dnstap reconnect, to start reconnecting to output */
334 void dtio_reconnect_timeout_cb(int fd, short bits, void* arg);
336 /** callback for the dnstap events, to write to the output */
337 void dtio_output_cb(int fd, short bits, void* arg);
339 /** callback for the dnstap commandpipe, to stop the dnstap IO */
340 void dtio_cmd_cb(int fd, short bits, void* arg);
342 /** callback for the timer when the thread stops and wants to finish up */
343 void dtio_stop_timer_cb(int fd, short bits, void* arg);
345 /** callback for the output when the thread stops and wants to finish up */
346 void dtio_stop_ev_cb(int fd, short bits, void* arg);
348 /** callback for unbound-dnstap-socket */
349 void dtio_tap_callback(int fd, short bits, void* arg);
351 /** callback for unbound-dnstap-socket */
352 void dtio_mainfdcallback(int fd, short bits, void* arg);
354 #endif /* DTSTREAM_H */