client_lib.c

linux集群服务器软件代码包

/* $Id: client_lib.c,v 1.21 2005/02/17 18:20:03 gshi Exp $ */
/* 
 * client_lib: heartbeat API client side code
 *
 * Copyright (C) 2000 Alan Robertson <[e-mail]>
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 * 
 * This library 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
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

/*
 *	Here's our approach:
 *
 *	Have each application connect to heartbeat via our IPC layer.
 *		This IPC layer currently uses sockets and provides
 *		a suitable authorization API.
 *
 *	We can validate permissions for "sniffing" using the builtin
 *		IPC authorization API.
 *
 *	This code thankfully no longer uses FIFOs.
 *
 */

#include <portability.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/utsname.h>
#include <sys/time.h>
#include <sys/stat.h>
#include <stdarg.h>
#include <heartbeat.h>
#include <hb_api_core.h>
#include <hb_api.h>
#include <clplumbing/ttylock.h>
#include <glib.h>

struct sys_config *		config  = NULL;

int			netstring_format = TRUE;

struct stringlist {
	char *			value;
	struct stringlist *	next;
};


/*
 *	Queue of messages to be read later...
 */
struct MsgQueue {
	struct ha_msg *		value;
	struct MsgQueue *	next;
	struct MsgQueue *	prev;
};

typedef struct gen_callback {
	char *			msgtype;
	llc_msg_callback_t 	cf;
	void *			pd;
	struct gen_callback*	next;
}gen_callback_t;

#define	MXFIFOPATH	128
#define	HBPREFIX	"LCK..hbapi:"

/* Order sequence */
typedef struct order_seq {
	char			to_node[HOSTLENG];
	seqno_t			seqno;
	struct order_seq *	next;
}order_seq_t;

/* Order Queue */
struct orderQ {
	struct ha_msg *		orderQ[MAXMSGHIST];
	int			curr_index;
	seqno_t			curr_oseqno;
	seqno_t			curr_gen;
	seqno_t			curr_client_gen;
	seqno_t			first_msg_seq;
	seqno_t			first_msg_gen;
	seqno_t			first_msg_client_gen;
	struct orderQ		*backupQ;
};

typedef struct order_queue {
	char			from_node[HOSTLENG];
	struct orderQ		node;
	struct orderQ		cluster;
	struct order_queue*	next;
	struct ha_msg*		leave_msg;
        int			client_leaving;
}order_queue_t;

/*
 *	Our heartbeat private data
 */
typedef struct llc_private {
	const char *		PrivateId;	/* A "magic cookie */
	llc_nstatus_callback_t	node_callback;	/* Node status callback fcn */
	void*			node_private;	/* node status callback data*/
	llc_ifstatus_callback_t	if_callback;	/* IF status callback fcn */
	void*			if_private;	/* IF status callback data */
	llc_cstatus_callback_t	cstatus_callback;/*Client status callback fcn */
	void*			client_private;	/* client status callback data*/
	struct gen_callback*	genlist;	/* List of general callbacks*/
	IPC_Channel*		chan;		/* IPC communication channel*/
	struct stringlist *	nodelist;	/* List of nodes from query */
	struct stringlist *	iflist;		/* List of IFs from query */
	int			SignedOn;	/* 1 if we're signed on */
	int			iscasual;	/* 1 if casual client */
	long			deadtime_ms;	/* heartbeat's deadtime */
	long			keepalive_ms;	/* HB's keepalive time*/
	int			logfacility;	/* HB's logging facility */
	struct stringlist*	nextnode;	/* Next node for walknode */
	struct stringlist*	nextif;		/* Next interface for walkif*/
	/* Messages to be read after current call completes */
	struct MsgQueue *	firstQdmsg;
	struct MsgQueue *	lastQdmsg;
	/* The next two items are for ordered message delivery */
	order_seq_t		order_seq_head;	/* head of order_seq list */
	order_queue_t*		order_queue_head;/* head of order queue */
}llc_private_t;

static const char * OurID = "Heartbeat private data";	/* "Magic cookie" */

