root/src/stratcon_iep.c

Revision 699c97c25d44d177d2979ba0008d46658927a912, 27.3 kB (checked in by Theo Schlossnagle <jesus@omniti.com>, 9 years ago)

works for me. Ensures there is a (perhaps non-connected) DAG of statement dependencies and runs them, then loads data, then runs queries. refs #162

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