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