#define ISOURS(l) (l && l->ll_cluster_private &&			\
		(((llc_private_t*)(l->ll_cluster_private))->PrivateId) == OurID)
#define SEQGAP		16
#define DEBUGORDER	0

static void		ClearLog(void);
			/* Common code for request messages */
static struct ha_msg*	hb_api_boilerplate(const char * apitype);
static int		hb_api_signon(struct ll_cluster*, const char * clientid);
static int		hb_api_signoff(struct ll_cluster*);
static int		hb_api_setfilter(struct ll_cluster*, unsigned);
static void		destroy_stringlist(struct stringlist *);
static struct stringlist*
			new_stringlist(const char *);
static int		get_nodelist(llc_private_t*);
static void		zap_nodelist(llc_private_t*);
static int		get_iflist(llc_private_t*, const char *host);
static void		zap_iflist(llc_private_t*);
static void		zap_order_seq(llc_private_t* pi);
static void		zap_order_queue(llc_private_t* pi);
static int		enqueue_msg(llc_private_t*,struct ha_msg*);
static struct ha_msg*	dequeue_msg(llc_private_t*);
static gen_callback_t*	search_gen_callback(const char * type, llc_private_t*);
static int		add_gen_callback(const char * msgtype
,	llc_private_t*, llc_msg_callback_t, void*);
static int		del_gen_callback(llc_private_t*, const char * msgtype);

static struct ha_msg*	read_api_msg(llc_private_t*);
static struct ha_msg*	read_cstatus_respond_msg(llc_private_t*pi, int timeout);
static struct ha_msg*	read_hb_msg(ll_cluster_t*, int blocking);

static int		hb_api_setsignal(ll_cluster_t*, int nsig);
static int set_msg_callback
			(ll_cluster_t*, const char * msgtype
,			llc_msg_callback_t callback, void * p);
static int
set_nstatus_callback (ll_cluster_t*
,		llc_nstatus_callback_t cbf, 	void * p);
static int
set_cstatus_callback (ll_cluster_t*
,		llc_cstatus_callback_t cbf, void * p);
static int
set_ifstatus_callback (ll_cluster_t* ci
,		llc_ifstatus_callback_t cbf, void * p);
static int init_nodewalk (ll_cluster_t*);
static const char * nextnode (ll_cluster_t* ci);
static int init_ifwalk (ll_cluster_t* ci, const char * host);
static const char *	get_nodestatus(ll_cluster_t*, const char *host);
static const char *
get_clientstatus(ll_cluster_t*, const char *host, const char *clientid
,	int timeout);
static const char *	get_nodetype(ll_cluster_t*, const char *host);
static const char *	get_ifstatus(ll_cluster_t*, const char *host
,	const char * intf);
static char *		get_parameter(ll_cluster_t*, const char* pname);
static const char *	get_resources(ll_cluster_t*);
static int		get_inputfd(ll_cluster_t*);
static IPC_Channel*	get_ipcchan(ll_cluster_t*);
static int		msgready(ll_cluster_t*);
static int		setfmode(ll_cluster_t*, unsigned mode);
static int		sendclustermsg(ll_cluster_t*, struct ha_msg* msg);
static int		sendnodemsg(ll_cluster_t*, struct ha_msg* msg
,			const char * nodename);

STATIC void		add_order_seq(llc_private_t*, struct ha_msg* msg);
static int		send_ordered_clustermsg(ll_cluster_t* lcl, struct ha_msg* msg);
static int		send_ordered_nodemsg(ll_cluster_t* lcl, struct ha_msg* msg
,			const char * nodename);
static const char *	APIError(ll_cluster_t*);
static int		CallbackCall(llc_private_t* p, struct ha_msg * msg);
static struct ha_msg *	read_msg_w_callbacks(ll_cluster_t* llc, int blocking);
static int		rcvmsg(ll_cluster_t* llc, int blocking);

volatile struct process_info *	curproc = NULL;
static char		OurPid[16];
static const char *	OurClientID = NULL;
static char 		OurNode[SYS_NMLN];
static ll_cluster_t*	hb_cluster_new(void);

