Merge pull request #606 from getcwd/master

plugins/alarm_curl ehancement

core/alarm.c

@@ -363,6 +363,19 @@ void uwsgi_alarm_thread_start() {
 	}
 }
 
+void uwsgi_alarm_trigger_uai(struct uwsgi_alarm_instance *uai, char *msg, size_t len) {
+	struct iovec iov[2];
+	iov[0].iov_base = &uai;
+	iov[0].iov_len = sizeof(long);
+	iov[1].iov_base = msg;
+	iov[1].iov_len = len;
+
+	// now send the message to the alarm thread
+	if (writev(uwsgi.alarm_thread->pipe[0], iov, 2) != (ssize_t) (len+sizeof(long))) {
+		uwsgi_error("[uwsgi-alarm-error] uwsgi_alarm_trigger()/writev()");
+	}
+}
+
 #ifdef UWSGI_PCRE
 // check if a log should raise an alarm
 void uwsgi_alarm_log_check(char *msg, size_t len) {
@@ -372,7 +385,14 @@ void uwsgi_alarm_log_check(char *msg, size_t len) {
 	while (ual) {
 		if (uwsgi_regexp_match(ual->pattern, ual->pattern_extra, msg, len) >= 0) {
 			if (!ual->negate) {
-				uwsgi_alarm_log_run(ual, msg, len);
+				struct uwsgi_alarm_ll *uall = ual->alarms;
+				while (uall) {
+					if (uwsgi.alarm_cheap)
+						uwsgi_alarm_trigger_uai(uall->alarm, msg, len);
+					else
+						uwsgi_alarm_run(uall->alarm, msg, len);
+					uall = uall->next;
+				}
 			}
 			else {
 				break;
@@ -397,18 +417,6 @@ void uwsgi_alarm_run(struct uwsgi_alarm_instance *uai, char *msg, size_t len) {
 	uai->last_msg_size = len;
 }
 
-#ifdef UWSGI_PCRE
-// call the alarms mapped to a log line
-void uwsgi_alarm_log_run(struct uwsgi_alarm_log *ual, char *msg, size_t len) {
-	struct uwsgi_alarm_ll *uall = ual->alarms;
-	while (uall) {
-		uwsgi_alarm_run(uall->alarm, msg, len);
-		uall = uall->next;
-	}
-}
-#endif
-
-
 // this is the api function workers,mules and whatever you want can call from code
 void uwsgi_alarm_trigger(char *alarm_instance_name, char *msg, size_t len) {
 	if (!uwsgi.alarm_thread) return;
@@ -416,16 +424,7 @@ void uwsgi_alarm_trigger(char *alarm_instance_name, char *msg, size_t len) {
 	struct uwsgi_alarm_instance *uai = uwsgi_alarm_get_instance(alarm_instance_name);
 	if (!uai) return;
 
-	struct iovec iov[2];
-	iov[0].iov_base = &uai;
-	iov[0].iov_len = sizeof(long);
-	iov[1].iov_base = msg;
-	iov[1].iov_len = len;
-
-	// now send the message to the alarm thread
-	if (writev(uwsgi.alarm_thread->pipe[0], iov, 2) != (ssize_t) (len+sizeof(long))) {
-		uwsgi_error("[uwsgi-alarm-error] uwsgi_alarm_trigger()/writev()");
-	}
+	uwsgi_alarm_trigger_uai(uai, msg, len);
 }
 
 struct uwsgi_alarm_fd *uwsgi_add_alarm_fd(int fd, char *alarm, size_t buf_len, char *msg, size_t msg_len) {

core/utils.c

@@ -3752,7 +3752,7 @@ static void *uwsgi_thread_run(void *arg) {
 	return NULL;
 }
 
-struct uwsgi_thread *uwsgi_thread_new(void (*func) (struct uwsgi_thread *)) {
+struct uwsgi_thread *uwsgi_thread_new_with_data(void (*func) (struct uwsgi_thread *), void *data) {
 
 	struct uwsgi_thread *ut = uwsgi_calloc(sizeof(struct uwsgi_thread));
 
@@ -3769,6 +3769,7 @@ struct uwsgi_thread *uwsgi_thread_new(void (*func) (struct uwsgi_thread *)) {
 	uwsgi_socket_nb(ut->pipe[1]);
 
 	ut->func = func;
+	ut->data = data;
 
 	pthread_attr_init(&ut->tattr);
 	pthread_attr_setdetachstate(&ut->tattr, PTHREAD_CREATE_DETACHED);
@@ -3788,6 +3789,10 @@ error:
 	return NULL;
 }
 
+struct uwsgi_thread *uwsgi_thread_new(void (*func) (struct uwsgi_thread *)) {
+	return uwsgi_thread_new_with_data(func, NULL);
+}
+
 int uwsgi_kvlist_parse(char *src, size_t len, char list_separator, char kv_separator, ...) {
 	size_t i;
 	va_list ap;

core/uwsgi.c

@@ -673,6 +673,7 @@ static struct uwsgi_option uwsgi_base_options[] = {
 	{"use-abort", no_argument, 0, "call abort() on segfault/fpe, could be useful for generating a core dump", uwsgi_opt_true, &uwsgi.use_abort, 0},
 
 	{"alarm", required_argument, 0, "create a new alarm, syntax: <alarm> <plugin:args>", uwsgi_opt_add_string_list, &uwsgi.alarm_list, UWSGI_OPT_MASTER},
+	{"alarm-cheap", required_argument, 0, "use main alarm thread rather than create dedicated threads for curl-based alarms", uwsgi_opt_true, &uwsgi.alarm_cheap, 0},
 	{"alarm-freq", required_argument, 0, "tune the anti-loop alam system (default 3 seconds)", uwsgi_opt_set_int, &uwsgi.alarm_freq, 0},
 	{"alarm-fd", required_argument, 0, "raise the specified alarm when an fd is read for read (by default it reads 1 byte, set 8 for eventfd)", uwsgi_opt_add_string_list, &uwsgi.alarm_fd_list, UWSGI_OPT_MASTER},
 	{"alarm-segfault", required_argument, 0, "raise the specified alarm when the segmentation fault handler is executed", uwsgi_opt_add_string_list, &uwsgi.alarm_segfault, UWSGI_OPT_MASTER},

plugins/airbrake/airbrake_plugin.c

@@ -271,12 +271,11 @@ static void uwsgi_airbrake_loop(struct uwsgi_thread *ut) {
 }
 
 static void uwsgi_airbrake_init(struct uwsgi_alarm_instance *uai) {
-	struct uwsgi_thread *ut = uwsgi_thread_new(uwsgi_airbrake_loop);
-	if (!ut) return;
-	uai->data_ptr = ut;
 	struct uwsgi_airbrake_config *uacc = uwsgi_calloc(sizeof(struct uwsgi_airbrake_config));
 	uacc->arg = uai->arg;
-	ut->data = uacc;
+	struct uwsgi_thread *ut = uwsgi_thread_new_with_data(uwsgi_airbrake_loop, uacc);
+	if (!ut) return;
+	uai->data_ptr = ut;
 }
 
 static void uwsgi_airbrake_func(struct uwsgi_alarm_instance *uai, char *msg, size_t len) {

plugins/alarm_curl/alarm_curl_plugin.c

@@ -3,9 +3,18 @@
 
 extern struct uwsgi_server uwsgi;
 
+struct uwsgi_alarm_curl {
+	CURL	*curl;
+	struct uwsgi_thread *ut;
+	int	pos;
+	int	blen;
+	char	*body;
+	int	hlen;
+	char	hdr[];
+};
+
 struct uwsgi_alarm_curl_config {
-	int first;
-	char *arg;
+	char *url;
 	char *subject;
 	char *to;
 };
@@ -17,7 +26,7 @@ struct uwsgi_alarm_curl_opt {
 };
 
 
-#ifdef CURLOPT_MAIL_RCPT
+#ifdef CURLPROTO_SMTP
 static void uwsgi_alarm_curl_to(CURL *curl, CURLoption option, char *arg, struct uwsgi_alarm_curl_config *uacc) {
 	uacc->to = arg;
 	struct curl_slist *list = NULL;
@@ -29,6 +38,7 @@ static void uwsgi_alarm_curl_to(CURL *curl, CURLoption option, char *arg, struct
 		p = strtok_r(NULL, ",", &ctx);
 	}
 	curl_easy_setopt(curl, option, list);
+	free(items);
 }
 #endif
 
@@ -44,16 +54,33 @@ static void uwsgi_alarm_curl_set_subject(CURL *curl, CURLoption option, char *ar
 	uacc->subject = arg;	
 }
 
-static struct uwsgi_alarm_curl_opt uaco[] = {
-	{"url", CURLOPT_URL, NULL},
-#ifdef CURLOPT_MAIL_RCPT
-	{"mail_to", CURLOPT_MAIL_RCPT, uwsgi_alarm_curl_to },
+static void uwsgi_alarm_curl_url(CURL *curl, CURLoption option, char *arg, struct uwsgi_alarm_curl_config *uacc)
+{
+#ifndef CURLPROTO_SMTP
+	if (!uwsgi_strnicmp(arg, 4, "smtp", 4)) {
+		uwsgi_error("Please update libcurl to use SMTP protocol.\n");
+		exit(1);
+	}
 #endif
-#ifdef CURLOPT_MAIL_FROM
+	uacc->url = arg;
+	curl_easy_setopt(curl, option, arg);
+}
+
+static void uwsgi_alarm_curl_ssl_insecure(CURL *curl, CURLoption option, char *arg, struct uwsgi_alarm_curl_config *uacc)
+{
+	curl_easy_setopt(curl, CURLOPT_SSL_VERIFYPEER, 0L);
+	curl_easy_setopt(curl, CURLOPT_SSL_VERIFYHOST, 0L);
+}
+
+static struct uwsgi_alarm_curl_opt uaco[] = {
+	{"url", CURLOPT_URL, uwsgi_alarm_curl_url},
+#ifdef CURLPROTO_SMTP
+	{"mail_to", CURLOPT_MAIL_RCPT, uwsgi_alarm_curl_to},
 	{"mail_from", CURLOPT_MAIL_FROM, NULL},
 #endif
 	{"subject", 0, uwsgi_alarm_curl_set_subject},
 	{"ssl", CURLOPT_USE_SSL, uwsgi_alarm_curl_ssl},
+	{"ssl_insecure", 0, uwsgi_alarm_curl_ssl_insecure},
 	{"auth_user", CURLOPT_USERNAME, NULL},
 	{"auth_pass", CURLOPT_PASSWORD, NULL},
 	{"method", CURLOPT_CUSTOMREQUEST, NULL},
@@ -65,14 +92,10 @@ static struct uwsgi_alarm_curl_opt uaco[] = {
 static void uwsgi_alarm_curl_setopt(CURL *curl, char *opt, struct uwsgi_alarm_curl_config *uacc) {
 	struct uwsgi_alarm_curl_opt *o = uaco;
 	char *equal = strchr(opt,'=');
-	if (!equal) {
-		if (!uacc->first) {
-			curl_easy_setopt(curl, CURLOPT_URL, opt);
-			uacc->first = 1;
-		}
+	if (!equal && !uacc->url) {
+		uwsgi_alarm_curl_url(curl, CURLOPT_URL, opt, uacc);
 		return;
 	}
-	uacc->first = 1;
 	*equal = 0;
 	while(o->name) {
 		if (!strcmp(o->name, opt)) {
@@ -82,91 +105,137 @@ static void uwsgi_alarm_curl_setopt(CURL *curl, char *opt, struct uwsgi_alarm_cu
 			else {
 				curl_easy_setopt(curl, o->option, equal+1);
 			}
-			goto end;
+			break;
 		}
 		o++;
 	}
-end:
-	*equal = '=';
 }
 
 static size_t uwsgi_alarm_curl_read_callback(void *ptr, size_t size, size_t nmemb, void *userp) {
-	struct uwsgi_thread *ut = (struct uwsgi_thread *) userp;
+	struct uwsgi_alarm_curl *uac = userp;
 	size_t full_size = size * nmemb;
-	size_t remains = ut->len - ut->pos;
-	struct uwsgi_alarm_curl_config *uacc = ut->data;
+	int remains = full_size;
 
-	if (remains == 0) return 0;
-
-	if (ut->custom0 == 0) {
-		size_t newline = 0;
-		size_t required = 1;
-		char *addr = ptr;
-		if (uacc->to) required += 4 + strlen(uacc->to) + 1;
-		if (uacc->subject) required += 9 + strlen(uacc->subject) + 1;
-		if (required > full_size) goto skip;
-
-
-		if (uacc->to) {
-			memcpy(addr, "To: ", 4); addr+=4;
-			memcpy(addr, uacc->to, strlen(uacc->to)); addr += strlen(uacc->to);
-			*addr ++= '\n';
-			newline = 1;
+	if (uac->pos < uac->hlen) {
+		if (remains > uac->hlen - uac->pos) {
+			memcpy(ptr, uac->hdr + uac->pos, uac->hlen - uac->pos);
+			ptr += uac->hlen - uac->pos;
+			remains -= uac->hlen - uac->pos;
+			uac->pos = uac->hlen;
+		} else {
+			memcpy(ptr, uac->hdr + uac->pos, remains);
+			uac->pos += remains;
+			return full_size;
 		}
-
-		if (uacc->subject) {
-			memcpy(addr, "Subject: ", 9); addr+=9;
-			memcpy(addr, uacc->subject, strlen(uacc->subject)); addr += strlen(uacc->subject);
-			*addr ++= '\n';
-			newline = 1;
-		}
-skip:
-		if (newline > 0) {
-			*addr = '\n';
-		}
-		ut->custom0 = 1;
-		return required;
 	}
 
-	if (full_size < remains) {
-		remains = full_size;
+	if (remains > uac->blen + uac->hlen - uac->pos) {
+		memcpy(ptr, uac->body + uac->pos - uac->hlen, uac->blen + uac->hlen - uac->pos);
+		remains -= uac->blen + uac->hlen - uac->pos;
+		uac->pos = uac->blen + uac->hlen;
+		return full_size - remains;
 	}
 
-	memcpy(ptr, ut->buf + ut->pos, remains);
-	ut->pos += remains;
-
-	return remains;	
+	memcpy(ptr, uac->body + uac->pos - uac->hlen, remains);
+	uac->pos += remains;
+	return full_size;
 }
 
-static void uwsgi_alarm_curl_loop(struct uwsgi_thread *ut) {
-	int interesting_fd;
-	ut->buf = uwsgi_malloc(uwsgi.log_master_bufsize);
+static struct uwsgi_alarm_curl *uwsgi_alarm_curl_alloc(struct uwsgi_alarm_curl_config *uacc)
+{
+	char *addr;
+	struct uwsgi_alarm_curl *uac;
+	size_t required = 0;
 
+	if (uacc->to) required += 4 + strlen(uacc->to) + 2;
+	if (uacc->subject) required += 9 + strlen(uacc->subject) + 2;
+	if (required)
+		required += 2;	/* newline between MIME header and body */
+
+	uac = uwsgi_malloc(sizeof(*uac) + required);
+	uac->hlen = required;
+	addr = uac->hdr;
+
+	if (uacc->to) {
+		memcpy(addr, "To: ", 4); addr += 4;
+		memcpy(addr, uacc->to, strlen(uacc->to)); addr += strlen(uacc->to);
+		*addr++ = '\r';
+		*addr++ = '\n';
+	}
+
+	if (uacc->subject) {
+		memcpy(addr, "Subject: ", 9); addr += 9;
+		memcpy(addr, uacc->subject, strlen(uacc->subject)); addr += strlen(uacc->subject);
+		*addr++ = '\r';
+		*addr++ = '\n';
+	}
+
+	if (required) {
+		*addr++ = '\r';
+		*addr = '\n';
+	}
+
+	return uac;
+}
+
+static struct uwsgi_alarm_curl *uwsgi_alarm_curl_init_curl(struct uwsgi_alarm_instance *uai) {
 	CURL *curl = curl_easy_init();
-	// ARGH !!!
-	if (!curl) return;
+	if (!curl) {
+		uwsgi_error("Failed to initialize libcurl.\n");
+		exit(1);
+	}
 
 	curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, uwsgi.socket_timeout);
 	curl_easy_setopt(curl, CURLOPT_TIMEOUT, uwsgi.socket_timeout);
 	curl_easy_setopt(curl, CURLOPT_READFUNCTION, uwsgi_alarm_curl_read_callback);
-	curl_easy_setopt(curl, CURLOPT_READDATA, ut);
 	curl_easy_setopt(curl, CURLOPT_UPLOAD, 1L);
 	curl_easy_setopt(curl, CURLOPT_POST, 1L);
 	struct curl_slist *expect = NULL; expect = curl_slist_append(expect, "Expect:");
 	curl_easy_setopt(curl, CURLOPT_HTTPHEADER, expect);
 	curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1);
 
-	struct uwsgi_alarm_curl_config *uacc = (struct uwsgi_alarm_curl_config *) ut->data;
-	char *opts = uwsgi_str(uacc->arg);
+	struct uwsgi_alarm_curl_config uacc;
+	memset(&uacc, 0, sizeof(uacc));
+	char *opts = uwsgi_str(uai->arg);
 
 	// fill curl options
 	char *ctx = NULL;
 	char *p = strtok_r(opts, ";", &ctx);
 	while(p) {
-		uwsgi_alarm_curl_setopt(curl, uwsgi_str(p), uacc);
+		uwsgi_alarm_curl_setopt(curl, p, &uacc);
 		p = strtok_r(NULL, ";", &ctx);
 	}
 
+	if (!uacc.url) {
+		uwsgi_error("An URL is required to trigger curl-based alarm.\n");
+		exit(1);
+	}
+
+	struct uwsgi_alarm_curl *uac = uwsgi_alarm_curl_alloc(&uacc);
+	curl_easy_setopt(curl, CURLOPT_READDATA, uac);
+	free(opts);
+	uac->curl = curl;
+	uai->data_ptr = uac;
+
+	return uac;
+}
+
+static void uwsgi_alarm_curl_call_curl(struct uwsgi_alarm_curl *uac, char *msg, int len)
+{
+	uac->pos = 0;
+	uac->body = msg;
+	uac->blen = len;
+	curl_easy_setopt(uac->curl, CURLOPT_INFILESIZE, uac->hlen + uac->blen);
+	CURLcode res = curl_easy_perform(uac->curl);
+	if (res != CURLE_OK)
+		uwsgi_log_alarm("-curl] curl_easy_perform() failed: %s\n", curl_easy_strerror(res));
+}
+
+static void uwsgi_alarm_curl_loop(struct uwsgi_thread *ut) {
+	int interesting_fd;
+	struct uwsgi_alarm_curl *uac = uwsgi_alarm_curl_init_curl(ut->data);
+	uac->ut = ut;
+	ut->buf = uwsgi_malloc(uwsgi.log_master_bufsize);
 	for(;;) {
 		int ret = event_queue_wait(ut->queue, -1, &interesting_fd);	
 		if (ret < 0) return;
@@ -174,31 +243,26 @@ static void uwsgi_alarm_curl_loop(struct uwsgi_thread *ut) {
 		if (interesting_fd != ut->pipe[1]) continue;
 		ssize_t rlen = read(ut->pipe[1], ut->buf, uwsgi.log_master_bufsize);
 		if (rlen <= 0) continue;
-		ut->pos = 0;
-		ut->len = (size_t) rlen;
-		ut->custom0 = 0;
-		curl_easy_setopt(curl, CURLOPT_INFILESIZE_LARGE, (curl_off_t) ut->len);
-		CURLcode res = curl_easy_perform(curl);
-		if (res != CURLE_OK) {
-			uwsgi_log_alarm("-curl] curl_easy_perform() failed: %s\n", curl_easy_strerror(res));
-		}
-		
+		uwsgi_alarm_curl_call_curl(uac, ut->buf, rlen);
 	}
 }
 
 static void uwsgi_alarm_curl_init(struct uwsgi_alarm_instance *uai) {
-	struct uwsgi_thread *ut = uwsgi_thread_new(uwsgi_alarm_curl_loop);
-	if (!ut) return;
-	uai->data_ptr = ut;
-	struct uwsgi_alarm_curl_config *uacc = uwsgi_calloc(sizeof(struct uwsgi_alarm_curl_config));
-	uacc->arg = uai->arg;
-	ut->data = uacc;
+	if (uwsgi.alarm_cheap)
+		uwsgi_alarm_curl_init_curl(uai);
+	else
+		uwsgi_thread_new_with_data(uwsgi_alarm_curl_loop, uai);
 }
 
 // pipe the message into the thread;
 static void uwsgi_alarm_curl_func(struct uwsgi_alarm_instance *uai, char *msg, size_t len) {
-	struct uwsgi_thread *ut = (struct uwsgi_thread *) uai->data_ptr;
-	ut->rlen = write(ut->pipe[0], msg, len);
+	struct uwsgi_alarm_curl *uac = uai->data_ptr;
+	if (uwsgi.alarm_cheap)
+		uwsgi_alarm_curl_call_curl(uac, msg, len);
+	else {
+		struct uwsgi_thread *ut = uac->ut;
+		ut->rlen = write(ut->pipe[0], msg, len);
+	}
 }
 
 static void uwsgi_alarm_curl_load(void) {

plugins/alarm_xmpp/alarm_xmpp_plugin.c

@@ -4,10 +4,9 @@ void uwsgi_alarm_xmpp_loop(struct uwsgi_thread *);
 
 static void uwsgi_alarm_xmpp_init(struct uwsgi_alarm_instance *uai) {
 
-	struct uwsgi_thread *ut = uwsgi_thread_new(uwsgi_alarm_xmpp_loop);
+	struct uwsgi_thread *ut = uwsgi_thread_new_with_data(uwsgi_alarm_xmpp_loop, uai->arg);
 	if (!ut) return;
 	uai->data_ptr = ut;
-	ut->data = uai->arg;
 }
 
 // pipe the message into the thread;

uwsgi.h

@@ -2708,6 +2708,7 @@ struct uwsgi_server {
 	char *subscription_notify_socket;
 
 	int mule_reload_mercy;
+	int alarm_cheap;
 };
 
 struct uwsgi_rpc {
@@ -4149,7 +4150,6 @@ char *uwsgi_elf_section(char *, char *, size_t *);
 
 void uwsgi_alarm_log_check(char *, size_t);
 void uwsgi_alarm_run(struct uwsgi_alarm_instance *, char *, size_t);
-void uwsgi_alarm_log_run(struct uwsgi_alarm_log *, char *, size_t);
 void uwsgi_register_alarm(char *, void (*)(struct uwsgi_alarm_instance *), void (*)(struct uwsgi_alarm_instance *, char *, size_t));
 void uwsgi_register_embedded_alarms();
 void uwsgi_alarms_init();
@@ -4175,6 +4175,7 @@ struct uwsgi_thread {
 	void (*func) (struct uwsgi_thread *);
 };
 struct uwsgi_thread *uwsgi_thread_new(void (*)(struct uwsgi_thread *));
+struct uwsgi_thread *uwsgi_thread_new_with_data(void (*)(struct uwsgi_thread *), void *data);
 
 struct uwsgi_offload_request {
 	// the request socket