root/src/stratcon_iep.c

Revision 1879de96c939c4bf7efb77e8ce76a0827308b2f9, 21.2 kB (checked in by Phil Maddox <philip.maddox@circonus.com>, 1 day ago)

Up Maximum Event Age From 10 Seconds To 30 Seconds

10 seconds is too tight a maximum age for events.... up it to 30 seconds
instead.

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