root/src/stratcon_iep.c

Revision 24eac7c456bb2802cf4a695e79f812cc3a0b4b0c, 18.1 kB (checked in by Theo Schlossnagle <jesus@omniti.com>, 5 years ago)

variety of stuff. fixing some bugs and adding checks into the data model, refs #122

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