2 * collectd - src/utils_ovs.c
4 * Copyright(c) 2016 Intel Corporation. All rights reserved.
6 * Permission is hereby granted, free of charge, to any person obtaining a copy of
7 * this software and associated documentation files (the "Software"), to deal in
8 * the Software without restriction, including without limitation the rights to
9 * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
10 * of the Software, and to permit persons to whom the Software is furnished to do
11 * so, subject to the following conditions:
13 * The above copyright notice and this permission notice shall be included in all
14 * copies or substantial portions of the Software.
16 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
19 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
21 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
25 * Volodymyr Mytnyk <volodymyrx.mytnyk@intel.com>
27 * OVS DB API internal architecture diagram
28 * +------------------------------------------------------------------------------+
29 * |OVS plugin |OVS utils |
30 * | | +------------------------+ |
31 * | | | echo handler | JSON request/ |
32 * | | +--+ (ovs_db_table_echo_cb) +<---+---------+ update event/ |
33 * | | | | | | | result |
34 * | | | +------------------------+ | | |
35 * | | | | +----+---+--------+ |
36 * | +----------+ | | +------------------------+ | | | | |
37 * | | update | | | | update handler | | | YAJL | JSON | |
38 * | | callback +<-------+(ovs_db_table_update_cp)+<---+ | parser | reader | |
39 * | +----------+ | | | | | | | | |
40 * | | | +------------------------+ | +--------+---+----+ |
42 * | +----------+ | | +------------------------+ | | |
43 * | | result | | | | result handler | | | |
44 * | | callback +<-------+ (ovs_db_result_cb) +<---+ JSON raw | |
45 * | +----------+ | | | | data | |
46 * | | | +------------------------+ | |
48 * | | | +------------------+ +------------+----+ |
49 * | +----------+ | | |thread| | |thread| | |
50 * | | init | | | | | reconnect | | |
51 * | | callback +<---------+ EVENT WORKER +<------------+ POLL WORKER | |
52 * | +----------+ | | +------------------+ +--------+--------+ |
54 * +----------------+-------------------------------------------------------------+
56 * JSON|echo reply raw|data
58 * +-------------------+----------------------------------------------+-----------+
60 * +-------------------------------------------------------------------------------
64 /* collectd headers */
68 #include "utils_ovs.h"
70 /* system libraries */
71 #include <semaphore.h>
72 #include <arpa/inet.h>
76 #define OVS_ERROR(fmt, ...) do { \
77 ERROR("ovs_utils: "fmt, ## __VA_ARGS__); } while (0)
78 #define OVS_DEBUG(fmt, ...) do { \
79 DEBUG("%s:%d:%s(): "fmt, __FILE__, __LINE__, __FUNCTION__, \
80 ## __VA_ARGS__); } while (0)
82 #define OVS_DB_POLL_TIMEOUT 1 /* poll receive timeout (sec) */
83 #define OVS_DB_POLL_READ_BLOCK_SIZE 5 /* read block size (bytes) */
84 #define OVS_DB_DEFAULT_DB_NAME "Open_vSwitch"
85 #define OVS_DB_RECONNECT_TIMEOUT 1 /* reconnect timeout (sec) */
87 #define OVS_DB_EVENT_TIMEOUT 5 /* event thread timeout (sec) */
88 #define OVS_DB_EVENT_TERMINATE 1
89 #define OVS_DB_EVENT_CONN_ESTABLISHED 2
90 #define OVS_DB_EVENT_CONN_TERMINATED 3
92 #define OVS_DB_POLL_STATE_RUNNING 1
93 #define OVS_DB_POLL_STATE_EXITING 2
95 #define OVS_DB_SEND_REQ_TIMEOUT 5 /* send request timeout (sec) */
97 #define OVS_YAJL_CALL(func, ...) \
99 yajl_gen_ret = yajl_gen_status_ok; \
100 if ((yajl_gen_ret = func(__VA_ARGS__)) != yajl_gen_status_ok) \
101 goto yajl_gen_failure; \
103 #define OVS_YAJL_ERROR_BUFFER_SIZE 1024
104 #define OVS_ERROR_BUFF_SIZE 512
105 #define OVS_UID_STR_SIZE 17 /* 64-bit HEX string len + '\0' */
107 /* JSON reader internal data */
108 struct ovs_json_reader_s {
114 typedef struct ovs_json_reader_s ovs_json_reader_t;
116 /* Result callback declaration */
117 struct ovs_result_cb_s {
119 ovs_db_result_cb_t call;
121 typedef struct ovs_result_cb_s ovs_result_cb_t;
123 /* Table callback declaration */
124 struct ovs_table_cb_s {
125 ovs_db_table_cb_t call;
127 typedef struct ovs_table_cb_s ovs_table_cb_t;
129 /* Callback declaration */
130 struct ovs_callback_s {
133 ovs_result_cb_t result;
134 ovs_table_cb_t table;
136 struct ovs_callback_s *next;
137 struct ovs_callback_s *prev;
139 typedef struct ovs_callback_s ovs_callback_t;
141 /* Connection declaration */
148 struct sockaddr_in s_inet;
149 struct sockaddr_un s_unix;
152 typedef struct ovs_conn_s ovs_conn_t;
154 /* Event thread data declaration */
155 struct ovs_event_thread_s {
157 pthread_mutex_t mutex;
161 typedef struct ovs_event_thread_s ovs_event_thread_t;
163 /* Poll thread data declaration */
164 struct ovs_poll_thread_s {
166 pthread_mutex_t mutex;
169 typedef struct ovs_poll_thread_s ovs_poll_thread_t;
171 /* OVS DB internal data declaration */
173 ovs_poll_thread_t poll_thread;
174 ovs_event_thread_t event_thread;
175 pthread_mutex_t mutex;
176 ovs_callback_t *remote_cb;
177 ovs_db_callback_t cb;
180 typedef struct ovs_db_s ovs_db_t;
182 /* Post an event to event thread.
183 * Possible events are:
184 * OVS_DB_EVENT_TERMINATE
185 * OVS_DB_EVENT_CONN_ESTABLISHED
186 * OVS_DB_EVENT_CONN_TERMINATED
189 ovs_db_event_post(ovs_db_t *pdb, int event)
191 pthread_mutex_lock(&pdb->event_thread.mutex);
192 pdb->event_thread.value = event;
193 pthread_mutex_unlock(&pdb->event_thread.mutex);
194 pthread_cond_signal(&pdb->event_thread.cond);
197 /* Check if POLL thread is still running. Returns
198 * 1 if running otherwise 0 is returned */
200 ovs_db_poll_is_running(ovs_db_t *pdb)
203 pthread_mutex_lock(&pdb->poll_thread.mutex);
204 state = pdb->poll_thread.state;
205 pthread_mutex_unlock(&pdb->poll_thread.mutex);
206 return (state == OVS_DB_POLL_STATE_RUNNING);
209 /* Terminate POLL thread */
211 ovs_db_poll_terminate(ovs_db_t *pdb)
213 pthread_mutex_lock(&pdb->poll_thread.mutex);
214 pdb->poll_thread.state = OVS_DB_POLL_STATE_EXITING;
215 pthread_mutex_unlock(&pdb->poll_thread.mutex);
218 /* Generate unique identifier (UID). It is used by OVS DB API
219 * to set "id" field for any OVS DB JSON request. */
224 clock_gettime(CLOCK_MONOTONIC, &ts);
225 return ((ts.tv_sec << 32) | (ts.tv_nsec & UINT32_MAX));
229 * Callback API. These function are used to store
230 * registered callbacks in OVS DB API.
233 /* Add new callback into OVS DB object */
235 ovs_db_callback_add(ovs_db_t *pdb, ovs_callback_t *new_cb)
237 pthread_mutex_lock(&pdb->mutex);
239 pdb->remote_cb->prev = new_cb;
240 new_cb->next = pdb->remote_cb;
242 pdb->remote_cb = new_cb;
243 pthread_mutex_unlock(&pdb->mutex);
246 /* Remove callback from OVS DB object */
248 ovs_db_callback_remove(ovs_db_t *pdb, ovs_callback_t *del_cb)
250 ovs_callback_t *pre_cb = del_cb->prev;
251 ovs_callback_t *next_cb = del_cb->next;
253 pthread_mutex_lock(&pdb->mutex);
255 next_cb->prev = del_cb->prev;
258 pre_cb->next = del_cb->next;
260 pdb->remote_cb = del_cb->next;
263 pthread_mutex_unlock(&pdb->mutex);
266 /* Remove all callbacks form OVS DB object */
268 ovs_db_callback_remove_all(ovs_db_t *pdb)
270 pthread_mutex_lock(&pdb->mutex);
271 for (ovs_callback_t *del_cb = pdb->remote_cb; pdb->remote_cb;
272 del_cb = pdb->remote_cb) {
273 pdb->remote_cb = pdb->remote_cb->next;
276 pdb->remote_cb = NULL;
277 pthread_mutex_unlock(&pdb->mutex);
280 /* Get/find callback in OVS DB object by UID. Returns pointer
281 * to requested callback otherwise NULL is returned */
282 static ovs_callback_t *
283 ovs_db_callback_get(ovs_db_t *pdb, uint64_t uid)
285 pthread_mutex_lock(&pdb->mutex);
286 for (ovs_callback_t *cb = pdb->remote_cb; cb != NULL; cb = cb->next)
287 if (cb->uid == uid) {
288 pthread_mutex_unlock(&pdb->mutex);
291 pthread_mutex_unlock(&pdb->mutex);
295 /* Send all requested data to the socket. Returns 0 if
296 * ALL request data has been sent otherwise negative value
299 ovs_db_data_send(const ovs_db_t *pdb, const char *data, size_t len)
306 if ((nbytes = send(pdb->conn.sock, data + off, rem, 0)) <= 0)
308 rem -= (size_t)nbytes;
309 off += (size_t)nbytes;
314 /* Parse OVS server URL.
316 * "tcp:a.b.c.d:port" - define TCP connection (INET domain)
317 * "unix:file" - define UNIX socket file (UNIX domain)
320 ovs_db_url_parse(const char *surl, ovs_conn_t *conn)
323 char *nexttok = NULL;
329 if ((surl == NULL) || (strlen(surl) < 1))
334 in_str = sstrdup(surl);
335 if ((nexttok = strtok_r(in_str, ":", &saveptr)) != NULL) {
336 if (strcmp("tcp", nexttok) == 0) {
337 tmp_conn.domain = AF_INET;
338 tmp_conn.type = SOCK_STREAM;
339 tmp_conn.addr_size = sizeof(tmp_conn.addr.s_inet);
340 } else if (strcmp("unix", nexttok) == 0) {
341 tmp_conn.domain = AF_UNIX;
342 tmp_conn.type = SOCK_STREAM;
343 tmp_conn.addr_size = sizeof(tmp_conn.addr.s_unix);
349 /* parse url depending on domain */
350 if ((nexttok = strtok_r(NULL, ":", &saveptr)) != NULL) {
351 if (tmp_conn.domain == AF_UNIX) {
353 tmp_conn.addr.s_inet.sin_family = AF_UNIX;
354 sstrncpy(tmp_conn.addr.s_unix.sun_path, nexttok, strlen(nexttok) + 1);
357 tmp_conn.addr.s_inet.sin_family = AF_INET;
359 inet_pton(AF_INET, nexttok, (void *)&tmp_conn.addr.s_inet.sin_addr);
361 if ((nexttok = strtok_r(NULL, ":", &saveptr)) != NULL)
362 tmp_conn.addr.s_inet.sin_port = htons(atoi(nexttok));
370 /* save result and return success */
376 OVS_ERROR("%s() : invalid OVS DB URL provided");
382 * YAJL (Yet Another JSON Library) helper functions
383 * Documentation (https://lloyd.github.io/yajl/)
386 /* Add null-terminated string into YAJL generator handle (JSON object).
387 * Similar function to yajl_gen_string() but takes null-terminated string
388 * instead of string and its length.
390 * jgen - YAJL generator handle allocated by yajl_gen_alloc()
391 * string - Null-terminated string
393 static inline yajl_gen_status
394 ovs_yajl_gen_tstring(yajl_gen hander, const char *string)
396 return yajl_gen_string(hander, string, strlen(string));
399 /* Add YAJL value into YAJL generator handle (JSON object)
401 * jgen - YAJL generator handle allocated by yajl_gen_alloc()
402 * jval - YAJL value usually returned by yajl_tree_get()
404 static yajl_gen_status
405 ovs_yajl_gen_val(yajl_gen jgen, yajl_val jval)
407 size_t array_len = 0;
408 yajl_val *jvalues = NULL;
409 yajl_val jobj_value = NULL;
410 const char *obj_key = NULL;
412 yajl_gen_status yajl_gen_ret;
414 if (YAJL_IS_STRING(jval))
415 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, YAJL_GET_STRING(jval));
416 else if (YAJL_IS_DOUBLE(jval))
417 OVS_YAJL_CALL(yajl_gen_double, jgen, YAJL_GET_DOUBLE(jval));
418 else if (YAJL_IS_INTEGER(jval))
419 OVS_YAJL_CALL(yajl_gen_double, jgen, YAJL_GET_INTEGER(jval));
420 else if (YAJL_IS_TRUE(jval))
421 OVS_YAJL_CALL(yajl_gen_bool, jgen, 1);
422 else if (YAJL_IS_FALSE(jval))
423 OVS_YAJL_CALL(yajl_gen_bool, jgen, 0);
424 else if (YAJL_IS_NULL(jval))
425 OVS_YAJL_CALL(yajl_gen_null, jgen);
426 else if (YAJL_IS_ARRAY(jval)) {
427 /* create new array and add all elements into the array */
428 array_len = YAJL_GET_ARRAY(jval)->len;
429 jvalues = YAJL_GET_ARRAY(jval)->values;
430 OVS_YAJL_CALL(yajl_gen_array_open, jgen);
431 for (int i = 0; i < array_len; i++)
432 OVS_YAJL_CALL(ovs_yajl_gen_val, jgen, jvalues[i]);
433 OVS_YAJL_CALL(yajl_gen_array_close, jgen);
434 } else if (YAJL_IS_OBJECT(jval)) {
435 /* create new object and add all elements into the object */
436 OVS_YAJL_CALL(yajl_gen_map_open, jgen);
437 obj_len = YAJL_GET_OBJECT(jval)->len;
438 for (int i = 0; i < obj_len; i++) {
439 obj_key = YAJL_GET_OBJECT(jval)->keys[i];
440 jobj_value = YAJL_GET_OBJECT(jval)->values[i];
441 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, obj_key);
442 OVS_YAJL_CALL(ovs_yajl_gen_val, jgen, jobj_value);
444 OVS_YAJL_CALL(yajl_gen_map_close, jgen);
446 OVS_ERROR("%s() unsupported value type %d (skip)", __FUNCTION__,
448 goto yajl_gen_failure;
450 return yajl_gen_status_ok;
453 OVS_ERROR("%s() error to generate value", __FUNCTION__);
457 /* OVS DB echo request handler. When OVS DB sends
458 * "echo" request to the client, client should generate
459 * "echo" replay with the same content received in the
462 ovs_db_table_echo_cb(const ovs_db_t *pdb, yajl_val jnode)
468 const char *resp = NULL;
469 const char *params_path[] = {"params", NULL};
470 const char *id_path[] = {"id", NULL};
471 yajl_gen_status yajl_gen_ret;
473 if ((jgen = yajl_gen_alloc(NULL)) == NULL)
476 /* check & get request attributes */
477 if ((jparams = yajl_tree_get(jnode, params_path, yajl_t_array)) == NULL ||
478 ((jid = yajl_tree_get(jnode, id_path, yajl_t_any)) == NULL)) {
479 OVS_ERROR("parse echo request failed");
480 goto yajl_gen_failure;
483 /* generate JSON echo response */
484 OVS_YAJL_CALL(yajl_gen_map_open, jgen);
486 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "result");
487 OVS_YAJL_CALL(ovs_yajl_gen_val, jgen, jparams);
489 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "error");
490 OVS_YAJL_CALL(yajl_gen_null, jgen);
492 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "id");
493 OVS_YAJL_CALL(ovs_yajl_gen_val, jgen, jid);
495 OVS_YAJL_CALL(yajl_gen_map_close, jgen);
496 OVS_YAJL_CALL(yajl_gen_get_buf, jgen, (const unsigned char **)&resp,
499 /* send the response */
500 OVS_DEBUG("response: %s", resp);
501 if (ovs_db_data_send(pdb, resp, resp_len) < 0) {
502 OVS_ERROR("send echo reply failed");
503 goto yajl_gen_failure;
505 /* clean up and return success */
506 yajl_gen_clear(jgen);
511 yajl_gen_clear(jgen);
515 /* Get OVS DB registered callback by YAJL val. The YAJL
516 * value should be YAJL string (UID). Returns NULL if
517 * callback hasn't been found.
519 static ovs_callback_t *
520 ovs_db_table_callback_get(ovs_db_t *pdb, yajl_val jid)
523 const char *suid = NULL;
526 if (jid && YAJL_IS_STRING(jid)) {
527 suid = YAJL_GET_STRING(jid);
528 uid = (uint64_t) strtoul(suid, &endptr, 16);
529 if (*endptr == '\0' && uid)
530 return ovs_db_callback_get(pdb, uid);
536 /* OVS DB table update event handler.
537 * This callback is called by POLL thread if OVS DB
538 * table update callback is received from the DB
539 * server. Once registered callback found, it's called
540 * by this handler. */
542 ovs_db_table_update_cb(ovs_db_t *pdb, yajl_val jnode)
544 ovs_callback_t *cb = NULL;
547 yajl_val jtable_updates;
548 yajl_val jtable_update;
550 const char *table_name = NULL;
551 const char *params_path[] = {"params", NULL};
552 const char *id_path[] = {"id", NULL};
554 /* check & get request attributes */
555 if ((jparams = yajl_tree_get(jnode, params_path, yajl_t_array)) == NULL ||
556 (yajl_tree_get(jnode, id_path, yajl_t_null) == NULL))
559 /* check array length: [<json-value>, <table-updates>] */
560 if (YAJL_GET_ARRAY(jparams)->len != 2)
563 jvalue = YAJL_GET_ARRAY(jparams)->values[0];
564 jtable_updates = YAJL_GET_ARRAY(jparams)->values[1];
565 if ((!YAJL_IS_OBJECT(jtable_updates)) || (!YAJL_IS_STRING(jvalue)))
568 /* find registered callback based on <json-value> */
569 cb = ovs_db_table_callback_get(pdb, jvalue);
570 if (cb == NULL || cb->table.call == NULL)
573 /* call registered callback */
574 cb->table.call(jtable_updates);
578 OVS_ERROR("invalid OVS DB table update event");
582 /* OVS DB result request handler.
583 * This callback is called by POLL thread if OVS DB
584 * result reply is received from the DB server.
585 * Once registered callback found, it's called
586 * by this handler. */
588 ovs_db_result_cb(ovs_db_t *pdb, yajl_val jnode)
590 ovs_callback_t *cb = NULL;
594 const char *result_path[] = {"result", NULL};
595 const char *error_path[] = {"error", NULL};
596 const char *id_path[] = {"id", NULL};
598 jresult = yajl_tree_get(jnode, result_path, yajl_t_any);
599 jerror = yajl_tree_get(jnode, error_path, yajl_t_any);
600 jid = yajl_tree_get(jnode, id_path, yajl_t_string);
602 /* check & get result attributes */
603 if (!jresult || !jerror || !jid)
606 /* try to find registered callback */
607 cb = ovs_db_table_callback_get(pdb, jid);
608 if (cb != NULL && cb->result.call != NULL) {
609 /* call registered callback */
610 cb->result.call(jresult, jerror);
611 /* unlock owner of the reply */
612 sem_post(&cb->result.sync);
618 /* Handle JSON data (one request) and call
619 * appropriate event OVS DB handler. Currently,
620 * update callback 'ovs_db_table_update_cb' and
621 * result callback 'ovs_db_result_cb' is supported.
624 ovs_db_json_data_process(ovs_db_t *pdb, const char *data, size_t len)
626 const char *method = NULL;
627 char yajl_errbuf[OVS_YAJL_ERROR_BUFFER_SIZE];
628 const char *method_path[] = {"method", NULL};
629 const char *result_path[] = {"result", NULL};
631 yajl_val jnode, jval;
633 /* duplicate the data to make null-terminated string
634 * required for yajl_tree_parse() */
635 if ((sjson = strndup(data, len)) == NULL)
638 OVS_DEBUG("%s", sjson);
640 /* parse json data */
641 jnode = yajl_tree_parse(sjson, yajl_errbuf, sizeof(yajl_errbuf));
643 OVS_ERROR("yajl_tree_parse() %s", yajl_errbuf);
647 /* get method name */
648 if (jval = yajl_tree_get(jnode, method_path, yajl_t_string)) {
649 method = YAJL_GET_STRING(jval);
650 if (strcmp("echo", method) == 0) {
651 /* echo request from the server */
652 if (ovs_db_table_echo_cb(pdb, jnode) < 0)
653 OVS_ERROR("handle echo request failed");
654 } else if (strcmp("update", method) == 0) {
655 /* update notification */
656 if (ovs_db_table_update_cb(pdb, jnode) < 0)
657 OVS_ERROR("handle update notification failed");
659 } else if (jval = yajl_tree_get(jnode, result_path, yajl_t_object)) {
660 /* result notification */
661 if (ovs_db_result_cb(pdb, jnode) < 0)
662 OVS_ERROR("handle result reply failed");
666 yajl_tree_free(jnode);
672 * JSON reader implementation.
674 * This module process raw JSON data (byte stream) and
675 * returns fully-fledged JSON data which can be processed
676 * (parsed) by YAJL later.
679 /* Allocate JSON reader instance */
680 static inline ovs_json_reader_t *
681 ovs_json_reader_alloc()
683 ovs_json_reader_t *jreader = NULL;
685 if ((jreader = calloc(sizeof(ovs_json_reader_t), 1)) == NULL)
691 /* Push raw data into into the JSON reader for processing */
693 ovs_json_reader_push_data(ovs_json_reader_t *jreader,
694 const char *data, size_t data_len)
696 char *new_buff = NULL;
697 size_t available = jreader->buff_size - jreader->buff_offset;
699 /* check/update required memory space */
700 if (available < data_len) {
701 OVS_DEBUG("Reallocate buffer [size=%d, available=%d required=%d]",
702 (int)jreader->buff_size, (int)available, (int)data_len);
704 /* allocate new chunk of memory */
705 new_buff = realloc(jreader->buff_ptr, (jreader->buff_size + data_len));
706 if (new_buff == NULL)
709 /* point to new allocated memory */
710 jreader->buff_ptr = new_buff;
711 jreader->buff_size += data_len;
714 /* store input data */
715 memcpy(jreader->buff_ptr + jreader->buff_offset, data, data_len);
716 jreader->buff_offset += data_len;
720 /* Pop one fully-fledged JSON if already exists. Returns 0 if
721 * completed JSON already exists otherwise negative value is
724 ovs_json_reader_pop(ovs_json_reader_t *jreader,
725 const char **json_ptr, size_t *json_len_ptr)
731 /* search open/close brace */
732 for (int i = jreader->json_offset; i < jreader->buff_offset; i++) {
733 if (jreader->buff_ptr[i] == '{') {
735 } else if (jreader->buff_ptr[i] == '}')
739 *json_ptr = jreader->buff_ptr + jreader->json_offset;
740 *json_len_ptr = json_len + 1;
741 jreader->json_offset = i + 1;
745 /* increase JSON data length */
750 if (jreader->json_offset) {
751 if (jreader->json_offset < jreader->buff_offset) {
752 /* shift data to the beginning of the buffer
753 * and zero rest of the buffer data */
754 json = &jreader->buff_ptr[jreader->json_offset];
755 json_len = jreader->buff_offset - jreader->json_offset;
756 for (int i = 0; i < jreader->buff_size; i++)
757 jreader->buff_ptr[i] = ((i < json_len) ? (json[i]) : (0));
758 jreader->buff_offset = json_len;
760 /* reset the buffer */
761 jreader->buff_offset = 0;
763 /* data is at the beginning of the buffer */
764 jreader->json_offset = 0;
770 /* Reset JSON reader. It is useful when start processing
771 * new raw data. E.g.: in case of lost stream connection.
774 ovs_json_reader_reset(ovs_json_reader_t *jreader)
777 jreader->buff_offset = 0;
778 jreader->json_offset = 0;
782 /* Release internal data allocated for JSON reader */
784 ovs_json_reader_free(ovs_json_reader_t *jreader)
787 free(jreader->buff_ptr);
792 /* Reconnect to OVD DB and call init OVS DB callback
793 * 'init_cb' if connection has been established.
796 ovs_db_reconnect(ovs_db_t *pdb)
798 char errbuff[OVS_ERROR_BUFF_SIZE];
800 /* remove all registered OVS DB table/result callbacks */
801 ovs_db_callback_remove_all(pdb);
803 /* open new socket */
804 if ((pdb->conn.sock = socket(pdb->conn.domain, pdb->conn.type, 0)) < 0) {
805 sstrerror(errno, errbuff, sizeof(errbuff));
806 OVS_ERROR("socket(): %s", errbuff);
810 /* try to connect to server */
811 if (connect(pdb->conn.sock, (struct sockaddr *)&pdb->conn.addr,
812 pdb->conn.addr_size) < 0) {
813 sstrerror(errno, errbuff, sizeof(errbuff));
814 OVS_ERROR("connect(): %s", errbuff);
815 close(pdb->conn.sock);
819 /* send notification to event thread */
820 ovs_db_event_post(pdb, OVS_DB_EVENT_CONN_ESTABLISHED);
824 /* POLL worker thread.
825 * It listens on OVS DB connection for incoming
826 * requests/reply/events etc. Also, it reconnects to OVS DB
827 * if connection has been lost.
830 ovs_poll_worker(void *arg)
832 ovs_db_t *pdb = (ovs_db_t *)arg; /* pointer to OVS DB */
833 ovs_json_reader_t *jreader = NULL;
837 char buff[OVS_DB_POLL_READ_BLOCK_SIZE];
838 struct pollfd poll_fd;
841 if ((jreader = ovs_json_reader_alloc()) == NULL) {
842 OVS_ERROR("initialize json reader failed");
846 /* start polling data */
847 poll_fd.fd = pdb->conn.sock;
848 poll_fd.events = POLLIN | POLLPRI;
852 while (ovs_db_poll_is_running(pdb)) {
853 poll_ret = poll(&poll_fd, 1, /* ms */ OVS_DB_POLL_TIMEOUT * 1000);
855 if (poll_fd.revents & POLLNVAL) {
856 /* invalid file descriptor, reconnect */
857 if (ovs_db_reconnect(pdb) != 0) {
858 /* sleep awhile until next reconnect */
859 usleep(OVS_DB_RECONNECT_TIMEOUT * 1000000);
861 ovs_json_reader_reset(jreader);
862 poll_fd.fd = pdb->conn.sock;
863 } else if ((poll_fd.revents & POLLERR) || (poll_fd.revents & POLLHUP)) {
864 /* connection is broken */
865 OVS_ERROR("poll() peer closed its end of the channel");
866 ovs_db_event_post(pdb, OVS_DB_EVENT_CONN_TERMINATED);
868 } else if ((poll_fd.revents & POLLIN) || (poll_fd.revents & POLLPRI)) {
869 /* read incoming data */
870 nbytes = recv(poll_fd.fd, buff, OVS_DB_POLL_READ_BLOCK_SIZE, 0);
872 OVS_DEBUG("recv(): received %d bytes of data", (int)nbytes);
873 ovs_json_reader_push_data(jreader, buff, nbytes);
874 while (!ovs_json_reader_pop(jreader, &json, &json_len))
875 /* process JSON data */
876 ovs_db_json_data_process(pdb, json, json_len);
877 } else if (nbytes == 0) {
878 OVS_ERROR("recv() peer has performed an orderly shutdown");
879 ovs_db_event_post(pdb, OVS_DB_EVENT_CONN_TERMINATED);
882 OVS_ERROR("recv() receive data error");
885 } /* poll() POLLIN & POLLPRI */
886 } else if (poll_ret == 0)
887 OVS_DEBUG("poll() timeout");
889 OVS_ERROR("poll() error");
895 OVS_DEBUG("poll thread has been completed");
896 ovs_json_reader_free(jreader);
897 pthread_exit((void *)0);
901 /* EVENT worker thread.
902 * Perform task based on incoming events. This
903 * task can be done asynchronously which allows to
904 * handle OVD DB callback like 'init_cb'.
907 ovs_event_worker(void *arg)
910 ovs_db_t *pdb = (ovs_db_t *)arg;
913 while (pdb->event_thread.value != OVS_DB_EVENT_TERMINATE) {
914 /* wait for an event */
915 clock_gettime(CLOCK_REALTIME, &ts);
916 ts.tv_sec += (OVS_DB_EVENT_TIMEOUT);
917 ret = pthread_cond_timedwait(&pdb->event_thread.cond,
918 &pdb->event_thread.mutex, &ts);
920 /* handle the event */
921 OVS_DEBUG("handle event %d", pdb->event_thread.value);
922 switch (pdb->event_thread.value) {
923 case OVS_DB_EVENT_CONN_ESTABLISHED:
924 if (pdb->cb.post_conn_init)
925 pdb->cb.post_conn_init(pdb);
927 case OVS_DB_EVENT_CONN_TERMINATED:
928 if (pdb->cb.post_conn_terminate)
929 pdb->cb.post_conn_terminate();
932 OVS_DEBUG("unknown event received");
935 } else if (ret == ETIMEDOUT) {
937 OVS_DEBUG("no event received (timeout)");
940 /* unexpected error */
941 OVS_ERROR("pthread_cond_timedwait() failed");
947 OVS_DEBUG("event thread has been completed");
948 pthread_exit((void *)0);
952 /* Stop EVENT thread */
954 ovs_db_event_thread_stop(ovs_db_t *pdb)
956 ovs_db_event_post(pdb, OVS_DB_EVENT_TERMINATE);
957 if (pthread_join(pdb->event_thread.tid, NULL) != 0)
959 pthread_mutex_unlock(&pdb->event_thread.mutex);
960 pthread_mutex_destroy(&pdb->event_thread.mutex);
964 /* Stop POLL thread */
966 ovs_db_poll_thread_stop(ovs_db_t *pdb)
968 ovs_db_poll_terminate(pdb);
969 if (pthread_join(pdb->poll_thread.tid, NULL) != 0)
971 pthread_mutex_destroy(&pdb->poll_thread.mutex);
976 * Public OVS DB API implementation
980 ovs_db_init(const char *surl, ovs_db_callback_t *cb)
982 pthread_mutexattr_t mutex_attr;
983 ovs_db_t *pdb = NULL;
985 /* allocate db data & fill it */
986 if ((pdb = calloc(1, sizeof(*pdb))) == NULL)
989 /* convert string url to socket addr */
990 if (ovs_db_url_parse(surl, &pdb->conn) < 0)
993 /* setup OVS DB callbacks */
997 /* prepare event thread */
998 pthread_cond_init(&pdb->event_thread.cond, NULL);
999 pthread_mutex_init(&pdb->event_thread.mutex, NULL);
1000 pthread_mutex_lock(&pdb->event_thread.mutex);
1001 if (plugin_thread_create(&pdb->event_thread.tid, NULL,
1002 ovs_event_worker, pdb) != 0) {
1003 OVS_ERROR("event worker start failed");
1007 /* prepare polling thread */
1008 ovs_db_reconnect(pdb);
1009 pdb->poll_thread.state = OVS_DB_POLL_STATE_RUNNING;
1010 pthread_mutex_init(&pdb->poll_thread.mutex, NULL);
1011 if (plugin_thread_create(&pdb->poll_thread.tid, NULL,
1012 ovs_poll_worker, pdb) != 0) {
1013 OVS_ERROR("pull worker start failed");
1017 /* init OVS DB mutex */
1018 if (pthread_mutexattr_init(&mutex_attr) ||
1019 pthread_mutexattr_settype(&mutex_attr, PTHREAD_MUTEX_RECURSIVE) ||
1020 pthread_mutex_init(&pdb->mutex, &mutex_attr)) {
1021 OVS_ERROR("OVS DB mutex init failed");
1025 /* return db to the caller */
1030 /* close connection */
1031 close(pdb->conn.sock);
1032 if (pdb->event_thread.tid != 0)
1033 /* stop event thread */
1034 if (ovs_db_event_thread_stop(pdb) < 0)
1035 OVS_ERROR("stop event thread failed");
1036 if (pdb->poll_thread.tid != 0)
1037 /* stop poll thread */
1038 if (ovs_db_poll_thread_stop(pdb) < 0)
1039 OVS_ERROR("stop poll thread failed");
1045 ovs_db_send_request(ovs_db_t *pdb, const char *method,
1046 const char *params, ovs_db_result_cb_t cb)
1049 yajl_gen_status yajl_gen_ret;
1052 ovs_callback_t *new_cb = NULL;
1054 char uid_buff[OVS_UID_STR_SIZE];
1055 const char *req = NULL;
1060 if (!pdb || !method || !params)
1063 if ((jgen = yajl_gen_alloc(NULL)) == NULL)
1066 /* try to parse params */
1067 if ((jparams = yajl_tree_parse(params, NULL, 0)) == NULL) {
1068 OVS_ERROR("params is not a JSON string");
1069 yajl_gen_clear(jgen);
1073 /* generate method field */
1074 OVS_YAJL_CALL(yajl_gen_map_open, jgen);
1076 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "method");
1077 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, method);
1079 /* generate params field */
1080 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "params");
1081 OVS_YAJL_CALL(ovs_yajl_gen_val, jgen, jparams);
1082 yajl_tree_free(jparams);
1084 /* generate id field */
1085 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "id");
1086 uid = ovs_uid_generate();
1087 ssnprintf(uid_buff, sizeof(uid_buff), "%" PRIX64, uid);
1088 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, uid_buff);
1090 OVS_YAJL_CALL(yajl_gen_map_close, jgen);
1093 /* register result callback */
1094 if ((new_cb = malloc(sizeof(ovs_callback_t))) == NULL)
1095 goto yajl_gen_failure;
1097 /* add new callback to front */
1098 sem_init(&new_cb->result.sync, 0, 0);
1099 new_cb->result.call = cb;
1101 ovs_db_callback_add(pdb, new_cb);
1104 /* send the request */
1105 OVS_YAJL_CALL(yajl_gen_get_buf, jgen, (const unsigned char **)&req,
1107 OVS_DEBUG("%s", req);
1108 if (!ovs_db_data_send(pdb, req, req_len)) {
1110 /* wait for result */
1111 clock_gettime(CLOCK_REALTIME, &ts);
1112 ts.tv_sec += OVS_DB_SEND_REQ_TIMEOUT;
1113 if (sem_timedwait(&new_cb->result.sync, &ts) < 0) {
1114 OVS_ERROR("%s() no replay received within %d sec", __FUNCTION__,
1115 OVS_DB_SEND_REQ_TIMEOUT);
1120 OVS_ERROR("ovs_db_data_send() failed");
1126 /* destroy callback */
1127 sem_destroy(&new_cb->result.sync);
1128 ovs_db_callback_remove(pdb, new_cb);
1131 /* release memory */
1132 yajl_gen_clear(jgen);
1133 return (yajl_gen_ret != yajl_gen_status_ok) ? (-1) : ret;
1137 ovs_db_table_cb_register(ovs_db_t *pdb, const char *tb_name,
1138 const char **tb_column, ovs_db_table_cb_t update_cb,
1139 ovs_db_result_cb_t result_cb, unsigned int flags)
1142 yajl_gen_status yajl_gen_ret;
1143 ovs_callback_t *new_cb = NULL;
1144 char uid_str[OVS_UID_STR_SIZE];
1150 if (pdb == NULL || tb_name == NULL || update_cb == NULL)
1153 if ((jgen = yajl_gen_alloc(NULL)) == NULL)
1156 /* register table update callback */
1157 if ((new_cb = malloc(sizeof(ovs_callback_t))) == NULL)
1160 /* add new callback to front */
1161 new_cb->table.call = update_cb;
1162 new_cb->uid = ovs_uid_generate();
1163 ovs_db_callback_add(pdb, new_cb);
1165 /* make update notification request
1166 * [<db-name>, <json-value>, <monitor-requests>] */
1167 OVS_YAJL_CALL(yajl_gen_array_open, jgen);
1169 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, OVS_DB_DEFAULT_DB_NAME);
1171 /* uid string <json-value> */
1172 ssnprintf(uid_str, sizeof(uid_str), "%" PRIX64, new_cb->uid);
1173 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, uid_str);
1175 /* <monitor-requests> */
1176 OVS_YAJL_CALL(yajl_gen_map_open, jgen);
1178 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, tb_name);
1179 OVS_YAJL_CALL(yajl_gen_array_open, jgen);
1181 /* <monitor-request> */
1182 OVS_YAJL_CALL(yajl_gen_map_open, jgen);
1185 /* columns within the table to be monitored */
1186 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "columns");
1187 OVS_YAJL_CALL(yajl_gen_array_open, jgen);
1188 for (; *tb_column; tb_column++)
1189 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, *tb_column);
1190 OVS_YAJL_CALL(yajl_gen_array_close, jgen);
1192 /* specify select option */
1193 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "select");
1195 OVS_YAJL_CALL(yajl_gen_map_open, jgen);
1197 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "initial");
1198 OVS_YAJL_CALL(yajl_gen_bool, jgen,
1199 flags & OVS_DB_TABLE_CB_FLAG_INITIAL);
1200 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "insert");
1201 OVS_YAJL_CALL(yajl_gen_bool, jgen,
1202 flags & OVS_DB_TABLE_CB_FLAG_INSERT);
1203 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "delete");
1204 OVS_YAJL_CALL(yajl_gen_bool, jgen,
1205 flags & OVS_DB_TABLE_CB_FLAG_DELETE);
1206 OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "modify");
1207 OVS_YAJL_CALL(yajl_gen_bool, jgen,
1208 flags & OVS_DB_TABLE_CB_FLAG_MODIFY);
1210 OVS_YAJL_CALL(yajl_gen_map_close, jgen);
1213 OVS_YAJL_CALL(yajl_gen_map_close, jgen);
1215 OVS_YAJL_CALL(yajl_gen_array_close, jgen);
1217 OVS_YAJL_CALL(yajl_gen_map_close, jgen);
1219 OVS_YAJL_CALL(yajl_gen_array_close, jgen);
1221 /* make a request to subscribe to given table */
1222 OVS_YAJL_CALL(yajl_gen_get_buf, jgen, (const unsigned char **)¶ms,
1224 if (ovs_db_send_request(pdb, "monitor", params, result_cb) < 0) {
1225 OVS_ERROR("Failed to subscribe to \"%s\" table", tb_name);
1230 /* release memory */
1231 yajl_gen_clear(jgen);
1236 ovs_db_destroy(ovs_db_t *pdb)
1245 /* try to lock the structure before releasing */
1246 if (ret = pthread_mutex_lock(&pdb->mutex)) {
1247 OVS_ERROR("pthread_mutex_lock() DB mutext lock failed (%d)", ret);
1251 /* stop poll thread */
1252 if (ovs_db_event_thread_stop(pdb) < 0) {
1253 OVS_ERROR("stop poll thread failed");
1257 /* stop event thread */
1258 if (ovs_db_poll_thread_stop(pdb) < 0) {
1259 OVS_ERROR("stop event thread failed");
1263 /* unsubscribe callbacks */
1264 ovs_db_callback_remove_all(pdb);
1266 /* close connection */
1268 close(pdb->conn.sock);
1270 /* release DB handler */
1271 pthread_mutex_unlock(&pdb->mutex);
1272 pthread_mutex_destroy(&pdb->mutex);
1278 * Public OVS utils API implementation
1281 /* Get YAJL value by key from YAJL dictionary */
1283 ovs_utils_get_value_by_key(yajl_val jval, const char *key)
1285 const char *obj_key = NULL;
1288 if (!YAJL_IS_OBJECT(jval) || !key)
1291 /* find a value by key */
1292 for (int i = 0; i < YAJL_GET_OBJECT(jval)->len; i++) {
1293 obj_key = YAJL_GET_OBJECT(jval)->keys[i];
1294 if (strcmp(obj_key, key) == 0)
1295 return YAJL_GET_OBJECT(jval)->values[i];