lrmd.c

在LINUX下实现HA的源代码

/*
 * Local Resource Manager Daemon
 *
 * Author: Huang Zhen <[e-mail]>
 * Partly contributed by Andrew Beekhof <[e-mail]> 
 * Copyright (c) 2004 International Business Machines
 *
 * 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
 *
 */
#include <portability.h>
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <stdarg.h>
#include <string.h>
#include <errno.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <dirent.h>

#include <glib.h>
#include <heartbeat.h>
#include <pils/plugin.h>
#include <pils/generic.h>
#include <clplumbing/cl_log.h>
#include <clplumbing/ipc.h>
#include <clplumbing/GSource.h>
#include <clplumbing/lsb_exitcodes.h>
#include <clplumbing/cl_signal.h>

#include <ha_msg.h>
#include <lrm/lrm_api.h>
#include <lrm/lrm_msg.h>
#include <lrm/raexec.h>

#define	MAX_PID_LEN 256

#define OPTARGS		"kh"
#define PID_FILE 	HA_VARRUNDIR"/lrmd.pid"
#define DAEMON_LOG   	"/var/log/lrmd.log"
#define DAEMON_DEBUG 	"/var/log/lrmd.debug"
#define PLUGIN_DIR	"/usr/lib/heartbeat/plugins"
#define RA_PLUGIN_DIR	"/usr/lib/heartbeat/plugins/RAExec"

typedef struct  
{
	char*		app_name;	
	pid_t		pid;
	gid_t		gid;
	uid_t		uid;

	IPC_Channel*	ch_cmd;
	IPC_Channel*	ch_cbk;

	GCHSource*	g_src;
}lrmd_client_t;

typedef struct lrmd_rsc lrmd_rsc_t;
typedef struct lrmd_mon lrmd_mon_t;
typedef struct lrmd_op	lrmd_op_t;

struct lrmd_op
{
	lrmd_rsc_t*	rsc;
	lrmd_client_t*	client;
	char*		app_name;
	int		call_id;
	int		callback_id;
	guint 		timeout_tag;
	lrmd_mon_t*	mon;
	struct ha_msg*	msg;
};

struct lrmd_mon
{
	mon_mode_t	mode;
	lrmd_rsc_t*	rsc;
	lrmd_client_t*	client;
	char*		app_name;
	int		call_id;
	int		interval;
	int		target;
	guint		timeout_tag;
	int		pending_op;
	gboolean	is_deleted;
	int		last_status;
	struct ha_msg*	msg;
};

struct lrmd_rsc
{
	rsc_id_t	id;
	char*		name;
	char*		type;
	GHashTable* 	params;

	GList*		op_list;
	GList*		mon_list;
	lrmd_op_t*	last_op;
};

/* glib loop call back functions */ 
static gboolean on_connect_cmd(IPC_Channel* ch_cmd, gpointer user_data);
static gboolean on_connect_cbk(IPC_Channel* ch_cbk, gpointer user_data);
static gboolean on_recieve_cmd(IPC_Channel* ch_cmd, gpointer user_data);
static gboolean on_timeout_monitor(gpointer data);
static gboolean on_timeout_op_done(gpointer data);
/* static gboolean on_idle (gpointer data); */ 
gboolean on_idle (gpointer data);
static void on_remove_client(gpointer user_data);

/* message handlers */ 
static int on_msg_unregister(lrmd_client_t* client, struct ha_msg* msg);
static int on_msg_register(lrmd_client_t* client, struct ha_msg* msg);
static int on_msg_get_ra_types(lrmd_client_t* client, struct ha_msg* msg);
static int on_msg_add_rsc(lrmd_client_t* client, struct ha_msg* msg);
static int on_msg_get_rsc(lrmd_client_t* client, struct ha_msg* msg);
static int on_msg_get_all(lrmd_client_t* client, struct ha_msg* msg);
static int on_msg_del_rsc(lrmd_client_t* client, struct ha_msg* msg);
static int on_msg_perform_op(lrmd_client_t* client, struct ha_msg* msg);
static int on_msg_set_monitor(lrmd_client_t* client, struct ha_msg* msg);
static int on_msg_get_state(lrmd_client_t* client, struct ha_msg* msg);

/* functions wrap the call to ra plugins */ 
static int check_ra_rc(struct RAExecOps * RAExec, int callback_id, int* rc,
					   char** data);
static int perform_ra_op(struct RAExecOps * RAExec, lrmd_op_t* op);