static void ha_api_perror(const char * fmt, ...) G_GNUC_PRINTF(1,2);
static void ha_api_log(int priority, const char * fmt, ...) G_GNUC_PRINTF(2,3);


#define	ZAPMSG(m)	{ha_msg_del(m); (m) = NULL;}

/*
 * All the boilerplate common to creating heartbeat API request
 * messages.
 */
static struct ha_msg*
hb_api_boilerplate(const char * apitype)
{
	struct ha_msg*	msg;

	if ((msg = ha_msg_new(4)) == NULL) {
		ha_api_log(LOG_ERR, "boilerplate: out of memory");
		return msg;
	}

	/* Message type: API request */
	if (ha_msg_add(msg, F_TYPE, T_APIREQ) != HA_OK) {
		ha_api_log(LOG_ERR, "boilerplate: cannot add F_TYPE field");
		ZAPMSG(msg);
		return msg;
	}
	/* Add field for API request type */
	if (ha_msg_add(msg, F_APIREQ, apitype) != HA_OK) {
		ha_api_log(LOG_ERR, "boilerplate: cannot add F_APIREQ field");
		ZAPMSG(msg);
		return msg;
	}
	/* Add field for destination */
	if (ha_msg_add(msg, F_TO, OurNode) != HA_OK) {
		ha_api_log(LOG_ERR, "boilerplate: cannot add F_TO field");
		ZAPMSG(msg);
		return msg;
	}
	/* Add our PID to the message */
	if (ha_msg_add(msg, F_PID, OurPid) != HA_OK) {
		ha_api_log(LOG_ERR, "boilerplate: cannot add F_PID field");
		ZAPMSG(msg);
		return msg;
	}
	
	/* Add our client ID to the message */
	if (ha_msg_add(msg, F_FROMID, OurClientID) != HA_OK) {
		ha_api_log(LOG_ERR, "boilerplate: cannot add F_FROMID field");
		ZAPMSG(msg);
		return msg;
	}
	return(msg);
}

/*
 * Sign ourselves on as a heartbeat client process.
 */
