ovs-thread: Quiesce in xpthread_barrier_wait().
[sliver-openvswitch.git] / lib / sflow_receiver.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 #ifndef __CHECKER__            /* Don't run sparse on anything in this file. */
9
10 #include <assert.h>
11 #include "sflow_api.h"
12
13 static void resetSampleCollector(SFLReceiver *receiver);
14 static void sendSample(SFLReceiver *receiver);
15 static void sflError(SFLReceiver *receiver, char *errm);
16 inline static void putNet32(SFLReceiver *receiver, u_int32_t val);
17 inline static void putAddress(SFLReceiver *receiver, SFLAddress *addr);
18 #ifdef SFLOW_DO_SOCKET
19 static void initSocket(SFLReceiver *receiver);
20 #endif
21
22 /*_________________--------------------------__________________
23   _________________    sfl_receiver_init     __________________
24   -----------------__________________________------------------
25 */
26
27 void sfl_receiver_init(SFLReceiver *receiver, SFLAgent *agent)
28 {
29     /* first clear everything */
30     memset(receiver, 0, sizeof(*receiver));
31
32     /* now copy in the parameters */
33     receiver->agent = agent;
34
35     /* set defaults */
36     receiver->sFlowRcvrMaximumDatagramSize = SFL_DEFAULT_DATAGRAM_SIZE;
37     receiver->sFlowRcvrPort = SFL_DEFAULT_COLLECTOR_PORT;
38
39 #ifdef SFLOW_DO_SOCKET
40     /* initialize the socket address */
41     initSocket(receiver);
42 #endif
43
44     /* preset some of the header fields */
45     receiver->sampleCollector.datap = receiver->sampleCollector.data;
46     putNet32(receiver, SFLDATAGRAM_VERSION5);
47     putAddress(receiver, &agent->myIP);
48     putNet32(receiver, agent->subId);
49
50     /* prepare to receive the first sample */
51     resetSampleCollector(receiver);
52 }
53
54 /*_________________---------------------------__________________
55   _________________      reset                __________________
56   -----------------___________________________------------------
57
58   called on timeout, or when owner string is cleared
59 */
60
61 static void reset(SFLReceiver *receiver) {
62     // ask agent to tell samplers and pollers to stop sending samples
63     sfl_agent_resetReceiver(receiver->agent, receiver);
64     // reinitialize
65     sfl_receiver_init(receiver, receiver->agent);
66 }
67
68 #ifdef SFLOW_DO_SOCKET
69 /*_________________---------------------------__________________
70   _________________      initSocket           __________________
71   -----------------___________________________------------------
72 */
73
74 static void initSocket(SFLReceiver *receiver) {
75     if(receiver->sFlowRcvrAddress.type == SFLADDRESSTYPE_IP_V6) {
76         struct sockaddr_in6 *sa6 = &receiver->receiver6;
77         sa6->sin6_port = htons((u_int16_t)receiver->sFlowRcvrPort);
78         sa6->sin6_family = AF_INET6;
79         sa6->sin6_addr = receiver->sFlowRcvrAddress.address.ip_v6;
80     }
81     else {
82         struct sockaddr_in *sa4 = &receiver->receiver4;
83         sa4->sin_port = htons((u_int16_t)receiver->sFlowRcvrPort);
84         sa4->sin_family = AF_INET;
85         sa4->sin_addr = receiver->sFlowRcvrAddress.address.ip_v4;
86     }
87 }
88 #endif
89
90 /*_________________----------------------------------------_____________
91   _________________          MIB Vars                      _____________
92   -----------------________________________________________-------------
93 */
94
95 char * sfl_receiver_get_sFlowRcvrOwner(SFLReceiver *receiver) {
96     return receiver->sFlowRcvrOwner;
97 }
98 void sfl_receiver_set_sFlowRcvrOwner(SFLReceiver *receiver, char *sFlowRcvrOwner) {
99     receiver->sFlowRcvrOwner = sFlowRcvrOwner;
100     if(sFlowRcvrOwner == NULL || sFlowRcvrOwner[0] == '\0') {
101         // reset condition! owner string was cleared
102         reset(receiver);
103     }
104 }
105 time_t sfl_receiver_get_sFlowRcvrTimeout(SFLReceiver *receiver) {
106     return receiver->sFlowRcvrTimeout;
107 }
108 void sfl_receiver_set_sFlowRcvrTimeout(SFLReceiver *receiver, time_t sFlowRcvrTimeout) {
109     receiver->sFlowRcvrTimeout =sFlowRcvrTimeout;
110 }
111 u_int32_t sfl_receiver_get_sFlowRcvrMaximumDatagramSize(SFLReceiver *receiver) {
112     return receiver->sFlowRcvrMaximumDatagramSize;
113 }
114 void sfl_receiver_set_sFlowRcvrMaximumDatagramSize(SFLReceiver *receiver, u_int32_t sFlowRcvrMaximumDatagramSize) {
115     u_int32_t mdz = sFlowRcvrMaximumDatagramSize;
116     if(mdz < SFL_MIN_DATAGRAM_SIZE) mdz = SFL_MIN_DATAGRAM_SIZE;
117     receiver->sFlowRcvrMaximumDatagramSize = mdz;
118 }
119 SFLAddress *sfl_receiver_get_sFlowRcvrAddress(SFLReceiver *receiver) {
120     return &receiver->sFlowRcvrAddress;
121 }
122 void sfl_receiver_set_sFlowRcvrAddress(SFLReceiver *receiver, SFLAddress *sFlowRcvrAddress) {
123     if(sFlowRcvrAddress) receiver->sFlowRcvrAddress = *sFlowRcvrAddress; // structure copy
124 #ifdef SFLOW_DO_SOCKET
125     initSocket(receiver);
126 #endif
127 }
128 u_int32_t sfl_receiver_get_sFlowRcvrPort(SFLReceiver *receiver) {
129     return receiver->sFlowRcvrPort;
130 }
131 void sfl_receiver_set_sFlowRcvrPort(SFLReceiver *receiver, u_int32_t sFlowRcvrPort) {
132     receiver->sFlowRcvrPort = sFlowRcvrPort;
133     // update the socket structure
134 #ifdef SFLOW_DO_SOCKET
135     initSocket(receiver);
136 #endif
137 }
138
139 /*_________________---------------------------__________________
140   _________________   sfl_receiver_tick       __________________
141   -----------------___________________________------------------
142 */
143
144 void sfl_receiver_tick(SFLReceiver *receiver, time_t now)
145 {
146     // if there are any samples to send, flush them now
147     if(receiver->sampleCollector.numSamples > 0) sendSample(receiver);
148     // check the timeout
149     if(receiver->sFlowRcvrTimeout && (u_int32_t)receiver->sFlowRcvrTimeout != 0xFFFFFFFF) {
150         // count down one tick and reset if we reach 0
151         if(--receiver->sFlowRcvrTimeout == 0) reset(receiver);
152     }
153 }
154
155 /*_________________-----------------------------__________________
156   _________________   receiver write utilities  __________________
157   -----------------_____________________________------------------
158 */
159
160 inline static void put32(SFLReceiver *receiver, u_int32_t val)
161 {
162     *receiver->sampleCollector.datap++ = val;
163 }
164
165 inline static void putNet32(SFLReceiver *receiver, u_int32_t val)
166 {
167     *receiver->sampleCollector.datap++ = htonl(val);
168 }
169
170 inline static void putNet32_run(SFLReceiver *receiver, void *obj, size_t quads)
171 {
172     u_int32_t *from = (u_int32_t *)obj;
173     while(quads--) putNet32(receiver, *from++);
174 }
175
176 inline static void putNet64(SFLReceiver *receiver, u_int64_t val64)
177 {
178     u_int32_t *firstQuadPtr = receiver->sampleCollector.datap;
179     // first copy the bytes in
180     memcpy((u_char *)firstQuadPtr, &val64, 8);
181     if(htonl(1) != 1) {
182         // swap the bytes, and reverse the quads too
183         u_int32_t tmp = *receiver->sampleCollector.datap++;
184         *firstQuadPtr = htonl(*receiver->sampleCollector.datap);
185         *receiver->sampleCollector.datap++ = htonl(tmp);
186     }
187     else receiver->sampleCollector.datap += 2;
188 }
189
190 inline static void put128(SFLReceiver *receiver, u_char *val)
191 {
192     memcpy(receiver->sampleCollector.datap, val, 16);
193     receiver->sampleCollector.datap += 4;
194 }
195
196 inline static void putString(SFLReceiver *receiver, SFLString *s)
197 {
198     putNet32(receiver, s->len);
199     memcpy(receiver->sampleCollector.datap, s->str, s->len);
200     receiver->sampleCollector.datap += (s->len + 3) / 4; /* pad to 4-byte boundary */
201     if ((s->len % 4) != 0){
202         u_int8_t padding = 4 - (s->len % 4);
203         memset(((u_int8_t*)receiver->sampleCollector.datap)-padding, 0, padding);
204     }
205 }
206
207 inline static u_int32_t stringEncodingLength(SFLString *s) {
208     // answer in bytes,  so remember to mulitply by 4 after rounding up to nearest 4-byte boundary
209     return 4 + (((s->len + 3) / 4) * 4);
210 }
211
212 inline static void putAddress(SFLReceiver *receiver, SFLAddress *addr)
213 {
214     // encode unspecified addresses as IPV4:0.0.0.0 - or should we flag this as an error?
215     if(addr->type == 0) {
216         putNet32(receiver, SFLADDRESSTYPE_IP_V4);
217         put32(receiver, 0);
218     }
219     else {
220         putNet32(receiver, addr->type);
221         if(addr->type == SFLADDRESSTYPE_IP_V4) put32(receiver, addr->address.ip_v4.addr);
222         else put128(receiver, addr->address.ip_v6.addr);
223     }
224 }
225
226 inline static u_int32_t addressEncodingLength(SFLAddress *addr) {
227     return (addr->type == SFLADDRESSTYPE_IP_V6) ? 20 : 8;  // type + address (unspecified == IPV4)
228 }
229
230 inline static void putMACAddress(SFLReceiver *receiver, u_int8_t *mac)
231 {
232     memcpy(receiver->sampleCollector.datap, mac, 6);
233     receiver->sampleCollector.datap += 2;
234 }
235
236 inline static void putSwitch(SFLReceiver *receiver, SFLExtended_switch *sw)
237 {
238     putNet32(receiver, sw->src_vlan);
239     putNet32(receiver, sw->src_priority);
240     putNet32(receiver, sw->dst_vlan);
241     putNet32(receiver, sw->dst_priority);
242 }
243
244 inline static void putRouter(SFLReceiver *receiver, SFLExtended_router *router)
245 {
246     putAddress(receiver, &router->nexthop);
247     putNet32(receiver, router->src_mask);
248     putNet32(receiver, router->dst_mask);
249 }
250
251 inline static u_int32_t routerEncodingLength(SFLExtended_router *router) {
252     return addressEncodingLength(&router->nexthop) + 8;
253 }
254
255 inline static void putGateway(SFLReceiver *receiver, SFLExtended_gateway *gw)
256 {
257     putAddress(receiver, &gw->nexthop);
258     putNet32(receiver, gw->as);
259     putNet32(receiver, gw->src_as);
260     putNet32(receiver, gw->src_peer_as);
261     putNet32(receiver, gw->dst_as_path_segments);
262     {
263         u_int32_t seg = 0;
264         for(; seg < gw->dst_as_path_segments; seg++) {
265             putNet32(receiver, gw->dst_as_path[seg].type);
266             putNet32(receiver, gw->dst_as_path[seg].length);
267             putNet32_run(receiver, gw->dst_as_path[seg].as.seq, gw->dst_as_path[seg].length);
268         }
269     }
270     putNet32(receiver, gw->communities_length);
271     putNet32_run(receiver, gw->communities, gw->communities_length);
272     putNet32(receiver, gw->localpref);
273 }
274
275 inline static u_int32_t gatewayEncodingLength(SFLExtended_gateway *gw) {
276     u_int32_t elemSiz = addressEncodingLength(&gw->nexthop);
277     u_int32_t seg = 0;
278     elemSiz += 16; // as, src_as, src_peer_as, dst_as_path_segments
279     for(; seg < gw->dst_as_path_segments; seg++) {
280         elemSiz += 8; // type, length
281         elemSiz += 4 * gw->dst_as_path[seg].length; // set/seq bytes
282     }
283     elemSiz += 4; // communities_length
284     elemSiz += 4 * gw->communities_length; // communities
285     elemSiz += 4; // localpref
286     return elemSiz;
287 }
288
289 inline static void putUser(SFLReceiver *receiver, SFLExtended_user *user)
290 {
291     putNet32(receiver, user->src_charset);
292     putString(receiver, &user->src_user);
293     putNet32(receiver, user->dst_charset);
294     putString(receiver, &user->dst_user);
295 }
296
297 inline static u_int32_t userEncodingLength(SFLExtended_user *user) {
298     return 4
299         + stringEncodingLength(&user->src_user)
300         + 4
301         + stringEncodingLength(&user->dst_user);
302 }
303
304 inline static void putUrl(SFLReceiver *receiver, SFLExtended_url *url)
305 {
306     putNet32(receiver, url->direction);
307     putString(receiver, &url->url);
308     putString(receiver, &url->host);
309 }
310
311 inline static u_int32_t urlEncodingLength(SFLExtended_url *url) {
312     return 4
313         + stringEncodingLength(&url->url)
314         + stringEncodingLength(&url->host);
315 }
316
317 inline static void putLabelStack(SFLReceiver *receiver, SFLLabelStack *labelStack)
318 {
319     putNet32(receiver, labelStack->depth);
320     putNet32_run(receiver, labelStack->stack, labelStack->depth);
321 }
322
323 inline static u_int32_t labelStackEncodingLength(SFLLabelStack *labelStack) {
324     return 4 + (4 * labelStack->depth);
325 }
326
327 inline static void putMpls(SFLReceiver *receiver, SFLExtended_mpls *mpls)
328 {
329     putAddress(receiver, &mpls->nextHop);
330     putLabelStack(receiver, &mpls->in_stack);
331     putLabelStack(receiver, &mpls->out_stack);
332 }
333
334 inline static u_int32_t mplsEncodingLength(SFLExtended_mpls *mpls) {
335     return addressEncodingLength(&mpls->nextHop)
336         + labelStackEncodingLength(&mpls->in_stack)
337         + labelStackEncodingLength(&mpls->out_stack);
338 }
339
340 inline static void putNat(SFLReceiver *receiver, SFLExtended_nat *nat)
341 {
342     putAddress(receiver, &nat->src);
343     putAddress(receiver, &nat->dst);
344 }
345
346 inline static u_int32_t natEncodingLength(SFLExtended_nat *nat) {
347     return addressEncodingLength(&nat->src)
348         + addressEncodingLength(&nat->dst);
349 }
350
351 inline static void putMplsTunnel(SFLReceiver *receiver, SFLExtended_mpls_tunnel *tunnel)
352 {
353     putString(receiver, &tunnel->tunnel_lsp_name);
354     putNet32(receiver, tunnel->tunnel_id);
355     putNet32(receiver, tunnel->tunnel_cos);
356 }
357
358 inline static u_int32_t mplsTunnelEncodingLength(SFLExtended_mpls_tunnel *tunnel) {
359     return stringEncodingLength(&tunnel->tunnel_lsp_name) + 8;
360 }
361
362 inline static void putMplsVc(SFLReceiver *receiver, SFLExtended_mpls_vc *vc)
363 {
364     putString(receiver, &vc->vc_instance_name);
365     putNet32(receiver, vc->vll_vc_id);
366     putNet32(receiver, vc->vc_label_cos);
367 }
368
369 inline static u_int32_t mplsVcEncodingLength(SFLExtended_mpls_vc *vc) {
370     return stringEncodingLength( &vc->vc_instance_name) + 8;
371 }
372
373 inline static void putMplsFtn(SFLReceiver *receiver, SFLExtended_mpls_FTN *ftn)
374 {
375     putString(receiver, &ftn->mplsFTNDescr);
376     putNet32(receiver, ftn->mplsFTNMask);
377 }
378
379 inline static u_int32_t mplsFtnEncodingLength(SFLExtended_mpls_FTN *ftn) {
380     return stringEncodingLength( &ftn->mplsFTNDescr) + 4;
381 }
382
383 inline static void putMplsLdpFec(SFLReceiver *receiver, SFLExtended_mpls_LDP_FEC *ldpfec)
384 {
385     putNet32(receiver, ldpfec->mplsFecAddrPrefixLength);
386 }
387
388 inline static u_int32_t mplsLdpFecEncodingLength(SFLExtended_mpls_LDP_FEC *ldpfec) {
389     return 4;
390 }
391
392 inline static void putVlanTunnel(SFLReceiver *receiver, SFLExtended_vlan_tunnel *vlanTunnel)
393 {
394     putLabelStack(receiver, &vlanTunnel->stack);
395 }
396
397 inline static u_int32_t vlanTunnelEncodingLength(SFLExtended_vlan_tunnel *vlanTunnel) {
398     return labelStackEncodingLength(&vlanTunnel->stack);
399 }
400
401
402 inline static void putGenericCounters(SFLReceiver *receiver, SFLIf_counters *counters)
403 {
404     putNet32(receiver, counters->ifIndex);
405     putNet32(receiver, counters->ifType);
406     putNet64(receiver, counters->ifSpeed);
407     putNet32(receiver, counters->ifDirection);
408     putNet32(receiver, counters->ifStatus);
409     putNet64(receiver, counters->ifInOctets);
410     putNet32(receiver, counters->ifInUcastPkts);
411     putNet32(receiver, counters->ifInMulticastPkts);
412     putNet32(receiver, counters->ifInBroadcastPkts);
413     putNet32(receiver, counters->ifInDiscards);
414     putNet32(receiver, counters->ifInErrors);
415     putNet32(receiver, counters->ifInUnknownProtos);
416     putNet64(receiver, counters->ifOutOctets);
417     putNet32(receiver, counters->ifOutUcastPkts);
418     putNet32(receiver, counters->ifOutMulticastPkts);
419     putNet32(receiver, counters->ifOutBroadcastPkts);
420     putNet32(receiver, counters->ifOutDiscards);
421     putNet32(receiver, counters->ifOutErrors);
422     putNet32(receiver, counters->ifPromiscuousMode);
423 }
424
425
426 /*_________________-----------------------------__________________
427   _________________      computeFlowSampleSize  __________________
428   -----------------_____________________________------------------
429 */
430
431 static int computeFlowSampleSize(SFLReceiver *receiver, SFL_FLOW_SAMPLE_TYPE *fs)
432 {
433     SFLFlow_sample_element *elem = fs->elements;
434 #ifdef SFL_USE_32BIT_INDEX
435     u_int siz = 52; /* tag, length, sequence_number, ds_class, ds_index, sampling_rate,
436                        sample_pool, drops, inputFormat, input, outputFormat, output, number of elements */
437 #else
438     u_int siz = 40; /* tag, length, sequence_number, source_id, sampling_rate,
439                        sample_pool, drops, input, output, number of elements */
440 #endif
441
442     fs->num_elements = 0; /* we're going to count them again even if this was set by the client */
443     for(; elem != NULL; elem = elem->nxt) {
444         u_int elemSiz = 0;
445         fs->num_elements++;
446         siz += 8; /* tag, length */
447         switch(elem->tag) {
448         case SFLFLOW_HEADER:
449             elemSiz = 16; /* header_protocol, frame_length, stripped, header_length */
450             elemSiz += ((elem->flowType.header.header_length + 3) / 4) * 4; /* header, rounded up to nearest 4 bytes */
451             break;
452         case SFLFLOW_ETHERNET: elemSiz = sizeof(SFLSampled_ethernet); break;
453         case SFLFLOW_IPV4: elemSiz = sizeof(SFLSampled_ipv4); break;
454         case SFLFLOW_IPV6: elemSiz = sizeof(SFLSampled_ipv6); break;
455         case SFLFLOW_EX_SWITCH: elemSiz = sizeof(SFLExtended_switch); break;
456         case SFLFLOW_EX_ROUTER: elemSiz = routerEncodingLength(&elem->flowType.router); break;
457         case SFLFLOW_EX_GATEWAY: elemSiz = gatewayEncodingLength(&elem->flowType.gateway); break;
458         case SFLFLOW_EX_USER: elemSiz = userEncodingLength(&elem->flowType.user); break;
459         case SFLFLOW_EX_URL: elemSiz = urlEncodingLength(&elem->flowType.url); break;
460         case SFLFLOW_EX_MPLS: elemSiz = mplsEncodingLength(&elem->flowType.mpls); break;
461         case SFLFLOW_EX_NAT: elemSiz = natEncodingLength(&elem->flowType.nat); break;
462         case SFLFLOW_EX_MPLS_TUNNEL: elemSiz = mplsTunnelEncodingLength(&elem->flowType.mpls_tunnel); break;
463         case SFLFLOW_EX_MPLS_VC: elemSiz = mplsVcEncodingLength(&elem->flowType.mpls_vc); break;
464         case SFLFLOW_EX_MPLS_FTN: elemSiz = mplsFtnEncodingLength(&elem->flowType.mpls_ftn); break;
465         case SFLFLOW_EX_MPLS_LDP_FEC: elemSiz = mplsLdpFecEncodingLength(&elem->flowType.mpls_ldp_fec); break;
466         case SFLFLOW_EX_VLAN_TUNNEL: elemSiz = vlanTunnelEncodingLength(&elem->flowType.vlan_tunnel); break;
467         default:
468             sflError(receiver, "unexpected packet_data_tag");
469             return -1;
470             break;
471         }
472         // cache the element size, and accumulate it into the overall FlowSample size
473         elem->length = elemSiz;
474         siz += elemSiz;
475     }
476
477     return siz;
478 }
479
480 /*_________________-------------------------------__________________
481   _________________ sfl_receiver_writeFlowSample  __________________
482   -----------------_______________________________------------------
483 */
484
485 int sfl_receiver_writeFlowSample(SFLReceiver *receiver, SFL_FLOW_SAMPLE_TYPE *fs)
486 {
487     int packedSize;
488     if(fs == NULL) return -1;
489     if((packedSize = computeFlowSampleSize(receiver, fs)) == -1) return -1;
490
491     // check in case this one sample alone is too big for the datagram
492     // in fact - if it is even half as big then we should ditch it. Very
493     // important to avoid overruning the packet buffer.
494     if(packedSize > (int)(receiver->sFlowRcvrMaximumDatagramSize / 2)) {
495         sflError(receiver, "flow sample too big for datagram");
496         return -1;
497     }
498
499     // if the sample pkt is full enough so that this sample might put
500     // it over the limit, then we should send it now before going on.
501     if((receiver->sampleCollector.pktlen + packedSize) >= receiver->sFlowRcvrMaximumDatagramSize)
502         sendSample(receiver);
503
504     receiver->sampleCollector.numSamples++;
505
506 #ifdef SFL_USE_32BIT_INDEX
507     putNet32(receiver, SFLFLOW_SAMPLE_EXPANDED);
508 #else
509     putNet32(receiver, SFLFLOW_SAMPLE);
510 #endif
511
512     putNet32(receiver, packedSize - 8); // don't include tag and len
513     putNet32(receiver, fs->sequence_number);
514
515 #ifdef SFL_USE_32BIT_INDEX
516     putNet32(receiver, fs->ds_class);
517     putNet32(receiver, fs->ds_index);
518 #else
519     putNet32(receiver, fs->source_id);
520 #endif
521
522     putNet32(receiver, fs->sampling_rate);
523     putNet32(receiver, fs->sample_pool);
524     putNet32(receiver, fs->drops);
525
526 #ifdef SFL_USE_32BIT_INDEX
527     putNet32(receiver, fs->inputFormat);
528     putNet32(receiver, fs->input);
529     putNet32(receiver, fs->outputFormat);
530     putNet32(receiver, fs->output);
531 #else
532     putNet32(receiver, fs->input);
533     putNet32(receiver, fs->output);
534 #endif
535
536     putNet32(receiver, fs->num_elements);
537
538     {
539         SFLFlow_sample_element *elem = fs->elements;
540         for(; elem != NULL; elem = elem->nxt) {
541
542             putNet32(receiver, elem->tag);
543             putNet32(receiver, elem->length); // length cached in computeFlowSampleSize()
544
545             switch(elem->tag) {
546             case SFLFLOW_HEADER:
547                 putNet32(receiver, elem->flowType.header.header_protocol);
548                 putNet32(receiver, elem->flowType.header.frame_length);
549                 putNet32(receiver, elem->flowType.header.stripped);
550                 putNet32(receiver, elem->flowType.header.header_length);
551                 /* the header */
552                 memcpy(receiver->sampleCollector.datap, elem->flowType.header.header_bytes, elem->flowType.header.header_length);
553                 /* round up to multiple of 4 to preserve alignment */
554                 receiver->sampleCollector.datap += ((elem->flowType.header.header_length + 3) / 4);
555                 break;
556             case SFLFLOW_ETHERNET:
557                 putNet32(receiver, elem->flowType.ethernet.eth_len);
558                 putMACAddress(receiver, elem->flowType.ethernet.src_mac);
559                 putMACAddress(receiver, elem->flowType.ethernet.dst_mac);
560                 putNet32(receiver, elem->flowType.ethernet.eth_type);
561                 break;
562             case SFLFLOW_IPV4:
563                 putNet32(receiver, elem->flowType.ipv4.length);
564                 putNet32(receiver, elem->flowType.ipv4.protocol);
565                 put32(receiver, elem->flowType.ipv4.src_ip.addr);
566                 put32(receiver, elem->flowType.ipv4.dst_ip.addr);
567                 putNet32(receiver, elem->flowType.ipv4.src_port);
568                 putNet32(receiver, elem->flowType.ipv4.dst_port);
569                 putNet32(receiver, elem->flowType.ipv4.tcp_flags);
570                 putNet32(receiver, elem->flowType.ipv4.tos);
571                 break;
572             case SFLFLOW_IPV6:
573                 putNet32(receiver, elem->flowType.ipv6.length);
574                 putNet32(receiver, elem->flowType.ipv6.protocol);
575                 put128(receiver, elem->flowType.ipv6.src_ip.addr);
576                 put128(receiver, elem->flowType.ipv6.dst_ip.addr);
577                 putNet32(receiver, elem->flowType.ipv6.src_port);
578                 putNet32(receiver, elem->flowType.ipv6.dst_port);
579                 putNet32(receiver, elem->flowType.ipv6.tcp_flags);
580                 putNet32(receiver, elem->flowType.ipv6.priority);
581                 break;
582             case SFLFLOW_EX_SWITCH: putSwitch(receiver, &elem->flowType.sw); break;
583             case SFLFLOW_EX_ROUTER: putRouter(receiver, &elem->flowType.router); break;
584             case SFLFLOW_EX_GATEWAY: putGateway(receiver, &elem->flowType.gateway); break;
585             case SFLFLOW_EX_USER: putUser(receiver, &elem->flowType.user); break;
586             case SFLFLOW_EX_URL: putUrl(receiver, &elem->flowType.url); break;
587             case SFLFLOW_EX_MPLS: putMpls(receiver, &elem->flowType.mpls); break;
588             case SFLFLOW_EX_NAT: putNat(receiver, &elem->flowType.nat); break;
589             case SFLFLOW_EX_MPLS_TUNNEL: putMplsTunnel(receiver, &elem->flowType.mpls_tunnel); break;
590             case SFLFLOW_EX_MPLS_VC: putMplsVc(receiver, &elem->flowType.mpls_vc); break;
591             case SFLFLOW_EX_MPLS_FTN: putMplsFtn(receiver, &elem->flowType.mpls_ftn); break;
592             case SFLFLOW_EX_MPLS_LDP_FEC: putMplsLdpFec(receiver, &elem->flowType.mpls_ldp_fec); break;
593             case SFLFLOW_EX_VLAN_TUNNEL: putVlanTunnel(receiver, &elem->flowType.vlan_tunnel); break;
594             default:
595                 sflError(receiver, "unexpected packet_data_tag");
596                 return -1;
597                 break;
598             }
599         }
600     }
601
602     // sanity check
603     assert(((u_char *)receiver->sampleCollector.datap
604             - (u_char *)receiver->sampleCollector.data
605             - receiver->sampleCollector.pktlen)  == (u_int32_t)packedSize);
606
607     // update the pktlen
608     receiver->sampleCollector.pktlen = (u_char *)receiver->sampleCollector.datap - (u_char *)receiver->sampleCollector.data;
609     return packedSize;
610 }
611
612 /*_________________-----------------------------__________________
613   _________________ computeCountersSampleSize   __________________
614   -----------------_____________________________------------------
615 */
616
617 static int computeCountersSampleSize(SFLReceiver *receiver, SFL_COUNTERS_SAMPLE_TYPE *cs)
618 {
619     SFLCounters_sample_element *elem = cs->elements;
620 #ifdef SFL_USE_32BIT_INDEX
621     u_int siz = 24; /* tag, length, sequence_number, ds_class, ds_index, number of elements */
622 #else
623     u_int siz = 20; /* tag, length, sequence_number, source_id, number of elements */
624 #endif
625
626     cs->num_elements = 0; /* we're going to count them again even if this was set by the client */
627     for(; elem != NULL; elem = elem->nxt) {
628         u_int elemSiz = 0;
629         cs->num_elements++;
630         siz += 8; /* tag, length */
631         switch(elem->tag) {
632         case SFLCOUNTERS_GENERIC:  elemSiz = sizeof(elem->counterBlock.generic); break;
633         case SFLCOUNTERS_ETHERNET: elemSiz = sizeof(elem->counterBlock.ethernet); break;
634         case SFLCOUNTERS_TOKENRING: elemSiz = sizeof(elem->counterBlock.tokenring); break;
635         case SFLCOUNTERS_VG: elemSiz = sizeof(elem->counterBlock.vg); break;
636         case SFLCOUNTERS_VLAN: elemSiz = sizeof(elem->counterBlock.vlan); break;
637         default:
638             sflError(receiver, "unexpected counters_tag");
639             return -1;
640             break;
641         }
642         // cache the element size, and accumulate it into the overall FlowSample size
643         elem->length = elemSiz;
644         siz += elemSiz;
645     }
646     return siz;
647 }
648
649 /*_________________----------------------------------__________________
650   _________________ sfl_receiver_writeCountersSample __________________
651   -----------------__________________________________------------------
652 */
653
654 int sfl_receiver_writeCountersSample(SFLReceiver *receiver, SFL_COUNTERS_SAMPLE_TYPE *cs)
655 {
656     int packedSize;
657     if(cs == NULL) return -1;
658     // if the sample pkt is full enough so that this sample might put
659     // it over the limit, then we should send it now.
660     if((packedSize = computeCountersSampleSize(receiver, cs)) == -1) return -1;
661
662     // check in case this one sample alone is too big for the datagram
663     // in fact - if it is even half as big then we should ditch it. Very
664     // important to avoid overruning the packet buffer.
665     if(packedSize > (int)(receiver->sFlowRcvrMaximumDatagramSize / 2)) {
666         sflError(receiver, "counters sample too big for datagram");
667         return -1;
668     }
669
670     if((receiver->sampleCollector.pktlen + packedSize) >= receiver->sFlowRcvrMaximumDatagramSize)
671         sendSample(receiver);
672
673     receiver->sampleCollector.numSamples++;
674
675 #ifdef SFL_USE_32BIT_INDEX
676     putNet32(receiver, SFLCOUNTERS_SAMPLE_EXPANDED);
677 #else
678     putNet32(receiver, SFLCOUNTERS_SAMPLE);
679 #endif
680
681     putNet32(receiver, packedSize - 8); // tag and length not included
682     putNet32(receiver, cs->sequence_number);
683
684 #ifdef SFL_USE_32BIT_INDEX
685     putNet32(receiver, cs->ds_class);
686     putNet32(receiver, cs->ds_index);
687 #else
688     putNet32(receiver, cs->source_id);
689 #endif
690
691     putNet32(receiver, cs->num_elements);
692
693     {
694         SFLCounters_sample_element *elem = cs->elements;
695         for(; elem != NULL; elem = elem->nxt) {
696
697             putNet32(receiver, elem->tag);
698             putNet32(receiver, elem->length); // length cached in computeCountersSampleSize()
699
700             switch(elem->tag) {
701             case SFLCOUNTERS_GENERIC:
702                 putGenericCounters(receiver, &(elem->counterBlock.generic));
703                 break;
704             case SFLCOUNTERS_ETHERNET:
705                 // all these counters are 32-bit
706                 putNet32_run(receiver, &elem->counterBlock.ethernet, sizeof(elem->counterBlock.ethernet) / 4);
707                 break;
708             case SFLCOUNTERS_TOKENRING:
709                 // all these counters are 32-bit
710                 putNet32_run(receiver, &elem->counterBlock.tokenring, sizeof(elem->counterBlock.tokenring) / 4);
711                 break;
712             case SFLCOUNTERS_VG:
713                 // mixed sizes
714                 putNet32(receiver, elem->counterBlock.vg.dot12InHighPriorityFrames);
715                 putNet64(receiver, elem->counterBlock.vg.dot12InHighPriorityOctets);
716                 putNet32(receiver, elem->counterBlock.vg.dot12InNormPriorityFrames);
717                 putNet64(receiver, elem->counterBlock.vg.dot12InNormPriorityOctets);
718                 putNet32(receiver, elem->counterBlock.vg.dot12InIPMErrors);
719                 putNet32(receiver, elem->counterBlock.vg.dot12InOversizeFrameErrors);
720                 putNet32(receiver, elem->counterBlock.vg.dot12InDataErrors);
721                 putNet32(receiver, elem->counterBlock.vg.dot12InNullAddressedFrames);
722                 putNet32(receiver, elem->counterBlock.vg.dot12OutHighPriorityFrames);
723                 putNet64(receiver, elem->counterBlock.vg.dot12OutHighPriorityOctets);
724                 putNet32(receiver, elem->counterBlock.vg.dot12TransitionIntoTrainings);
725                 putNet64(receiver, elem->counterBlock.vg.dot12HCInHighPriorityOctets);
726                 putNet64(receiver, elem->counterBlock.vg.dot12HCInNormPriorityOctets);
727                 putNet64(receiver, elem->counterBlock.vg.dot12HCOutHighPriorityOctets);
728                 break;
729             case SFLCOUNTERS_VLAN:
730                 // mixed sizes
731                 putNet32(receiver, elem->counterBlock.vlan.vlan_id);
732                 putNet64(receiver, elem->counterBlock.vlan.octets);
733                 putNet32(receiver, elem->counterBlock.vlan.ucastPkts);
734                 putNet32(receiver, elem->counterBlock.vlan.multicastPkts);
735                 putNet32(receiver, elem->counterBlock.vlan.broadcastPkts);
736                 putNet32(receiver, elem->counterBlock.vlan.discards);
737                 break;
738             default:
739                 sflError(receiver, "unexpected counters_tag");
740                 return -1;
741                 break;
742             }
743         }
744     }
745     // sanity check
746     assert(((u_char *)receiver->sampleCollector.datap
747             - (u_char *)receiver->sampleCollector.data
748             - receiver->sampleCollector.pktlen)  == (u_int32_t)packedSize);
749
750     // update the pktlen
751     receiver->sampleCollector.pktlen = (u_char *)receiver->sampleCollector.datap - (u_char *)receiver->sampleCollector.data;
752     return packedSize;
753 }
754
755 /*_________________---------------------------------__________________
756   _________________ sfl_receiver_samplePacketsSent  __________________
757   -----------------_________________________________------------------
758 */
759
760 u_int32_t sfl_receiver_samplePacketsSent(SFLReceiver *receiver)
761 {
762     return receiver->sampleCollector.packetSeqNo;
763 }
764
765 /*_________________---------------------------__________________
766   _________________     sendSample            __________________
767   -----------------___________________________------------------
768 */
769
770 static void sendSample(SFLReceiver *receiver)
771 {
772     /* construct and send out the sample, then reset for the next one... */
773     /* first fill in the header with the latest values */
774     /* version, agent_address and sub_agent_id were pre-set. */
775     u_int32_t hdrIdx = (receiver->agent->myIP.type == SFLADDRESSTYPE_IP_V6) ? 7 : 4;
776     receiver->sampleCollector.data[hdrIdx++] = htonl(++receiver->sampleCollector.packetSeqNo); /* seq no */
777     receiver->sampleCollector.data[hdrIdx++] = htonl((receiver->agent->now - receiver->agent->bootTime) * 1000); /* uptime */
778     receiver->sampleCollector.data[hdrIdx++] = htonl(receiver->sampleCollector.numSamples); /* num samples */
779     /* send */
780     if(receiver->agent->sendFn) (*receiver->agent->sendFn)(receiver->agent->magic,
781                                                            receiver->agent,
782                                                            receiver,
783                                                            (u_char *)receiver->sampleCollector.data,
784                                                            receiver->sampleCollector.pktlen);
785     else {
786 #ifdef SFLOW_DO_SOCKET
787         /* send it myself */
788         if (receiver->sFlowRcvrAddress.type == SFLADDRESSTYPE_IP_V6) {
789             u_int32_t soclen = sizeof(struct sockaddr_in6);
790             int result = sendto(receiver->agent->receiverSocket6,
791                                 receiver->sampleCollector.data,
792                                 receiver->sampleCollector.pktlen,
793                                 0,
794                                 (struct sockaddr *)&receiver->receiver6,
795                                 soclen);
796             if(result == -1 && errno != EINTR) sfl_agent_sysError(receiver->agent, "receiver", "IPv6 socket sendto error");
797             if(result == 0) sfl_agent_error(receiver->agent, "receiver", "IPv6 socket sendto returned 0");
798         }
799         else {
800             u_int32_t soclen = sizeof(struct sockaddr_in);
801             int result = sendto(receiver->agent->receiverSocket4,
802                                 receiver->sampleCollector.data,
803                                 receiver->sampleCollector.pktlen,
804                                 0,
805                                 (struct sockaddr *)&receiver->receiver4,
806                                 soclen);
807             if(result == -1 && errno != EINTR) sfl_agent_sysError(receiver->agent, "receiver", "socket sendto error");
808             if(result == 0) sfl_agent_error(receiver->agent, "receiver", "socket sendto returned 0");
809         }
810 #endif
811     }
812
813     /* reset for the next time */
814     resetSampleCollector(receiver);
815 }
816
817 /*_________________---------------------------__________________
818   _________________   resetSampleCollector    __________________
819   -----------------___________________________------------------
820 */
821
822 static void resetSampleCollector(SFLReceiver *receiver)
823 {
824     receiver->sampleCollector.pktlen = 0;
825     receiver->sampleCollector.numSamples = 0;
826     /* point the datap to just after the header */
827     receiver->sampleCollector.datap = (receiver->agent->myIP.type == SFLADDRESSTYPE_IP_V6) ?
828         (receiver->sampleCollector.data + 10) :  (receiver->sampleCollector.data + 7);
829
830     receiver->sampleCollector.pktlen = (u_char *)receiver->sampleCollector.datap - (u_char *)receiver->sampleCollector.data;
831 }
832
833 /*_________________---------------------------__________________
834   _________________         sflError          __________________
835   -----------------___________________________------------------
836 */
837
838 static void sflError(SFLReceiver *receiver, char *msg)
839 {
840     sfl_agent_error(receiver->agent, "receiver", msg);
841     resetSampleCollector(receiver);
842 }
843
844 #endif  /* !__CHECKER__ */