return 0;
}
-struct web_client *web_client_create(int listener)
-{
+struct web_client *web_client_create(int listener) {
struct web_client *w;
w = callocz(1, sizeof(struct web_client));
if(w->prev) w->prev->next = w->next;
if(w->next) w->next->prev = w->prev;
- if(w->response.header_output) buffer_free(w->response.header_output);
- if(w->response.header) buffer_free(w->response.header);
- if(w->response.data) buffer_free(w->response.data);
+ buffer_free(w->response.header_output);
+ buffer_free(w->response.header);
+ buffer_free(w->response.data);
if(w->ifd != -1) close(w->ifd);
if(w->ofd != -1 && w->ofd != w->ifd) close(w->ofd);
freez(w);
}
cleanup:
- if(dimensions)
- buffer_free(dimensions);
+ buffer_free(dimensions);
return ret;
}
buffer_strcat(w->response.data, ");");
cleanup:
- if(dimensions) buffer_free(dimensions);
+ buffer_free(dimensions);
return ret;
}
}
int validate_stream_api_key(const char *key) {
+ if(appconfig_get_boolean(&stream_config, key, "enabled", 0))
+ return 1;
+
return 0;
}
int web_client_stream_request(RRDHOST *host, struct web_client *w, char *url) {
- info("STREAM request from client '%s:%s', starting as host '%s'", w->client_ip, w->client_port, host->hostname);
+ info("STREAM [%s]:%s: client connection.", w->client_ip, w->client_port);
- char *key = NULL;
+ char *key = NULL, *hostname = NULL, *machine_guid = NULL, *os = NULL;
+ int update_every = default_rrd_update_every;
+ int history = default_rrd_history_entries;
+ RRD_MEMORY_MODE mode = default_rrd_memory_mode;
+ int health_enabled = default_health_enabled;
while(url) {
char *value = mystrsep(&url, "?&");
if(!strcmp(name, "key"))
key = value;
+ else if(!strcmp(name, "hostname"))
+ hostname = value;
+ else if(!strcmp(name, "machine_guid"))
+ machine_guid = value;
+ else if(!strcmp(name, "update_every"))
+ update_every = (int)strtoul(value, NULL, 0);
+ else if(!strcmp(name, "os"))
+ os = value;
}
if(!key || !*key) {
return 401;
}
+ if(!hostname || !*hostname) {
+ error("STREAM [%s]:%s: request without a hostname. Forbidding access.", w->client_ip, w->client_port);
+ buffer_flush(w->response.data);
+ buffer_sprintf(w->response.data, "You need to send a hostname too.");
+ return 400;
+ }
+
+ if(!machine_guid || !*machine_guid) {
+ error("STREAM [%s]:%s: request without a machine GUID. Forbidding access.", w->client_ip, w->client_port);
+ buffer_flush(w->response.data);
+ buffer_sprintf(w->response.data, "You need to send a machine GUID too.");
+ return 400;
+ }
+
if(!validate_stream_api_key(key)) {
error("STREAM [%s]:%s: API key '%s' is not allowed. Forbidding access.", w->client_ip, w->client_port, key);
buffer_flush(w->response.data);
return 401;
}
+ if(!appconfig_get_boolean(&stream_config, machine_guid, "enabled", 1)) {
+ error("STREAM [%s]:%s: machine GUID '%s' is not allowed. Forbidding access.", w->client_ip, w->client_port, machine_guid);
+ buffer_flush(w->response.data);
+ buffer_sprintf(w->response.data, "Your machine guide is not permitted access.");
+ return 404;
+ }
+
+ update_every = (int)appconfig_get_number(&stream_config, machine_guid, "update every", update_every);
+ if(update_every < 0) update_every = 1;
+
+ history = (int)appconfig_get_number(&stream_config, key, "default history", history);
+ history = (int)appconfig_get_number(&stream_config, machine_guid, "history", history);
+ if(history < 5) history = 5;
+
+ mode = rrd_memory_mode_id(appconfig_get(&stream_config, key, "default memory mode", rrd_memory_mode_name(mode)));
+ mode = rrd_memory_mode_id(appconfig_get(&stream_config, machine_guid, "memory mode", rrd_memory_mode_name(mode)));
+
+ health_enabled = appconfig_get_boolean_ondemand(&stream_config, key, "health enabled by default", health_enabled);
+ health_enabled = appconfig_get_boolean_ondemand(&stream_config, machine_guid, "health enabled", health_enabled);
+
+ if(strcmp(machine_guid, "localhost"))
+ host = localhost;
+ else
+ host = rrdhost_find_or_create(hostname, machine_guid, os, update_every, history, mode, health_enabled?1:0);
+
+ info("STREAM request from client '%s:%s' for host '%s' with machine_guid '%s': update every = %d, history = %d, memory mode = %s, health %s",
+ w->client_ip, w->client_port,
+ hostname, machine_guid,
+ update_every,
+ history,
+ rrd_memory_mode_name(mode),
+ (health_enabled == CONFIG_BOOLEAN_NO)?"disabled":((health_enabled == CONFIG_BOOLEAN_YES)?"enabled":"auto")
+ );
+
struct plugind cd = {
.enabled = 1,
.update_every = default_rrd_update_every,
return 500;
}
+ rrdhost_wrlock(host);
+ host->use_counter++;
+ rrdhost_unlock(host);
+
// call the plugins.d processor to receive the metrics
- info("STREAM [%s]:%s: connecting client to plugins.d.", w->client_ip, w->client_port);
+ info("STREAM [%s]:%s: connecting client to plugins.d on host '%s' with machine GUID '%s'.", w->client_ip, w->client_port, host->hostname, host->machine_guid);
size_t count = pluginsd_process(host, &cd, fp, 1);
- error("STREAM [%s]:%s: client disconnected.", w->client_ip, w->client_port);
+ error("STREAM [%s]:%s: client disconnected (host '%s', machine GUID '%s').", w->client_ip, w->client_port, host->hostname, host->machine_guid);
+
+ rrdhost_wrlock(host);
+ host->use_counter--;
+ if(!host->use_counter && health_enabled == CONFIG_BOOLEAN_AUTO)
+ host->health_enabled = 0;
+ rrdhost_unlock(host);
- // close all sockets, to let the socket worker we are done
+ // cleanup
fclose(fp);
w->ifd = -1;
- if(w->ofd != -1 && w->ofd != w->ifd) {
- close(w->ofd);
- w->ofd = -1;
- }
// this will not send anything
// the socket is closed
encoded_url = s = &s[8];
w->mode = WEB_CLIENT_MODE_OPTIONS;
}
+ else if(!strncmp(s, "STREAM ", 7)) {
+ encoded_url = s = &s[7];
+ w->mode = WEB_CLIENT_MODE_STREAM;
+ }
else {
w->wait_receive = 0;
return HTTP_VALIDATION_NOT_SUPPORTED;
hash_graph = 0,
hash_list = 0,
hash_all_json = 0,
- hash_host = 0,
- hash_stream = 0;
+ hash_host = 0;
#ifdef NETDATA_INTERNAL_CHECKS
static uint32_t hash_exit = 0, hash_debug = 0, hash_mirror = 0;
hash_list = simple_hash("list");
hash_all_json = simple_hash("all.json");
hash_host = simple_hash("host");
- hash_stream = simple_hash("stream");
#ifdef NETDATA_INTERNAL_CHECKS
hash_exit = simple_hash("exit");
hash_debug = simple_hash("debug");
debug(D_WEB_CLIENT_ACCESS, "%llu: host switch request ...", w->id);
return web_client_switch_host(host, w, url);
}
- else if(unlikely(hash == hash_stream && strcmp(tok, "stream") == 0)) {
- debug(D_WEB_CLIENT_ACCESS, "%llu: stream request ...", w->id);
- return web_client_stream_request(host, w, url);
- }
else if(unlikely(hash == hash_netdata_conf && strcmp(tok, "netdata.conf") == 0)) {
debug(D_WEB_CLIENT_ACCESS, "%llu: Sending netdata.conf ...", w->id);
w->response.data->contenttype = CT_TEXT_PLAIN;
buffer_flush(w->response.data);
- generate_config(w->response.data, 0);
+ config_generate(w->response.data, 0);
return 200;
}
else if(unlikely(hash == hash_data && strcmp(tok, WEB_PATH_DATA) == 0)) { // "data"
switch(http_request_validate(w)) {
case HTTP_VALIDATION_OK:
- if(unlikely(w->mode == WEB_CLIENT_MODE_OPTIONS)) {
- w->response.data->contenttype = CT_TEXT_PLAIN;
- buffer_flush(w->response.data);
- buffer_strcat(w->response.data, "OK");
- w->response.code = 200;
+ switch(w->mode) {
+ case WEB_CLIENT_MODE_STREAM:
+ w->response.code = web_client_stream_request(localhost, w, w->decoded_url);
+ return;
+
+ case WEB_CLIENT_MODE_OPTIONS:
+ w->response.data->contenttype = CT_TEXT_PLAIN;
+ buffer_flush(w->response.data);
+ buffer_strcat(w->response.data, "OK");
+ w->response.code = 200;
+ break;
+
+ case WEB_CLIENT_MODE_FILECOPY:
+ case WEB_CLIENT_MODE_NORMAL:
+ w->response.code = web_client_process_url(localhost, w, w->decoded_url);
+ break;
}
- else
- w->response.code = web_client_process_url(localhost, w, w->decoded_url);
break;
case HTTP_VALIDATION_INCOMPLETE:
else w->wait_send = 0;
switch(w->mode) {
+ case WEB_CLIENT_MODE_STREAM:
+ debug(D_WEB_CLIENT, "%llu: STREAM done.", w->id);
+ break;
+
case WEB_CLIENT_MODE_OPTIONS:
debug(D_WEB_CLIENT, "%llu: Done preparing the OPTIONS response. Sending data (%zu bytes) to client.", w->id, w->response.data->len);
break;
break;
default:
- fatal("%llu: Unknown client mode %d.", w->id, w->mode);
+ fatal("%llu: Unknown client mode %u.", w->id, w->mode);
break;
}
}
// if the sockets are closed, may have transferred this client
// to plugins.d
- if(w->ifd == -1 && w->ofd == -1) break;
+ if(unlikely(w->mode == WEB_CLIENT_MODE_STREAM))
+ break;
}
}
projects / netdata.git / blobdiff