static int
hb_api_signon(struct ll_cluster* cinfo, const char * clientid)
{
	struct ha_msg*	request;
	struct ha_msg*	reply;
	struct utsname	un;
	int		rc;
	const char *	result;
	int		iscasual;
	llc_private_t* pi;
	const char	*tmpstr;
        char		regpath[] = API_REGSOCK;
	char		path[] = IPC_PATH_ATTR;
	GHashTable*	wchanattrs;
	char		cuid[20];
	char		cgid[20];

	if (!ISOURS(cinfo)) {
		ha_api_log(LOG_ERR, "hb_api_signon: bad cinfo");
		return HA_FAIL;
	}
	pi = (llc_private_t*)cinfo->ll_cluster_private;

	/* Re-sign ourselves back on */
	if (pi->SignedOn) {
		hb_api_signoff(cinfo);
	}

	snprintf(OurPid, sizeof(OurPid), "%d", getpid());

	/* Create our client id */
	if (clientid != NULL) {
		OurClientID = clientid;
		iscasual = 0;
	}else{
		OurClientID = OurPid;
		iscasual = 1;
	}

	pi->iscasual = iscasual;


	if (uname(&un) < 0) {
		ha_api_perror("uname failure");
		return HA_FAIL;
	}
        memset(OurNode, 0, sizeof(OurNode));
	strncpy(OurNode, un.nodename, sizeof(OurNode)-1);
	g_strdown(OurNode);

	/* Initialize order_seq_head */
	pi->order_seq_head.seqno = 1;
	pi->order_seq_head.to_node[0] = '\0';
	pi->order_seq_head.next = NULL;

	/* Initialize order_queue_head */
	pi->order_queue_head = NULL;

	/* Crank out the boilerplate */
	if ((request = hb_api_boilerplate(API_SIGNON)) == NULL) {
		return HA_FAIL;
	}
	snprintf(cuid, sizeof(cuid)-1, "%ld",  (long)geteuid());
	/* Add our UID to the message */
	if (ha_msg_add(request, F_UID, cuid) != HA_OK) {
		ha_api_log(LOG_ERR, "hb_api_signon: cannot add F_UID field");
		ZAPMSG(request);
		return HA_FAIL;
	}
	snprintf(cgid, sizeof(cgid)-1, "%ld",  (long)getegid());
	/* Add our GID to the message */
	if (ha_msg_add(request, F_GID, cgid) != HA_OK) {
		ha_api_log(LOG_ERR, "hb_api_signon: cannot add F_GID field");
		ZAPMSG(request);
		return HA_FAIL;
	}
	wchanattrs = g_hash_table_new(g_str_hash, g_str_equal);
        g_hash_table_insert(wchanattrs, path, regpath);

	/* Connect to the heartbeat API server */

	if ((pi->chan = ipc_channel_constructor(IPC_ANYTYPE, wchanattrs))
	==	NULL) {
		ha_api_log(LOG_ERR, "hb_api_signon: Can't connect"
		" to heartbeat");
		ZAPMSG(request);
		return HA_FAIL;
	}

	pi->chan->should_send_blocking = TRUE;

        if (pi->chan->ops->initiate_connection(pi->chan) != IPC_OK) {
		ha_api_log(LOG_ERR, "hb_api_signon: Can't initiate"
		" connection  to heartbeat");
		ZAPMSG(request);
                return HA_FAIL;
        }


	/* Send the registration request message */
	if (msg2ipcchan(request, pi->chan) != HA_OK) {
		pi->chan->ops->destroy(pi->chan);
		pi->chan = NULL;
		ha_api_perror("can't send message to IPC");
		ZAPMSG(request);
		return HA_FAIL;
	}
	

	ZAPMSG(request);
	pi->chan->ops->waitout(pi->chan);

	/* Read the reply... */
	if ((reply=read_api_msg(pi)) == NULL) {
		return HA_FAIL;
	}

	/* Get the return code */
	if ((result = ha_msg_value(reply, F_APIRESULT)) != NULL
	&&	strcmp(result, API_OK) == 0) {
		rc = HA_OK;
		pi->SignedOn = 1;

		if ((tmpstr = ha_msg_value(reply, F_DEADTIME)) == NULL
		||	sscanf(tmpstr, "%lx", (unsigned long*)&(pi->deadtime_ms)) != 1) {
			ha_api_log(LOG_ERR
			,	"hb_api_signon: Can't get deadtime ");
			ZAPMSG(reply);
			return HA_FAIL;
		}
		if ((tmpstr = ha_msg_value(reply, F_KEEPALIVE)) == NULL
		||	sscanf(tmpstr, "%lx", (unsigned long*)&(pi->keepalive_ms)) != 1) {
			ha_api_log(LOG_ERR
			,	"hb_api_signon: Can't get keepalive time ");
			ZAPMSG(reply);
			return HA_FAIL;
		}
		if ((tmpstr = ha_msg_value(reply, F_NODENAME)) == NULL
		||	strlen(tmpstr) >= sizeof(OurNode)) {
			ha_api_log(LOG_ERR
			,	"hb_api_signon: Can't get local node name");
			ZAPMSG(reply);
			return HA_FAIL;
		}else{
			strncpy(OurNode, tmpstr, sizeof(OurNode)-1);
			OurNode[sizeof(OurNode)-1] = EOS;
		}
		/* Sometimes they don't use syslog logging... */
		tmpstr = ha_msg_value(reply, F_LOGFACILITY);
		if (tmpstr == NULL
		||	sscanf(tmpstr, "%d", &(pi->logfacility)) != 1) {
			pi->logfacility = -1;
		}
	}else{
		rc = HA_FAIL;
	}
	ZAPMSG(reply);

	return rc;
}

/*
 * Sign off (disconnect) as a heartbeat client process.
 */
