cli.c

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