[Cluster-devel] [PATCH] rgmanager: dbus notifications (Merged)

[Fabio M. Di Nitto <fdinitto@redhat.com>]

You  might as well cherry pick the build system patch to enable/disable
dbus in the same patch set.

That way we have no delta and we can avoid possible merge conflicts later.

Fabio

On 02/03/2011 01:37 AM, Lon Hohberger wrote:
> Includes aforementioned makefile fix
> 
> Everything in one patch for easy reading.
> 
> Resolves: rhbz#657756
> 
> Signed-off-by: Lon Hohberger <lhh@redhat.com>
> ---
>  rgmanager/include/rg_dbus.h         |   20 +++
>  rgmanager/man/rgmanager.8           |    2 +
>  rgmanager/src/daemons/Makefile      |    7 +-
>  rgmanager/src/daemons/main.c        |   19 ++-
>  rgmanager/src/daemons/update-dbus.c |  266 +++++++++++++++++++++++++++++++++++
>  5 files changed, 309 insertions(+), 5 deletions(-)
>  create mode 100644 rgmanager/include/rg_dbus.h
>  create mode 100644 rgmanager/src/daemons/update-dbus.c
> 
> diff --git a/rgmanager/include/rg_dbus.h b/rgmanager/include/rg_dbus.h
> new file mode 100644
> index 0000000..3ef5ae1
> --- /dev/null
> +++ b/rgmanager/include/rg_dbus.h
> @@ -0,0 +1,20 @@
> +#ifndef _RGM_DBUS_H
> +#define _RGM_DBUS_H
> +
> +int rgm_dbus_init(void);
> +int rgm_dbus_release(void);
> +extern int rgm_dbus_notify;
> +
> +#ifdef DBUS
> +
> +#define RGM_DBUS_DEFAULT 1
> +#define RGM_DBUS_UPDATE (rgm_dbus_notify?rgm_dbus_update:0)
> +int32_t rgm_dbus_update(char *key, uint64_t view, void *data, uint32_t size);
> +
> +#else
> +
> +#define RGM_DBUS_DEFAULT 0
> +#define RGM_DBUS_UPDATE NULL
> +
> +#endif /* DBUS */
> +#endif
> diff --git a/rgmanager/man/rgmanager.8 b/rgmanager/man/rgmanager.8
> index 39b2195..2b018d2 100644
> --- a/rgmanager/man/rgmanager.8
> +++ b/rgmanager/man/rgmanager.8
> @@ -371,6 +371,8 @@ will leave the virtual machine running.
>  Run in the foreground (do not fork).
>  .IP \-d
>  Enable debug-level logging.
> +.IP \-q
> +Disable DBus signals which are normally sent when services change state.
>  .IP \-w
>  Disable internal process monitoring (for debugging).
>  .IP \-N
> diff --git a/rgmanager/src/daemons/Makefile b/rgmanager/src/daemons/Makefile
> index 94e0dbb..4dbbe5a 100644
> --- a/rgmanager/src/daemons/Makefile
> +++ b/rgmanager/src/daemons/Makefile
> @@ -29,6 +29,7 @@ OBJS1=	fo_domain.o \
>  	service_op.o \
>  	slang_event.o \
>  	event_config.o \
> +	update-dbus.o \
>  	watchdog.o
>  
>  OBJS2=	test-noccs.o \
> @@ -40,12 +41,13 @@ OBJS2=	test-noccs.o \
>  	rg_locks-noccs.o \
>  	event_config-noccs.o
>  
> -CFLAGS += -DSHAREDIR=\"${sharedir}\" -D_GNU_SOURCE
> +CFLAGS += -DDBUS -DSHAREDIR=\"${sharedir}\" -D_GNU_SOURCE
>  CFLAGS += -fPIC
>  CFLAGS += -I${ccsincdir} -I${cmanincdir} -I${dlmincdir} -I${logtincdir}
>  CFLAGS += `xml2-config --cflags` -I${slangincdir}
>  CFLAGS += -I$(S)/../../include
>  CFLAGS += -I${incdir}
> +CFLAGS += `pkg-config --cflags dbus-1`
>  
>  NOCCS_CFLAGS += -DNO_CCS
>  
> @@ -59,6 +61,7 @@ DLM_LDFLAGS += -L${dlmlibdir} -ldlm
>  XML2_LDFLAGS += `xml2-config --libs`
>  SLANG_LDFLAGS += -L${slanglibdir} -lslang
>  EXTRA_LDFLAGS += -lpthread
> +DBUS_LDFLAGS += `pkg-config --libs dbus-1`
>  
>  LDDEPS += ../clulib/libclulib.a
>  
> @@ -66,7 +69,7 @@ ${TARGET1}: ${OBJS1} ${LDDEPS}
>  	$(CC) -o $@ $^ $(CCS_LDFLAGS) $(CMAN_LDFLAGS) \
>  			$(DLM_LDFLAGS) $(XML2_LDFLAGS) \
>  			$(SLANG_LDFLAGS) $(EXTRA_LDFLAGS) \
> -			$(LOGSYS_LDFLAGS) $(LD_FLAGS)
> +			$(LOGSYS_LDFLAGS) $(LD_FLAGS) $(DBUS_LDFLAGS)
>  
>  #
>  # Our test program links against the local allocator so that
> diff --git a/rgmanager/src/daemons/main.c b/rgmanager/src/daemons/main.c
> index 4e34246..02cd2dc 100644
> --- a/rgmanager/src/daemons/main.c
> +++ b/rgmanager/src/daemons/main.c
> @@ -23,6 +23,7 @@
>  #include <members.h>
>  #include <daemon_init.h>
>  #include <groups.h>
> +#include <rg_dbus.h>
>  
>  #ifdef WRAP_THREADS
>  void dump_thread_states(FILE *);
> @@ -43,7 +44,6 @@ static uint8_t ALIGNED port = RG_PORT;
>  static char *rgmanager_lsname = (char *)"rgmanager"; /* XXX default */
>  static int status_poll_interval = DEFAULT_CHECK_INTERVAL;
>  
> -
>  static void
>  segfault(int __attribute__ ((unused)) sig)
>  {
> @@ -684,6 +684,8 @@ event_loop(msgctx_t *localctx, msgctx_t *clusterctx)
>  		dump_internal_state("/var/lib/cluster/rgmanager-dump");
>  	}
>  
> +	rgm_dbus_init();
> +
>  	while (running && (tv.tv_sec || tv.tv_usec)) {
>  		FD_ZERO(&rfds);
>  		max = -1;
> @@ -949,7 +951,7 @@ main(int argc, char **argv)
>  	pthread_t th;
>  	cman_handle_t clu = NULL;
>  
> -	while ((rv = getopt(argc, argv, "wfdN")) != EOF) {
> +	while ((rv = getopt(argc, argv, "wfdNq")) != EOF) {
>  		switch (rv) {
>  		case 'w':
>  			wd = 0;
> @@ -963,6 +965,9 @@ main(int argc, char **argv)
>  		case 'f':
>  			foreground = 1;
>  			break;
> +		case 'q':
> +			rgm_dbus_notify = 0;
> +			break;
>  		default:
>  			return 1;
>  			break;
> @@ -1032,6 +1037,12 @@ main(int argc, char **argv)
>  	configure_rgmanager(-1, debug, &cluster_timeout);
>  	logt_print(LOG_NOTICE, "Resource Group Manager Starting\n");
>  
> +	if (rgm_dbus_notify && rgm_dbus_init() != 0) {
> +		rgm_dbus_notify = 0;
> +		logt_print(LOG_NOTICE, "Failed to initialize DBus; "
> +			   "notifications disabled\n");
> +	}
> +
>  	if (init_resource_groups(0, do_init) != 0) {
>  		logt_print(LOG_CRIT, "#8: Couldn't initialize services\n");
>  		goto out_ls;
> @@ -1074,7 +1085,8 @@ main(int argc, char **argv)
>  
>  	ds_key_init("rg_lockdown", 32, 10);
>  #else
> -	if (vf_init(me.cn_nodeid, port, NULL, NULL, cluster_timeout) != 0) {
> +	if (vf_init(me.cn_nodeid, port, NULL, RGM_DBUS_UPDATE,
> +		    cluster_timeout) != 0) {
>  		logt_print(LOG_CRIT, "#11: Couldn't set up VF listen socket\n");
>  		goto out_ls;
>  	}
> @@ -1106,6 +1118,7 @@ out_ls:
>  	clu_lock_finished(rgmanager_lsname);
>  
>  out:
> +	rgm_dbus_release();
>  	logt_print(LOG_NOTICE, "Shutdown complete, exiting\n");
>  	cman_finish(clu);
>  	
> diff --git a/rgmanager/src/daemons/update-dbus.c b/rgmanager/src/daemons/update-dbus.c
> new file mode 100644
> index 0000000..f465d1d
> --- /dev/null
> +++ b/rgmanager/src/daemons/update-dbus.c
> @@ -0,0 +1,266 @@
> +/* DBus notifications */
> +#include <stdint.h>
> +#include <rg_dbus.h>
> +#include <errno.h>
> +
> +#ifdef DBUS
> +
> +#include <stdio.h>
> +#include <stdint.h>
> +#include <resgroup.h>
> +#include <poll.h>
> +#include <dbus/dbus.h>
> +#include <liblogthread.h>
> +#include <members.h>
> +
> +
> +#define DBUS_RGM_NAME	"com.redhat.cluster.rgmanager"
> +#define DBUS_RGM_IFACE	"com.redhat.cluster.rgmanager"
> +#define DBUS_RGM_PATH	"/com/redhat/cluster/rgmanager"
> +
> +static DBusConnection *db = NULL;
> +static pthread_mutex_t mu = PTHREAD_MUTEX_INITIALIZER;
> +static pthread_t th = 0;
> +#endif
> +
> +/* Set this to the desired value prior to calling rgm_dbus_init() */
> +int rgm_dbus_notify = RGM_DBUS_DEFAULT;
> +
> +
> +int 
> +rgm_dbus_init(void)
> +#ifdef DBUS
> +{
> +	DBusConnection *dbc = NULL;
> +	DBusError err;
> +
> +	if (!rgm_dbus_notify)
> +		return 0;
> +
> +	pthread_mutex_lock(&mu);
> +	if (db) {
> +		pthread_mutex_unlock(&mu);
> +		return 0;
> +	}
> +
> +	dbus_error_init(&err);
> +
> +	dbc = dbus_bus_get_private(DBUS_BUS_SYSTEM, &err);
> +	if (!dbc) {
> +		logt_print(LOG_DEBUG,
> +			   "DBus Failed to initialize: dbus_bus_get: %s\n",
> +			   err.message);
> +		dbus_error_free(&err);
> +		pthread_mutex_unlock(&mu);
> +		return -1;
> +	}
> +
> +	dbus_connection_set_exit_on_disconnect(dbc, FALSE);
> +
> +	db = dbc;
> +	pthread_mutex_unlock(&mu);
> +	logt_print(LOG_DEBUG, "DBus Notifications Initialized\n");
> +	return 0;
> +}
> +#else
> +{
> +	errno = ENOSYS;
> +	return -1;
> +}
> +#endif
> +
> +
> +#ifdef DBUS
> +static int
> +_rgm_dbus_release(void)
> +{
> +	pthread_t t;
> +
> +	if (!db)
> +		return 0;
> +
> +	/* tell thread to exit - not sure how to tell dbus
> +	 * to wake up, so just have it poll XXX */
> +
> +	/* if the thread left because the dbus connection died,
> +	   this block is avoided */
> +	if (th) {
> +		t = th;
> +		th = 0;
> +		pthread_join(t, NULL);
> +	}
> +
> +	dbus_connection_close(db);
> +	dbus_connection_unref(db);
> +	db = NULL;
> +
> +	logt_print(LOG_DEBUG, "DBus Released\n");
> +	return 0;
> +}
> +#endif
> +
> +
> +/* Clean shutdown (e.g. when exiting */
> +int
> +rgm_dbus_release(void)
> +#ifdef DBUS
> +{
> +	int ret;
> +
> +	pthread_mutex_lock(&mu);
> +	ret = _rgm_dbus_release();
> +	pthread_mutex_unlock(&mu);
> +	return ret;
> +}
> +#else
> +{
> +	return 0;
> +}
> +#endif
> +
> +
> +#ifdef DBUS
> +/* Auto-flush thread.  Since sending only guarantees queueing,
> + * we need this thread to push things out over dbus in the
> + * background */
> +static void *
> +_dbus_auto_flush(void *arg)
> +{
> +	/* DBus connection functions are thread safe */
> +	dbus_connection_ref(db);
> +	while (dbus_connection_read_write(db, 500)) {
> +		if (!th)
> +			break;	
> +	}
> +
> +	dbus_connection_unref(db);
> +	th = 0;
> +	return NULL;
> +}
> +
> +
> +static int
> +_rgm_dbus_notify(const char *svcname,
> +		 const char *svcstatus,
> +		 const char *svcflags,
> +		 const char *svcowner,
> +		 const char *svclast)
> +{
> +	DBusMessage *msg = NULL;
> +	int ret = -1;
> +
> +	if (!db) {
> +		goto out_free;
> +	}
> +
> +	pthread_mutex_lock(&mu);
> +
> +	/* Check to ensure the connection is still valid. If it
> +	 * isn't, clean up and shut down the dbus connection.
> +	 *
> +	 * The main rgmanager thread will periodically try to
> +	 * reinitialize the dbus notification subsystem unless
> +	 * the administrator ran rgmanager with the -D command
> +	 * line option.
> +	 */
> +	if (dbus_connection_get_is_connected(db) != TRUE) {
> +		goto out_unlock;
> +	}
> +
> +	if (!th) {
> +		/* start auto-flush thread if needed */
> +		pthread_create(&th, NULL, _dbus_auto_flush, NULL);
> +	}
> +
> +	if (!(msg = dbus_message_new_signal(DBUS_RGM_PATH,
> +	      				    DBUS_RGM_IFACE,
> +	      				    "ServiceStateChange"))) {
> +		goto out_unlock;
> +	}
> +
> +	if (!dbus_message_append_args(msg,
> +	 			      DBUS_TYPE_STRING, &svcname,
> +	 			      DBUS_TYPE_STRING, &svcstatus,
> +	 			      DBUS_TYPE_STRING, &svcflags,
> + 				      DBUS_TYPE_STRING, &svcowner,
> + 				      DBUS_TYPE_STRING, &svclast,
> +	    			      DBUS_TYPE_INVALID)) {
> +		goto out_unlock;
> +	}
> +
> +	dbus_connection_send(db, msg, NULL);
> +	ret = 0;
> +
> +out_unlock:
> +	pthread_mutex_unlock(&mu);
> +	if (msg)
> +		dbus_message_unref(msg);
> +out_free:
> +	return ret;
> +}
> +
> +
> +/*
> + * view-formation callback function
> + */
> +int32_t
> +rgm_dbus_update(char *key, uint64_t view, void *data, uint32_t size)
> +{
> +	char flags[64];
> +	rg_state_t *st;
> +	cluster_member_list_t *m = NULL;
> +	const char *owner;
> +	const char *last;
> +	int ret = 0;
> +
> +	if (!rgm_dbus_notify)
> +		goto out_free;
> +	if (!db)
> +		goto out_free;
> +	if (view == 1)
> +		goto out_free;
> +	if (size != (sizeof(*st)))
> +		goto out_free;
> +
> +	st = (rg_state_t *)data;
> +	swab_rg_state_t(st);
> +
> +	/* Don't send transitional states */
> +	if (st->rs_state == RG_STATE_STARTING ||
> +	    st->rs_state == RG_STATE_STOPPING)
> +		goto out_free;
> +
> +	m = member_list();
> +	if (!m)
> +		goto out_free;
> +
> +	owner = memb_id_to_name(m, st->rs_owner);
> +	last = memb_id_to_name(m, st->rs_last_owner);
> +
> +	if (!owner)
> +		owner = "(none)";
> +	if (!last)
> +		last = "(none)";
> +
> +	flags[0] = 0;
> +	rg_flags_str(flags, sizeof(flags), st->rs_flags, (char *)" ");
> +	if (flags[0] == 0)
> +		snprintf(flags, sizeof(flags), "(none)");
> +
> +	ret = _rgm_dbus_notify(st->rs_name,
> +			       rg_state_str(st->rs_state),
> +			       (char *)flags, owner, last);
> +
> +	if (ret < 0) {
> +		logt_print(LOG_ERR, "Error sending update for %s; "
> +			   "DBus notifications disabled\n", key);
> +		rgm_dbus_release();
> +	}
> +
> +out_free:
> +	if (m)
> +		free_member_list(m);
> +	free(data);
> +	return 0;
> +}
> +#endif