X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fwrite_mongodb.c;h=24151ceb720a3a29a055a0a24e7cdc6993661d45;hb=f2daf5deea1d50dbb834cffa0c8106b5456c9d82;hp=4f866bc84fc9158a2a266518fe2942a63dfa52f1;hpb=47d9b018d30223078ea44dee5305dfa2225ce95b;p=collectd.git diff --git a/src/write_mongodb.c b/src/write_mongodb.c index 4f866bc8..24151ceb 100644 --- a/src/write_mongodb.c +++ b/src/write_mongodb.c @@ -1,6 +1,6 @@ /** * collectd - src/write_mongodb.c - * Copyright (C) 2010-2012 Florian Forster + * Copyright (C) 2010-2013 Florian Forster * Copyright (C) 2010 Akkarit Sangpetch * Copyright (C) 2012 Chris Lundquist * @@ -23,7 +23,7 @@ * DEALINGS IN THE SOFTWARE. * * Authors: - * Florian Forster + * Florian Forster * Akkarit Sangpetch * Chris Lundquist **/ @@ -43,6 +43,11 @@ #endif #include +#if (MONGO_MAJOR == 0) && (MONGO_MINOR < 8) +# define bson_alloc() bson_create() +# define bson_dealloc(b) bson_dispose(b) +#endif + struct wm_node_s { char name[DATA_MAX_NAME_LEN]; @@ -51,7 +56,10 @@ struct wm_node_s int port; int timeout; - int connected; + /* Authentication information */ + char *db; + char *user; + char *passwd; _Bool store_rates; @@ -63,70 +71,145 @@ typedef struct wm_node_s wm_node_t; /* * Functions */ -static int wm_write (const data_set_t *ds, /* {{{ */ +static bson *wm_create_bson (const data_set_t *ds, /* {{{ */ const value_list_t *vl, - user_data_t *ud) + _Bool store_rates) { - wm_node_t *node = ud->data; - char collection_name[512]; - int status; + bson *ret; + gauge_t *rates; int i; - bson record; - gauge_t *rates = NULL; - if (node->store_rates) + ret = bson_alloc (); /* matched by bson_dealloc() */ + if (ret == NULL) + { + ERROR ("write_mongodb plugin: bson_create failed."); + return (NULL); + } + + if (store_rates) { rates = uc_get_rate (ds, vl); if (rates == NULL) { ERROR ("write_mongodb plugin: uc_get_rate() failed."); - return (-1); + return (NULL); } } + else + { + rates = NULL; + } - ssnprintf(collection_name, sizeof (collection_name), "collectd.%s", vl->plugin); - - bson_init(&record); - bson_append_date (&record, "time", (bson_date_t) CDTIME_T_TO_MS (vl->time)); - bson_append_string (&record, "host", vl->host); - bson_append_string (&record, "plugin_instance", vl->plugin_instance); - bson_append_string (&record, "type", vl->type); - bson_append_string (&record, "type_instance", vl->type_instance); + bson_init (ret); /* matched by bson_destroy() */ + bson_append_date (ret, "time", (bson_date_t) CDTIME_T_TO_MS (vl->time)); + bson_append_string (ret, "host", vl->host); + bson_append_string (ret, "plugin", vl->plugin); + bson_append_string (ret, "plugin_instance", vl->plugin_instance); + bson_append_string (ret, "type", vl->type); + bson_append_string (ret, "type_instance", vl->type_instance); + bson_append_start_array (ret, "values"); /* {{{ */ for (i = 0; i < ds->ds_num; i++) { + char key[16]; + + ssnprintf (key, sizeof (key), "%i", i); + if (ds->ds[i].type == DS_TYPE_GAUGE) - bson_append_double(&record, ds->ds[i].name, vl->values[i].gauge); - else if (node->store_rates) - bson_append_double(&record, ds->ds[i].name, (double) rates[i]); + bson_append_double(ret, key, vl->values[i].gauge); + else if (store_rates) + bson_append_double(ret, key, (double) rates[i]); else if (ds->ds[i].type == DS_TYPE_COUNTER) - bson_append_long(&record, ds->ds[i].name, vl->values[i].counter); + bson_append_long(ret, key, vl->values[i].counter); else if (ds->ds[i].type == DS_TYPE_DERIVE) - bson_append_long(&record, ds->ds[i].name, vl->values[i].derive); + bson_append_long(ret, key, vl->values[i].derive); else if (ds->ds[i].type == DS_TYPE_ABSOLUTE) - bson_append_long(&record, ds->ds[i].name, vl->values[i].absolute); + bson_append_long(ret, key, vl->values[i].absolute); else assert (23 == 42); } - /* We must finish the record, other wise the insert will fail */ - bson_finish(&record); + bson_append_finish_array (ret); /* }}} values */ + + bson_append_start_array (ret, "dstypes"); /* {{{ */ + for (i = 0; i < ds->ds_num; i++) + { + char key[16]; + + ssnprintf (key, sizeof (key), "%i", i); + + if (store_rates) + bson_append_string (ret, key, "gauge"); + else + bson_append_string (ret, key, DS_TYPE_TO_STRING (ds->ds[i].type)); + } + bson_append_finish_array (ret); /* }}} dstypes */ + + bson_append_start_array (ret, "dsnames"); /* {{{ */ + for (i = 0; i < ds->ds_num; i++) + { + char key[16]; + + ssnprintf (key, sizeof (key), "%i", i); + bson_append_string (ret, key, ds->ds[i].name); + } + bson_append_finish_array (ret); /* }}} dsnames */ + + bson_finish (ret); sfree (rates); + return (ret); +} /* }}} bson *wm_create_bson */ + +static int wm_write (const data_set_t *ds, /* {{{ */ + const value_list_t *vl, + user_data_t *ud) +{ + wm_node_t *node = ud->data; + char collection_name[512]; + bson *bson_record; + int status; + + ssnprintf (collection_name, sizeof (collection_name), "collectd.%s", + vl->plugin); + + bson_record = wm_create_bson (ds, vl, node->store_rates); + if (bson_record == NULL) + return (ENOMEM); pthread_mutex_lock (&node->lock); - if (node->connected == 0) + if (!mongo_is_connected (node->conn)) { - status = mongo_connect(node->conn, node->host, node->port); + INFO ("write_mongodb plugin: Connecting to [%s]:%i", + (node->host != NULL) ? node->host : "localhost", + (node->port != 0) ? node->port : MONGO_DEFAULT_PORT); + status = mongo_connect (node->conn, node->host, node->port); if (status != MONGO_OK) { - ERROR ("write_mongodb plugin: Connecting to host \"%s\" (port %i) failed.", + ERROR ("write_mongodb plugin: Connecting to [%s]:%i failed.", (node->host != NULL) ? node->host : "localhost", (node->port != 0) ? node->port : MONGO_DEFAULT_PORT); - mongo_destroy(node->conn); + mongo_destroy (node->conn); pthread_mutex_unlock (&node->lock); return (-1); } + if ((node->db != NULL) && (node->user != NULL) && (node->passwd != NULL)) + { + status = mongo_cmd_authenticate (node->conn, + node->db, node->user, node->passwd); + if (status != MONGO_OK) + { + ERROR ("write_mongodb plugin: Authenticating to [%s]%i for database " + "\"%s\" as user \"%s\" failed.", + (node->host != NULL) ? node->host : "localhost", + (node->port != 0) ? node->port : MONGO_DEFAULT_PORT, + node->db, node->user); + mongo_destroy (node->conn); + pthread_mutex_unlock (&node->lock); + return (-1); + } + } + if (node->timeout > 0) { status = mongo_set_op_timeout (node->conn, node->timeout); if (status != MONGO_OK) { @@ -134,30 +217,39 @@ static int wm_write (const data_set_t *ds, /* {{{ */ node->timeout, node->conn->errstr); } } - - node->connected = 1; } /* Assert if the connection has been established */ - assert (node->connected == 1); - - DEBUG ( "write_mongodb plugin: writing record"); - /* bson_print(&record); */ + assert (mongo_is_connected (node->conn)); - status = mongo_insert(node->conn,collection_name,&record); + #if MONGO_MINOR >= 6 + /* There was an API change in 0.6.0 as linked below */ + /* https://github.com/mongodb/mongo-c-driver/blob/master/HISTORY.md */ + status = mongo_insert (node->conn, collection_name, bson_record, NULL); + #else + status = mongo_insert (node->conn, collection_name, bson_record); + #endif - if(status != MONGO_OK) + if (status != MONGO_OK) { ERROR ( "write_mongodb plugin: error inserting record: %d", node->conn->err); - if (node->conn->err == MONGO_BSON_INVALID) + if (node->conn->err != MONGO_BSON_INVALID) ERROR ("write_mongodb plugin: %s", node->conn->errstr); - else if (record.err) - ERROR ("write_mongodb plugin: %s", record.errstr); + else + ERROR ("write_mongodb plugin: Invalid BSON structure, error = %#x", + (unsigned int) bson_record->err); + + /* Disconnect except on data errors. */ + if ((node->conn->err != MONGO_BSON_INVALID) + && (node->conn->err != MONGO_BSON_NOT_FINISHED)) + mongo_destroy (node->conn); } pthread_mutex_unlock (&node->lock); + /* free our resource as not to leak memory */ - bson_destroy(&record); + bson_destroy (bson_record); /* matches bson_init() */ + bson_dealloc (bson_record); /* matches bson_alloc() */ return (0); } /* }}} int wm_write */ @@ -169,11 +261,8 @@ static void wm_config_free (void *ptr) /* {{{ */ if (node == NULL) return; - if (node->connected != 0) - { - mongo_destroy(node->conn); - node->connected = 0; - } + if (mongo_is_connected (node->conn)) + mongo_destroy (node->conn); sfree (node->host); sfree (node); @@ -189,6 +278,7 @@ static int wm_config_node (oconfig_item_t *ci) /* {{{ */ if (node == NULL) return (ENOMEM); memset (node, 0, sizeof (*node)); + mongo_init (node->conn); node->host = NULL; node->store_rates = 1; pthread_mutex_init (&node->lock, /* attr = */ NULL); @@ -220,6 +310,12 @@ static int wm_config_node (oconfig_item_t *ci) /* {{{ */ status = cf_util_get_int (child, &node->timeout); else if (strcasecmp ("StoreRates", child->key) == 0) status = cf_util_get_boolean (child, &node->store_rates); + else if (strcasecmp ("Database", child->key) == 0) + status = cf_util_get_string (child, &node->db); + else if (strcasecmp ("User", child->key) == 0) + status = cf_util_get_string (child, &node->user); + else if (strcasecmp ("Password", child->key) == 0) + status = cf_util_get_string (child, &node->passwd); else WARNING ("write_mongodb plugin: Ignoring unknown config option \"%s\".", child->key); @@ -228,6 +324,20 @@ static int wm_config_node (oconfig_item_t *ci) /* {{{ */ break; } /* for (i = 0; i < ci->children_num; i++) */ + if ((node->db != NULL) || (node->user != NULL) || (node->passwd != NULL)) + { + if ((node->db == NULL) || (node->user == NULL) || (node->passwd == NULL)) + { + WARNING ("write_mongodb plugin: Authentication requires the " + "\"Database\", \"User\" and \"Password\" options to be specified, " + "but at last one of them is missing. Authentication will NOT be " + "used."); + sfree (node->db); + sfree (node->user); + sfree (node->passwd); + } + } + if (status == 0) { char cb_name[DATA_MAX_NAME_LEN];