root/src/stratcon_jlog_streamer.c

Revision 8e9cf57ebf8fd9d4375a893d47cef1aaecce0662, 59.0 kB (checked in by Theo Schlossnagle <jesus@omniti.com>, 2 years ago)

on-disk backing store implementation

  • 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 "dtrace_probes.h"
35 #include "eventer/eventer.h"
36 #include "noit_conf.h"
37 #include "utils/noit_hash.h"
38 #include "utils/noit_log.h"
39 #include "utils/noit_getip.h"
40 #include "noit_jlog_listener.h"
41 #include "noit_rest.h"
42 #include "stratcon_datastore.h"
43 #include "stratcon_jlog_streamer.h"
44 #include "stratcon_iep.h"
45
46 #include <unistd.h>
47 #include <assert.h>
48 #include <errno.h>
49 #include <sys/types.h>
50 #include <sys/socket.h>
51 #ifdef HAVE_SYS_FILIO_H
52 #include <sys/filio.h>
53 #endif
54 #include <netinet/in.h>
55 #include <sys/un.h>
56 #include <arpa/inet.h>
57
58 pthread_mutex_t noits_lock;
59 noit_hash_table noits = NOIT_HASH_EMPTY;
60 pthread_mutex_t noit_ip_by_cn_lock;
61 noit_hash_table noit_ip_by_cn = NOIT_HASH_EMPTY;
62 static uuid_t self_stratcon_id;
63 static char self_stratcon_hostname[256] = "\0";
64 static struct sockaddr_in self_stratcon_ip;
65 static noit_boolean stratcon_selfcheck_extended_id = noit_true;
66
67 static struct timeval DEFAULT_NOIT_PERIOD_TV = { 5UL, 0UL };
68
69 static void noit_connection_initiate_connection(noit_connection_ctx_t *ctx);
70
71 static const char *feed_type_to_str(int jlog_feed_cmd) {
72   switch(jlog_feed_cmd) {
73     case NOIT_JLOG_DATA_FEED: return "durable/storage";
74     case NOIT_JLOG_DATA_TEMP_FEED: return "transient/iep";
75   }
76   return "unknown";
77 }
78
79 #define GET_EXPECTED_CN(nctx, cn) do { \
80   void *vcn; \
81   cn = NULL; \
82   if(nctx->config && \
83      noit_hash_retrieve(nctx->config, "cn", 2, &vcn)) { \
84      cn = vcn; \
85   } \
86 } while(0)
87 #define GET_FEEDTYPE(nctx, feedtype) do { \
88   jlog_streamer_ctx_t *_jctx = nctx->consumer_ctx; \
89   feedtype = "unknown"; \
90   if(_jctx->push == stratcon_datastore_push) \
91     feedtype = "storage"; \
92   else if(_jctx->push == stratcon_iep_line_processor) \
93     feedtype = "iep"; \
94 } while(0)
95
96 static int
97 remote_str_sort(const void *a, const void *b) {
98   int rv;
99   noit_connection_ctx_t * const *actx = a;
100   noit_connection_ctx_t * const *bctx = b;
101   jlog_streamer_ctx_t *ajctx = (*actx)->consumer_ctx;
102   jlog_streamer_ctx_t *bjctx = (*bctx)->consumer_ctx;
103   rv = strcmp((*actx)->remote_str, (*bctx)->remote_str);
104   if(rv) return rv;
105   return (ajctx->jlog_feed_cmd < bjctx->jlog_feed_cmd) ? -1 :
106            ((ajctx->jlog_feed_cmd == bjctx->jlog_feed_cmd) ? 0 : 1);
107 }
108 static void
109 nc_print_noit_conn_brief(noit_console_closure_t ncct,
110                           noit_connection_ctx_t *ctx) {
111   jlog_streamer_ctx_t *jctx = ctx->consumer_ctx;
112   struct timeval now, diff, session_duration;
113   const char *feedtype = "unknown";
114   const char *lasttime = "never";
115   if(ctx->last_connect.tv_sec != 0) {
116     char cmdbuf[4096];
117     time_t r = ctx->last_connect.tv_sec;
118     struct tm tbuf, *tm;
119     tm = gmtime_r(&r, &tbuf);
120     strftime(cmdbuf, sizeof(cmdbuf), "%Y-%m-%d %H:%M:%S UTC", tm);
121     lasttime = cmdbuf;
122   }
123   nc_printf(ncct, "%s [%s]:\n\tLast connect: %s\n", ctx->remote_str,
124             ctx->remote_cn ? "connected" :
125                              (ctx->retry_event ? "disconnected" :
126                                                    "connecting"), lasttime);
127   if(ctx->e) {
128     char buff[128];
129     const char *addrstr = NULL;
130     struct sockaddr_in6 addr6;
131     socklen_t len = sizeof(addr6);
132     if(getsockname(ctx->e->fd, (struct sockaddr *)&addr6, &len) == 0) {
133       unsigned short port = 0;
134       if(addr6.sin6_family == AF_INET) {
135         addrstr = inet_ntop(addr6.sin6_family,
136                             &((struct sockaddr_in *)&addr6)->sin_addr,
137                             buff, sizeof(buff));
138         memcpy(&port, &(&addr6)->sin6_port, sizeof(port));
139         port = ntohs(port);
140       }
141       else if(addr6.sin6_family == AF_INET6) {
142         addrstr = inet_ntop(addr6.sin6_family, &addr6.sin6_addr,
143                             buff, sizeof(buff));
144         port = ntohs(addr6.sin6_port);
145       }
146       if(addrstr != NULL)
147         nc_printf(ncct, "\tLocal address is %s:%u\n", buff, port);
148       else
149         nc_printf(ncct, "\tLocal address not interpretable\n");
150     }
151     else {
152       nc_printf(ncct, "\tLocal address error[%d]: %s\n",
153                 ctx->e->fd, strerror(errno));
154     }
155   }
156   feedtype = feed_type_to_str(ntohl(jctx->jlog_feed_cmd));
157   nc_printf(ncct, "\tJLog event streamer [%s]\n", feedtype);
158   gettimeofday(&now, NULL);
159   if(ctx->retry_event) {
160     sub_timeval(ctx->retry_event->whence, now, &diff);
161     nc_printf(ncct, "\tNext attempt in %lld.%06us\n",
162               (long long)diff.tv_sec, (unsigned int) diff.tv_usec);
163   }
164   else if(ctx->remote_cn) {
165     nc_printf(ncct, "\tRemote CN: '%s'\n",
166               ctx->remote_cn ? ctx->remote_cn : "???");
167     if(ctx->consumer_callback == stratcon_jlog_recv_handler) {
168       struct timeval last;
169       double session_duration_seconds;
170       const char *state = "unknown";
171
172       switch(jctx->state) {
173         case JLOG_STREAMER_WANT_INITIATE: state = "initiate"; break;
174         case JLOG_STREAMER_WANT_COUNT: state = "waiting for next batch"; break;
175         case JLOG_STREAMER_WANT_ERROR: state = "waiting for error"; break;
176         case JLOG_STREAMER_WANT_HEADER: state = "reading header"; break;
177         case JLOG_STREAMER_WANT_BODY: state = "reading body"; break;
178         case JLOG_STREAMER_IS_ASYNC: state = "asynchronously processing"; break;
179         case JLOG_STREAMER_WANT_CHKPT: state = "checkpointing"; break;
180       }
181       last.tv_sec = jctx->header.tv_sec;
182       last.tv_usec = jctx->header.tv_usec;
183       sub_timeval(now, last, &diff);
184       sub_timeval(now, ctx->last_connect, &session_duration);
185       session_duration_seconds = session_duration.tv_sec +
186                                  (double)session_duration.tv_usec/1000000.0;
187       nc_printf(ncct, "\tState: %s\n"
188                       "\tNext checkpoint: [%08x:%08x]\n"
189                       "\tLast event: %lld.%06us ago\n"
190                       "\tEvents this session: %llu (%0.2f/s)\n"
191                       "\tOctets this session: %llu (%0.2f/s)\n",
192                 state,
193                 jctx->header.chkpt.log, jctx->header.chkpt.marker,
194                 (long long)diff.tv_sec, (unsigned int)diff.tv_usec,
195                 jctx->total_events,
196                 (double)jctx->total_events/session_duration_seconds,
197                 jctx->total_bytes_read,
198                 (double)jctx->total_bytes_read/session_duration_seconds);
199     }
200     else {
201       nc_printf(ncct, "\tUnknown type.\n");
202     }
203   }
204 }
205
206 jlog_streamer_ctx_t *
207 stratcon_jlog_streamer_datastore_ctx_alloc(void) {
208   jlog_streamer_ctx_t *ctx;
209   ctx = stratcon_jlog_streamer_ctx_alloc();
210   ctx->jlog_feed_cmd = htonl(NOIT_JLOG_DATA_FEED);
211   ctx->push = stratcon_datastore_push;
212   return ctx;
213 }
214 jlog_streamer_ctx_t *
215 stratcon_jlog_streamer_ctx_alloc(void) {
216   jlog_streamer_ctx_t *ctx;
217   ctx = calloc(1, sizeof(*ctx));
218   return ctx;
219 }
220 noit_connection_ctx_t *
221 noit_connection_ctx_alloc(void) {
222   noit_connection_ctx_t *ctx, **pctx;
223   ctx = calloc(1, sizeof(*ctx));
224   ctx->refcnt = 1;
225   pctx = malloc(sizeof(*pctx));
226   *pctx = ctx;
227   pthread_mutex_lock(&noits_lock);
228   noit_hash_store(&noits, (const char *)pctx, sizeof(*pctx), ctx);
229   pthread_mutex_unlock(&noits_lock);
230   return ctx;
231 }
232 int
233 noit_connection_reinitiate(eventer_t e, int mask, void *closure,
234                          struct timeval *now) {
235   noit_connection_ctx_t *ctx = closure;
236   ctx->retry_event = NULL;
237   noit_connection_initiate_connection(closure);
238   return 0;
239 }
240 void
241 noit_connection_schedule_reattempt(noit_connection_ctx_t *ctx,
242                                    struct timeval *now) {
243   struct timeval __now, interval;
244   const char *v, *feedtype, *cn_expected;
245   u_int32_t min_interval = 1000, max_interval = 8000;
246
247   noit_connection_disable_timeout(ctx);
248   if(ctx->remote_cn) {
249     free(ctx->remote_cn);
250     ctx->remote_cn = NULL;
251   }
252   if(noit_hash_retr_str(ctx->config,
253                         "reconnect_initial_interval",
254                         strlen("reconnect_initial_interval"),
255                         &v)) {
256     min_interval = MAX(atoi(v), 100); /* .1 second minimum */
257   }
258   if(noit_hash_retr_str(ctx->config,
259                         "reconnect_maximum_interval",
260                         strlen("reconnect_maximum_interval"),
261                         &v)) {
262     max_interval = MIN(atoi(v), 3600*1000); /* 1 hour maximum */
263   }
264   if(ctx->current_backoff == 0) ctx->current_backoff = min_interval;
265   else {
266     ctx->current_backoff *= 2;
267     ctx->current_backoff = MAX(min_interval, ctx->current_backoff);
268     ctx->current_backoff = MIN(max_interval, ctx->current_backoff);
269   }
270   if(!now) {
271     gettimeofday(&__now, NULL);
272     now = &__now;
273   }
274   interval.tv_sec = ctx->current_backoff / 1000;
275   interval.tv_usec = (ctx->current_backoff % 1000) * 1000;
276   noitL(noit_debug, "Next jlog_streamer attempt in %ums\n",
277         ctx->current_backoff);
278   if(ctx->retry_event)
279     eventer_remove(ctx->retry_event);
280   else
281     ctx->retry_event = eventer_alloc();
282   ctx->retry_event->callback = noit_connection_reinitiate;
283   ctx->retry_event->closure = ctx;
284   ctx->retry_event->mask = EVENTER_TIMER;
285   add_timeval(*now, interval, &ctx->retry_event->whence);
286   GET_EXPECTED_CN(ctx, cn_expected);
287   GET_FEEDTYPE(ctx, feedtype);
288   STRATCON_NOIT_RESCHEDULE(-1, (char *)feedtype, ctx->remote_str,
289                            (char *)cn_expected, ctx->current_backoff);
290   eventer_add(ctx->retry_event);
291 }
292 static void
293 noit_connection_ctx_free(noit_connection_ctx_t *ctx) {
294   if(ctx->remote_cn) free(ctx->remote_cn);
295   if(ctx->remote_str) free(ctx->remote_str);
296   if(ctx->retry_event) {
297     eventer_remove(ctx->retry_event);
298     eventer_free(ctx->retry_event);
299   }
300   if(ctx->timeout_event) {
301     eventer_remove(ctx->timeout_event);
302     eventer_free(ctx->timeout_event);
303   }
304   ctx->consumer_free(ctx->consumer_ctx);
305   free(ctx);
306 }
307 void
308 noit_connection_ctx_deref(noit_connection_ctx_t *ctx) {
309   if(noit_atomic_dec32(&ctx->refcnt) == 0)
310     noit_connection_ctx_free(ctx);
311 }
312 void
313 noit_connection_ctx_dealloc(noit_connection_ctx_t *ctx) {
314   noit_connection_ctx_t **pctx = &ctx;
315   pthread_mutex_lock(&noits_lock);
316   noit_hash_delete(&noits, (const char *)pctx, sizeof(*pctx),
317                    free, (void (*)(void *))noit_connection_ctx_deref);
318   pthread_mutex_unlock(&noits_lock);
319 }
320 void
321 jlog_streamer_ctx_free(void *cl) {
322   jlog_streamer_ctx_t *ctx = cl;
323   if(ctx->buffer) free(ctx->buffer);
324   free(ctx);
325 }
326
327 #define Eread(a,b) e->opset->read(e->fd, (a), (b), &mask, e)
328 static int
329 __read_on_ctx(eventer_t e, jlog_streamer_ctx_t *ctx, int *newmask) {
330   int len, mask;
331   while(ctx->bytes_read < ctx->bytes_expected) {
332     len = Eread(ctx->buffer + ctx->bytes_read,
333                 ctx->bytes_expected - ctx->bytes_read);
334     if(len < 0) {
335       *newmask = mask;
336       return -1;
337     }
338     /* if we get 0 inside SSL, and there was a real error, we
339      * will actually get a -1 here.
340      * if(len == 0) return ctx->bytes_read;
341      */
342     ctx->total_bytes_read += len;
343     ctx->bytes_read += len;
344   }
345   assert(ctx->bytes_read == ctx->bytes_expected);
346   return ctx->bytes_read;
347 }
348 #define FULLREAD(e,ctx,size) do { \
349   int mask, len; \
350   if(!ctx->bytes_expected) { \
351     ctx->bytes_expected = size; \
352     if(ctx->buffer) free(ctx->buffer); \
353     ctx->buffer = malloc(size + 1); \
354     if(ctx->buffer == NULL) { \
355       noitL(noit_error, "malloc(%lu) failed.\n", (long unsigned int)size + 1); \
356       goto socket_error; \
357     } \
358     ctx->buffer[size] = '\0'; \
359   } \
360   len = __read_on_ctx(e, ctx, &mask); \
361   if(len < 0) { \
362     if(errno == EAGAIN) return mask | EVENTER_EXCEPTION; \
363     noitL(noit_error, "[%s] SSL read error: %s\n", nctx->remote_str, strerror(errno)); \
364     goto socket_error; \
365   } \
366   ctx->bytes_read = 0; \
367   ctx->bytes_expected = 0; \
368   if(len != size) { \
369     noitL(noit_error, "[%s] SSL short read [%d] (%d/%lu).  Reseting connection.\n", \
370           nctx->remote_str, ctx->state, len, (long unsigned int)size); \
371     goto socket_error; \
372   } \
373 } while(0)
374
375 int
376 noit_connection_session_timeout(eventer_t e, int mask, void *closure,
377                                 struct timeval *now) {
378   noit_connection_ctx_t *nctx = closure;
379   eventer_t fde = nctx->e;
380   nctx->timeout_event = NULL;
381   noitL(noit_error, "Timing out jlog session: %s\n",
382         nctx->remote_cn ? nctx->remote_cn : "(null)");
383   if(fde)
384     eventer_trigger(fde, EVENTER_EXCEPTION);
385   return 0;
386 }
387 int
388 stratcon_jlog_recv_handler(eventer_t e, int mask, void *closure,
389                            struct timeval *now) {
390   noit_connection_ctx_t *nctx = closure;
391   jlog_streamer_ctx_t *ctx = nctx->consumer_ctx;
392   jlog_streamer_ctx_t dummy;
393   int len;
394   jlog_id n_chkpt;
395   const char *cn_expected, *feedtype;
396   GET_EXPECTED_CN(nctx, cn_expected);
397   GET_FEEDTYPE(nctx, feedtype);
398
399   if(mask & EVENTER_EXCEPTION || nctx->wants_shutdown) {
400     if(write(e->fd, e, 0) == -1)
401       noitL(noit_error, "socket error: %s\n", strerror(errno));
402  socket_error:
403     STRATCON_NOIT_CONNECT_CLOSE(e->fd, (char *)feedtype, nctx->remote_str,
404                                 (char *)cn_expected,
405                                 nctx->wants_shutdown, errno);
406     ctx->state = JLOG_STREAMER_WANT_INITIATE;
407     ctx->count = 0;
408     ctx->needs_chkpt = 0;
409     ctx->bytes_read = 0;
410     ctx->bytes_expected = 0;
411     if(ctx->buffer) free(ctx->buffer);
412     ctx->buffer = NULL;
413     noit_connection_schedule_reattempt(nctx, now);
414     eventer_remove_fd(e->fd);
415     nctx->e = NULL;
416     e->opset->close(e->fd, &mask, e);
417     return 0;
418   }
419
420   noit_connection_update_timeout(nctx);
421   while(1) {
422     switch(ctx->state) {
423       case JLOG_STREAMER_WANT_INITIATE:
424         len = e->opset->write(e->fd, &ctx->jlog_feed_cmd,
425                               sizeof(ctx->jlog_feed_cmd),
426                               &mask, e);
427         if(len < 0) {
428           if(errno == EAGAIN) return mask | EVENTER_EXCEPTION;
429           goto socket_error;
430         }
431         if(len != sizeof(ctx->jlog_feed_cmd)) {
432           noitL(noit_error, "short write [%d/%d] on initiating stream.\n",
433                 (int)len, (int)sizeof(ctx->jlog_feed_cmd));
434           goto socket_error;
435         }
436         ctx->state = JLOG_STREAMER_WANT_COUNT;
437         break;
438
439       case JLOG_STREAMER_WANT_ERROR:
440         FULLREAD(e, ctx, 0 - ctx->count);
441         noitL(noit_error, "[%s] %.*s\n", nctx->remote_str,
442               0 - ctx->count, ctx->buffer);
443         free(ctx->buffer); ctx->buffer = NULL;
444         goto socket_error;
445         break;
446
447       case JLOG_STREAMER_WANT_COUNT:
448         FULLREAD(e, ctx, sizeof(u_int32_t));
449         memcpy(&dummy.count, ctx->buffer, sizeof(u_int32_t));
450         ctx->count = ntohl(dummy.count);
451         ctx->needs_chkpt = 0;
452         free(ctx->buffer); ctx->buffer = NULL;
453         STRATCON_NOIT_STREAM_COUNT(e->fd, (char *)feedtype,
454                                    nctx->remote_str, (char *)cn_expected,
455                                    ctx->count);
456         if(ctx->count < 0)
457           ctx->state = JLOG_STREAMER_WANT_ERROR;
458         else
459           ctx->state = JLOG_STREAMER_WANT_HEADER;
460         break;
461
462       case JLOG_STREAMER_WANT_HEADER:
463         if(ctx->count == 0) {
464           ctx->state = JLOG_STREAMER_WANT_COUNT;
465           break;
466         }
467         FULLREAD(e, ctx, sizeof(ctx->header));
468         memcpy(&dummy.header, ctx->buffer, sizeof(ctx->header));
469         ctx->header.chkpt.log = ntohl(dummy.header.chkpt.log);
470         ctx->header.chkpt.marker = ntohl(dummy.header.chkpt.marker);
471         ctx->header.tv_sec = ntohl(dummy.header.tv_sec);
472         ctx->header.tv_usec = ntohl(dummy.header.tv_usec);
473         ctx->header.message_len = ntohl(dummy.header.message_len);
474         STRATCON_NOIT_STREAM_HEADER(e->fd, (char *)feedtype,
475                                     nctx->remote_str, (char *)cn_expected,
476                                     ctx->header.chkpt.log, ctx->header.chkpt.marker,
477                                     ctx->header.tv_sec, ctx->header.tv_usec,
478                                     ctx->header.message_len);
479         free(ctx->buffer); ctx->buffer = NULL;
480         ctx->state = JLOG_STREAMER_WANT_BODY;
481         break;
482
483       case JLOG_STREAMER_WANT_BODY:
484         FULLREAD(e, ctx, (unsigned long)ctx->header.message_len);
485         STRATCON_NOIT_STREAM_BODY(e->fd, (char *)feedtype,
486                                   nctx->remote_str, (char *)cn_expected,
487                                   ctx->header.chkpt.log, ctx->header.chkpt.marker,
488                                   ctx->header.tv_sec, ctx->header.tv_usec,
489                                   ctx->buffer);
490         if(ctx->header.message_len > 0) {
491           ctx->needs_chkpt = 1;
492           ctx->push(DS_OP_INSERT, &nctx->r.remote, nctx->remote_cn,
493                     ctx->buffer, NULL);
494         }
495         else if(ctx->buffer)
496           free(ctx->buffer);
497         /* Don't free the buffer, it's used by the datastore process. */
498         ctx->buffer = NULL;
499         ctx->count--;
500         ctx->total_events++;
501         if(ctx->count == 0 && ctx->needs_chkpt) {
502           eventer_t completion_e;
503           eventer_remove_fd(e->fd);
504           completion_e = eventer_alloc();
505           memcpy(completion_e, e, sizeof(*e));
506           nctx->e = completion_e;
507           completion_e->mask = EVENTER_READ | EVENTER_WRITE | EVENTER_EXCEPTION;
508           ctx->state = JLOG_STREAMER_IS_ASYNC;
509           ctx->push(DS_OP_CHKPT, &nctx->r.remote, nctx->remote_cn,
510                     NULL, completion_e);
511           noitL(noit_debug, "Pushing %s batch async [%s]: [%u/%u]\n",
512                 feed_type_to_str(ntohl(ctx->jlog_feed_cmd)),
513                 nctx->remote_cn ? nctx->remote_cn : "(null)",
514                 ctx->header.chkpt.log, ctx->header.chkpt.marker);
515           noit_connection_disable_timeout(nctx);
516           return 0;
517         }
518         else if(ctx->count == 0)
519           ctx->state = JLOG_STREAMER_WANT_CHKPT;
520         else
521           ctx->state = JLOG_STREAMER_WANT_HEADER;
522         break;
523
524       case JLOG_STREAMER_IS_ASYNC:
525         ctx->state = JLOG_STREAMER_WANT_CHKPT; /* falls through */
526       case JLOG_STREAMER_WANT_CHKPT:
527         noitL(noit_debug, "Pushing %s checkpoint [%s]: [%u/%u]\n",
528               feed_type_to_str(ntohl(ctx->jlog_feed_cmd)),
529               nctx->remote_cn ? nctx->remote_cn : "(null)",
530               ctx->header.chkpt.log, ctx->header.chkpt.marker);
531         n_chkpt.log = htonl(ctx->header.chkpt.log);
532         n_chkpt.marker = htonl(ctx->header.chkpt.marker);
533
534         /* screw short writes.  I'd rather die than not write my data! */
535         len = e->opset->write(e->fd, &n_chkpt, sizeof(jlog_id),
536                               &mask, e);
537         if(len < 0) {
538           if(errno == EAGAIN) return mask | EVENTER_EXCEPTION;
539           goto socket_error;
540         }
541         if(len != sizeof(jlog_id)) {
542           noitL(noit_error, "short write on checkpointing stream.\n");
543           goto socket_error;
544         }
545         STRATCON_NOIT_STREAM_CHECKPOINT(e->fd, (char *)feedtype,
546                                         nctx->remote_str, (char *)cn_expected,
547                                         ctx->header.chkpt.log, ctx->header.chkpt.marker);
548         ctx->state = JLOG_STREAMER_WANT_COUNT;
549         break;
550     }
551   }
552   /* never get here */
553 }
554
555 int
556 noit_connection_ssl_upgrade(eventer_t e, int mask, void *closure,
557                             struct timeval *now) {
558   noit_connection_ctx_t *nctx = closure;
559   int rv;
560   const char *error = NULL, *cn_expected, *feedtype;
561
562   GET_EXPECTED_CN(nctx, cn_expected);
563   GET_FEEDTYPE(nctx, feedtype);
564   STRATCON_NOIT_CONNECT_SSL(e->fd, (char *)feedtype, nctx->remote_str,
565                             (char *)cn_expected);
566   rv = eventer_SSL_connect(e, &mask);
567   if(rv > 0) {
568     eventer_ssl_ctx_t *sslctx;
569     e->callback = nctx->consumer_callback;
570     /* We must make a copy of the acceptor_closure_t for each new
571      * connection.
572      */
573     if((sslctx = eventer_get_eventer_ssl_ctx(e)) != NULL) {
574       const char *cn, *end;
575       cn = eventer_ssl_get_peer_subject(sslctx);
576       if(cn && (cn = strstr(cn, "CN=")) != NULL) {
577         cn += 3;
578         end = cn;
579         while(*end && *end != '/') end++;
580         nctx->remote_cn = malloc(end - cn + 1);
581         memcpy(nctx->remote_cn, cn, end - cn);
582         nctx->remote_cn[end-cn] = '\0';
583       }
584       if(cn_expected && (!nctx->remote_cn ||
585                          strcmp(nctx->remote_cn, cn_expected))) {
586         error = "jlog connect CN mismatch\n";
587         goto error;
588       }
589     }
590     STRATCON_NOIT_CONNECT_SSL_SUCCESS(e->fd, (char *)feedtype,
591                                       nctx->remote_str, (char *)cn_expected);
592     return e->callback(e, mask, e->closure, now);
593   }
594   if(errno == EAGAIN) return mask | EVENTER_EXCEPTION;
595   noitL(noit_debug, "jlog streamer SSL upgrade failed.\n");
596
597  error:
598   STRATCON_NOIT_CONNECT_SSL_FAILED(e->fd, (char *)feedtype,
599                                    nctx->remote_str, (char *)cn_expected,
600                                    (char *)error, errno);
601   if(error) noitL(noit_error, "%s", error);
602   eventer_remove_fd(e->fd);
603   nctx->e = NULL;
604   e->opset->close(e->fd, &mask, e);
605   noit_connection_schedule_reattempt(nctx, now);
606   return 0;
607 }
608 int
609 noit_connection_complete_connect(eventer_t e, int mask, void *closure,
610                                  struct timeval *now) {
611   noit_connection_ctx_t *nctx = closure;
612   const char *cert, *key, *ca, *ciphers, *crl = NULL, *cn_expected, *feedtype;
613   char remote_str[128], tmp_str[128];
614   eventer_ssl_ctx_t *sslctx;
615   int aerrno, len;
616   socklen_t aerrno_len = sizeof(aerrno);
617
618   GET_EXPECTED_CN(nctx, cn_expected);
619   GET_FEEDTYPE(nctx, feedtype);
620   if(getsockopt(e->fd,SOL_SOCKET,SO_ERROR, &aerrno, &aerrno_len) == 0)
621     if(aerrno != 0) goto connect_error;
622   aerrno = 0;
623
624   if(mask & EVENTER_EXCEPTION) {
625     if(aerrno == 0 && (write(e->fd, e, 0) == -1))
626       aerrno = errno;
627  connect_error:
628     switch(nctx->r.remote.sa_family) {
629       case AF_INET:
630         len = sizeof(struct sockaddr_in);
631         inet_ntop(nctx->r.remote.sa_family, &nctx->r.remote_in.sin_addr,
632                   tmp_str, len);
633         snprintf(remote_str, sizeof(remote_str), "%s:%d",
634                  tmp_str, ntohs(nctx->r.remote_in.sin_port));
635         break;
636       case AF_INET6:
637         len = sizeof(struct sockaddr_in6);
638         inet_ntop(nctx->r.remote.sa_family, &nctx->r.remote_in6.sin6_addr,
639                   tmp_str, len);
640         snprintf(remote_str, sizeof(remote_str), "%s:%d",
641                  tmp_str, ntohs(nctx->r.remote_in6.sin6_port));
642        break;
643       case AF_UNIX:
644         snprintf(remote_str, sizeof(remote_str), "%s", nctx->r.remote_un.sun_path);
645         break;
646       default:
647         snprintf(remote_str, sizeof(remote_str), "(unknown)");
648     }
649     noitL(noit_error, "Error connecting to %s: %s\n",
650           remote_str, strerror(aerrno));
651     STRATCON_NOIT_CONNECT_FAILED(e->fd, (char *)feedtype, remote_str,
652                                  (char *)cn_expected, aerrno);
653     eventer_remove_fd(e->fd);
654     nctx->e = NULL;
655     e->opset->close(e->fd, &mask, e);
656     noit_connection_schedule_reattempt(nctx, now);
657     return 0;
658   }
659
660 #define SSLCONFGET(var,name) do { \
661   if(!noit_hash_retr_str(nctx->sslconfig, name, strlen(name), \
662                          &var)) var = NULL; } while(0)
663   SSLCONFGET(cert, "certificate_file");
664   SSLCONFGET(key, "key_file");
665   SSLCONFGET(ca, "ca_chain");
666   SSLCONFGET(ciphers, "ciphers");
667   SSLCONFGET(crl, "crl");
668   sslctx = eventer_ssl_ctx_new(SSL_CLIENT, cert, key, ca, ciphers);
669   if(!sslctx) goto connect_error;
670   if(crl) {
671     if(!eventer_ssl_use_crl(sslctx, crl)) {
672       noitL(noit_error, "Failed to load CRL from %s\n", crl);
673       eventer_ssl_ctx_free(sslctx);
674       goto connect_error;
675     }
676   }
677
678   memcpy(&nctx->last_connect, now, sizeof(*now));
679   eventer_ssl_ctx_set_verify(sslctx, eventer_ssl_verify_cert,
680                              nctx->sslconfig);
681   EVENTER_ATTACH_SSL(e, sslctx);
682   e->callback = noit_connection_ssl_upgrade;
683   STRATCON_NOIT_CONNECT_SUCCESS(e->fd, (char *)feedtype, nctx->remote_str,
684                                 (char *)cn_expected);
685   return e->callback(e, mask, closure, now);
686 }
687 static void
688 noit_connection_initiate_connection(noit_connection_ctx_t *nctx) {
689   struct timeval __now;
690   const char *cn_expected, *feedtype;
691   eventer_t e;
692   int rv, fd = -1;
693 #ifdef SO_KEEPALIVE
694   int optval;
695   socklen_t optlen = sizeof(optval);
696 #endif
697
698   GET_EXPECTED_CN(nctx, cn_expected);
699   GET_FEEDTYPE(nctx, feedtype);
700   nctx->e = NULL;
701   if(nctx->wants_permanent_shutdown) {
702     STRATCON_NOIT_SHUTDOWN_PERMANENT(-1, (char *)feedtype,
703                                      nctx->remote_str, (char *)cn_expected);
704     noit_connection_ctx_dealloc(nctx);
705     return;
706   }
707   /* Open a socket */
708   fd = socket(nctx->r.remote.sa_family, SOCK_STREAM, 0);
709   if(fd < 0) goto reschedule;
710
711   /* Make it non-blocking */
712   if(eventer_set_fd_nonblocking(fd)) goto reschedule;
713 #define set_or_bail(type, opt, val) do { \
714   optval = val; \
715   optlen = sizeof(optval); \
716   if(setsockopt(fd, type, opt, &optval, optlen) < 0) { \
717     noitL(noit_error, "Cannot set " #type "/" #opt " on jlog socket: %s\n", \
718           strerror(errno)); \
719     goto reschedule; \
720   } \
721 } while(0)
722 #ifdef SO_KEEPALIVE
723   set_or_bail(SOL_SOCKET, SO_KEEPALIVE, 1);
724 #endif
725 #ifdef TCP_KEEPALIVE_THRESHOLD
726   set_or_bail(IPPROTO_TCP, TCP_KEEPALIVE_THRESHOLD, 10 * 1000);
727 #endif
728 #ifdef TCP_KEEPALIVE_ABORT_THRESHOLD
729   set_or_bail(IPPROTO_TCP, TCP_KEEPALIVE_ABORT_THRESHOLD, 30 * 1000);
730 #endif
731 #ifdef TCP_CONN_NOTIFY_THRESHOLD
732   set_or_bail(IPPROTO_TCP, TCP_CONN_NOTIFY_THRESHOLD, 10 * 1000);
733 #endif
734 #ifdef TCP_CONN_ABORT_THRESHOLD
735   set_or_bail(IPPROTO_TCP, TCP_CONN_ABORT_THRESHOLD, 30 * 1000);
736 #endif
737
738   /* Initiate a connection */
739   rv = connect(fd, &nctx->r.remote, nctx->remote_len);
740   if(rv == -1 && errno != EINPROGRESS) goto reschedule;
741
742   /* Register a handler for connection completion */
743   e = eventer_alloc();
744   e->fd = fd;
745   e->mask = EVENTER_READ | EVENTER_WRITE | EVENTER_EXCEPTION;
746   e->callback = noit_connection_complete_connect;
747   e->closure = nctx;
748   nctx->e = e;
749   eventer_add(e);
750
751   STRATCON_NOIT_CONNECT(e->fd, (char *)feedtype, nctx->remote_str,
752                         (char *)cn_expected);
753   noit_connection_update_timeout(nctx);
754   return;
755
756  reschedule:
757   if(fd >= 0) close(fd);
758   gettimeofday(&__now, NULL);
759   noit_connection_schedule_reattempt(nctx, &__now);
760   return;
761 }
762
763 int
764 noit_connection_update_timeout(noit_connection_ctx_t *nctx) {
765   struct timeval now, diff;
766   if(nctx->max_silence == 0) return 0;
767
768   diff.tv_sec = nctx->max_silence / 1000;
769   diff.tv_usec = (nctx->max_silence % 1000) * 1000;
770   gettimeofday(&now, NULL);
771
772   if(!nctx->timeout_event) {
773     nctx->timeout_event = eventer_alloc();
774     nctx->timeout_event->mask = EVENTER_TIMER;
775     nctx->timeout_event->closure = nctx;
776     nctx->timeout_event->callback = noit_connection_session_timeout;
777     add_timeval(now, diff, &nctx->timeout_event->whence);
778     eventer_add(nctx->timeout_event);
779   }
780   else {
781     add_timeval(now, diff, &nctx->timeout_event->whence);
782     eventer_update(nctx->timeout_event, EVENTER_TIMER);
783   }
784   return 0;
785 }
786
787 int
788 noit_connection_disable_timeout(noit_connection_ctx_t *nctx) {
789   if(nctx->timeout_event) {
790     eventer_remove(nctx->timeout_event);
791     eventer_free(nctx->timeout_event);
792     nctx->timeout_event = NULL;
793   }
794   return 0;
795 }
796
797 int
798 initiate_noit_connection(const char *host, unsigned short port,
799                          noit_hash_table *sslconfig, noit_hash_table *config,
800                          eventer_func_t handler, void *closure,
801                          void (*freefunc)(void *)) {
802   noit_connection_ctx_t *ctx;
803   const char *stimeout;
804   int8_t family;
805   int rv;
806   union {
807     struct in_addr addr4;
808     struct in6_addr addr6;
809   } a;
810
811   if(host[0] == '/') {
812     family = AF_UNIX;
813   }
814   else {
815     family = AF_INET;
816     rv = inet_pton(family, host, &a);
817     if(rv != 1) {
818       family = AF_INET6;
819       rv = inet_pton(family, host, &a);
820       if(rv != 1) {
821         noitL(noit_stderr, "Cannot translate '%s' to IP\n", host);
822         return -1;
823       }
824     }
825   }
826
827   ctx = noit_connection_ctx_alloc();
828   ctx->remote_str = calloc(1, strlen(host) + 7);
829   snprintf(ctx->remote_str, strlen(host) + 7,
830            "%s:%d", host, port);
831  
832   memset(&ctx->r, 0, sizeof(ctx->r));
833   if(family == AF_UNIX) {
834     struct sockaddr_un *s = &ctx->r.remote_un;
835     s->sun_family = AF_UNIX;
836     strncpy(s->sun_path, host, sizeof(s->sun_path)-1);
837     ctx->remote_len = sizeof(*s);
838   }
839   else if(family == AF_INET) {
840     struct sockaddr_in *s = &ctx->r.remote_in;
841     s->sin_family = family;
842     s->sin_port = htons(port);
843     memcpy(&s->sin_addr, &a, sizeof(struct in_addr));
844     ctx->remote_len = sizeof(*s);
845   }
846   else {
847     struct sockaddr_in6 *s = &ctx->r.remote_in6;
848     s->sin6_family = family;
849     s->sin6_port = htons(port);
850     memcpy(&s->sin6_addr, &a, sizeof(a));
851     ctx->remote_len = sizeof(*s);
852   }
853
854   if(ctx->sslconfig)
855     noit_hash_delete_all(ctx->sslconfig, free, free);
856   else
857     ctx->sslconfig = calloc(1, sizeof(noit_hash_table));
858   noit_hash_merge_as_dict(ctx->sslconfig, sslconfig);
859   if(ctx->config)
860     noit_hash_delete_all(ctx->config, free, free);
861   else
862     ctx->config = calloc(1, sizeof(noit_hash_table));
863   noit_hash_merge_as_dict(ctx->config, config);
864
865   if(noit_hash_retr_str(ctx->config, "timeout", strlen("timeout"), &stimeout))
866     ctx->max_silence = atoi(stimeout);
867   else
868     ctx->max_silence = DEFAULT_NOIT_CONNECTION_TIMEOUT;
869   ctx->consumer_callback = handler;
870   ctx->consumer_free = freefunc;
871   ctx->consumer_ctx = closure;
872   noit_connection_initiate_connection(ctx);
873   return 0;
874 }
875
876 void
877 stratcon_streamer_connection(const char *toplevel, const char *destination,
878                              eventer_func_t handler,
879                              void *(*handler_alloc)(void), void *handler_ctx,
880                              void (*handler_free)(void *)) {
881   int i, cnt = 0;
882   noit_conf_section_t *noit_configs;
883   char path[256];
884
885   snprintf(path, sizeof(path), "/%s/noits//noit", toplevel ? toplevel : "*");
886   noit_configs = noit_conf_get_sections(NULL, path, &cnt);
887   noitL(noit_error, "Found %d %s stanzas\n", cnt, path);
888   for(i=0; i<cnt; i++) {
889     char address[256];
890     unsigned short port;
891     int portint;
892     noit_hash_table *sslconfig, *config;
893
894     if(!noit_conf_get_stringbuf(noit_configs[i],
895                                 "ancestor-or-self::node()/@address",
896                                 address, sizeof(address))) {
897       noitL(noit_error, "address attribute missing in noit %d\n", i+1);
898       continue;
899     }
900     /* if destination is specified, exact match it */
901     if(destination && strcmp(address, destination)) continue;
902
903     if(!noit_conf_get_int(noit_configs[i],
904                           "ancestor-or-self::node()/@port", &portint))
905       portint = 0;
906     port = (unsigned short) portint;
907     if(address[0] != '/' && (portint == 0 || (port != portint))) {
908       /* UNIX sockets don't require a port (they'll ignore it if specified */
909       noitL(noit_stderr,
910             "Invalid port [%d] specified in stanza %d\n", port, i+1);
911       continue;
912     }
913     sslconfig = noit_conf_get_hash(noit_configs[i], "sslconfig");
914     config = noit_conf_get_hash(noit_configs[i], "config");
915
916     noitL(noit_error, "initiating to %s\n", address);
917     initiate_noit_connection(address, port, sslconfig, config,
918                              handler,
919                              handler_alloc ? handler_alloc() : handler_ctx,
920                              handler_free);
921     noit_hash_destroy(sslconfig,free,free);
922     free(sslconfig);
923     noit_hash_destroy(config,free,free);
924     free(config);
925   }
926   free(noit_configs);
927 }
928 int
929 stratcon_find_noit_ip_by_cn(const char *cn, char *ip, int len) {
930   int rv = -1;
931   void *vip;
932   pthread_mutex_lock(&noit_ip_by_cn_lock);
933   if(noit_hash_retrieve(&noit_ip_by_cn, cn, strlen(cn), &vip)) {
934     int new_len;
935     char *new_ip = (char *)vip;
936     new_len = strlen(new_ip);
937     strlcpy(ip, new_ip, len);
938     if(new_len >= len) rv = new_len+1;
939     else rv = 0;
940   }
941   pthread_mutex_unlock(&noit_ip_by_cn_lock);
942   return rv;
943 }
944 void
945 stratcon_jlog_streamer_recache_noit() {
946   int di, cnt;
947   noit_conf_section_t *noit_configs;
948   noit_configs = noit_conf_get_sections(NULL, "//noits//noit", &cnt);
949   pthread_mutex_lock(&noit_ip_by_cn_lock);
950   noit_hash_delete_all(&noit_ip_by_cn, free, free);
951   for(di=0; di<cnt; di++) {
952     char address[64];
953     if(noit_conf_get_stringbuf(noit_configs[di], "self::node()/@address",
954                                  address, sizeof(address))) {
955       char expected_cn[256];
956       if(noit_conf_get_stringbuf(noit_configs[di], "self::node()/config/cn",
957                                  expected_cn, sizeof(expected_cn)))
958         noit_hash_store(&noit_ip_by_cn,
959                         strdup(expected_cn), strlen(expected_cn),
960                         strdup(address));
961     }
962   }
963   free(noit_configs);
964   pthread_mutex_unlock(&noit_ip_by_cn_lock);
965 }
966 void
967 stratcon_jlog_streamer_reload(const char *toplevel) {
968   /* flush and repopulate the cn cache */
969   stratcon_jlog_streamer_recache_noit();
970   if(!stratcon_datastore_get_enabled()) return;
971   stratcon_streamer_connection(toplevel, NULL,
972                                stratcon_jlog_recv_handler,
973                                (void *(*)())stratcon_jlog_streamer_datastore_ctx_alloc,
974                                NULL,
975                                jlog_streamer_ctx_free);
976 }
977
978 char *
979 stratcon_console_noit_opts(noit_console_closure_t ncct,
980                            noit_console_state_stack_t *stack,
981                            noit_console_state_t *dstate,
982                            int argc, char **argv, int idx) {
983   if(argc == 1) {
984     noit_hash_iter iter = NOIT_HASH_ITER_ZERO;
985     const char *key_id;
986     int klen, i = 0;
987     void *vconn, *vcn;
988     noit_connection_ctx_t *ctx;
989     noit_hash_table dedup = NOIT_HASH_EMPTY;
990
991     pthread_mutex_lock(&noits_lock);
992     while(noit_hash_next(&noits, &iter, &key_id, &klen, &vconn)) {
993       ctx = (noit_connection_ctx_t *)vconn;
994       vcn = NULL;
995       if(ctx->config && noit_hash_retrieve(ctx->config, "cn", 2, &vcn) &&
996          !noit_hash_store(&dedup, vcn, strlen(vcn), NULL)) {
997         if(!strncmp(vcn, argv[0], strlen(argv[0]))) {
998           if(idx == i) {
999             pthread_mutex_unlock(&noits_lock);
1000             noit_hash_destroy(&dedup, NULL, NULL);
1001             return strdup(vcn);
1002           }
1003           i++;
1004         }
1005       }
1006       if(ctx->remote_str &&
1007          !noit_hash_store(&dedup, ctx->remote_str, strlen(ctx->remote_str), NULL)) {
1008         if(!strncmp(ctx->remote_str, argv[0], strlen(argv[0]))) {
1009           if(idx == i) {
1010             pthread_mutex_unlock(&noits_lock);
1011             noit_hash_destroy(&dedup, NULL, NULL);
1012             return strdup(ctx->remote_str);
1013           }
1014           i++;
1015         }
1016       }
1017     }
1018     pthread_mutex_unlock(&noits_lock);
1019     noit_hash_destroy(&dedup, NULL, NULL);
1020   }
1021   if(argc == 2)
1022     return noit_console_opt_delegate(ncct, stack, dstate, argc-1, argv+1, idx);
1023   return NULL;
1024 }
1025 static int
1026 stratcon_console_show_noits(noit_console_closure_t ncct,
1027                             int argc, char **argv,
1028                             noit_console_state_t *dstate,
1029                             void *closure) {
1030   noit_hash_iter iter = NOIT_HASH_ITER_ZERO;
1031   const char *key_id, *ecn;
1032   int klen, n = 0, i;
1033   void *vconn;
1034   noit_connection_ctx_t **ctx;
1035
1036   if(closure != (void *)0 && argc == 0) {
1037     nc_printf(ncct, "takes an argument\n");
1038     return 0;
1039   }
1040   if(closure == (void *)0 && argc > 0) {
1041     nc_printf(ncct, "takes no arguments\n");
1042     return 0;
1043   }
1044   pthread_mutex_lock(&noits_lock);
1045   ctx = malloc(sizeof(*ctx) * noits.size);
1046   while(noit_hash_next(&noits, &iter, &key_id, &klen,
1047                        &vconn)) {
1048     ctx[n] = (noit_connection_ctx_t *)vconn;
1049     if(argc == 0 ||
1050        !strcmp(ctx[n]->remote_str, argv[0]) ||
1051        (ctx[n]->config && noit_hash_retr_str(ctx[n]->config, "cn", 2, &ecn) &&
1052         !strcmp(ecn, argv[0]))) {
1053       noit_atomic_inc32(&ctx[n]->refcnt);
1054       n++;
1055     }
1056   }
1057   pthread_mutex_unlock(&noits_lock);
1058   qsort(ctx, n, sizeof(*ctx), remote_str_sort);
1059   for(i=0; i<n; i++) {
1060     nc_print_noit_conn_brief(ncct, ctx[i]);
1061     noit_connection_ctx_deref(ctx[i]);
1062   }
1063   free(ctx);
1064   return 0;
1065 }
1066
1067 static void
1068 emit_noit_info_metrics(struct timeval *now, const char *uuid_str,
1069                        noit_connection_ctx_t *nctx) {
1070   struct timeval last, session_duration, diff;
1071   u_int64_t session_duration_ms, last_event_ms;
1072   jlog_streamer_ctx_t *jctx = nctx->consumer_ctx;
1073   char str[1024], *wr;
1074   int len;
1075   const char *cn_expected;
1076   const char *feedtype = "unknown";
1077
1078   GET_FEEDTYPE(nctx, feedtype);
1079   if(NULL != (wr = strchr(feedtype, '/'))) feedtype = wr+1;
1080
1081   GET_EXPECTED_CN(nctx, cn_expected);
1082   if(!cn_expected) return;
1083
1084   snprintf(str, sizeof(str), "M\t%lu.%03lu\t%s\t%s`%s`",
1085            (long unsigned int)now->tv_sec,
1086            (long unsigned int)now->tv_usec/1000UL,
1087            uuid_str, cn_expected, feedtype);
1088   wr = str + strlen(str);
1089   len = sizeof(str) - (wr - str);
1090
1091   /* Now we write NAME TYPE VALUE into wr each time and push it */
1092 #define push_noit_m_str(name, value) do { \
1093   snprintf(wr, len, "%s\ts\t%s\n", name, value); \
1094   stratcon_datastore_push(DS_OP_INSERT, \
1095                           (struct sockaddr *)&self_stratcon_ip, \
1096                           self_stratcon_hostname, strdup(str), NULL); \
1097   stratcon_iep_line_processor(DS_OP_INSERT, \
1098                               (struct sockaddr *)&self_stratcon_ip, \
1099                               self_stratcon_hostname, strdup(str), NULL); \
1100 } while(0)
1101 #define push_noit_m_u64(name, value) do { \
1102   snprintf(wr, len, "%s\tL\t%llu\n", name, (long long unsigned int)value); \
1103   stratcon_datastore_push(DS_OP_INSERT, \
1104                           (struct sockaddr *)&self_stratcon_ip, \
1105                           self_stratcon_hostname, strdup(str), NULL); \
1106   stratcon_iep_line_processor(DS_OP_INSERT, \
1107                               (struct sockaddr *)&self_stratcon_ip, \
1108                               self_stratcon_hostname, strdup(str), NULL); \
1109 } while(0)
1110
1111   last.tv_sec = jctx->header.tv_sec;
1112   last.tv_usec = jctx->header.tv_usec;
1113   sub_timeval(*now, last, &diff);
1114   last_event_ms = diff.tv_sec * 1000 + diff.tv_usec / 1000;
1115   sub_timeval(*now, nctx->last_connect, &session_duration);
1116   session_duration_ms = session_duration.tv_sec * 1000 +
1117                         session_duration.tv_usec / 1000;
1118
1119   push_noit_m_str("state", nctx->remote_cn ? "connected" :
1120                              (nctx->retry_event ? "disconnected" :
1121                                                   "connecting"));
1122   push_noit_m_u64("last_event_age_ms", last_event_ms);
1123   push_noit_m_u64("session_length_ms", session_duration_ms);
1124 }
1125 static int
1126 periodic_noit_metrics(eventer_t e, int mask, void *closure,
1127                       struct timeval *now) {
1128   struct timeval whence = DEFAULT_NOIT_PERIOD_TV;
1129   noit_connection_ctx_t **ctxs;
1130   noit_hash_iter iter = NOIT_HASH_ITER_ZERO;
1131   const char *key_id;
1132   void *vconn;
1133   int klen, n = 0, i;
1134   char str[1024];
1135   char uuid_str[1024], tmp_uuid_str[UUID_STR_LEN+1];
1136   struct timeval epoch, diff;
1137   u_int64_t uptime = 0;
1138   char ip_str[128];
1139
1140   inet_ntop(AF_INET, &self_stratcon_ip.sin_addr, ip_str,
1141             sizeof(ip_str));
1142
1143   uuid_str[0] = '\0';
1144   uuid_unparse_lower(self_stratcon_id, tmp_uuid_str);
1145   if(stratcon_selfcheck_extended_id) {
1146     strlcat(uuid_str, ip_str, sizeof(uuid_str)-37);
1147     strlcat(uuid_str, "`selfcheck`selfcheck`", sizeof(uuid_str)-37);
1148   }
1149   strlcat(uuid_str, tmp_uuid_str, sizeof(uuid_str));
1150
1151 #define PUSH_BOTH(type, str) do { \
1152   stratcon_datastore_push(type, \
1153                           (struct sockaddr *)&self_stratcon_ip, \
1154                           self_stratcon_hostname, str, NULL); \
1155   stratcon_iep_line_processor(type, \
1156                               (struct sockaddr *)&self_stratcon_ip, \
1157                               self_stratcon_hostname, str, NULL); \
1158 } while(0)
1159
1160   if(closure == NULL) {
1161     /* Only do this the first time we get called */
1162     snprintf(str, sizeof(str), "C\t%lu.%03lu\t%s\t%s\tstratcon\t%s\n",
1163              (long unsigned int)now->tv_sec,
1164              (long unsigned int)now->tv_usec/1000UL, uuid_str, ip_str,
1165              self_stratcon_hostname);
1166     PUSH_BOTH(DS_OP_INSERT, strdup(str));
1167   }
1168
1169   pthread_mutex_lock(&noits_lock);
1170   ctxs = malloc(sizeof(*ctxs) * noits.size);
1171   while(noit_hash_next(&noits, &iter, &key_id, &klen,
1172                        &vconn)) {
1173     ctxs[n] = (noit_connection_ctx_t *)vconn;
1174     noit_atomic_inc32(&ctxs[n]->refcnt);
1175     n++;
1176   }
1177   pthread_mutex_unlock(&noits_lock);
1178
1179   snprintf(str, sizeof(str), "S\t%lu.%03lu\t%s\tG\tA\t0\tok %d noits\n",
1180            (long unsigned int)now->tv_sec,
1181            (long unsigned int)now->tv_usec/1000UL, uuid_str, n);
1182   PUSH_BOTH(DS_OP_INSERT, strdup(str));
1183
1184   if(eventer_get_epoch(&epoch) != 0)
1185     memcpy(&epoch, now, sizeof(epoch));
1186   sub_timeval(*now, epoch, &diff);
1187   uptime = diff.tv_sec;
1188   snprintf(str, sizeof(str), "M\t%lu.%03lu\t%s\tuptime\tL\t%llu\n",
1189            (long unsigned int)now->tv_sec,
1190            (long unsigned int)now->tv_usec/1000UL,
1191            uuid_str, (long long unsigned int)uptime);
1192   PUSH_BOTH(DS_OP_INSERT, strdup(str));
1193
1194   for(i=0; i<n; i++) {
1195     emit_noit_info_metrics(now, uuid_str, ctxs[i]);
1196     noit_connection_ctx_deref(ctxs[i]);
1197   }
1198   free(ctxs);
1199   PUSH_BOTH(DS_OP_CHKPT, NULL);
1200
1201   add_timeval(e->whence, whence, &whence);
1202   eventer_add_at(periodic_noit_metrics, (void *)0x1, whence);
1203   return 0;
1204 }
1205
1206 static int
1207 rest_show_noits(noit_http_rest_closure_t *restc,
1208                 int npats, char **pats) {
1209   xmlDocPtr doc;
1210   xmlNodePtr root;
1211   noit_hash_table seen = NOIT_HASH_EMPTY;
1212   noit_hash_iter iter = NOIT_HASH_ITER_ZERO;
1213   char path[256];
1214   const char *key_id;
1215   const char *type = NULL, *want_cn = NULL;
1216   int klen, n = 0, i, di, cnt;
1217   void *vconn;
1218   noit_connection_ctx_t **ctxs;
1219   noit_conf_section_t *noit_configs;
1220   struct timeval now, diff, last;
1221   xmlNodePtr node;
1222   noit_http_request *req = noit_http_session_request(restc->http_ctx);
1223
1224   noit_http_process_querystring(req);
1225   type = noit_http_request_querystring(req, "type");
1226   want_cn = noit_http_request_querystring(req, "cn");
1227
1228   gettimeofday(&now, NULL);
1229
1230   pthread_mutex_lock(&noits_lock);
1231   ctxs = malloc(sizeof(*ctxs) * noits.size);
1232   while(noit_hash_next(&noits, &iter, &key_id, &klen,
1233                        &vconn)) {
1234     ctxs[n] = (noit_connection_ctx_t *)vconn;
1235     noit_atomic_inc32(&ctxs[n]->refcnt);
1236     n++;
1237   }
1238   pthread_mutex_unlock(&noits_lock);
1239   qsort(ctxs, n, sizeof(*ctxs), remote_str_sort);
1240
1241   doc = xmlNewDoc((xmlChar *)"1.0");
1242   root = xmlNewDocNode(doc, NULL, (xmlChar *)"noits", NULL);
1243   xmlDocSetRootElement(doc, root);
1244
1245   for(i=0; i<n; i++) {
1246     char buff[256];
1247     const char *feedtype = "unknown", *state = "unknown";
1248     noit_connection_ctx_t *ctx = ctxs[i];
1249     jlog_streamer_ctx_t *jctx = ctx->consumer_ctx;
1250
1251     feedtype = feed_type_to_str(ntohl(jctx->jlog_feed_cmd));
1252
1253     /* If the user requested a specific type and we're not it, skip. */
1254     if(type && strcmp(feedtype, type)) continue;
1255     /* If the user wants a specific CN... limit to that. */
1256     if(want_cn && (!ctx->remote_cn || strcmp(want_cn, ctx->remote_cn)))
1257       continue;
1258
1259     node = xmlNewNode(NULL, (xmlChar *)"noit");
1260     snprintf(buff, sizeof(buff), "%llu.%06d",
1261              (long long unsigned)ctx->last_connect.tv_sec,
1262              (int)ctx->last_connect.tv_usec);
1263     xmlSetProp(node, (xmlChar *)"last_connect", (xmlChar *)buff);
1264     xmlSetProp(node, (xmlChar *)"state", ctx->remote_cn ?
1265                (xmlChar *)"connected" :
1266                (ctx->retry_event ? (xmlChar *)"disconnected" :
1267                                     (xmlChar *)"connecting"));
1268     if(ctx->e) {
1269       char buff[128];
1270       const char *addrstr = NULL;
1271       struct sockaddr_in6 addr6;
1272       socklen_t len = sizeof(addr6);
1273       if(getsockname(ctx->e->fd, (struct sockaddr *)&addr6, &len) == 0) {
1274         unsigned short port = 0;
1275         if(addr6.sin6_family == AF_INET) {
1276           addrstr = inet_ntop(addr6.sin6_family,
1277                               &((struct sockaddr_in *)&addr6)->sin_addr,
1278                               buff, sizeof(buff));
1279           memcpy(&port, &(&addr6)->sin6_port, sizeof(port));
1280           port = ntohs(port);
1281         }
1282         else if(addr6.sin6_family == AF_INET6) {
1283           addrstr = inet_ntop(addr6.sin6_family, &addr6.sin6_addr,
1284                               buff, sizeof(buff));
1285           port = ntohs(addr6.sin6_port);
1286         }
1287         if(addrstr != NULL) {
1288           snprintf(buff + strlen(buff), sizeof(buff) - strlen(buff),
1289                    ":%u", port);
1290           xmlSetProp(node, (xmlChar *)"local", (xmlChar *)buff);
1291         }
1292       }
1293     }
1294     noit_hash_replace(&seen, strdup(ctx->remote_str), strlen(ctx->remote_str),
1295                       0, free, NULL);
1296     xmlSetProp(node, (xmlChar *)"remote", (xmlChar *)ctx->remote_str);
1297     xmlSetProp(node, (xmlChar *)"type", (xmlChar *)feedtype);
1298     if(ctx->retry_event) {
1299       sub_timeval(ctx->retry_event->whence, now, &diff);
1300       snprintf(buff, sizeof(buff), "%llu.%06d",
1301                (long long unsigned)diff.tv_sec, (int)diff.tv_usec);
1302       xmlSetProp(node, (xmlChar *)"next_attempt", (xmlChar *)buff);
1303     }
1304     else if(ctx->remote_cn) {
1305       if(ctx->remote_cn)
1306         xmlSetProp(node, (xmlChar *)"remote_cn", (xmlChar *)ctx->remote_cn);
1307  
1308       switch(jctx->state) {
1309         case JLOG_STREAMER_WANT_INITIATE: state = "initiate"; break;
1310         case JLOG_STREAMER_WANT_COUNT: state = "waiting for next batch"; break;
1311         case JLOG_STREAMER_WANT_ERROR: state = "waiting for error"; break;
1312         case JLOG_STREAMER_WANT_HEADER: state = "reading header"; break;
1313         case JLOG_STREAMER_WANT_BODY: state = "reading body"; break;
1314         case JLOG_STREAMER_IS_ASYNC: state = "asynchronously processing"; break;
1315         case JLOG_STREAMER_WANT_CHKPT: state = "checkpointing"; break;
1316       }
1317       xmlSetProp(node, (xmlChar *)"state", (xmlChar *)state);
1318       snprintf(buff, sizeof(buff), "%08x:%08x",
1319                jctx->header.chkpt.log, jctx->header.chkpt.marker);
1320       xmlSetProp(node, (xmlChar *)"checkpoint", (xmlChar *)buff);
1321       snprintf(buff, sizeof(buff), "%llu",
1322                (unsigned long long)jctx->total_events);
1323       xmlSetProp(node, (xmlChar *)"session_events", (xmlChar *)buff);
1324       snprintf(buff, sizeof(buff), "%llu",
1325                (unsigned long long)jctx->total_bytes_read);
1326       xmlSetProp(node, (xmlChar *)"session_bytes", (xmlChar *)buff);
1327  
1328       sub_timeval(now, ctx->last_connect, &diff);
1329       snprintf(buff, sizeof(buff), "%lld.%06d",
1330                (long long)diff.tv_sec, (int)diff.tv_usec);
1331       xmlSetProp(node, (xmlChar *)"session_duration", (xmlChar *)buff);
1332  
1333       if(jctx->header.tv_sec) {
1334         last.tv_sec = jctx->header.tv_sec;
1335         last.tv_usec = jctx->header.tv_usec;
1336         snprintf(buff, sizeof(buff), "%llu.%06d",
1337                  (unsigned long long)last.tv_sec, (int)last.tv_usec);
1338         xmlSetProp(node, (xmlChar *)"last_event", (xmlChar *)buff);
1339         sub_timeval(now, last, &diff);
1340         snprintf(buff, sizeof(buff), "%lld.%06d",
1341                  (long long)diff.tv_sec, (int)diff.tv_usec);
1342         xmlSetProp(node, (xmlChar *)"last_event_age", (xmlChar *)buff);
1343       }
1344     }
1345
1346     xmlAddChild(root, node);
1347     noit_connection_ctx_deref(ctx);
1348   }
1349   free(ctxs);
1350
1351   if(!type || !strcmp(type, "configured")) {
1352     snprintf(path, sizeof(path), "//noits//noit");
1353     noit_configs = noit_conf_get_sections(NULL, path, &cnt);
1354     for(di=0; di<cnt; di++) {
1355       char address[64], port_str[32], remote_str[98];
1356       char expected_cn_buff[256], *expected_cn = NULL;
1357       if(noit_conf_get_stringbuf(noit_configs[di], "self::node()/config/cn",
1358                                  expected_cn_buff, sizeof(expected_cn_buff)))
1359         expected_cn = expected_cn_buff;
1360       if(want_cn && (!expected_cn || strcmp(want_cn, expected_cn))) continue;
1361       if(noit_conf_get_stringbuf(noit_configs[di], "self::node()/@address",
1362                                  address, sizeof(address))) {
1363         void *v;
1364         if(!noit_conf_get_stringbuf(noit_configs[di], "self::node()/@port",
1365                                    port_str, sizeof(port_str)))
1366           strlcpy(port_str, "43191", sizeof(port_str));
1367
1368         /* If the user wants a specific CN... limit to that. */
1369           if(want_cn && (!expected_cn || strcmp(want_cn, expected_cn)))
1370             continue;
1371
1372         snprintf(remote_str, sizeof(remote_str), "%s:%s", address, port_str);
1373         if(!noit_hash_retrieve(&seen, remote_str, strlen(remote_str), &v)) {
1374           node = xmlNewNode(NULL, (xmlChar *)"noit");
1375           xmlSetProp(node, (xmlChar *)"remote", (xmlChar *)remote_str);
1376           xmlSetProp(node, (xmlChar *)"type", (xmlChar *)"configured");
1377           if(expected_cn)
1378             xmlSetProp(node, (xmlChar *)"cn", (xmlChar *)expected_cn);
1379           xmlAddChild(root, node);
1380         }
1381       }
1382     }
1383     free(noit_configs);
1384   }
1385   noit_hash_destroy(&seen, free, NULL);
1386
1387   noit_http_response_ok(restc->http_ctx, "text/xml");
1388   noit_http_response_xml(restc->http_ctx, doc);
1389   noit_http_response_end(restc->http_ctx);
1390   xmlFreeDoc(doc);
1391   return 0;
1392 }
1393 static int
1394 stratcon_add_noit(const char *target, unsigned short port,
1395                   const char *cn) {
1396   int cnt;
1397   char path[256];
1398   char port_str[6];
1399   noit_conf_section_t *noit_configs, parent;
1400   xmlNodePtr newnoit, config, cnnode;
1401
1402   snprintf(path, sizeof(path),
1403            "//noits//noit[@address=\"%s\" and @port=\"%d\"]", target, port);
1404   noit_configs = noit_conf_get_sections(NULL, path, &cnt);
1405   free(noit_configs);
1406   if(cnt != 0) return -1;
1407
1408   parent = noit_conf_get_section(NULL, "//noits");
1409   if(!parent) return -1;
1410   snprintf(port_str, sizeof(port_str), "%d", port);
1411   newnoit = xmlNewNode(NULL, (xmlChar *)"noit");
1412   xmlSetProp(newnoit, (xmlChar *)"address", (xmlChar *)target);
1413   xmlSetProp(newnoit, (xmlChar *)"port", (xmlChar *)port_str);
1414   xmlAddChild(parent, newnoit);
1415   if(cn) {
1416     config = xmlNewNode(NULL, (xmlChar *)"config");
1417     cnnode = xmlNewNode(NULL, (xmlChar *)"cn");
1418     xmlNodeAddContent(cnnode, (xmlChar *)cn);
1419     xmlAddChild(config, cnnode);
1420     xmlAddChild(newnoit, config);
1421     pthread_mutex_lock(&noit_ip_by_cn_lock);
1422     noit_hash_replace(&noit_ip_by_cn, strdup(cn), strlen(cn),
1423                       strdup(target), free, free);
1424     pthread_mutex_unlock(&noit_ip_by_cn_lock);
1425   }
1426   if(stratcon_datastore_get_enabled())
1427     stratcon_streamer_connection(NULL, target,
1428                                  stratcon_jlog_recv_handler,
1429                                  (void *(*)())stratcon_jlog_streamer_datastore_ctx_alloc,
1430                                  NULL,
1431                                  jlog_streamer_ctx_free);
1432   if(stratcon_iep_get_enabled())
1433     stratcon_streamer_connection(NULL, target,
1434                                  stratcon_jlog_recv_handler,
1435                                  (void *(*)())stratcon_jlog_streamer_iep_ctx_alloc,
1436                                  NULL,
1437                                  jlog_streamer_ctx_free);
1438   return 1;
1439 }
1440 static int
1441 stratcon_remove_noit(const char *target, unsigned short port) {
1442   noit_hash_iter iter = NOIT_HASH_ITER_ZERO;
1443   const char *key_id;
1444   int klen, n = -1, i, cnt = 0;
1445   void *vconn;
1446   noit_connection_ctx_t **ctx;
1447   noit_conf_section_t *noit_configs;
1448   char path[256];
1449   char remote_str[256];
1450
1451   snprintf(remote_str, sizeof(remote_str), "%s:%d", target, port);
1452
1453   snprintf(path, sizeof(path),
1454            "//noits//noit[@address=\"%s\" and @port=\"%d\"]", target, port);
1455   noit_configs = noit_conf_get_sections(NULL, path, &cnt);
1456   for(i=0; i<cnt; i++) {
1457     char expected_cn[256];
1458     if(noit_conf_get_stringbuf(noit_configs[i], "self::node()/config/cn",
1459                                expected_cn, sizeof(expected_cn))) {
1460       pthread_mutex_lock(&noit_ip_by_cn_lock);
1461       noit_hash_delete(&noit_ip_by_cn, expected_cn, strlen(expected_cn),
1462                        free, free);
1463       pthread_mutex_unlock(&noit_ip_by_cn_lock);
1464     }
1465     CONF_REMOVE(noit_configs[i]);
1466     xmlUnlinkNode(noit_configs[i]);
1467     xmlFreeNode(noit_configs[i]);
1468     n = 0;
1469   }
1470   free(noit_configs);
1471
1472   pthread_mutex_lock(&noits_lock);
1473   ctx = malloc(sizeof(*ctx) * noits.size);
1474   while(noit_hash_next(&noits, &iter, &key_id, &klen,
1475                        &vconn)) {
1476     if(!strcmp(((noit_connection_ctx_t *)vconn)->remote_str, remote_str)) {
1477       ctx[n] = (noit_connection_ctx_t *)vconn;
1478       noit_atomic_inc32(&ctx[n]->refcnt);
1479       n++;
1480     }
1481   }
1482   pthread_mutex_unlock(&noits_lock);
1483   for(i=0; i<n; i++) {
1484     noit_connection_ctx_dealloc(ctx[i]); /* once for the record */
1485     noit_connection_ctx_deref(ctx[i]);   /* once for the aboce inc32 */
1486   }
1487   free(ctx);
1488   return n;
1489 }
1490 static int
1491 rest_set_noit(noit_http_rest_closure_t *restc,
1492               int npats, char **pats) {
1493   const char *cn = NULL;
1494   noit_http_session_ctx *ctx = restc->http_ctx;
1495   noit_http_request *req = noit_http_session_request(ctx);
1496   unsigned short port = 43191;
1497   if(npats < 1 || npats > 2)
1498     noit_http_response_server_error(ctx, "text/xml");
1499   if(npats == 2) port = atoi(pats[1]);
1500   noit_http_process_querystring(req);
1501   cn = noit_http_request_querystring(req, "cn");
1502   if(stratcon_add_noit(pats[0], port, cn) >= 0)
1503     noit_http_response_ok(ctx, "text/xml");
1504   else
1505     noit_http_response_standard(ctx, 409, "EXISTS", "text/xml");
1506   if(noit_conf_write_file(NULL) != 0)
1507     noitL(noit_error, "local config write failed\n");
1508   noit_conf_mark_changed();
1509   noit_http_response_end(ctx);
1510   return 0;
1511 }
1512 static int
1513 rest_delete_noit(noit_http_rest_closure_t *restc,
1514                  int npats, char **pats) {
1515   noit_http_session_ctx *ctx = restc->http_ctx;
1516   unsigned short port = 43191;
1517   if(npats < 1 || npats > 2)
1518     noit_http_response_server_error(ctx, "text/xml");
1519   if(npats == 2) port = atoi(pats[1]);
1520   if(stratcon_remove_noit(pats[0], port) >= 0)
1521     noit_http_response_ok(ctx, "text/xml");
1522   else
1523     noit_http_response_not_found(ctx, "text/xml");
1524   if(noit_conf_write_file(NULL) != 0)
1525     noitL(noit_error, "local config write failed\n");
1526   noit_conf_mark_changed();
1527   noit_http_response_end(ctx);
1528   return 0;
1529 }
1530 static int
1531 stratcon_console_conf_noits(noit_console_closure_t ncct,
1532                             int argc, char **argv,
1533                             noit_console_state_t *dstate,
1534                             void *closure) {
1535   char *cp, target[128];
1536   unsigned short port = 43191;
1537   int adding = (int)(vpsized_int)closure;
1538   if(argc != 1)
1539     return -1;
1540
1541   cp = strchr(argv[0], ':');
1542   if(cp) {
1543     strlcpy(target, argv[0], MIN(sizeof(target), cp-argv[0]+1));
1544     port = atoi(cp+1);
1545   }
1546   else strlcpy(target, argv[0], sizeof(target));
1547   if(adding) {
1548     if(stratcon_add_noit(target, port, NULL) >= 0) {
1549       nc_printf(ncct, "Added noit at %s:%d\n", target, port);
1550     }
1551     else {
1552       nc_printf(ncct, "Failed to add noit at %s:%d\n", target, port);
1553     }
1554   }
1555   else {
1556     if(stratcon_remove_noit(target, port) >= 0) {
1557       nc_printf(ncct, "Removed noit at %s:%d\n", target, port);
1558     }
1559     else {
1560       nc_printf(ncct, "Failed to remove noit at %s:%d\n", target, port);
1561     }
1562   }
1563   return 0;
1564 }
1565
1566 static void
1567 register_console_streamer_commands() {
1568   noit_console_state_t *tl;
1569   cmd_info_t *showcmd, *confcmd, *conftcmd, *conftnocmd;
1570
1571   tl = noit_console_state_initial();
1572   showcmd = noit_console_state_get_cmd(tl, "show");
1573   assert(showcmd && showcmd->dstate);
1574   confcmd = noit_console_state_get_cmd(tl, "configure");
1575   conftcmd = noit_console_state_get_cmd(confcmd->dstate, "terminal");
1576   conftnocmd = noit_console_state_get_cmd(conftcmd->dstate, "no");
1577
1578   noit_console_state_add_cmd(conftcmd->dstate,
1579     NCSCMD("noit", stratcon_console_conf_noits, NULL, NULL, (void *)1));
1580   noit_console_state_add_cmd(conftnocmd->dstate,
1581     NCSCMD("noit", stratcon_console_conf_noits, NULL, NULL, (void *)0));
1582
1583   noit_console_state_add_cmd(showcmd->dstate,
1584     NCSCMD("noit", stratcon_console_show_noits,
1585            stratcon_console_noit_opts, NULL, (void *)1));
1586   noit_console_state_add_cmd(showcmd->dstate,
1587     NCSCMD("noits", stratcon_console_show_noits, NULL, NULL, NULL));
1588 }
1589
1590 void
1591 stratcon_jlog_streamer_init(const char *toplevel) {
1592   struct timeval whence = DEFAULT_NOIT_PERIOD_TV;
1593   struct in_addr remote;
1594   char uuid_str[UUID_STR_LEN + 1];
1595
1596   pthread_mutex_init(&noits_lock, NULL);
1597   pthread_mutex_init(&noit_ip_by_cn_lock, NULL);
1598   eventer_name_callback("noit_connection_reinitiate",
1599                         noit_connection_reinitiate);
1600   eventer_name_callback("stratcon_jlog_recv_handler",
1601                         stratcon_jlog_recv_handler);
1602   eventer_name_callback("noit_connection_ssl_upgrade",
1603                         noit_connection_ssl_upgrade);
1604   eventer_name_callback("noit_connection_complete_connect",
1605                         noit_connection_complete_connect);
1606   eventer_name_callback("noit_connection_session_timeout",
1607                         noit_connection_session_timeout);
1608   register_console_streamer_commands();
1609   stratcon_jlog_streamer_reload(toplevel);
1610   stratcon_streamer_connection(toplevel, "", NULL, NULL, NULL, NULL);
1611   assert(noit_http_rest_register_auth(
1612     "GET", "/noits/", "^show$", rest_show_noits,
1613              noit_http_rest_client_cert_auth
1614   ) == 0);
1615   assert(noit_http_rest_register_auth(
1616     "PUT", "/noits/", "^set/([^/:]+)$", rest_set_noit,
1617              noit_http_rest_client_cert_auth
1618   ) == 0);
1619   assert(noit_http_rest_register_auth(
1620     "PUT", "/noits/", "^set/([^/:]+):(\\d+)$", rest_set_noit,
1621              noit_http_rest_client_cert_auth
1622   ) == 0);
1623   assert(noit_http_rest_register_auth(
1624     "DELETE", "/noits/", "^delete/([^/:]+)$", rest_delete_noit,
1625              noit_http_rest_client_cert_auth
1626   ) == 0);
1627   assert(noit_http_rest_register_auth(
1628     "DELETE", "/noits/", "^delete/([^/:]+):(\\d+)$", rest_delete_noit,
1629              noit_http_rest_client_cert_auth
1630   ) == 0);
1631
1632   uuid_clear(self_stratcon_id);
1633
1634   if(noit_conf_get_stringbuf(NULL, "/stratcon/@id",
1635                              uuid_str, sizeof(uuid_str)) &&
1636      uuid_parse(uuid_str, self_stratcon_id) == 0) {
1637     int period;
1638     noit_conf_get_boolean(NULL, "/stratcon/@extended_id",
1639                           &stratcon_selfcheck_extended_id);
1640     /* If a UUID was provided for stratcon itself, we will report metrics
1641      * on a large variety of things (including all noits).
1642      */
1643     if(noit_conf_get_int(NULL, "/stratcon/@metric_period", &period) &&
1644        period > 0) {
1645       DEFAULT_NOIT_PERIOD_TV.tv_sec = period / 1000;
1646       DEFAULT_NOIT_PERIOD_TV.tv_usec = (period % 1000) * 1000;
1647     }
1648     self_stratcon_ip.sin_family = AF_INET;
1649     remote.s_addr = 0xffffffff;
1650     noit_getip_ipv4(remote, &self_stratcon_ip.sin_addr);
1651     gethostname(self_stratcon_hostname, sizeof(self_stratcon_hostname));
1652     eventer_add_in(periodic_noit_metrics, NULL, whence);
1653   }
1654 }
1655
Note: See TracBrowser for help on using the browser.