root/src/stratcon_iep.c

Revision 66b174ed5ef14f2ccffa1e5db926e291a58b1f7d, 27.3 kB (checked in by Theo Schlossnagle <jesus@omniti.com>, 5 years ago)

embarassingly backwards, fixes #166

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