X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=src%2Ftools%2Fgsprintconsole_ves.c;h=f82be92e5cb93213369a962ebbd79c21e9a4d5a3;hb=refs%2Fheads%2Fmaster;hp=bdaa2773d6dc52890e715d8eb1a3b10a2bab5afb;hpb=c9783d8ea8b85d810483559e50dbf2297109e349;p=com%2Fgs-lite.git diff --git a/src/tools/gsprintconsole_ves.c b/src/tools/gsprintconsole_ves.c index bdaa277..f82be92 100644 --- a/src/tools/gsprintconsole_ves.c +++ b/src/tools/gsprintconsole_ves.c @@ -34,6 +34,14 @@ #include #include +#include +#include + +#ifdef RMR_ENABLED +#include +#include +#include +#endif #include "gsconfig.h" #include "gstypes.h" @@ -48,6 +56,9 @@ static char linebuf[MAXLINE]; int listensockfd=0; int fd=0; +// how frequently we will log stats (expressed in tuples posted) +#define STAT_FREQUENCY 5 + // Not all systems have timersub defined so make sure its ther #ifndef timersub @@ -88,13 +99,13 @@ static void wait_for_client() { /* make sure we can reuse the common port rapidly */ if (setsockopt(listensockfd, SOL_SOCKET, SO_REUSEPORT, (gs_sp_t )&on, sizeof(on)) != 0) { - gslog(LOG_EMERG,"Error::could not set socket option\n"); + gslog(LOG_EMERG,"Error::could not set socket option"); exit(1); } #endif if (setsockopt(listensockfd, SOL_SOCKET, SO_REUSEADDR, (gs_sp_t )&on, sizeof(on)) != 0) { - gslog(LOG_EMERG,"Error::could not set socket option\n"); + gslog(LOG_EMERG,"Error::could not set socket option"); exit(1); } @@ -166,6 +177,12 @@ int main(int argc, char* argv[]) { endpoint dummyep; gs_uint32_t tip1,tip2,tip3,tip4; gs_sp_t instance_name; + gs_sp_t rmr_port = NULL; + +#ifdef RMR_ENABLED + // RMR-related parameters + gs_int32_t rmr_mtype = MC_REPORT; +#endif gs_sp_t curl_address = NULL; endpoint curl_endpoint; @@ -173,16 +190,31 @@ int main(int argc, char* argv[]) { gs_sp_t curl_auth = NULL; gs_uint32_t http_code; - gs_uint32_t ves_version=5; - + gs_uint32_t ves_version=7; gs_uint32_t tlimit = 0; // time limit in seconds time_t start_time, curr_time; + + gs_uint64_t post_success_cnt = 0ULL; + gs_uint64_t post_failure_cnt = 0ULL; + +#ifdef RMR_ENABLED + void* mrc; //msg router context + struct epoll_event events[1]; // list of events to give to epoll + struct epoll_event epe; // event definition for event to listen to + gs_int32_t ep_fd = -1; // epoll's file des (given to epoll_wait) + gs_int32_t rcv_fd; // file des that NNG tickles -- give this to epoll to listen on + gs_int32_t nready; // number of events ready for receive + rmr_mbuf_t* rmr_sbuf; // send buffer + rmr_mbuf_t* rmr_rbuf; // received buffer + + gs_uint64_t rmr_post_success_cnt = 0ULL; + gs_uint64_t rmr_post_failure_cnt = 0ULL; +#endif gsopenlog(argv[0]); - - while ((ch = getopt(argc, argv, "l:p:r:vXDC:U:A:V:")) != -1) { + while ((ch = getopt(argc, argv, "l:p:r:vXDC:U:R:A:V:")) != -1) { switch (ch) { case 'r': bufsz=atoi(optarg); @@ -214,6 +246,9 @@ int main(int argc, char* argv[]) { curl_endpoint.ip=htonl(tip1<<24|tip2<<16|tip3<<8|tip4); curl_endpoint.port=htons(curl_endpoint.port); break; + case 'R': + rmr_port=strdup(optarg); + break; case 'U': curl_url = strdup(optarg); break; @@ -222,7 +257,7 @@ int main(int argc, char* argv[]) { break; default: usage: - fprintf(stderr, "usage: %s [-r ] [-p ] [-l ] [-v] [-X] [-D] [-C :] [-U ] [-A ] [-V ] : query param1 param2...\n", + fprintf(stderr, "usage: %s [-r ] [-p ] [-l ] [-v] [-X] [-D] [-C :] [-U ] [-A ] [-V ] [-R ] : query param1 param2...\n", *argv); exit(1); } @@ -248,7 +283,7 @@ int main(int argc, char* argv[]) { } if (get_initinstance(gshub,instance_name,&dummyep,1)!=0) { - gslog(LOG_EMERG,"Did not receive signal that GS is initiated\n"); + gslog(LOG_EMERG,"Did not receive signal that GS is initiated"); } @@ -268,7 +303,45 @@ int main(int argc, char* argv[]) { argv +=2; if (argc < 1) goto usage; - + + if (rmr_port) { +#ifdef RMR_ENABLED + /* initialize RMR library */ + if( (mrc = rmr_init( rmr_port, 1400, RMRFL_NONE )) == NULL ) { + fprintf(stderr, "%s::error:unable to initialise RMR\n", me); + exit( 1 ); + } + + rcv_fd = rmr_get_rcvfd( mrc ); // set up epoll things, start by getting the FD from MRr + if( rcv_fd < 0 ) { + fprintf(stderr, "%s::error:unable to set up polling fd\n", me); + exit( 1 ); + } + if( (ep_fd = epoll_create1( 0 )) < 0 ) { + fprintf(stderr, "%s::error:unable to create epoll fd: %d\n", me, 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, "%s::error:epoll_ctl status not 0 : %s\n", me, strerror(errno)); + exit( 1 ); + } + + rmr_sbuf = rmr_alloc_msg( mrc, MAXLINE ); // alloc first send buffer; subsequent buffers allcoated on send + rmr_rbuf = NULL; // don't need to alloc receive buffer + + while( ! rmr_ready( mrc ) ) { // must have a route table before we can send; wait til RMr say it has one + sleep( 10 ); + } + fprintf( stderr, "%s: RMR is ready\n", argv[0]); +#else + fprintf(stderr,"Runtime libraries built without RMR support. Rebuild with RMR_ENABLED defined in gsoptions.h\n"); + exit(0); +#endif + } + /* initialize host library and the sgroup */ if (verbose>=2) fprintf(stderr,"Initializing gscp\n"); @@ -384,7 +457,7 @@ int main(int argc, char* argv[]) { long unsigned int lineno=0; long unsigned int seqno=0; - unsigned int measurement_interval; + double measurement_interval; while((code=ftaapp_get_tuple(&rfta_id,&rsize,rbuf,2*MAXTUPLESZ,0))>=0) { lineno++; if (dump) // -D in command line @@ -414,7 +487,7 @@ int main(int argc, char* argv[]) { "\"eventName\": \"Measurement_MC_%s\", " "\"lastEpochMicrosec\": %s, " "\"priority\": \"Normal\", " - "\"reportingEntityName\": \"gs-lite MC\", " + "\"reportingEntityName\": \"GS-LITE MC\", " "\"sequence\": %u, " "\"sourceName\": \"meas_cmpgn_xapp\", " "\"startEpochMicrosec\": %s, " @@ -433,7 +506,7 @@ int main(int argc, char* argv[]) { "\"eventName\": \"Measurement_MC_%s\", " "\"lastEpochMicrosec\": %s, " "\"priority\": \"Normal\", " - "\"reportingEntityName\": \"gs-lite MC\", " + "\"reportingEntityName\": \"GS-LITE MC\", " "\"sequence\": %u, " "\"sourceName\": \"meas_cmpgn_xapp\", " "\"startEpochMicrosec\": %s, " @@ -463,7 +536,7 @@ int main(int argc, char* argv[]) { else pos += snprintf(linebuf+pos,MAXLINE-pos," \"%s\": \"%d\"",field_names[y], ar.r.i); if(y==measurement_interval_pos) - measurement_interval = (unsigned int)ar.r.i; + measurement_interval = (double)ar.r.i; break; case UINT_TYPE: if(ves_version < 7) @@ -471,7 +544,7 @@ int main(int argc, char* argv[]) { else pos += snprintf(linebuf+pos,MAXLINE-pos," \"%s\": \"%u\"",field_names[y], ar.r.ui); if(y==measurement_interval_pos) - measurement_interval = (unsigned int)ar.r.ui; + measurement_interval = (double)ar.r.ui; break; case IP_TYPE: if(ves_version < 7) @@ -521,7 +594,7 @@ int main(int argc, char* argv[]) { else pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"%u\"}",field_names[y], ar.r.ui); if(y==measurement_interval_pos) - measurement_interval = (unsigned int)ar.r.ui; + measurement_interval = (double)ar.r.ui; break; case BOOL_TYPE: if(ves_version < 7){ @@ -544,7 +617,7 @@ int main(int argc, char* argv[]) { else pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"%llu\"",field_names[y], ar.r.ul); if(y==measurement_interval_pos) - measurement_interval = (unsigned int)ar.r.ul; + measurement_interval = (double)ar.r.ul; break; case LLONG_TYPE: if(ves_version < 7) @@ -552,13 +625,15 @@ int main(int argc, char* argv[]) { else pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"%lld\"",field_names[y], ar.r.l); if(y==measurement_interval_pos) - measurement_interval = (unsigned int)ar.r.l; + measurement_interval = (double)ar.r.l; break; case FLOAT_TYPE: if(ves_version < 7) pos += snprintf(linebuf+pos,MAXLINE-pos,"{\"name\": \"%s\", \"value\": \"%f\"}",field_names[y], ar.r.f); else pos += snprintf(linebuf+pos,MAXLINE-pos,"\"%s\": \"%f\"",field_names[y], ar.r.f); + if(y==measurement_interval_pos) + measurement_interval = (double)ar.r.f; break; case TIMEVAL_TYPE: { @@ -609,27 +684,54 @@ int main(int argc, char* argv[]) { } if(ves_version < 7){ snprintf(linebuf+pos, MAXLINE-pos, - "], \"measurementInterval\": %u, \"measurementsForVfScalingVersion\": 1" + "], \"measurementInterval\": %f, \"measurementsForVfScalingVersion\": 1" "}}}\n", measurement_interval ); }else{ snprintf(linebuf+pos, MAXLINE-pos, - "}, \"measurementInterval\": %u, \"measurementFieldsVersion\": \"4.0\"" + "}, \"measurementInterval\": %f, \"measurementFieldsVersion\": \"4.0\"" "}}}\n", measurement_interval ); } + +#ifdef RMR_ENABLED + if (rmr_port) { + rmr_sbuf->mtype = rmr_mtype; // fill in the message bits + rmr_sbuf->len = strlen(linebuf) + 1; // our receiver likely wants a nice acsii-z string + memcpy(rmr_sbuf->payload, linebuf, rmr_sbuf->len); + rmr_sbuf->state = 0; + rmr_sbuf = rmr_send_msg( mrc, rmr_sbuf); // send it (send returns an empty payload on success, or the original payload on fail/retry) + while( rmr_sbuf->state == RMR_ERR_RETRY ) { // soft failure (device busy?) retry + rmr_sbuf = rmr_send_msg( mrc, rmr_sbuf); // retry send until it's good (simple test; real programmes should do better) + } + if(rmr_sbuf->state != RMR_OK) { + gslog(LOG_WARNING, "rmr_send_msg() failure, strerror(errno) is %s", strerror(errno)); + rmr_post_failure_cnt++; + } else + rmr_post_success_cnt++; + if (((rmr_post_success_cnt+rmr_post_failure_cnt) % STAT_FREQUENCY) == 0) + gslog(LOG_WARNING, "%s: successful RMR posts - %llu, failed RMR posts - %llu", argv[0], rmr_post_success_cnt, rmr_post_failure_cnt); + } +#endif + if(curl_address==NULL){ - emit_line(); + if (!rmr_port) // if neither VES collector nor RMR is specified print to standard output + emit_line(); }else{ http_post_request_hdr(curl_endpoint, curl_url, linebuf, &http_code, curl_auth); if(http_code != 200 && http_code != 202){ - gslog(LOG_WARNING, "http return code is %d\n",http_code); - } + post_failure_cnt++; + gslog(LOG_WARNING, "http return code is %d",http_code); + } else { + post_success_cnt++; + } + if (((post_success_cnt+post_failure_cnt) % STAT_FREQUENCY) == 0) + gslog(LOG_WARNING, "%s: successful ves posts - %llu, failed ves posts - %llu", argv[0], post_success_cnt, post_failure_cnt); } if (verbose!=0) fflush(stdout); } else { if (rfta_id.streamid != fta_id.streamid) - fprintf(stderr,"Got unkown streamid %llu \n",rfta_id.streamid); + fprintf(stderr,"Got unknown streamid %llu \n",rfta_id.streamid); } // whenever we receive a temp tuple check if we reached time limit