1 |
/* |
---|
2 |
* Copyright (c) 2007, OmniTI Computer Consulting, Inc. |
---|
3 |
* All rights reserved. |
---|
4 |
* |
---|
5 |
* Redistribution and use in source and binary forms, with or without |
---|
6 |
* modification, are permitted provided that the following conditions are |
---|
7 |
* met: |
---|
8 |
* |
---|
9 |
* * Redistributions of source code must retain the above copyright |
---|
10 |
* notice, this list of conditions and the following disclaimer. |
---|
11 |
* * Redistributions in binary form must reproduce the above |
---|
12 |
* copyright notice, this list of conditions and the following |
---|
13 |
* disclaimer in the documentation and/or other materials provided |
---|
14 |
* with the distribution. |
---|
15 |
* * Neither the name OmniTI Computer Consulting, Inc. nor the names |
---|
16 |
* of its contributors may be used to endorse or promote products |
---|
17 |
* derived from this software without specific prior written |
---|
18 |
* permission. |
---|
19 |
* |
---|
20 |
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
---|
21 |
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
---|
22 |
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
---|
23 |
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
---|
24 |
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
---|
25 |
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
---|
26 |
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
---|
27 |
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
---|
28 |
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
---|
29 |
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
---|
30 |
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
---|
31 |
*/ |
---|
32 |
|
---|
33 |
#include "noit_defines.h" |
---|
34 |
#include "eventer/eventer.h" |
---|
35 |
#include "utils/noit_log.h" |
---|
36 |
#include "utils/noit_b64.h" |
---|
37 |
#include "noit_jlog_listener.h" |
---|
38 |
#include "stratcon_jlog_streamer.h" |
---|
39 |
#include "stratcon_datastore.h" |
---|
40 |
#include "stratcon_iep.h" |
---|
41 |
#include "noit_conf.h" |
---|
42 |
#include "noit_check.h" |
---|
43 |
|
---|
44 |
#include <sys/types.h> |
---|
45 |
#include <sys/stat.h> |
---|
46 |
#include <fcntl.h> |
---|
47 |
#include <unistd.h> |
---|
48 |
#include <sys/fcntl.h> |
---|
49 |
#ifdef HAVE_SYS_FILIO_H |
---|
50 |
#include <sys/filio.h> |
---|
51 |
#endif |
---|
52 |
#include <assert.h> |
---|
53 |
#ifdef OPENWIRE |
---|
54 |
#include "amqcs.h" |
---|
55 |
#else |
---|
56 |
#include "stomp/stomp.h" |
---|
57 |
#endif |
---|
58 |
|
---|
59 |
eventer_jobq_t iep_jobq; |
---|
60 |
static noit_log_stream_t noit_iep = NULL; |
---|
61 |
static noit_spinlock_t iep_conn_cnt = 0; |
---|
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 |
char *doc_str; |
---|
79 |
apr_pool_t *pool; |
---|
80 |
}; |
---|
81 |
|
---|
82 |
static void |
---|
83 |
start_iep_daemon(); |
---|
84 |
|
---|
85 |
|
---|
86 |
static float |
---|
87 |
stratcon_iep_age_from_line(char *data, struct timeval now) { |
---|
88 |
float n, t; |
---|
89 |
if(data && (*data == 'S' || *data == 'M')) { |
---|
90 |
if(data[1] != '\t') return 0; |
---|
91 |
t = strtof(data + 2, NULL); |
---|
92 |
n = (float)now.tv_sec + (float)now.tv_usec / 1000000.0; |
---|
93 |
return n - t; |
---|
94 |
} |
---|
95 |
return 0; |
---|
96 |
} |
---|
97 |
|
---|
98 |
struct statement_node { |
---|
99 |
char *id; |
---|
100 |
char *statement; |
---|
101 |
char *provides; |
---|
102 |
int marked; /* helps with identifying cycles */ |
---|
103 |
int nrequires; |
---|
104 |
struct statement_node **requires; |
---|
105 |
}; |
---|
106 |
static void |
---|
107 |
statement_node_free(void *vstmt) { |
---|
108 |
struct statement_node *stmt = vstmt; |
---|
109 |
if(stmt->id) free(stmt->id); |
---|
110 |
if(stmt->statement) free(stmt->statement); |
---|
111 |
if(stmt->provides) free(stmt->provides); |
---|
112 |
if(stmt->requires) free(stmt->requires); |
---|
113 |
} |
---|
114 |
static int |
---|
115 |
stmt_mark_dag(struct statement_node *stmt, int mgen) { |
---|
116 |
int i; |
---|
117 |
assert(stmt->marked <= mgen); |
---|
118 |
if(stmt->marked == mgen) return -1; |
---|
119 |
if(stmt->marked > 0) return 0; /* validated in a previous sweep */ |
---|
120 |
stmt->marked = mgen; |
---|
121 |
for(i=0; i<stmt->nrequires; i++) |
---|
122 |
if(stmt_mark_dag(stmt->requires[i], mgen) < 0) return -1; |
---|
123 |
return 0; |
---|
124 |
} |
---|
125 |
static void |
---|
126 |
submit_statement_node(struct statement_node *stmt) { |
---|
127 |
int line_len, i; |
---|
128 |
char *line, *cp; |
---|
129 |
|
---|
130 |
if(stmt->marked) return; |
---|
131 |
for(i=0; i<stmt->nrequires; i++) |
---|
132 |
submit_statement_node(stmt->requires[i]); |
---|
133 |
|
---|
134 |
line_len = 3 /* 2 tabs + \0 */ + |
---|
135 |
1 /* 'D' */ + 1 /* '\n' */ + |
---|
136 |
strlen(stmt->id) + strlen(stmt->statement); |
---|
137 |
line = malloc(line_len); |
---|
138 |
snprintf(line, line_len, "D\t%s\t%s\n", stmt->id, stmt->statement); |
---|
139 |
cp = line; |
---|
140 |
while(cp[0] && cp[1]) { |
---|
141 |
if(*cp == '\n') *cp = ' '; |
---|
142 |
cp++; |
---|
143 |
} |
---|
144 |
noitL(noit_error, "submitting statement: %s\n", line); |
---|
145 |
stratcon_iep_line_processor(DS_OP_INSERT, NULL, NULL, line, NULL); |
---|
146 |
stmt->marked = 1; |
---|
147 |
} |
---|
148 |
void stratcon_iep_submit_statements() { |
---|
149 |
int i, cnt = 0; |
---|
150 |
noit_conf_section_t *statement_configs; |
---|
151 |
char path[256]; |
---|
152 |
struct statement_node *stmt; |
---|
153 |
void *vstmt; |
---|
154 |
noit_hash_table stmt_by_id = NOIT_HASH_EMPTY; |
---|
155 |
noit_hash_table stmt_by_provider = NOIT_HASH_EMPTY; |
---|
156 |
noit_hash_iter iter = NOIT_HASH_ITER_ZERO; |
---|
157 |
const char *key; |
---|
158 |
int klen, mgen = 0; |
---|
159 |
|
---|
160 |
snprintf(path, sizeof(path), "/stratcon/iep/queries//statement"); |
---|
161 |
statement_configs = noit_conf_get_sections(NULL, path, &cnt); |
---|
162 |
noitL(noit_debug, "Found %d %s stanzas\n", cnt, path); |
---|
163 |
|
---|
164 |
/* Phase 1: sweep in all the statements */ |
---|
165 |
for(i=0; i<cnt; i++) { |
---|
166 |
char id[UUID_STR_LEN+1]; |
---|
167 |
char provides[256]; |
---|
168 |
char *statement; |
---|
169 |
|
---|
170 |
if(!noit_conf_get_stringbuf(statement_configs[i], |
---|
171 |
"self::node()/@id", |
---|
172 |
id, sizeof(id))) { |
---|
173 |
noitL(noit_iep, "No uuid specified in query\n"); |
---|
174 |
continue; |
---|
175 |
} |
---|
176 |
if(!noit_conf_get_stringbuf(statement_configs[i], |
---|
177 |
"ancestor-or-self::node()/@provides", |
---|
178 |
provides, sizeof(provides))) { |
---|
179 |
provides[0] = '\0'; |
---|
180 |
} |
---|
181 |
if(!noit_conf_get_string(statement_configs[i], "self::node()/epl", |
---|
182 |
&statement)) { |
---|
183 |
noitL(noit_iep, "No contents specified in statement\n"); |
---|
184 |
continue; |
---|
185 |
} |
---|
186 |
stmt = calloc(1, sizeof(*stmt)); |
---|
187 |
stmt->id = strdup(id); |
---|
188 |
stmt->statement = statement; |
---|
189 |
stmt->provides = provides[0] ? strdup(provides) : NULL; |
---|
190 |
if(!noit_hash_store(&stmt_by_id, stmt->id, strlen(stmt->id), stmt)) { |
---|
191 |
noitL(noit_error, "Duplicate statement id: %s\n", stmt->id); |
---|
192 |
exit(-1); |
---|
193 |
} |
---|
194 |
if(stmt->provides) { |
---|
195 |
if(!noit_hash_store(&stmt_by_provider, stmt->provides, |
---|
196 |
strlen(stmt->provides), stmt)) { |
---|
197 |
noitL(noit_error, "Two statements provide: '%s'\n", stmt->provides); |
---|
198 |
exit(-1); |
---|
199 |
} |
---|
200 |
} |
---|
201 |
} |
---|
202 |
|
---|
203 |
/* Phase 2: load the requires graph */ |
---|
204 |
for(i=0; i<cnt; i++) { |
---|
205 |
char id[UUID_STR_LEN+1]; |
---|
206 |
int rcnt, j; |
---|
207 |
char *requires; |
---|
208 |
noit_conf_section_t *reqs; |
---|
209 |
|
---|
210 |
if(!noit_conf_get_stringbuf(statement_configs[i], |
---|
211 |
"self::node()/@id", |
---|
212 |
id, sizeof(id))) { |
---|
213 |
noitL(noit_iep, "No uuid specified in query\n"); |
---|
214 |
continue; |
---|
215 |
} |
---|
216 |
if(!noit_hash_retrieve(&stmt_by_id, id, strlen(id), &vstmt)) { |
---|
217 |
noitL(noit_error, "Cannot find statement: %s\n", id); |
---|
218 |
exit(-1); |
---|
219 |
} |
---|
220 |
stmt = vstmt; |
---|
221 |
reqs = noit_conf_get_sections(statement_configs[i], |
---|
222 |
"self::node()/requires", &rcnt); |
---|
223 |
if(rcnt > 0) { |
---|
224 |
stmt->requires = malloc(rcnt * sizeof(*(stmt->requires))); |
---|
225 |
for(j=0; j<rcnt; j++) { |
---|
226 |
void *vrstmt; |
---|
227 |
if(!noit_conf_get_string(reqs[j], "self::node()", |
---|
228 |
&requires) || requires[0] == '\0') { |
---|
229 |
continue; |
---|
230 |
} |
---|
231 |
if(!noit_hash_retrieve(&stmt_by_provider, requires, strlen(requires), |
---|
232 |
&vrstmt)) { |
---|
233 |
noitL(noit_error, |
---|
234 |
"Statement %s requires %s which no one provides.\n", |
---|
235 |
stmt->id, requires); |
---|
236 |
exit(-1); |
---|
237 |
} |
---|
238 |
stmt->requires[stmt->nrequires++] = vrstmt; |
---|
239 |
} |
---|
240 |
} |
---|
241 |
} |
---|
242 |
|
---|
243 |
/* Phase 3: Recursive sweep and mark to detect cycles. |
---|
244 |
We're walking the graph backwards here from dependent to provider, |
---|
245 |
but a cycle is a cycle, so this validates the graph. */ |
---|
246 |
while(noit_hash_next(&stmt_by_id, &iter, &key, &klen, &vstmt)) { |
---|
247 |
stmt = vstmt; |
---|
248 |
if(stmt_mark_dag(stmt, ++mgen) < 0) { |
---|
249 |
noitL(noit_error, "Statement %s has a cyclic requirement\n", stmt->id); |
---|
250 |
exit(-1); |
---|
251 |
} |
---|
252 |
} |
---|
253 |
|
---|
254 |
/* Phase 4: clean the markings */ |
---|
255 |
mgen = 0; |
---|
256 |
memset(&iter, 0, sizeof(iter)); |
---|
257 |
while(noit_hash_next(&stmt_by_id, &iter, &key, &klen, &vstmt)) { |
---|
258 |
stmt = vstmt; |
---|
259 |
stmt->marked = 0; |
---|
260 |
} |
---|
261 |
|
---|
262 |
/* Phase 5: do the load */ |
---|
263 |
memset(&iter, 0, sizeof(iter)); |
---|
264 |
while(noit_hash_next(&stmt_by_id, &iter, &key, &klen, &vstmt)) { |
---|
265 |
stmt = vstmt; |
---|
266 |
submit_statement_node(stmt); |
---|
267 |
} |
---|
268 |
|
---|
269 |
noit_hash_destroy(&stmt_by_provider, NULL, NULL); |
---|
270 |
noit_hash_destroy(&stmt_by_id, NULL, statement_node_free); |
---|
271 |
free(statement_configs); |
---|
272 |
} |
---|
273 |
|
---|
274 |
void stratcon_iep_submit_queries() { |
---|
275 |
int i, cnt = 0; |
---|
276 |
noit_conf_section_t *query_configs; |
---|
277 |
char path[256]; |
---|
278 |
|
---|
279 |
snprintf(path, sizeof(path), "/stratcon/iep/queries//query"); |
---|
280 |
query_configs = noit_conf_get_sections(NULL, path, &cnt); |
---|
281 |
noitL(noit_debug, "Found %d %s stanzas\n", cnt, path); |
---|
282 |
for(i=0; i<cnt; i++) { |
---|
283 |
char id[UUID_STR_LEN+1]; |
---|
284 |
char topic[256]; |
---|
285 |
char *query; |
---|
286 |
char *line; |
---|
287 |
int line_len; |
---|
288 |
|
---|
289 |
if(!noit_conf_get_stringbuf(query_configs[i], |
---|
290 |
"self::node()/@id", |
---|
291 |
id, sizeof(id))) { |
---|
292 |
noitL(noit_iep, "No uuid specified in query\n"); |
---|
293 |
continue; |
---|
294 |
} |
---|
295 |
if(!noit_conf_get_stringbuf(query_configs[i], |
---|
296 |
"ancestor-or-self::node()/@topic", |
---|
297 |
topic, sizeof(topic))) { |
---|
298 |
noitL(noit_iep, "No topic specified in query\n"); |
---|
299 |
continue; |
---|
300 |
} |
---|
301 |
if(!noit_conf_get_string(query_configs[i], "self::node()/epl", |
---|
302 |
&query)) { |
---|
303 |
noitL(noit_iep, "No contents specified in query\n"); |
---|
304 |
continue; |
---|
305 |
} |
---|
306 |
line_len = 4 /* 3 tabs + \0 */ + |
---|
307 |
1 /* 'Q' */ + 1 /* '\n' */ + |
---|
308 |
strlen(id) + strlen(topic) + strlen(query); |
---|
309 |
line = malloc(line_len); |
---|
310 |
snprintf(line, line_len, "Q\t%s\t%s\t%s\n", id, topic, query); |
---|
311 |
free(query); |
---|
312 |
query = line; |
---|
313 |
while(query[0] && query[1]) { |
---|
314 |
if(*query == '\n') *query = ' '; |
---|
315 |
query++; |
---|
316 |
} |
---|
317 |
stratcon_iep_line_processor(DS_OP_INSERT, NULL, NULL, line, NULL); |
---|
318 |
} |
---|
319 |
free(query_configs); |
---|
320 |
} |
---|
321 |
|
---|
322 |
static |
---|
323 |
struct iep_thread_driver *stratcon_iep_get_connection() { |
---|
324 |
apr_status_t rc; |
---|
325 |
struct iep_thread_driver *driver; |
---|
326 |
driver = pthread_getspecific(iep_connection); |
---|
327 |
if(!driver) { |
---|
328 |
driver = calloc(1, sizeof(*driver)); |
---|
329 |
pthread_setspecific(iep_connection, driver); |
---|
330 |
} |
---|
331 |
|
---|
332 |
if(!driver->pool) { |
---|
333 |
if(apr_pool_create(&driver->pool, NULL) != APR_SUCCESS) return NULL; |
---|
334 |
} |
---|
335 |
|
---|
336 |
if(!driver->connection) { |
---|
337 |
int port; |
---|
338 |
char hostname[128]; |
---|
339 |
if(!noit_conf_get_int(NULL, "/stratcon/iep/stomp/port", &port)) |
---|
340 |
port = 61613; |
---|
341 |
if(!noit_conf_get_stringbuf(NULL, "/stratcon/iep/stomp/hostname", |
---|
342 |
hostname, sizeof(hostname))) |
---|
343 |
strlcpy(hostname, "127.0.0.1", sizeof(hostname)); |
---|
344 |
#ifdef OPENWIRE |
---|
345 |
memset(&driver->connect_options, 0, sizeof(driver->connect_options)); |
---|
346 |
strlcpy(driver->connect_options.hostname, hostname, |
---|
347 |
sizeof(driver->connect_options.hostname)); |
---|
348 |
driver->connect_options.port = port; |
---|
349 |
if(amqcs_connect(&driver->connection, &driver->connect_options, |
---|
350 |
driver->pool) != APR_SUCCESS) { |
---|
351 |
noitL(noit_error, "MQ connection failed\n"); |
---|
352 |
return NULL; |
---|
353 |
} |
---|
354 |
#else |
---|
355 |
if(stomp_connect(&driver->connection, hostname, port, |
---|
356 |
driver->pool)!= APR_SUCCESS) { |
---|
357 |
noitL(noit_error, "MQ connection failed\n"); |
---|
358 |
stomp_disconnect(&driver->connection); |
---|
359 |
return NULL; |
---|
360 |
} |
---|
361 |
|
---|
362 |
{ |
---|
363 |
stomp_frame frame; |
---|
364 |
char username[128]; |
---|
365 |
char password[128]; |
---|
366 |
char* exchange = malloc(128); |
---|
367 |
frame.command = "CONNECT"; |
---|
368 |
frame.headers = apr_hash_make(driver->pool); |
---|
369 |
// This is for RabbitMQ Support |
---|
370 |
if((noit_conf_get_stringbuf(NULL, "/stratcon/iep/stomp/username", |
---|
371 |
username, sizeof(username))) && |
---|
372 |
(noit_conf_get_stringbuf(NULL, "/stratcon/iep/stomp/password", |
---|
373 |
password, sizeof(password)))) |
---|
374 |
{ |
---|
375 |
apr_hash_set(frame.headers, "login", APR_HASH_KEY_STRING, username); |
---|
376 |
apr_hash_set(frame.headers, "passcode", APR_HASH_KEY_STRING, password); |
---|
377 |
} |
---|
378 |
|
---|
379 |
|
---|
380 |
// This is for RabbitMQ support |
---|
381 |
driver->exchange = NULL; |
---|
382 |
if(noit_conf_get_stringbuf(NULL, "/stratcon/iep/stomp/exchange", |
---|
383 |
exchange, 128)) |
---|
384 |
{ |
---|
385 |
if (!driver->exchange) |
---|
386 |
driver->exchange = exchange; |
---|
387 |
else |
---|
388 |
free(exchange); |
---|
389 |
apr_hash_set(frame.headers, "exchange", APR_HASH_KEY_STRING, driver->exchange); |
---|
390 |
} |
---|
391 |
|
---|
392 |
frame.body = NULL; |
---|
393 |
frame.body_length = -1; |
---|
394 |
rc = stomp_write(driver->connection, &frame, driver->pool); |
---|
395 |
if(rc != APR_SUCCESS) { |
---|
396 |
noitL(noit_error, "MQ STOMP CONNECT failed, %d\n", rc); |
---|
397 |
stomp_disconnect(&driver->connection); |
---|
398 |
return NULL; |
---|
399 |
} |
---|
400 |
} |
---|
401 |
#endif |
---|
402 |
stratcon_iep_submit_statements(); |
---|
403 |
stratcon_datastore_iep_check_preload(); |
---|
404 |
stratcon_iep_submit_queries(); |
---|
405 |
} |
---|
406 |
|
---|
407 |
return driver; |
---|
408 |
} |
---|
409 |
|
---|
410 |
static int |
---|
411 |
setup_iep_connection_callback(eventer_t e, int mask, void *closure, |
---|
412 |
struct timeval *now) { |
---|
413 |
noit_spinlock_unlock(&iep_conn_cnt); |
---|
414 |
stratcon_iep_line_processor(DS_OP_INSERT, NULL, NULL, NULL, NULL); |
---|
415 |
return 0; |
---|
416 |
} |
---|
417 |
|
---|
418 |
static void |
---|
419 |
setup_iep_connection_later(int seconds) { |
---|
420 |
eventer_t newe; |
---|
421 |
if(!noit_spinlock_trylock(&iep_conn_cnt)) return; |
---|
422 |
newe = eventer_alloc(); |
---|
423 |
gettimeofday(&newe->whence, NULL); |
---|
424 |
newe->whence.tv_sec += seconds; |
---|
425 |
newe->mask = EVENTER_TIMER; |
---|
426 |
newe->callback = setup_iep_connection_callback; |
---|
427 |
newe->closure = NULL; |
---|
428 |
eventer_add(newe); |
---|
429 |
} |
---|
430 |
|
---|
431 |
static int |
---|
432 |
stratcon_iep_submitter(eventer_t e, int mask, void *closure, |
---|
433 |
struct timeval *now) { |
---|
434 |
float age; |
---|
435 |
struct iep_job_closure *job = closure; |
---|
436 |
struct iep_thread_driver *driver; |
---|
437 |
/* We only play when it is an asynch event */ |
---|
438 |
if(!(mask & EVENTER_ASYNCH_WORK)) return 0; |
---|
439 |
|
---|
440 |
if(mask & EVENTER_ASYNCH_CLEANUP) { |
---|
441 |
/* free all the memory associated with the batch */ |
---|
442 |
if(job) { |
---|
443 |
if(job->line) free(job->line); |
---|
444 |
if(job->remote) free(job->remote); |
---|
445 |
if(job->doc_str) free(job->doc_str); |
---|
446 |
if(job->pool) apr_pool_destroy(job->pool); |
---|
447 |
free(job); |
---|
448 |
} |
---|
449 |
return 0; |
---|
450 |
} |
---|
451 |
driver = stratcon_iep_get_connection(); |
---|
452 |
if(!driver) setup_iep_connection_later(1); |
---|
453 |
|
---|
454 |
if(!job->line || job->line[0] == '\0') return 0; |
---|
455 |
|
---|
456 |
if((age = stratcon_iep_age_from_line(job->line, *now)) > 60) { |
---|
457 |
noitL(noit_debug, "Skipping old event %f second old.\n", age); |
---|
458 |
return 0; |
---|
459 |
} |
---|
460 |
/* Submit */ |
---|
461 |
if(driver && driver->pool && driver->connection) { |
---|
462 |
apr_status_t rc; |
---|
463 |
int line_len = strlen(job->line); |
---|
464 |
int remote_len = strlen(job->remote); |
---|
465 |
#ifdef OPENWIRE |
---|
466 |
ow_ActiveMQQueue *dest; |
---|
467 |
ow_ActiveMQTextMessage *message; |
---|
468 |
|
---|
469 |
apr_pool_create(&job->pool, driver->pool); |
---|
470 |
message = ow_ActiveMQTextMessage_create(job->pool); |
---|
471 |
message->content = |
---|
472 |
ow_byte_array_create_with_data(job->pool,strlen(job->doc_str), |
---|
473 |
job->doc_str); |
---|
474 |
dest = ow_ActiveMQQueue_create(job->pool); |
---|
475 |
dest->physicalName = ow_string_create_from_cstring(job->pool,"TEST.QUEUE"); |
---|
476 |
rc = amqcs_send(driver->connection, |
---|
477 |
(ow_ActiveMQDestination*)dest, |
---|
478 |
(ow_ActiveMQMessage*)message, |
---|
479 |
1,4,0,job->pool); |
---|
480 |
if(rc != APR_SUCCESS) { |
---|
481 |
noitL(noit_error, "MQ send failed, disconnecting\n"); |
---|
482 |
if(driver->connection) amqcs_disconnect(&driver->connection); |
---|
483 |
driver->connection = NULL; |
---|
484 |
} |
---|
485 |
#else |
---|
486 |
stomp_frame out; |
---|
487 |
|
---|
488 |
job->doc_str = (char*)calloc(line_len + 1 /* \t */ + |
---|
489 |
remote_len + 2, 1); |
---|
490 |
strncpy(job->doc_str, job->line, 2); |
---|
491 |
strncat(job->doc_str, job->remote, remote_len); |
---|
492 |
strncat(job->doc_str, "\t", 1); |
---|
493 |
strncat(job->doc_str, job->line + 2, line_len - 2); |
---|
494 |
|
---|
495 |
apr_pool_create(&job->pool, driver->pool); |
---|
496 |
|
---|
497 |
out.command = "SEND"; |
---|
498 |
out.headers = apr_hash_make(job->pool); |
---|
499 |
if (driver->exchange) |
---|
500 |
apr_hash_set(out.headers, "exchange", APR_HASH_KEY_STRING, driver->exchange); |
---|
501 |
|
---|
502 |
apr_hash_set(out.headers, "destination", APR_HASH_KEY_STRING, "/queue/noit.firehose"); |
---|
503 |
apr_hash_set(out.headers, "ack", APR_HASH_KEY_STRING, "auto"); |
---|
504 |
|
---|
505 |
out.body_length = -1; |
---|
506 |
out.body = job->doc_str; |
---|
507 |
rc = stomp_write(driver->connection, &out, job->pool); |
---|
508 |
if(rc != APR_SUCCESS) { |
---|
509 |
noitL(noit_error, "STOMP send failed, disconnecting\n"); |
---|
510 |
if(driver->connection) stomp_disconnect(&driver->connection); |
---|
511 |
driver->connection = NULL; |
---|
512 |
} |
---|
513 |
#endif |
---|
514 |
} |
---|
515 |
else { |
---|
516 |
noitL(noit_iep, "no iep connection, skipping line: '%s'\n", job->line); |
---|
517 |
} |
---|
518 |
return 0; |
---|
519 |
} |
---|
520 |
|
---|
521 |
void |
---|
522 |
stratcon_iep_line_processor(stratcon_datastore_op_t op, |
---|
523 |
struct sockaddr *remote, const char *remote_cn, |
---|
524 |
void *operand, eventer_t completion) { |
---|
525 |
int len; |
---|
526 |
char remote_str[128]; |
---|
527 |
struct iep_job_closure *jc; |
---|
528 |
eventer_t newe; |
---|
529 |
struct timeval __now, iep_timeout = { 20L, 0L }; |
---|
530 |
/* We only care about inserts */ |
---|
531 |
|
---|
532 |
if(op == DS_OP_CHKPT) { |
---|
533 |
eventer_add(completion); |
---|
534 |
return; |
---|
535 |
} |
---|
536 |
if(op != DS_OP_INSERT) return; |
---|
537 |
|
---|
538 |
snprintf(remote_str, sizeof(remote_str), "%s", "0.0.0.0"); |
---|
539 |
if(remote) { |
---|
540 |
switch(remote->sa_family) { |
---|
541 |
case AF_INET: |
---|
542 |
len = sizeof(struct sockaddr_in); |
---|
543 |
inet_ntop(remote->sa_family, &((struct sockaddr_in *)remote)->sin_addr, |
---|
544 |
remote_str, len); |
---|
545 |
break; |
---|
546 |
case AF_INET6: |
---|
547 |
len = sizeof(struct sockaddr_in6); |
---|
548 |
inet_ntop(remote->sa_family, &((struct sockaddr_in6 *)remote)->sin6_addr, |
---|
549 |
remote_str, len); |
---|
550 |
break; |
---|
551 |
case AF_UNIX: |
---|
552 |
len = SUN_LEN(((struct sockaddr_un *)remote)); |
---|
553 |
snprintf(remote_str, sizeof(remote_str), "%s", ((struct sockaddr_un *)remote)->sun_path); |
---|
554 |
break; |
---|
555 |
} |
---|
556 |
} |
---|
557 |
|
---|
558 |
/* process operand and push onto queue */ |
---|
559 |
gettimeofday(&__now, NULL); |
---|
560 |
newe = eventer_alloc(); |
---|
561 |
newe->mask = EVENTER_ASYNCH; |
---|
562 |
add_timeval(__now, iep_timeout, &newe->whence); |
---|
563 |
newe->callback = stratcon_iep_submitter; |
---|
564 |
jc = calloc(1, sizeof(*jc)); |
---|
565 |
jc->line = operand; |
---|
566 |
jc->remote = strdup(remote_str); |
---|
567 |
newe->closure = jc; |
---|
568 |
|
---|
569 |
eventer_add_asynch(&iep_jobq, newe); |
---|
570 |
} |
---|
571 |
|
---|
572 |
static void connection_destroy(void *vd) { |
---|
573 |
struct iep_thread_driver *driver = vd; |
---|
574 |
#ifdef OPENWIRE |
---|
575 |
if(driver->connection) amqcs_disconnect(&driver->connection); |
---|
576 |
#else |
---|
577 |
if(driver->connection) stomp_disconnect(&driver->connection); |
---|
578 |
if(driver->exchange) free(driver->exchange); |
---|
579 |
#endif |
---|
580 |
if(driver->pool) apr_pool_destroy(driver->pool); |
---|
581 |
free(driver); |
---|
582 |
} |
---|
583 |
|
---|
584 |
jlog_streamer_ctx_t * |
---|
585 |
stratcon_jlog_streamer_iep_ctx_alloc(void) { |
---|
586 |
jlog_streamer_ctx_t *ctx; |
---|
587 |
ctx = stratcon_jlog_streamer_ctx_alloc(); |
---|
588 |
ctx->jlog_feed_cmd = htonl(NOIT_JLOG_DATA_TEMP_FEED); |
---|
589 |
ctx->push = stratcon_iep_line_processor; |
---|
590 |
return ctx; |
---|
591 |
} |
---|
592 |
|
---|
593 |
struct iep_daemon_info { |
---|
594 |
pid_t child; |
---|
595 |
int stdin_pipe[2]; |
---|
596 |
int stderr_pipe[2]; |
---|
597 |
char *directory; |
---|
598 |
char *command; |
---|
599 |
}; |
---|
600 |
|
---|
601 |
static void |
---|
602 |
iep_daemon_info_free(struct iep_daemon_info *info) { |
---|
603 |
if(!info) return; |
---|
604 |
if(info->directory) free(info->directory); |
---|
605 |
if(info->command) free(info->command); |
---|
606 |
if(info->stdin_pipe[0] >= 0) close(info->stdin_pipe[0]); |
---|
607 |
if(info->stdin_pipe[1] >= 0) close(info->stdin_pipe[1]); |
---|
608 |
if(info->stderr_pipe[0] >= 0) close(info->stderr_pipe[0]); |
---|
609 |
if(info->stderr_pipe[1] >= 0) close(info->stderr_pipe[1]); |
---|
610 |
free(info); |
---|
611 |
} |
---|
612 |
|
---|
613 |
static int |
---|
614 |
stratcon_iep_err_handler(eventer_t e, int mask, void *closure, |
---|
615 |
struct timeval *now) { |
---|
616 |
int len, newmask; |
---|
617 |
char buff[4096]; |
---|
618 |
struct iep_daemon_info *info = (struct iep_daemon_info *)closure; |
---|
619 |
|
---|
620 |
if(mask & EVENTER_EXCEPTION) { |
---|
621 |
int rv; |
---|
622 |
read_error: |
---|
623 |
kill(info->child, SIGKILL); |
---|
624 |
if(waitpid(info->child, &rv, 0) != info->child) { |
---|
625 |
noitL(noit_error, "Failed to reap IEP daemon\n"); |
---|
626 |
exit(-1); |
---|
627 |
} |
---|
628 |
noitL(noit_error, "IEP daemon is done, starting a new one\n"); |
---|
629 |
start_iep_daemon(); |
---|
630 |
eventer_remove_fd(e->fd); |
---|
631 |
iep_daemon_info_free(info); |
---|
632 |
return 0; |
---|
633 |
} |
---|
634 |
while(1) { |
---|
635 |
len = e->opset->read(e->fd, buff, sizeof(buff)-1, &newmask, e); |
---|
636 |
if(len == -1 && (errno == EAGAIN || errno == EINTR)) |
---|
637 |
return newmask | EVENTER_EXCEPTION; |
---|
638 |
if(len <= 0) goto read_error; |
---|
639 |
assert(len < sizeof(buff)); |
---|
640 |
buff[len] = '\0'; |
---|
641 |
noitL(noit_iep, "%s", buff); |
---|
642 |
} |
---|
643 |
} |
---|
644 |
|
---|
645 |
static void |
---|
646 |
start_iep_daemon() { |
---|
647 |
eventer_t newe; |
---|
648 |
struct iep_daemon_info *info; |
---|
649 |
|
---|
650 |
info = calloc(1, sizeof(*info)); |
---|
651 |
info->stdin_pipe[0] = info->stdin_pipe[1] = -1; |
---|
652 |
info->stderr_pipe[0] = info->stderr_pipe[1] = -1; |
---|
653 |
|
---|
654 |
if(!noit_conf_get_string(NULL, "/stratcon/iep/start/@directory", |
---|
655 |
&info->directory)) |
---|
656 |
info->directory = strdup("."); |
---|
657 |
if(!noit_conf_get_string(NULL, "/stratcon/iep/start/@command", |
---|
658 |
&info->command)) { |
---|
659 |
noitL(noit_error, "No IEP start command provided. You're on your own.\n"); |
---|
660 |
setup_iep_connection_later(0); |
---|
661 |
return; |
---|
662 |
} |
---|
663 |
if(pipe(info->stdin_pipe) != 0 || |
---|
664 |
pipe(info->stderr_pipe) != 0) { |
---|
665 |
noitL(noit_error, "pipe: %s\n", strerror(errno)); |
---|
666 |
goto bail; |
---|
667 |
} |
---|
668 |
info->child = fork(); |
---|
669 |
if(info->child == -1) { |
---|
670 |
noitL(noit_error, "fork: %s\n", strerror(errno)); |
---|
671 |
goto bail; |
---|
672 |
} |
---|
673 |
if(info->child == 0) { |
---|
674 |
char *argv[3] = { "run-iep", NULL, NULL }; |
---|
675 |
int stdout_fileno; |
---|
676 |
|
---|
677 |
argv[1] = noit_conf_config_filename(); |
---|
678 |
|
---|
679 |
if(chdir(info->directory) != 0) { |
---|
680 |
noitL(noit_error, "Starting IEP daemon, chdir failed: %s\n", |
---|
681 |
strerror(errno)); |
---|
682 |
exit(-1); |
---|
683 |
} |
---|
684 |
|
---|
685 |
close(info->stdin_pipe[1]); |
---|
686 |
close(info->stderr_pipe[0]); |
---|
687 |
dup2(info->stdin_pipe[0], 0); |
---|
688 |
dup2(info->stderr_pipe[1], 2); |
---|
689 |
stdout_fileno = open("/dev/null", O_WRONLY); |
---|
690 |
dup2(stdout_fileno, 1); |
---|
691 |
|
---|
692 |
exit(execv(info->command, argv)); |
---|
693 |
} |
---|
694 |
/* in the parent */ |
---|
695 |
close(info->stdin_pipe[0]); |
---|
696 |
info->stdin_pipe[0] = -1; |
---|
697 |
close(info->stderr_pipe[1]); |
---|
698 |
info->stderr_pipe[1] = -1; |
---|
699 |
if(eventer_set_fd_nonblocking(info->stderr_pipe[0])) { |
---|
700 |
goto bail; |
---|
701 |
} |
---|
702 |
|
---|
703 |
newe = eventer_alloc(); |
---|
704 |
newe->fd = info->stderr_pipe[0]; |
---|
705 |
newe->mask = EVENTER_READ | EVENTER_EXCEPTION; |
---|
706 |
newe->callback = stratcon_iep_err_handler; |
---|
707 |
newe->closure = info; |
---|
708 |
eventer_add(newe); |
---|
709 |
info = NULL; |
---|
710 |
|
---|
711 |
/* This will induce a stomp connection which will initialize esper */ |
---|
712 |
setup_iep_connection_later(1); |
---|
713 |
|
---|
714 |
return; |
---|
715 |
|
---|
716 |
bail: |
---|
717 |
if(info) { |
---|
718 |
iep_daemon_info_free(info); |
---|
719 |
} |
---|
720 |
noitL(noit_error, "Failed to start IEP daemon\n"); |
---|
721 |
exit(-1); |
---|
722 |
return; |
---|
723 |
} |
---|
724 |
|
---|
725 |
void |
---|
726 |
stratcon_iep_init() { |
---|
727 |
noit_boolean disabled = noit_false; |
---|
728 |
apr_initialize(); |
---|
729 |
atexit(apr_terminate); |
---|
730 |
|
---|
731 |
if(noit_conf_get_boolean(NULL, "/stratcon/iep/@disabled", &disabled) && |
---|
732 |
disabled == noit_true) { |
---|
733 |
noitL(noit_error, "IEP system is disabled!\n"); |
---|
734 |
return; |
---|
735 |
} |
---|
736 |
|
---|
737 |
noit_iep = noit_log_stream_find("error/iep"); |
---|
738 |
if(!noit_iep) noit_iep = noit_error; |
---|
739 |
|
---|
740 |
eventer_name_callback("stratcon_iep_submitter", stratcon_iep_submitter); |
---|
741 |
eventer_name_callback("setup_iep_connection_callback", setup_iep_connection_callback); |
---|
742 |
pthread_key_create(&iep_connection, connection_destroy); |
---|
743 |
|
---|
744 |
/* start up a thread pool of one */ |
---|
745 |
memset(&iep_jobq, 0, sizeof(iep_jobq)); |
---|
746 |
eventer_jobq_init(&iep_jobq, "iep_submitter"); |
---|
747 |
iep_jobq.backq = eventer_default_backq(); |
---|
748 |
eventer_jobq_increase_concurrency(&iep_jobq); |
---|
749 |
|
---|
750 |
start_iep_daemon(); |
---|
751 |
|
---|
752 |
/* setup our live jlog stream */ |
---|
753 |
stratcon_streamer_connection(NULL, NULL, |
---|
754 |
stratcon_jlog_recv_handler, |
---|
755 |
(void *(*)())stratcon_jlog_streamer_iep_ctx_alloc, |
---|
756 |
NULL, |
---|
757 |
jlog_streamer_ctx_free); |
---|
758 |
} |
---|
759 |
|
---|