src/rmr/common/src/rtc_static.c

   1 // : vi ts=4 sw=4 noet :
   2 /*
   3 ==================================================================================
   4         Copyright (c) 2019 Nokia
   5         Copyright (c) 2018-2019 AT&T Intellectual Property.
   6
   7    Licensed under the Apache License, Version 2.0 (the "License");
   8    you may not use this file except in compliance with the License.
   9    You may obtain a copy of the License at
  10
  11            http://www.apache.org/licenses/LICENSE-2.0
  12
  13    Unless required by applicable law or agreed to in writing, software
  14    distributed under the License is distributed on an "AS IS" BASIS,
  15    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16    See the License for the specific language governing permissions and
  17    limitations under the License.
  18 ==================================================================================
  19 */
  20
  21 /*
  22         Mnemonic:       rt_collector.c
  23         Abstract:       The route table collector is started as a separate pthread and
  24                                 is responsible for listening for route table updates from a
  25                                 route manager or route table generator process.
  26
  27         Author:         E. Scott Daniels
  28         Date:           29 November 2018 (extracted to common 13 March 2019)
  29 */
  30
  31 #ifndef _rt_collector_c
  32 #define _rt_collector_c
  33
  34 #include <stdio.h>
  35 #include <stdlib.h>
  36 #include <netdb.h>
  37 #include <errno.h>
  38 #include <string.h>
  39 #include <errno.h>
  40 #include <fcntl.h>
  41 #include <sys/types.h>
  42 #include <sys/stat.h>
  43 #include <unistd.h>
  44
  45 /*
  46         Route Table Collector
  47         A side thread which opens a socket and subscribes to a routing table generator.
  48         It may do other things along the way (latency measurements?).
  49
  50         The pointer is a pointer to the context.
  51
  52         Listens for records from the route table generation publisher, expecting
  53         one of the following, newline terminated, ASCII records:
  54                 rte|msg-type||]name:port,name:port,...;name:port,...                    // route table entry with one or more groups of endpoints
  55                 new|start                                                               // start of new table
  56                 new|end                                                                 // end of new table; complete
  57
  58                 Name must be a host name which can be looked up via gethostbyname() (DNS).
  59
  60                 Multiple endpoints (name:port) may be given separated by a comma; an endpoint is selected using round robin
  61                         for each message of the type that is sent.
  62
  63                 Multiple endpoint groups can be given as a comma separated list of endpoints, separated by semicolons:
  64                                 group1n1:port,group1n2:port,group1n3:port;group2n1:port,group2n2:port
  65
  66                 If multiple groups are given, when send() is called for the cooresponding message type,
  67                 the message will be sent to one endpoint in each group.
  68
  69                 msg-type is the numeric message type (e.g. 102). If it is given as n,name then it is assumed
  70                 that the entry applies only to the instance running with the hostname 'name.'
  71
  72         Buffers received from the route table generator can contain multiple newline terminated
  73         records, but each buffer must be less than 4K in length, and the last record in a
  74         buffere may NOT be split across buffers.
  75
  76         Other chores:
  77         In addition to the primary task of getting, vetting, and installing a new route table, or
  78         updates to the existing table, this thread will periodically cause the send counts for each
  79         endpoint known to be written to standard error. The frequency is once every 180 seconds, and
  80         more frequently if verbose mode (see ENV_VERBOSE_FILE) is > 0.
  81 */
  82 static void* rtc( void* vctx ) {
  83         uta_ctx_t*      ctx;                                    // context user has -- where we pin the route table
  84         uta_ctx_t*      pvt_cx;                                 // private context for session with rtg
  85         rmr_mbuf_t*     msg = NULL;                             // message from rtg
  86         char*   payload;                                        // payload in the message
  87         size_t  mlen;
  88         size_t  clen;                                           // length to copy and mark
  89         char*   port;                                           // a port number we listen/connect to
  90         char*   fport;                                          // pointer to the real buffer to free
  91         size_t  buf_size;                                       // nng needs var pointer not just size?
  92         char*   nextr;                                          // pointer at next record in the message
  93         char*   curr;                                           // current record
  94         int     i;
  95         long    blabber = 0;                            // time of last blabber so we don't flood if rtg goes bad
  96         int             cstate = -1;                            // connection state to rtg
  97         int             state;                                          // processing state of some nng function
  98         char*   tokens[128];
  99         char    wbuf[128];
 100         char*   pbuf = NULL;
 101         int             pbuf_size = 0;                          // number allocated in pbuf
 102         int             ntoks;
 103         int             raw_interface = 1;                      // rtg is using raw NNG/Nano not RMr to send updates
 104         int             vfd = -1;                                       // verbose file des if we have one
 105         int             vlevel = 0;                                     // how chatty we should be 0== no nattering allowed
 106         char*   eptr;
 107         int             epfd = -1;                                      // fd for epoll so we can multi-task
 108         struct  epoll_event events[1];          // list of events to give to epoll; we only have one we care about
 109         struct  epoll_event epe;                        // event definition for event to listen to
 110         int             rcv_fd = -1;                            // pollable file des from NNG to use for timeout
 111         int             count_delay = 30;                       // number of seconds between writing count info; initially every 30s
 112         int             bump_freq = 0;                          // time at which we will bump count frequency to every 5 minutes
 113
 114
 115         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
 116                 fprintf( stderr, "[CRI] rmr_rtc: internal mishap: context passed in was nil\n" );
 117                 return NULL;
 118         }
 119
 120         if( (eptr = getenv( ENV_VERBOSE_FILE )) != NULL ) {
 121                 vfd = open( eptr, O_RDONLY );
 122                 if( vfd >= 0 ) {
 123                         wbuf[0] = 0;
 124                         lseek( vfd, 0, 0 );
 125                         read( vfd, wbuf, 10 );
 126                         vlevel = atoi( wbuf );
 127                 }
 128         }
 129
 130         read_static_rt( ctx, vlevel );                                          // seed the route table if one provided
 131
 132         if( (port = getenv( ENV_RTG_PORT )) == NULL || ! *port ) {              // port we need to open to listen for RTG connections
 133                 port = strdup( DEF_RTG_PORT );
 134         } else {
 135                 port = strdup( port );
 136         }
 137
 138         if( (curr = getenv( ENV_RTG_RAW )) != NULL ) {
 139                 raw_interface = atoi( curr ) > 0;                               // if > 0 we assume that rtg messages are NOT coming from an RMr based process
 140         }
 141
 142         fport = port;           // must hold to free
 143
 144         ntoks = uta_tokenise( port, tokens, 120, ':' );                 // assume tcp:port, but it could be port or old style host:port
 145         switch( ntoks ) {
 146                 case 1:
 147                                 port = tokens[0];                       // just the port
 148                                 break;
 149
 150                 case 2:
 151                                 port = tokens[1];                       // tcp:port or :port
 152                                 break;
 153
 154                 default:
 155                                 port = DEF_RTG_PORT;            // this shouldn't happen, but parnioia is good
 156                                 break;
 157         }
 158
 159         if( (pvt_cx = init( port, MAX_RTG_MSG_SZ, FL_NOTHREAD )) == NULL ) {                            // open a private context
 160                 fprintf( stderr, "[CRI] rmr_rtc: unable to initialise listen port for RTG (pvt_cx)\n" );
 161
 162                 while( TRUE ) {                                                                                         // no listen port, just dump counts now and then
 163                         sleep( count_delay );
 164                         rt_epcounts( ctx->rtable, ctx->my_name );
 165                 }
 166
 167                 free( fport );                                  // parinoid free and return
 168                 return NULL;
 169         }
 170
 171         if( (rcv_fd = rmr_get_rcvfd( pvt_cx )) >= 0 ) {            // get the epoll fd for the rtg socket
 172                 if( rcv_fd < 0 ) {
 173                         fprintf( stderr, "[WARN] cannot get epoll fd for rtg session; stats will generate only after update from rt manager\n" );
 174                 } else {
 175                         if( (epfd = epoll_create1( 0 )) < 0 ) {
 176                                 fprintf( stderr, "[WARN] stats will generate only after rt manager update; unable to create epoll fd for rtg session: %s\n", strerror( errno ) );
 177                                 rcv_fd = -1;
 178                         } else {
 179                                 epe.events = EPOLLIN;
 180                                 epe.data.fd = rcv_fd;
 181
 182                                 if( epoll_ctl( epfd, EPOLL_CTL_ADD, rcv_fd, &epe ) != 0 )  {
 183                                         fprintf( stderr, "[WARN] stats will generate only after rt manager update; unable to init epoll_ctl: %s\n", strerror( errno ) );
 184                                         rcv_fd = -1;
 185                                 }
 186                         }
 187                 }
 188         }
 189
 190         if( DEBUG ) fprintf( stderr, "[DBUG] rtc thread is running and listening; listening for rtg conns on %s\n", port );
 191         free( fport );
 192
 193         // future:  if we need to register with the rtg, then build a message and send it through a wormhole here
 194
 195         bump_freq = time( NULL ) + 300;                         // after 5 minutes we decrease the count frequency
 196         blabber = 0;
 197         while( 1 ) {                    // until the cows return, pigs fly, or somesuch event likely not to happen
 198                 while( msg == NULL || msg->len <= 0 ) {                                                 // until we actually have something from the other side
 199                         if( rcv_fd < 0 || epoll_wait( epfd, events, 1, 1000 ) > 0 )  {  // skip epoll if init failed, else block for max 1 sec
 200                                 if( raw_interface ) {
 201                                         msg = (rmr_mbuf_t *) rcv_payload( pvt_cx, msg );                // receive from non-RMr sender
 202                                 } else {
 203                                         msg = rmr_rcv_msg( pvt_cx, msg );               // receive from an RMr sender
 204                                 }
 205                         } else {                                                                                                        // no msg, do extra tasks
 206                                 if( msg != NULL ) {                                                                             // if we were working with a message; ensure no len
 207                                         msg->len = 0;
 208                                         msg->state = RMR_ERR_TIMEOUT;
 209                                 }
 210                         }
 211
 212                         if( time( NULL ) > blabber  ) {
 213                                 blabber = time( NULL ) + count_delay;                                   // set next time to blabber, then do so
 214                                 if( blabber > bump_freq ) {
 215                                         count_delay = 300;
 216                                 }
 217                                 rt_epcounts( ctx->rtable, ctx->my_name );
 218                         }
 219                 }
 220
 221                 if( vfd >= 0 ) {                                                        // if file open, check for change to vlevel
 222                         wbuf[0] = 0;
 223                         lseek( vfd, 0, 0 );
 224                         read( vfd, wbuf, 10 );
 225                         vlevel = atoi( wbuf );
 226                 }
 227
 228                 if( msg != NULL && msg->len > 0 ) {
 229                         payload = msg->payload;
 230                         mlen = msg->len;                                        // usable bytes in the payload
 231                         if( vlevel > 1 ) {
 232                                 fprintf( stderr, "[DBUG] rmr_rtc: received rt message; %d bytes (%s)\n", (int) mlen, msg->payload );
 233                         } else {
 234                                 if( DEBUG > 1 || (vlevel > 0) ) fprintf( stderr, "[DBUG] rmr_rtc: received rt message; %d bytes\n", (int) mlen );
 235                         }
 236
 237                         if( pbuf_size <= mlen ) {
 238                                 if( pbuf ) {
 239                                         free( pbuf );
 240                                 }
 241                                 if( mlen < 512 ) {
 242                                         pbuf_size = 512;
 243                                 } else {
 244                                         pbuf_size = mlen * 2;
 245                                 }
 246                                 pbuf = (char *) malloc( sizeof( char ) * pbuf_size );
 247                         }
 248                         memcpy( pbuf, payload, mlen );
 249                         pbuf[mlen] = 0;                                                                         // don't depend on sender making this a legit string
 250
 251                         curr = pbuf;
 252                         while( curr ) {                                                         // loop over each record in the buffer
 253                                 nextr = strchr( curr, '\n' );                   // allow multiple newline records, find end of current and mark
 254
 255                                 if( nextr ) {
 256                                         *(nextr++) = 0;
 257                                 }
 258
 259                                 if( vlevel > 1 ) {
 260                                         fprintf( stderr, "[DBUG] rmr_rtc: processing (%s)\n", curr );
 261                                 }
 262                                 parse_rt_rec( ctx, curr, vlevel );              // parse record and add to in progress table
 263
 264                                 curr = nextr;
 265                         }
 266
 267                         if( ctx->shutdown ) {           // mostly for testing, but allows user app to close us down if rmr_*() function sets this
 268                                 break;
 269                         }
 270
 271                         msg->len = 0;                           // force back into the listen loop
 272                 }
 273         }
 274
 275         return NULL;    // unreachable, but some compilers don't see that and complain.
 276 }
 277
 278
 279 #endif