clearlinux/uwsgi
2
0
Fork 0
mirror of https://github.com/clearlinux/uwsgi.git synced 2026-09-08 14:41:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

better cluster node management

Browse Source
This commit is contained in:
roberto@sirius
2011-01-01 11:53:06 +01:00
parent a43bc93492
commit 31892a69df
4 changed files with 143 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
master.c
+20 -13
View File
@@ -19,23 +19,31 @@ void get_linux_tcp_info(int fd) {
}
#endif
char *new_cluster_hostname;
char *new_cluster_address;
char *new_cluster_workers;
void print_dict(char *key, uint16_t keylen, char *val, uint16_t vallen, void *data) {
void manage_cluster_announce(char *key, uint16_t keylen, char *val, uint16_t vallen, void *data) {
char *tmpstr;
struct uwsgi_cluster_node *ucn = (struct uwsgi_cluster_node *) data;
uwsgi_log("%.*s = %.*s\n", keylen, key, vallen, val);
if (!uwsgi_strncmp("hostname", 8, key, keylen)) {
new_cluster_hostname = uwsgi_concat2n(val, vallen, "", 0);
strncpy(ucn->nodename, val, UMIN(vallen, 255));
}
if (!uwsgi_strncmp("address", 7, key, keylen)) {
new_cluster_address = uwsgi_concat2n(val, vallen, "", 0);
strncpy(ucn->name, val, UMIN(vallen, 100));
}
if (!uwsgi_strncmp("workers", 7, key, keylen)) {
new_cluster_workers = uwsgi_concat2n(val, vallen, "", 0);
tmpstr = uwsgi_concat2n(val, vallen, "", 0);
ucn->workers = atoi(tmpstr);
free(tmpstr);
}
if (!uwsgi_strncmp("requests", 8, key, keylen)) {
tmpstr = uwsgi_concat2n(val, vallen, "", 0);
ucn->requests = strtoul(tmpstr, NULL, 0);
free(tmpstr);
}
}
@@ -78,6 +86,7 @@ void master_loop(char **argv, char **environ) {
char *cptrbuf;
uint16_t ustrlen;
struct uwsgi_header *uh;
struct uwsgi_cluster_node nucn;
#endif
#endif
@@ -383,12 +392,10 @@ void master_loop(char **argv, char **environ) {
switch(uwsgi.wsgi_requests[0]->uh.modifier1) {
case 95:
new_cluster_hostname = NULL;
new_cluster_address = NULL;
new_cluster_workers = NULL;
uwsgi_hooked_parse(uwsgi.wsgi_requests[0]->buffer, uwsgi.wsgi_requests[0]->uh.pktsize, print_dict, NULL);
if (new_cluster_hostname && new_cluster_address && new_cluster_workers) {
uwsgi_cluster_add_node(new_cluster_address, atoi(new_cluster_workers), CLUSTER_NODE_DYNAMIC);
memset(&nucn, 0, sizeof(struct uwsgi_cluster_node));
uwsgi_hooked_parse(uwsgi.wsgi_requests[0]->buffer, uwsgi.wsgi_requests[0]->uh.pktsize, manage_cluster_announce, &nucn);
if (nucn.name[0] != 0) {
uwsgi_cluster_add_node(&nucn, CLUSTER_NODE_DYNAMIC);
}
break;
case 96:
plugins/python/uwsgi_pymodule.c
+39
View File
@@ -1845,6 +1845,43 @@ PyObject *py_uwsgi_suspend(PyObject * self, PyObject * args) {
}
#ifdef UWSGI_MULTICAST
PyObject *py_uwsgi_cluster(PyObject * self, PyObject * args) {
if (uwsgi.cluster) {
return PyString_FromString(uwsgi.cluster);
}
Py_INCREF(Py_None);
return Py_None;
}
PyObject *py_uwsgi_cluster_node_name(PyObject * self, PyObject * args) {
struct uwsgi_cluster_node *ucn;
int i;
char *node = NULL;
if (!PyArg_ParseTuple(args, "s:cluster_node_name", &node)) {
return NULL;
}
if (node == NULL) {
return PyString_FromString(uwsgi.hostname);
}
for (i = 0; i < MAX_CLUSTER_NODES; i++) {
ucn = &uwsgi.shared->nodes[i];
if (ucn->name[0] != 0) {
if (!strcmp(ucn->name, node)) {
return PyString_FromString(ucn->nodename);
}
}
}
Py_INCREF(Py_None);
return Py_None;
}
PyObject *py_uwsgi_cluster_nodes(PyObject * self, PyObject * args) {
struct uwsgi_cluster_node *ucn;
@@ -1919,6 +1956,8 @@ static PyMethodDef uwsgi_advanced_methods[] = {
#ifdef UWSGI_MULTICAST
{"send_multicast_message", py_uwsgi_multicast, METH_VARARGS, ""},
{"cluster_nodes", py_uwsgi_cluster_nodes, METH_VARARGS, ""},
{"cluster_node_name", py_uwsgi_cluster_node_name, METH_VARARGS, ""},
{"cluster", py_uwsgi_cluster, METH_VARARGS, ""},
{"cluster_best_node", py_uwsgi_cluster_best_node, METH_VARARGS, ""},
#endif
#ifdef UWSGI_ASYNC
uwsgi.c
+78 -4
View File
@@ -620,7 +620,7 @@ options_parsed:
#ifdef UWSGI_PROXY
if (uwsgi.proxy_add_me) {
uwsgi_cluster_add_node(uwsgi.sockets[0].name, 1, CLUSTER_NODE_STATIC);
uwsgi_cluster_simple_add_node(uwsgi.sockets[0].name, 1, CLUSTER_NODE_STATIC);
}
#endif
@@ -1674,7 +1674,7 @@ end:
uwsgi.proxy_add_me = 1;
}
else {
uwsgi_cluster_add_node(optarg, 1, CLUSTER_NODE_STATIC);
uwsgi_cluster_simple_add_node(optarg, 1, CLUSTER_NODE_STATIC);
}
return 1;
case LONG_ARGS_PROXY:
@@ -2045,7 +2045,7 @@ end:
}
void uwsgi_cluster_add_node(char *nodename, int workers, int type) {
void uwsgi_cluster_simple_add_node(char *nodename, int workers, int type) {
int i;
struct uwsgi_cluster_node *ucn;
@@ -2107,6 +2107,66 @@ void uwsgi_cluster_add_node(char *nodename, int workers, int type) {
uwsgi_log("unable to add node %s\n", nodename);
}
void uwsgi_cluster_add_node(struct uwsgi_cluster_node *nucn, int type) {
int i;
struct uwsgi_cluster_node *ucn;
char *tcp_port;
uwsgi_log("adding node\n");
tcp_port = strchr(nucn->name, ':');
if (tcp_port == NULL) {
fprintf(stdout, "invalid cluster node name %s\n", nucn->name);
return;
}
// first check for already present node
for (i = 0; i < MAX_CLUSTER_NODES; i++) {
ucn = &uwsgi.shared->nodes[i];
if (ucn->name[0] != 0) {
if (!strcmp(ucn->name, nucn->name)) {
ucn->status = UWSGI_NODE_OK;
ucn->last_seen = time(NULL);
// update requests
ucn->requests = nucn->requests;
return;
}
}
}
for (i = 0; i < MAX_CLUSTER_NODES; i++) {
ucn = &uwsgi.shared->nodes[i];
if (ucn->name[0] == 0) {
memcpy(ucn->name, nucn->name, strlen(nucn->name) + 1);
ucn->workers = nucn->workers;
ucn->ucn_addr.sin_family = AF_INET;
ucn->ucn_addr.sin_port = htons(atoi(tcp_port + 1));
tcp_port[0] = 0;
if (nucn->name[0] == 0) {
ucn->ucn_addr.sin_addr.s_addr = INADDR_ANY;
} else {
uwsgi_log("%s\n", nucn->name);
ucn->ucn_addr.sin_addr.s_addr = inet_addr(nucn->name);
}
ucn->type = type;
// here memory can be freed, as it is allocated by uwsgi_concat2n
if (type != CLUSTER_NODE_DYNAMIC) {
tcp_port[0] = ':';
}
ucn->last_seen = time(NULL);
ucn->requests = nucn->requests;
uwsgi_log("[uWSGI cluster] added node %s\n", ucn->name);
return;
}
}
uwsgi_log("unable to add node %s\n", nucn->name);
}
void build_options() {
int i;
@@ -2254,13 +2314,15 @@ int uwsgi_cluster_add_me() {
const char *key1 = "hostname";
const char *key2 = "address";
const char *key3 = "workers";
const char *key4 = "requests";
char *ptrbuf ;
uint16_t ustrlen;
char numproc[6];
snprintf(numproc, 6, "%d", uwsgi.numproc);
size_t len = 2 + strlen(key1) + 2 + strlen(uwsgi.hostname) + 2 + strlen(key2) + 2 + strlen(uwsgi.sockets[0].name) + 2 + strlen(key3) + 2 + strlen(numproc);
size_t len = 2 + strlen(key1) + 2 + strlen(uwsgi.hostname) + 2 + strlen(key2) + 2 + strlen(uwsgi.sockets[0].name) + 2 + strlen(key3) + 2 + strlen(numproc) + 2 + strlen(key4) + 2 + 1;
char *buf = uwsgi_malloc( len );
ptrbuf = buf;
@@ -2296,11 +2358,23 @@ int uwsgi_cluster_add_me() {
*ptrbuf++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ptrbuf, numproc, strlen(numproc)); ptrbuf+=strlen(numproc);
ustrlen = strlen(key4);
*ptrbuf++ = (uint8_t) (ustrlen & 0xff);
*ptrbuf++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ptrbuf, key4, strlen(key4)); ptrbuf+=strlen(key4);
ustrlen = 1;
*ptrbuf++ = (uint8_t) (ustrlen & 0xff);
*ptrbuf++ = (uint8_t) ((ustrlen >>8) & 0xff);
memcpy(ptrbuf, "0", 1); ptrbuf+=1;
uwsgi_string_sendto(uwsgi.cluster_fd, 95, 0, (struct sockaddr *) &uwsgi.mc_cluster_addr, sizeof(uwsgi.mc_cluster_addr), buf, len);
free(buf);
uwsgi_log("add_me() successfull\n");
return 0;
}
uwsgi.h
+6 -1
View File
@@ -907,6 +907,8 @@ struct uwsgi_lb_node {
struct uwsgi_cluster_node {
char name[101];
char nodename[0xff];
struct sockaddr_in ucn_addr;
int type;
@@ -919,6 +921,8 @@ struct uwsgi_cluster_node {
int errors;
time_t last_choosen;
int requests;
};
@@ -1102,7 +1106,8 @@ void uwsgi_proxy(int);
pid_t proxy_start(int);
#endif
void uwsgi_cluster_add_node(char *, int, int);
void uwsgi_cluster_add_node(struct uwsgi_cluster_node *, int);
void uwsgi_cluster_simple_add_node(char *, int, int);
int uwsgi_ping_node(int, struct wsgi_request *);
struct http_status_codes {