/* Utility functions */ 
static int flush_op(lrmd_op_t* op);
static int perform_op(lrmd_rsc_t* rsc);
static int check_op(lrmd_op_t* op);
static int op_done(lrmd_op_t* op);
static void free_mon(lrmd_mon_t* mon);
static void free_rsc(lrmd_rsc_t* rsc);
static int send_rc_msg ( IPC_Channel* ch, int rc);
static lrmd_client_t* lookup_client (pid_t pid);
static lrmd_rsc_t* lookup_rsc (rsc_id_t rid);
static struct ha_msg* op_to_msg(lrmd_op_t* op);

static void lrmd_log (int priority, int level, const char* fmt);


/* msg dispatch table */ 
typedef int (*msg_handler)(lrmd_client_t* client, struct ha_msg* msg);
struct msg_map
{
	const char* 	msg_type;
	gboolean	need_return_rc;
	msg_handler	handler;
};

struct msg_map msg_maps[] = {
	{UNREGISTER,	TRUE,	on_msg_unregister},
	{REGISTER,	TRUE,	on_msg_register},
	{GETRATYPES,	FALSE,	on_msg_get_ra_types},
	{ADDRSC,	TRUE,	on_msg_add_rsc},
	{GETRSC,	FALSE,	on_msg_get_rsc},
	{GETALLRCSES,	FALSE,	on_msg_get_all},
	{DELRSC,	TRUE,	on_msg_del_rsc},
	{PERFORMOP,	TRUE,	on_msg_perform_op},
	{FLUSHOPS,	TRUE,	on_msg_perform_op},
	{SETMONITOR,	TRUE,	on_msg_set_monitor},
	{GETRSCSTATE,	FALSE,	on_msg_get_state},
};

GMainLoop* mainloop 		= NULL;
GList* client_list 		= NULL;
GList* rsc_list 		= NULL;
static int call_id 		= 1;
const char* lrm_system_name 	= "lrmd";
GHashTable * RAExecFuncs 	= NULL;
GList* ra_list			= NULL;

/*
 * Daemon functions
 *
 * copy from the code of Andrew Beekhof <[e-mail]>
 */
void usage(const char* cmd, int exit_status);
int init_start(void);
void lrmd_shutdown(int nsig);
int init_stop(const char *pid_file);
long get_running_pid(const char *pid_file, gboolean* anypidfile);
void register_pid(const char *pid_file, gboolean do_fork,
			void (*shutdown)(int nsig));

int
main(int argc, char ** argv)
{
	int req_stop = FALSE;
	int argerr = 0;
	int flag;

	cl_log_set_entity(lrm_system_name);
	cl_log_enable_stderr(TRUE);
	cl_log_set_facility(LOG_USER);

	while ((flag = getopt(argc, argv, OPTARGS)) != EOF) {
		switch(flag) {
			case 'k':		/* Stop (kill) */
				req_stop = TRUE;
				break;
			case 'h':		/* Help message */
				usage(lrm_system_name, LSB_EXIT_OK);
				break;
			default:
				++argerr;
				break;
		}
	}

	if (optind > argc) {
		++argerr;
	}

	if (argerr) {
		usage(lrm_system_name,LSB_EXIT_OK);
	}

	if (req_stop) {
		return init_stop(PID_FILE);
	}
	return init_start();
}

int
init_stop(const char *pid_file)
{
	long	pid;
	int	rc = LSB_EXIT_OK;

	if (pid_file == NULL) {
		cl_log(LOG_ERR, "No pid file specified to kill process");
		return LSB_EXIT_GENERIC;
	}

	pid =	get_running_pid(pid_file, NULL);

	if (pid > 0) {
		if (CL_KILL((pid_t)pid, SIGTERM) < 0) {
			rc = (errno == EPERM
			      ?	LSB_EXIT_EPERM : LSB_EXIT_GENERIC);
			fprintf(stderr, "Cannot kill pid %ld\n", pid);
		}else{
			cl_log(LOG_INFO,
			       "Signal sent to pid=%ld,"
			       " waiting for process to exit",
			       pid);

			while (CL_PID_EXISTS(pid)) {
				sleep(1);
			}
		}
	}
 	return rc;
}

long
get_running_pid(const char *pid_file, gboolean* anypidfile)
{
	long    pid;
	FILE *  lockfd;
	lockfd = fopen(pid_file, "r");

	if (anypidfile) {
		*anypidfile = (lockfd != NULL);
	}

	if (lockfd != NULL && fscanf(lockfd, "%ld", &pid) == 1 && pid > 0) {
		if (CL_PID_EXISTS((pid_t)pid)) {
			fclose(lockfd);
			return(pid);
		}
	}
	if (lockfd != NULL) {
		fclose(lockfd);
	}
	return(-1L);
}

