duct.c

   1 /*
   2  *  innduct
   3  *  tailing reliable realtime streaming feeder for inn
   4  *  innduct.c - main program, option parsing and startup
   5  *
   6  *  Copyright (C) 2010 Ian Jackson <ijackson@chiark.greenend.org.uk>
   7  *
   8  *  This program is free software: you can redistribute it and/or modify
   9  *  it under the terms of the GNU General Public License as published by
  10  *  the Free Software Foundation, either version 3 of the License, or
  11  *  (at your option) any later version.
  12  *
  13  *  This program is distributed in the hope that it will be useful,
  14  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  *  GNU General Public License for more details.
  17  *
  18  *  You should have received a copy of the GNU General Public License
  19  *  along with this program.  If not, see <http://www.gnu.org/licenses/>.
  20  *
  21  *  (I believe that when you compile and link this as part of the inn2
  22  *  build, with the Makefile runes I have provided, all the libraries
  23  *  and files which end up included in innduct are licence-compatible
  24  *  with GPLv3.  If not then please let me know.  -Ian Jackson.)
  25  */
  26
  27 #include "innduct.h"
  28
  29 const char *sms_names[]= {
  30 #define SMS_DEF_NAME(s) #s ,
  31   SMS_LIST(SMS_DEF_NAME)
  32   0
  33 };
  34
  35
  36 /*----- general operational variables -----*/
  37
  38 /* main initialises */
  39 oop_source *loop;
  40 ConnList conns;
  41 char *path_lock, *path_flushing, *path_defer, *path_dump;
  42 char *globpat_backlog;
  43 pid_t self_pid;
  44 int *lowvol_perperiod;
  45 int lowvol_circptr;
  46 int lowvol_total; /* does not include current period */
  47
  48 /*---------- configuration option variables ----------*/
  49 /* when changing defaults, remember to update the manpage */
  50
  51 const char *sitename, *remote_host;
  52 const char *feedfile, *path_run, *path_cli, *path_cli_dir;
  53 int quiet_multiple=0;
  54 int interactive=0, try_filemon=1;
  55 int try_stream=1;
  56 int port=119;
  57 const char *inndconffile;
  58
  59 int max_connections=10;
  60 int max_queue_per_conn=200;
  61 int target_max_feedfile_size=100000;
  62 int period_seconds=30;
  63 int filepoll_seconds=5;
  64 int max_queue_per_ipf=-1;
  65
  66 int connection_setup_timeout=200;
  67 int inndcomm_flush_timeout=100;
  68
  69 double nocheck_thresh= 95.0; /* converted from percentage by main */
  70 double nocheck_decay= 100; /* conv'd from articles to lambda by main */
  71
  72 /* all these are initialised to seconds, and converted to periods in main */
  73 int reconnect_delay_periods=1000;
  74 int flushfail_retry_periods=1000;
  75 int backlog_retry_minperiods=100;
  76 int backlog_spontrescan_periods=300;
  77 int spontaneous_flush_periods=100000;
  78 int max_separated_periods=2000;
  79 int need_activity_periods=1000;
  80 int lowvol_thresh=3;
  81 int lowvol_periods=1000;
  82
  83 double max_bad_data_ratio= 1; /* conv'd from percentage by main */
  84 int max_bad_data_initial= 30;
  85   /* in one corrupt 4096-byte block the number of newlines has
  86    * mean 16 and standard deviation 3.99.  30 corresponds to z=+3.5 */
  87
  88 /*========== main program ==========*/
  89
  90 static void postfork_inputfile(InputFile *ipf) {
  91   if (!ipf) return;
  92   xclose(ipf->fd, "(in child) input file ", ipf->path);
  93 }
  94
  95 static void postfork_stdio(FILE *f, const char *what, const char *what2) {
  96   /* we have no stdio streams that are buffered long-term */
  97   if (!f) return;
  98   if (fclose(f)) syscrash("(in child) close %s%s", what, what2?what2:0);
  99 }
 100
 101 void postfork(void) {
 102   in_child= 1;
 103
 104   xsigsetdefault(SIGTERM);
 105   xsigsetdefault(SIGINT);
 106   xsigsetdefault(SIGPIPE);
 107   if (terminate_sig_flag) raise(terminate_sig_flag);
 108
 109   postfork_inputfile(main_input_file);
 110   postfork_inputfile(flushing_input_file);
 111
 112   Conn *conn;
 113   FOR_CONN(conn)
 114     conn_closefd(conn,"(in child) ");
 115
 116   postfork_stdio(defer, "defer file ", path_defer);
 117 }
 118
 119 typedef struct Every Every;
 120 struct Every {
 121   struct timeval interval;
 122   int fixed_rate;
 123   void (*f)(void);
 124 };
 125
 126 static void every_schedule(Every *e, struct timeval base);
 127
 128 static void *every_happens(oop_source *lp, struct timeval base, void *e_v) {
 129   Every *e= e_v;
 130   e->f();
 131   if (!e->fixed_rate) xgettimeofday(&base);
 132   every_schedule(e, base);
 133   return OOP_CONTINUE;
 134 }
 135
 136 static void every_schedule(Every *e, struct timeval base) {
 137   struct timeval when;
 138   timeradd(&base, &e->interval, &when);
 139   loop->on_time(loop, when, every_happens, e);
 140 }
 141
 142 static void every(int interval, int fixed_rate, void (*f)(void)) {
 143   NEW_DECL(Every *,e);
 144   e->interval.tv_sec= interval;
 145   e->interval.tv_usec= 0;
 146   e->fixed_rate= fixed_rate;
 147   e->f= f;
 148   struct timeval now;
 149   xgettimeofday(&now);
 150   every_schedule(e, now);
 151 }
 152
 153 void period(void) {
 154   char *dipf_main=     dbg_report_ipf(main_input_file);
 155   char *dipf_flushing= dbg_report_ipf(flushing_input_file);
 156   char *dipf_backlog=  dbg_report_ipf(backlog_input_file);
 157
 158   dbg("PERIOD"
 159       " sms=%s[%d] conns=%d until_connect=%d"
 160       " input_files main:%s flushing:%s backlog:%s[%d]"
 161       " children connecting=%ld inndcomm=%ld lowvol_total=%d"
 162       ,
 163       sms_names[sms], until_flush, conns.count, until_connect,
 164       dipf_main, dipf_flushing, dipf_backlog, until_backlog_nextscan,
 165       (long)connecting_child, (long)inndcomm_child, lowvol_total
 166       );
 167
 168   free(dipf_main);
 169   free(dipf_flushing);
 170   free(dipf_backlog);
 171
 172   if (until_connect) until_connect--;
 173
 174   inputfile_queue_check_expired(backlog_input_file);
 175   poll_backlog_file();
 176   if (!backlog_input_file) close_defer(); /* want to start on a new backlog */
 177   statemc_period_poll();
 178   check_assign_articles();
 179   check_idle_conns();
 180 }
 181
 182
 183 /*========== option parsing ==========*/
 184
 185 static void vbadusage(const char *fmt, va_list al) NORET_PRINTF(1,0);
 186 static void vbadusage(const char *fmt, va_list al) {
 187   char *m= xvasprintf(fmt,al);
 188   fprintf(stderr, "bad usage: %s\n"
 189           "say --help for help, or read the manpage\n",
 190           m);
 191   if (interactive < 2)
 192     syslog(LOG_ERR,"innduct: invoked with bad usage: %s",m);
 193   exit(8);
 194 }
 195
 196 /*---------- generic option parser ----------*/
 197
 198 static void badusage(const char *fmt, ...) NORET_PRINTF(1,2);
 199 static void badusage(const char *fmt, ...) {
 200   va_list al;
 201   va_start(al,fmt);
 202   vbadusage(fmt,al);
 203 }
 204
 205 enum OptFlags {
 206   of_seconds= 001000u,
 207   of_boolean= 002000u,
 208 };
 209
 210 typedef struct Option Option;
 211 typedef void OptionParser(const Option*, const char *val);
 212
 213 struct Option {
 214   int shrt;
 215   const char *lng, *formarg;
 216   void *store;
 217   OptionParser *fn;
 218   int intval;
 219 };
 220
 221 static void parse_options(const Option *options, char ***argvp) {
 222   /* on return *argvp is first non-option arg; argc is not updated */
 223
 224   for (;;) {
 225     const char *arg= *++(*argvp);
 226     if (!arg) break;
 227     if (*arg != '-') break;
 228     if (!strcmp(arg,"--")) { arg= *++(*argvp); break; }
 229     int a;
 230     while ((a= *++arg)) {
 231       const Option *o;
 232       if (a=='-') {
 233         arg++;
 234         char *equals= strchr(arg,'=');
 235         unsigned len= equals ? (size_t)(equals - arg) : strlen(arg);
 236         for (o=options; o->shrt || o->lng; o++)
 237           if (strlen(o->lng) == len && !memcmp(o->lng,arg,len))
 238             goto found_long;
 239         badusage("unknown long option --%s",arg);
 240       found_long:
 241         if (!o->formarg) {
 242           if (equals) badusage("option --%s does not take a value",o->lng);
 243           arg= 0;
 244         } else if (equals) {
 245           arg= equals+1;
 246         } else {
 247           arg= *++(*argvp);
 248           if (!arg) badusage("option --%s needs a value for %s",
 249                              o->lng, o->formarg);
 250         }
 251         o->fn(o, arg);
 252         break; /* eaten the whole argument now */
 253       }
 254       for (o=options; o->shrt || o->lng; o++)
 255         if (a == o->shrt)
 256           goto found_short;
 257       badusage("unknown short option -%c",a);
 258     found_short:
 259       if (!o->formarg) {
 260         o->fn(o,0);
 261       } else {
 262         if (!*++arg) {
 263           arg= *++(*argvp);
 264           if (!arg) badusage("option -%c needs a value for %s",
 265                              o->shrt, o->formarg);
 266         }
 267         o->fn(o,arg);
 268         break; /* eaten the whole argument now */
 269       }
 270     }
 271   }
 272 }
 273
 274 #define DELIMPERHAPS(delim,str)  (str) ? (delim) : "", (str) ? (str) : ""
 275
 276 static void print_options(const Option *options, FILE *f) {
 277   const Option *o;
 278   for (o=options; o->shrt || o->lng; o++) {
 279     char shrt[2] = { o->shrt, 0 };
 280     char *optspec= xasprintf("%s%s%s%s%s",
 281                              o->shrt ? "-" : "", shrt,
 282                              o->shrt && o->lng ? "|" : "",
 283                              DELIMPERHAPS("--", o->lng));
 284     fprintf(f, "  %s%s%s\n", optspec, DELIMPERHAPS(" ", o->formarg));
 285     free(optspec);
 286   }
 287 }
 288
 289 /*---------- specific option types ----------*/
 290
 291 static void op_integer(const Option *o, const char *val) {
 292   char *ep;
 293   errno= 0;
 294   unsigned long ul= strtoul(val,&ep,10);
 295   if (*ep || ep==val || errno || ul>INT_MAX)
 296     badusage("bad integer value for %s",o->lng);
 297   int *store= o->store;
 298   *store= ul;
 299 }
 300
 301 static void op_double(const Option *o, const char *val) {
 302   int *store= o->store;
 303   char *ep;
 304   errno= 0;
 305   *store= strtod(val, &ep);
 306   if (*ep || ep==val || errno)
 307     badusage("bad floating point value for %s",o->lng);
 308 }
 309
 310 static void op_string(const Option *o, const char *val) {
 311   const char **store= o->store;
 312   *store= val;
 313 }
 314
 315 static void op_seconds(const Option *o, const char *val) {
 316   int *store= o->store;
 317   char *ep;
 318   int unit;
 319
 320   double v= strtod(val,&ep);
 321   if (ep==val) badusage("bad time/duration value for %s",o->lng);
 322
 323   if (!*ep || !strcmp(ep,"s") || !strcmp(ep,"sec")) unit= 1;
 324   else if (!strcmp(ep,"m") || !strcmp(ep,"min"))    unit= 60;
 325   else if (!strcmp(ep,"h") || !strcmp(ep,"hour"))   unit= 3600;
 326   else if (!strcmp(ep,"d") || !strcmp(ep,"day"))    unit= 86400;
 327   else if (!strcmp(ep,"das")) unit= 10;
 328   else if (!strcmp(ep,"hs"))  unit= 100;
 329   else if (!strcmp(ep,"ks"))  unit= 1000;
 330   else if (!strcmp(ep,"Ms"))  unit= 1000000;
 331   else badusage("bad units %s for time/duration value for %s",ep,o->lng);
 332
 333   v *= unit;
 334   v= ceil(v);
 335   if (v > INT_MAX) badusage("time/duration value for %s out of range",o->lng);
 336   *store= v;
 337 }
 338
 339 static void op_setint(const Option *o, const char *val) {
 340   int *store= o->store;
 341   *store= o->intval;
 342 }
 343
 344 /*---------- specific options ----------*/
 345
 346 static void help(const Option *o, const char *val);
 347
 348 static const Option innduct_options[]= {
 349 {'f',"feedfile",         "F",     &feedfile,                 op_string      },
 350 {'q',"quiet-multiple",   0,       &quiet_multiple,           op_setint, 1   },
 351 {0,"no-daemon",          0,       &interactive,              op_setint, 1   },
 352 {0,"interactive",        0,       &interactive,              op_setint, 2   },
 353 {0,"no-streaming",       0,       &try_stream,               op_setint, 0   },
 354 {0,"no-filemon",         0,       &try_filemon,              op_setint, 0   },
 355 {'C',"inndconf",         "F",     &inndconffile,             op_string      },
 356 {'P',"port",             "PORT",  &port,                     op_integer     },
 357 {0,"chdir",              "DIR",   &path_run,                 op_string      },
 358 {0,"cli",            "DIR/|PATH", &path_cli,                 op_string      },
 359 {0,"help",               0,       0,                         help           },
 360
 361 {0,"max-connections",    "N",     &max_connections,          op_integer     },
 362 {0,"max-queue-per-conn", "N",     &max_queue_per_conn,       op_integer     },
 363 {0,"max-queue-per-file", "N",     &max_queue_per_ipf,        op_integer     },
 364 {0,"feedfile-flush-size","BYTES", &target_max_feedfile_size, op_integer     },
 365 {0,"period-interval",    "TIME",  &period_seconds,           op_seconds     },
 366
 367 {0,"connection-timeout",   "TIME",  &connection_setup_timeout, op_seconds   },
 368 {0,"stuck-flush-timeout",  "TIME",  &inndcomm_flush_timeout,   op_seconds   },
 369 {0,"feedfile-poll",        "TIME",  &filepoll_seconds,         op_seconds   },
 370
 371 {0,"no-check-proportion",   "PERCENT",   &nocheck_thresh,       op_double   },
 372 {0,"no-check-response-time","ARTICLES",  &nocheck_decay,        op_double   },
 373
 374 {0,"reconnect-interval",     "PERIOD", &reconnect_delay_periods,  op_seconds },
 375 {0,"flush-retry-interval",   "PERIOD", &flushfail_retry_periods,  op_seconds },
 376 {0,"earliest-deferred-retry","PERIOD", &backlog_retry_minperiods, op_seconds },
 377 {0,"backlog-rescan-interval","PERIOD",&backlog_spontrescan_periods,op_seconds},
 378 {0,"max-flush-interval",     "PERIOD", &spontaneous_flush_periods,op_seconds },
 379 {0,"flush-finish-timeout",   "PERIOD", &max_separated_periods,    op_seconds },
 380 {0,"idle-timeout",           "PERIOD", &need_activity_periods,    op_seconds },
 381 {0,"low-volume-thresh",      "PERIOD", &lowvol_thresh,            op_integer },
 382 {0,"low-volume-window",      "PERIOD", &lowvol_periods,           op_seconds },
 383
 384 {0,"max-bad-input-data-ratio","PERCENT", &max_bad_data_ratio,   op_double    },
 385 {0,"max-bad-input-data-init", "PERCENT", &max_bad_data_initial, op_integer   },
 386
 387 {0,0}
 388 };
 389
 390 static void printusage(FILE *f) {
 391   fputs("usage: innduct [options] site [fqdn]\n"
 392         "available options are:\n", f);
 393   print_options(innduct_options, f);
 394 }
 395
 396 static void help(const Option *o, const char *val) {
 397   printusage(stdout);
 398   if (ferror(stdout) || fflush(stdout)) {
 399     perror("innduct: writing help");
 400     exit(12);
 401   }
 402   exit(0);
 403 }
 404
 405 static void convert_to_periods_rndup(int *store) {
 406   *store += period_seconds-1;
 407   *store /= period_seconds;
 408 }
 409
 410 static int path_ends_slash(const char *specified) {
 411   int l= strlen(specified);
 412   assert(l);
 413   return specified[l-1] == '/';
 414 }
 415
 416 static int innduct_fatal_cleanup(void) { return 12; } /* used for libinn die */
 417
 418 int main(int argc, char **argv) {
 419   /* set up libinn logging */
 420   message_program_name= "innduct";
 421   message_fatal_cleanup= innduct_fatal_cleanup;
 422
 423 #define INNLOGSET_CALL(fn, pfx, sysloglevel)    \
 424   message_handlers_##fn(1, duct_log_##fn);
 425   INNLOGSETS(INNLOGSET_CALL)
 426
 427   if (!argv[1]) {
 428     printusage(stderr);
 429     exit(8);
 430   }
 431
 432   parse_options(innduct_options, &argv);
 433
 434   /* arguments */
 435
 436   sitename= *argv++;
 437   if (!sitename) badusage("need site name argument");
 438
 439   if (*argv) remote_host= *argv++;
 440   else remote_host= sitename;
 441
 442   if (*argv) badusage("too many non-option arguments");
 443
 444   /* defaults */
 445
 446   int r= innconf_read(inndconffile);
 447   if (!r) badusage("could not read inn.conf");
 448
 449   if (!remote_host) remote_host= sitename;
 450
 451   if (nocheck_thresh < 0 || nocheck_thresh > 100)
 452     badusage("nocheck threshold percentage must be between 0..100");
 453   nocheck_thresh *= 0.01;
 454
 455   if (nocheck_decay < 0.1)
 456     badusage("nocheck decay articles must be at least 0.1");
 457   nocheck_decay= pow(0.5, 1.0/nocheck_decay);
 458
 459   convert_to_periods_rndup(&reconnect_delay_periods);
 460   convert_to_periods_rndup(&flushfail_retry_periods);
 461   convert_to_periods_rndup(&backlog_retry_minperiods);
 462   convert_to_periods_rndup(&backlog_spontrescan_periods);
 463   convert_to_periods_rndup(&spontaneous_flush_periods);
 464   convert_to_periods_rndup(&max_separated_periods);
 465   convert_to_periods_rndup(&need_activity_periods);
 466   convert_to_periods_rndup(&lowvol_periods);
 467
 468   if (max_bad_data_ratio < 0 || max_bad_data_ratio > 100)
 469     badusage("bad input data ratio must be between 0..100");
 470   max_bad_data_ratio *= 0.01;
 471
 472   if (!path_run)
 473     path_run= innconf->pathrun;
 474
 475   if (!feedfile) feedfile= sitename;
 476   if (!feedfile[0]) badusage("feed filename, if specified, must be nonempty");
 477   if (path_ends_slash(feedfile))
 478     feedfile= xasprintf("%s%s", feedfile, sitename);
 479   if (feedfile[0] != '/')
 480     feedfile= xasprintf("%s/%s", innconf->pathoutgoing, feedfile);
 481
 482   if (!path_cli) {
 483     path_cli_dir= "innduct";
 484   } else if (!path_cli[0] || !strcmp(path_cli,"none")) {
 485     path_cli= 0; /* ok, don't then */
 486   } else if (path_ends_slash(path_cli)) {
 487     path_cli_dir= xasprintf("%.*s", strlen(path_cli)-1, path_cli);
 488   }
 489   if (path_cli_dir)
 490     path_cli= xasprintf("%s/%s", path_cli_dir, sitename);
 491
 492   if (max_queue_per_ipf<0)
 493     max_queue_per_ipf= max_queue_per_conn * 2;
 494
 495   const char *feedfile_forbidden= "?*[~#";
 496   int c;
 497   while ((c= *feedfile_forbidden++))
 498     if (strchr(feedfile, c))
 499       badusage("feed filename may not contain metacharacter %c",c);
 500
 501   int i;
 502   lowvol_perperiod= xcalloc(sizeof(*lowvol_perperiod), lowvol_periods);
 503   for (i=0; i<lowvol_periods; i++) {
 504     lowvol_perperiod[i]= lowvol_thresh;
 505     lowvol_total += lowvol_thresh;
 506   }
 507   lowvol_total -= lowvol_thresh;
 508
 509   /* set things up */
 510
 511   path_lock=        xasprintf("%s_lock",      feedfile);
 512   path_flushing=    xasprintf("%s_flushing",  feedfile);
 513   path_defer=       xasprintf("%s_defer",     feedfile);
 514   path_dump=        xasprintf("%s_dump",      feedfile);
 515   globpat_backlog=  xasprintf("%s_backlog*",  feedfile);
 516
 517   oop_source_sys *sysloop= oop_sys_new();
 518   if (!sysloop) syscrash("could not create liboop event loop");
 519   loop= (oop_source*)sysloop;
 520
 521   LIST_INIT(conns);
 522
 523   if (interactive < 1) {
 524     for (i=3; i<255; i++)
 525       /* do this now before we open syslog, etc. */
 526       close(i);
 527   }
 528
 529   logv_prefix= xasprintf("%s| ", sitename);
 530   if (interactive < 2) {
 531     openlog("innduct",LOG_NDELAY|LOG_PID,LOG_NEWS);
 532     logv_use_syslog= 1;
 533   }
 534
 535   if (interactive < 1) {
 536     int null= open("/dev/null",O_RDWR);
 537     if (null<0) sysdie("failed to open /dev/null");
 538     dup2(null,0);
 539     dup2(null,1);
 540     dup2(null,2);
 541     xclose(null, "/dev/null original fd",0);
 542
 543     pid_t child1= xfork("daemonise first fork");
 544     if (child1) _exit(0);
 545
 546     pid_t sid= setsid();
 547     if (sid == -1) sysdie("setsid failed");
 548
 549     pid_t child2= xfork("daemonise second fork");
 550     if (child2) _exit(0);
 551   }
 552
 553   self_pid= getpid();
 554   if (self_pid==-1) syscrash("getpid");
 555
 556   r= chdir(path_run);
 557   if (r) sysdie("could not chdir to pathrun %s", path_run);
 558
 559   statemc_lock();
 560
 561   init_signals();
 562
 563   notice("starting");
 564
 565   int val= 1;
 566   r= SMsetup(SM_PREOPEN, &val); if (!r) warn("SMsetup SM_PREOPEN failed");
 567   r= SMinit(); if (!r) die("storage manager initialisation (SMinit) failed");
 568
 569   if (interactive >= 2)
 570     cli_stdio();
 571
 572   cli_init();
 573
 574   int filemon_ok= 0;
 575   if (!try_filemon) {
 576     notice("filemon: suppressed by command line option, polling");
 577   } else {
 578     filemon_ok= filemon_method_init();
 579     if (!filemon_ok)
 580       warn("filemon: no file monitoring available, polling");
 581   }
 582   if (!filemon_ok)
 583     every(filepoll_seconds,0,filepoll);
 584
 585   every(period_seconds,1,period);
 586
 587   statemc_init();
 588
 589   /* let's go */
 590
 591   void *run= oop_sys_run(sysloop);
 592   assert(run == OOP_ERROR);
 593   syscrash("event loop failed");
 594 }