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(&stream_config, key, "enabled", 0))
+ 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) {
- char *key = NULL, *hostname = NULL, *machine_guid = NULL;
+ info("STREAM [%s]:%s: client connection.", w->client_ip, w->client_port);
+
+ 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;
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 404;
}
- // update_every = (int)appconfig_get_number(&stream_config, key, "default update every", update_every);
update_every = (int)appconfig_get_number(&stream_config, machine_guid, "update every", update_every);
if(update_every < 0) update_every = 1;
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);
- host = rrdhost_find_or_create(hostname, machine_guid, update_every, history, mode, health_enabled?1:0);
+ 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,
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 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 (host '%s', machine GUID '%s').", w->client_ip, w->client_port, host->hostname, host->machine_guid);
- if(health_enabled == CONFIG_BOOLEAN_AUTO)
+
+ 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;
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