cli.c

   1 /*
   2  *  innduct
   3  *  tailing reliable realtime streaming feeder for inn
   4  *  cli.c - command and control connections
   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 /*========== command and control (CLI) connections ==========*/
  30
  31 static int cli_master;
  32
  33 typedef struct CliConn CliConn;
  34 struct CliConn {
  35   void (*destroy)(CliConn*);
  36   int fd;
  37   oop_read *rd;
  38   FILE *out;
  39   union {
  40     struct sockaddr sa;
  41     struct sockaddr_un un;
  42   } sa;
  43   socklen_t salen;
  44 };
  45
  46 static const oop_rd_style cli_rd_style= {
  47   OOP_RD_DELIM_STRIP, '\n',
  48   OOP_RD_NUL_FORBID,
  49   OOP_RD_SHORTREC_FORBID
  50 };
  51
  52 static void cli_destroy(CliConn *cc) {
  53   cc->destroy(cc);
  54 }
  55
  56 static void cli_checkouterr(CliConn *cc /* may destroy*/) {
  57   if (ferror(cc->out) | fflush(cc->out)) {
  58     info("CTRL%d write error %s", cc->fd, strerror(errno));
  59     cli_destroy(cc);
  60   }
  61 }
  62
  63 static void cli_prompt(CliConn *cc /* may destroy*/) {
  64   fprintf(cc->out, "%s| ", sitename);
  65   cli_checkouterr(cc);
  66 }
  67
  68 struct CliCommand {
  69   const char *cmd;
  70   void (*f)(CliConn *cc, const CliCommand *ccmd,
  71             const char *arg, size_t argsz);
  72   void *xdata;
  73   int xval;
  74 };
  75
  76 static const CliCommand cli_commands[];
  77
  78 #define CCMD(wh)                                                \
  79   static void ccmd_##wh(CliConn *cc, const CliCommand *c,       \
  80                         const char *arg, size_t argsz)
  81
  82 CCMD(help) {
  83   fputs("commands:\n", cc->out);
  84   const CliCommand *ccmd;
  85   for (ccmd=cli_commands; ccmd->cmd; ccmd++)
  86     fprintf(cc->out, " %s\n", ccmd->cmd);
  87   fputs("NB: permissible arguments are not shown above."
  88         "  Not all commands listed are safe.  See innduct(8).\n", cc->out);
  89 }
  90
  91 CCMD(flush) {
  92   int ok= trigger_flush_ok("manual request");
  93   if (!ok) fprintf(cc->out,"already flushing (state is %s)\n", sms_names[sms]);
  94 }
  95
  96 CCMD(stop) {
  97   preterminate();
  98   notice("terminating (CTRL%d)",cc->fd);
  99   raise_default(SIGTERM);
 100   abort();
 101 }
 102
 103 CCMD(dump);
 104
 105 /* messing with our head: */
 106 CCMD(period) { period(); }
 107 CCMD(setintarg) { *(int*)c->xdata= atoi(arg); }
 108 CCMD(setint) { *(int*)c->xdata= c->xval; }
 109 CCMD(setint_period) { *(int*)c->xdata= c->xval; period(); }
 110
 111 static const CliCommand cli_commands[]= {
 112   { "h",             ccmd_help      },
 113   { "flush",         ccmd_flush     },
 114   { "stop",          ccmd_stop      },
 115   { "dump q",        ccmd_dump, 0,0 },
 116   { "dump a",        ccmd_dump, 0,1 },
 117
 118   { "p",             ccmd_period    },
 119
 120 #define POKES(cmd,func)                                                 \
 121   { cmd "flush",     func,           &until_flush,             1 },     \
 122   { cmd "conn",      func,           &until_connect,           0 },     \
 123   { cmd "blscan",    func,           &until_backlog_nextscan,  0 },
 124 POKES("next ", ccmd_setint)
 125 POKES("prod ", ccmd_setint_period)
 126
 127   { "pretend flush", ccmd_setintarg, &simulate_flush             },
 128   { "wedge blscan",  ccmd_setint,    &until_backlog_nextscan, -1 },
 129   { 0 }
 130 };
 131
 132 static void *cli_rd_ok(oop_source *lp, oop_read *oread, oop_rd_event ev,
 133                        const char *errmsg, int errnoval,
 134                        const char *data, size_t recszu, void *cc_v) {
 135   CliConn *cc= cc_v;
 136
 137   if (!data) {
 138     info("CTRL%d closed", cc->fd);
 139     cc->destroy(cc);
 140     return OOP_CONTINUE;
 141   }
 142
 143   if (recszu == 0) goto prompt;
 144   assert(recszu <= INT_MAX);
 145   int recsz= recszu;
 146
 147   const CliCommand *ccmd;
 148   for (ccmd=cli_commands; ccmd->cmd; ccmd++) {
 149     int l= strlen(ccmd->cmd);
 150     if (recsz < l) continue;
 151     if (recsz > l && data[l] != ' ') continue;
 152     if (memcmp(data, ccmd->cmd, l)) continue;
 153
 154     int argl= (int)recsz - (l+1);
 155     ccmd->f(cc, ccmd, argl>=0 ? data+l+1 : 0, argl);
 156     goto prompt;
 157   }
 158
 159   fputs("unknown command; h for help\n", cc->out);
 160
 161  prompt:
 162   cli_prompt(cc);
 163   return OOP_CONTINUE;
 164 }
 165
 166 static void *cli_rd_err(oop_source *lp, oop_read *oread, oop_rd_event ev,
 167                         const char *errmsg, int errnoval,
 168                         const char *data, size_t recsz, void *cc_v) {
 169   CliConn *cc= cc_v;
 170
 171   info("CTRL%d read error %s", cc->fd, errmsg);
 172   cc->destroy(cc);
 173   return OOP_CONTINUE;
 174 }
 175
 176 static int cli_conn_startup(CliConn *cc /* may destroy*/,
 177                                 const char *how) {
 178   cc->rd= oop_rd_new_fd(loop, cc->fd, 0,0);
 179   if (!cc->rd) { warn("oop_rd_new_fd cli failed"); return -1; }
 180
 181   int er= oop_rd_read(cc->rd, &cli_rd_style, MAX_CLI_COMMAND,
 182                       cli_rd_ok, cc,
 183                       cli_rd_err, cc);
 184   if (er) { errno= er; syswarn("oop_rd_read cli failed"); return -1; }
 185
 186   info("CTRL%d %s ready", cc->fd, how);
 187   cli_prompt(cc);
 188   return 0;
 189 }
 190
 191 static void cli_stdio_destroy(CliConn *cc) {
 192   if (cc->rd) {
 193     oop_rd_cancel(cc->rd);
 194     errno= oop_rd_delete_tidy(cc->rd);
 195     if (errno) syswarn("oop_rd_delete tidy failed (no-nonblock stdin?)");
 196   }
 197   free(cc);
 198 }
 199
 200 void cli_stdio(void) {
 201   NEW_DECL(CliConn *,cc);
 202   cc->destroy= cli_stdio_destroy;
 203
 204   cc->fd= 0;
 205   cc->out= stdout;
 206   int r= cli_conn_startup(cc,"stdio");
 207   if (r) cc->destroy(cc);
 208 }
 209
 210 static void cli_accepted_destroy(CliConn *cc) {
 211   if (cc->rd) {
 212     oop_rd_cancel(cc->rd);
 213     oop_rd_delete_kill(cc->rd);
 214   }
 215   if (cc->out) { fclose(cc->out); cc->fd=0; }
 216   close_perhaps(&cc->fd);
 217   free(cc);
 218 }
 219
 220 static void *cli_master_readable(oop_source *lp, int master,
 221                                  oop_event ev, void *u) {
 222   NEW_DECL(CliConn *,cc);
 223   cc->destroy= cli_accepted_destroy;
 224
 225   cc->salen= sizeof(cc->sa);
 226   cc->fd= accept(master, &cc->sa.sa, &cc->salen);
 227   if (cc->fd<0) { syswarn("error accepting cli connection"); goto x; }
 228
 229   cc->out= fdopen(cc->fd, "w");
 230   if (!cc->out) { syswarn("error fdopening accepted cli connection"); goto x; }
 231
 232   int r= cli_conn_startup(cc, "accepted");
 233   if (r) goto x;
 234
 235   return OOP_CONTINUE;
 236
 237  x:
 238   cc->destroy(cc);
 239   return OOP_CONTINUE;
 240 }
 241
 242 #define NOCLI(...) do{                                          \
 243     syswarn("no cli listener, because failed to " __VA_ARGS__); \
 244     goto nocli;                                                 \
 245   }while(0)
 246
 247 void cli_init(void) {
 248   union {
 249     struct sockaddr sa;
 250     struct sockaddr_un un;
 251   } sa;
 252
 253   memset(&sa,0,sizeof(sa));
 254   int maxlen= sizeof(sa.un.sun_path);
 255
 256   if (!path_cli) {
 257     info("control command line disabled");
 258     return;
 259   }
 260
 261   int pathlen= strlen(path_cli);
 262   if (pathlen > maxlen) {
 263     warn("no cli listener, because cli socket path %s too long (%d>%d)",
 264          path_cli, pathlen, maxlen);
 265     return;
 266   }
 267
 268   if (path_cli_dir) {
 269     int r= mkdir(path_cli_dir, 0700);
 270     if (r && errno!=EEXIST)
 271       NOCLI("create cli socket directory %s", path_cli_dir);
 272   }
 273
 274   int r= unlink(path_cli);
 275   if (r && errno!=ENOENT)
 276     NOCLI("remove old cli socket %s", path_cli);
 277
 278   cli_master= socket(PF_UNIX, SOCK_STREAM, 0);
 279   if (cli_master<0) NOCLI("create new cli master socket");
 280
 281   int sl= pathlen + offsetof(struct sockaddr_un, sun_path);
 282   sa.un.sun_family= AF_UNIX;
 283   memcpy(sa.un.sun_path, path_cli, pathlen);
 284
 285   r= bind(cli_master, &sa.sa, sl);
 286   if (r) NOCLI("bind to cli socket path %s", sa.un.sun_path);
 287
 288   r= listen(cli_master, 5);
 289   if (r) NOCLI("listen to cli master socket");
 290
 291   xsetnonblock(cli_master, 1);
 292
 293   loop->on_fd(loop, cli_master, OOP_READ, cli_master_readable, 0);
 294   info("cli ready, listening on %s", path_cli);
 295
 296   return;
 297
 298  nocli:
 299   xclose_perhaps(&cli_master, "cli master",0);
 300   return;
 301 }
 302
 303 /*========== dumping state ==========*/
 304
 305 static void dump_article_list(FILE *f, const CliCommand *c,
 306                               const ArticleList *al) {
 307   fprintf(f, " count=%d\n", al->count);
 308   if (!c->xval) return;
 309
 310   int i; Article *art;
 311   for (i=0, art=LIST_HEAD(*al); art; i++, art=LIST_NEXT(art)) {
 312     fprintf(f," #%05d %-11s", i, artstate_names[art->state]);
 313     DUMPV("%p", art->,ipf);
 314     DUMPV("%d", art->,missing);
 315     DUMPV("%lu", (unsigned long)art->,offset);
 316     DUMPV("%d", art->,blanklen);
 317     DUMPV("%d", art->,midlen);
 318     fprintf(f, " %s %s\n", TokenToText(art->token), art->messageid);
 319   }
 320 }
 321
 322 static void dump_input_file(FILE *f, const CliCommand *c,
 323                             InputFile *ipf, const char *wh) {
 324   char *dipf= dbg_report_ipf(ipf);
 325   fprintf(f,"input %s %s", wh, dipf);
 326   free(dipf);
 327
 328   if (ipf) {
 329     DUMPV("%d", ipf->,readcount_ok);
 330     DUMPV("%d", ipf->,readcount_blank);
 331     DUMPV("%d", ipf->,readcount_err);
 332     DUMPV("%d", ipf->,count_nooffer_missing);
 333   }
 334   fprintf(f,"\n");
 335   if (ipf) {
 336     ArtState state; const char *const *statename;
 337     for (state=0, statename=artstate_names; *statename; state++,statename++) {
 338 #define RC_DUMP_FMT(x) " " #x "=%d"
 339 #define RC_DUMP_VAL(x) ,ipf->counts[state][RC_##x]
 340       fprintf(f,"input %s counts %-11s"
 341               RESULT_COUNTS(RC_DUMP_FMT,RC_DUMP_FMT) "\n",
 342               wh, *statename
 343               RESULT_COUNTS(RC_DUMP_VAL,RC_DUMP_VAL));
 344     }
 345     fprintf(f,"input %s queue", wh);
 346     dump_article_list(f,c,&ipf->queue);
 347   }
 348 }
 349
 350 CCMD(dump) {
 351   int i;
 352   fprintf(cc->out, "dumping state to %s\n", path_dump);
 353   FILE *f= fopen(path_dump, "w");
 354   if (!f) { fprintf(cc->out, "failed: open: %s\n", strerror(errno)); return; }
 355
 356   fprintf(f,"general");
 357   DUMPV("%s", sms_names,[sms]);
 358   DUMPV("%d", ,until_flush);
 359   DUMPV("%ld", (long),self_pid);
 360   DUMPV("%p", , defer);
 361   DUMPV("%d", , until_connect);
 362   DUMPV("%d", , until_backlog_nextscan);
 363   DUMPV("%d", , simulate_flush);
 364   fprintf(f,"\nnocheck");
 365   DUMPV("%#.10f", , accept_proportion);
 366   DUMPV("%d", , nocheck);
 367   DUMPV("%d", , nocheck_reported);
 368   fprintf(f,"\n");
 369
 370   fprintf(f,"special");
 371   DUMPV("%ld", (long),connecting_child);
 372   DUMPV("%d", , connecting_fdpass_sock);
 373   DUMPV("%d", , cli_master);
 374   fprintf(f,"\n");
 375
 376   fprintf(f,"lowvol");
 377   DUMPV("%d", , lowvol_circptr);
 378   DUMPV("%d", , lowvol_total);
 379   fprintf(f,":");
 380   for (i=0; i<lowvol_periods; i++) {
 381     fprintf(f," ");
 382     if (i==lowvol_circptr) fprintf(f,"*");
 383     fprintf(f,"%d",lowvol_perperiod[i]);
 384   }
 385   fprintf(f,"\n");
 386
 387   fprintf(f,"filemon ");
 388   filemon_method_dump_info(f);
 389
 390   dump_input_file(f,c, main_input_file,     "main"    );
 391   dump_input_file(f,c, flushing_input_file, "flushing");
 392   dump_input_file(f,c, backlog_input_file,  "backlog" );
 393
 394   fprintf(f,"conns count=%d\n", conns.count);
 395
 396   Conn *conn;
 397   FOR_CONN(conn) {
 398
 399     fprintf(f,"C%d",conn->fd);
 400     DUMPV("%p",conn->,rd);             DUMPV("%d",conn->,max_queue);
 401     DUMPV("%d",conn->,stream);         DUMPV("\"%s\"",conn->,quitting);
 402     DUMPV("%d",conn->,since_activity);
 403     fprintf(f,"\n");
 404
 405     fprintf(f,"C%d waiting", conn->fd); dump_article_list(f,c,&conn->waiting);
 406     fprintf(f,"C%d priority",conn->fd); dump_article_list(f,c,&conn->priority);
 407     fprintf(f,"C%d sent",    conn->fd); dump_article_list(f,c,&conn->sent);
 408
 409     fprintf(f,"C%d xmit xmitu=%d\n", conn->fd, conn->xmitu);
 410     for (i=0; i<conn->xmitu; i++) {
 411       const struct iovec *iv= &conn->xmit[i];
 412       const XmitDetails *xd= &conn->xmitd[i];
 413       char *dinfo;
 414       switch (xd->kind) {
 415       case xk_Const:    dinfo= xasprintf("Const");                 break;
 416       case xk_Artdata:  dinfo= xasprintf("A%p", xd->info.sm_art);  break;
 417       default:
 418         abort();
 419       }
 420       fprintf(f," #%03d %-11s l=%d %s\n", i, dinfo, iv->iov_len,
 421               sanitise(iv->iov_base, iv->iov_len));
 422       free(dinfo);
 423     }
 424   }
 425
 426   fprintf(f,"paths");
 427   DUMPV("%s", , feedfile);
 428   DUMPV("%s", , path_cli);
 429   DUMPV("%s", , path_lock);
 430   DUMPV("%s", , path_flushing);
 431   DUMPV("%s", , path_defer);
 432   DUMPV("%s", , path_dump);
 433   DUMPV("%s", , globpat_backlog);
 434   fprintf(f,"\n");
 435
 436   if (!!ferror(f) + !!fclose(f)) {
 437     fprintf(cc->out, "failed: write: %s\n", strerror(errno));
 438     return;
 439   }
 440 }