2 * collectd - src/write_stackdriver.c
5 * Copyright (C) 2017 Florian Forster
7 * Permission to use, copy, modify, and/or distribute this software for any
8 * purpose with or without fee is hereby granted, provided that the above
9 * copyright notice and this permission notice appear in all copies.
11 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
12 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
13 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
14 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
15 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
16 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
17 * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
20 * Florian Forster <octo at collectd.org>
26 #include "configfile.h"
28 #include "utils_format_stackdriver.h"
29 #include "utils_gce.h"
30 #include "utils_oauth.h"
32 #include <curl/curl.h>
34 #include <yajl/yajl_tree.h>
40 #define GCM_API_URL "https://monitoring.googleapis.com/v3"
43 #ifndef MONITORING_SCOPE
44 #define MONITORING_SCOPE "https://www.googleapis.com/auth/monitoring"
47 struct wg_callback_s {
52 sd_resource_t *resource;
56 sd_output_t *formatter;
58 char curl_errbuf[CURL_ERROR_SIZE];
60 size_t timeseries_count;
61 cdtime_t send_buffer_init_time;
65 typedef struct wg_callback_s wg_callback_t;
71 typedef struct wg_memory_s wg_memory_t;
73 static size_t wg_write_memory_cb(void *contents, size_t size,
74 size_t nmemb, /* {{{ */
76 size_t realsize = size * nmemb;
77 wg_memory_t *mem = (wg_memory_t *)userp;
79 if (0x7FFFFFF0 < mem->size || 0x7FFFFFF0 - mem->size < realsize) {
80 ERROR("integer overflow");
84 mem->memory = (char *)realloc((void *)mem->memory, mem->size + realsize + 1);
85 if (mem->memory == NULL) {
87 ERROR("wg_write_memory_cb: not enough memory (realloc returned NULL)");
91 memcpy(&(mem->memory[mem->size]), contents, realsize);
92 mem->size += realsize;
93 mem->memory[mem->size] = 0;
95 } /* }}} size_t wg_write_memory_cb */
97 static char *wg_get_authorization_header(wg_callback_t *cb) { /* {{{ */
99 char access_token[256];
100 char authorization_header[256];
102 assert((cb->auth != NULL) || gce_check());
103 if (cb->auth != NULL)
104 status = oauth_access_token(cb->auth, access_token, sizeof(access_token));
106 status = gce_access_token(cb->email, access_token, sizeof(access_token));
108 ERROR("write_stackdriver plugin: Failed to get access token");
112 status = snprintf(authorization_header, sizeof(authorization_header),
113 "Authorization: Bearer %s", access_token);
114 if ((status < 1) || ((size_t)status >= sizeof(authorization_header)))
117 return strdup(authorization_header);
118 } /* }}} char *wg_get_authorization_header */
125 static api_error_t *parse_api_error(char const *body) {
127 yajl_val root = yajl_tree_parse(body, errbuf, sizeof(errbuf));
129 ERROR("write_stackdriver plugin: yajl_tree_parse failed: %s", errbuf);
133 api_error_t *err = calloc(1, sizeof(*err));
135 ERROR("write_stackdriver plugin: calloc failed");
136 yajl_tree_free(root);
140 yajl_val code = yajl_tree_get(root, (char const *[]){"error", "code", NULL},
142 if (YAJL_IS_INTEGER(code)) {
143 err->code = YAJL_GET_INTEGER(code);
146 yajl_val message = yajl_tree_get(
147 root, (char const *[]){"error", "message", NULL}, yajl_t_string);
148 if (YAJL_IS_STRING(message)) {
149 char const *m = YAJL_GET_STRING(message);
151 err->message = strdup(m);
158 static char *api_error_string(api_error_t *err, char *buffer,
159 size_t buffer_size) {
161 strncpy(buffer, "Unknown error (API error is NULL)", buffer_size);
162 } else if (err->message == NULL) {
163 snprintf(buffer, buffer_size, "API error %d", err->code);
165 snprintf(buffer, buffer_size, "API error %d: %s", err->code, err->message);
170 #define API_ERROR_STRING(err) api_error_string(err, (char[1024]){""}, 1024)
172 // do_post does a HTTP POST request, assuming a JSON payload and using OAuth
173 // authentication. Returns -1 on error and the HTTP status code otherwise.
174 // ret_content, if not NULL, will contain the server's response.
175 // If ret_content is provided and the server responds with a 4xx or 5xx error,
176 // an appropriate message will be logged.
177 static int do_post(wg_callback_t *cb, char const *url, void const *payload,
178 wg_memory_t *ret_content) {
179 if (cb->curl == NULL) {
180 cb->curl = curl_easy_init();
181 if (cb->curl == NULL) {
182 ERROR("write_stackdriver plugin: curl_easy_init() failed");
186 curl_easy_setopt(cb->curl, CURLOPT_ERRORBUFFER, cb->curl_errbuf);
187 curl_easy_setopt(cb->curl, CURLOPT_NOSIGNAL, 1L);
190 curl_easy_setopt(cb->curl, CURLOPT_POST, 1L);
191 curl_easy_setopt(cb->curl, CURLOPT_URL, url);
193 long timeout_ms = 2 * CDTIME_T_TO_MS(plugin_get_interval());
194 if (timeout_ms < 10000) {
197 curl_easy_setopt(cb->curl, CURLOPT_TIMEOUT_MS, timeout_ms);
200 char *auth_header = wg_get_authorization_header(cb);
201 if (auth_header == NULL) {
202 ERROR("write_stackdriver plugin: getting access token failed with");
206 struct curl_slist *headers =
207 curl_slist_append(NULL, "Content-Type: application/json");
208 headers = curl_slist_append(headers, auth_header);
209 curl_easy_setopt(cb->curl, CURLOPT_HTTPHEADER, headers);
211 curl_easy_setopt(cb->curl, CURLOPT_POSTFIELDS, payload);
213 curl_easy_setopt(cb->curl, CURLOPT_WRITEFUNCTION,
214 ret_content ? wg_write_memory_cb : NULL);
215 curl_easy_setopt(cb->curl, CURLOPT_WRITEDATA, ret_content);
217 int status = curl_easy_perform(cb->curl);
219 /* clean up that has to happen in any case */
220 curl_slist_free_all(headers);
222 curl_easy_setopt(cb->curl, CURLOPT_HTTPHEADER, NULL);
223 curl_easy_setopt(cb->curl, CURLOPT_WRITEFUNCTION, NULL);
224 curl_easy_setopt(cb->curl, CURLOPT_WRITEDATA, NULL);
226 if (status != CURLE_OK) {
227 ERROR("write_stackdriver plugin: POST %s failed: %s", url, cb->curl_errbuf);
228 sfree(ret_content->memory);
229 ret_content->size = 0;
234 curl_easy_getinfo(cb->curl, CURLINFO_RESPONSE_CODE, &http_code);
236 if (ret_content != NULL) {
237 if ((status >= 400) && (status < 500)) {
238 ERROR("write_stackdriver plugin: POST %s: %s", url,
239 API_ERROR_STRING(parse_api_error(ret_content->memory)));
240 } else if (status >= 500) {
241 WARNING("write_stackdriver plugin: POST %s: %s", url,
242 ret_content->memory);
246 return (int)http_code;
249 static int wg_call_metricdescriptor_create(wg_callback_t *cb,
250 char const *payload) {
252 snprintf(url, sizeof(url), "%s/projects/%s/metricDescriptors", cb->url,
254 wg_memory_t response = {0};
256 int status = do_post(cb, url, payload, &response);
258 ERROR("write_stackdriver plugin: POST %s failed", url);
261 sfree(response.memory);
264 ERROR("write_stackdriver plugin: POST %s: unexpected response code: got "
270 } /* int wg_call_metricdescriptor_create */
272 static int wg_call_timeseries_write(wg_callback_t *cb, char const *payload) {
274 snprintf(url, sizeof(url), "%s/projects/%s/timeSeries", cb->url, cb->project);
275 wg_memory_t response = {0};
277 int status = do_post(cb, url, payload, &response);
279 ERROR("write_stackdriver plugin: POST %s failed", url);
282 sfree(response.memory);
285 ERROR("write_stackdriver plugin: POST %s: unexpected response code: got "
291 } /* int wg_call_timeseries_write */
293 static void wg_reset_buffer(wg_callback_t *cb) /* {{{ */
295 cb->timeseries_count = 0;
296 cb->send_buffer_init_time = cdtime();
297 } /* }}} wg_reset_buffer */
299 static int wg_callback_init(wg_callback_t *cb) /* {{{ */
301 if (cb->curl != NULL)
304 cb->formatter = sd_output_create(cb->resource);
305 if (cb->formatter == NULL) {
306 ERROR("write_stackdriver plugin: sd_output_create failed.");
310 cb->curl = curl_easy_init();
311 if (cb->curl == NULL) {
312 ERROR("write_stackdriver plugin: curl_easy_init failed.");
316 curl_easy_setopt(cb->curl, CURLOPT_NOSIGNAL, 1L);
317 curl_easy_setopt(cb->curl, CURLOPT_USERAGENT,
318 PACKAGE_NAME "/" PACKAGE_VERSION);
319 curl_easy_setopt(cb->curl, CURLOPT_ERRORBUFFER, cb->curl_errbuf);
323 } /* }}} int wg_callback_init */
325 static int wg_flush_nolock(cdtime_t timeout, wg_callback_t *cb) /* {{{ */
327 if (cb->timeseries_count == 0) {
328 cb->send_buffer_init_time = cdtime();
332 /* timeout == 0 => flush unconditionally */
334 cdtime_t now = cdtime();
336 if ((cb->send_buffer_init_time + timeout) > now)
340 char *payload = sd_output_reset(cb->formatter);
341 int status = wg_call_timeseries_write(cb, payload);
343 ERROR("write_stackdriver plugin: Sending buffer failed with status %d.",
349 } /* }}} wg_flush_nolock */
351 static int wg_flush(cdtime_t timeout, /* {{{ */
352 const char *identifier __attribute__((unused)),
353 user_data_t *user_data) {
357 if (user_data == NULL)
360 cb = user_data->data;
362 pthread_mutex_lock(&cb->lock);
364 if (cb->curl == NULL) {
365 status = wg_callback_init(cb);
367 ERROR("write_stackdriver plugin: wg_callback_init failed.");
368 pthread_mutex_unlock(&cb->lock);
373 status = wg_flush_nolock(timeout, cb);
374 pthread_mutex_unlock(&cb->lock);
377 } /* }}} int wg_flush */
379 static void wg_callback_free(void *data) /* {{{ */
381 wg_callback_t *cb = data;
385 sd_output_destroy(cb->formatter);
386 cb->formatter = NULL;
392 oauth_destroy(cb->auth);
394 curl_easy_cleanup(cb->curl);
398 } /* }}} void wg_callback_free */
400 static int wg_metric_descriptors_create(wg_callback_t *cb, const data_set_t *ds,
401 const value_list_t *vl) {
403 for (size_t i = 0; i < ds->ds_num; i++) {
406 int status = sd_format_metric_descriptor(buffer, sizeof(buffer), ds, vl, i);
408 ERROR("write_stackdriver plugin: sd_format_metric_descriptor failed "
415 status = wg_call_metricdescriptor_create(cb, buffer);
417 ERROR("write_stackdriver plugin: wg_call_metricdescriptor_create failed "
425 return sd_output_register_metric(cb->formatter, ds, vl);
426 } /* }}} int wg_metric_descriptors_create */
428 static int wg_write(const data_set_t *ds, const value_list_t *vl, /* {{{ */
429 user_data_t *user_data) {
430 wg_callback_t *cb = user_data->data;
434 pthread_mutex_lock(&cb->lock);
436 if (cb->curl == NULL) {
437 int status = wg_callback_init(cb);
439 ERROR("write_stackdriver plugin: wg_callback_init failed.");
440 pthread_mutex_unlock(&cb->lock);
447 status = sd_output_add(cb->formatter, ds, vl);
448 if (status == 0) { /* success */
450 } else if (status == ENOBUFS) { /* success, flush */
451 wg_flush_nolock(0, cb);
454 } else if (status == EEXIST) {
455 /* metric already in the buffer; flush and retry */
456 wg_flush_nolock(0, cb);
458 } else if (status == ENOENT) {
459 /* new metric, create metric descriptor first */
460 status = wg_metric_descriptors_create(cb, ds, vl);
471 cb->timeseries_count++;
474 pthread_mutex_unlock(&cb->lock);
476 } /* }}} int wg_write */
478 static void wg_check_scope(char const *email) /* {{{ */
480 char *scope = gce_scope(email);
482 WARNING("write_stackdriver plugin: Unable to determine scope of this "
487 if (strstr(scope, MONITORING_SCOPE) == NULL) {
490 /* Strip trailing newline characers for printing. */
491 scope_len = strlen(scope);
492 while ((scope_len > 0) && (iscntrl((int)scope[scope_len - 1])))
493 scope[--scope_len] = 0;
495 WARNING("write_stackdriver plugin: The determined scope of this instance "
496 "(\"%s\") does not contain the monitoring scope (\"%s\"). You need "
497 "to add this scope to the list of scopes passed to gcutil with "
498 "--service_account_scopes when creating the instance. "
499 "Alternatively, to use this plugin on an instance which does not "
500 "have this scope, use a Service Account.",
501 scope, MONITORING_SCOPE);
505 } /* }}} void wg_check_scope */
507 static int wg_config_resource(oconfig_item_t *ci, wg_callback_t *cb) /* {{{ */
509 if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_STRING)) {
510 ERROR("write_stackdriver plugin: The \"%s\" option requires exactly one "
516 char *resource_type = ci->values[0].value.string;
518 if (cb->resource != NULL) {
519 sd_resource_destroy(cb->resource);
522 cb->resource = sd_resource_create(resource_type);
523 if (cb->resource == NULL) {
524 ERROR("write_stackdriver plugin: sd_resource_create(\"%s\") failed.",
529 for (int i = 0; i < ci->children_num; i++) {
530 oconfig_item_t *child = ci->children + i;
532 if (strcasecmp("Label", child->key) == 0) {
533 if ((child->values_num != 2) ||
534 (child->values[0].type != OCONFIG_TYPE_STRING) ||
535 (child->values[1].type != OCONFIG_TYPE_STRING)) {
536 ERROR("write_stackdriver plugin: The \"Label\" option needs exactly "
537 "two string arguments.");
541 sd_resource_add_label(cb->resource, child->values[0].value.string,
542 child->values[1].value.string);
547 } /* }}} int wg_config_resource */
549 static int wg_config(oconfig_item_t *ci) /* {{{ */
555 wg_callback_t *cb = calloc(1, sizeof(*cb));
557 ERROR("write_stackdriver plugin: calloc failed.");
560 cb->url = strdup(GCM_API_URL);
561 pthread_mutex_init(&cb->lock, /* attr = */ NULL);
563 char *credential_file = NULL;
565 for (int i = 0; i < ci->children_num; i++) {
566 oconfig_item_t *child = ci->children + i;
567 if (strcasecmp("Project", child->key) == 0)
568 cf_util_get_string(child, &cb->project);
569 else if (strcasecmp("Email", child->key) == 0)
570 cf_util_get_string(child, &cb->email);
571 else if (strcasecmp("Url", child->key) == 0)
572 cf_util_get_string(child, &cb->url);
573 else if (strcasecmp("CredentialFile", child->key) == 0)
574 cf_util_get_string(child, &credential_file);
575 else if (strcasecmp("Resource", child->key) == 0)
576 wg_config_resource(child, cb);
578 ERROR("write_stackdriver plugin: Invalid configuration option: %s.",
580 wg_callback_free(cb);
585 /* Set up authentication */
586 /* Option 1: Credentials file given => use service account */
587 if (credential_file != NULL) {
589 oauth_create_google_file(credential_file, MONITORING_SCOPE);
590 if (cfg.oauth == NULL) {
591 ERROR("write_stackdriver plugin: oauth_create_google_file failed");
592 wg_callback_free(cb);
595 cb->auth = cfg.oauth;
597 if (cb->project == NULL) {
598 cb->project = cfg.project_id;
599 INFO("write_stackdriver plugin: Automatically detected project ID: "
603 sfree(cfg.project_id);
606 /* Option 2: Look for credentials in well-known places */
607 if (cb->auth == NULL) {
608 oauth_google_t cfg = oauth_create_google_default(MONITORING_SCOPE);
609 cb->auth = cfg.oauth;
611 if (cb->project == NULL) {
612 cb->project = cfg.project_id;
613 INFO("write_stackdriver plugin: Automatically detected project ID: "
617 sfree(cfg.project_id);
621 if ((cb->auth != NULL) && (cb->email != NULL)) {
622 NOTICE("write_stackdriver plugin: A service account email was configured "
624 "not used for authentication because %s used instead.",
625 (credential_file != NULL) ? "a credential file was"
626 : "application default credentials were");
629 /* Option 3: Running on GCE => use metadata service */
630 if ((cb->auth == NULL) && gce_check()) {
631 wg_check_scope(cb->email);
632 } else if (cb->auth == NULL) {
633 ERROR("write_stackdriver plugin: Unable to determine credentials. Please "
635 "specify the \"Credentials\" option or set up Application Default "
637 wg_callback_free(cb);
641 if ((cb->project == NULL) && gce_check()) {
642 cb->project = gce_project_id();
644 if (cb->project == NULL) {
645 ERROR("write_stackdriver plugin: Unable to determine the project number. "
646 "Please specify the \"Project\" option manually.");
647 wg_callback_free(cb);
651 if ((cb->resource == NULL) && gce_check()) {
652 /* TODO(octo): add error handling */
653 cb->resource = sd_resource_create("gce_instance");
654 sd_resource_add_label(cb->resource, "project_id", gce_project_id());
655 sd_resource_add_label(cb->resource, "instance_id", gce_instance_id());
656 sd_resource_add_label(cb->resource, "zone", gce_zone());
658 if (cb->resource == NULL) {
659 /* TODO(octo): add error handling */
660 cb->resource = sd_resource_create("global");
661 sd_resource_add_label(cb->resource, "project_id", cb->project);
664 DEBUG("write_stackdriver plugin: Registering write callback with URL %s",
666 assert((cb->auth != NULL) || gce_check());
668 user_data_t user_data = {
671 plugin_register_flush("write_stackdriver", wg_flush, &user_data);
673 user_data.free_func = wg_callback_free;
674 plugin_register_write("write_stackdriver", wg_write, &user_data);
677 } /* }}} int wg_config */
679 static int wg_init(void) {
681 /* Call this while collectd is still single-threaded to avoid
682 * initialization issues in libgcrypt. */
683 curl_global_init(CURL_GLOBAL_SSL);
686 } /* }}} int wg_init */
688 void module_register(void) /* {{{ */
690 plugin_register_complex_config("write_stackdriver", wg_config);
691 plugin_register_init("write_stackdriver", wg_init);
692 } /* }}} void module_register */