Merge remote branch 'martins/nodestatus'

(This used to be ctdb commit adcb076612c56f92cfb6f3ba90721981a067a494)

8 files changed, 680 insertions, 146 deletions

diff --git a/ctdb/Makefile.in b/ctdb/Makefile.in
index 3b9648dfe5..2dd2bb7fb4 100755
--- a/ctdb/Makefile.in
+++ b/ctdb/Makefile.in
@@ -84,6 +84,7 @@ TEST_BINS=tests/bin/ctdb_bench tests/bin/ctdb_fetch tests/bin/ctdb_fetch_one \
 	tests/bin/ctdb_randrec tests/bin/ctdb_persistent \
 	tests/bin/ctdb_traverse tests/bin/rb_test tests/bin/ctdb_transaction \
 	tests/bin/ctdb_takeover_tests tests/bin/ctdb_update_record \
+	tests/bin/ctdb_parse_nodestring \
 	@INFINIBAND_BINS@
 
 BINS = bin/ctdb @CTDB_SCSI_IO@ bin/smnotify bin/ping_pong bin/ltdbtool @CTDB_PMDA@
@@ -234,19 +235,25 @@ tests/bin/ctdb_transaction: $(CTDB_CLIENT_OBJ) tests/src/ctdb_transaction.o
 	@$(CC) $(CFLAGS) -o $@ tests/src/ctdb_transaction.o $(CTDB_CLIENT_OBJ) $(LIB_FLAGS)
 
 CTDB_SERVER_MOST_OBJ = $(CTDB_SERVER_OBJ:server/ctdbd.o=)
-CTDB_TEST_C = $(CTDB_SERVER_MOST_OBJ:.o=.c) tests/src/ctdbd_test.c
+CTDBD_TEST_C = $(CTDB_SERVER_MOST_OBJ:.o=.c) tests/src/ctdbd_test.c
+
+CTDB_TEST_C = $(CTDB_CLIENT_OBJ:.o=.c) tools/ctdb.c tools/ctdb_vacuum.c  tests/src/ctdb_test.c
+
 CTDB_TEST_OBJ =  @TALLOC_OBJ@ @TDB_OBJ@ \
 	@LIBREPLACEOBJ@ $(EXTRA_OBJ) @TEVENT_OBJ@ $(SOCKET_WRAPPER_OBJ)
 
-CTDB_TEST_DEPENDS = $(CTDB_TEST_OBJ) $(CTDB_SERVER_MOST_C) \
-	tests/src/ctdbd_test.c
-
-tests/src/ctdb_takeover_tests.o: tests/src/ctdb_takeover_tests.c $(CTDB_TEST_C)
+tests/src/ctdb_takeover_tests.o: tests/src/ctdb_takeover_tests.c $(CTDBD_TEST_C)
 
 tests/bin/ctdb_takeover_tests: $(CTDB_TEST_OBJ) tests/src/ctdb_takeover_tests.o
 	@echo Linking $@
 	@$(CC) $(CFLAGS) -o $@ tests/src/ctdb_takeover_tests.o $(CTDB_TEST_OBJ) $(LIB_FLAGS)
 
+tests/src/ctdb_parse_nodestring.o: tests/src/ctdb_parse_nodestring.c tests/src/libctdb_test.c $(CTDB_TEST_C)
+
+tests/bin/ctdb_parse_nodestring: $(CTDB_TEST_OBJ) tests/src/ctdb_parse_nodestring.o
+	@echo Linking $@
+	@$(CC) $(CFLAGS) -o $@ tests/src/ctdb_parse_nodestring.o $(CTDB_TEST_OBJ) @POPT_OBJ@ $(LIB_FLAGS)
+
 tests/bin/ibwrapper_test: $(CTDB_CLIENT_OBJ) ib/ibwrapper_test.o
 	@echo Linking $@
 	@$(CC) $(CFLAGS) -o $@ ib/ibwrapper_test.o $(CTDB_CLIENT_OBJ) $(LIB_FLAGS)
