root/src/stratcon_iep.c

Revision 8bc0f312a7762512a81b3b7c4894866b15be8e07, 23.5 kB (checked in by Theo Schlossnagle <jesus@omniti.com>, 4 years ago)

only allow one postponed iep reconnect at a time

  • Property mode set to 100644
Line 
1 /*
2  * Copyright (c) 2007, OmniTI Computer Consulting, Inc.
3  * All rights reserved.
4  *
5  * Redistribution and use in source and binary forms, with or without
6  * modification, are permitted provided that the following conditions are
7  * met:
8  *
9  *     * Redistributions of source code must retain the above copyright
10  *       notice, this list of conditions and the following disclaimer.
11  *     * Redistributions in binary form must reproduce the above
12  *       copyright notice, this list of conditions and the following
13  *       disclaimer in the documentation and/or other materials provided
14  *       with the distribution.
15  *     * Neither the name OmniTI Computer Consulting, Inc. nor the names
16  *       of its contributors may be used to endorse or promote products
17  *       derived from this software without specific prior written
18  *       permission.
19  *
20  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
21  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
22  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
23  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
24  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
25  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
26  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
27  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
28  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
29  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
30  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
31  */
32
33 #include "noit_defines.h"
34 #include "eventer/eventer.h"
35 #include "utils/noit_log.h"
36 #include "utils/noit_b64.h"
37 #include "noit_jlog_listener.h"
38 #include "stratcon_jlog_streamer.h"
39 #include "stratcon_datastore.h"
40 #include "stratcon_iep.h"
41 #include "noit_conf.h"
42 #include "noit_check.h"
43
44 #include <sys/types.h>
45 #include <sys/stat.h>
46 #include <fcntl.h>
47 #include <unistd.h>
48 #include <sys/fcntl.h>
49 #ifdef HAVE_SYS_FILIO_H
50 #include <sys/filio.h>
51 #endif
52 #include <assert.h>
53 #ifdef OPENWIRE
54 #include "amqcs.h"
55 #else
56 #include "stomp/stomp.h"
57 #endif
58
59 eventer_jobq_t iep_jobq;
60 static noit_log_stream_t noit_iep = NULL;
61 static noit_spinlock_t iep_conn_cnt = 0;
62
63 struct iep_thread_driver {
64 #ifdef OPENWIRE
65   amqcs_connect_options connect_options;
66   amqcs_connection *connection;
67 #else
68   stomp_connection *connection;
69 #endif
70   apr_pool_t *pool;
71   char* exchange;
72 };
73 pthread_key_t iep_connection;
74
75 struct iep_job_closure {
76   char *line;       /* This is a copy and gets trashed during processing */
77   char *remote;
78   char *doc_str;
79   apr_pool_t *pool;
80 };
81
82 static void
83 start_iep_daemon();
84
85
86 static float
87 stratcon_iep_age_from_line(char *data, struct timeval now) {
88   float n, t;
89   if(data && (*data == 'S' || *data == 'M')) {
90     if(data[1] != '\t') return 0;
91     t = strtof(data + 2, NULL);
92     n = (float)now.tv_sec + (float)now.tv_usec / 1000000.0;
93     return n - t;
94   }
95   return 0;
96 }
97
98 struct statement_node {
99   char *id;
100   char *statement;
101   char *provides;
102   int marked; /* helps with identifying cycles */
103   int nrequires;
104   struct statement_node **requires;
105 };
106 static void
107 statement_node_free(void *vstmt) {
108   struct statement_node *stmt = vstmt;
109   if(stmt->id) free(stmt->id);
110   if(stmt->statement) free(stmt->statement);
111   if(stmt->provides) free(stmt->provides);
112   if(stmt->requires) free(stmt->requires);
113 }
114 static int
115 stmt_mark_dag(struct statement_node *stmt, int mgen) {
116   int i;
117   assert(stmt->marked <= mgen);
118   if(stmt->marked == mgen) return -1;
119   if(stmt->marked > 0) return 0; /* validated in a previous sweep */
120   stmt->marked = mgen;
121   for(i=0; i<stmt->nrequires; i++)
122     if(stmt_mark_dag(stmt->requires[i], mgen) < 0) return -1;
123   return 0;
124 }
125 static void
126 submit_statement_node(struct statement_node *stmt) {
127   int line_len, i;
128   char *line, *cp;
129
130   if(stmt->marked) return;
131   for(i=0; i<stmt->nrequires; i++)
132     submit_statement_node(stmt->requires[i]);
133
134   line_len = 3 /* 2 tabs + \0 */ +
135              1 /* 'D' */ + 1 /* '\n' */ +
136              strlen(stmt->id) + strlen(stmt->statement);
137   line = malloc(line_len);
138   snprintf(line, line_len, "D\t%s\t%s\n", stmt->id, stmt->statement);
139   cp = line;
140   while(cp[0] && cp[1]) {
141     if(*cp == '\n') *cp = ' ';
142     cp++;
143   }
144   noitL(noit_error, "submitting statement: %s\n", line);
145   stratcon_iep_line_processor(DS_OP_INSERT, NULL, NULL, line, NULL);
146   stmt->marked = 1;
147 }
148 void stratcon_iep_submit_statements() {
149   int i, cnt = 0;
150   noit_conf_section_t *statement_configs;
151   char path[256];
152   struct statement_node *stmt;
153   void *vstmt;
154   noit_hash_table stmt_by_id = NOIT_HASH_EMPTY;
155   noit_hash_table stmt_by_provider = NOIT_HASH_EMPTY;
156   noit_hash_iter iter = NOIT_HASH_ITER_ZERO;
157   const char *key;
158   int klen, mgen = 0;
159
160   snprintf(path, sizeof(path), "/stratcon/iep/queries//statement");
161   statement_configs = noit_conf_get_sections(NULL, path, &cnt);
162   noitL(noit_debug, "Found %d %s stanzas\n", cnt, path);
163
164   /* Phase 1: sweep in all the statements */
165   for(i=0; i<cnt; i++) {
166     char id[UUID_STR_LEN+1];
167     char provides[256];
168     char *statement;
169
170     if(!noit_conf_get_stringbuf(statement_configs[i],
171                                 "self::node()/@id",
172                                 id, sizeof(id))) {
173       noitL(noit_iep, "No uuid specified in query\n");
174       continue;
175     }
176     if(!noit_conf_get_stringbuf(statement_configs[i],
177                                 "ancestor-or-self::node()/@provides",
178                                 provides, sizeof(provides))) {
179       provides[0] = '\0';
180     }
181     if(!noit_conf_get_string(statement_configs[i], "self::node()/epl",
182                              &statement)) {
183       noitL(noit_iep, "No contents specified in statement\n");
184       continue;
185     }
186     stmt = calloc(1, sizeof(*stmt));
187     stmt->id = strdup(id);
188     stmt->statement = statement;
189     stmt->provides = provides[0] ? strdup(provides) : NULL;
190     if(!noit_hash_store(&stmt_by_id, stmt->id, strlen(stmt->id), stmt)) {
191       noitL(noit_error, "Duplicate statement id: %s\n", stmt->id);
192       exit(-1);
193     }
194     if(stmt->provides) {
195       if(!noit_hash_store(&stmt_by_provider, stmt->provides,
196                           strlen(stmt->provides), stmt)) {
197         noitL(noit_error, "Two statements provide: '%s'\n", stmt->provides);
198         exit(-1);
199       }
200     }
201   }
202
203   /* Phase 2: load the requires graph */
204   for(i=0; i<cnt; i++) {
205     char id[UUID_STR_LEN+1];
206     int rcnt, j;
207     char *requires;
208     noit_conf_section_t *reqs;
209
210     if(!noit_conf_get_stringbuf(statement_configs[i],
211                                 "self::node()/@id",
212                                 id, sizeof(id))) {
213       noitL(noit_iep, "No uuid specified in query\n");
214       continue;
215     }
216     if(!noit_hash_retrieve(&stmt_by_id, id, strlen(id), &vstmt)) {
217       noitL(noit_error, "Cannot find statement: %s\n", id);
218       exit(-1);
219     }
220     stmt = vstmt;
221     reqs = noit_conf_get_sections(statement_configs[i],
222                                   "self::node()/requires", &rcnt);
223     if(rcnt > 0) {
224       stmt->requires = malloc(rcnt * sizeof(*(stmt->requires)));
225       for(j=0; j<rcnt; j++) {
226         void *vrstmt;
227         if(!noit_conf_get_string(reqs[j], "self::node()",
228                                  &requires) || requires[0] == '\0') {
229           continue;
230         }
231         if(!noit_hash_retrieve(&stmt_by_provider, requires, strlen(requires),
232                                &vrstmt)) {
233           noitL(noit_error,
234                 "Statement %s requires %s which no one provides.\n",
235                 stmt->id, requires);
236           exit(-1);
237         }
238         stmt->requires[stmt->nrequires++] = vrstmt;
239       }
240     }
241   }
242
243   /* Phase 3: Recursive sweep and mark to detect cycles.
244      We're walking the graph backwards here from dependent to provider,
245      but a cycle is a cycle, so this validates the graph. */
246   while(noit_hash_next(&stmt_by_id, &iter, &key, &klen, &vstmt)) {
247     stmt = vstmt;
248     if(stmt_mark_dag(stmt, ++mgen) < 0) {
249       noitL(noit_error, "Statement %s has a cyclic requirement\n", stmt->id);
250       exit(-1);
251     }
252   }
253
254   /* Phase 4: clean the markings */
255   mgen = 0;
256   memset(&iter, 0, sizeof(iter));
257   while(noit_hash_next(&stmt_by_id, &iter, &key, &klen, &vstmt)) {
258     stmt = vstmt;
259     stmt->marked = 0;
260   }
261
262   /* Phase 5: do the load */
263   memset(&iter, 0, sizeof(iter));
264   while(noit_hash_next(&stmt_by_id, &iter, &key, &klen, &vstmt)) {
265     stmt = vstmt;
266     submit_statement_node(stmt);
267   }
268
269   noit_hash_destroy(&stmt_by_provider, NULL, NULL);
270   noit_hash_destroy(&stmt_by_id, NULL, statement_node_free);
271   free(statement_configs);
272 }
273
274 void stratcon_iep_submit_queries() {
275   int i, cnt = 0;
276   noit_conf_section_t *query_configs;
277   char path[256];
278
279   snprintf(path, sizeof(path), "/stratcon/iep/queries//query");
280   query_configs = noit_conf_get_sections(NULL, path, &cnt);
281   noitL(noit_debug, "Found %d %s stanzas\n", cnt, path);
282   for(i=0; i<cnt; i++) {
283     char id[UUID_STR_LEN+1];
284     char topic[256];
285     char *query;
286     char *line;
287     int line_len;
288
289     if(!noit_conf_get_stringbuf(query_configs[i],
290                                 "self::node()/@id",
291                                 id, sizeof(id))) {
292       noitL(noit_iep, "No uuid specified in query\n");
293       continue;
294     }
295     if(!noit_conf_get_stringbuf(query_configs[i],
296                                 "ancestor-or-self::node()/@topic",
297                                 topic, sizeof(topic))) {
298       noitL(noit_iep, "No topic specified in query\n");
299       continue;
300     }
301     if(!noit_conf_get_string(query_configs[i], "self::node()/epl",
302                              &query)) {
303       noitL(noit_iep, "No contents specified in query\n");
304       continue;
305     }
306     line_len = 4 /* 3 tabs + \0 */ +
307                1 /* 'Q' */ + 1 /* '\n' */ +
308                strlen(id) + strlen(topic) + strlen(query);
309     line = malloc(line_len);
310     snprintf(line, line_len, "Q\t%s\t%s\t%s\n", id, topic, query);
311     free(query);
312     query = line;
313     while(query[0] && query[1]) {
314       if(*query == '\n') *query = ' ';
315       query++;
316     }
317     stratcon_iep_line_processor(DS_OP_INSERT, NULL, NULL, line, NULL);
318   }
319   free(query_configs);
320 }
321
322 static
323 struct iep_thread_driver *stratcon_iep_get_connection() {
324   apr_status_t rc;
325   struct iep_thread_driver *driver;
326   driver = pthread_getspecific(iep_connection);
327   if(!driver) {
328     driver = calloc(1, sizeof(*driver));
329     pthread_setspecific(iep_connection, driver);
330   }
331
332   if(!driver->pool) {
333     if(apr_pool_create(&driver->pool, NULL) != APR_SUCCESS) return NULL;
334   }
335
336   if(!driver->connection) {
337     int port;
338     char hostname[128];
339     if(!noit_conf_get_int(NULL, "/stratcon/iep/stomp/port", &port))
340       port = 61613;
341     if(!noit_conf_get_stringbuf(NULL, "/stratcon/iep/stomp/hostname",
342                                 hostname, sizeof(hostname)))
343       strlcpy(hostname, "127.0.0.1", sizeof(hostname));
344 #ifdef OPENWIRE
345     memset(&driver->connect_options, 0, sizeof(driver->connect_options));
346     strlcpy(driver->connect_options.hostname, hostname,
347             sizeof(driver->connect_options.hostname));
348     driver->connect_options.port = port;
349     if(amqcs_connect(&driver->connection, &driver->connect_options,
350                      driver->pool) != APR_SUCCESS) {
351       noitL(noit_error, "MQ connection failed\n");
352       return NULL;
353     }
354 #else
355     if(stomp_connect(&driver->connection, hostname, port,
356                      driver->pool)!= APR_SUCCESS) {
357       noitL(noit_error, "MQ connection failed\n");
358       stomp_disconnect(&driver->connection);
359       return NULL;
360     }
361
362     {
363       stomp_frame frame;
364       char username[128];
365       char password[128];
366       char* exchange = malloc(128);
367       frame.command = "CONNECT";
368       frame.headers = apr_hash_make(driver->pool);
369       // This is for RabbitMQ Support
370       if((noit_conf_get_stringbuf(NULL, "/stratcon/iep/stomp/username",
371                                   username, sizeof(username))) &&
372          (noit_conf_get_stringbuf(NULL, "/stratcon/iep/stomp/password",
373                                   password, sizeof(password))))
374       {
375         apr_hash_set(frame.headers, "login", APR_HASH_KEY_STRING, username);
376         apr_hash_set(frame.headers, "passcode", APR_HASH_KEY_STRING, password);
377       }
378
379
380       // This is for RabbitMQ support
381       driver->exchange = NULL;
382       if(noit_conf_get_stringbuf(NULL, "/stratcon/iep/stomp/exchange",
383                                   exchange, 128))
384       {
385         if (!driver->exchange)
386           driver->exchange = exchange;
387         else
388           free(exchange);
389         apr_hash_set(frame.headers, "exchange", APR_HASH_KEY_STRING, driver->exchange);
390       }
391
392       frame.body = NULL;
393       frame.body_length = -1;
394       rc = stomp_write(driver->connection, &frame, driver->pool);
395       if(rc != APR_SUCCESS) {
396         noitL(noit_error, "MQ STOMP CONNECT failed, %d\n", rc);
397         stomp_disconnect(&driver->connection);
398         return NULL;
399       }
400     } 
401     {
402       stomp_frame *frame;
403       rc = stomp_read(driver->connection, &frame, driver->pool);
404       if (rc != APR_SUCCESS) {
405         noitL(noit_error, "MQ STOMP CONNECT bad response, %d\n", rc);
406         stomp_disconnect(&driver->connection);
407         return NULL;
408       }
409      }
410 #endif
411      stratcon_iep_submit_statements();
412      stratcon_datastore_iep_check_preload();
413      stratcon_iep_submit_queries();
414   }
415
416   return driver;
417 }
418
419 static int
420 setup_iep_connection_callback(eventer_t e, int mask, void *closure,
421                               struct timeval *now) {
422   noit_spinlock_unlock(&iep_conn_cnt);
423   stratcon_iep_line_processor(DS_OP_INSERT, NULL, NULL, NULL, NULL);
424   return 0;
425 }
426
427 static void
428 setup_iep_connection_later(int seconds) {
429   eventer_t newe;
430   if(!noit_spinlock_trylock(&iep_conn_cnt)) return;
431   newe = eventer_alloc();
432   gettimeofday(&newe->whence, NULL);
433   newe->whence.tv_sec += seconds;
434   newe->mask = EVENTER_TIMER;
435   newe->callback = setup_iep_connection_callback;
436   newe->closure = NULL;
437   eventer_add(newe);
438 }
439
440 static int
441 stratcon_iep_submitter(eventer_t e, int mask, void *closure,
442                        struct timeval *now) {
443   float age;
444   struct iep_job_closure *job = closure;
445   struct iep_thread_driver *driver;
446   /* We only play when it is an asynch event */
447   if(!(mask & EVENTER_ASYNCH_WORK)) return 0;
448
449   if(mask & EVENTER_ASYNCH_CLEANUP) {
450     /* free all the memory associated with the batch */
451     if(job) {
452       if(job->line) free(job->line);
453       if(job->remote) free(job->remote);
454       if(job->doc_str) free(job->doc_str);
455       if(job->pool) apr_pool_destroy(job->pool);
456       free(job);
457     }
458     return 0;
459   }
460   driver = stratcon_iep_get_connection();
461   if(!driver) setup_iep_connection_later(1);
462
463   if(!job->line || job->line[0] == '\0') return 0;
464
465   if((age = stratcon_iep_age_from_line(job->line, *now)) > 60) {
466     noitL(noit_debug, "Skipping old event %f second old.\n", age);
467     return 0;
468   }
469   /* Submit */
470   if(driver && driver->pool && driver->connection) {
471     apr_status_t rc;
472     int line_len = strlen(job->line);
473     int remote_len = strlen(job->remote);
474 #ifdef OPENWIRE
475     ow_ActiveMQQueue *dest;
476     ow_ActiveMQTextMessage *message;
477
478     apr_pool_create(&job->pool, driver->pool);
479     message = ow_ActiveMQTextMessage_create(job->pool);
480     message->content =
481       ow_byte_array_create_with_data(job->pool,strlen(job->doc_str),
482                                      job->doc_str);
483     dest = ow_ActiveMQQueue_create(job->pool);
484     dest->physicalName = ow_string_create_from_cstring(job->pool,"TEST.QUEUE");         
485     rc = amqcs_send(driver->connection,
486                     (ow_ActiveMQDestination*)dest,
487                     (ow_ActiveMQMessage*)message,
488                     1,4,0,job->pool);
489     if(rc != APR_SUCCESS) {
490       noitL(noit_error, "MQ send failed, disconnecting\n");
491       if(driver->connection) amqcs_disconnect(&driver->connection);
492       driver->connection = NULL;
493     }
494 #else
495     stomp_frame out;
496
497     job->doc_str = (char*)calloc(line_len + 1 /* \t */ +
498         remote_len + 2, 1);
499     strncpy(job->doc_str, job->line, 2);
500     strncat(job->doc_str, job->remote, remote_len);
501     strncat(job->doc_str, "\t", 1);
502     strncat(job->doc_str, job->line + 2, line_len - 2);
503
504     apr_pool_create(&job->pool, driver->pool);
505
506     out.command = "SEND";
507     out.headers = apr_hash_make(job->pool);
508     if (driver->exchange)
509       apr_hash_set(out.headers, "exchange", APR_HASH_KEY_STRING, driver->exchange);
510
511     apr_hash_set(out.headers, "destination", APR_HASH_KEY_STRING, "/queue/noit.firehose");
512     apr_hash_set(out.headers, "ack", APR_HASH_KEY_STRING, "auto");
513  
514     out.body_length = -1;
515     out.body = job->doc_str;
516     rc = stomp_write(driver->connection, &out, job->pool);
517     if(rc != APR_SUCCESS) {
518       noitL(noit_error, "STOMP send failed, disconnecting\n");
519       if(driver->connection) stomp_disconnect(&driver->connection);
520       driver->connection = NULL;
521     }
522 #endif
523   }
524   else {
525     noitL(noit_iep, "no iep connection, skipping line: '%s'\n", job->line);
526   }
527   return 0;
528 }
529
530 void
531 stratcon_iep_line_processor(stratcon_datastore_op_t op,
532                             struct sockaddr *remote, const char *remote_cn,
533                             void *operand, eventer_t completion) {
534   int len;
535   char remote_str[128];
536   struct iep_job_closure *jc;
537   eventer_t newe;
538   struct timeval __now, iep_timeout = { 20L, 0L };
539   /* We only care about inserts */
540
541   if(op == DS_OP_CHKPT) {
542     eventer_add(completion);
543     return;
544   }
545   if(op != DS_OP_INSERT) return;
546
547   snprintf(remote_str, sizeof(remote_str), "%s", "0.0.0.0");
548   if(remote) {
549     switch(remote->sa_family) {
550       case AF_INET:
551         len = sizeof(struct sockaddr_in);
552         inet_ntop(remote->sa_family, &((struct sockaddr_in *)remote)->sin_addr,
553                   remote_str, len);
554         break;
555       case AF_INET6:
556        len = sizeof(struct sockaddr_in6);
557         inet_ntop(remote->sa_family, &((struct sockaddr_in6 *)remote)->sin6_addr,
558                   remote_str, len);
559        break;
560       case AF_UNIX:
561         len = SUN_LEN(((struct sockaddr_un *)remote));
562         snprintf(remote_str, sizeof(remote_str), "%s", ((struct sockaddr_un *)remote)->sun_path);
563         break;
564     }
565   }
566
567   /* process operand and push onto queue */
568   gettimeofday(&__now, NULL);
569   newe = eventer_alloc();
570   newe->mask = EVENTER_ASYNCH;
571   add_timeval(__now, iep_timeout, &newe->whence);
572   newe->callback = stratcon_iep_submitter;
573   jc = calloc(1, sizeof(*jc));
574   jc->line = operand;
575   jc->remote = strdup(remote_str);
576   newe->closure = jc;
577
578   eventer_add_asynch(&iep_jobq, newe);
579 }
580
581 static void connection_destroy(void *vd) {
582   struct iep_thread_driver *driver = vd;
583 #ifdef OPENWIRE
584   if(driver->connection) amqcs_disconnect(&driver->connection);
585 #else
586   if(driver->connection) stomp_disconnect(&driver->connection);
587   if(driver->exchange) free(driver->exchange);
588 #endif
589   if(driver->pool) apr_pool_destroy(driver->pool);
590   free(driver);
591 }
592
593 jlog_streamer_ctx_t *
594 stratcon_jlog_streamer_iep_ctx_alloc(void) {
595   jlog_streamer_ctx_t *ctx;
596   ctx = stratcon_jlog_streamer_ctx_alloc();
597   ctx->jlog_feed_cmd = htonl(NOIT_JLOG_DATA_TEMP_FEED);
598   ctx->push = stratcon_iep_line_processor;
599   return ctx;
600 }
601
602 struct iep_daemon_info {
603   pid_t child;
604   int stdin_pipe[2];
605   int stderr_pipe[2];
606   char *directory;
607   char *command;
608 };
609
610 static void
611 iep_daemon_info_free(struct iep_daemon_info *info) {
612   if(!info) return;
613   if(info->directory) free(info->directory);
614   if(info->command) free(info->command);
615   if(info->stdin_pipe[0] >= 0) close(info->stdin_pipe[0]);
616   if(info->stdin_pipe[1] >= 0) close(info->stdin_pipe[1]);
617   if(info->stderr_pipe[0] >= 0) close(info->stderr_pipe[0]);
618   if(info->stderr_pipe[1] >= 0) close(info->stderr_pipe[1]);
619   free(info);
620 }
621
622 static int
623 stratcon_iep_err_handler(eventer_t e, int mask, void *closure,
624                          struct timeval *now) {
625   int len, newmask;
626   char buff[4096];
627   struct iep_daemon_info *info = (struct iep_daemon_info *)closure;
628
629   if(mask & EVENTER_EXCEPTION) {
630     int rv;
631    read_error:
632     kill(info->child, SIGKILL);
633     if(waitpid(info->child, &rv, 0) != info->child) {
634       noitL(noit_error, "Failed to reap IEP daemon\n");
635       exit(-1);
636     }
637     noitL(noit_error, "IEP daemon is done, starting a new one\n");
638     start_iep_daemon();
639     eventer_remove_fd(e->fd);
640     iep_daemon_info_free(info);
641     return 0;
642   }
643   while(1) {
644     len = e->opset->read(e->fd, buff, sizeof(buff)-1, &newmask, e);
645     if(len == -1 && (errno == EAGAIN || errno == EINTR))
646       return newmask | EVENTER_EXCEPTION;
647     if(len <= 0) goto read_error;
648     assert(len < sizeof(buff));
649     buff[len] = '\0';
650     noitL(noit_iep, "%s", buff);
651   }
652 }
653
654 static void
655 start_iep_daemon() {
656   eventer_t newe;
657   struct iep_daemon_info *info;
658
659   info = calloc(1, sizeof(*info));
660   info->stdin_pipe[0] = info->stdin_pipe[1] = -1;
661   info->stderr_pipe[0] = info->stderr_pipe[1] = -1;
662
663   if(!noit_conf_get_string(NULL, "/stratcon/iep/start/@directory",
664                            &info->directory))
665     info->directory = strdup(".");
666   if(!noit_conf_get_string(NULL, "/stratcon/iep/start/@command",
667                            &info->command)) {
668     noitL(noit_error, "No IEP start command provided.  You're on your own.\n");
669     setup_iep_connection_later(0);
670     return;
671   }
672   if(pipe(info->stdin_pipe) != 0 ||
673      pipe(info->stderr_pipe) != 0) {
674     noitL(noit_error, "pipe: %s\n", strerror(errno));
675     goto bail;
676   }
677   info->child = fork();
678   if(info->child == -1) {
679     noitL(noit_error, "fork: %s\n", strerror(errno));
680     goto bail;
681   }
682   if(info->child == 0) {
683     char *argv[3] = { "run-iep", NULL, NULL };
684     int stdout_fileno;
685
686     argv[1] = noit_conf_config_filename();
687
688     if(chdir(info->directory) != 0) {
689       noitL(noit_error, "Starting IEP daemon, chdir failed: %s\n",
690             strerror(errno));
691       exit(-1);
692     }
693
694     close(info->stdin_pipe[1]);
695     close(info->stderr_pipe[0]);
696     dup2(info->stdin_pipe[0], 0);
697     dup2(info->stderr_pipe[1], 2);
698     stdout_fileno = open("/dev/null", O_WRONLY);
699     dup2(stdout_fileno, 1);
700
701     exit(execv(info->command, argv));
702   }
703   /* in the parent */
704   close(info->stdin_pipe[0]);
705   info->stdin_pipe[0] = -1;
706   close(info->stderr_pipe[1]);
707   info->stderr_pipe[1] = -1;
708   if(eventer_set_fd_nonblocking(info->stderr_pipe[0])) {
709     goto bail;
710   }
711
712   newe = eventer_alloc();
713   newe->fd = info->stderr_pipe[0];
714   newe->mask = EVENTER_READ | EVENTER_EXCEPTION;
715   newe->callback = stratcon_iep_err_handler;
716   newe->closure = info;
717   eventer_add(newe);
718   info = NULL;
719
720   /* This will induce a stomp connection which will initialize esper */
721   setup_iep_connection_later(1);
722
723   return;
724
725  bail:
726   if(info) {
727     iep_daemon_info_free(info);
728   }
729   noitL(noit_error, "Failed to start IEP daemon\n");
730   exit(-1);
731   return;
732 }
733
734 void
735 stratcon_iep_init() {
736   noit_boolean disabled = noit_false;
737   apr_initialize();
738   atexit(apr_terminate);   
739
740   if(noit_conf_get_boolean(NULL, "/stratcon/iep/@disabled", &disabled) &&
741      disabled == noit_true) {
742     noitL(noit_error, "IEP system is disabled!\n");
743     return;
744   }
745
746   noit_iep = noit_log_stream_find("error/iep");
747   if(!noit_iep) noit_iep = noit_error;
748
749   eventer_name_callback("stratcon_iep_submitter", stratcon_iep_submitter);
750   eventer_name_callback("setup_iep_connection_callback", setup_iep_connection_callback);
751   pthread_key_create(&iep_connection, connection_destroy);
752
753   /* start up a thread pool of one */
754   memset(&iep_jobq, 0, sizeof(iep_jobq));
755   eventer_jobq_init(&iep_jobq, "iep_submitter");
756   iep_jobq.backq = eventer_default_backq();
757   eventer_jobq_increase_concurrency(&iep_jobq);
758
759   start_iep_daemon();
760
761   /* setup our live jlog stream */
762   stratcon_streamer_connection(NULL, NULL,
763                                stratcon_jlog_recv_handler,
764                                (void *(*)())stratcon_jlog_streamer_iep_ctx_alloc,
765                                NULL,
766                                jlog_streamer_ctx_free);
767 }
768
Note: See TracBrowser for help on using the browser.