examples/health_check.c - oransc/ric-plt/lib/rmr - Gitiles

 // :vim ts=4 sw=4 noet:
 /*
 ==================================================================================
 	Copyright (c) 2019 Nokia
 	Copyright (c) 2018-2019 AT&T Intellectual Property.

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

 	   http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
 ==================================================================================
 */

 /*
 	Mnemonic:	health_check.c
 	Abstract:	This is a simple programme which sends a 'health check' message to
 				an application and waits for a response. By default, the application
 				is assumed to be running on the local host, and listening on 4560,
 				but both host and port can be configured as needed. Connection is
 				made via a wormhole, so there is no need for a routing table.

 				The application being checked is expected to recognise the health
 				check message type, and to return the message using the RMR return
 				to sender function after changing the message type to  "health response,"
 				and leaving the remainder of the payload _unchanged_.

 				A timestamp is placed into the outbound payload, and the round trip
 				latency is reported (the reason the pinged application should not modify
 				the payload.


 				Command line options and parameters:
 					[-h host:port]		target
 					[-n num-msgs]		total number to send
 					[-t seconds]		max timeout per message

 				Route table:  While we don't need a route table to do wormhole sends we
 				do need for RMR to initialise an empty one. To avoid having to have a
 				dummy table on disk somewhere, we'll create one and "point" RMR at it.

 	Date:		9 August 2019
 	Author:		E. Scott Daniels
 */

 #include <unistd.h>
 #include <errno.h>
 #include <string.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <sys/epoll.h>
 #include <time.h>
 #include <fcntl.h>

 #include <rmr/rmr.h>
 // include message types header

 #ifndef HEALTH_CHECK
 #define HEALTH_CHECK	100		// message types
 #define HEALTH_RESP		101
 #endif

 /*
 	Our message payload.
 */
 typedef struct mpl {
 	char	msg[512];				// message for human consumption
 	struct	timespec out_ts;		// time this payload was sent
 } mpl_t;

 // ---------------------------------------------------------------------------
 /*
 	Very simple checksum over a buffer.
 */
 static int sum( unsigned char* buf, int len ) {
 	int sum = 0;
 	int	i = 0;
 	unsigned char*	last;

 	last = buf + len;
 	while( buf < last ) {
 		sum += *(buf++) + i++;
 	}

 	return sum % 255;
 }

 /*
 	Compute the elapsed time between ts1 and ts2.
 	Returns mu-seconds.
 */
 static int elapsed( struct timespec* start_ts, struct timespec* end_ts ) {
 	long long start;
 	long long end;
 	int bin;

 	start = ( start_ts->tv_sec * 1000000000) + start_ts->tv_nsec;
 	end = ( end_ts->tv_sec * 1000000000) + end_ts->tv_nsec;

 	bin = (end - start) / 1000;			// to mu-sec
 	//bin = (end - start);

 	return bin;
 }

 /*
 	See if my id string is in the buffer immediately after the first >.
 	Return 1 if so, 0 if not.
 */
 static int vet_received( char* me, char* buf ) {
 	char*	ch;

 	if( (ch = strchr( buf, '>' )) == NULL ) {
 		return 0;
 	}

 	return strcmp( me, ch+1 ) == 0;
 }

 /*
 	Create an empty route table and set an environment var for RMR to find.
 	This must be called before initialising RMR.
 */
 static void mk_rt( ) {
 	int 	fd;
 	char	fnb[128];
 	char*	contents = "newrt|start\nnewrt|end\n";

 	snprintf( fnb, sizeof( fnb ), "/tmp/health_check.rt" );
 	fd = open( fnb, O_CREAT | O_WRONLY, 0664 );
 	if( fd < 0 ) {
 		fprintf( stderr, "[FAIL] could not create dummy route table: %s %s\n", fnb, strerror( errno ) );
 		return;
 	}

 	write( fd, contents, strlen( contents ) );
 	if( (close( fd ) < 0 ) ) {
 		fprintf( stderr, "[FAIL] couldn't close dummy route table: %s: %s\n", fnb, strerror( errno ) );
 		return;
 	}

 	setenv( "RMR_SEED_RT", fnb, 0 );		// set it, but don't overwrite it
 }

 int main( int argc, char** argv ) {
 	void* mrc;      						// msg router context
 	rmr_mbuf_t*		mbuf;					// message buffer
 	mpl_t*	payload;						// the payload in a message
 	int		ai = 1;							// arg index
 	long	timeout;
 	long	max_timeout = 5;				// -t to overrride
 	char*	target = "localhost:4560";		// address of target to ping
 	char*	listen_port;					// the port we open for "backhaul" connections (random)
 	char*	tok;							// pointer at token in a buffer
 	int		i;
 	char	wbuf[1024];
 	char	me[128];						// who I am to vet rts was actually from me
 	int		rand_port = 0;					// -r sets and causes us to generate a random listen port
 	int		whid;							// id of wormhole
 	int		num2send = 1;					// number of messages to send
 	int     ep_fd = -1;						// epoll's file des (given to epoll_wait)
 	int		rcv_fd;    						// file des that NNG tickles -- give this to epoll to listen on
 	int		nready;							// number of events ready for receive
 	int		count = 0;
 	int		errors = 0;
 	int		cksum;							// computed simple checksum
 	struct	timespec in_ts;					// time we got response
 	struct	epoll_event events[1];			// list of events to give to epoll
 	struct	epoll_event epe;				// event definition for event to listen to

 	// ---- simple arg parsing ------
 	while( ai < argc ) {
 		if( *argv[ai] == '-' ) {
 			switch( argv[ai][1] ) {
 				case 'h':					// host port
 					ai++;
 					target = strdup( argv[ai] );
 					break;

 				case 'n':					// num to send
 					ai++;
 					num2send = atoi( argv[ai] );
 					break;

 				case 'r':					// generate random listen port
 					rand_port = 1;
 					;;

 				case 't':					// timeout
 					ai++;
 					max_timeout = atoi( argv[ai] );
 					break;

 				default:
 					fprintf( stderr, "[FAIL] unrecognised option: %s\n", argv[ai] );
 					exit( 1 );
 			}

 			ai++;
 		} else {
 			break;		// not an option, leave with a1 @ first positional parm
 		}
 	}

 	if( rand_port ) {
 		srand( time( NULL ) );
 		snprintf( wbuf, sizeof( wbuf ), "%d", 43000 + (rand() % 1000) );			// random listen port
 		listen_port = strdup( wbuf );
 	} else {
 		listen_port = "43086";
 	}


 	mk_rt();				// create a dummy route table so we don't have errors/hang

 	fprintf( stderr, "[INFO] listen port: %s; sending %d messages\n", listen_port, num2send );

 	if( (mrc = rmr_init( listen_port, 1400, 0 )) == NULL ) {		// start without route table listener thread
 		fprintf( stderr, "[FAIL] unable to initialise RMr\n" );
 		exit( 1 );
 	}
 	fprintf( stderr, "[INFO] RMR initialised\n" );

 	if( (rcv_fd = rmr_get_rcvfd( mrc )) < 0 ) {			// if we can't get an epoll FD, then we can't timeout; abort
 		fprintf( stderr, "[FAIL] unable to get an epoll FD\n" );
 		exit( 1 );
 	}

 	if( (ep_fd = epoll_create1( 0 )) < 0 ) {
 		fprintf( stderr, "[FAIL] unable to create epoll fd: %d\n", errno );
 		exit( 1 );
 	}
 	epe.events = EPOLLIN;
 	epe.data.fd = rcv_fd;

 	if( epoll_ctl( ep_fd, EPOLL_CTL_ADD, rcv_fd, &epe ) != 0 )  {
 		fprintf( stderr, "[FAIL] epoll_ctl status not 0 : %s\n", strerror( errno ) );
 		exit( 1 );
 	}

 	while( ! rmr_ready( mrc ) ) {
 		sleep( 1 );
 	}

 	mbuf = rmr_alloc_msg( mrc, sizeof( *payload ) + 100 );		// send buffer with a bit of padding

 	fprintf( stderr, "[INFO] starting session with %s, starting to send\n", target );
 	whid = rmr_wh_open( mrc, target );								// open a wormhole directly to the target
 	if( whid < 0 ) {
 		fprintf( stderr, "[FAIL] unable to connect to %s\n", target );
 		exit( 2 );
 	}

 	fprintf( stderr, "[INFO] connected to %s, starting to send\n", target );
 	rmr_set_stimeout( mrc, 3 );									// we let rmr retry failures for up to 3 "rounds"

 	gethostname( wbuf, sizeof( wbuf ) );
 	snprintf( me, sizeof( me ), "%s-%d", wbuf, getpid( ) );

 	errors = 0;
 	while( count < num2send ) {								// we send n messages after the first message is successful
 		if( !mbuf ) {
 			fprintf( stderr, "[FAIL] mbuf is nil?\n" );
 			exit( 1 );
 		}

 		payload = (mpl_t *) mbuf->payload;

 		snprintf( wbuf, sizeof( payload->msg ), "%s count=%d %d", me, count, rand() );
 		snprintf( mbuf->payload, 1024, "%d|%s", sum( wbuf , strlen( wbuf ) ), wbuf );

 		mbuf->mtype = HEALTH_CHECK;
 		mbuf->sub_id = -1;
 		mbuf->len =  sizeof( *payload );
 		mbuf->state = 0;

 		clock_gettime( CLOCK_REALTIME, &payload->out_ts );		// mark time out
 		mbuf = rmr_wh_send_msg( mrc, whid, mbuf );

 		if( mbuf->state == RMR_OK ) {							// good send, wait for response
 			nready = epoll_wait( ep_fd, events, 1, max_timeout * 1000 );
 			if( nready > 0 ) {
 				clock_gettime( CLOCK_REALTIME, &in_ts );		// mark response received time

 				mbuf = rmr_rcv_msg( mrc, mbuf );
 				payload = (mpl_t *) mbuf->payload;
 				tok = strchr( payload->msg, '|' );				// find end of chksum
 				if( tok ) {
 					tok++;
 					cksum = sum( tok, strlen( tok ) );
 					if( cksum != atoi( payload->msg ) ) {
 						fprintf( stderr, "[WRN] response to msg %d received, cksum mismatch; expected %d, got %d\n",
 							count+1, atoi( payload->msg ), cksum );
 					} else {
 						fprintf( stderr, "[INFO] response to msg %d received, %d mu-sec\n",  count+1, elapsed( &payload->out_ts, &in_ts ) );
 					}
 				}
 			} else {
 				fprintf( stderr, "[ERR] timeout waiting for response to message %d\n", count+1 );
 				errors++;
 			}
 		} else {
 			fprintf( stderr, "[ERR] send failed: %d\n", mbuf->state );
 		}

 		count++;
 		sleep( 1 );
 	}

 	rmr_wh_close( mrc, whid );

 	return errors = 0;
 }
	// :vim ts=4 sw=4 noet:
	/*
	==================================================================================
	Copyright (c) 2019 Nokia
	Copyright (c) 2018-2019 AT&T Intellectual Property.

	Licensed under the Apache License, Version 2.0 (the "License");
	you may not use this file except in compliance with the License.
	You may obtain a copy of the License at

	http://www.apache.org/licenses/LICENSE-2.0

	Unless required by applicable law or agreed to in writing, software
	distributed under the License is distributed on an "AS IS" BASIS,
	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	See the License for the specific language governing permissions and
	limitations under the License.
	==================================================================================
	*/

	/*
	Mnemonic: health_check.c
	Abstract: This is a simple programme which sends a 'health check' message to
	an application and waits for a response. By default, the application
	is assumed to be running on the local host, and listening on 4560,
	but both host and port can be configured as needed. Connection is
	made via a wormhole, so there is no need for a routing table.

	The application being checked is expected to recognise the health
	check message type, and to return the message using the RMR return
	to sender function after changing the message type to "health response,"
	and leaving the remainder of the payload _unchanged_.

	A timestamp is placed into the outbound payload, and the round trip
	latency is reported (the reason the pinged application should not modify
	the payload.


	Command line options and parameters:
	[-h host:port] target
	[-n num-msgs] total number to send
	[-t seconds] max timeout per message

	Route table: While we don't need a route table to do wormhole sends we
	do need for RMR to initialise an empty one. To avoid having to have a
	dummy table on disk somewhere, we'll create one and "point" RMR at it.

	Date: 9 August 2019
	Author: E. Scott Daniels
	*/

	#include <unistd.h>
	#include <errno.h>
	#include <string.h>
	#include <stdio.h>
	#include <stdlib.h>
	#include <sys/epoll.h>
	#include <time.h>
	#include <fcntl.h>

	#include <rmr/rmr.h>
	// include message types header

	#ifndef HEALTH_CHECK
	#define HEALTH_CHECK 100 // message types
	#define HEALTH_RESP 101
	#endif

	/*
	Our message payload.
	*/
	typedef struct mpl {
	char msg[512]; // message for human consumption
	struct timespec out_ts; // time this payload was sent
	} mpl_t;

	// ---------------------------------------------------------------------------
	/*
	Very simple checksum over a buffer.
	*/
	static int sum( unsigned char* buf, int len ) {
	int sum = 0;
	int i = 0;
	unsigned char* last;

	last = buf + len;
	while( buf < last ) {
	sum += *(buf++) + i++;
	}

	return sum % 255;
	}

	/*
	Compute the elapsed time between ts1 and ts2.
	Returns mu-seconds.
	*/
	static int elapsed( struct timespec* start_ts, struct timespec* end_ts ) {
	long long start;
	long long end;
	int bin;

	start = ( start_ts->tv_sec * 1000000000) + start_ts->tv_nsec;
	end = ( end_ts->tv_sec * 1000000000) + end_ts->tv_nsec;

	bin = (end - start) / 1000; // to mu-sec
	//bin = (end - start);

	return bin;
	}

	/*
	See if my id string is in the buffer immediately after the first >.
	Return 1 if so, 0 if not.
	*/
	static int vet_received( char* me, char* buf ) {
	char* ch;

	if( (ch = strchr( buf, '>' )) == NULL ) {
	return 0;
	}

	return strcmp( me, ch+1 ) == 0;
	}

	/*
	Create an empty route table and set an environment var for RMR to find.
	This must be called before initialising RMR.
	*/
	static void mk_rt( ) {
	int fd;
	char fnb[128];
	char* contents = "newrt\|start\nnewrt\|end\n";

	snprintf( fnb, sizeof( fnb ), "/tmp/health_check.rt" );
	fd = open( fnb, O_CREAT \| O_WRONLY, 0664 );
	if( fd < 0 ) {
	fprintf( stderr, "[FAIL] could not create dummy route table: %s %s\n", fnb, strerror( errno ) );
	return;
	}

	write( fd, contents, strlen( contents ) );
	if( (close( fd ) < 0 ) ) {
	fprintf( stderr, "[FAIL] couldn't close dummy route table: %s: %s\n", fnb, strerror( errno ) );
	return;
	}

	setenv( "RMR_SEED_RT", fnb, 0 ); // set it, but don't overwrite it
	}

	int main( int argc, char** argv ) {
	void* mrc; // msg router context
	rmr_mbuf_t* mbuf; // message buffer
	mpl_t* payload; // the payload in a message
	int ai = 1; // arg index
	long timeout;
	long max_timeout = 5; // -t to overrride
	char* target = "localhost:4560"; // address of target to ping
	char* listen_port; // the port we open for "backhaul" connections (random)
	char* tok; // pointer at token in a buffer
	int i;
	char wbuf[1024];
	char me[128]; // who I am to vet rts was actually from me
	int rand_port = 0; // -r sets and causes us to generate a random listen port
	int whid; // id of wormhole
	int num2send = 1; // number of messages to send
	int ep_fd = -1; // epoll's file des (given to epoll_wait)
	int rcv_fd; // file des that NNG tickles -- give this to epoll to listen on
	int nready; // number of events ready for receive
	int count = 0;
	int errors = 0;
	int cksum; // computed simple checksum
	struct timespec in_ts; // time we got response
	struct epoll_event events[1]; // list of events to give to epoll
	struct epoll_event epe; // event definition for event to listen to

	// ---- simple arg parsing ------
	while( ai < argc ) {
	if( *argv[ai] == '-' ) {
	switch( argv[ai][1] ) {
	case 'h': // host port
	ai++;
	target = strdup( argv[ai] );
	break;

	case 'n': // num to send
	ai++;
	num2send = atoi( argv[ai] );
	break;

	case 'r': // generate random listen port
	rand_port = 1;
	;;

	case 't': // timeout
	ai++;
	max_timeout = atoi( argv[ai] );
	break;

	default:
	fprintf( stderr, "[FAIL] unrecognised option: %s\n", argv[ai] );
	exit( 1 );
	}

	ai++;
	} else {
	break; // not an option, leave with a1 @ first positional parm
	}
	}

	if( rand_port ) {
	srand( time( NULL ) );
	snprintf( wbuf, sizeof( wbuf ), "%d", 43000 + (rand() % 1000) ); // random listen port
	listen_port = strdup( wbuf );
	} else {
	listen_port = "43086";
	}


	mk_rt(); // create a dummy route table so we don't have errors/hang

	fprintf( stderr, "[INFO] listen port: %s; sending %d messages\n", listen_port, num2send );

	if( (mrc = rmr_init( listen_port, 1400, 0 )) == NULL ) { // start without route table listener thread
	fprintf( stderr, "[FAIL] unable to initialise RMr\n" );
	exit( 1 );
	}
	fprintf( stderr, "[INFO] RMR initialised\n" );

	if( (rcv_fd = rmr_get_rcvfd( mrc )) < 0 ) { // if we can't get an epoll FD, then we can't timeout; abort
	fprintf( stderr, "[FAIL] unable to get an epoll FD\n" );
	exit( 1 );
	}

	if( (ep_fd = epoll_create1( 0 )) < 0 ) {
	fprintf( stderr, "[FAIL] unable to create epoll fd: %d\n", errno );
	exit( 1 );
	}
	epe.events = EPOLLIN;
	epe.data.fd = rcv_fd;

	if( epoll_ctl( ep_fd, EPOLL_CTL_ADD, rcv_fd, &epe ) != 0 ) {
	fprintf( stderr, "[FAIL] epoll_ctl status not 0 : %s\n", strerror( errno ) );
	exit( 1 );
	}

	while( ! rmr_ready( mrc ) ) {
	sleep( 1 );
	}

	mbuf = rmr_alloc_msg( mrc, sizeof( *payload ) + 100 ); // send buffer with a bit of padding

	fprintf( stderr, "[INFO] starting session with %s, starting to send\n", target );
	whid = rmr_wh_open( mrc, target ); // open a wormhole directly to the target
	if( whid < 0 ) {
	fprintf( stderr, "[FAIL] unable to connect to %s\n", target );
	exit( 2 );
	}

	fprintf( stderr, "[INFO] connected to %s, starting to send\n", target );
	rmr_set_stimeout( mrc, 3 ); // we let rmr retry failures for up to 3 "rounds"

	gethostname( wbuf, sizeof( wbuf ) );
	snprintf( me, sizeof( me ), "%s-%d", wbuf, getpid( ) );

	errors = 0;
	while( count < num2send ) { // we send n messages after the first message is successful
	if( !mbuf ) {
	fprintf( stderr, "[FAIL] mbuf is nil?\n" );
	exit( 1 );
	}

	payload = (mpl_t *) mbuf->payload;

	snprintf( wbuf, sizeof( payload->msg ), "%s count=%d %d", me, count, rand() );
	snprintf( mbuf->payload, 1024, "%d\|%s", sum( wbuf , strlen( wbuf ) ), wbuf );

	mbuf->mtype = HEALTH_CHECK;
	mbuf->sub_id = -1;
	mbuf->len = sizeof( *payload );
	mbuf->state = 0;

	clock_gettime( CLOCK_REALTIME, &payload->out_ts ); // mark time out
	mbuf = rmr_wh_send_msg( mrc, whid, mbuf );

	if( mbuf->state == RMR_OK ) { // good send, wait for response
	nready = epoll_wait( ep_fd, events, 1, max_timeout * 1000 );
	if( nready > 0 ) {
	clock_gettime( CLOCK_REALTIME, &in_ts ); // mark response received time

	mbuf = rmr_rcv_msg( mrc, mbuf );
	payload = (mpl_t *) mbuf->payload;
	tok = strchr( payload->msg, '\|' ); // find end of chksum
	if( tok ) {
	tok++;
	cksum = sum( tok, strlen( tok ) );
	if( cksum != atoi( payload->msg ) ) {
	fprintf( stderr, "[WRN] response to msg %d received, cksum mismatch; expected %d, got %d\n",
	count+1, atoi( payload->msg ), cksum );
	} else {
	fprintf( stderr, "[INFO] response to msg %d received, %d mu-sec\n", count+1, elapsed( &payload->out_ts, &in_ts ) );
	}
	}
	} else {
	fprintf( stderr, "[ERR] timeout waiting for response to message %d\n", count+1 );
	errors++;
	}
	} else {
	fprintf( stderr, "[ERR] send failed: %d\n", mbuf->state );
	}

	count++;
	sleep( 1 );
	}

	rmr_wh_close( mrc, whid );

	return errors = 0;
	}