void
usage(const char* cmd, int exit_status)
{
	FILE* stream;

	stream = exit_status ? stderr : stdout;

	fprintf(stream, "usage: %s [-kh]\n", cmd);
	fflush(stream);
	
	exit(exit_status);
}

void
lrmd_shutdown(int nsig)
{
	static int shuttingdown = 0;
	CL_SIGNAL(nsig, lrmd_shutdown);

	if (!shuttingdown) {
		shuttingdown = 1;
	}
	if (mainloop != NULL && g_main_is_running(mainloop)) {
		g_main_quit(mainloop);
	}else {
		exit(LSB_EXIT_OK);
	}
}

void
register_pid(const char *pid_file,gboolean do_fork,void (*shutdown)(int nsig))
{
	int	j;
	long	pid;
	FILE *	lockfd;

	if (do_fork) {
		pid = fork();

		if (pid < 0) {
			cl_log(LOG_CRIT, "cannot start daemon");
			exit(LSB_EXIT_GENERIC);
		}else if (pid > 0) {
			exit(LSB_EXIT_OK);
		}
	}

	lockfd = fopen(pid_file, "w");
	if (lockfd == NULL) {
		cl_log(LOG_CRIT, "cannot create pid file: %s", pid_file);
		exit(LSB_EXIT_GENERIC);
	}else{
		pid = getpid();
		fprintf(lockfd, "%ld\n", pid);
		fclose(lockfd);
	}

	umask(022);

	for (j=0; j < 3; ++j) {
		close(j);
		(void)open("/dev/null", j == 0 ? O_RDONLY : O_RDONLY);
	}
	CL_IGNORE_SIG(SIGINT);
	CL_IGNORE_SIG(SIGHUP);
	CL_SIGNAL(SIGTERM, shutdown);
}

/* main loop of the daemon*/
int
init_start ()
{
	long pid;
	/* load RA plugins */ 
	PILPluginUniv * PluginLoadingSystem = NULL;
	PILGenericIfMgmtRqst RegisterRqsts[]= {
		{"RAExec", &RAExecFuncs, NULL, NULL, NULL},
		{ NULL, NULL, NULL, NULL, NULL} };
	DIR* dir = NULL;
	struct dirent* subdir;
	char* ra_name;
	char* dot = NULL;
	/*
	 *create the waiting connections
	 *one for register the client,
	 *the other is for create the callback channel
	 */

	IPC_WaitConnection* conn_cmd = NULL;
	IPC_WaitConnection* conn_cbk = NULL;

	GHashTable* conn_cmd_attrs;
	GHashTable* conn_cbk_attrs;

	char path[] = IPC_PATH_ATTR;
	char cmd_path[] = LRM_CMDPATH;
	char cbk_path[] = LRM_CALLBACKPATH;



	if ((pid = get_running_pid(PID_FILE, NULL)) > 0) {
		cl_log(LOG_CRIT, "already running: [pid %ld].", pid);
		exit(LSB_EXIT_OK);
	}

	register_pid(PID_FILE, TRUE, FALSE);

	cl_log_set_logfile(DAEMON_LOG);
	cl_log_set_debugfile(DAEMON_DEBUG);


	PluginLoadingSystem = NewPILPluginUniv (PLUGIN_DIR);
	PILLoadPlugin(PluginLoadingSystem , "InterfaceMgr", "generic" ,
				  &RegisterRqsts);

	dir = opendir(RA_PLUGIN_DIR);
	if (NULL == dir) {
		lrmd_log(LOG_ERR, -1, "main: can not open RA plugin dir.");
		return 1;
	}

	while ( NULL != (subdir = readdir(dir))) {
		/* skip . and .. */ 
		if ( '.' == subdir->d_name[0]) {
			continue;
		}
		/* skip the other type files */ 
		if (NULL == strstr(subdir->d_name, ".so")) {
			continue;
		}
		/* remove the ".so" */ 
		dot = strchr(subdir->d_name,'.');
		if (NULL != dot) {
			int len = (int)(dot - subdir->d_name);
			ra_name = strndup(subdir->d_name,len);
		}
		else {
			ra_name = g_strdup(subdir->d_name);
		}
		PILLoadPlugin(PluginLoadingSystem , "RAExec", ra_name, NULL);
		ra_list = g_list_append(ra_list,ra_name);
	}

	lrmd_log(LOG_INFO, 1, "main: start.");

	/*Create a waiting connection to accept command connect from client*/
	conn_cmd_attrs = g_hash_table_new(g_str_hash, g_str_equal);
	g_hash_table_insert(conn_cmd_attrs, path, cmd_path);
	conn_cmd = ipc_wait_conn_constructor(IPC_ANYTYPE, conn_cmd_attrs);
	if (NULL == conn_cmd) {
		lrmd_log(LOG_ERR, -1,
			"main: can not create wait connection for command.");
		return 1;
	}

	/*Create a source to handle new connect rquests for command*/
	G_main_add_IPC_WaitConnection( G_PRIORITY_HIGH, conn_cmd, NULL, FALSE,
				   on_connect_cmd, conn_cmd, NULL);

	/*
	 *Create a waiting connection to accept the callback connect from client
	*/
	conn_cbk_attrs = g_hash_table_new(g_str_hash, g_str_equal);
	g_hash_table_insert(conn_cbk_attrs, path, cbk_path);
	conn_cbk = ipc_wait_conn_constructor( IPC_ANYTYPE, conn_cbk_attrs);
	if (NULL == conn_cbk) {
		lrmd_log(LOG_ERR, -1,
			"main: can not create wait connection for callback.");
		return 1;
	}

	/*Create a source to handle new connect rquests for callback*/
	G_main_add_IPC_WaitConnection( G_PRIORITY_HIGH, conn_cbk, NULL, FALSE,
	                               on_connect_cbk, conn_cbk, NULL);
	g_idle_add(on_idle,NULL);
/* 	g_timeout_add(5, on_idle,NULL); */ 

	/*Create the mainloop and run it*/
	mainloop = g_main_new(FALSE);
	lrmd_log(LOG_INFO, 0, "main: run the loop...");
	g_main_run(mainloop);

	conn_cmd->ops->destroy(conn_cmd);
	conn_cmd = NULL;

	conn_cbk->ops->destroy(conn_cbk);
	conn_cbk = NULL;

	if (unlink(PID_FILE) == 0) {
		cl_log(LOG_INFO, "[%s] stopped", lrm_system_name);
	}

	lrmd_log(LOG_INFO, -1, "main: end.");

	return 0;
}

