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;
}
return 200;
}
+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 [%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;
+ int health_enabled = default_health_enabled;
+
+ while(url) {
+ char *value = mystrsep(&url, "?&");
+ if(!value || !*value) continue;
+
+ char *name = mystrsep(&value, "=");
+ if(!name || !*name) continue;
+ if(!value || !*value) continue;
+
+ 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) {
+ error("STREAM [%s]:%s: request without an API key. Forbidding access.", w->client_ip, w->client_port);
+ buffer_flush(w->response.data);
+ buffer_sprintf(w->response.data, "You need an API key for this request.");
+ 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);
+ buffer_sprintf(w->response.data, "Your API key is not permitted access.");
+ 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,
+ .pid = 0,
+ .serial_failures = 0,
+ .successful_collections = 0,
+ .obsolete = 0,
+ .started_t = now_realtime_sec(),
+ .next = NULL,
+ };
+
+ // put the client IP and port into the buffers used by plugins.d
+ snprintfz(cd.id, CONFIG_MAX_NAME, "%s:%s", w->client_ip, w->client_port);
+ snprintfz(cd.filename, FILENAME_MAX, "%s:%s", w->client_ip, w->client_port);
+ snprintfz(cd.fullfilename, FILENAME_MAX, "%s:%s", w->client_ip, w->client_port);
+ snprintfz(cd.cmd, PLUGINSD_CMD_MAX, "%s:%s", w->client_ip, w->client_port);
+
+ info("STREAM [%s]:%s: sending STREAM to initiate streaming...", w->client_ip, w->client_port);
+ if(send_timeout(w->ifd, "STREAM", 6, 0, 60) != 6) {
+ error("STREAM [%s]:%s: cannot send STREAM.", w->client_ip, w->client_port);
+ buffer_flush(w->response.data);
+ buffer_sprintf(w->response.data, "Failed to reply back with STREAM");
+ return 400;
+ }
+
+ // remove the non-blocking flag from the socket
+ if(fcntl(w->ifd, F_SETFL, fcntl(w->ifd, F_GETFL, 0) & ~O_NONBLOCK) == -1)
+ error("STREAM [%s]:%s: cannot remove the non-blocking flag from socket %d", w->client_ip, w->client_port, w->ifd);
+
+ /*
+ char buffer[1000 + 1];
+ ssize_t len;
+ while((len = read(w->ifd, buffer, 1000)) != -1) {
+ buffer[len] = '\0';
+ fprintf(stderr, "BEGIN READ %zu bytes\n%s\nEND READ\n", (size_t)len, buffer);
+ }
+ */
+
+ // convert the socket to a FILE *
+ FILE *fp = fdopen(w->ifd, "r");
+ if(!fp) {
+ error("STREAM [%s]:%s: failed to get a FILE for FD %d.", w->client_ip, w->client_port, w->ifd);
+ buffer_flush(w->response.data);
+ buffer_sprintf(w->response.data, "Failed to get a FILE for an FD.");
+ 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);
+
+ rrdhost_wrlock(host);
+ host->use_counter--;
+ if(!host->use_counter && health_enabled == CONFIG_BOOLEAN_AUTO)
+ host->health_enabled = 0;
+ rrdhost_unlock(host);
+
+ // cleanup
+ fclose(fp);
+ w->ifd = -1;
+
+ // this will not send anything
+ // the socket is closed
+ buffer_flush(w->response.data);
+ if(count) return 200;
+ return 400;
+}
+
const char *web_content_type_to_string(uint8_t contenttype) {
switch(contenttype) {
case CT_TEXT_HTML:
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;
static inline int web_client_process_url(RRDHOST *host, struct web_client *w, char *url);
static inline int web_client_switch_host(RRDHOST *host, struct web_client *w, char *url) {
+ static uint32_t hash_localhost = 0;
+
+ if(unlikely(!hash_localhost)) {
+ hash_localhost = simple_hash("localhost");
+ }
+
if(host != localhost) {
buffer_flush(w->response.data);
buffer_strcat(w->response.data, "Nesting of hosts is not allowed.");
if(tok && *tok) {
debug(D_WEB_CLIENT, "%llu: Searching for host with name '%s'.", w->id, tok);
+ // copy the URL, we need it to serve files
+ w->last_url[0] = '/';
+ if(url && *url) strncpyz(&w->last_url[1], url, URL_MAX - 1);
+ else w->last_url[1] = '\0';
+
uint32_t hash = simple_hash(tok);
+
+ if(unlikely(hash == hash_localhost && !strcmp(tok, "localhost")))
+ return web_client_process_url(localhost, w, url);
+
+ rrd_rdlock();
RRDHOST *h;
- for(h = localhost; h; h = h->next) {
+ rrdhost_foreach_read(h) {
if(unlikely((hash == h->hash_hostname && !strcmp(tok, h->hostname)) ||
- (hash == h->hash_machine_guid && !strcmp(tok, h->machine_guid))))
+ (hash == h->hash_machine_guid && !strcmp(tok, h->machine_guid)))) {
+ rrd_unlock();
return web_client_process_url(h, w, url);
+ }
}
+ rrd_unlock();
}
buffer_flush(w->response.data);
- buffer_strcat(w->response.data, "Host is not found: ");
+ buffer_strcat(w->response.data, "This netdata does not maintain a database for host: ");
buffer_strcat_htmlescape(w->response.data, tok?tok:"");
return 404;
}
uint32_t hash = simple_hash(tok);
debug(D_WEB_CLIENT, "%llu: Processing command '%s'.", w->id, tok);
- if(unlikely(hash == hash_host && strcmp(tok, "host") == 0)) {
- debug(D_WEB_CLIENT_ACCESS, "%llu: host switch request ...", w->id);
- return web_client_switch_host(host, w, url);
- }
if(unlikely(hash == hash_api && strcmp(tok, "api") == 0)) {
debug(D_WEB_CLIENT_ACCESS, "%llu: API request ...", w->id);
return web_client_api_request(host, w, url);
}
+ else if(unlikely(hash == hash_host && strcmp(tok, "host") == 0)) {
+ debug(D_WEB_CLIENT_ACCESS, "%llu: host switch request ...", w->id);
+ return web_client_switch_host(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"
debug(D_WEB_CLIENT_ACCESS, "%llu: Sending list of RRD_STATS...", w->id);
buffer_flush(w->response.data);
- RRDSET *st = host->rrdset_root;
+ RRDSET *st;
- for ( ; st ; st = st->next )
- buffer_sprintf(w->response.data, "%s\n", st->name);
+ rrdhost_rdlock(host);
+ rrdset_foreach_read(st, host) buffer_sprintf(w->response.data, "%s\n", st->name);
+ rrdhost_unlock(host);
return 200;
}
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;
}
}
struct web_client *w = ptr;
struct pollfd fds[2], *ifd, *ofd;
- int retval, fdmax = 0, timeout;
+ int retval, timeout;
+ nfds_t fdmax = 0;
log_access("%llu: %s port %s connected on thread task id %d", w->id, w->client_ip, w->client_port, gettid());
for(;;) {
+ if(unlikely(netdata_exit)) break;
+
if(unlikely(w->dead)) {
debug(D_WEB_CLIENT, "%llu: client is dead.", w->id);
break;
timeout = web_client_timeout * 1000;
retval = poll(fds, fdmax, timeout);
+ if(unlikely(netdata_exit)) break;
+
if(unlikely(retval == -1)) {
if(errno == EAGAIN || errno == EINTR) {
debug(D_WEB_CLIENT, "%llu: EAGAIN received.", w->id);
break;
}
+ if(unlikely(netdata_exit)) break;
+
int used = 0;
if(w->wait_send && ofd->revents & POLLOUT) {
used++;
}
}
+ if(unlikely(netdata_exit)) break;
+
if(w->wait_receive && (ifd->revents & POLLIN || ifd->revents & POLLPRI)) {
used++;
if(web_client_receive(w) < 0) {
if(w->mode == WEB_CLIENT_MODE_NORMAL) {
debug(D_WEB_CLIENT, "%llu: Attempting to process received data.", w->id);
web_client_process_request(w);
+
+ // if the sockets are closed, may have transferred this client
+ // to plugins.d
+ if(unlikely(w->mode == WEB_CLIENT_MODE_STREAM))
+ break;
}
}
projects / netdata.git / blobdiff