static int
hb_api_signoff(struct ll_cluster* cinfo)
{
	struct ha_msg*	request;
	llc_private_t* pi;

	if (!ISOURS(cinfo)) {
		ha_api_log(LOG_ERR, "hb_api_signoff: bad cinfo");
		return HA_FAIL;
	}
	pi = (llc_private_t*)cinfo->ll_cluster_private;

	if (!pi->SignedOn) {
		/* Already signed off... No value in returning an error... */
		return HA_OK;
	}

	if ((request = hb_api_boilerplate(API_SIGNOFF)) == NULL) {
		ha_api_log(LOG_ERR, "hb_api_signoff: can't create msg");
		return HA_FAIL;
	}
	
	/* Send the message */
	if (msg2ipcchan(request, pi->chan) != HA_OK) {
		ZAPMSG(request);
		ha_api_perror("can't send message to IPC");
		return HA_FAIL;
	}
	pi->chan->ops->waitout(pi->chan);
	ZAPMSG(request);
	OurClientID = NULL;
	pi->chan->ops->destroy(pi->chan);
	pi->SignedOn = 0;
	zap_order_seq(pi);
	zap_order_queue(pi);

	return HA_OK;
}
/*
 * delete:  destroy the heartbeat API object
 */
static int
hb_api_delete(struct ll_cluster* ci)
{
	llc_private_t* pi;
	if (!ISOURS(ci)) {
		ha_api_log(LOG_ERR, "hb_api_delete: bad cinfo");
		return HA_FAIL;
	}
	pi = (llc_private_t*)ci->ll_cluster_private;

	/* Sign off */
	hb_api_signoff(ci);

	/* Free up interface and node lists */
	zap_iflist(pi);
	zap_nodelist(pi);

	/* What about our message queue? */

	/* Free up the private information */
	memset(pi, 0, sizeof(*pi));
	ha_free(pi);

	/* Free up the generic (llc) information */
	memset(ci, 0, sizeof(*ci));
	ha_free(ci);
	return HA_OK;
}

/*
 * Set message filter mode.
 */
static int
hb_api_setfilter(struct ll_cluster* ci, unsigned fmask)
{
	struct ha_msg*	request;
	struct ha_msg*	reply;
	int		rc;
	const char *	result;
	char		filtermask[32];
	llc_private_t* pi;

	if (!ISOURS(ci)) {
		ha_api_log(LOG_ERR, "hb_api_setfilter: bad cinfo");
		return HA_FAIL;
	}
	pi = (llc_private_t*)ci->ll_cluster_private;

	if (!pi->SignedOn) {
		ha_api_log(LOG_ERR, "not signed on");
		return HA_FAIL;
	}

	if ((request = hb_api_boilerplate(API_SETFILTER)) == NULL) {
		ha_api_log(LOG_ERR, "hb_api_setfilter: can't create msg");
		return HA_FAIL;
	}

	/* Format the filtermask information in hex */
	snprintf(filtermask, sizeof(filtermask), "%x", fmask);
	if (ha_msg_add(request, F_FILTERMASK, filtermask) != HA_OK) {
		ha_api_log(LOG_ERR, "hb_api_setfilter: cannot add field/2");
		ZAPMSG(request);
		return HA_FAIL;
	}
	
	/* Send the message */
	if (msg2ipcchan(request, pi->chan) != HA_OK) {
		ZAPMSG(request);
		ha_api_perror("can't send message to IPC");
		return HA_FAIL;
	}
	ZAPMSG(request);


	/* Read reply... */
	if ((reply=read_api_msg(pi)) == NULL) {
		return HA_FAIL;
	}
	if ((result = ha_msg_value(reply, F_APIRESULT)) != NULL
	&&	strcmp(result, API_OK) == 0) {
		rc = HA_OK;
	}else{
		rc = HA_FAIL;
	}
	ZAPMSG(reply);

	return rc;
}

/*
 * Set signal for message notification.
 * This is not believed to be a security hole :-)
 */
static int
hb_api_setsignal(ll_cluster_t* lcl, int nsig)
{
	struct ha_msg*	request;