src/admin.c

   1 /*
   2  * PgBouncer - Lightweight connection pooler for PostgreSQL.
   3  *
   4  * Copyright (c) 2007-2009  Marko Kreen, Skype Technologies OÜ
   5  *
   6  * Permission to use, copy, modify, and/or distribute this software for any
   7  * purpose with or without fee is hereby granted, provided that the above
   8  * copyright notice and this permission notice appear in all copies.
   9  *
  10  * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
  11  * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  12  * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
  13  * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
  14  * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
  15  * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
  16  * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
  17  */
  18
  19 /*
  20  * Admin console commands.
  21  */
  22
  23 #include "bouncer.h"
  24
  25 #include <regex.h>
  26
  27 /* regex elements */
  28 #define WS0     "[ \t\n\r]*"
  29 #define WS1     "[ \t\n\r]+"
  30 #define WORD    "([0-9a-z_]+)"
  31 #define STRING  "'(([^']*|'')*)'"
  32
  33 /* possible max + 1 */
  34 #define MAX_GROUPS 10
  35
  36 /* group numbers */
  37 #define CMD_NAME 1
  38 #define CMD_ARG 3
  39 #define SET_KEY 1
  40 #define SET_VAL 3
  41
  42 typedef bool (*cmd_func_t)(PgSocket *admin, const char *arg);
  43 struct cmd_lookup {
  44         const char *word;
  45         cmd_func_t func;
  46 };
  47
  48 /* CMD [arg]; */
  49 static const char cmd_normal_rx[] =
  50 "^" WS0 WORD "(" WS1 WORD ")?" WS0 "(;" WS0 ")?$";
  51
  52 /* SET with simple value */
  53 static const char cmd_set_word_rx[] =
  54 "^" WS0 "set" WS1 WORD WS0 "(=|to)" WS0 WORD WS0 "(;" WS0 ")?$";
  55
  56 /* SET with quoted value */
  57 static const char cmd_set_str_rx[] =
  58 "^" WS0 "set" WS1 WORD WS0 "(=|to)" WS0 STRING WS0 "(;" WS0 ")?$";
  59
  60 /* compiled regexes */
  61 static regex_t rc_cmd;
  62 static regex_t rc_set_word;
  63 static regex_t rc_set_str;
  64
  65 static PgPool *admin_pool;
  66
  67 /* only valid during processing */
  68 static const char *current_query;
  69
  70 static bool syntax_error(PgSocket *admin)
  71 {
  72         return admin_error(admin, "invalid command '%s', use SHOW HELP;",
  73                            current_query ? current_query : "<no query>");
  74 }
  75
  76 static bool exec_cmd(struct cmd_lookup *lookup, PgSocket *admin,
  77                      const char *cmd, const char *arg)
  78 {
  79         for (; lookup->word; lookup++) {
  80                 if (strcasecmp(lookup->word, cmd) == 0)
  81                         return lookup->func(admin, arg);
  82         }
  83         return syntax_error(admin);
  84 }
  85
  86 bool admin_error(PgSocket *admin, const char *fmt, ...)
  87 {
  88         char str[1024];
  89         va_list ap;
  90         bool res = true;
  91
  92         va_start(ap, fmt);
  93         vsnprintf(str, sizeof(str), fmt, ap);
  94         va_end(ap);
  95
  96         log_error("%s", str);
  97         if (admin)
  98                 res = send_pooler_error(admin, true, str);
  99         return res;
 100 }
 101
 102 static int count_paused_databases(void)
 103 {
 104         List *item;
 105         PgDatabase *db;
 106         int cnt = 0;
 107
 108         statlist_for_each(item, &database_list) {
 109                 db = container_of(item, PgDatabase, head);
 110                 cnt += db->db_paused;
 111         }
 112         return cnt;
 113 }
 114
 115 static int count_db_active(PgDatabase *db)
 116 {
 117         List *item;
 118         PgPool *pool;
 119         int cnt = 0;
 120
 121         statlist_for_each(item, &pool_list) {
 122                 pool = container_of(item, PgPool, head);
 123                 if (pool->db != db)
 124                         continue;
 125                 cnt += pool_server_count(pool);
 126         }
 127         return cnt;
 128 }
 129
 130 bool admin_flush(PgSocket *admin, PktBuf *buf, const char *desc)
 131 {
 132         pktbuf_write_CommandComplete(buf, desc);
 133         pktbuf_write_ReadyForQuery(buf);
 134         return pktbuf_send_queued(buf, admin);
 135 }
 136
 137 bool admin_ready(PgSocket *admin, const char *desc)
 138 {
 139         PktBuf buf;
 140         uint8_t tmp[512];
 141         pktbuf_static(&buf, tmp, sizeof(tmp));
 142         pktbuf_write_CommandComplete(&buf, desc);
 143         pktbuf_write_ReadyForQuery(&buf);
 144         return pktbuf_send_immidiate(&buf, admin);
 145 }
 146
 147 /*
 148  * some silly clients start actively messing with server parameters
 149  * without checking if thats necessary.  Fake some env for them.
 150  */
 151 struct FakeParam {
 152         const char *name;
 153         const char *value;
 154 };
 155
 156 static const struct FakeParam fake_param_list[] = {
 157         { "client_encoding", "UTF-8" },
 158         { "default_transaction_isolation", "read committed" },
 159         { "datestyle", "ISO" },
 160         { "timezone", "GMT" },
 161         { NULL },
 162 };
 163
 164 /* fake result send, returns if handled */
 165 static bool fake_show(PgSocket *admin, const char *name)
 166 {
 167         PktBuf *buf;
 168         const struct FakeParam *p;
 169         bool got = false;
 170
 171         for (p = fake_param_list; p->name; p++) {
 172                 if (strcasecmp(name, p->name) == 0) {
 173                         got = true;
 174                         break;
 175                 }
 176         }
 177
 178         if (got) {
 179                 buf = pktbuf_dynamic(256);
 180                 if (buf) {
 181                         pktbuf_write_RowDescription(buf, "s", p->name);
 182                         pktbuf_write_DataRow(buf, "s", p->value);
 183                         admin_flush(admin, buf, "SHOW");
 184                 } else
 185                         admin_error(admin, "no mem");
 186         }
 187         return got;
 188 }
 189
 190 static bool fake_set(PgSocket *admin, const char *key, const char *val)
 191 {
 192         PktBuf *buf;
 193         const struct FakeParam *p;
 194         bool got = false;
 195
 196         for (p = fake_param_list; p->name; p++) {
 197                 if (strcasecmp(key, p->name) == 0) {
 198                         got = true;
 199                         break;
 200                 }
 201         }
 202
 203         if (got) {
 204                 buf = pktbuf_dynamic(256);
 205                 if (buf) {
 206                         pktbuf_write_Notice(buf, "SET ignored");
 207                         admin_flush(admin, buf, "SET");
 208                 } else
 209                         admin_error(admin, "no mem");
 210         }
 211         return got;
 212 }
 213
 214 /* Command: SET key = val; */
 215 static bool admin_set(PgSocket *admin, const char *key, const char *val)
 216 {
 217         char tmp[512];
 218
 219         if (fake_set(admin, key, val))
 220                 return true;
 221
 222         if (admin->admin_user) {
 223                 if (set_config_param(bouncer_params, key, val, true, admin)) {
 224                         snprintf(tmp, sizeof(tmp), "SET %s=%s", key, val);
 225                         return admin_ready(admin, tmp);
 226                 } else {
 227                         return admin_error(admin, "SET failed");
 228                 }
 229         } else
 230                 return admin_error(admin, "admin access needed");
 231 }
 232
 233 /* send a row with sendmsg, optionally attaching a fd */
 234 static bool send_one_fd(PgSocket *admin,
 235                         int fd, const char *task,
 236                         const char *user, const char *db,
 237                         const char *addr, int port,
 238                         uint64_t ckey, int link,
 239                         const char *client_enc,
 240                         const char *std_strings,
 241                         const char *datestyle,
 242                         const char *timezone)
 243 {
 244         struct msghdr msg;
 245         struct cmsghdr *cmsg;
 246         int res;
 247         struct iovec iovec;
 248         uint8_t pktbuf[STARTUP_BUF * 2];
 249         uint8_t cntbuf[CMSG_SPACE(sizeof(int))];
 250
 251         iovec.iov_base = pktbuf;
 252         BUILD_DataRow(res, pktbuf, sizeof(pktbuf), "issssiqissss",
 253                       fd, task, user, db, addr, port, ckey, link,
 254                       client_enc, std_strings, datestyle, timezone);
 255         if (res < 0)
 256                 return false;
 257         iovec.iov_len = res;
 258
 259         /* sending fds */
 260         memset(&msg, 0, sizeof(msg));
 261         msg.msg_iov = &iovec;
 262         msg.msg_iovlen = 1;
 263
 264         /* attach a fd */
 265         if (admin->remote_addr.is_unix && admin->own_user) {
 266                 msg.msg_control = cntbuf;
 267                 msg.msg_controllen = sizeof(cntbuf);
 268
 269                 cmsg = CMSG_FIRSTHDR(&msg);
 270                 cmsg->cmsg_level = SOL_SOCKET;
 271                 cmsg->cmsg_type = SCM_RIGHTS;
 272                 cmsg->cmsg_len = CMSG_LEN(sizeof(int));
 273
 274                 memcpy(CMSG_DATA(cmsg), &fd, sizeof(int));
 275                 msg.msg_controllen = cmsg->cmsg_len;
 276         }
 277
 278         slog_debug(admin, "sending socket list: fd=%d, len=%d",
 279                    fd, (int)msg.msg_controllen);
 280         res = safe_sendmsg(sbuf_socket(&admin->sbuf), &msg, 0);
 281         if (res < 0) {
 282                 log_error("send_one_fd: sendmsg error: %s", strerror(errno));
 283                 return false;
 284         } else if ((size_t)res != iovec.iov_len) {
 285                 log_error("send_one_fd: partial sendmsg");
 286                 return false;
 287         }
 288         return true;
 289 }
 290
 291 /* send a row with sendmsg, optionally attaching a fd */
 292 static bool show_one_fd(PgSocket *admin, PgSocket *sk)
 293 {
 294         PgAddr *addr = &sk->remote_addr;
 295         MBuf tmp;
 296         VarCache *v = &sk->vars;
 297
 298         mbuf_init(&tmp, sk->cancel_key, 8);
 299
 300         return send_one_fd(admin, sbuf_socket(&sk->sbuf),
 301                            is_server_socket(sk) ? "server" : "client",
 302                            sk->auth_user ? sk->auth_user->name : NULL,
 303                            sk->pool ? sk->pool->db->name : NULL,
 304                            addr->is_unix ? "unix" : inet_ntoa(addr->ip_addr),
 305                            addr->port,
 306                            mbuf_get_uint64(&tmp),
 307                            sk->link ? sbuf_socket(&sk->link->sbuf) : 0,
 308                            v->client_encoding[0] ? v->client_encoding : NULL,
 309                            v->std_strings[0] ? v->std_strings : NULL,
 310                            v->datestyle[0] ? v->datestyle : NULL,
 311                            v->timezone[0] ? v->timezone : NULL);
 312 }
 313
 314 /* send a row with sendmsg, optionally attaching a fd */
 315 static bool show_pooler_fds(PgSocket *admin)
 316 {
 317         int fd_net, fd_unix;
 318         bool res = true;
 319
 320         get_pooler_fds(&fd_net, &fd_unix);
 321
 322         if (fd_net)
 323                 res = send_one_fd(admin, fd_net, "pooler", NULL, NULL,
 324                                   cf_listen_addr, cf_listen_port, 0, 0,
 325                                   NULL, NULL, NULL, NULL);
 326         if (fd_unix && res)
 327                 res = send_one_fd(admin, fd_unix, "pooler", NULL, NULL,
 328                                   "unix", cf_listen_port, 0, 0,
 329                                   NULL, NULL, NULL, NULL);
 330         return res;
 331 }
 332
 333 static bool show_fds_from_list(PgSocket *admin, StatList *list)
 334 {
 335         List *item;
 336         PgSocket *sk;
 337         bool res = true;
 338
 339         statlist_for_each(item, list) {
 340                 sk = container_of(item, PgSocket, head);
 341                 res = show_one_fd(admin, sk);
 342                 if (!res)
 343                         break;
 344         }
 345         return res;
 346 }
 347
 348 /*
 349  * Command: SHOW FDS
 350  *
 351  * If privileged connection, send also actual fds
 352  */
 353 static bool admin_show_fds(PgSocket *admin, const char *arg)
 354 {
 355         List *item;
 356         PgPool *pool;
 357         bool res;
 358
 359         /*
 360          * Dangerous to show to everybody:
 361          * - can lock pooler as code flips async option
 362          * - show cancel keys for all users
 363          */
 364         if (!admin->admin_user)
 365                 return admin_error(admin, "admin access needed");
 366
 367         /*
 368          * It's very hard to send it reliably over in async manner,
 369          * so turn async off for this resultset.
 370          */
 371         socket_set_nonblocking(sbuf_socket(&admin->sbuf), 0);
 372
 373         /*
 374          * send resultset
 375          */
 376         SEND_RowDescription(res, admin, "issssiqissss",
 377                                  "fd", "task",
 378                                  "user", "database",
 379                                  "addr", "port",
 380                                  "cancel", "link",
 381                                  "client_encoding", "std_strings",
 382                                  "datestyle", "timezone");
 383         if (res)
 384                 res = show_pooler_fds(admin);
 385
 386         if (res)
 387                 res = show_fds_from_list(admin, &login_client_list);
 388
 389         statlist_for_each(item, &pool_list) {
 390                 pool = container_of(item, PgPool, head);
 391                 if (pool->db->admin)
 392                         continue;
 393                 res = res && show_fds_from_list(admin, &pool->active_client_list);
 394                 res = res && show_fds_from_list(admin, &pool->waiting_client_list);
 395                 res = res && show_fds_from_list(admin, &pool->active_server_list);
 396                 res = res && show_fds_from_list(admin, &pool->idle_server_list);
 397                 res = res && show_fds_from_list(admin, &pool->used_server_list);
 398                 res = res && show_fds_from_list(admin, &pool->tested_server_list);
 399                 res = res && show_fds_from_list(admin, &pool->new_server_list);
 400                 if (!res)
 401                         break;
 402         }
 403         if (res)
 404                 res = admin_ready(admin, "SHOW");
 405
 406         /* turn async back on */
 407         socket_set_nonblocking(sbuf_socket(&admin->sbuf), 1);
 408
 409         return res;
 410 }
 411
 412 /* Command: SHOW DATABASES */
 413 static bool admin_show_databases(PgSocket *admin, const char *arg)
 414 {
 415         PgDatabase *db;
 416         List *item;
 417         char *host;
 418         const char *f_user;
 419         PktBuf *buf;
 420
 421         buf = pktbuf_dynamic(256);
 422         if (!buf) {
 423                 admin_error(admin, "no mem");
 424                 return true;
 425         }
 426
 427         pktbuf_write_RowDescription(buf, "ssissii",
 428                                     "name", "host", "port",
 429                                     "database", "force_user", "pool_size", "reserve_pool");
 430         statlist_for_each(item, &database_list) {
 431                 db = container_of(item, PgDatabase, head);
 432
 433                 if (!db->addr.is_unix) {
 434                         host = inet_ntoa(db->addr.ip_addr);
 435                 } else
 436                         host = NULL;
 437
 438                 f_user = db->forced_user ? db->forced_user->name : NULL;
 439                 pktbuf_write_DataRow(buf, "ssissii",
 440                                      db->name, host, db->addr.port,
 441                                      db->dbname, f_user,
 442                                      db->pool_size,
 443                                      db->res_pool_size);
 444         }
 445         admin_flush(admin, buf, "SHOW");
 446         return true;
 447 }
 448
 449
 450 /* Command: SHOW LISTS */
 451 static bool admin_show_lists(PgSocket *admin, const char *arg)
 452 {
 453         PktBuf *buf = pktbuf_dynamic(256);
 454         if (!buf) {
 455                 admin_error(admin, "no mem");
 456                 return true;
 457         }
 458         pktbuf_write_RowDescription(buf, "si", "list", "items");
 459 #define SENDLIST(name, size) pktbuf_write_DataRow(buf, "si", (name), (size))
 460         SENDLIST("databases", statlist_count(&database_list));
 461         SENDLIST("users", statlist_count(&user_list));
 462         SENDLIST("pools", statlist_count(&pool_list));
 463         SENDLIST("free_clients", objcache_free_count(client_cache));
 464         SENDLIST("used_clients", objcache_active_count(client_cache));
 465         SENDLIST("login_clients", statlist_count(&login_client_list));
 466         SENDLIST("free_servers", objcache_free_count(server_cache));
 467         SENDLIST("used_servers", objcache_active_count(server_cache));
 468         admin_flush(admin, buf, "SHOW");
 469         return true;
 470 }
 471
 472 /* Command: SHOW USERS */
 473 static bool admin_show_users(PgSocket *admin, const char *arg)
 474 {
 475         PgUser *user;
 476         List *item;
 477         PktBuf *buf = pktbuf_dynamic(256);
 478         if (!buf) {
 479                 admin_error(admin, "no mem");
 480                 return true;
 481         }
 482         pktbuf_write_RowDescription(buf, "s", "name");
 483         statlist_for_each(item, &user_list) {
 484                 user = container_of(item, PgUser, head);
 485                 pktbuf_write_DataRow(buf, "s", user->name);
 486         }
 487         admin_flush(admin, buf, "SHOW");
 488         return true;
 489 }
 490
 491 #define SKF_STD "sssssisiTTss"
 492 #define SKF_DBG "sssssisiTTssiiiiiii"
 493
 494 static void socket_header(PktBuf *buf, bool debug)
 495 {
 496         pktbuf_write_RowDescription(buf, debug ? SKF_DBG : SKF_STD,
 497                                     "type", "user", "database", "state",
 498                                     "addr", "port", "local_addr", "local_port",
 499                                     "connect_time", "request_time",
 500                                     "ptr", "link",
 501                                     "recv_pos", "pkt_pos", "pkt_remain",
 502                                     "send_pos", "send_remain",
 503                                     "pkt_avail", "send_avail");
 504 }
 505
 506 static void adr2txt(const PgAddr *adr, char *dst, unsigned dstlen)
 507 {
 508         if (adr->is_unix) {
 509                 safe_strcpy(dst, "unix", dstlen);
 510         } else {
 511                 char *tmp = inet_ntoa(adr->ip_addr);
 512                 safe_strcpy(dst, tmp, dstlen);
 513         }
 514 }
 515
 516 static void socket_row(PktBuf *buf, PgSocket *sk, const char *state, bool debug)
 517 {
 518         int pkt_avail = 0, send_avail = 0;
 519         char ptrbuf[128], linkbuf[128];
 520         char l_addr[32], r_addr[32];
 521         IOBuf *io = sk->sbuf.io;
 522
 523         if (io) {
 524                 pkt_avail = iobuf_amount_parse(sk->sbuf.io);
 525                 send_avail = iobuf_amount_pending(sk->sbuf.io);
 526         }
 527
 528         adr2txt(&sk->remote_addr, r_addr, sizeof(r_addr));
 529         adr2txt(&sk->local_addr, l_addr, sizeof(l_addr));
 530
 531         snprintf(ptrbuf, sizeof(ptrbuf), "%p", sk);
 532         if (sk->link)
 533                 snprintf(linkbuf, sizeof(linkbuf), "%p", sk->link);
 534         else
 535                 linkbuf[0] = 0;
 536
 537         pktbuf_write_DataRow(buf, debug ? SKF_DBG : SKF_STD,
 538                              is_server_socket(sk) ? "S" :"C",
 539                              sk->auth_user ? sk->auth_user->name : "(nouser)",
 540                              sk->pool ? sk->pool->db->name : "(nodb)",
 541                              state, r_addr, sk->remote_addr.port,
 542                              l_addr, sk->local_addr.port,
 543                              sk->connect_time,
 544                              sk->request_time,
 545                              ptrbuf, linkbuf,
 546                              io ? io->recv_pos : 0,
 547                              io ? io->parse_pos : 0,
 548                              sk->sbuf.pkt_remain,
 549                              io ? io->done_pos : 0,
 550                              0,
 551                              pkt_avail, send_avail);
 552 }
 553
 554 /* Helper for SHOW CLIENTS */
 555 static void show_socket_list(PktBuf *buf, StatList *list, const char *state, bool debug)
 556 {
 557         List *item;
 558         PgSocket *sk;
 559
 560         statlist_for_each(item, list) {
 561                 sk = container_of(item, PgSocket, head);
 562                 socket_row(buf, sk, state, debug);
 563         }
 564 }
 565
 566 /* Command: SHOW CLIENTS */
 567 static bool admin_show_clients(PgSocket *admin, const char *arg)
 568 {
 569         List *item;
 570         PgPool *pool;
 571         PktBuf *buf = pktbuf_dynamic(256);
 572
 573         if (!buf) {
 574                 admin_error(admin, "no mem");
 575                 return true;
 576         }
 577
 578         socket_header(buf, false);
 579         statlist_for_each(item, &pool_list) {
 580                 pool = container_of(item, PgPool, head);
 581
 582                 show_socket_list(buf, &pool->active_client_list, "active", false);
 583                 show_socket_list(buf, &pool->waiting_client_list, "waiting", false);
 584         }
 585
 586         admin_flush(admin, buf, "SHOW");
 587         return true;
 588 }
 589
 590 /* Command: SHOW SERVERS */
 591 static bool admin_show_servers(PgSocket *admin, const char *arg)
 592 {
 593         List *item;
 594         PgPool *pool;
 595         PktBuf *buf;
 596
 597         buf = pktbuf_dynamic(256);
 598         if (!buf) {
 599                 admin_error(admin, "no mem");
 600                 return true;
 601         }
 602
 603         socket_header(buf, false);
 604         statlist_for_each(item, &pool_list) {
 605                 pool = container_of(item, PgPool, head);
 606                 show_socket_list(buf, &pool->active_server_list, "active", false);
 607                 show_socket_list(buf, &pool->idle_server_list, "idle", false);
 608                 show_socket_list(buf, &pool->used_server_list, "used", false);
 609                 show_socket_list(buf, &pool->tested_server_list, "tested", false);
 610                 show_socket_list(buf, &pool->new_server_list, "new", false);
 611         }
 612         admin_flush(admin, buf, "SHOW");
 613         return true;
 614 }
 615
 616 /* Command: SHOW SOCKETS */
 617 static bool admin_show_sockets(PgSocket *admin, const char *arg)
 618 {
 619         List *item;
 620         PgPool *pool;
 621         PktBuf *buf;
 622
 623         buf = pktbuf_dynamic(256);
 624         if (!buf) {
 625                 admin_error(admin, "no mem");
 626                 return true;
 627         }
 628
 629         socket_header(buf, true);
 630         statlist_for_each(item, &pool_list) {
 631                 pool = container_of(item, PgPool, head);
 632                 show_socket_list(buf, &pool->active_client_list, "cl_active", true);
 633                 show_socket_list(buf, &pool->waiting_client_list, "cl_waiting", true);
 634
 635                 show_socket_list(buf, &pool->active_server_list, "sv_active", true);
 636                 show_socket_list(buf, &pool->idle_server_list, "sv_idle", true);
 637                 show_socket_list(buf, &pool->used_server_list, "sv_used", true);
 638                 show_socket_list(buf, &pool->tested_server_list, "sv_tested", true);
 639                 show_socket_list(buf, &pool->new_server_list, "sv_login", true);
 640         }
 641         show_socket_list(buf, &login_client_list, "cl_login", true);
 642         admin_flush(admin, buf, "SHOW");
 643         return true;
 644 }
 645
 646 static void show_active_socket_list(PktBuf *buf, StatList *list, const char *state)
 647 {
 648         List *item;
 649         statlist_for_each(item, list) {
 650                 PgSocket *sk = container_of(item, PgSocket, head);
 651                 if (!sbuf_is_empty(&sk->sbuf))
 652                         socket_row(buf, sk, state, true);
 653         }
 654 }
 655
 656 /* Command: SHOW ACTIVE_SOCKETS */
 657 static bool admin_show_active_sockets(PgSocket *admin, const char *arg)
 658 {
 659         List *item;
 660         PgPool *pool;
 661         PktBuf *buf;
 662
 663         buf = pktbuf_dynamic(256);
 664         if (!buf) {
 665                 admin_error(admin, "no mem");
 666                 return true;
 667         }
 668
 669         socket_header(buf, true);
 670         statlist_for_each(item, &pool_list) {
 671                 pool = container_of(item, PgPool, head);
 672                 show_active_socket_list(buf, &pool->active_client_list, "cl_active");
 673                 show_active_socket_list(buf, &pool->waiting_client_list, "cl_waiting");
 674
 675                 show_active_socket_list(buf, &pool->active_server_list, "sv_active");
 676                 show_active_socket_list(buf, &pool->idle_server_list, "sv_idle");
 677                 show_active_socket_list(buf, &pool->used_server_list, "sv_used");
 678                 show_active_socket_list(buf, &pool->tested_server_list, "sv_tested");
 679                 show_active_socket_list(buf, &pool->new_server_list, "sv_login");
 680         }
 681         show_active_socket_list(buf, &login_client_list, "cl_login");
 682         admin_flush(admin, buf, "SHOW");
 683         return true;
 684 }
 685
 686 /* Command: SHOW POOLS */
 687 static bool admin_show_pools(PgSocket *admin, const char *arg)
 688 {
 689         List *item;
 690         PgPool *pool;
 691         PktBuf *buf;
 692         PgSocket *waiter;
 693         usec_t now = get_cached_time();
 694
 695         buf = pktbuf_dynamic(256);
 696         if (!buf) {
 697                 admin_error(admin, "no mem");
 698                 return true;
 699         }
 700         pktbuf_write_RowDescription(buf, "ssiiiiiiii",
 701                                     "database", "user",
 702                                     "cl_active", "cl_waiting",
 703                                     "sv_active", "sv_idle",
 704                                     "sv_used", "sv_tested",
 705                                     "sv_login", "maxwait");
 706         statlist_for_each(item, &pool_list) {
 707                 pool = container_of(item, PgPool, head);
 708                 waiter = first_socket(&pool->waiting_client_list);
 709                 pktbuf_write_DataRow(buf, "ssiiiiiiii",
 710                                      pool->db->name, pool->user->name,
 711                                      statlist_count(&pool->active_client_list),
 712                                      statlist_count(&pool->waiting_client_list),
 713                                      statlist_count(&pool->active_server_list),
 714                                      statlist_count(&pool->idle_server_list),
 715                                      statlist_count(&pool->used_server_list),
 716                                      statlist_count(&pool->tested_server_list),
 717                                      statlist_count(&pool->new_server_list),
 718                                      /* how long is the oldest client waited */
 719                                      (waiter && waiter->query_start)
 720                                      ?  (int)((now - waiter->query_start) / USEC) : 0);
 721         }
 722         admin_flush(admin, buf, "SHOW");
 723         return true;
 724 }
 725
 726 static void slab_stat_cb(void *arg, const char *slab_name,
 727                          unsigned size, unsigned free,
 728                          unsigned total)
 729 {
 730         PktBuf *buf = arg;
 731         unsigned alloc = total * size;
 732         pktbuf_write_DataRow(buf, "siiii", slab_name,
 733                              size, total - free, free, alloc);
 734 }
 735
 736 /* Command: SHOW MEM */
 737 static bool admin_show_mem(PgSocket *admin, const char *arg)
 738 {
 739         PktBuf *buf;
 740
 741         buf = pktbuf_dynamic(256);
 742         if (!buf) {
 743                 admin_error(admin, "no mem");
 744                 return true;
 745         }
 746         pktbuf_write_RowDescription(buf, "siiii", "name",
 747                                     "size", "used", "free", "memtotal");
 748         objcache_stats(slab_stat_cb, buf);
 749         admin_flush(admin, buf, "SHOW");
 750         return true;
 751 }
 752
 753 /* Command: SHOW CONFIG */
 754 static bool admin_show_config(PgSocket *admin, const char *arg)
 755 {
 756         ConfElem *cf;
 757         int i = 0;
 758         PktBuf *buf;
 759
 760         buf = pktbuf_dynamic(256);
 761         if (!buf) {
 762                 admin_error(admin, "no mem");
 763                 return true;
 764         }
 765
 766         pktbuf_write_RowDescription(buf, "sss", "key", "value", "changeable");
 767         while (1) {
 768                 cf = &bouncer_params[i++];
 769                 if (!cf->name)
 770                         break;
 771
 772                 pktbuf_write_DataRow(buf, "sss",
 773                                      cf->name, conf_to_text(cf),
 774                                      cf->reloadable ? "yes" : "no");
 775         }
 776         admin_flush(admin, buf, "SHOW");
 777         return true;
 778 }
 779
 780 /* Command: RELOAD */
 781 static bool admin_cmd_reload(PgSocket *admin, const char *arg)
 782 {
 783         if (arg && *arg)
 784                 return syntax_error(admin);
 785
 786         if (!admin->admin_user)
 787                 return admin_error(admin, "admin access needed");
 788
 789         log_info("RELOAD command issued");
 790         load_config(true);
 791         return admin_ready(admin, "RELOAD");
 792 }
 793
 794 /* Command: SHUTDOWN */
 795 static bool admin_cmd_shutdown(PgSocket *admin, const char *arg)
 796 {
 797         if (arg && *arg)
 798                 return syntax_error(admin);
 799
 800         if (!admin->admin_user)
 801                 return admin_error(admin, "admin access needed");
 802
 803         /*
 804          * note: new pooler expects unix socket file gone when it gets
 805          * event from fd.  currently atexit() cleanup should be called
 806          * before closing open sockets.
 807          */
 808         log_info("SHUTDOWN command issued");
 809         cf_shutdown = 2;
 810         event_loopbreak();
 811
 812         return true;
 813 }
 814
 815 static void full_resume(void)
 816 {
 817         int tmp_mode = cf_pause_mode;
 818         cf_pause_mode = P_NONE;
 819         if (tmp_mode == P_SUSPEND)
 820                 resume_all();
 821
 822         /* avoid surprise later if cf_shutdown stays set */
 823         if (cf_shutdown) {
 824                 log_info("canceling shutdown");
 825                 cf_shutdown = 0;
 826         }
 827 }
 828
 829 /* Command: RESUME */
 830 static bool admin_cmd_resume(PgSocket *admin, const char *arg)
 831 {
 832         if (!admin->admin_user)
 833                 return admin_error(admin, "admin access needed");
 834
 835         if (!arg[0]) {
 836                 log_info("RESUME command issued");
 837                 if (cf_pause_mode != P_NONE)
 838                         full_resume();
 839                 else
 840                         return admin_error(admin, "Pooler is not paused/suspended");
 841         } else {
 842                 PgDatabase *db = find_database(arg);
 843                 log_info("PAUSE '%s' command issued", arg);
 844                 if (db == NULL)
 845                         return admin_error(admin, "no such database: %s", arg);
 846                 if (!db->db_paused)
 847                         return admin_error(admin, "database %s is not paused", arg);
 848                 db->db_paused = 0;
 849         }
 850         return admin_ready(admin, "RESUME");
 851 }
 852
 853 /* Command: SUSPEND */
 854 static bool admin_cmd_suspend(PgSocket *admin, const char *arg)
 855 {
 856         if (arg && *arg)
 857                 return syntax_error(admin);
 858
 859         if (!admin->admin_user)
 860                 return admin_error(admin, "admin access needed");
 861
 862         if (cf_pause_mode)
 863                 return admin_error(admin, "already suspended/paused");
 864
 865         /* suspend needs to be able to flush buffers */
 866         if (count_paused_databases() > 0)
 867                 return admin_error(admin, "cannot suspend with paused databases");
 868
 869         log_info("SUSPEND command issued");
 870         cf_pause_mode = P_SUSPEND;
 871         admin->wait_for_response = 1;
 872         suspend_pooler();
 873
 874         g_suspend_start = get_cached_time();
 875
 876         return true;
 877 }
 878
 879 /* Command: PAUSE */
 880 static bool admin_cmd_pause(PgSocket *admin, const char *arg)
 881 {
 882         if (!admin->admin_user)
 883                 return admin_error(admin, "admin access needed");
 884
 885         if (cf_pause_mode)
 886                 return admin_error(admin, "already suspended/paused");
 887
 888         if (!arg[0]) {
 889                 log_info("PAUSE command issued");
 890                 cf_pause_mode = P_PAUSE;
 891                 admin->wait_for_response = 1;
 892         } else {
 893                 PgDatabase *db;
 894                 log_info("PAUSE '%s' command issued", arg);
 895                 db = find_database(arg);
 896                 if (db == NULL)
 897                         return admin_error(admin, "no such database: %s", arg);
 898                 if (db == admin->pool->db)
 899                         return admin_error(admin, "cannot pause admin db: %s", arg);
 900                 db->db_paused = 1;
 901                 if (count_db_active(db) > 0)
 902                         admin->wait_for_response = 1;
 903                 else
 904                         return admin_ready(admin, "PAUSE");
 905         }
 906
 907         return true;
 908 }
 909
 910 /* extract substring from regex group */
 911 static void copy_arg(const char *src, regmatch_t *glist,
 912                      int gnum, char *dst, unsigned dstmax)
 913 {
 914         regmatch_t *g = &glist[gnum];
 915         unsigned len = g->rm_eo - g->rm_so;
 916         if (len < dstmax)
 917                 memcpy(dst, src + g->rm_so, len);
 918         else
 919                 len = 0;
 920         dst[len] = 0;
 921 }
 922
 923 /* extract quoted substring from regex group */
 924 static void copy_arg_unquote(const char *str, regmatch_t *glist,
 925                              int gnum, char *dst, int dstmax)
 926 {
 927         regmatch_t *g = &glist[gnum];
 928         int len = g->rm_eo - g->rm_so;
 929         const char *src = str + g->rm_so;
 930         const char *end = src + len;
 931
 932         if (len < dstmax) {
 933                 len = 0;
 934                 while (src < end) {
 935                         if (src[0] == '\'' && src[1] == '\'') {
 936                                 *dst++ = '\'';
 937                                 src += 2;
 938                         } else
 939                                 *dst++ = *src++;
 940                 }
 941         }
 942         *dst = 0;
 943 }
 944
 945 static bool admin_show_help(PgSocket *admin, const char *arg)
 946 {
 947         bool res;
 948         SEND_generic(res, admin, 'N',
 949                 "sssss",
 950                 "SNOTICE", "C00000", "MConsole usage",
 951                 "D\n\tSHOW HELP|CONFIG|DATABASES"
 952                 "|POOLS|CLIENTS|SERVERS|VERSION\n"
 953                 "\tSHOW STATS|FDS|SOCKETS|ACTIVE_SOCKETS|LISTS|MEM\n"
 954                 "\tSET key = arg\n"
 955                 "\tRELOAD\n"
 956                 "\tPAUSE [<db>]\n"
 957                 "\tSUSPEND\n"
 958                 "\tRESUME [<db>]\n"
 959                 "\tSHUTDOWN", "");
 960         if (res)
 961                 res = admin_ready(admin, "SHOW");
 962         return res;
 963 }
 964
 965 static bool admin_show_version(PgSocket *admin, const char *arg)
 966 {
 967         bool res;
 968         SEND_generic(res, admin, 'N',
 969                 "ssss", "SNOTICE", "C00000",
 970                 "M" FULLVER, "");
 971         if (res)
 972                 res = admin_ready(admin, "SHOW");
 973         return res;
 974 }
 975
 976 static bool admin_show_stats(PgSocket *admin, const char *arg)
 977 {
 978         return admin_database_stats(admin, &pool_list);
 979 }
 980
 981 static bool admin_show_totals(PgSocket *admin, const char *arg)
 982 {
 983         return show_stat_totals(admin, &pool_list);
 984 }
 985
 986
 987 static struct cmd_lookup show_map [] = {
 988         {"clients", admin_show_clients},
 989         {"config", admin_show_config},
 990         {"databases", admin_show_databases},
 991         {"fds", admin_show_fds},
 992         {"help", admin_show_help},
 993         {"lists", admin_show_lists},
 994         {"pools", admin_show_pools},
 995         {"servers", admin_show_servers},
 996         {"sockets", admin_show_sockets},
 997         {"active_sockets", admin_show_active_sockets},
 998         {"stats", admin_show_stats},
 999         {"users", admin_show_users},
1000         {"version", admin_show_version},
1001         {"totals", admin_show_totals},
1002         {"mem", admin_show_mem},
1003         {NULL, NULL}
1004 };
1005
1006 static bool admin_cmd_show(PgSocket *admin, const char *arg)
1007 {
1008         if (fake_show(admin, arg))
1009                 return true;
1010         return exec_cmd(show_map, admin, arg, NULL);
1011 }
1012
1013 static struct cmd_lookup cmd_list [] = {
1014         {"pause", admin_cmd_pause},
1015         {"reload", admin_cmd_reload},
1016         {"resume", admin_cmd_resume},
1017         {"select", admin_cmd_show},
1018         {"show", admin_cmd_show},
1019         {"shutdown", admin_cmd_shutdown},
1020         {"suspend", admin_cmd_suspend},
1021         {NULL, NULL}
1022 };
1023
1024 /* handle user query */
1025 static bool admin_parse_query(PgSocket *admin, const char *q)
1026 {
1027         regmatch_t grp[MAX_GROUPS];
1028         char cmd[16];
1029         char arg[64];
1030         char val[256];
1031         bool res;
1032
1033         current_query = q;
1034
1035         if (regexec(&rc_cmd, q, MAX_GROUPS, grp, 0) == 0) {
1036                 copy_arg(q, grp, CMD_NAME, cmd, sizeof(cmd));
1037                 copy_arg(q, grp, CMD_ARG, arg, sizeof(arg));
1038                 res = exec_cmd(cmd_list, admin, cmd, arg);
1039         } else if (regexec(&rc_set_str, q, MAX_GROUPS, grp, 0) == 0) {
1040                 copy_arg(q, grp, SET_KEY, arg, sizeof(arg));
1041                 copy_arg_unquote(q, grp, SET_VAL, val, sizeof(val));
1042                 if (!arg[0] || !val[0]) {
1043                         res = admin_error(admin, "bad arguments");
1044                 } else
1045                         res = admin_set(admin, arg, val);
1046         } else if (regexec(&rc_set_word, q, MAX_GROUPS, grp, 0) == 0) {
1047                 copy_arg(q, grp, SET_KEY, arg, sizeof(arg));
1048                 copy_arg(q, grp, SET_VAL, val, sizeof(val));
1049                 if (!arg[0] || !val[0]) {
1050                         res = admin_error(admin, "bad arguments");
1051                 } else
1052                         res = admin_set(admin, arg, val);
1053         } else
1054                 res = syntax_error(admin);
1055
1056         current_query = NULL;
1057
1058         if (!res)
1059                 disconnect_client(admin, true, "failure");
1060         return res;
1061 }
1062
1063 /* handle packets */
1064 bool admin_handle_client(PgSocket *admin, PktHdr *pkt)
1065 {
1066         const char *q;
1067         bool res;
1068
1069         /* dont tolerate partial packets */
1070         if (incomplete_pkt(pkt)) {
1071                 disconnect_client(admin, true, "incomplete pkt");
1072                 return false;
1073         }
1074
1075         switch (pkt->type) {
1076         case 'Q':
1077                 q = mbuf_get_string(&pkt->data);
1078                 if (!q) {
1079                         disconnect_client(admin, true, "incomplete query");
1080                         return false;
1081                 }
1082                 log_debug("got admin query: %s", q);
1083                 res = admin_parse_query(admin, q);
1084                 if (res)
1085                         sbuf_prepare_skip(&admin->sbuf, pkt->len);
1086                 return res;
1087         case 'X':
1088                 disconnect_client(admin, false, "close req");
1089                 break;
1090         default:
1091                 admin_error(admin, "unsupported pkt type: %d", pkt_desc(pkt));
1092                 disconnect_client(admin, true, "bad pkt");
1093                 break;
1094         }
1095         return false;
1096 }
1097
1098 /**
1099  * Client is unauthenticated, look if it wants to connect
1100  * to special "pgbouncer" user.
1101  */
1102 bool admin_pre_login(PgSocket *client)
1103 {
1104         uid_t peer_uid = -1;
1105         gid_t peer_gid = -1;
1106         int res;
1107         const char *username = client->auth_user->name;
1108
1109         client->admin_user = 0;
1110         client->own_user = 0;
1111
1112         /* tag same uid as special */
1113         if (client->remote_addr.is_unix) {
1114                 res = getpeereid(sbuf_socket(&client->sbuf), &peer_uid, &peer_gid);
1115                 if (res >= 0 && peer_uid == getuid()
1116                         && strcmp("pgbouncer", username) == 0)
1117                 {
1118                         client->own_user = 1;
1119                         client->admin_user = 1;
1120                         slog_info(client, "pgbouncer access from unix socket");
1121                         return true;
1122                 }
1123         }
1124
1125         /*
1126          * auth_mode=any does not keep original username around,
1127          * so username based checks do not work.
1128          */
1129         if (cf_auth_type == AUTH_ANY) {
1130                 if (cf_log_connections)
1131                         slog_info(client, "auth_mode=any: allowing anybody in as admin");
1132                 client->admin_user = 1;
1133                 return true;
1134         }
1135
1136         if (strlist_contains(cf_admin_users, username)) {
1137                 client->admin_user = 1;
1138                 return true;
1139         } else if (strlist_contains(cf_stats_users, username)) {
1140                 return true;
1141         }
1142         disconnect_client(client, true, "not allowed");
1143         return false;
1144 }
1145
1146 /* init special database and query parsing */
1147 void admin_setup(void)
1148 {
1149         PgDatabase *db;
1150         PgPool *pool;
1151         PgUser *user;
1152         PktBuf msg;
1153         int res;
1154
1155         /* fake database */
1156         db = add_database("pgbouncer");
1157         if (!db)
1158                 fatal("no memory for admin database");
1159
1160         db->addr.port = cf_listen_port;
1161         db->addr.is_unix = 1;
1162         db->pool_size = 2;
1163         db->admin = 1;
1164         if (!force_user(db, "pgbouncer", ""))
1165                 fatal("no mem on startup - cannot alloc pgbouncer user");
1166
1167         /* fake pool */
1168         pool = get_pool(db, db->forced_user);
1169         if (!pool)
1170                 fatal("cannot create admin pool?");
1171         admin_pool = pool;
1172
1173         /* fake user, with disabled psw */
1174         user = add_user("pgbouncer", "");
1175         if (!user)
1176                 fatal("cannot create admin user?");
1177
1178         /* prepare welcome */
1179         pktbuf_static(&msg, pool->welcome_msg, sizeof(pool->welcome_msg));
1180         pktbuf_write_AuthenticationOk(&msg);
1181         pktbuf_write_ParameterStatus(&msg, "server_version", "8.0/bouncer");
1182         pktbuf_write_ParameterStatus(&msg, "client_encoding", "UNICODE");
1183         pktbuf_write_ParameterStatus(&msg, "server_encoding", "SQL_ASCII");
1184         pktbuf_write_ParameterStatus(&msg, "is_superuser", "on");
1185
1186         pool->welcome_msg_len = pktbuf_written(&msg);
1187         pool->welcome_msg_ready = 1;
1188
1189         pktbuf_static(&msg, db->startup_params, sizeof(db->startup_params));
1190         pktbuf_put_string(&msg, "database");
1191         db->dbname = (char *)db->startup_params + pktbuf_written(&msg);
1192         pktbuf_put_string(&msg, "pgbouncer");
1193         db->startup_params_len = pktbuf_written(&msg);
1194
1195         /* initialize regexes */
1196         res = regcomp(&rc_cmd, cmd_normal_rx, REG_EXTENDED | REG_ICASE);
1197         if (res != 0)
1198                 fatal("cmd regex compilation error");
1199         res = regcomp(&rc_set_word, cmd_set_word_rx, REG_EXTENDED | REG_ICASE);
1200         if (res != 0)
1201                 fatal("set/word regex compilation error");
1202         res = regcomp(&rc_set_str, cmd_set_str_rx, REG_EXTENDED | REG_ICASE);
1203         if (res != 0)
1204                 fatal("set/str regex compilation error");
1205 }
1206
1207 void admin_pause_done(void)
1208 {
1209         List *item, *tmp;
1210         PgSocket *admin;
1211         bool res;
1212
1213         statlist_for_each_safe(item, &admin_pool->active_client_list, tmp) {
1214                 admin = container_of(item, PgSocket, head);
1215                 if (!admin->wait_for_response)
1216                         continue;
1217
1218                 res = false;
1219                 switch (cf_pause_mode) {
1220                 case P_PAUSE:
1221                         res = admin_ready(admin, "PAUSE");
1222                         break;
1223                 case P_SUSPEND:
1224                         res = admin_ready(admin, "SUSPEND");
1225                         break;
1226                 default:
1227                         if (count_paused_databases() > 0)
1228                                 res = admin_ready(admin, "PAUSE");
1229                         else
1230                                 /* fixme */
1231                                 fatal("admin_pause_done: bad state");
1232                 }
1233
1234                 if (!res)
1235                         disconnect_client(admin, false, "dead admin");
1236                 else
1237                         admin->wait_for_response = 0;
1238         }
1239
1240         if (statlist_empty(&admin_pool->active_client_list)
1241             && cf_pause_mode == P_SUSPEND)
1242         {
1243                 log_info("Admin disappeared when suspended, doing RESUME");
1244                 cf_pause_mode = P_NONE;
1245                 resume_all();
1246         }
1247 }
1248
1249 /* admin on console has pressed ^C */
1250 void admin_handle_cancel(PgSocket *admin)
1251 {
1252         bool res;
1253
1254         /* weird, but no reason to fail */
1255         if (!admin->wait_for_response)
1256                 slog_warning(admin, "admin cancel request for non-waiting client?");
1257
1258         if (cf_pause_mode != P_NONE)
1259                 full_resume();
1260
1261         /* notify readiness */
1262         SEND_ReadyForQuery(res, admin);
1263         if (!res)
1264                 disconnect_client(admin, false, "readiness send failed");
1265 }
1266