root/src/stratcon_iep.c

Revision 88a71780101cbf23034aa0cb840f9f0368fda2dd, 19.6 kB (checked in by Theo Schlossnagle <jesus@omniti.com>, 6 years ago)

fixes #126

  • 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 <unistd.h>
45 #include <sys/fcntl.h>
46 #include <assert.h>
47 #include <libxml/parser.h>
48 #include <libxml/tree.h>
49 #include <libxml/xmlsave.h>
50 #ifdef OPENWIRE
51 #include "amqcs.h"
52 #else
53 #include "stomp/stomp.h"
54 #endif
55
56 eventer_jobq_t iep_jobq;
57
58 struct iep_thread_driver {
59 #ifdef OPENWIRE
60   amqcs_connect_options connect_options;
61   amqcs_connection *connection;
62 #else
63   stomp_connection *connection;
64 #endif
65   apr_pool_t *pool;
66 };
67 pthread_key_t iep_connection;
68
69 struct iep_job_closure {
70   char *line;       /* This is a copy and gets trashed during processing */
71   char *remote;
72   xmlDocPtr doc;
73   char *doc_str;
74   apr_pool_t *pool;
75 };
76
77 static void
78 start_iep_daemon();
79
80 static int
81 bust_to_parts(char *in, char **p, int len) {
82   int cnt = 0;
83   char *s = in;
84   while(cnt < len) {
85     p[cnt++] = s;
86     while(*s && *s != '\t') s++;
87     if(!*s) break;
88     *s++ = '\0';
89   }
90   while(*s) s++; /* Move to end */
91   if(s > in && *(s-1) == '\n') *(s-1) = '\0'; /* chomp */
92   return cnt;
93 }
94
95 #define ADDCHILD(a,b) \
96   xmlNewTextChild(root, NULL, (xmlChar *)(a), (xmlChar *)(b))
97 #define NEWDOC(xmldoc,n,stanza) do { \
98   xmlNodePtr root; \
99   xmldoc = xmlNewDoc((xmlChar *)"1.0"); \
100   root = xmlNewDocNode(xmldoc, NULL, (xmlChar *)(n), NULL); \
101   xmlDocSetRootElement(xmldoc, root); \
102   stanza \
103 } while(0)
104
105
106 static xmlDocPtr
107 stratcon_iep_doc_from_status(char *data, char *remote) {
108   xmlDocPtr doc;
109   char *parts[7];
110   if(bust_to_parts(data, parts, 7) != 7) return NULL;
111   /* 'S' TIMESTAMP UUID STATE AVAILABILITY DURATION STATUS_MESSAGE */
112   NEWDOC(doc, "NoitStatus",
113          {
114            ADDCHILD("remote", remote);
115            ADDCHILD("id", parts[2]);
116            ADDCHILD("state", parts[3]);
117            ADDCHILD("availability", parts[4]);
118            ADDCHILD("duration", parts[5]);
119            ADDCHILD("status", parts[6]);
120          });
121   return doc;
122 }
123
124 static xmlDocPtr
125 stratcon_iep_doc_from_check(char *data, char *remote) {
126   xmlDocPtr doc;
127   char *parts[6];
128   if(bust_to_parts(data, parts, 6) != 6) return NULL;
129   /* 'C' TIMESTAMP UUID TARGET MODULE NAME */
130   NEWDOC(doc, "NoitCheck",
131          {
132            ADDCHILD("remote", remote);
133            ADDCHILD("id", parts[2]);
134            ADDCHILD("target", parts[3]);
135            ADDCHILD("module", parts[4]);
136            ADDCHILD("name", parts[5]);
137          });
138 noitL(noit_error,"Submitting check %s\n", parts[2]);
139   return doc;
140 }
141
142 static xmlDocPtr
143 stratcon_iep_doc_from_metric(char *data, char *remote) {
144   xmlDocPtr doc;
145   char *parts[6];
146   const char *rootname = "NoitMetricNumeric";
147   const char *valuename = "value";
148   if(bust_to_parts(data, parts, 6) != 6) return NULL;
149   /*  'M' TIMESTAMP UUID NAME TYPE VALUE */
150
151   if(*parts[4] == METRIC_STRING) {
152     rootname = "NoitMetricText";
153     valuename = "message";
154   }
155   NEWDOC(doc, rootname,
156          {
157            ADDCHILD("remote", remote);
158            ADDCHILD("id", parts[2]);
159            ADDCHILD("name", parts[3]);
160            ADDCHILD(valuename, parts[5]);
161          });
162   return doc;
163 }
164
165 static xmlDocPtr
166 stratcon_iep_doc_from_query(char *data, char *remote) {
167   xmlDocPtr doc;
168   char *parts[4];
169   if(bust_to_parts(data, parts, 4) != 4) return NULL;
170   /*  'Q' ID NAME QUERY  */
171
172   NEWDOC(doc, "StratconQuery",
173          {
174            ADDCHILD("id", parts[1]);
175            ADDCHILD("name", parts[2]);
176            ADDCHILD("expression", parts[3]);
177          });
178   return doc;
179 }
180
181 static xmlDocPtr
182 stratcon_iep_doc_from_querystop(char *data, char *remote) {
183   xmlDocPtr doc;
184   char *parts[2];
185   if(bust_to_parts(data, parts, 2) != 2) return NULL;
186   /*  'Q' ID */
187
188   NEWDOC(doc, "StratconQueryStop",
189          {
190            xmlNodeSetContent(root, (xmlChar *)parts[1]);
191          });
192   return doc;
193 }
194
195 static xmlDocPtr
196 stratcon_iep_doc_from_line(char *data, char *remote) {
197   if(data) {
198     switch(*data) {
199       case 'C': return stratcon_iep_doc_from_check(data, remote);
200       case 'S': return stratcon_iep_doc_from_status(data, remote);
201       case 'M': return stratcon_iep_doc_from_metric(data, remote);
202       case 'Q': return stratcon_iep_doc_from_query(data, remote);
203       case 'q': return stratcon_iep_doc_from_querystop(data, remote);
204     }
205   }
206   return NULL;
207 }
208
209 static float
210 stratcon_iep_age_from_line(char *data, struct timeval now) {
211   float n, t;
212   if(data && (*data == 'S' || *data == 'M')) {
213     if(data[1] != '\t') return 0;
214     t = strtof(data + 2, NULL);
215     n = (float)now.tv_sec + (float)now.tv_usec / 1000000.0;
216     return n - t;
217   }
218   return 0;
219 }
220
221 void stratcon_iep_submit_queries() {
222   int i, cnt = 0;
223   noit_conf_section_t *query_configs;
224   char path[256];
225
226   snprintf(path, sizeof(path), "/stratcon/iep/queries//query");
227   query_configs = noit_conf_get_sections(NULL, path, &cnt);
228   noitL(noit_debug, "Found %d %s stanzas\n", cnt, path);
229   for(i=0; i<cnt; i++) {
230     char id[UUID_STR_LEN];
231     char topic[256];
232     char *query;
233     char *line;
234     int line_len;
235
236     if(!noit_conf_get_stringbuf(query_configs[i],
237                                 "self::node()/@id",
238                                 id, sizeof(id))) {
239       noitL(noit_error, "No uuid specified in query\n");
240       continue;
241     }
242     if(!noit_conf_get_stringbuf(query_configs[i],
243                                 "ancestor-or-self::node()/@topic",
244                                 topic, sizeof(topic))) {
245       noitL(noit_error, "No topic specified in query\n");
246       continue;
247     }
248     if(!noit_conf_get_string(query_configs[i], "self::node()",
249                              &query)) {
250       noitL(noit_error, "No contents specified in query\n");
251       continue;
252     }
253     line_len = 4 /* 3 tabs + \0 */ +
254                1 /* 'Q' */ + 1 /* '\n' */ +
255                strlen(id) + strlen(topic) + strlen(query);
256     line = malloc(line_len);
257     snprintf(line, line_len, "Q\t%s\t%s\t%s\n", id, topic, query);
258     free(query);
259     query = line;
260     while(query[0] && query[1]) {
261       if(*query == '\n') *query = ' ';
262       query++;
263     }
264     stratcon_iep_line_processor(DS_OP_INSERT, NULL, line);
265     free(line);
266   }
267 }
268
269 static char *
270 stratcon__xml_doc_to_str(xmlDocPtr doc) {
271   char *rv;
272   xmlSaveCtxtPtr savectx;
273   xmlBufferPtr xmlbuffer;
274   xmlbuffer = xmlBufferCreate();
275   savectx = xmlSaveToBuffer(xmlbuffer, "utf8", 1);
276   xmlSaveDoc(savectx, doc);
277   xmlSaveClose(savectx);
278   rv = strdup((const char *)xmlBufferContent(xmlbuffer));
279   xmlBufferFree(xmlbuffer);
280   return rv;
281 }
282
283 static
284 struct iep_thread_driver *stratcon_iep_get_connection() {
285   apr_status_t rc;
286   struct iep_thread_driver *driver;
287   driver = pthread_getspecific(iep_connection);
288   if(!driver) {
289     driver = calloc(1, sizeof(*driver));
290     pthread_setspecific(iep_connection, driver);
291   }
292
293   if(!driver->pool) {
294     if(apr_pool_create(&driver->pool, NULL) != APR_SUCCESS) return NULL;
295   }
296
297   if(!driver->connection) {
298     int port;
299     char hostname[128];
300     if(!noit_conf_get_int(NULL, "/stratcon/iep/port", &port))
301       port = 61613;
302     if(!noit_conf_get_stringbuf(NULL, "/stratcon/iep/hostname",
303                                 hostname, sizeof(hostname)))
304       strlcpy(hostname, "127.0.0.1", sizeof(hostname));
305 #ifdef OPENWIRE
306     memset(&driver->connect_options, 0, sizeof(driver->connect_options));
307     strlcpy(driver->connect_options.hostname, hostname,
308             sizeof(driver->connect_options.hostname));
309     driver->connect_options.port = port;
310     if(amqcs_connect(&driver->connection, &driver->connect_options,
311                      driver->pool) != APR_SUCCESS) {
312       noitL(noit_error, "MQ connection failed\n");
313       return NULL;
314     }
315 #else
316     if(stomp_connect(&driver->connection, hostname, port,
317                      driver->pool)!= APR_SUCCESS) {
318       noitL(noit_error, "MQ connection failed\n");
319       return NULL;
320     }
321
322     {
323       stomp_frame frame;
324       frame.command = "CONNECT";
325       frame.headers = apr_hash_make(driver->pool);
326 /*
327       We don't use login/pass
328       apr_hash_set(frame.headers, "login", APR_HASH_KEY_STRING, "");
329       apr_hash_set(frame.headers, "passcode", APR_HASH_KEY_STRING, "");
330 */
331       frame.body = NULL;
332       frame.body_length = -1;
333       rc = stomp_write(driver->connection, &frame, driver->pool);
334       if(rc != APR_SUCCESS) {
335         noitL(noit_error, "MQ STOMP CONNECT failed, %d\n", rc);
336         stomp_disconnect(&driver->connection);
337         return NULL;
338       }
339     } 
340     {
341       stomp_frame *frame;
342       rc = stomp_read(driver->connection, &frame, driver->pool);
343       if (rc != APR_SUCCESS) {
344         noitL(noit_error, "MQ STOMP CONNECT bad response, %d\n", rc);
345         stomp_disconnect(&driver->connection);
346         return NULL;
347       }
348       noitL(noit_error, "Response: %s, %s\n", frame->command, frame->body);
349      }
350 #endif
351      stratcon_datastore_iep_check_preload();
352      stratcon_iep_submit_queries();
353   }
354
355   return driver;
356 }
357
358 static int
359 stratcon_iep_submitter(eventer_t e, int mask, void *closure,
360                        struct timeval *now) {
361   float age;
362   struct iep_job_closure *job = closure;
363   /* We only play when it is an asynch event */
364   if(!(mask & EVENTER_ASYNCH_WORK)) return 0;
365
366   if(mask & EVENTER_ASYNCH_CLEANUP) {
367     /* free all the memory associated with the batch */
368     if(job) {
369       if(job->line) free(job->line);
370       if(job->remote) free(job->remote);
371       if(job->doc_str) free(job->doc_str);
372       if(job->doc) xmlFreeDoc(job->doc);
373       if(job->pool) apr_pool_destroy(job->pool);
374       free(job);
375     }
376     return 0;
377   }
378
379   if((age = stratcon_iep_age_from_line(job->line, *now)) > 60) {
380     noitL(noit_debug, "Skipping old event %f second old.\n", age);
381     return 0;
382   }
383   job->doc = stratcon_iep_doc_from_line(job->line, job->remote);
384   if(job->doc) {
385     job->doc_str = stratcon__xml_doc_to_str(job->doc);
386     if(job->doc_str) {
387       /* Submit */
388       struct iep_thread_driver *driver;
389       driver = stratcon_iep_get_connection();
390       if(driver && driver->pool && driver->connection) {
391         apr_status_t rc;
392 #ifdef OPENWIRE
393         ow_ActiveMQQueue *dest;
394         ow_ActiveMQTextMessage *message;
395
396         apr_pool_create(&job->pool, driver->pool);
397         message = ow_ActiveMQTextMessage_create(job->pool);
398         message->content =
399           ow_byte_array_create_with_data(job->pool,strlen(job->doc_str),
400                                          job->doc_str);
401         dest = ow_ActiveMQQueue_create(job->pool);
402         dest->physicalName = ow_string_create_from_cstring(job->pool,"TEST.QUEUE");         
403         rc = amqcs_send(driver->connection,
404                         (ow_ActiveMQDestination*)dest,
405                         (ow_ActiveMQMessage*)message,
406                         1,4,0,job->pool);
407         if(rc != APR_SUCCESS) {
408           noitL(noit_error, "MQ send failed, disconnecting\n");
409           if(driver->connection) amqcs_disconnect(&driver->connection);
410           driver->connection = NULL;
411         }
412 #else
413         stomp_frame out;
414
415         apr_pool_create(&job->pool, driver->pool);
416
417         out.command = "SEND";
418         out.headers = apr_hash_make(job->pool);
419         apr_hash_set(out.headers, "destination", APR_HASH_KEY_STRING, "/queue/noit.firehose");
420         apr_hash_set(out.headers, "ack", APR_HASH_KEY_STRING, "auto");
421      
422         out.body_length = -1;
423         out.body = job->doc_str;
424         rc = stomp_write(driver->connection, &out, job->pool);
425         if(rc != APR_SUCCESS) {
426           noitL(noit_error, "STOMP send failed, disconnecting\n");
427           if(driver->connection) stomp_disconnect(&driver->connection);
428           driver->connection = NULL;
429         }
430 #endif
431       }
432       else {
433         noitL(noit_error, "Not submitting event, no MQ\n");
434       }
435     }
436   }
437   else {
438     noitL(noit_error, "no iep handler for: '%s'\n", job->line);
439   }
440   return 0;
441 }
442
443 void
444 stratcon_iep_line_processor(stratcon_datastore_op_t op,
445                             struct sockaddr *remote, void *operand) {
446   int len;
447   char remote_str[128];
448   struct iep_job_closure *jc;
449   eventer_t newe;
450   struct timeval __now, iep_timeout = { 20L, 0L };
451   /* We only care about inserts */
452
453   if(op == DS_OP_CHKPT) {
454     eventer_add((eventer_t) operand);
455     return;
456   }
457   if(op != DS_OP_INSERT) return;
458
459   snprintf(remote_str, sizeof(remote_str), "%s", "0.0.0.0");
460   if(remote) {
461     switch(remote->sa_family) {
462       case AF_INET:
463         len = sizeof(struct sockaddr_in);
464         inet_ntop(remote->sa_family, &((struct sockaddr_in *)remote)->sin_addr,
465                   remote_str, len);
466         break;
467       case AF_INET6:
468        len = sizeof(struct sockaddr_in6);
469         inet_ntop(remote->sa_family, &((struct sockaddr_in6 *)remote)->sin6_addr,
470                   remote_str, len);
471        break;
472       case AF_UNIX:
473         len = SUN_LEN(((struct sockaddr_un *)remote));
474         snprintf(remote_str, sizeof(remote_str), "%s", ((struct sockaddr_un *)remote)->sun_path);
475         break;
476     }
477   }
478
479   /* process operand and push onto queue */
480   gettimeofday(&__now, NULL);
481   newe = eventer_alloc();
482   newe->mask = EVENTER_ASYNCH;
483   add_timeval(__now, iep_timeout, &newe->whence);
484   newe->callback = stratcon_iep_submitter;
485   jc = calloc(1, sizeof(*jc));
486   jc->line = strdup(operand);
487   jc->remote = strdup(remote_str);
488   newe->closure = jc;
489
490   eventer_add_asynch(&iep_jobq, newe);
491 }
492
493 static void connection_destroy(void *vd) {
494   struct iep_thread_driver *driver = vd;
495 #ifdef OPENWIRE
496   if(driver->connection) amqcs_disconnect(&driver->connection);
497 #else
498   if(driver->connection) stomp_disconnect(&driver->connection);
499 #endif
500   if(driver->pool) apr_pool_destroy(driver->pool);
501   free(driver);
502 }
503
504 jlog_streamer_ctx_t *
505 stratcon_jlog_streamer_iep_ctx_alloc(void) {
506   jlog_streamer_ctx_t *ctx;
507   ctx = stratcon_jlog_streamer_ctx_alloc();
508   ctx->jlog_feed_cmd = htonl(NOIT_JLOG_DATA_TEMP_FEED);
509   ctx->push = stratcon_iep_line_processor;
510   return ctx;
511 }
512
513 struct iep_daemon_info {
514   pid_t child;
515   int stdin_pipe[2];
516   int stderr_pipe[2];
517   char *directory;
518   char *command;
519 };
520
521 static void
522 iep_daemon_info_free(struct iep_daemon_info *info) {
523   if(!info) return;
524   if(info->directory) free(info->directory);
525   if(info->command) free(info->command);
526   if(info->stdin_pipe[0] >= 0) close(info->stdin_pipe[0]);
527   if(info->stdin_pipe[1] >= 0) close(info->stdin_pipe[1]);
528   if(info->stderr_pipe[0] >= 0) close(info->stderr_pipe[0]);
529   if(info->stderr_pipe[1] >= 0) close(info->stderr_pipe[1]);
530   free(info);
531 }
532
533 static int
534 stratcon_iep_err_handler(eventer_t e, int mask, void *closure,
535                          struct timeval *now) {
536   int len, newmask;
537   char buff[4096];
538   struct iep_daemon_info *info = (struct iep_daemon_info *)closure;
539
540   if(mask & EVENTER_EXCEPTION) {
541     int rv;
542    read_error:
543     kill(SIGKILL, info->child);
544     if(waitpid(info->child, &rv, 0) != info->child) {
545       noitL(noit_error, "Failed to reap IEP daemon\n");
546       exit(-1);
547     }
548     noitL(noit_error, "IEP daemon is done, starting a new one\n");
549     start_iep_daemon();
550     eventer_remove_fd(e->fd);
551     e->opset->close(e->fd, &newmask, e);
552     return 0;
553   }
554   while(1) {
555     len = e->opset->read(e->fd, buff, sizeof(buff)-1, &newmask, e);
556     if(len == -1 && (errno == EAGAIN || errno == EINTR))
557       return newmask | EVENTER_EXCEPTION;
558     if(len <= 0) goto read_error;
559     assert(len < sizeof(buff));
560     buff[len] = '\0';
561     noitL(noit_error, "IEP: %s", buff);
562   }
563 }
564
565 static void
566 start_iep_daemon() {
567   eventer_t newe;
568   struct iep_daemon_info *info;
569
570   info = calloc(1, sizeof(*info));
571   info->stdin_pipe[0] = info->stdin_pipe[1] = -1;
572   info->stderr_pipe[0] = info->stderr_pipe[1] = -1;
573
574   if(!noit_conf_get_string(NULL, "/stratcon/iep/start/@directory",
575                            &info->directory))
576     info->directory = strdup(".");
577   if(!noit_conf_get_string(NULL, "/stratcon/iep/start/@command",
578                            &info->command)) {
579     noitL(noit_error, "No IEP start command provided.  You're on your own.\n");
580     goto bail;
581   }
582   if(pipe(info->stdin_pipe) != 0 ||
583      pipe(info->stderr_pipe) != 0) {
584     noitL(noit_error, "pipe: %s\n", strerror(errno));
585     goto bail;
586   }
587   info->child = fork();
588   if(info->child == -1) {
589     noitL(noit_error, "fork: %s\n", strerror(errno));
590     goto bail;
591   }
592   if(info->child == 0) {
593     char *argv[2] = { "run-iep", NULL };
594     int stdout_fileno;
595
596     if(chdir(info->directory) != 0) {
597       noitL(noit_error, "Starting IEP daemon, chdir failed: %s\n",
598             strerror(errno));
599       exit(-1);
600     }
601
602     close(info->stdin_pipe[1]);
603     close(info->stderr_pipe[0]);
604     dup2(info->stdin_pipe[0], 0);
605     dup2(info->stderr_pipe[1], 2);
606     stdout_fileno = open("/dev/null", O_WRONLY);
607     dup2(stdout_fileno, 1);
608
609     exit(execv(info->command, argv));
610   }
611   /* in the parent */
612   socklen_t on = 1;
613
614   close(info->stdin_pipe[0]);
615   info->stdin_pipe[0] = -1;
616   close(info->stderr_pipe[1]);
617   info->stderr_pipe[1] = -1;
618   if(ioctl(info->stderr_pipe[0], FIONBIO, &on)) {
619     goto bail;
620   }
621
622   newe = eventer_alloc();
623   newe->fd = info->stderr_pipe[0];
624   newe->mask = EVENTER_READ | EVENTER_EXCEPTION;
625   newe->callback = stratcon_iep_err_handler;
626   newe->closure = info;
627   eventer_add(newe);
628   info = NULL;
629
630   return;
631
632  bail:
633   if(info) {
634     iep_daemon_info_free(info);
635   }
636   noitL(noit_error, "Failed to start IEP daemon\n");
637   exit(-1);
638   return;
639 }
640
641 void
642 stratcon_iep_init() {
643   noit_boolean disabled = noit_false;
644   apr_initialize();
645   atexit(apr_terminate);   
646
647   if(noit_conf_get_boolean(NULL, "/stratcon/iep/@disabled", &disabled) &&
648      disabled == noit_true) {
649     noitL(noit_error, "IEP system is disabled!\n");
650     return;
651   }
652
653   start_iep_daemon();
654
655   eventer_name_callback("stratcon_iep_submitter", stratcon_iep_submitter);
656   pthread_key_create(&iep_connection, connection_destroy);
657
658   /* start up a thread pool of one */
659   memset(&iep_jobq, 0, sizeof(iep_jobq));
660   eventer_jobq_init(&iep_jobq, "iep_submitter");
661   iep_jobq.backq = eventer_default_backq();
662   eventer_jobq_increase_concurrency(&iep_jobq);
663
664   /* setup our live jlog stream */
665   stratcon_streamer_connection(NULL, NULL,
666                                stratcon_jlog_recv_handler,
667                                (void *(*)())stratcon_jlog_streamer_iep_ctx_alloc,
668                                NULL,
669                                jlog_streamer_ctx_free);
670 }
671
Note: See TracBrowser for help on using the browser.