root/src/stratcon_iep.c

Revision 5360a1ee7f4ade3bd6299e566204ee1820aeffa3, 23.4 kB (checked in by Theo Schlossnagle <jesus@omniti.com>, 5 years ago)

hefty patch. First part of stratcond support for horizontal partitioning of storage. refs #150

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