root/src/stratcon_iep.c

Revision 222027809f476c8187ccf88ac8b565d3787094c4, 19.4 kB (checked in by Theo Schlossnagle <jesus@omniti.com>, 5 years ago)

make iep uses dedicated logging stream for more flexibility. Fix the numeric event representaiton in Java. Demonstrates we need an object-based event for Text stuff too... as the other doesn't handle null right.

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