diff --git a/ctdb/include/ctdb_protocol.h b/ctdb/include/ctdb_protocol.h
index d96f565eb8..70197d94bd 100644
--- a/ctdb/include/ctdb_protocol.h
+++ b/ctdb/include/ctdb_protocol.h
@@ -182,6 +182,8 @@ struct ctdb_call_info {
 #define CTDB_BROADCAST_VNNMAP 0xF0000003
 /* send a broadcast to all connected nodes */
 #define CTDB_BROADCAST_CONNECTED 0xF0000004
+/* send a broadcast to selected connected nodes */
+#define CTDB_MULTICAST 0xF0000005
 
 /* the key used for transaction locking on persistent databases */
 #define CTDB_TRANSACTION_LOCK_KEY "__transaction_lock__"
diff --git a/ctdb/tests/src/ctdb_parse_nodestring.c b/ctdb/tests/src/ctdb_parse_nodestring.c
new file mode 100644
index 0000000000..e467c36870
--- /dev/null
+++ b/ctdb/tests/src/ctdb_parse_nodestring.c
@@ -0,0 +1,126 @@
+/* 
+   Tests for tools/ctdb.c:parse_nodestring()
+
+   Copyright (C) Martin Schwenke 2011
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 3 of the License, or
+   (at your option) any later version.
+   
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+   
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "ctdb_test.c"
+#include "libctdb_test.c"
+
+static void test_read_nodemap(void)
+{
+	struct ctdb_connection *ctdb = ctdb_connect("foo", NULL, NULL);
+
+	libctdb_test_read_nodemap(ctdb);
+	libctdb_test_print_nodemap(ctdb);
+
+	ctdb_disconnect(ctdb);
+}
+
+static const char * decode_pnn_mode(uint32_t pnn_mode)
+{
+	int i;
+	static const struct {
+		uint32_t mode;
+		const char *name;
+	} pnn_modes[] = {
+		{ CTDB_CURRENT_NODE,        "CURRENT_NODE" },
+		{ CTDB_BROADCAST_ALL,       "BROADCAST_ALL" },
+		{ CTDB_BROADCAST_VNNMAP,    "BROADCAST_VNNMAP" },
+		{ CTDB_BROADCAST_CONNECTED, "BROADCAST_CONNECTED" },
+		{ CTDB_MULTICAST,           "MULTICAST" },
+	};
+
+	for (i = 0; i < ARRAY_SIZE(pnn_modes); i++) {
+		if (pnn_mode == pnn_modes[i].mode) {
+			return pnn_modes[i].name;
+		}
+	}
+
+	return "PNN";
+}
+
+static void print_nodes(uint32_t *nodes, uint32_t pnn_mode)
+{
+	int i;
+
+	printf("NODES:");
+	for (i = 0; i < talloc_array_length(nodes); i++) {
+		printf(" %lu", (unsigned long) nodes[i]);
+	}
+	printf("\n");
+
+	printf("PNN MODE: %s (%lu)\n",
+	       decode_pnn_mode(pnn_mode), (unsigned long) pnn_mode);
+}
+
+static void test_parse_nodestring(const char *nodestring_s,
+				  const char *dd_ok_s)
+{
+	const char *nodestring;
+	bool dd_ok;
+	struct ctdb_connection *ctdb;
+	uint32_t *nodes;
+	uint32_t pnn_mode;
+
+	nodestring = strcmp("", nodestring_s) == 0 ? NULL : nodestring_s;
+
+	if (strcasecmp(dd_ok_s, "yes") == 0 ||
+	    strcmp(dd_ok_s, "true") == 0) {
+		dd_ok = true;
+	} else {
+		dd_ok = false;
+	}
+
+	ctdb  = ctdb_connect("foo", NULL, NULL);
+	ctdb_connection = ctdb;
+
+	libctdb_test_read_nodemap(ctdb);
+
+	if (parse_nodestring(NULL, nodestring, dd_ok, &nodes, &pnn_mode)) {
+		print_nodes(nodes, pnn_mode);
+	}
+
+	ctdb_disconnect(ctdb);
+}
+
+static void usage(void)
+{
+	fprintf(stderr, "usage: ctdb_parse_nodelist <op>\n");
+	exit(1);
+}
+
+int main(int argc, const char *argv[])
+{
+	LogLevel = DEBUG_DEBUG;
+	if (getenv("CTDB_TEST_LOGLEVEL")) {
+		LogLevel = atoi(getenv("CTDB_TEST_LOGLEVEL"));
+	}
+
+	if (argc < 2) {
+		usage();
+	}
+
+	if (argc == 2 && strcmp(argv[1], "read_nodemap") == 0) {
+		test_read_nodemap();
+	} else if (argc == 4 && strcmp(argv[1], "parse_nodestring") == 0) {
+		test_parse_nodestring(argv[2], argv[3]);
+	} else {
+		usage();
+	}
+
+	return 0;
+}
diff --git a/ctdb/tests/src/ctdb_takeover_tests.c b/ctdb/tests/src/ctdb_takeover_tests.c
index 19cb5a6f39..96cb9e5532 100644
--- a/ctdb/tests/src/ctdb_takeover_tests.c
+++ b/ctdb/tests/src/ctdb_takeover_tests.c
@@ -97,8 +97,7 @@ void ctdb_test_read_ctdb_public_ip_list(void)
 	talloc_free(tmp_ctx);
 }
 
-/* Format of each line is "IP pnn" - the separator has to be at least
- * 1 space (not a tab or whatever - a space!).
+/* Format of each line is "IP CURRENT_PNN ALLOWED_PNN,...".
  */
 static bool
 read_ctdb_public_ip_info(TALLOC_CTX *ctx,
diff --git a/ctdb/tests/src/ctdb_test.c b/ctdb/tests/src/ctdb_test.c
new file mode 100644
index 0000000000..533f3f48a1
--- /dev/null
+++ b/ctdb/tests/src/ctdb_test.c
@@ -0,0 +1,60 @@
+/*
+   ctdb test include file
+
+   Copyright (C) Martin Schwenke  2011
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 3 of the License, or
+   (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, see <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef _CTDBD_TEST_C
+#define _CTDBD_TEST_C
+
+#define main(argc, argv) main_foobar(argc, argv)
+#define usage usage_foobar
+#include "tools/ctdb.c"
+#undef main
+#undef usage
+
+#undef TIMELIMIT
+#include "tools/ctdb_vacuum.c"
+
+/* UTIL_OBJ */
+#include "lib/util/idtree.c"
+#include "lib/util/db_wrap.c"
+#include "lib/util/strlist.c"
+#include "lib/util/util.c"
+#include "lib/util/util_time.c"
+#include "lib/util/util_file.c"
+#include "lib/util/fault.c"
+#include "lib/util/substitute.c"
+#include "lib/util/signal.c"
+
+/* CTDB_COMMON_OBJ */
+#include "common/ctdb_io.c"
+#include "common/ctdb_util.c"
+#include "common/ctdb_ltdb.c"
+#include "common/ctdb_message.c"
+#include "common/cmdline.c"
+#include "lib/util/debug.c"
+#include "common/rb_tree.c"
+#ifdef _LINUX_ERRNO_H
+#include "common/system_linux.c"
+#endif
+#include "common/system_common.c"
+#include "common/ctdb_logging.c"
+
+/* CTDB_CLIENT_OBJ */
+#include "client/ctdb_client.c"
+
+#endif /* _CTDBD_TEST_C */
diff --git a/ctdb/tests/src/libctdb_test.c b/ctdb/tests/src/libctdb_test.c
new file mode 100644
index 0000000000..9e9c31eff5
--- /dev/null
+++ b/ctdb/tests/src/libctdb_test.c
@@ -0,0 +1,231 @@
+/*
+   Test stubs and support functions for some libctdb functions
+
+   Copyright (C) Martin Schwenke  2011
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 3 of the License, or
+   (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include <syslog.h>
+#include "ctdb.h"
+
+/* Can't use the real definition, since including libctdb_private.h
+ * causes macro conflicts */
+struct ctdb_connection {
+	struct ctdb_node_map *nodemap;
+	uint32_t current_node;
+	uint32_t recmaster;
+	struct ctdb_ifaces_list *ifaces;
+};
+
+
+/* Read a nodemap from stdin.  Each line looks like:
+ *  <PNN> <FLAGS> [RECMASTER] [CURRENT]
+ * EOF or a blank line terminates input.
+ */
+void libctdb_test_read_nodemap(struct ctdb_connection *ctdb)
+{
+	char line[1024];
+
+	ctdb->nodemap = (struct ctdb_node_map *) malloc(sizeof(uint32_t));
+	if (ctdb->nodemap == NULL) {
+		DEBUG(DEBUG_ERR, ("OOM allocating nodemap\n"));
+		exit (1);
+	}
+	ctdb->nodemap->num = 0;
+
+	while ((fgets(line, sizeof(line), stdin) != NULL) &&
+	       (line[0] != '\n')) {
+		uint32_t pnn, flags;
+		char *tok, *t;
+
+		/* Get rid of pesky newline */
+		if ((t = strchr(line, '\n')) != NULL) {
+			*t = '\0';
+		}
+
+		/* Get PNN */
+		tok = strtok(line, " \t");
+		if (tok == NULL) {
+			DEBUG(DEBUG_ERR, (__location__ " WARNING, bad line ignoed :%s\n", line));
+			continue;
+		}
+		pnn = (uint32_t)strtoul(tok, NULL, 0);
+
+		/* Get flags */
+		tok = strtok(NULL, " \t");
+		if (tok == NULL) {
+			DEBUG(DEBUG_ERR, (__location__ " WARNING, bad line ignored :%s\n", line));
+			continue;
+		}
+		flags = (uint32_t)strtoul(tok, NULL, 0);
+
+		tok = strtok(NULL, " \t");
+		while (tok != NULL) {
+			if (strcmp(tok, "CURRENT") == 0) {
+				ctdb->current_node = pnn;
+			} else if (strcmp(tok, "RECMASTER") == 0) {
+				ctdb->recmaster = pnn;
+			}
+			tok = strtok(NULL, " \t");
+		}
+
+		ctdb->nodemap = (struct ctdb_node_map *) realloc(ctdb->nodemap, sizeof(uint32_t) + (ctdb->nodemap->num + 1) * sizeof(struct ctdb_node_and_flags));
+		if (ctdb->nodemap == NULL) {
+			DEBUG(DEBUG_ERR, ("OOM allocating nodemap\n"));
+			exit (1);
+		}
+
+		ctdb->nodemap->nodes[ctdb->nodemap->num].pnn = pnn;
+		ctdb->nodemap->nodes[ctdb->nodemap->num].flags = flags;
+		ctdb->nodemap->num++;
+	}
+}
+
+void libctdb_test_print_nodemap(struct ctdb_connection *ctdb)
+{
+	int i;
+
+	for (i = 0; i < ctdb->nodemap->num; i++) {
+		printf("%ld\t0x%lx%s%s\n",
+		       (unsigned long) ctdb->nodemap->nodes[i].pnn,
+		       (unsigned long) ctdb->nodemap->nodes[i].flags,
+		       ctdb->nodemap->nodes[i].pnn == ctdb->current_node ? "\tCURRENT" : "",
+		       ctdb->nodemap->nodes[i].pnn == ctdb->recmaster ? "\tRECMASTER" : "");
+	}
+}
+
+/* Stubs... */
+
+bool ctdb_getnodemap(struct ctdb_connection *ctdb,
+		     uint32_t destnode, struct ctdb_node_map **nodemap)
+{
+	*nodemap = ctdb->nodemap;
+	return true;
+}
+
+void ctdb_free_nodemap(struct ctdb_node_map *nodemap)
+{
+	return;
+}
+
+bool ctdb_getifaces(struct ctdb_connection *ctdb,
+		    uint32_t destnode, struct ctdb_ifaces_list **ifaces)
+{
+	*ifaces = ctdb->ifaces;
+	return false; /* Not implemented */
+}
+
+void ctdb_free_ifaces(struct ctdb_ifaces_list *ifaces)
+{
+	return;
+}
+
+bool ctdb_getpnn(struct ctdb_connection *ctdb,
+		 uint32_t destnode,
+		 uint32_t *pnn)
+{
+	if (destnode == CTDB_CURRENT_NODE) {
+		*pnn = ctdb->current_node;
+	} else {
+		*pnn = destnode;
+	}
+	return true;
+}
+
+bool ctdb_getrecmode(struct ctdb_connection *ctdb,
+		     uint32_t destnode,
+		     uint32_t *recmode)
+{
+	*recmode = 0;
+	return true;
+}
+
+bool ctdb_getrecmaster(struct ctdb_connection *ctdb,
+		       uint32_t destnode,
+		       uint32_t *recmaster)
+{
+	*recmaster = ctdb->recmaster;
+	return true;
+}
+
+bool
+ctdb_getdbseqnum(struct ctdb_connection *ctdb,
+		 uint32_t destnode,
+		 uint32_t dbid,
+		 uint64_t *seqnum)
+{
+	*seqnum = 0;
+	return false; /* Not implemented */
+}
+
+bool
+ctdb_check_message_handlers(struct ctdb_connection *ctdb,
+			   uint32_t destnode,
+			   uint32_t num,
+			   uint64_t *mhs,
+			   uint8_t *result)
+{
+	*result = 0;
+	return false; /* Not implemented */
+}
+
+/* Not a stub, a copy */
+void ctdb_log_file(FILE *outf, int priority, const char *format, va_list ap)
+{
+	fprintf(outf, "%s:",
+		priority == LOG_EMERG ? "EMERG" :
+		priority == LOG_ALERT ? "ALERT" :
+		priority == LOG_CRIT ? "CRIT" :
+		priority == LOG_ERR ? "ERR" :
+		priority == LOG_WARNING ? "WARNING" :
+		priority == LOG_NOTICE ? "NOTICE" :
+		priority == LOG_INFO ? "INFO" :
+		priority == LOG_DEBUG ? "DEBUG" :
+		"Unknown Error Level");
+
+	vfprintf(outf, format, ap);
+	if (priority == LOG_ERR) {
+		fprintf(outf, " (%s)", strerror(errno));
+	}
+	fprintf(outf, "\n");
+}
+
+/* Remove type-safety macro. */
+#undef ctdb_connect
+struct ctdb_connection *ctdb_connect(const char *addr,
+				     ctdb_log_fn_t log_func, void *log_priv)
+{
+	struct ctdb_connection *ctdb;
+
+	ctdb = malloc(sizeof(struct ctdb_connection));
+	if (ctdb == NULL) {
+		DEBUG(DEBUG_ERR, ("OOM allocating ctdb_connection\n"));
+		exit (1);
+	}
+
+	ctdb->nodemap = NULL;
+	ctdb->current_node = 0;
+	ctdb->recmaster = 0;
+
+	return ctdb;
+}
+
+void ctdb_disconnect(struct ctdb_connection *ctdb)
+{
+	if (ctdb->nodemap != NULL) {
+		free(ctdb->nodemap);
+	}
+	free(ctdb);
+}
diff --git a/ctdb/tools/ctdb.c b/ctdb/tools/ctdb.c
index 793e98f132..77c45dc58e 100644
--- a/ctdb/tools/ctdb.c
+++ b/ctdb/tools/ctdb.c
@@ -43,6 +43,7 @@ static void usage(void);
 static struct {
 	int timelimit;
 	uint32_t pnn;
+	uint32_t *nodes;
 	int machinereadable;
 	int verbose;
 	int maxruntime;
@@ -66,46 +67,150 @@ static int control_version(struct ctdb_context *ctdb, int argc, const char **arg
 }
 #endif
 
+#define CTDB_NOMEM_ABORT(p) do { if (!(p)) {				\
+		DEBUG(DEBUG_ALERT,("ctdb fatal error: %s\n",		\
+				   "Out of memory in " __location__ ));	\
+		abort();						\
+	}} while (0)
 
-/*
-  verify that a node exists and is reachable
+/* Pretty print the flags to a static buffer in human-readable format.
+ * This never returns NULL!
  */
-static void verify_node(struct ctdb_context *ctdb)
+static const char *pretty_print_flags(uint32_t flags)
 {
-	int ret;
-	struct ctdb_node_map *nodemap=NULL;
+	int j;
+	static const struct {
+		uint32_t flag;
+		const char *name;
+	} flag_names[] = {
+		{ NODE_FLAGS_DISCONNECTED,	    "DISCONNECTED" },
+		{ NODE_FLAGS_PERMANENTLY_DISABLED,  "DISABLED" },
+		{ NODE_FLAGS_BANNED,		    "BANNED" },
+		{ NODE_FLAGS_UNHEALTHY,		    "UNHEALTHY" },
+		{ NODE_FLAGS_DELETED,		    "DELETED" },
+		{ NODE_FLAGS_STOPPED,		    "STOPPED" },
+		{ NODE_FLAGS_INACTIVE,		    "INACTIVE" },
+	};
+	static char flags_str[512]; /* Big enough to contain all flag names */
 
-	if (options.pnn == CTDB_CURRENT_NODE) {
-		return;
+	flags_str[0] = '\0';
+	for (j=0;j<ARRAY_SIZE(flag_names);j++) {
+		if (flags & flag_names[j].flag) {
+			if (flags_str[0] == '\0') {
+				(void) strcpy(flags_str, flag_names[j].name);
+			} else {
+				(void) strcat(flags_str, "|");
+				(void) strcat(flags_str, flag_names[j].name);
+			}
+		}
 	}
-	if (options.pnn == CTDB_BROADCAST_ALL) {
-		return;
+	if (flags_str[0] == '\0') {
+		(void) strcpy(flags_str, "OK");
 	}
 
-	/* verify the node exists */
-	if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap) != 0) {
+	return flags_str;
+}
+
+/* Parse a nodestring.  Parameter dd_ok controls what happens to nodes
+ * that are disconnected or deleted.  If dd_ok is true those nodes are
+ * included in the output list of nodes.  If dd_ok is false, those
+ * nodes are filtered from the "all" case and cause an error if
+ * explicitly specified.
+ */
+static bool parse_nodestring(struct ctdb_context *ctdb,
+			     const char * nodestring,
+			     bool dd_ok,
+			     uint32_t **nodes,
+			     uint32_t *pnn_mode)
+{
+	int n;
+	uint32_t i;
+	struct ctdb_node_map *nodemap;
+       
+	*nodes = NULL;
+
+	if (!ctdb_getnodemap(ctdb_connection, CTDB_CURRENT_NODE, &nodemap)) {
 		DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
 		exit(10);
 	}
-	if (options.pnn >= nodemap->num) {
-		DEBUG(DEBUG_ERR, ("Node %u does not exist\n", options.pnn));
-		exit(ERR_NONODE);
-	}
-	if (nodemap->nodes[options.pnn].flags & NODE_FLAGS_DELETED) {
-		DEBUG(DEBUG_ERR, ("Node %u is DELETED\n", options.pnn));
-		exit(ERR_DISNODE);
-	}
-	if (nodemap->nodes[options.pnn].flags & NODE_FLAGS_DISCONNECTED) {
-		DEBUG(DEBUG_ERR, ("Node %u is DISCONNECTED\n", options.pnn));
-		exit(ERR_DISNODE);
-	}
 
-	/* verify we can access the node */
-	ret = ctdb_ctrl_getpnn(ctdb, TIMELIMIT(), options.pnn);
-	if (ret == -1) {
-		DEBUG(DEBUG_ERR,("Can not access node. Node is not operational.\n"));
-		exit(10);
+	if (nodestring != NULL) {
+		*nodes = talloc_array(ctdb, uint32_t, 0);
+		CTDB_NOMEM_ABORT(*nodes);
+               
+		n = 0;
+
+		if (strcmp(nodestring, "all") == 0) {
+			*pnn_mode = CTDB_BROADCAST_ALL;
+
+			/* all */
+			for (i = 0; i < nodemap->num; i++) {
+				if ((nodemap->nodes[i].flags & 
+				     (NODE_FLAGS_DISCONNECTED |
+				      NODE_FLAGS_DELETED)) && !dd_ok) {
+					continue;
+				}
+				*nodes = talloc_realloc(ctdb, *nodes,
+							uint32_t, n+1);
+				CTDB_NOMEM_ABORT(*nodes);
+				(*nodes)[n] = i;
+				n++;
+			}
+		} else {
+			/* x{,y...} */
+			char *ns, *tok;
+                       
+			ns = talloc_strdup(ctdb, nodestring);
+			tok = strtok(ns, ",");
+			while (tok != NULL) {
+				uint32_t pnn;
+				i = (uint32_t)strtoul(tok, NULL, 0);
+				if (i >= nodemap->num) {
+					DEBUG(DEBUG_ERR, ("Node %u does not exist\n", i));
+					exit(ERR_NONODE);
+				}
+				if ((nodemap->nodes[i].flags & 
+				     (NODE_FLAGS_DISCONNECTED |
+				      NODE_FLAGS_DELETED)) && !dd_ok) {
+					DEBUG(DEBUG_ERR, ("Node %u has status %s\n", i, pretty_print_flags(nodemap->nodes[i].flags)));
+					exit(ERR_DISNODE);
+				}
+				if (!ctdb_getpnn(ctdb_connection, i, &pnn)) {
+					DEBUG(DEBUG_ERR, ("Can not access node %u. Node is not operational.\n", i));
+					exit(10);
+				}
+
+				*nodes = talloc_realloc(ctdb, *nodes,
+							uint32_t, n+1);
+				CTDB_NOMEM_ABORT(*nodes);
+
+				(*nodes)[n] = i;
+				n++;
+
+				tok = strtok(NULL, ",");
+			}
+			talloc_free(ns);
+
+			if (n == 1) {
+				*pnn_mode = (*nodes)[0];
+			} else {
+				*pnn_mode = CTDB_MULTICAST;
+			}
+		}
+	} else {
+		/* default - no nodes specified */
+		*nodes = talloc_array(ctdb, uint32_t, 1);
+		CTDB_NOMEM_ABORT(*nodes);
+
+		if (!ctdb_getpnn(ctdb_connection, CTDB_CURRENT_NODE,
+				 &((*nodes)[0]))) {
+			return false;
+		}
 	}
+
+	ctdb_free_nodemap(nodemap);
+
+	return true;
 }
 
 /*
@@ -607,6 +712,57 @@ static int control_xpnn(struct ctdb_context *ctdb, int argc, const char **argv)
 	return -1;
 }
 
+/* Helpers for ctdb status
+ */
+static bool is_partially_online(struct ctdb_node_and_flags *node)
+{
+	int j;
+	bool ret = false;
+
+	if (node->flags == 0) {
+		struct ctdb_ifaces_list *ifaces;
+
+		if (ctdb_getifaces(ctdb_connection, node->pnn, &ifaces)) {
+			for (j=0; j < ifaces->num; j++) {
+				if (ifaces->ifaces[j].link_state != 0) {
+					continue;
+				}
+				ret = true;
+				break;
+			}
+			ctdb_free_ifaces(ifaces);
+		}
+	}
+
+	return ret;
+}
+
+static int control_status_1_machine(int mypnn, struct ctdb_node_and_flags *node)
+{
+	printf(":%d:%s:%d:%d:%d:%d:%d:%d:%d:%c:\n", node->pnn,
+	       ctdb_addr_to_str(&node->addr),
+	       !!(node->flags&NODE_FLAGS_DISCONNECTED),
+	       !!(node->flags&NODE_FLAGS_BANNED),
+	       !!(node->flags&NODE_FLAGS_PERMANENTLY_DISABLED),
+	       !!(node->flags&NODE_FLAGS_UNHEALTHY),
+	       !!(node->flags&NODE_FLAGS_STOPPED),
+	       !!(node->flags&NODE_FLAGS_INACTIVE),
+	       is_partially_online(node) ? 1 : 0,
+	       (node->pnn == mypnn)?'Y':'N');
+
+	return node->flags;
+}
+
+static int control_status_1_human(int mypnn, struct ctdb_node_and_flags *node)
+{
+       printf("pnn:%d %-16s %s%s\n", node->pnn,
+              ctdb_addr_to_str(&node->addr),
+              is_partially_online(node) ? "PARTIALLYONLINE" : pretty_print_flags(node->flags),
+              node->pnn == mypnn?" (THIS NODE)":"");
+
+       return node->flags;
+}
+
 /*
   display remote ctdb status
  */
@@ -633,100 +789,21 @@ static int control_status(struct ctdb_context *ctdb, int argc, const char **argv
 		printf(":Node:IP:Disconnected:Banned:Disabled:Unhealthy:Stopped"
 		       ":Inactive:PartiallyOnline:ThisNode:\n");
 		for (i=0;i<nodemap->num;i++) {
-			int partially_online = 0;
-			int j;
-
 			if (nodemap->nodes[i].flags & NODE_FLAGS_DELETED) {
 				continue;
 			}
-			if (nodemap->nodes[i].flags == 0) {
-				struct ctdb_ifaces_list *ifaces;
-
-				ret = ctdb_getifaces(ctdb_connection,
-						     nodemap->nodes[i].pnn,
-						     &ifaces);
-				if (ret == 0) {
-					for (j=0; j < ifaces->num; j++) {
-						if (ifaces->ifaces[j].link_state != 0) {
-							continue;
-						}
-						partially_online = 1;
-						break;
-					}
-					ctdb_free_ifaces(ifaces);
-				}
-			}
-			printf(":%d:%s:%d:%d:%d:%d:%d:%d:%d:%c:\n", nodemap->nodes[i].pnn,
-				ctdb_addr_to_str(&nodemap->nodes[i].addr),
-			       !!(nodemap->nodes[i].flags&NODE_FLAGS_DISCONNECTED),
-			       !!(nodemap->nodes[i].flags&NODE_FLAGS_BANNED),
-			       !!(nodemap->nodes[i].flags&NODE_FLAGS_PERMANENTLY_DISABLED),
-			       !!(nodemap->nodes[i].flags&NODE_FLAGS_UNHEALTHY),
-			       !!(nodemap->nodes[i].flags&NODE_FLAGS_STOPPED),
-			       !!(nodemap->nodes[i].flags&NODE_FLAGS_INACTIVE),
-			       partially_online,
-			       (nodemap->nodes[i].pnn == mypnn)?'Y':'N');
+			(void) control_status_1_machine(mypnn,
+							&nodemap->nodes[i]);
 		}
 		return 0;
 	}
 
 	printf("Number of nodes:%d\n", nodemap->num);
 	for(i=0;i<nodemap->num;i++){
-		static const struct {
-			uint32_t flag;
-			const char *name;
-		} flag_names[] = {
-			{ NODE_FLAGS_DISCONNECTED,          "DISCONNECTED" },
-			{ NODE_FLAGS_PERMANENTLY_DISABLED,  "DISABLED" },
-			{ NODE_FLAGS_BANNED,                "BANNED" },
-			{ NODE_FLAGS_UNHEALTHY,             "UNHEALTHY" },
-			{ NODE_FLAGS_DELETED,               "DELETED" },
-			{ NODE_FLAGS_STOPPED,               "STOPPED" },
-			{ NODE_FLAGS_INACTIVE,              "INACTIVE" },
-		};
-		char *flags_str = NULL;
-		int j;
-
 		if (nodemap->nodes[i].flags & NODE_FLAGS_DELETED) {
 			continue;
 		}
-		if (nodemap->nodes[i].flags == 0) {
-			struct ctdb_control_get_ifaces *ifaces;
-
-			ret = ctdb_ctrl_get_ifaces(ctdb, TIMELIMIT(),
-						   nodemap->nodes[i].pnn,
-						   ctdb, &ifaces);
-			if (ret == 0) {
-				for (j=0; j < ifaces->num; j++) {
-					if (ifaces->ifaces[j].link_state != 0) {
-						continue;
-					}
-					flags_str = talloc_strdup(ctdb, "PARTIALLYONLINE");
-					break;
-				}
-				talloc_free(ifaces);
-			}
-		}
-		for (j=0;j<ARRAY_SIZE(flag_names);j++) {
-			if (nodemap->nodes[i].flags & flag_names[j].flag) {
-				if (flags_str == NULL) {
-					flags_str = talloc_strdup(ctdb, flag_names[j].name);
-				} else {
-					flags_str = talloc_asprintf_append(flags_str, "|%s",
-									   flag_names[j].name);
-				}
-				CTDB_NO_MEMORY_FATAL(ctdb, flags_str);
-			}
-		}
-		if (flags_str == NULL) {
-			flags_str = talloc_strdup(ctdb, "OK");
-			CTDB_NO_MEMORY_FATAL(ctdb, flags_str);
-		}
-		printf("pnn:%d %-16s %s%s\n", nodemap->nodes[i].pnn,
-		       ctdb_addr_to_str(&nodemap->nodes[i].addr),
-		       flags_str,
-		       nodemap->nodes[i].pnn == mypnn?" (THIS NODE)":"");
-		talloc_free(flags_str);
+		(void) control_status_1_human(mypnn, &nodemap->nodes[i]);
 	}
 
 	ret = ctdb_ctrl_getvnnmap(ctdb, TIMELIMIT(), options.pnn, ctdb, &vnnmap);
@@ -759,6 +836,50 @@ static int control_status(struct ctdb_context *ctdb, int argc, const char **argv
 	return 0;
 }
 
+static int control_nodestatus(struct ctdb_context *ctdb, int argc, const char **argv)
+{
+	int i, ret;
+	struct ctdb_node_map *nodemap=NULL;
+	uint32_t * nodes;
+	uint32_t pnn_mode, mypnn;
+
+	if (argc > 1) {
+		usage();
+	}
+
+	if (!parse_nodestring(ctdb, argc == 1 ? argv[0] : NULL, false,
+			      &nodes, &pnn_mode)) {
+		return -1;
+	}
+
+	if (options.machinereadable) {
+		printf(":Node:IP:Disconnected:Banned:Disabled:Unhealthy:Stopped"
+		       ":Inactive:PartiallyOnline:ThisNode:\n");
+	} else if (pnn_mode == CTDB_BROADCAST_ALL) {
+		printf("Number of nodes:%d\n", (int) talloc_array_length(nodes));
+	}
+
+	if (!ctdb_getpnn(ctdb_connection, CTDB_CURRENT_NODE, &mypnn)) {
+		DEBUG(DEBUG_ERR, ("Unable to get PNN from local node\n"));
+		return -1;
+	}
+
+	if (!ctdb_getnodemap(ctdb_connection, CTDB_CURRENT_NODE, &nodemap)) {
+		DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
+		return -1;
+	}
+
+	for (i = 0; i < talloc_array_length(nodes); i++) {
+		if (options.machinereadable) {
+			ret |= control_status_1_machine(mypnn,
+							&nodemap->nodes[nodes[i]]);
+		} else {
+			ret |= control_status_1_human(mypnn,
+						      &nodemap->nodes[nodes[i]]);
+		}
+	}
+	return ret;
+}
 
 struct natgw_node {
 	struct natgw_node *next;
@@ -5204,6 +5325,7 @@ static const struct {
 	{ "writekey", 	     control_writekey,      	true,	false,  "write to a database key", "<tdb-file> <key> <value>" },
 	{ "checktcpport",    control_chktcpport,      	false,	true,  "check if a service is bound to a specific tcp port or not", "<port>" },
 	{ "getdbseqnum",     control_getdbseqnum,       false,	false, "get the sequence number off a database", "<dbid>" },
+	{ "nodestatus",      control_nodestatus,        true,   false,  "show and return node status" },
 };
 
 /*
@@ -5356,45 +5478,32 @@ int main(int argc, const char *argv[])
 		exit(1);
 	}				
 
-	/* setup the node number to contact */
-	if (nodestring != NULL) {
-		if (strcmp(nodestring, "all") == 0) {
-			options.pnn = CTDB_BROADCAST_ALL;
-		} else {
-			options.pnn = strtoul(nodestring, NULL, 0);
-		}
+	/* setup the node number(s) to contact */
+	if (!parse_nodestring(ctdb, nodestring, true,
+			      &options.nodes, &options.pnn)) {
+		usage();
 	}
 
-	/* verify the node exists */
-	verify_node(ctdb);
-
 	if (options.pnn == CTDB_CURRENT_NODE) {
-		int pnn;
-		pnn = ctdb_ctrl_getpnn(ctdb, TIMELIMIT(), options.pnn);		
-		if (pnn == -1) {
-			return -1;
-		}
-		options.pnn = pnn;
+		options.pnn = options.nodes[0];
 	}
 
 	if (ctdb_commands[i].auto_all && 
-	    options.pnn == CTDB_BROADCAST_ALL) {
-		uint32_t *nodes;
-		uint32_t num_nodes;
+	    ((options.pnn == CTDB_BROADCAST_ALL) ||
+	     (options.pnn == CTDB_MULTICAST))) {
 		int j;
-		ret = 0;
 
-		nodes = ctdb_get_connected_nodes(ctdb, TIMELIMIT(), ctdb, &num_nodes);
-		CTDB_NO_MEMORY(ctdb, nodes);
-	
-		for (j=0;j<num_nodes;j++) {
-			options.pnn = nodes[j];
+		ret = 0;
+		for (j = 0; j < talloc_array_length(options.nodes); j++) {
+			options.pnn = options.nodes[j];
 			ret |= ctdb_commands[i].fn(ctdb, extra_argc-1, extra_argv+1);
 		}
-		talloc_free(nodes);
 	} else {
 		ret = ctdb_commands[i].fn(ctdb, extra_argc-1, extra_argv+1);
 	}
 
+	ctdb_disconnect(ctdb_connection);
+	talloc_free(ctdb);
+
 	return ret;
 }
diff --git a/ctdb/tools/ctdb_vacuum.c b/ctdb/tools/ctdb_vacuum.c
index 419e660a7f..3bde7b384d 100644
--- a/ctdb/tools/ctdb_vacuum.c
+++ b/ctdb/tools/ctdb_vacuum.c
@@ -478,7 +478,7 @@ int ctdb_vacuum(struct ctdb_context *ctdb, int argc, const char **argv)
 	return 0;
 }
 
-struct traverse_state {
+struct vacuum_traverse_state {
 	bool error;
 	struct tdb_context *dest_db;
 };
@@ -488,7 +488,7 @@ struct traverse_state {
  */
 static int repack_traverse(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *private)
 {
-	struct traverse_state *state = (struct traverse_state *)private;
+	struct vacuum_traverse_state *state = (struct vacuum_traverse_state *)private;
 	if (tdb_store(state->dest_db, key, data, TDB_INSERT) != 0) {
 		state->error = true;
 		return -1;
@@ -502,7 +502,7 @@ static int repack_traverse(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data,
 static int ctdb_repack_tdb(struct tdb_context *tdb)
 {
 	struct tdb_context *tmp_db;
-	struct traverse_state state;
+	struct vacuum_traverse_state state;
 
 	if (tdb_transaction_start(tdb) != 0) {
 		DEBUG(DEBUG_ERR,(__location__ " Failed to start transaction\n"));