/*
 *GLoop Message Handlers
 */
gboolean
on_connect_cmd (IPC_Channel* ch, gpointer user_data)
{
	lrmd_client_t* client = NULL;

	lrmd_log(LOG_INFO, 1, "on_connect_cmd: start.");
	/* check paremeters */ 
	if (NULL == ch) {
		lrmd_log(LOG_ERR, -1, "on_connect_cmd: channel is null");
		return TRUE;
	}
	/* create new client */ 
	/* the register will be finished in on_msg_register */ 
	client = g_new(lrmd_client_t, 1);
	client->app_name = NULL;
	client->ch_cmd = ch;
	client->g_src = G_main_add_IPC_Channel(G_PRIORITY_DEFAULT,
				ch, FALSE, on_recieve_cmd, (gpointer)client,
				on_remove_client);

	lrmd_log(LOG_INFO, -1, "on_connect_cmd: end.");

	return TRUE;
}

gboolean
on_connect_cbk (IPC_Channel* ch, gpointer user_data)
{
	/*client connect for create the second channel for call back*/
	pid_t pid;
	struct ha_msg* msg = NULL;
	const char* type = NULL;
	lrmd_client_t* client = NULL;

	lrmd_log(LOG_INFO, 1, "on_connect_cbk: start.");
	if (NULL == ch) {
		lrmd_log(LOG_INFO, -1, "on_connect_cbk: channel is null");
		return TRUE;
	}

	/*get the message */
	msg = msgfromIPC_noauth(ch);
	if (NULL == msg) {
		lrmd_log(LOG_ERR, -1, "on_connect_cbk: can not recieve msg");
		return TRUE;
	}

	/*check if it is a register message*/
	type = ha_msg_value(msg, F_LRM_TYPE);
	if (0 != strncmp(type, REGISTER, strlen(REGISTER))) {
		lrmd_log(LOG_ERR, -1, "on_connect_cbk: msg is not register");
		send_rc_msg(ch, HA_FAIL);
		return TRUE;
	}

	/*get the pid of client */
	if (HA_OK != ha_msg_value_int(msg, F_LRM_PID, &pid)) {
		lrmd_log(LOG_ERR, -1, "on_connect_cbk: can not get pid");
		send_rc_msg(ch, HA_FAIL);
		return TRUE;
	}

	/*get the client in the client list*/