lib/sflow_agent.c

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