ctdb/tests/src/db_test_tool.c

   1 /*
   2    CTDB DB test tool
   3
   4    Copyright (C) Martin Schwenke  2019
   5
   6    Parts based on ctdb.c, event_tool.c:
   7
   8    Copyright (C) Amitay Isaacs  2015, 2018
   9
  10    This program is free software; you can redistribute it and/or modify
  11    it under the terms of the GNU General Public License as published by
  12    the Free Software Foundation; either version 3 of the License, or
  13    (at your option) any later version.
  14
  15    This program is distributed in the hope that it will be useful,
  16    but WITHOUT ANY WARRANTY; without even the implied warranty of
  17    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  18    GNU General Public License for more details.
  19
  20    You should have received a copy of the GNU General Public License
  21    along with this program; if not, see <http://www.gnu.org/licenses/>.
  22 */
  23
  24 #include "replace.h"
  25 #include "system/filesys.h"
  26 #include "system/network.h"
  27 #include "system/time.h"
  28
  29 #include <ctype.h>
  30 #include <popt.h>
  31 #include <talloc.h>
  32 #include <tevent.h>
  33
  34 #include "lib/util/debug.h"
  35 #include "lib/util/sys_rw.h"
  36 #include "lib/util/util.h"
  37 #include "lib/util/smb_strtox.h"
  38 #include "lib/tdb_wrap/tdb_wrap.h"
  39
  40 #include "common/cmdline.h"
  41 #include "common/logging.h"
  42 #include "common/path.h"
  43 #include "common/event_script.h"
  44 #include "common/system_socket.h"
  45
  46 #include "protocol/protocol.h"
  47 #include "protocol/protocol_api.h"
  48 #include "protocol/protocol_util.h"
  49
  50 #include "client/client.h"
  51 #include "client/client_sync.h"
  52
  53 struct tdb_context *client_db_tdb(struct ctdb_db_context *db);
  54
  55 #define TIMEOUT()       tevent_timeval_current_ofs(ctx->timelimit, 0)
  56
  57 struct db_test_tool_context {
  58         struct cmdline_context *cmdline;
  59         struct tevent_context *ev;
  60         struct ctdb_client_context *client;
  61         uint32_t destnode;
  62         uint32_t timelimit;
  63 };
  64
  65 /*
  66  * If this is ever consolodated into a larger test tool then these
  67  * forward declarations can be moved to an include file
  68  */
  69 int db_test_tool_init(TALLOC_CTX *mem_ctx,
  70                       const char *prog,
  71                       struct poptOption *options,
  72                       int argc,
  73                       const char **argv,
  74                       bool parse_options,
  75                       struct db_test_tool_context **result);
  76 int db_test_tool_run(struct db_test_tool_context *ctx, int *result);
  77
  78 static int db_test_get_lmaster(TALLOC_CTX *mem_ctx,
  79                                int argc,
  80                                const char **argv,
  81                                void *private_data)
  82 {
  83         struct db_test_tool_context *ctx = talloc_get_type_abort(
  84                 private_data, struct db_test_tool_context);
  85         struct ctdb_vnn_map *vnnmap;
  86         TDB_DATA key;
  87         uint32_t idx, lmaster;
  88         unsigned int hash;
  89         int ret = 0;
  90
  91         if (argc != 1) {
  92                 cmdline_usage(ctx->cmdline, "get-lmaster");
  93                 return 1;
  94         }
  95
  96         ret = ctdb_ctrl_getvnnmap(mem_ctx,
  97                                   ctx->ev,
  98                                   ctx->client,
  99                                   CTDB_CURRENT_NODE,
 100                                   TIMEOUT(),
 101                                   &vnnmap);
 102         if (ret != 0) {
 103                 D_ERR("Control GETVNN_MAP failed, ret=%d\n", ret);
 104                 return ret;
 105         }
 106
 107         key.dsize = strlen(argv[0]);
 108         key.dptr = (uint8_t *)discard_const(argv[0]);
 109
 110         hash = tdb_jenkins_hash(&key);
 111         idx =  hash % vnnmap->size;
 112         lmaster = vnnmap->map[idx];
 113
 114         printf("%"PRId32"\n", lmaster);
 115
 116         return 0;
 117 }
 118
 119 static struct ctdb_dbid *db_find(TALLOC_CTX *mem_ctx,
 120                                  struct db_test_tool_context *ctx,
 121                                  struct ctdb_dbid_map *dbmap,
 122                                  const char *db_name)
 123 {
 124         struct ctdb_dbid *db = NULL;
 125         const char *name;
 126         unsigned int i;
 127         int ret;
 128
 129         for (i=0; i<dbmap->num; i++) {
 130                 ret = ctdb_ctrl_get_dbname(mem_ctx,
 131                                            ctx->ev,
 132                                            ctx->client,
 133                                            ctx->destnode,
 134                                            TIMEOUT(),
 135                                            dbmap->dbs[i].db_id,
 136                                            &name);
 137                 if (ret != 0) {
 138                         return NULL;
 139                 }
 140
 141                 if (strcmp(db_name, name) == 0) {
 142                         talloc_free(discard_const(name));
 143                         db = &dbmap->dbs[i];
 144                         break;
 145                 }
 146         }
 147
 148         return db;
 149 }
 150
 151 static bool db_exists(TALLOC_CTX *mem_ctx,
 152                       struct db_test_tool_context *ctx,
 153                       const char *db_arg,
 154                       uint32_t *db_id,
 155                       const char **db_name,
 156                       uint8_t *db_flags)
 157 {
 158         struct ctdb_dbid_map *dbmap;
 159         struct ctdb_dbid *db = NULL;
 160         uint32_t id = 0;
 161         const char *name = NULL;
 162         unsigned int i;
 163         int ret = 0;
 164
 165         ret = ctdb_ctrl_get_dbmap(mem_ctx,
 166                                   ctx->ev,
 167                                   ctx->client,
 168                                   ctx->destnode,
 169                                   TIMEOUT(),
 170                                   &dbmap);
 171         if (ret != 0) {
 172                 return false;
 173         }
 174
 175         if (strncmp(db_arg, "0x", 2) == 0) {
 176                 id = smb_strtoul(db_arg, NULL, 0, &ret, SMB_STR_STANDARD);
 177                 if (ret != 0) {
 178                         return false;
 179                 }
 180                 for (i=0; i<dbmap->num; i++) {
 181                         if (id == dbmap->dbs[i].db_id) {
 182                                 db = &dbmap->dbs[i];
 183                                 break;
 184                         }
 185                 }
 186         } else {
 187                 name = db_arg;
 188                 db = db_find(mem_ctx, ctx, dbmap, name);
 189         }
 190
 191         if (db == NULL) {
 192                 fprintf(stderr, "No database matching '%s' found\n", db_arg);
 193                 return false;
 194         }
 195
 196         if (name == NULL) {
 197                 ret = ctdb_ctrl_get_dbname(mem_ctx,
 198                                            ctx->ev,
 199                                            ctx->client,
 200                                            ctx->destnode,
 201                                            TIMEOUT(),
 202                                            id,
 203                                            &name);
 204                 if (ret != 0) {
 205                         return false;
 206                 }
 207         }
 208
 209         if (db_id != NULL) {
 210                 *db_id = db->db_id;
 211         }
 212         if (db_name != NULL) {
 213                 *db_name = talloc_strdup(mem_ctx, name);
 214         }
 215         if (db_flags != NULL) {
 216                 *db_flags = db->flags;
 217         }
 218         return true;
 219 }
 220
 221 static int db_test_fetch_local_delete(TALLOC_CTX *mem_ctx,
 222                                       int argc,
 223                                       const char **argv,
 224                                       void *private_data)
 225 {
 226         struct db_test_tool_context *ctx = talloc_get_type_abort(
 227                 private_data, struct db_test_tool_context);
 228         struct ctdb_db_context *db = NULL;
 229         struct ctdb_record_handle *h = NULL;
 230         struct tdb_context *tdb;
 231         struct ctdb_ltdb_header header;
 232         const char *db_name;
 233         TDB_DATA key, data;
 234         uint32_t db_id;
 235         uint8_t db_flags;
 236         size_t len;
 237         uint8_t *buf;
 238         size_t np;
 239         int ret;
 240
 241         if (argc != 2) {
 242                 cmdline_usage(ctx->cmdline, "fetch-local-delete");
 243                 return 1;
 244         }
 245
 246         if (! db_exists(mem_ctx, ctx, argv[0], &db_id, &db_name, &db_flags)) {
 247                 return ENOENT;
 248         }
 249
 250         if (db_flags & (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED)) {
 251                 D_ERR("DB %s is not a volatile database\n", db_name);
 252                 return EINVAL;
 253         }
 254
 255         ret = ctdb_attach(ctx->ev,
 256                           ctx->client,
 257                           TIMEOUT(),
 258                           db_name,
 259                           db_flags,
 260                           &db);
 261         if (ret != 0) {
 262                 D_ERR("Failed to attach to DB %s\n", db_name);
 263                 return ret;
 264         }
 265
 266         key.dsize = strlen(argv[1]);
 267         key.dptr = (uint8_t *)discard_const(argv[1]);
 268
 269         ret = ctdb_fetch_lock(mem_ctx,
 270                               ctx->ev,
 271                               ctx->client,
 272                               db,
 273                               key,
 274                               false,
 275                               &h,
 276                               &header,
 277                               NULL);
 278         if (ret != 0) {
 279                 D_ERR("Failed to fetch record for key %s\n", argv[1]);
 280                 goto done;
 281         }
 282
 283         len = ctdb_ltdb_header_len(&header);
 284         buf = talloc_size(mem_ctx, len);
 285         if (buf == NULL) {
 286                 D_ERR("Memory allocation error\n");
 287                 ret = ENOMEM;
 288                 goto done;
 289         }
 290
 291         ctdb_ltdb_header_push(&header, buf, &np);
 292
 293         data.dsize = np;
 294         data.dptr = buf;
 295
 296         tdb = client_db_tdb(db);
 297
 298         ret = tdb_store(tdb, key, data, TDB_REPLACE);
 299         TALLOC_FREE(buf);
 300         if (ret != 0) {
 301                 D_ERR("fetch_lock delete: %s tdb_store failed, %s\n",
 302                       db_name,
 303                       tdb_errorstr(tdb));
 304         }
 305
 306 done:
 307         TALLOC_FREE(h);
 308
 309         return ret;
 310 }
 311
 312 #define ISASCII(x) (isprint(x) && ! strchr("\"\\", (x)))
 313
 314 static void dump(const char *name, uint8_t *dptr, size_t dsize)
 315 {
 316         size_t i;
 317
 318         fprintf(stdout, "%s(%zu) = \"", name, dsize);
 319         for (i = 0; i < dsize; i++) {
 320                 if (ISASCII(dptr[i])) {
 321                         fprintf(stdout, "%c", dptr[i]);
 322                 } else {
 323                         fprintf(stdout, "\\%02X", dptr[i]);
 324                 }
 325         }
 326         fprintf(stdout, "\"\n");
 327 }
 328
 329 static void dump_ltdb_header(struct ctdb_ltdb_header *header)
 330 {
 331         fprintf(stdout, "dmaster: %u\n", header->dmaster);
 332         fprintf(stdout, "rsn: %" PRIu64 "\n", header->rsn);
 333         fprintf(stdout, "flags: 0x%08x", header->flags);
 334         if (header->flags & CTDB_REC_FLAG_MIGRATED_WITH_DATA) {
 335                 fprintf(stdout, " MIGRATED_WITH_DATA");
 336         }
 337         if (header->flags & CTDB_REC_FLAG_VACUUM_MIGRATED) {
 338                 fprintf(stdout, " VACUUM_MIGRATED");
 339         }
 340         if (header->flags & CTDB_REC_FLAG_AUTOMATIC) {
 341                 fprintf(stdout, " AUTOMATIC");
 342         }
 343         if (header->flags & CTDB_REC_RO_HAVE_DELEGATIONS) {
 344                 fprintf(stdout, " RO_HAVE_DELEGATIONS");
 345         }
 346         if (header->flags & CTDB_REC_RO_HAVE_READONLY) {
 347                 fprintf(stdout, " RO_HAVE_READONLY");
 348         }
 349         if (header->flags & CTDB_REC_RO_REVOKING_READONLY) {
 350                 fprintf(stdout, " RO_REVOKING_READONLY");
 351         }
 352         if (header->flags & CTDB_REC_RO_REVOKE_COMPLETE) {
 353                 fprintf(stdout, " RO_REVOKE_COMPLETE");
 354         }
 355         fprintf(stdout, "\n");
 356
 357 }
 358
 359 static int db_test_local_lock(TALLOC_CTX *mem_ctx,
 360                               int argc,
 361                               const char **argv,
 362                               void *private_data)
 363 {
 364         struct db_test_tool_context *ctx = talloc_get_type_abort(
 365                 private_data, struct db_test_tool_context);
 366         struct ctdb_db_context *db;
 367         const char *db_name;
 368         int pipefd[2];
 369         TDB_DATA key;
 370         uint32_t db_id;
 371         uint8_t db_flags;
 372         pid_t pid;
 373         int ret;
 374
 375         if (argc != 2) {
 376                 cmdline_usage(ctx->cmdline, "local-lock");
 377                 return 1;
 378         }
 379
 380
 381         if (! db_exists(mem_ctx, ctx, argv[0], &db_id, &db_name, &db_flags)) {
 382                 D_ERR("DB %s not attached\n", db_name);
 383                 return 1;
 384         }
 385
 386         if (db_flags & (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED)) {
 387                 D_ERR("DB %s is not a volatile database\n", db_name);
 388                 return 1;
 389         }
 390
 391         ret = ctdb_attach(ctx->ev,
 392                           ctx->client,
 393                           TIMEOUT(),
 394                           db_name,
 395                           db_flags,
 396                           &db);
 397         if (ret != 0) {
 398                 D_ERR("Failed to attach to DB %s\n", db_name);
 399                 return 1;
 400         }
 401
 402         ret = pipe(pipefd);
 403         if (ret != 0) {
 404                 DBG_ERR("Failed to create pipe\n");
 405                 return 1;
 406         }
 407
 408         pid = fork();
 409         if (pid < 0) {
 410                 DBG_ERR("Failed to fork()\n");
 411                 return 1;
 412         }
 413
 414         if (pid != 0) {
 415                 ssize_t nread;
 416                 int status;
 417
 418                 close(pipefd[1]);
 419
 420                 nread = sys_read(pipefd[0], &status, sizeof(status));
 421                 if (nread < 0 || (size_t)nread != sizeof(status)) {
 422                         status = EINVAL;
 423                 }
 424
 425                 if (status == 0) {
 426                         printf("OK %d\n", pid);
 427                 } else {
 428                         printf("FAIL %d\n", status);
 429                 }
 430                 fflush(stdout);
 431
 432                 return status;
 433         }
 434
 435         close(pipefd[0]);
 436
 437         key.dsize = strlen(argv[1]);
 438         key.dptr = (uint8_t *)discard_const(argv[1]);
 439
 440         ret = tdb_chainlock(client_db_tdb(db), key);
 441         if (ret != 0) {
 442                 D_ERR("Failed to lock chain for key %s\n", argv[1]);
 443                 goto fail;
 444         }
 445
 446         sys_write(pipefd[1], &ret, sizeof(ret));
 447
 448         fclose(stdin);
 449         fclose(stdout);
 450         fclose(stderr);
 451
 452         /* Hold the lock- the caller should SIGTERM to release the lock */
 453         sleep(120);
 454         exit(1);
 455
 456 fail:
 457         sys_write(pipefd[1], &ret, sizeof(ret));
 458         return ret;
 459 }
 460
 461 static int db_test_local_read(TALLOC_CTX *mem_ctx,
 462                               int argc,
 463                               const char **argv,
 464                               void *private_data)
 465 {
 466         struct db_test_tool_context *ctx = talloc_get_type_abort(
 467                 private_data, struct db_test_tool_context);
 468         struct ctdb_db_context *db;
 469         struct ctdb_ltdb_header header;
 470         const char *db_name;
 471         TDB_DATA key, data;
 472         uint32_t db_id;
 473         uint8_t db_flags;
 474         size_t np;
 475         int ret;
 476
 477         if (argc != 2) {
 478                 cmdline_usage(ctx->cmdline, "local-read");
 479                 return 1;
 480         }
 481
 482         if (! db_exists(mem_ctx, ctx, argv[0], &db_id, &db_name, &db_flags)) {
 483                 return ENOENT;
 484         }
 485
 486         if (db_flags & (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED)) {
 487                 D_ERR("DB %s is not a volatile database\n", db_name);
 488                 return EINVAL;
 489         }
 490
 491         ret = ctdb_attach(ctx->ev,
 492                           ctx->client,
 493                           TIMEOUT(),
 494                           db_name,
 495                           db_flags,
 496                           &db);
 497         if (ret != 0) {
 498                 D_ERR("Failed to attach to DB %s\n", db_name);
 499                 return ret;
 500         }
 501
 502         key.dsize = strlen(argv[1]);
 503         key.dptr = (uint8_t *)discard_const(argv[1]);
 504
 505         data = tdb_fetch(client_db_tdb(db), key);
 506
 507         if (data.dptr == NULL) {
 508                 D_ERR("No record for key %s\n", argv[1]);
 509                 return 1;
 510         }
 511
 512         if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
 513                 D_ERR("Invalid record for key %s\n", argv[1]);
 514                 free(data.dptr);
 515                 return 1;
 516         }
 517
 518         ret = ctdb_ltdb_header_pull(data.dptr, data.dsize, &header, &np);
 519         if (ret != 0) {
 520                 D_ERR("Failed to parse header from data\n");
 521                 free(data.dptr);
 522                 return 1;
 523         }
 524
 525         dump_ltdb_header(&header);
 526         dump("data", data.dptr + np, data.dsize - np);
 527
 528         free(data.dptr);
 529
 530         return 0;
 531 }
 532
 533 static int db_test_vacuum(TALLOC_CTX *mem_ctx,
 534                           int argc,
 535                           const char **argv,
 536                           void *private_data)
 537 {
 538         struct db_test_tool_context *ctx = talloc_get_type_abort(
 539                 private_data, struct db_test_tool_context);
 540         struct ctdb_db_vacuum db_vacuum;
 541         struct ctdb_req_control request;
 542         struct ctdb_reply_control *reply;
 543         const char *db_arg;
 544         uint32_t db_id;
 545         const char *db_name;
 546         uint8_t db_flags;
 547         int ret = 0;
 548
 549         if (argc != 1 && argc != 2) {
 550                 cmdline_usage(ctx->cmdline, "vacuum");
 551                 return 1;
 552         }
 553
 554         db_arg = argv[0];
 555
 556         db_vacuum.full_vacuum_run = false;
 557         if (argc == 2) {
 558                 if (strcmp(argv[1], "full") == 0) {
 559                         db_vacuum.full_vacuum_run = true;
 560                 } else {
 561                         cmdline_usage(ctx->cmdline, "vacuum");
 562                         return 1;
 563                 }
 564         }
 565
 566         if (! db_exists(mem_ctx, ctx, db_arg, &db_id, &db_name, &db_flags)) {
 567                 return ENOENT;
 568         }
 569
 570         if (db_flags & (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED)) {
 571                 D_ERR("DB %s is not a volatile database\n", db_name);
 572                 return EINVAL;
 573         }
 574
 575         db_vacuum.db_id = db_id;
 576
 577         ctdb_req_control_db_vacuum(&request, &db_vacuum);
 578
 579         ret = ctdb_client_control(mem_ctx,
 580                                   ctx->ev,
 581                                   ctx->client,
 582                                   ctx->destnode,
 583                                   TIMEOUT(),
 584                                   &request,
 585                                   &reply);
 586         if (ret != 0) {
 587                 D_ERR("Control DB_VACUUM failed to node %u, ret=%d\n",
 588                       ctx->destnode,
 589                       ret);
 590                 return ret;
 591         }
 592
 593
 594         ret = ctdb_reply_control_db_vacuum(reply);
 595         if (ret != 0) {
 596                 D_ERR("Control DB_VACUUM failed, ret=%d\n", ret);
 597                 return ret;
 598         }
 599
 600         return 0;
 601 }
 602
 603 struct cmdline_command db_test_commands[] = {
 604         {
 605                 .name     = "get-lmaster",
 606                 .fn       = db_test_get_lmaster,
 607                 .msg_help = "Print lmaster for key",
 608                 .msg_args = "<key>"
 609         },
 610         {
 611                 .name     = "fetch-local-delete",
 612                 .fn       = db_test_fetch_local_delete,
 613                 .msg_help = "Fetch record and delete from local database",
 614                 .msg_args = "<dbname|dbid> <key>"
 615         },
 616         {
 617                 .name     = "local-lock",
 618                 .fn       = db_test_local_lock,
 619                 .msg_help = "Lock a record in a local database",
 620                 .msg_args = "<dbname|dbid> <key>"
 621         },
 622         {
 623                 .name     = "local-read",
 624                 .fn       = db_test_local_read,
 625                 .msg_help = "Read a record from local database",
 626                 .msg_args = "<dbname|dbid> <key>"
 627         },
 628         {
 629                 .name     = "vacuum",
 630                 .fn       = db_test_vacuum,
 631                 .msg_help = "Vacuum a database",
 632                 .msg_args = "<dbname|dbid> [full]"
 633         },
 634         CMDLINE_TABLEEND
 635 };
 636
 637 int db_test_tool_init(TALLOC_CTX *mem_ctx,
 638                       const char *prog,
 639                       struct poptOption *options,
 640                       int argc,
 641                       const char **argv,
 642                       bool parse_options,
 643                       struct db_test_tool_context **result)
 644 {
 645         struct db_test_tool_context *ctx;
 646         int ret;
 647
 648         ctx = talloc_zero(mem_ctx, struct db_test_tool_context);
 649         if (ctx == NULL) {
 650                 D_ERR("Memory allocation error\n");
 651                 return ENOMEM;
 652         }
 653
 654         ret = cmdline_init(mem_ctx,
 655                            prog,
 656                            options,
 657                            NULL,
 658                            db_test_commands,
 659                            &ctx->cmdline);
 660         if (ret != 0) {
 661                 D_ERR("Failed to initialize cmdline, ret=%d\n", ret);
 662                 talloc_free(ctx);
 663                 return ret;
 664         }
 665
 666         ret = cmdline_parse(ctx->cmdline, argc, argv, parse_options);
 667         if (ret != 0) {
 668                 cmdline_usage(ctx->cmdline, NULL);
 669                 talloc_free(ctx);
 670                 return ret;
 671         }
 672
 673         *result = ctx;
 674         return 0;
 675 }
 676
 677 int db_test_tool_run(struct db_test_tool_context *ctx, int *result)
 678 {
 679         char *ctdb_socket;
 680         int ret;
 681
 682         ctx->ev = tevent_context_init(ctx);
 683         if (ctx->ev == NULL) {
 684                 D_ERR("Failed to initialize tevent\n");
 685                 return ENOMEM;
 686         }
 687
 688         ctdb_socket = path_socket(ctx, "ctdbd");
 689         if (ctdb_socket == NULL) {
 690                 fprintf(stderr, "Memory allocation error\n");
 691                 return ENOMEM;
 692         }
 693
 694         ret = ctdb_client_init(ctx, ctx->ev, ctdb_socket, &ctx->client);
 695         if (ret != 0) {
 696                 D_ERR("Failed to connect to CTDB daemon (%s)\n", ctdb_socket);
 697                 return ret;
 698         }
 699
 700         ret = cmdline_run(ctx->cmdline, ctx, result);
 701         return ret;
 702 }
 703
 704 #ifdef CTDB_DB_TEST_TOOL
 705
 706 static struct {
 707         const char *debug;
 708         int destnode;
 709         int timelimit;
 710 } db_test_data = {
 711         .debug = "ERROR",
 712         .destnode = CTDB_CURRENT_NODE,
 713         .timelimit = 60,
 714 };
 715
 716 struct poptOption db_test_options[] = {
 717         {
 718                 .longName   = "debug",
 719                 .shortName  = 'd',
 720                 .argInfo    = POPT_ARG_STRING,
 721                 .arg        = &db_test_data.debug,
 722                 .val        = 0,
 723                 .descrip    = "debug level",
 724                 .argDescrip = "ERROR|WARNING|NOTICE|INFO|DEBUG"
 725         },
 726         {
 727                 .longName   = "node",
 728                 .shortName  = 'n',
 729                 .argInfo    = POPT_ARG_INT,
 730                 .arg        = &db_test_data.destnode,
 731                 .val        = 0,
 732                 .descrip    = "node number",
 733                 .argDescrip = "NUM"
 734         },
 735         {
 736                 .longName   = "timelimit",
 737                 .shortName  = 't',
 738                 .argInfo    = POPT_ARG_INT,
 739                 .arg        = &db_test_data.timelimit,
 740                 .val        = 0,
 741                 .descrip    = "control time limit",
 742                 .argDescrip = "SECONDS"
 743         },
 744         POPT_TABLEEND
 745 };
 746
 747 int main(int argc, const char **argv)
 748 {
 749         TALLOC_CTX *mem_ctx;
 750         struct db_test_tool_context *ctx;
 751         int ret, result = 0;
 752         int level;
 753         bool ok;
 754
 755         mem_ctx = talloc_new(NULL);
 756         if (mem_ctx == NULL) {
 757                 fprintf(stderr, "Memory allocation error\n");
 758                 exit(1);
 759         }
 760
 761         ret = db_test_tool_init(mem_ctx,
 762                                 "ctdb-db-test",
 763                                 db_test_options,
 764                                 argc,
 765                                 argv,
 766                                 true,
 767                                 &ctx);
 768         if (ret != 0) {
 769                 talloc_free(mem_ctx);
 770                 exit(1);
 771         }
 772
 773         setup_logging("ctdb-db-test", DEBUG_STDERR);
 774         ok = debug_level_parse(db_test_data.debug, &level);
 775         if (!ok) {
 776                 level = DEBUG_ERR;
 777         }
 778         debuglevel_set(level);
 779
 780         ctx->destnode = db_test_data.destnode;
 781         ctx->timelimit = db_test_data.timelimit;
 782
 783         ret = db_test_tool_run(ctx, &result);
 784         if (ret != 0) {
 785                 result = ret;
 786         }
 787
 788         talloc_free(mem_ctx);
 789         exit(result);
 790 }
 791
 792 #endif /* CTDB_DB_TEST_TOOL */