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