lib/sflow_agent.c

   1 /* Copyright (c) 2002-2009 InMon Corp. Licensed under the terms of either the
   2  *   Sun Industry Standards Source License 1.1, that is available at:
   3  *    http://host-sflow.sourceforge.net/sissl.html
   4  * or the InMon sFlow License, that is available at:
   5  *    http://www.inmon.com/technology/sflowlicense.txt
   6  */
   7
   8 #include "sflow_api.h"
   9
  10 static void * sflAlloc(SFLAgent *agent, size_t bytes);
  11 static void sflFree(SFLAgent *agent, void *obj);
  12 static void sfl_agent_jumpTableAdd(SFLAgent *agent, SFLSampler *sampler);
  13 static void sfl_agent_jumpTableRemove(SFLAgent *agent, SFLSampler *sampler);
  14
  15 /*________________--------------------------__________________
  16   ________________    sfl_agent_init        __________________
  17   ----------------__________________________------------------
  18 */
  19
  20 void sfl_agent_init(SFLAgent *agent,
  21                     SFLAddress *myIP, /* IP address of this agent in net byte order */
  22                     u_int32_t subId,  /* agent_sub_id */
  23                     time_t bootTime,  /* agent boot time */
  24                     time_t now,       /* time now */
  25                     void *magic,      /* ptr to pass back in logging and alloc fns */
  26                     allocFn_t allocFn,
  27                     freeFn_t freeFn,
  28                     errorFn_t errorFn,
  29                     sendFn_t sendFn)
  30 {
  31     /* first clear everything */
  32     memset(agent, 0, sizeof(*agent));
  33     /* now copy in the parameters */
  34     agent->myIP = *myIP; /* structure copy */
  35     agent->subId = subId;
  36     agent->bootTime = bootTime;
  37     agent->now = now;
  38     agent->magic = magic;
  39     agent->allocFn = allocFn;
  40     agent->freeFn = freeFn;
  41     agent->errorFn = errorFn;
  42     agent->sendFn = sendFn;
  43
  44 #ifdef SFLOW_DO_SOCKET
  45     if(sendFn == NULL) {
  46         /* open the socket - really need one for v4 and another for v6? */
  47         if((agent->receiverSocket4 = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) == -1)
  48             sfl_agent_sysError(agent, "agent", "IPv4 socket open failed");
  49         if((agent->receiverSocket6 = socket(AF_INET6, SOCK_DGRAM, IPPROTO_UDP)) == -1)
  50             sfl_agent_sysError(agent, "agent", "IPv6 socket open failed");
  51     }
  52 #endif
  53 }
  54
  55 /*_________________---------------------------__________________
  56   _________________   sfl_agent_release       __________________
  57   -----------------___________________________------------------
  58 */
  59
  60 void sfl_agent_release(SFLAgent *agent)
  61 {
  62     /* release and free the samplers, pollers and receivers */
  63     SFLSampler *sm = agent->samplers;
  64     SFLPoller *pl = agent->pollers;
  65     SFLReceiver *rcv = agent->receivers;
  66
  67     for(; sm != NULL; ) {
  68         SFLSampler *nextSm = sm->nxt;
  69         sflFree(agent, sm);
  70         sm = nextSm;
  71     }
  72     agent->samplers = NULL;
  73
  74     for(; pl != NULL; ) {
  75         SFLPoller *nextPl = pl->nxt;
  76         sflFree(agent, pl);
  77         pl = nextPl;
  78     }
  79     agent->pollers = NULL;
  80
  81     for(; rcv != NULL; ) {
  82         SFLReceiver *nextRcv = rcv->nxt;
  83         sflFree(agent, rcv);
  84         rcv = nextRcv;
  85     }
  86     agent->receivers = NULL;
  87
  88 #ifdef SFLOW_DO_SOCKET
  89     /* close the sockets */
  90     if(agent->receiverSocket4 > 0) close(agent->receiverSocket4);
  91     if(agent->receiverSocket6 > 0) close(agent->receiverSocket6);
  92 #endif
  93 }
  94
  95
  96 /*_________________---------------------------__________________
  97   _________________   sfl_agent_set_*         __________________
  98   -----------------___________________________------------------
  99 */
 100
 101 void sfl_agent_set_agentAddress(SFLAgent *agent, SFLAddress *addr)
 102 {
 103     if(addr && memcmp(addr, &agent->myIP, sizeof(agent->myIP)) != 0) {
 104         /* change of address */
 105         agent->myIP = *addr; /* structure copy */
 106         /* reset sequence numbers here? */
 107     }
 108 }
 109
 110 void sfl_agent_set_agentSubId(SFLAgent *agent, u_int32_t subId)
 111 {
 112     if(subId != agent->subId) {
 113         /* change of subId */
 114         agent->subId = subId;
 115         /* reset sequence numbers here? */
 116     }
 117 }
 118
 119 /*_________________---------------------------__________________
 120   _________________   sfl_agent_tick          __________________
 121   -----------------___________________________------------------
 122 */
 123
 124 void sfl_agent_tick(SFLAgent *agent, time_t now)
 125 {
 126     SFLReceiver *rcv = agent->receivers;
 127     SFLSampler *sm = agent->samplers;
 128     SFLPoller *pl = agent->pollers;
 129     agent->now = now;
 130     /* receivers use ticks to flush send data */
 131     for(; rcv != NULL; rcv = rcv->nxt) sfl_receiver_tick(rcv, now);
 132     /* samplers use ticks to decide when they are sampling too fast */
 133     for(; sm != NULL; sm = sm->nxt) sfl_sampler_tick(sm, now);
 134     /* pollers use ticks to decide when to ask for counters */
 135     for(; pl != NULL; pl = pl->nxt) sfl_poller_tick(pl, now);
 136 }
 137
 138 /*_________________---------------------------__________________
 139   _________________   sfl_agent_addReceiver   __________________
 140   -----------------___________________________------------------
 141 */
 142
 143 SFLReceiver *sfl_agent_addReceiver(SFLAgent *agent)
 144 {
 145     SFLReceiver *rcv = (SFLReceiver *)sflAlloc(agent, sizeof(SFLReceiver));
 146     sfl_receiver_init(rcv, agent);
 147     /* add to end of list - to preserve the receiver index numbers for existing receivers */
 148     {
 149         SFLReceiver *r, *prev = NULL;
 150         for(r = agent->receivers; r != NULL; prev = r, r = r->nxt);
 151         if(prev) prev->nxt = rcv;
 152         else agent->receivers = rcv;
 153         rcv->nxt = NULL;
 154     }
 155     return rcv;
 156 }
 157
 158 /*_________________---------------------------__________________
 159   _________________     sfl_dsi_compare       __________________
 160   -----------------___________________________------------------
 161
 162   Note that if there is a mixture of ds_classes for this agent, then
 163   the simple numeric comparison may not be correct - the sort order (for
 164   the purposes of the SNMP MIB) should really be determined by the OID
 165   that these numeric ds_class numbers are a shorthand for.  For example,
 166   ds_class == 0 means ifIndex, which is the oid "1.3.6.1.2.1.2.2.1"
 167 */
 168
 169 static inline int sfl_dsi_compare(SFLDataSource_instance *pdsi1, SFLDataSource_instance *pdsi2) {
 170     /* could have used just memcmp(),  but not sure if that would
 171        give the right answer on little-endian platforms. Safer to be explicit... */
 172     int cmp = pdsi2->ds_class - pdsi1->ds_class;
 173     if(cmp == 0) cmp = pdsi2->ds_index - pdsi1->ds_index;
 174     if(cmp == 0) cmp = pdsi2->ds_instance - pdsi1->ds_instance;
 175     return cmp;
 176 }
 177
 178 /*_________________---------------------------__________________
 179   _________________   sfl_agent_addSampler    __________________
 180   -----------------___________________________------------------
 181 */
 182
 183 SFLSampler *sfl_agent_addSampler(SFLAgent *agent, SFLDataSource_instance *pdsi)
 184 {
 185     /* Keep the list sorted. */
 186     SFLSampler *prev = NULL, *sm = agent->samplers;
 187     for(; sm != NULL; prev = sm, sm = sm->nxt) {
 188         int64_t cmp = sfl_dsi_compare(pdsi, &sm->dsi);
 189         if(cmp == 0) return sm;  /* found - return existing one */
 190         if(cmp < 0) break;       /* insert here */
 191     }
 192     /* either we found the insert point, or reached the end of the list...*/
 193
 194     {
 195         SFLSampler *newsm = (SFLSampler *)sflAlloc(agent, sizeof(SFLSampler));
 196         sfl_sampler_init(newsm, agent, pdsi);
 197         if(prev) prev->nxt = newsm;
 198         else agent->samplers = newsm;
 199         newsm->nxt = sm;
 200
 201         /* see if we should go in the ifIndex jumpTable */
 202         if(SFL_DS_CLASS(newsm->dsi) == 0) {
 203             SFLSampler *test = sfl_agent_getSamplerByIfIndex(agent, SFL_DS_INDEX(newsm->dsi));
 204             if(test && (SFL_DS_INSTANCE(newsm->dsi) < SFL_DS_INSTANCE(test->dsi))) {
 205                 /* replace with this new one because it has a lower ds_instance number */
 206                 sfl_agent_jumpTableRemove(agent, test);
 207                 test = NULL;
 208             }
 209             if(test == NULL) sfl_agent_jumpTableAdd(agent, newsm);
 210         }
 211         return newsm;
 212     }
 213 }
 214
 215 /*_________________---------------------------__________________
 216   _________________   sfl_agent_addPoller     __________________
 217   -----------------___________________________------------------
 218 */
 219
 220 SFLPoller *sfl_agent_addPoller(SFLAgent *agent,
 221                                SFLDataSource_instance *pdsi,
 222                                void *magic,         /* ptr to pass back in getCountersFn() */
 223                                getCountersFn_t getCountersFn)
 224 {
 225     /* keep the list sorted */
 226     SFLPoller *prev = NULL, *pl = agent->pollers;
 227     for(; pl != NULL; prev = pl, pl = pl->nxt) {
 228         int64_t cmp = sfl_dsi_compare(pdsi, &pl->dsi);
 229         if(cmp == 0) return pl;  /* found - return existing one */
 230         if(cmp < 0) break;       /* insert here */
 231     }
 232     /* either we found the insert point, or reached the end of the list... */
 233     {
 234         SFLPoller *newpl = (SFLPoller *)sflAlloc(agent, sizeof(SFLPoller));
 235         sfl_poller_init(newpl, agent, pdsi, magic, getCountersFn);
 236         if(prev) prev->nxt = newpl;
 237         else agent->pollers = newpl;
 238         newpl->nxt = pl;
 239         return newpl;
 240     }
 241 }
 242
 243 /*_________________---------------------------__________________
 244   _________________  sfl_agent_removeSampler  __________________
 245   -----------------___________________________------------------
 246 */
 247
 248 int sfl_agent_removeSampler(SFLAgent *agent, SFLDataSource_instance *pdsi)
 249 {
 250     /* find it, unlink it and free it */
 251     SFLSampler *prev = NULL, *sm = agent->samplers;
 252     for(; sm != NULL; prev = sm, sm = sm->nxt) {
 253         if(sfl_dsi_compare(pdsi, &sm->dsi) == 0) {
 254             if(prev == NULL) agent->samplers = sm->nxt;
 255             else prev->nxt = sm->nxt;
 256             sfl_agent_jumpTableRemove(agent, sm);
 257             sflFree(agent, sm);
 258             return 1;
 259         }
 260     }
 261     /* not found */
 262     return 0;
 263 }
 264
 265 /*_________________---------------------------__________________
 266   _________________  sfl_agent_removePoller   __________________
 267   -----------------___________________________------------------
 268 */
 269
 270 int sfl_agent_removePoller(SFLAgent *agent, SFLDataSource_instance *pdsi)
 271 {
 272     /* find it, unlink it and free it */
 273     SFLPoller *prev = NULL, *pl = agent->pollers;
 274     for(; pl != NULL; prev = pl, pl = pl->nxt) {
 275         if(sfl_dsi_compare(pdsi, &pl->dsi) == 0) {
 276             if(prev == NULL) agent->pollers = pl->nxt;
 277             else prev->nxt = pl->nxt;
 278             sflFree(agent, pl);
 279             return 1;
 280         }
 281     }
 282     /* not found */
 283     return 0;
 284 }
 285
 286 /*_________________--------------------------------__________________
 287   _________________  sfl_agent_jumpTableAdd        __________________
 288   -----------------________________________________------------------
 289 */
 290
 291 static void sfl_agent_jumpTableAdd(SFLAgent *agent, SFLSampler *sampler)
 292 {
 293     u_int32_t hashIndex = SFL_DS_INDEX(sampler->dsi) % SFL_HASHTABLE_SIZ;
 294     sampler->hash_nxt = agent->jumpTable[hashIndex];
 295     agent->jumpTable[hashIndex] = sampler;
 296 }
 297
 298 /*_________________--------------------------------__________________
 299   _________________  sfl_agent_jumpTableRemove     __________________
 300   -----------------________________________________------------------
 301 */
 302
 303 static void sfl_agent_jumpTableRemove(SFLAgent *agent, SFLSampler *sampler)
 304 {
 305     u_int32_t hashIndex = SFL_DS_INDEX(sampler->dsi) % SFL_HASHTABLE_SIZ;
 306     SFLSampler *search = agent->jumpTable[hashIndex], *prev = NULL;
 307     for( ; search != NULL; prev = search, search = search->hash_nxt) if(search == sampler) break;
 308     if(search) {
 309         // found - unlink
 310         if(prev) prev->hash_nxt = search->hash_nxt;
 311         else agent->jumpTable[hashIndex] = search->hash_nxt;
 312         search->hash_nxt = NULL;
 313     }
 314 }
 315
 316 /*_________________--------------------------------__________________
 317   _________________  sfl_agent_getSamplerByIfIndex __________________
 318   -----------------________________________________------------------
 319   fast lookup (pointers cached in hash table).  If there are multiple
 320   sampler instances for a given ifIndex, then this fn will return
 321   the one with the lowest instance number.  Since the samplers
 322   list is sorted, this means the other instances will be accesible
 323   by following the sampler->nxt pointer (until the ds_class
 324   or ds_index changes).  This is helpful if you need to offer
 325   the same flowSample to multiple samplers.
 326 */
 327
 328 SFLSampler *sfl_agent_getSamplerByIfIndex(SFLAgent *agent, u_int32_t ifIndex)
 329 {
 330     SFLSampler *search = agent->jumpTable[ifIndex % SFL_HASHTABLE_SIZ];
 331     for( ; search != NULL; search = search->hash_nxt) if(SFL_DS_INDEX(search->dsi) == ifIndex) break;
 332     return search;
 333 }
 334
 335 /*_________________---------------------------__________________
 336   _________________  sfl_agent_getSampler     __________________
 337   -----------------___________________________------------------
 338 */
 339
 340 SFLSampler *sfl_agent_getSampler(SFLAgent *agent, SFLDataSource_instance *pdsi)
 341 {
 342     /* find it and return it */
 343     SFLSampler *sm = agent->samplers;
 344     for(; sm != NULL; sm = sm->nxt)
 345         if(sfl_dsi_compare(pdsi, &sm->dsi) == 0) return sm;
 346     /* not found */
 347     return NULL;
 348 }
 349
 350 /*_________________---------------------------__________________
 351   _________________  sfl_agent_getPoller      __________________
 352   -----------------___________________________------------------
 353 */
 354
 355 SFLPoller *sfl_agent_getPoller(SFLAgent *agent, SFLDataSource_instance *pdsi)
 356 {
 357     /* find it and return it */
 358     SFLPoller *pl = agent->pollers;
 359     for(; pl != NULL; pl = pl->nxt)
 360         if(sfl_dsi_compare(pdsi, &pl->dsi) == 0) return pl;
 361     /* not found */
 362     return NULL;
 363 }
 364
 365 /*_________________---------------------------__________________
 366   _________________  sfl_agent_getReceiver    __________________
 367   -----------------___________________________------------------
 368 */
 369
 370 SFLReceiver *sfl_agent_getReceiver(SFLAgent *agent, u_int32_t receiverIndex)
 371 {
 372     u_int32_t rcvIdx = 0;
 373     SFLReceiver *rcv = agent->receivers;
 374     for(;  rcv != NULL; rcv = rcv->nxt)
 375         if(receiverIndex == ++rcvIdx) return rcv;
 376
 377     /* not found - ran off the end of the table */
 378     return NULL;
 379 }
 380
 381 /*_________________---------------------------__________________
 382   _________________ sfl_agent_getNextSampler  __________________
 383   -----------------___________________________------------------
 384 */
 385
 386 SFLSampler *sfl_agent_getNextSampler(SFLAgent *agent, SFLDataSource_instance *pdsi)
 387 {
 388     /* return the one lexograpically just after it - assume they are sorted
 389        correctly according to the lexographical ordering of the object ids */
 390     SFLSampler *sm = sfl_agent_getSampler(agent, pdsi);
 391     return sm ? sm->nxt : NULL;
 392 }
 393
 394 /*_________________---------------------------__________________
 395   _________________ sfl_agent_getNextPoller   __________________
 396   -----------------___________________________------------------
 397 */
 398
 399 SFLPoller *sfl_agent_getNextPoller(SFLAgent *agent, SFLDataSource_instance *pdsi)
 400 {
 401     /* return the one lexograpically just after it - assume they are sorted
 402        correctly according to the lexographical ordering of the object ids */
 403     SFLPoller *pl = sfl_agent_getPoller(agent, pdsi);
 404     return pl ? pl->nxt : NULL;
 405 }
 406
 407 /*_________________---------------------------__________________
 408   _________________ sfl_agent_getNextReceiver __________________
 409   -----------------___________________________------------------
 410 */
 411
 412 SFLReceiver *sfl_agent_getNextReceiver(SFLAgent *agent, u_int32_t receiverIndex)
 413 {
 414     return sfl_agent_getReceiver(agent, receiverIndex + 1);
 415 }
 416
 417
 418 /*_________________---------------------------__________________
 419   _________________ sfl_agent_resetReceiver   __________________
 420   -----------------___________________________------------------
 421 */
 422
 423 void sfl_agent_resetReceiver(SFLAgent *agent, SFLReceiver *receiver)
 424 {
 425     /* tell samplers and pollers to stop sending to this receiver */
 426     /* first get his receiverIndex */
 427     u_int32_t rcvIdx = 0;
 428     SFLReceiver *rcv = agent->receivers;
 429     for(; rcv != NULL; rcv = rcv->nxt) {
 430         rcvIdx++; /* thanks to Diego Valverde for pointing out this bugfix */
 431         if(rcv == receiver) {
 432             /* now tell anyone that is using it to stop */
 433             SFLSampler *sm = agent->samplers;
 434             SFLPoller *pl = agent->pollers;
 435
 436             for(; sm != NULL; sm = sm->nxt)
 437                 if(sfl_sampler_get_sFlowFsReceiver(sm) == rcvIdx) sfl_sampler_set_sFlowFsReceiver(sm, 0);
 438
 439             for(; pl != NULL; pl = pl->nxt)
 440                 if(sfl_poller_get_sFlowCpReceiver(pl) == rcvIdx) sfl_poller_set_sFlowCpReceiver(pl, 0);
 441
 442             break;
 443         }
 444     }
 445 }
 446
 447 /*_________________---------------------------__________________
 448   _________________     sfl_agent_error       __________________
 449   -----------------___________________________------------------
 450 */
 451 #define MAX_ERRMSG_LEN 1000
 452
 453 void sfl_agent_error(SFLAgent *agent, char *modName, char *msg)
 454 {
 455     char errm[MAX_ERRMSG_LEN];
 456     snprintf(errm, sizeof errm, "sfl_agent_error: %s: %s\n", modName, msg);
 457     if(agent->errorFn) (*agent->errorFn)(agent->magic, agent, errm);
 458     else {
 459         fprintf(stderr, "%s\n", errm);
 460         fflush(stderr);
 461     }
 462 }
 463
 464 /*_________________---------------------------__________________
 465   _________________     sfl_agent_sysError    __________________
 466   -----------------___________________________------------------
 467 */
 468
 469 void sfl_agent_sysError(SFLAgent *agent, char *modName, char *msg)
 470 {
 471     char errm[MAX_ERRMSG_LEN];
 472     snprintf(errm, sizeof errm, "sfl_agent_sysError: %s: %s (errno = %d - %s)\n", modName, msg, errno, strerror(errno));
 473     if(agent->errorFn) (*agent->errorFn)(agent->magic, agent, errm);
 474     else {
 475         fprintf(stderr, "%s\n", errm);
 476         fflush(stderr);
 477     }
 478 }
 479
 480
 481 /*_________________---------------------------__________________
 482   _________________       alloc and free      __________________
 483   -----------------___________________________------------------
 484 */
 485
 486 static void * sflAlloc(SFLAgent *agent, size_t bytes)
 487 {
 488     if(agent->allocFn) return (*agent->allocFn)(agent->magic, agent, bytes);
 489     else return SFL_ALLOC(bytes);
 490 }
 491
 492 static void sflFree(SFLAgent *agent, void *obj)
 493 {
 494     if(agent->freeFn) (*agent->freeFn)(agent->magic, agent, obj);
 495     else SFL_FREE(obj);
 496 }