root/src/stratcon_iep.c

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

openwire never worked, clean it out

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