* Volodymyr Mytnyk <volodymyrx.mytnyk@intel.com>
**/
+#include "collectd.h"
+
#include "common.h" /* auxiliary functions */
#include "utils_ovs.h" /* OVS helpers */
#define OVS_EVENTS_IFACE_UUID_SIZE 64
#define OVS_EVENTS_EXT_IFACE_ID_SIZE 64
#define OVS_EVENTS_EXT_VM_UUID_SIZE 64
-#define OVS_EVENTS_OVS_DB_URL_SIZE 64
#define OVS_EVENTS_PLUGIN "ovs_events"
#define OVS_EVENTS_CTX_LOCK \
for (int __i = ovs_events_ctx_lock(); __i != 0; __i = ovs_events_ctx_unlock())
/* OVS events configuration data */
struct ovs_events_config_s {
- _Bool send_notification; /* sent notification to collectd? */
- char ovs_db_server_url[OVS_EVENTS_OVS_DB_URL_SIZE]; /* OVS DB server URL */
- ovs_events_iface_list_t *ifaces; /* interface info */
+ _Bool send_notification; /* sent notification to collectd? */
+ char ovs_db_node[OVS_DB_ADDR_NODE_SIZE]; /* OVS DB node */
+ char ovs_db_serv[OVS_DB_ADDR_SERVICE_SIZE]; /* OVS DB service */
+ ovs_events_iface_list_t *ifaces; /* interface info */
};
typedef struct ovs_events_config_s ovs_events_config_t;
*/
static ovs_events_ctx_t ovs_events_ctx = {
.mutex = PTHREAD_MUTEX_INITIALIZER,
- .config = {.send_notification = 0, /* do not send notification */
- .ovs_db_server_url =
- "tcp:127.0.0.1:6640", /* use default OVS DB URL */
+ .config = {.send_notification = 0, /* do not send notification */
+ .ovs_db_node = "localhost", /* use default OVS DB node */
+ .ovs_db_serv = "6640", /* use default OVS DB service */
.ifaces = NULL},
.ovs_db_select_params = NULL,
.is_db_available = 0,
/* This function is used only by "OVS_EVENTS_CTX_LOCK" define (see above).
* It always returns 1 when context is locked.
*/
-static inline int ovs_events_ctx_lock() {
+static int ovs_events_ctx_lock() {
pthread_mutex_lock(&ovs_events_ctx.mutex);
return (1);
}
/* This function is used only by "OVS_EVENTS_CTX_LOCK" define (see above).
* It always returns 0 when context is unlocked.
*/
-static inline int ovs_events_ctx_unlock() {
+static int ovs_events_ctx_unlock() {
pthread_mutex_unlock(&ovs_events_ctx.mutex);
return (0);
}
/* Get OVS DB select parameter request based on rfc7047,
* "Transact" & "Select" section
*/
-static inline char *ovs_events_get_select_params() {
+static char *ovs_events_get_select_params() {
int ret = 0;
size_t buff_size = 0;
size_t offset = 0;
if (cf_util_get_boolean(child, &ovs_events_ctx.config.send_notification) <
0)
OVS_EVENTS_CONFIG_ERROR(child->key);
- } else if (strcasecmp("OvsDbServerUrl", child->key) == 0) {
- if (cf_util_get_string_buffer(
- child, ovs_events_ctx.config.ovs_db_server_url,
- sizeof(ovs_events_ctx.config.ovs_db_server_url)) < 0)
- OVS_EVENTS_CONFIG_ERROR(child->key);
+ } else if (strcasecmp("OvsDbAddress", child->key) == 0) {
+ if (child->values_num < 1) {
+ ERROR(OVS_EVENTS_PLUGIN ": invalid OVS DB address specified");
+ goto failure;
+ }
+ /* check node type and get the value */
+ if (child->values[0].type != OCONFIG_TYPE_STRING) {
+ ERROR(OVS_EVENTS_PLUGIN ": OVS DB node is not a string");
+ goto failure;
+ }
+ sstrncpy(ovs_events_ctx.config.ovs_db_node, child->values[0].value.string,
+ sizeof(ovs_events_ctx.config.ovs_db_node));
+ /* get OVS DB address service name (optional) */
+ if (child->values_num > 1) {
+ if (child->values[1].type != OCONFIG_TYPE_STRING) {
+ ERROR(OVS_EVENTS_PLUGIN ": OVS DB service is not a string");
+ goto failure;
+ }
+ sstrncpy(ovs_events_ctx.config.ovs_db_serv,
+ child->values[1].value.string,
+ sizeof(ovs_events_ctx.config.ovs_db_serv));
+ }
} else if (strcasecmp("Interfaces", child->key) == 0) {
for (int j = 0; j < child->values_num; j++) {
/* check value type */
ovs_db_callback_t cb = {.post_conn_init = ovs_events_conn_initialize,
.post_conn_terminate = ovs_events_conn_terminate};
- DEBUG(OVS_EVENTS_PLUGIN ": OVS DB url = %s",
- ovs_events_ctx.config.ovs_db_server_url);
+ DEBUG(OVS_EVENTS_PLUGIN ": OVS DB node = %s, service=%s",
+ ovs_events_ctx.config.ovs_db_node, ovs_events_ctx.config.ovs_db_serv);
/* generate OVS DB select condition based on list on configured interfaces */
ovs_events_ctx.ovs_db_select_params = ovs_events_get_select_params();
}
/* initialize OVS DB */
- ovs_db = ovs_db_init(ovs_events_ctx.config.ovs_db_server_url, &cb);
+ ovs_db = ovs_db_init(ovs_events_ctx.config.ovs_db_node,
+ ovs_events_ctx.config.ovs_db_serv, &cb);
if (ovs_db == NULL) {
ERROR(OVS_EVENTS_PLUGIN ": fail to connect to OVS DB server");
goto ovs_events_failure;
/* clang-format on */
/* collectd headers */
+#include "collectd.h"
+
#include "common.h"
/* private headers */
#include "utils_ovs.h"
/* system libraries */
+#if HAVE_NETDB_H
+#include <netdb.h>
+#endif
+#if HAVE_ARPA_INET_H
#include <arpa/inet.h>
+#endif
+#if HAVE_POLL_H
#include <poll.h>
-#include <semaphore.h>
+#endif
+#if HAVE_SYS_UN_H
#include <sys/un.h>
+#endif
+
+#include <semaphore.h>
#define OVS_ERROR(fmt, ...) \
do { \
};
typedef struct ovs_callback_s ovs_callback_t;
-/* Connection declaration */
-struct ovs_conn_s {
- int sock;
- int domain;
- int type;
- int addr_size;
- union {
- struct sockaddr_in s_inet;
- struct sockaddr_un s_unix;
- } addr;
-};
-typedef struct ovs_conn_s ovs_conn_t;
-
/* Event thread data declaration */
struct ovs_event_thread_s {
pthread_t tid;
pthread_mutex_t mutex;
ovs_callback_t *remote_cb;
ovs_db_callback_t cb;
- ovs_conn_t conn;
+ char service[OVS_DB_ADDR_SERVICE_SIZE];
+ char node[OVS_DB_ADDR_NODE_SIZE];
+ int sock;
};
-typedef struct ovs_db_s ovs_db_t;
/* Post an event to event thread.
* Possible events are:
/* Check if POLL thread is still running. Returns
* 1 if running otherwise 0 is returned */
-static inline int ovs_db_poll_is_running(ovs_db_t *pdb) {
+static _Bool ovs_db_poll_is_running(ovs_db_t *pdb) {
int state = 0;
pthread_mutex_lock(&pdb->poll_thread.mutex);
state = pdb->poll_thread.state;
}
/* Terminate POLL thread */
-static inline void ovs_db_poll_terminate(ovs_db_t *pdb) {
+static void ovs_db_poll_terminate(ovs_db_t *pdb) {
pthread_mutex_lock(&pdb->poll_thread.mutex);
pdb->poll_thread.state = OVS_DB_POLL_STATE_EXITING;
pthread_mutex_unlock(&pdb->poll_thread.mutex);
pthread_mutex_lock(&pdb->mutex);
for (ovs_callback_t *del_cb = pdb->remote_cb; pdb->remote_cb;
del_cb = pdb->remote_cb) {
- pdb->remote_cb = pdb->remote_cb->next;
+ pdb->remote_cb = del_cb->next;
free(del_cb);
}
pdb->remote_cb = NULL;
}
/* Get/find callback in OVS DB object by UID. Returns pointer
- * to requested callback otherwise NULL is returned */
+ * to requested callback otherwise NULL is returned.
+ *
+ * IMPORTANT NOTE:
+ * The OVS DB mutex should be locked by the caller
+ * to make sure that returned callback is still valid.
+ */
static ovs_callback_t *ovs_db_callback_get(ovs_db_t *pdb, uint64_t uid) {
- pthread_mutex_lock(&pdb->mutex);
for (ovs_callback_t *cb = pdb->remote_cb; cb != NULL; cb = cb->next)
- if (cb->uid == uid) {
- pthread_mutex_unlock(&pdb->mutex);
+ if (cb->uid == uid)
return cb;
- }
- pthread_mutex_unlock(&pdb->mutex);
return NULL;
}
size_t off = 0;
while (rem > 0) {
- if ((nbytes = send(pdb->conn.sock, data + off, rem, 0)) <= 0)
+ if ((nbytes = send(pdb->sock, data + off, rem, 0)) <= 0)
return (-1);
rem -= (size_t)nbytes;
off += (size_t)nbytes;
return (0);
}
-/* Parse OVS server URL.
- * Format of the URL:
- * "tcp:a.b.c.d:port" - define TCP connection (INET domain)
- * "unix:file" - define UNIX socket file (UNIX domain)
- */
-static int ovs_db_url_parse(const char *surl, ovs_conn_t *conn) {
- ovs_conn_t tmp_conn;
- char *nexttok = NULL;
- char *in_str = NULL;
- char *saveptr;
- int ret = 0;
-
- /* sanity check */
- if ((surl == NULL) || (strlen(surl) < 1))
- return (-1);
-
- /* parse domain */
- tmp_conn = *conn;
- in_str = sstrdup(surl);
- if ((nexttok = strtok_r(in_str, ":", &saveptr)) != NULL) {
- if (strcmp("tcp", nexttok) == 0) {
- tmp_conn.domain = AF_INET;
- tmp_conn.type = SOCK_STREAM;
- tmp_conn.addr_size = sizeof(tmp_conn.addr.s_inet);
- } else if (strcmp("unix", nexttok) == 0) {
- tmp_conn.domain = AF_UNIX;
- tmp_conn.type = SOCK_STREAM;
- tmp_conn.addr_size = sizeof(tmp_conn.addr.s_unix);
- } else
- goto failure;
- } else
- goto failure;
-
- /* parse url depending on domain */
- if ((nexttok = strtok_r(NULL, ":", &saveptr)) != NULL) {
- if (tmp_conn.domain == AF_UNIX) {
- /* <UNIX-NAME> */
- tmp_conn.addr.s_inet.sin_family = AF_UNIX;
- sstrncpy(tmp_conn.addr.s_unix.sun_path, nexttok, strlen(nexttok) + 1);
- } else {
- /* <IP:PORT> */
- tmp_conn.addr.s_inet.sin_family = AF_INET;
- ret = inet_pton(AF_INET, nexttok, (void *)&tmp_conn.addr.s_inet.sin_addr);
- if (ret == 1) {
- if ((nexttok = strtok_r(NULL, ":", &saveptr)) != NULL)
- tmp_conn.addr.s_inet.sin_port = htons(atoi(nexttok));
- else
- goto failure;
- } else
- goto failure;
- }
- }
-
- /* save result and return success */
- *conn = tmp_conn;
- sfree(in_str);
- return (0);
-
-failure:
- OVS_ERROR("invalid OVS DB URL provided [url=%s]", surl);
- sfree(in_str);
- return (-1);
-}
-
/*
* YAJL (Yet Another JSON Library) helper functions
* Documentation (https://lloyd.github.io/yajl/)
* jgen - YAJL generator handle allocated by yajl_gen_alloc()
* string - Null-terminated string
*/
-static inline yajl_gen_status ovs_yajl_gen_tstring(yajl_gen hander,
- const char *string) {
- return yajl_gen_string(hander, string, strlen(string));
+static yajl_gen_status ovs_yajl_gen_tstring(yajl_gen hander,
+ const char *string) {
+ return yajl_gen_string(hander, (const unsigned char *)string, strlen(string));
}
/* Add YAJL value into YAJL generator handle (JSON object)
/* Get OVS DB registered callback by YAJL val. The YAJL
* value should be YAJL string (UID). Returns NULL if
- * callback hasn't been found.
+ * callback hasn't been found. See also ovs_db_callback_get()
+ * description for addition info.
*/
static ovs_callback_t *ovs_db_table_callback_get(ovs_db_t *pdb, yajl_val jid) {
char *endptr = NULL;
goto ovs_failure;
/* find registered callback based on <json-value> */
+ pthread_mutex_lock(&pdb->mutex);
cb = ovs_db_table_callback_get(pdb, jvalue);
- if (cb == NULL || cb->table.call == NULL)
+ if (cb == NULL || cb->table.call == NULL) {
+ pthread_mutex_unlock(&pdb->mutex);
goto ovs_failure;
+ }
/* call registered callback */
cb->table.call(jtable_updates);
+ pthread_mutex_unlock(&pdb->mutex);
return 0;
ovs_failure:
return (-1);
/* try to find registered callback */
+ pthread_mutex_lock(&pdb->mutex);
cb = ovs_db_table_callback_get(pdb, jid);
if (cb != NULL && cb->result.call != NULL) {
/* call registered callback */
sem_post(&cb->result.sync);
}
+ pthread_mutex_unlock(&pdb->mutex);
return (0);
}
}
/* get method name */
- if (jval = yajl_tree_get(jnode, method_path, yajl_t_string)) {
+ if ((jval = yajl_tree_get(jnode, method_path, yajl_t_string)) != NULL) {
method = YAJL_GET_STRING(jval);
if (strcmp("echo", method) == 0) {
/* echo request from the server */
if (ovs_db_table_update_cb(pdb, jnode) < 0)
OVS_ERROR("handle update notification failed");
}
- } else if (jval = yajl_tree_get(jnode, result_path, yajl_t_any)) {
+ } else if ((jval = yajl_tree_get(jnode, result_path, yajl_t_any)) != NULL) {
/* result notification */
if (ovs_db_result_cb(pdb, jnode) < 0)
OVS_ERROR("handle result reply failed");
*/
/* Allocate JSON reader instance */
-static inline ovs_json_reader_t *ovs_json_reader_alloc() {
+static ovs_json_reader_t *ovs_json_reader_alloc() {
ovs_json_reader_t *jreader = NULL;
if ((jreader = calloc(sizeof(ovs_json_reader_t), 1)) == NULL)
}
/* Push raw data into into the JSON reader for processing */
-static inline int ovs_json_reader_push_data(ovs_json_reader_t *jreader,
- const char *data, size_t data_len) {
+static int ovs_json_reader_push_data(ovs_json_reader_t *jreader,
+ const char *data, size_t data_len) {
char *new_buff = NULL;
size_t available = jreader->buff_size - jreader->buff_offset;
/* Pop one fully-fledged JSON if already exists. Returns 0 if
* completed JSON already exists otherwise negative value is
* returned */
-static inline int ovs_json_reader_pop(ovs_json_reader_t *jreader,
- const char **json_ptr,
- size_t *json_len_ptr) {
+static int ovs_json_reader_pop(ovs_json_reader_t *jreader,
+ const char **json_ptr, size_t *json_len_ptr) {
size_t nbraces = 0;
size_t json_len = 0;
char *json = NULL;
/* Reset JSON reader. It is useful when start processing
* new raw data. E.g.: in case of lost stream connection.
*/
-static inline void ovs_json_reader_reset(ovs_json_reader_t *jreader) {
+static void ovs_json_reader_reset(ovs_json_reader_t *jreader) {
if (jreader) {
jreader->buff_offset = 0;
jreader->json_offset = 0;
}
/* Release internal data allocated for JSON reader */
-static inline void ovs_json_reader_free(ovs_json_reader_t *jreader) {
+static void ovs_json_reader_free(ovs_json_reader_t *jreader) {
if (jreader) {
free(jreader->buff_ptr);
free(jreader);
}
}
-/* Reconnect to OVD DB and call init OVS DB callback
- * 'init_cb' if connection has been established.
+/* Reconnect to OVD DB and call the OVS DB post connection init callback
+ * if connection has been established.
*/
static int ovs_db_reconnect(ovs_db_t *pdb) {
char errbuff[OVS_ERROR_BUFF_SIZE];
+ const char unix_prefix[] = "unix:";
+ struct addrinfo *result, *rp;
+ _Bool is_connected = 0;
+ struct sockaddr_un saunix;
/* remove all registered OVS DB table/result callbacks */
ovs_db_callback_remove_all(pdb);
- /* open new socket */
- if ((pdb->conn.sock = socket(pdb->conn.domain, pdb->conn.type, 0)) < 0) {
- sstrerror(errno, errbuff, sizeof(errbuff));
- OVS_ERROR("socket(): %s", errbuff);
- return (-1);
+ if (strncmp(pdb->node, unix_prefix, strlen(unix_prefix)) == 0) {
+ /* create unix socket address */
+ rp = calloc(1, sizeof(struct addrinfo));
+ struct sockaddr_un *sa_unix = calloc(1, sizeof(struct sockaddr_un));
+ if (rp == NULL || sa_unix == NULL) {
+ sfree(rp);
+ sfree(sa_unix);
+ return (1);
+ }
+ rp->ai_family = AF_UNIX;
+ rp->ai_socktype = SOCK_STREAM;
+ rp->ai_addrlen = sizeof(*sa_unix);
+ rp->ai_addr = (struct sockaddr *)sa_unix;
+ sa_unix->sun_family = rp->ai_family;
+ sstrncpy(sa_unix->sun_path, (pdb->node + strlen(unix_prefix)),
+ sizeof(sa_unix->sun_path));
+ result = rp;
+ } else {
+ /* intet socket address */
+ int ret = 0;
+ struct addrinfo hints;
+
+ /* setup criteria for selecting the socket address */
+ memset(&hints, 0, sizeof(hints));
+ hints.ai_family = AF_UNSPEC;
+ hints.ai_socktype = SOCK_STREAM;
+
+ /* get socket addresses */
+ if ((ret = getaddrinfo(pdb->node, pdb->service, &hints, &result)) != 0) {
+ OVS_ERROR("getaddrinfo(): %s", gai_strerror(ret));
+ return (1);
+ }
}
-
- /* try to connect to server */
- if (connect(pdb->conn.sock, (struct sockaddr *)&pdb->conn.addr,
- pdb->conn.addr_size) < 0) {
- sstrerror(errno, errbuff, sizeof(errbuff));
- OVS_ERROR("connect(): %s", errbuff);
- close(pdb->conn.sock);
- return (-1);
+ /* try to connect to the server */
+ for (rp = result; rp != NULL; rp = rp->ai_next) {
+ if ((pdb->sock = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol)) <
+ 0) {
+ sstrerror(errno, errbuff, sizeof(errbuff));
+ OVS_DEBUG("socket(): %s", errbuff);
+ continue;
+ }
+ if (connect(pdb->sock, rp->ai_addr, rp->ai_addrlen) < 0) {
+ sstrerror(errno, errbuff, sizeof(errbuff));
+ OVS_DEBUG("connect(): %s [family=%d]", errbuff, rp->ai_family);
+ close(pdb->sock);
+ } else {
+ is_connected = 1;
+ break;
+ }
}
/* send notification to event thread */
- ovs_db_event_post(pdb, OVS_DB_EVENT_CONN_ESTABLISHED);
- return (0);
+ if (is_connected)
+ ovs_db_event_post(pdb, OVS_DB_EVENT_CONN_ESTABLISHED);
+ else
+ OVS_ERROR("connect to \"%s\" failed", pdb->node);
+
+ freeaddrinfo(result);
+ return !is_connected;
}
/* POLL worker thread.
}
/* start polling data */
- poll_fd.fd = pdb->conn.sock;
+ poll_fd.fd = pdb->sock;
poll_fd.events = POLLIN | POLLPRI;
poll_fd.revents = 0;
usleep(OVS_DB_RECONNECT_TIMEOUT * 1000000);
}
ovs_json_reader_reset(jreader);
- poll_fd.fd = pdb->conn.sock;
+ poll_fd.fd = pdb->sock;
} else if ((poll_fd.revents & POLLERR) || (poll_fd.revents & POLLHUP)) {
/* connection is broken */
OVS_ERROR("poll() peer closed its end of the channel");
* Public OVS DB API implementation
*/
-ovs_db_t *ovs_db_init(const char *surl, ovs_db_callback_t *cb) {
+ovs_db_t *ovs_db_init(const char *node, const char *service,
+ ovs_db_callback_t *cb) {
pthread_mutexattr_t mutex_attr;
ovs_db_t *pdb = NULL;
if ((pdb = calloc(1, sizeof(*pdb))) == NULL)
return (NULL);
- /* convert string url to socket addr */
- if (ovs_db_url_parse(surl, &pdb->conn) < 0)
- goto failure;
+ /* node cannot be unset */
+ if (node == NULL || strlen(node) == 0)
+ return (NULL);
+
+ /* store the OVS DB address */
+ sstrncpy(pdb->node, node, sizeof(pdb->node));
+ if (service != NULL)
+ sstrncpy(pdb->service, service, sizeof(pdb->service));
/* setup OVS DB callbacks */
if (cb)
return pdb;
failure:
- if (pdb->conn.sock)
+ if (pdb->sock)
/* close connection */
- close(pdb->conn.sock);
+ close(pdb->sock);
if (pdb->event_thread.tid != 0)
/* stop event thread */
if (ovs_db_event_thread_stop(pdb) < 0)
return (-1);
/* try to lock the structure before releasing */
- if (ret = pthread_mutex_lock(&pdb->mutex)) {
+ if ((ret = pthread_mutex_lock(&pdb->mutex))) {
OVS_ERROR("pthread_mutex_lock() DB mutext lock failed (%d)", ret);
return (-1);
}
ovs_db_callback_remove_all(pdb);
/* close connection */
- if (pdb->conn.sock)
- close(pdb->conn.sock);
+ if (pdb->sock)
+ close(pdb->sock);
/* release DB handler */
pthread_mutex_unlock(&pdb->mutex);
projects / collectd.git / commitdiff