mir/ovs
2
0
Fork 0
mirror of https://github.com/openvswitch/ovs synced 2025-09-01 06:45:17 +00:00
Code Issues Releases Wiki Activity

Add InMon's sFlow Agent library to the build system.

Browse Source 
The C source and header files added in this commit is covered under the
InMon sFlow license at http://www.inmon.com/technology/sflowlicense.txt

The library requires -Wno-unused to compile without warnings, so this
commit adds that for building the sFlow code only.  Automake can only
change compiler flags on a per-library or per-program basis, so sFlow
is built as a separate library.

The library will be used in upcoming commits.
This commit is contained in:
Ben Pfaff
2009-11-20 09:45:26 -08:00
parent e4bfff8f0d
commit c72e245a0e
10 changed files with 2565 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
COPYING
View File

@@ -22,3 +22,7 @@ Public License, version 2.
Files under the xenserver directory are licensed on a file-by-file
basis. Some files are under an uncertain license that may not be
DFSG-compliant or GPL-compatible. Refer to each file for details.
Files lib/sflow*.[ch] are licensed under the terms of the InMon sFlow
licence that is available at:
http://www.inmon.com/technology/sflowlicense.txt

10
acinclude.m4
View File

@@ -235,4 +235,14 @@ dnl Example: OVS_ENABLE_OPTION([-Wdeclaration-after-statement])
AC_DEFUN([OVS_ENABLE_OPTION],
[OVS_CHECK_CC_OPTION([$1], [WARNING_FLAGS="$WARNING_FLAGS $1"])
AC_SUBST([WARNING_FLAGS])])
dnl OVS_CONDITIONAL_CC_OPTION([OPTION], [CONDITIONAL])
dnl Check whether the given C compiler OPTION is accepted.
dnl If so, enable the given Automake CONDITIONAL.
dnl Example: OVS_CONDITIONAL_CC_OPTION([-Wno-unused], [HAVE_WNO_UNUSED])
AC_DEFUN([OVS_CONDITIONAL_CC_OPTION],
[OVS_CHECK_CC_OPTION(
[$1], [ovs_have_cc_option=yes], [ovs_have_cc_option=no])
AM_CONDITIONAL([$2], [test $ovs_have_cc_option = yes])])
dnl ----------------------------------------------------------------------

1
configure.ac
View File

@@ -75,6 +75,7 @@ OVS_ENABLE_OPTION([-Wold-style-definition])
OVS_ENABLE_OPTION([-Wmissing-prototypes])
OVS_ENABLE_OPTION([-Wmissing-field-initializers])
OVS_ENABLE_OPTION([-Wno-override-init])
OVS_CONDITIONAL_CC_OPTION([-Wno-unused], [HAVE_WNO_UNUSED])
AC_ARG_VAR(KARCH, [Kernel Architecture String])
AC_SUBST(KARCH)

13
lib/automake.mk
View File

@@ -125,6 +125,19 @@ nodist_lib_libopenvswitch_a_SOURCES = \
lib/dirs.c
CLEANFILES += $(nodist_lib_libopenvswitch_a_SOURCES)
noinst_LIBRARIES += lib/libsflow.a
lib_libsflow_a_SOURCES = \
lib/sflow_api.h \
lib/sflow.h \
lib/sflow_agent.c \
lib/sflow_sampler.c \
lib/sflow_poller.c \
lib/sflow_receiver.c
lib_libsflow_a_CFLAGS = $(AM_CFLAGS)
if HAVE_WNO_UNUSED
lib_libsflow_a_CFLAGS += -Wno-unused
endif
if HAVE_NETLINK
lib_libopenvswitch_a_SOURCES += \
lib/netlink-protocol.h \

548
lib/sflow.h Normal file
View File

@@ -0,0 +1,548 @@
/* Copyright (c) 2002-2009 InMon Corp. Licensed under the terms of the InMon sFlow licence: */
/* http://www.inmon.com/technology/sflowlicense.txt */
#ifndef SFLOW_H
#define SFLOW_H 1
enum SFLAddress_type {
SFLADDRESSTYPE_IP_V4 = 1,
SFLADDRESSTYPE_IP_V6 = 2
};
typedef struct {
u_int32_t addr;
} SFLIPv4;
typedef struct {
u_char addr[16];
} SFLIPv6;
typedef union _SFLAddress_value {
SFLIPv4 ip_v4;
SFLIPv6 ip_v6;
} SFLAddress_value;
typedef struct _SFLAddress {
u_int32_t type; /* enum SFLAddress_type */
SFLAddress_value address;
} SFLAddress;
/* Packet header data */
#define SFL_DEFAULT_HEADER_SIZE 128
#define SFL_DEFAULT_COLLECTOR_PORT 6343
#define SFL_DEFAULT_SAMPLING_RATE 400
#define SFL_DEFAULT_POLLING_INTERVAL 30
/* The header protocol describes the format of the sampled header */
enum SFLHeader_protocol {
SFLHEADER_ETHERNET_ISO8023 = 1,
SFLHEADER_ISO88024_TOKENBUS = 2,
SFLHEADER_ISO88025_TOKENRING = 3,
SFLHEADER_FDDI = 4,
SFLHEADER_FRAME_RELAY = 5,
SFLHEADER_X25 = 6,
SFLHEADER_PPP = 7,
SFLHEADER_SMDS = 8,
SFLHEADER_AAL5 = 9,
SFLHEADER_AAL5_IP = 10, /* e.g. Cisco AAL5 mux */
SFLHEADER_IPv4 = 11,
SFLHEADER_IPv6 = 12,
SFLHEADER_MPLS = 13
};
/* raw sampled header */
typedef struct _SFLSampled_header {
u_int32_t header_protocol; /* (enum SFLHeader_protocol) */
u_int32_t frame_length; /* Original length of packet before sampling */
u_int32_t stripped; /* header/trailer bytes stripped by sender */
u_int32_t header_length; /* length of sampled header bytes to follow */
u_int8_t *header_bytes; /* Header bytes */
} SFLSampled_header;
/* decoded ethernet header */
typedef struct _SFLSampled_ethernet {
u_int32_t eth_len; /* The length of the MAC packet excluding
lower layer encapsulations */
u_int8_t src_mac[8]; /* 6 bytes + 2 pad */
u_int8_t dst_mac[8];
u_int32_t eth_type;
} SFLSampled_ethernet;
/* decoded IP version 4 header */
typedef struct _SFLSampled_ipv4 {
u_int32_t length; /* The length of the IP packet
excluding lower layer encapsulations */
u_int32_t protocol; /* IP Protocol type (for example, TCP = 6, UDP = 17) */
SFLIPv4 src_ip; /* Source IP Address */
SFLIPv4 dst_ip; /* Destination IP Address */
u_int32_t src_port; /* TCP/UDP source port number or equivalent */
u_int32_t dst_port; /* TCP/UDP destination port number or equivalent */
u_int32_t tcp_flags; /* TCP flags */
u_int32_t tos; /* IP type of service */
} SFLSampled_ipv4;
/* decoded IP version 6 data */
typedef struct _SFLSampled_ipv6 {
u_int32_t length; /* The length of the IP packet
excluding lower layer encapsulations */
u_int32_t protocol; /* IP Protocol type (for example, TCP = 6, UDP = 17) */
SFLIPv6 src_ip; /* Source IP Address */
SFLIPv6 dst_ip; /* Destination IP Address */
u_int32_t src_port; /* TCP/UDP source port number or equivalent */
u_int32_t dst_port; /* TCP/UDP destination port number or equivalent */
u_int32_t tcp_flags; /* TCP flags */
u_int32_t priority; /* IP priority */
} SFLSampled_ipv6;
/* Extended data types */
/* Extended switch data */
typedef struct _SFLExtended_switch {
u_int32_t src_vlan; /* The 802.1Q VLAN id of incomming frame */
u_int32_t src_priority; /* The 802.1p priority */
u_int32_t dst_vlan; /* The 802.1Q VLAN id of outgoing frame */
u_int32_t dst_priority; /* The 802.1p priority */
} SFLExtended_switch;
/* Extended router data */
typedef struct _SFLExtended_router {
SFLAddress nexthop; /* IP address of next hop router */
u_int32_t src_mask; /* Source address prefix mask bits */
u_int32_t dst_mask; /* Destination address prefix mask bits */
} SFLExtended_router;
/* Extended gateway data */
enum SFLExtended_as_path_segment_type {
SFLEXTENDED_AS_SET = 1, /* Unordered set of ASs */
SFLEXTENDED_AS_SEQUENCE = 2 /* Ordered sequence of ASs */
};
typedef struct _SFLExtended_as_path_segment {
u_int32_t type; /* enum SFLExtended_as_path_segment_type */
u_int32_t length; /* number of AS numbers in set/sequence */
union {
u_int32_t *set;
u_int32_t *seq;
} as;
} SFLExtended_as_path_segment;
typedef struct _SFLExtended_gateway {
SFLAddress nexthop; /* Address of the border router that should
be used for the destination network */
u_int32_t as; /* AS number for this gateway */
u_int32_t src_as; /* AS number of source (origin) */
u_int32_t src_peer_as; /* AS number of source peer */
u_int32_t dst_as_path_segments; /* number of segments in path */
SFLExtended_as_path_segment *dst_as_path; /* list of seqs or sets */
u_int32_t communities_length; /* number of communities */
u_int32_t *communities; /* set of communities */
u_int32_t localpref; /* LocalPref associated with this route */
} SFLExtended_gateway;
typedef struct _SFLString {
u_int32_t len;
char *str;
} SFLString;
/* Extended user data */
typedef struct _SFLExtended_user {
u_int32_t src_charset; /* MIBEnum value of character set used to encode a string - See RFC 2978
Where possible UTF-8 encoding (MIBEnum=106) should be used. A value
of zero indicates an unknown encoding. */
SFLString src_user;
u_int32_t dst_charset;
SFLString dst_user;
} SFLExtended_user;
/* Extended URL data */
enum SFLExtended_url_direction {
SFLEXTENDED_URL_SRC = 1, /* URL is associated with source address */
SFLEXTENDED_URL_DST = 2 /* URL is associated with destination address */
};
typedef struct _SFLExtended_url {
u_int32_t direction; /* enum SFLExtended_url_direction */
SFLString url; /* URL associated with the packet flow.
Must be URL encoded */
SFLString host; /* The host field from the HTTP header */
} SFLExtended_url;
/* Extended MPLS data */
typedef struct _SFLLabelStack {
u_int32_t depth;
u_int32_t *stack; /* first entry is top of stack - see RFC 3032 for encoding */
} SFLLabelStack;
typedef struct _SFLExtended_mpls {
SFLAddress nextHop; /* Address of the next hop */
SFLLabelStack in_stack;
SFLLabelStack out_stack;
} SFLExtended_mpls;
/* Extended NAT data
Packet header records report addresses as seen at the sFlowDataSource.
The extended_nat structure reports on translated source and/or destination
addesses for this packet. If an address was not translated it should
be equal to that reported for the header. */
typedef struct _SFLExtended_nat {
SFLAddress src; /* Source address */
SFLAddress dst; /* Destination address */
} SFLExtended_nat;
/* additional Extended MPLS stucts */
typedef struct _SFLExtended_mpls_tunnel {
SFLString tunnel_lsp_name; /* Tunnel name */
u_int32_t tunnel_id; /* Tunnel ID */
u_int32_t tunnel_cos; /* Tunnel COS value */
} SFLExtended_mpls_tunnel;
typedef struct _SFLExtended_mpls_vc {
SFLString vc_instance_name; /* VC instance name */
u_int32_t vll_vc_id; /* VLL/VC instance ID */
u_int32_t vc_label_cos; /* VC Label COS value */
} SFLExtended_mpls_vc;
/* Extended MPLS FEC
- Definitions from MPLS-FTN-STD-MIB mplsFTNTable */
typedef struct _SFLExtended_mpls_FTN {
SFLString mplsFTNDescr;
u_int32_t mplsFTNMask;
} SFLExtended_mpls_FTN;
/* Extended MPLS LVP FEC
- Definition from MPLS-LDP-STD-MIB mplsFecTable
Note: mplsFecAddrType, mplsFecAddr information available
from packet header */
typedef struct _SFLExtended_mpls_LDP_FEC {
u_int32_t mplsFecAddrPrefixLength;
} SFLExtended_mpls_LDP_FEC;
/* Extended VLAN tunnel information
Record outer VLAN encapsulations that have
been stripped. extended_vlantunnel information
should only be reported if all the following conditions are satisfied:
1. The packet has nested vlan tags, AND
2. The reporting device is VLAN aware, AND
3. One or more VLAN tags have been stripped, either
because they represent proprietary encapsulations, or
because switch hardware automatically strips the outer VLAN
encapsulation.
Reporting extended_vlantunnel information is not a substitute for
reporting extended_switch information. extended_switch data must
always be reported to describe the ingress/egress VLAN information
for the packet. The extended_vlantunnel information only applies to
nested VLAN tags, and then only when one or more tags has been
stripped. */
typedef SFLLabelStack SFLVlanStack;
typedef struct _SFLExtended_vlan_tunnel {
SFLVlanStack stack; /* List of stripped 802.1Q TPID/TCI layers. Each
TPID,TCI pair is represented as a single 32 bit
integer. Layers listed from outermost to
innermost. */
} SFLExtended_vlan_tunnel;
enum SFLFlow_type_tag {
/* enterprise = 0, format = ... */
SFLFLOW_HEADER = 1, /* Packet headers are sampled */
SFLFLOW_ETHERNET = 2, /* MAC layer information */
SFLFLOW_IPV4 = 3, /* IP version 4 data */
SFLFLOW_IPV6 = 4, /* IP version 6 data */
SFLFLOW_EX_SWITCH = 1001, /* Extended switch information */
SFLFLOW_EX_ROUTER = 1002, /* Extended router information */
SFLFLOW_EX_GATEWAY = 1003, /* Extended gateway router information */
SFLFLOW_EX_USER = 1004, /* Extended TACAS/RADIUS user information */
SFLFLOW_EX_URL = 1005, /* Extended URL information */
SFLFLOW_EX_MPLS = 1006, /* Extended MPLS information */
SFLFLOW_EX_NAT = 1007, /* Extended NAT information */
SFLFLOW_EX_MPLS_TUNNEL = 1008, /* additional MPLS information */
SFLFLOW_EX_MPLS_VC = 1009,
SFLFLOW_EX_MPLS_FTN = 1010,
SFLFLOW_EX_MPLS_LDP_FEC = 1011,
SFLFLOW_EX_VLAN_TUNNEL = 1012, /* VLAN stack */
};
typedef union _SFLFlow_type {
SFLSampled_header header;
SFLSampled_ethernet ethernet;
SFLSampled_ipv4 ipv4;
SFLSampled_ipv6 ipv6;
SFLExtended_switch sw;
SFLExtended_router router;
SFLExtended_gateway gateway;
SFLExtended_user user;
SFLExtended_url url;
SFLExtended_mpls mpls;
SFLExtended_nat nat;
SFLExtended_mpls_tunnel mpls_tunnel;
SFLExtended_mpls_vc mpls_vc;
SFLExtended_mpls_FTN mpls_ftn;
SFLExtended_mpls_LDP_FEC mpls_ldp_fec;
SFLExtended_vlan_tunnel vlan_tunnel;
} SFLFlow_type;
typedef struct _SFLFlow_sample_element {
struct _SFLFlow_sample_element *nxt;
u_int32_t tag; /* SFLFlow_type_tag */
u_int32_t length;
SFLFlow_type flowType;
} SFLFlow_sample_element;
enum SFL_sample_tag {
SFLFLOW_SAMPLE = 1, /* enterprise = 0 : format = 1 */
SFLCOUNTERS_SAMPLE = 2, /* enterprise = 0 : format = 2 */
SFLFLOW_SAMPLE_EXPANDED = 3, /* enterprise = 0 : format = 3 */
SFLCOUNTERS_SAMPLE_EXPANDED = 4 /* enterprise = 0 : format = 4 */
};
/* Format of a single flow sample */
typedef struct _SFLFlow_sample {
/* u_int32_t tag; */ /* SFL_sample_tag -- enterprise = 0 : format = 1 */
/* u_int32_t length; */
u_int32_t sequence_number; /* Incremented with each flow sample
generated */
u_int32_t source_id; /* fsSourceId */
u_int32_t sampling_rate; /* fsPacketSamplingRate */
u_int32_t sample_pool; /* Total number of packets that could have been
sampled (i.e. packets skipped by sampling
process + total number of samples) */
u_int32_t drops; /* Number of times a packet was dropped due to
lack of resources */
u_int32_t input; /* SNMP ifIndex of input interface.
0 if interface is not known. */
u_int32_t output; /* SNMP ifIndex of output interface,
0 if interface is not known.
Set most significant bit to indicate
multiple destination interfaces
(i.e. in case of broadcast or multicast)
and set lower order bits to indicate
number of destination interfaces.
Examples:
0x00000002 indicates ifIndex = 2
0x00000000 ifIndex unknown.
0x80000007 indicates a packet sent
to 7 interfaces.
0x80000000 indicates a packet sent to
an unknown number of
interfaces greater than 1.*/
u_int32_t num_elements;
SFLFlow_sample_element *elements;
} SFLFlow_sample;
/* same thing, but the expanded version (for full 32-bit ifIndex numbers) */
typedef struct _SFLFlow_sample_expanded {
/* u_int32_t tag; */ /* SFL_sample_tag -- enterprise = 0 : format = 1 */
/* u_int32_t length; */
u_int32_t sequence_number; /* Incremented with each flow sample
generated */
u_int32_t ds_class; /* EXPANDED */
u_int32_t ds_index; /* EXPANDED */
u_int32_t sampling_rate; /* fsPacketSamplingRate */
u_int32_t sample_pool; /* Total number of packets that could have been
sampled (i.e. packets skipped by sampling
process + total number of samples) */
u_int32_t drops; /* Number of times a packet was dropped due to
lack of resources */
u_int32_t inputFormat; /* EXPANDED */
u_int32_t input; /* SNMP ifIndex of input interface.
0 if interface is not known. */
u_int32_t outputFormat; /* EXPANDED */
u_int32_t output; /* SNMP ifIndex of output interface,
0 if interface is not known. */
u_int32_t num_elements;
SFLFlow_sample_element *elements;
} SFLFlow_sample_expanded;
/* Counter types */
/* Generic interface counters - see RFC 1573, 2233 */
typedef struct _SFLIf_counters {
u_int32_t ifIndex;
u_int32_t ifType;
u_int64_t ifSpeed;
u_int32_t ifDirection; /* Derived from MAU MIB (RFC 2668)
0 = unknown, 1 = full-duplex,
2 = half-duplex, 3 = in, 4 = out */
u_int32_t ifStatus; /* bit field with the following bits assigned:
bit 0 = ifAdminStatus (0 = down, 1 = up)
bit 1 = ifOperStatus (0 = down, 1 = up) */
u_int64_t ifInOctets;
u_int32_t ifInUcastPkts;
u_int32_t ifInMulticastPkts;
u_int32_t ifInBroadcastPkts;
u_int32_t ifInDiscards;
u_int32_t ifInErrors;
u_int32_t ifInUnknownProtos;
u_int64_t ifOutOctets;
u_int32_t ifOutUcastPkts;
u_int32_t ifOutMulticastPkts;
u_int32_t ifOutBroadcastPkts;
u_int32_t ifOutDiscards;
u_int32_t ifOutErrors;
u_int32_t ifPromiscuousMode;
} SFLIf_counters;
/* Ethernet interface counters - see RFC 2358 */
typedef struct _SFLEthernet_counters {
u_int32_t dot3StatsAlignmentErrors;
u_int32_t dot3StatsFCSErrors;
u_int32_t dot3StatsSingleCollisionFrames;
u_int32_t dot3StatsMultipleCollisionFrames;
u_int32_t dot3StatsSQETestErrors;
u_int32_t dot3StatsDeferredTransmissions;
u_int32_t dot3StatsLateCollisions;
u_int32_t dot3StatsExcessiveCollisions;
u_int32_t dot3StatsInternalMacTransmitErrors;
u_int32_t dot3StatsCarrierSenseErrors;
u_int32_t dot3StatsFrameTooLongs;
u_int32_t dot3StatsInternalMacReceiveErrors;
u_int32_t dot3StatsSymbolErrors;
} SFLEthernet_counters;
/* Token ring counters - see RFC 1748 */
typedef struct _SFLTokenring_counters {
u_int32_t dot5StatsLineErrors;
u_int32_t dot5StatsBurstErrors;
u_int32_t dot5StatsACErrors;
u_int32_t dot5StatsAbortTransErrors;
u_int32_t dot5StatsInternalErrors;
u_int32_t dot5StatsLostFrameErrors;
u_int32_t dot5StatsReceiveCongestions;
u_int32_t dot5StatsFrameCopiedErrors;
u_int32_t dot5StatsTokenErrors;
u_int32_t dot5StatsSoftErrors;
u_int32_t dot5StatsHardErrors;
u_int32_t dot5StatsSignalLoss;
u_int32_t dot5StatsTransmitBeacons;
u_int32_t dot5StatsRecoverys;
u_int32_t dot5StatsLobeWires;
u_int32_t dot5StatsRemoves;
u_int32_t dot5StatsSingles;
u_int32_t dot5StatsFreqErrors;
} SFLTokenring_counters;
/* 100 BaseVG interface counters - see RFC 2020 */
typedef struct _SFLVg_counters {
u_int32_t dot12InHighPriorityFrames;
u_int64_t dot12InHighPriorityOctets;
u_int32_t dot12InNormPriorityFrames;
u_int64_t dot12InNormPriorityOctets;
u_int32_t dot12InIPMErrors;
u_int32_t dot12InOversizeFrameErrors;
u_int32_t dot12InDataErrors;
u_int32_t dot12InNullAddressedFrames;
u_int32_t dot12OutHighPriorityFrames;
u_int64_t dot12OutHighPriorityOctets;
u_int32_t dot12TransitionIntoTrainings;
u_int64_t dot12HCInHighPriorityOctets;
u_int64_t dot12HCInNormPriorityOctets;
u_int64_t dot12HCOutHighPriorityOctets;
} SFLVg_counters;
typedef struct _SFLVlan_counters {
u_int32_t vlan_id;
u_int64_t octets;
u_int32_t ucastPkts;
u_int32_t multicastPkts;
u_int32_t broadcastPkts;
u_int32_t discards;
} SFLVlan_counters;
/* Counters data */
enum SFLCounters_type_tag {
/* enterprise = 0, format = ... */
SFLCOUNTERS_GENERIC = 1,
SFLCOUNTERS_ETHERNET = 2,
SFLCOUNTERS_TOKENRING = 3,
SFLCOUNTERS_VG = 4,
SFLCOUNTERS_VLAN = 5
};
typedef union _SFLCounters_type {
SFLIf_counters generic;
SFLEthernet_counters ethernet;
SFLTokenring_counters tokenring;
SFLVg_counters vg;
SFLVlan_counters vlan;
} SFLCounters_type;
typedef struct _SFLCounters_sample_element {
struct _SFLCounters_sample_element *nxt; /* linked list */
u_int32_t tag; /* SFLCounters_type_tag */
u_int32_t length;
SFLCounters_type counterBlock;
} SFLCounters_sample_element;
typedef struct _SFLCounters_sample {
/* u_int32_t tag; */ /* SFL_sample_tag -- enterprise = 0 : format = 2 */
/* u_int32_t length; */
u_int32_t sequence_number; /* Incremented with each counters sample
generated by this source_id */
u_int32_t source_id; /* fsSourceId */
u_int32_t num_elements;
SFLCounters_sample_element *elements;
} SFLCounters_sample;
/* same thing, but the expanded version, so ds_index can be a full 32 bits */
typedef struct _SFLCounters_sample_expanded {
/* u_int32_t tag; */ /* SFL_sample_tag -- enterprise = 0 : format = 2 */
/* u_int32_t length; */
u_int32_t sequence_number; /* Incremented with each counters sample
generated by this source_id */
u_int32_t ds_class; /* EXPANDED */
u_int32_t ds_index; /* EXPANDED */
u_int32_t num_elements;
SFLCounters_sample_element *elements;
} SFLCounters_sample_expanded;
#define SFLADD_ELEMENT(_sm, _el) do { (_el)->nxt = (_sm)->elements; (_sm)->elements = (_el); } while(0)
/* Format of a sample datagram */
enum SFLDatagram_version {
SFLDATAGRAM_VERSION2 = 2,
SFLDATAGRAM_VERSION4 = 4,
SFLDATAGRAM_VERSION5 = 5
};
typedef struct _SFLSample_datagram_hdr {
u_int32_t datagram_version; /* (enum SFLDatagram_version) = VERSION5 = 5 */
SFLAddress agent_address; /* IP address of sampling agent */
u_int32_t sub_agent_id; /* Used to distinguishing between datagram
streams from separate agent sub entities
within an device. */
u_int32_t sequence_number; /* Incremented with each sample datagram
generated */
u_int32_t uptime; /* Current time (in milliseconds since device
last booted). Should be set as close to
datagram transmission time as possible.*/
u_int32_t num_records; /* Number of tag-len-val flow/counter records to follow */
} SFLSample_datagram_hdr;
#define SFL_MAX_DATAGRAM_SIZE 1500
#define SFL_MIN_DATAGRAM_SIZE 200
#define SFL_DEFAULT_DATAGRAM_SIZE 1400
#define SFL_DATA_PAD 400
#endif /* SFLOW_H */

492
lib/sflow_agent.c Normal file
View File

@@ -0,0 +1,492 @@
/* Copyright (c) 2002-2009 InMon Corp. Licensed under the terms of the InMon sFlow licence: */
/* http://www.inmon.com/technology/sflowlicense.txt */
#include "sflow_api.h"
static void * sflAlloc(SFLAgent *agent, size_t bytes);
static void sflFree(SFLAgent *agent, void *obj);
static void sfl_agent_jumpTableAdd(SFLAgent *agent, SFLSampler *sampler);
static void sfl_agent_jumpTableRemove(SFLAgent *agent, SFLSampler *sampler);
/*________________--------------------------__________________
________________ sfl_agent_init __________________
----------------__________________________------------------
*/
void sfl_agent_init(SFLAgent *agent,
SFLAddress *myIP, /* IP address of this agent in net byte order */
u_int32_t subId, /* agent_sub_id */
time_t bootTime, /* agent boot time */
time_t now, /* time now */
void *magic, /* ptr to pass back in logging and alloc fns */
allocFn_t allocFn,
freeFn_t freeFn,
errorFn_t errorFn,
sendFn_t sendFn)
{
/* first clear everything */
memset(agent, 0, sizeof(*agent));
/* now copy in the parameters */
agent->myIP = *myIP; /* structure copy */
agent->subId = subId;
agent->bootTime = bootTime;
agent->now = now;
agent->magic = magic;
agent->allocFn = allocFn;
agent->freeFn = freeFn;
agent->errorFn = errorFn;
agent->sendFn = sendFn;
#ifdef SFLOW_DO_SOCKET
if(sendFn == NULL) {
/* open the socket - really need one for v4 and another for v6? */
if((agent->receiverSocket4 = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) == -1)
sfl_agent_sysError(agent, "agent", "IPv4 socket open failed");
if((agent->receiverSocket6 = socket(AF_INET6, SOCK_DGRAM, IPPROTO_UDP)) == -1)
sfl_agent_sysError(agent, "agent", "IPv6 socket open failed");
}
#endif
}
/*_________________---------------------------__________________
_________________ sfl_agent_release __________________
-----------------___________________________------------------
*/
void sfl_agent_release(SFLAgent *agent)
{
/* release and free the samplers, pollers and receivers */
SFLSampler *sm = agent->samplers;
SFLPoller *pl = agent->pollers;
SFLReceiver *rcv = agent->receivers;
for(; sm != NULL; ) {
SFLSampler *nextSm = sm->nxt;
sflFree(agent, sm);
sm = nextSm;
}
agent->samplers = NULL;
for(; pl != NULL; ) {
SFLPoller *nextPl = pl->nxt;
sflFree(agent, pl);
pl = nextPl;
}
agent->pollers = NULL;
for(; rcv != NULL; ) {
SFLReceiver *nextRcv = rcv->nxt;
sflFree(agent, rcv);
rcv = nextRcv;
}
agent->receivers = NULL;
#ifdef SFLOW_DO_SOCKET
/* close the sockets */
if(agent->receiverSocket4 > 0) close(agent->receiverSocket4);
if(agent->receiverSocket6 > 0) close(agent->receiverSocket6);
#endif
}
/*_________________---------------------------__________________
_________________ sfl_agent_set_* __________________
-----------------___________________________------------------
*/
void sfl_agent_set_agentAddress(SFLAgent *agent, SFLAddress *addr)
{
if(addr && memcmp(addr, &agent->myIP, sizeof(agent->myIP)) != 0) {
/* change of address */
agent->myIP = *addr; /* structure copy */
/* reset sequence numbers here? */
}
}
void sfl_agent_set_agentSubId(SFLAgent *agent, u_int32_t subId)
{
if(subId != agent->subId) {
/* change of subId */
agent->subId = subId;
/* reset sequence numbers here? */
}
}
/*_________________---------------------------__________________
_________________ sfl_agent_tick __________________
-----------------___________________________------------------
*/
void sfl_agent_tick(SFLAgent *agent, time_t now)
{
SFLReceiver *rcv = agent->receivers;
SFLSampler *sm = agent->samplers;
SFLPoller *pl = agent->pollers;
agent->now = now;
/* receivers use ticks to flush send data */
for(; rcv != NULL; rcv = rcv->nxt) sfl_receiver_tick(rcv, now);
/* samplers use ticks to decide when they are sampling too fast */
for(; sm != NULL; sm = sm->nxt) sfl_sampler_tick(sm, now);
/* pollers use ticks to decide when to ask for counters */
for(; pl != NULL; pl = pl->nxt) sfl_poller_tick(pl, now);
}
/*_________________---------------------------__________________
_________________ sfl_agent_addReceiver __________________
-----------------___________________________------------------
*/
SFLReceiver *sfl_agent_addReceiver(SFLAgent *agent)
{
SFLReceiver *rcv = (SFLReceiver *)sflAlloc(agent, sizeof(SFLReceiver));
sfl_receiver_init(rcv, agent);
/* add to end of list - to preserve the receiver index numbers for existing receivers */
{
SFLReceiver *r, *prev = NULL;
for(r = agent->receivers; r != NULL; prev = r, r = r->nxt);
if(prev) prev->nxt = rcv;
else agent->receivers = rcv;
rcv->nxt = NULL;
}
return rcv;
}
/*_________________---------------------------__________________
_________________ sfl_dsi_compare __________________
-----------------___________________________------------------
Note that if there is a mixture of ds_classes for this agent, then
the simple numeric comparison may not be correct - the sort order (for
the purposes of the SNMP MIB) should really be determined by the OID
that these numeric ds_class numbers are a shorthand for. For example,
ds_class == 0 means ifIndex, which is the oid "1.3.6.1.2.1.2.2.1"
*/
static inline int sfl_dsi_compare(SFLDataSource_instance *pdsi1, SFLDataSource_instance *pdsi2) {
/* could have used just memcmp(), but not sure if that would
give the right answer on little-endian platforms. Safer to be explicit... */
int cmp = pdsi2->ds_class - pdsi1->ds_class;
if(cmp == 0) cmp = pdsi2->ds_index - pdsi1->ds_index;
if(cmp == 0) cmp = pdsi2->ds_instance - pdsi1->ds_instance;
return cmp;
}
/*_________________---------------------------__________________
_________________ sfl_agent_addSampler __________________
-----------------___________________________------------------
*/
SFLSampler *sfl_agent_addSampler(SFLAgent *agent, SFLDataSource_instance *pdsi)
{
/* Keep the list sorted. */
SFLSampler *prev = NULL, *sm = agent->samplers;
for(; sm != NULL; prev = sm, sm = sm->nxt) {
int64_t cmp = sfl_dsi_compare(pdsi, &sm->dsi);
if(cmp == 0) return sm; /* found - return existing one */
if(cmp < 0) break; /* insert here */
}
/* either we found the insert point, or reached the end of the list...*/
{
SFLSampler *newsm = (SFLSampler *)sflAlloc(agent, sizeof(SFLSampler));
sfl_sampler_init(newsm, agent, pdsi);
if(prev) prev->nxt = newsm;
else agent->samplers = newsm;
newsm->nxt = sm;
/* see if we should go in the ifIndex jumpTable */
if(SFL_DS_CLASS(newsm->dsi) == 0) {
SFLSampler *test = sfl_agent_getSamplerByIfIndex(agent, SFL_DS_INDEX(newsm->dsi));
if(test && (SFL_DS_INSTANCE(newsm->dsi) < SFL_DS_INSTANCE(test->dsi))) {
/* replace with this new one because it has a lower ds_instance number */
sfl_agent_jumpTableRemove(agent, test);
test = NULL;
}
if(test == NULL) sfl_agent_jumpTableAdd(agent, newsm);
}
return newsm;
}
}
/*_________________---------------------------__________________
_________________ sfl_agent_addPoller __________________
-----------------___________________________------------------
*/
SFLPoller *sfl_agent_addPoller(SFLAgent *agent,
SFLDataSource_instance *pdsi,
void *magic, /* ptr to pass back in getCountersFn() */
getCountersFn_t getCountersFn)
{
/* keep the list sorted */
SFLPoller *prev = NULL, *pl = agent->pollers;
for(; pl != NULL; prev = pl, pl = pl->nxt) {
int64_t cmp = sfl_dsi_compare(pdsi, &pl->dsi);
if(cmp == 0) return pl; /* found - return existing one */
if(cmp < 0) break; /* insert here */
}
/* either we found the insert point, or reached the end of the list... */
{
SFLPoller *newpl = (SFLPoller *)sflAlloc(agent, sizeof(SFLPoller));
sfl_poller_init(newpl, agent, pdsi, magic, getCountersFn);
if(prev) prev->nxt = newpl;
else agent->pollers = newpl;
newpl->nxt = pl;
return newpl;
}
}
/*_________________---------------------------__________________
_________________ sfl_agent_removeSampler __________________
-----------------___________________________------------------
*/
int sfl_agent_removeSampler(SFLAgent *agent, SFLDataSource_instance *pdsi)
{
/* find it, unlink it and free it */
SFLSampler *prev = NULL, *sm = agent->samplers;
for(; sm != NULL; prev = sm, sm = sm->nxt) {
if(sfl_dsi_compare(pdsi, &sm->dsi) == 0) {
if(prev == NULL) agent->samplers = sm->nxt;
else prev->nxt = sm->nxt;
sfl_agent_jumpTableRemove(agent, sm);
sflFree(agent, sm);
return 1;
}
}
/* not found */
return 0;
}
/*_________________---------------------------__________________
_________________ sfl_agent_removePoller __________________
-----------------___________________________------------------
*/
int sfl_agent_removePoller(SFLAgent *agent, SFLDataSource_instance *pdsi)
{
/* find it, unlink it and free it */
SFLPoller *prev = NULL, *pl = agent->pollers;
for(; pl != NULL; prev = pl, pl = pl->nxt) {
if(sfl_dsi_compare(pdsi, &pl->dsi) == 0) {
if(prev == NULL) agent->pollers = pl->nxt;
else prev->nxt = pl->nxt;
sflFree(agent, pl);
return 1;
}
}
/* not found */
return 0;
}
/*_________________--------------------------------__________________
_________________ sfl_agent_jumpTableAdd __________________
-----------------________________________________------------------
*/
static void sfl_agent_jumpTableAdd(SFLAgent *agent, SFLSampler *sampler)
{
u_int32_t hashIndex = SFL_DS_INDEX(sampler->dsi) % SFL_HASHTABLE_SIZ;
sampler->hash_nxt = agent->jumpTable[hashIndex];
agent->jumpTable[hashIndex] = sampler;
}
/*_________________--------------------------------__________________
_________________ sfl_agent_jumpTableRemove __________________
-----------------________________________________------------------
*/
static void sfl_agent_jumpTableRemove(SFLAgent *agent, SFLSampler *sampler)
{
u_int32_t hashIndex = SFL_DS_INDEX(sampler->dsi) % SFL_HASHTABLE_SIZ;
SFLSampler *search = agent->jumpTable[hashIndex], *prev = NULL;
for( ; search != NULL; prev = search, search = search->hash_nxt) if(search == sampler) break;
if(search) {
// found - unlink
if(prev) prev->hash_nxt = search->hash_nxt;
else agent->jumpTable[hashIndex] = search->hash_nxt;
search->hash_nxt = NULL;
}
}
/*_________________--------------------------------__________________
_________________ sfl_agent_getSamplerByIfIndex __________________
-----------------________________________________------------------
fast lookup (pointers cached in hash table). If there are multiple
sampler instances for a given ifIndex, then this fn will return
the one with the lowest instance number. Since the samplers
list is sorted, this means the other instances will be accesible
by following the sampler->nxt pointer (until the ds_class
or ds_index changes). This is helpful if you need to offer
the same flowSample to multiple samplers.
*/
SFLSampler *sfl_agent_getSamplerByIfIndex(SFLAgent *agent, u_int32_t ifIndex)
{
SFLSampler *search = agent->jumpTable[ifIndex % SFL_HASHTABLE_SIZ];
for( ; search != NULL; search = search->hash_nxt) if(SFL_DS_INDEX(search->dsi) == ifIndex) break;
return search;
}
/*_________________---------------------------__________________
_________________ sfl_agent_getSampler __________________
-----------------___________________________------------------
*/
SFLSampler *sfl_agent_getSampler(SFLAgent *agent, SFLDataSource_instance *pdsi)
{
/* find it and return it */
SFLSampler *sm = agent->samplers;
for(; sm != NULL; sm = sm->nxt)
if(sfl_dsi_compare(pdsi, &sm->dsi) == 0) return sm;
/* not found */
return NULL;
}
/*_________________---------------------------__________________
_________________ sfl_agent_getPoller __________________
-----------------___________________________------------------
*/
SFLPoller *sfl_agent_getPoller(SFLAgent *agent, SFLDataSource_instance *pdsi)
{
/* find it and return it */
SFLPoller *pl = agent->pollers;
for(; pl != NULL; pl = pl->nxt)
if(sfl_dsi_compare(pdsi, &pl->dsi) == 0) return pl;
/* not found */
return NULL;
}
/*_________________---------------------------__________________
_________________ sfl_agent_getReceiver __________________
-----------------___________________________------------------
*/
SFLReceiver *sfl_agent_getReceiver(SFLAgent *agent, u_int32_t receiverIndex)
{
u_int32_t rcvIdx = 0;
SFLReceiver *rcv = agent->receivers;
for(; rcv != NULL; rcv = rcv->nxt)
if(receiverIndex == ++rcvIdx) return rcv;
/* not found - ran off the end of the table */
return NULL;
}
/*_________________---------------------------__________________
_________________ sfl_agent_getNextSampler __________________
-----------------___________________________------------------
*/
SFLSampler *sfl_agent_getNextSampler(SFLAgent *agent, SFLDataSource_instance *pdsi)
{
/* return the one lexograpically just after it - assume they are sorted
correctly according to the lexographical ordering of the object ids */
SFLSampler *sm = sfl_agent_getSampler(agent, pdsi);
return sm ? sm->nxt : NULL;
}
/*_________________---------------------------__________________
_________________ sfl_agent_getNextPoller __________________
-----------------___________________________------------------
*/
SFLPoller *sfl_agent_getNextPoller(SFLAgent *agent, SFLDataSource_instance *pdsi)
{
/* return the one lexograpically just after it - assume they are sorted
correctly according to the lexographical ordering of the object ids */
SFLPoller *pl = sfl_agent_getPoller(agent, pdsi);
return pl ? pl->nxt : NULL;
}
/*_________________---------------------------__________________
_________________ sfl_agent_getNextReceiver __________________
-----------------___________________________------------------
*/
SFLReceiver *sfl_agent_getNextReceiver(SFLAgent *agent, u_int32_t receiverIndex)
{
return sfl_agent_getReceiver(agent, receiverIndex + 1);
}
/*_________________---------------------------__________________
_________________ sfl_agent_resetReceiver __________________
-----------------___________________________------------------
*/
void sfl_agent_resetReceiver(SFLAgent *agent, SFLReceiver *receiver)
{
/* tell samplers and pollers to stop sending to this receiver */
/* first get his receiverIndex */
u_int32_t rcvIdx = 0;
SFLReceiver *rcv = agent->receivers;
for(; rcv != NULL; rcv = rcv->nxt) {
rcvIdx++; /* thanks to Diego Valverde for pointing out this bugfix */
if(rcv == receiver) {
/* now tell anyone that is using it to stop */
SFLSampler *sm = agent->samplers;
SFLPoller *pl = agent->pollers;
for(; sm != NULL; sm = sm->nxt)
if(sfl_sampler_get_sFlowFsReceiver(sm) == rcvIdx) sfl_sampler_set_sFlowFsReceiver(sm, 0);
for(; pl != NULL; pl = pl->nxt)
if(sfl_poller_get_sFlowCpReceiver(pl) == rcvIdx) sfl_poller_set_sFlowCpReceiver(pl, 0);
break;
}
}
}
/*_________________---------------------------__________________
_________________ sfl_agent_error __________________
-----------------___________________________------------------
*/
#define MAX_ERRMSG_LEN 1000
void sfl_agent_error(SFLAgent *agent, char *modName, char *msg)
{
char errm[MAX_ERRMSG_LEN];
sprintf(errm, "sfl_agent_error: %s: %s\n", modName, msg);
if(agent->errorFn) (*agent->errorFn)(agent->magic, agent, errm);
else {
fprintf(stderr, "%s\n", errm);
fflush(stderr);
}
}
/*_________________---------------------------__________________
_________________ sfl_agent_sysError __________________
-----------------___________________________------------------
*/
void sfl_agent_sysError(SFLAgent *agent, char *modName, char *msg)
{
char errm[MAX_ERRMSG_LEN];
sprintf(errm, "sfl_agent_sysError: %s: %s (errno = %d - %s)\n", modName, msg, errno, strerror(errno));
if(agent->errorFn) (*agent->errorFn)(agent->magic, agent, errm);
else {
fprintf(stderr, "%s\n", errm);
fflush(stderr);
}
}
/*_________________---------------------------__________________
_________________ alloc and free __________________
-----------------___________________________------------------
*/
static void * sflAlloc(SFLAgent *agent, size_t bytes)
{
if(agent->allocFn) return (*agent->allocFn)(agent->magic, agent, bytes);
else return SFL_ALLOC(bytes);
}
static void sflFree(SFLAgent *agent, void *obj)
{
if(agent->freeFn) (*agent->freeFn)(agent->magic, agent, obj);
else SFL_FREE(obj);
}

340
lib/sflow_api.h Normal file
View File

@@ -0,0 +1,340 @@
/* Copyright (c) 2002-2009 InMon Corp. Licensed under the terms of the InMon sFlow licence: */
/* http://www.inmon.com/technology/sflowlicense.txt */
#ifndef SFLOW_API_H
#define SFLOW_API_H 1
/* define SFLOW_DO_SOCKET to 1 if you want the agent
to send the packets itself, otherwise set the sendFn
callback in sfl_agent_init.*/
/* #define SFLOW_DO_SOCKET */
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <sys/types.h>
#include <arpa/inet.h> /* for htonl */
#ifdef SFLOW_DO_SOCKET
#include <sys/socket.h>
#include <netinet/in_systm.h>
#include <netinet/in.h>
#include <netinet/ip.h>
#endif
#include "sflow.h"
/* define SFLOW_SOFTWARE_SAMPLING to 1 if you need to use the
sfl_sampler_takeSample routine and give it every packet */
/* #define SFLOW_SOFTWARE_SAMPLING */
/*
uncomment this preprocessor flag (or compile with -DSFL_USE_32BIT_INDEX)
if your ds_index numbers can ever be >= 2^30-1 (i.e. >= 0x3FFFFFFF)
*/
/* #define SFL_USE_32BIT_INDEX */
/* Used to combine ds_class, ds_index and instance into
a single 64-bit number like this:
__________________________________
| cls| index | instance |
----------------------------------
but now is opened up to a 12-byte struct to ensure
that ds_index has a full 32-bit field, and to make
accessing the components simpler. The macros have
the same behavior as before, so this change should
be transparent. The only difference is that these
objects are now passed around by reference instead
of by value, and the comparison is done using a fn.
*/
typedef struct _SFLDataSource_instance {
u_int32_t ds_class;
u_int32_t ds_index;
u_int32_t ds_instance;
} SFLDataSource_instance;
#ifdef SFL_USE_32BIT_INDEX
#define SFL_FLOW_SAMPLE_TYPE SFLFlow_sample_expanded
#define SFL_COUNTERS_SAMPLE_TYPE SFLCounters_sample_expanded
#else
#define SFL_FLOW_SAMPLE_TYPE SFLFlow_sample
#define SFL_COUNTERS_SAMPLE_TYPE SFLCounters_sample
/* if index numbers are not going to use all 32 bits, then we can use
the more compact encoding, with the dataSource class and index merged */
#define SFL_DS_DATASOURCE(dsi) (((dsi).ds_class << 24) + (dsi).ds_index)
#endif
#define SFL_DS_INSTANCE(dsi) (dsi).ds_instance
#define SFL_DS_CLASS(dsi) (dsi).ds_class
#define SFL_DS_INDEX(dsi) (dsi).ds_index
#define SFL_DS_SET(dsi,clss,indx,inst) \
do { \
(dsi).ds_class = (clss); \
(dsi).ds_index = (indx); \
(dsi).ds_instance = (inst); \
} while(0)
typedef struct _SFLSampleCollector {
u_int32_t data[(SFL_MAX_DATAGRAM_SIZE + SFL_DATA_PAD) / sizeof(u_int32_t)];
u_int32_t *datap; /* packet fill pointer */
u_int32_t pktlen; /* accumulated size */
u_int32_t packetSeqNo;
u_int32_t numSamples;
} SFLSampleCollector;
struct _SFLAgent; /* forward decl */
typedef struct _SFLReceiver {
struct _SFLReceiver *nxt;
/* MIB fields */
char *sFlowRcvrOwner;
time_t sFlowRcvrTimeout;
u_int32_t sFlowRcvrMaximumDatagramSize;
SFLAddress sFlowRcvrAddress;
u_int32_t sFlowRcvrPort;
u_int32_t sFlowRcvrDatagramVersion;
/* public fields */
struct _SFLAgent *agent; /* pointer to my agent */
/* private fields */
SFLSampleCollector sampleCollector;
#ifdef SFLOW_DO_SOCKET
struct sockaddr_in receiver4;
struct sockaddr_in6 receiver6;
#endif
} SFLReceiver;
typedef struct _SFLSampler {
/* for linked list */
struct _SFLSampler *nxt;
/* for hash lookup table */
struct _SFLSampler *hash_nxt;
/* MIB fields */
SFLDataSource_instance dsi;
u_int32_t sFlowFsReceiver;
u_int32_t sFlowFsPacketSamplingRate;
u_int32_t sFlowFsMaximumHeaderSize;
/* public fields */
struct _SFLAgent *agent; /* pointer to my agent */
/* private fields */
SFLReceiver *myReceiver;
u_int32_t skip;
u_int32_t samplePool;
u_int32_t flowSampleSeqNo;
/* rate checking */
u_int32_t samplesThisTick;
u_int32_t samplesLastTick;
u_int32_t backoffThreshold;
} SFLSampler;
/* declare */
struct _SFLPoller;
typedef void (*getCountersFn_t)(void *magic, /* callback to get counters */
struct _SFLPoller *sampler, /* called with self */
SFL_COUNTERS_SAMPLE_TYPE *cs); /* struct to fill in */
typedef struct _SFLPoller {
/* for linked list */
struct _SFLPoller *nxt;
/* MIB fields */
SFLDataSource_instance dsi;
u_int32_t sFlowCpReceiver;
time_t sFlowCpInterval;
/* public fields */
struct _SFLAgent *agent; /* pointer to my agent */
void *magic; /* ptr to pass back in getCountersFn() */
getCountersFn_t getCountersFn;
u_int32_t bridgePort; /* port number local to bridge */
/* private fields */
SFLReceiver *myReceiver;
time_t countersCountdown;
u_int32_t countersSampleSeqNo;
} SFLPoller;
typedef void *(*allocFn_t)(void *magic, /* callback to allocate space on heap */
struct _SFLAgent *agent, /* called with self */
size_t bytes); /* bytes requested */
typedef int (*freeFn_t)(void *magic, /* callback to free space on heap */
struct _SFLAgent *agent, /* called with self */
void *obj); /* obj to free */
typedef void (*errorFn_t)(void *magic, /* callback to log error message */
struct _SFLAgent *agent, /* called with self */
char *msg); /* error message */
typedef void (*sendFn_t)(void *magic, /* optional override fn to send packet */
struct _SFLAgent *agent,
SFLReceiver *receiver,
u_char *pkt,
u_int32_t pktLen);
/* prime numbers are good for hash tables */
#define SFL_HASHTABLE_SIZ 199
typedef struct _SFLAgent {
SFLSampler *jumpTable[SFL_HASHTABLE_SIZ]; /* fast lookup table for samplers (by ifIndex) */
SFLSampler *samplers; /* the list of samplers */
SFLPoller *pollers; /* the list of samplers */
SFLReceiver *receivers; /* the array of receivers */
time_t bootTime; /* time when we booted or started */
time_t now; /* time now */
SFLAddress myIP; /* IP address of this node */
u_int32_t subId; /* sub_agent_id */
void *magic; /* ptr to pass back in logging and alloc fns */
allocFn_t allocFn;
freeFn_t freeFn;
errorFn_t errorFn;
sendFn_t sendFn;
#ifdef SFLOW_DO_SOCKET
int receiverSocket4;
int receiverSocket6;
#endif
} SFLAgent;
/* call this at the start with a newly created agent */
void sfl_agent_init(SFLAgent *agent,
SFLAddress *myIP, /* IP address of this agent */
u_int32_t subId, /* agent_sub_id */
time_t bootTime, /* agent boot time */
time_t now, /* time now */
void *magic, /* ptr to pass back in logging and alloc fns */
allocFn_t allocFn,
freeFn_t freeFn,
errorFn_t errorFn,
sendFn_t sendFn);
/* call this to create samplers */
SFLSampler *sfl_agent_addSampler(SFLAgent *agent, SFLDataSource_instance *pdsi);
/* call this to create pollers */
SFLPoller *sfl_agent_addPoller(SFLAgent *agent,
SFLDataSource_instance *pdsi,
void *magic, /* ptr to pass back in getCountersFn() */
getCountersFn_t getCountersFn);
/* call this to create receivers */
SFLReceiver *sfl_agent_addReceiver(SFLAgent *agent);
/* call this to remove samplers */
int sfl_agent_removeSampler(SFLAgent *agent, SFLDataSource_instance *pdsi);
/* call this to remove pollers */
int sfl_agent_removePoller(SFLAgent *agent, SFLDataSource_instance *pdsi);
/* note: receivers should not be removed. Typically the receivers
list will be created at init time and never changed */
/* call these fns to retrieve sampler, poller or receiver (e.g. for SNMP GET or GETNEXT operation) */
SFLSampler *sfl_agent_getSampler(SFLAgent *agent, SFLDataSource_instance *pdsi);
SFLSampler *sfl_agent_getNextSampler(SFLAgent *agent, SFLDataSource_instance *pdsi);
SFLPoller *sfl_agent_getPoller(SFLAgent *agent, SFLDataSource_instance *pdsi);
SFLPoller *sfl_agent_getNextPoller(SFLAgent *agent, SFLDataSource_instance *pdsi);
SFLReceiver *sfl_agent_getReceiver(SFLAgent *agent, u_int32_t receiverIndex);
SFLReceiver *sfl_agent_getNextReceiver(SFLAgent *agent, u_int32_t receiverIndex);
/* jump table access - for performance */
SFLSampler *sfl_agent_getSamplerByIfIndex(SFLAgent *agent, u_int32_t ifIndex);
/* call these functions to GET and SET MIB values */
/* receiver */
char * sfl_receiver_get_sFlowRcvrOwner(SFLReceiver *receiver);
void sfl_receiver_set_sFlowRcvrOwner(SFLReceiver *receiver, char *sFlowRcvrOwner);
time_t sfl_receiver_get_sFlowRcvrTimeout(SFLReceiver *receiver);
void sfl_receiver_set_sFlowRcvrTimeout(SFLReceiver *receiver, time_t sFlowRcvrTimeout);
u_int32_t sfl_receiver_get_sFlowRcvrMaximumDatagramSize(SFLReceiver *receiver);
void sfl_receiver_set_sFlowRcvrMaximumDatagramSize(SFLReceiver *receiver, u_int32_t sFlowRcvrMaximumDatagramSize);
SFLAddress *sfl_receiver_get_sFlowRcvrAddress(SFLReceiver *receiver);
void sfl_receiver_set_sFlowRcvrAddress(SFLReceiver *receiver, SFLAddress *sFlowRcvrAddress);
u_int32_t sfl_receiver_get_sFlowRcvrPort(SFLReceiver *receiver);
void sfl_receiver_set_sFlowRcvrPort(SFLReceiver *receiver, u_int32_t sFlowRcvrPort);
/* sampler */
u_int32_t sfl_sampler_get_sFlowFsReceiver(SFLSampler *sampler);
void sfl_sampler_set_sFlowFsReceiver(SFLSampler *sampler, u_int32_t sFlowFsReceiver);
u_int32_t sfl_sampler_get_sFlowFsPacketSamplingRate(SFLSampler *sampler);
void sfl_sampler_set_sFlowFsPacketSamplingRate(SFLSampler *sampler, u_int32_t sFlowFsPacketSamplingRate);
u_int32_t sfl_sampler_get_sFlowFsMaximumHeaderSize(SFLSampler *sampler);
void sfl_sampler_set_sFlowFsMaximumHeaderSize(SFLSampler *sampler, u_int32_t sFlowFsMaximumHeaderSize);
u_int32_t sfl_sampler_get_samplesLastTick(SFLSampler *sampler);
/* poller */
u_int32_t sfl_poller_get_sFlowCpReceiver(SFLPoller *poller);
void sfl_poller_set_sFlowCpReceiver(SFLPoller *poller, u_int32_t sFlowCpReceiver);
u_int32_t sfl_poller_get_sFlowCpInterval(SFLPoller *poller);
void sfl_poller_set_sFlowCpInterval(SFLPoller *poller, u_int32_t sFlowCpInterval);
/* fns to set the sflow agent address or sub-id */
void sfl_agent_set_agentAddress(SFLAgent *agent, SFLAddress *addr);
void sfl_agent_set_agentSubId(SFLAgent *agent, u_int32_t subId);
/* The poller may need a separate number to reference the local bridge port
to get counters if it is not the same as the global ifIndex */
void sfl_poller_set_bridgePort(SFLPoller *poller, u_int32_t port_no);
u_int32_t sfl_poller_get_bridgePort(SFLPoller *poller);
/* call this to indicate a discontinuity with a counter like samplePool so that the
sflow collector will ignore the next delta */
void sfl_sampler_resetFlowSeqNo(SFLSampler *sampler);
/* call this to indicate a discontinuity with one or more of the counters so that the
sflow collector will ignore the next delta */
void sfl_poller_resetCountersSeqNo(SFLPoller *poller);
#ifdef SFLOW_SOFTWARE_SAMLING
/* software sampling: call this with every packet - returns non-zero if the packet
should be sampled (in which case you then call sfl_sampler_writeFlowSample()) */
int sfl_sampler_takeSample(SFLSampler *sampler);
#endif
/* call this to set a maximum samples-per-second threshold. If the sampler reaches this
threshold it will automatically back off the sampling rate. A value of 0 disables the
mechanism */
void sfl_sampler_set_backoffThreshold(SFLSampler *sampler, u_int32_t samplesPerSecond);
u_int32_t sfl_sampler_get_backoffThreshold(SFLSampler *sampler);
/* call this once per second (N.B. not on interrupt stack i.e. not hard real-time) */
void sfl_agent_tick(SFLAgent *agent, time_t now);
/* call this with each flow sample */
void sfl_sampler_writeFlowSample(SFLSampler *sampler, SFL_FLOW_SAMPLE_TYPE *fs);
/* call this to push counters samples (usually done in the getCountersFn callback) */
void sfl_poller_writeCountersSample(SFLPoller *poller, SFL_COUNTERS_SAMPLE_TYPE *cs);
/* call this to deallocate resources */
void sfl_agent_release(SFLAgent *agent);
/* internal fns */
void sfl_receiver_init(SFLReceiver *receiver, SFLAgent *agent);
void sfl_sampler_init(SFLSampler *sampler, SFLAgent *agent, SFLDataSource_instance *pdsi);
void sfl_poller_init(SFLPoller *poller, SFLAgent *agent, SFLDataSource_instance *pdsi, void *magic, getCountersFn_t getCountersFn);
void sfl_receiver_tick(SFLReceiver *receiver, time_t now);
void sfl_poller_tick(SFLPoller *poller, time_t now);
void sfl_sampler_tick(SFLSampler *sampler, time_t now);
int sfl_receiver_writeFlowSample(SFLReceiver *receiver, SFL_FLOW_SAMPLE_TYPE *fs);
int sfl_receiver_writeCountersSample(SFLReceiver *receiver, SFL_COUNTERS_SAMPLE_TYPE *cs);
void sfl_agent_resetReceiver(SFLAgent *agent, SFLReceiver *receiver);
void sfl_agent_error(SFLAgent *agent, char *modName, char *msg);
void sfl_agent_sysError(SFLAgent *agent, char *modName, char *msg);
u_int32_t sfl_receiver_samplePacketsSent(SFLReceiver *receiver);
#define SFL_ALLOC malloc
#define SFL_FREE free
#endif /* SFLOW_API_H */

142
lib/sflow_poller.c Normal file
View File

@@ -0,0 +1,142 @@
/* Copyright (c) 2002-2009 InMon Corp. Licensed under the terms of the InMon sFlow licence: */
/* http://www.inmon.com/technology/sflowlicense.txt */
#include "sflow_api.h"
/*_________________--------------------------__________________
_________________ sfl_poller_init __________________
-----------------__________________________------------------
*/
void sfl_poller_init(SFLPoller *poller,
SFLAgent *agent,
SFLDataSource_instance *pdsi,
void *magic, /* ptr to pass back in getCountersFn() */
getCountersFn_t getCountersFn)
{
/* copy the dsi in case it points to poller->dsi, which we are about to clear */
SFLDataSource_instance dsi = *pdsi;
/* preserve the *nxt pointer too, in case we are resetting this poller and it is
already part of the agent's linked list (thanks to Matt Woodly for pointing this out) */
SFLPoller *nxtPtr = poller->nxt;
/* clear everything */
memset(poller, 0, sizeof(*poller));
/* restore the linked list ptr */
poller->nxt = nxtPtr;
/* now copy in the parameters */
poller->agent = agent;
poller->dsi = dsi; /* structure copy */
poller->magic = magic;
poller->getCountersFn = getCountersFn;
}
/*_________________--------------------------__________________
_________________ reset __________________
-----------------__________________________------------------
*/
static void reset(SFLPoller *poller)
{
SFLDataSource_instance dsi = poller->dsi;
sfl_poller_init(poller, poller->agent, &dsi, poller->magic, poller->getCountersFn);
}
/*_________________---------------------------__________________
_________________ MIB access __________________
-----------------___________________________------------------
*/
u_int32_t sfl_poller_get_sFlowCpReceiver(SFLPoller *poller) {
return poller->sFlowCpReceiver;
}
void sfl_poller_set_sFlowCpReceiver(SFLPoller *poller, u_int32_t sFlowCpReceiver) {
poller->sFlowCpReceiver = sFlowCpReceiver;
if(sFlowCpReceiver == 0) reset(poller);
else {
/* retrieve and cache a direct pointer to my receiver */
poller->myReceiver = sfl_agent_getReceiver(poller->agent, poller->sFlowCpReceiver);
}
}
u_int32_t sfl_poller_get_sFlowCpInterval(SFLPoller *poller) {
return poller->sFlowCpInterval;
}
void sfl_poller_set_sFlowCpInterval(SFLPoller *poller, u_int32_t sFlowCpInterval) {
poller->sFlowCpInterval = sFlowCpInterval;
/* Set the countersCountdown to be a randomly selected value between 1 and
sFlowCpInterval. That way the counter polling would be desynchronised
(on a 200-port switch, polling all the counters in one second could be harmful). */
poller->countersCountdown = 1 + (random() % sFlowCpInterval);
}
/*_________________---------------------------------__________________
_________________ bridge port __________________
-----------------_________________________________------------------
May need a separate number to reference the local bridge port
to get counters if it is not the same as the global ifIndex.
*/
void sfl_poller_set_bridgePort(SFLPoller *poller, u_int32_t port_no) {
poller->bridgePort = port_no;
}
u_int32_t sfl_poller_get_bridgePort(SFLPoller *poller) {
return poller->bridgePort;
}
/*_________________---------------------------------__________________
_________________ sequence number reset __________________
-----------------_________________________________------------------
Used to indicate a counter discontinuity
so that the sflow collector will know to ignore the next delta.
*/
void sfl_poller_resetCountersSeqNo(SFLPoller *poller) { poller->countersSampleSeqNo = 0; }
/*_________________---------------------------__________________
_________________ sfl_poller_tick __________________
-----------------___________________________------------------
*/
void sfl_poller_tick(SFLPoller *poller, time_t now)
{
if(poller->countersCountdown == 0) return; /* counters retrieval was not enabled */
if(poller->sFlowCpReceiver == 0) return;
if(--poller->countersCountdown == 0) {
if(poller->getCountersFn != NULL) {
/* call out for counters */
SFL_COUNTERS_SAMPLE_TYPE cs;
memset(&cs, 0, sizeof(cs));
poller->getCountersFn(poller->magic, poller, &cs);
/* this countersFn is expected to fill in some counter block elements
and then call sfl_poller_writeCountersSample(poller, &cs); */
}
/* reset the countdown */
poller->countersCountdown = poller->sFlowCpInterval;
}
}
/*_________________---------------------------------__________________
_________________ sfl_poller_writeCountersSample __________________
-----------------_________________________________------------------
*/
void sfl_poller_writeCountersSample(SFLPoller *poller, SFL_COUNTERS_SAMPLE_TYPE *cs)
{
/* fill in the rest of the header fields, and send to the receiver */
cs->sequence_number = ++poller->countersSampleSeqNo;
#ifdef SFL_USE_32BIT_INDEX
cs->ds_class = SFL_DS_CLASS(poller->dsi);
cs->ds_index = SFL_DS_INDEX(poller->dsi);
#else
cs->source_id = SFL_DS_DATASOURCE(poller->dsi);
#endif
/* sent to my receiver */
if(poller->myReceiver) sfl_receiver_writeCountersSample(poller->myReceiver, cs);
}

832
lib/sflow_receiver.c Normal file
View File

@@ -0,0 +1,832 @@
/* Copyright (c) 2002-2009 InMon Corp. Licensed under the terms of the InMon sFlow licence: */
/* http://www.inmon.com/technology/sflowlicense.txt */
#include <assert.h>
#include "sflow_api.h"
static void resetSampleCollector(SFLReceiver *receiver);
static void sendSample(SFLReceiver *receiver);
static void sflError(SFLReceiver *receiver, char *errm);
inline static void putNet32(SFLReceiver *receiver, u_int32_t val);
inline static void putAddress(SFLReceiver *receiver, SFLAddress *addr);
#ifdef SFLOW_DO_SOCKET
static void initSocket(SFLReceiver *receiver);
#endif
/*_________________--------------------------__________________
_________________ sfl_receiver_init __________________
-----------------__________________________------------------
*/
void sfl_receiver_init(SFLReceiver *receiver, SFLAgent *agent)
{
/* first clear everything */
memset(receiver, 0, sizeof(*receiver));
/* now copy in the parameters */
receiver->agent = agent;
/* set defaults */
receiver->sFlowRcvrMaximumDatagramSize = SFL_DEFAULT_DATAGRAM_SIZE;
receiver->sFlowRcvrPort = SFL_DEFAULT_COLLECTOR_PORT;
#ifdef SFLOW_DO_SOCKET
/* initialize the socket address */
initSocket(receiver);
#endif
/* preset some of the header fields */
receiver->sampleCollector.datap = receiver->sampleCollector.data;
putNet32(receiver, SFLDATAGRAM_VERSION5);
putAddress(receiver, &agent->myIP);
putNet32(receiver, agent->subId);
/* prepare to receive the first sample */
resetSampleCollector(receiver);
}
/*_________________---------------------------__________________
_________________ reset __________________
-----------------___________________________------------------
called on timeout, or when owner string is cleared
*/
static void reset(SFLReceiver *receiver) {
// ask agent to tell samplers and pollers to stop sending samples
sfl_agent_resetReceiver(receiver->agent, receiver);
// reinitialize
sfl_receiver_init(receiver, receiver->agent);
}
#ifdef SFLOW_DO_SOCKET
/*_________________---------------------------__________________
_________________ initSocket __________________
-----------------___________________________------------------
*/
static void initSocket(SFLReceiver *receiver) {
if(receiver->sFlowRcvrAddress.type == SFLADDRESSTYPE_IP_V6) {
struct sockaddr_in6 *sa6 = &receiver->receiver6;
sa6->sin6_port = htons((u_int16_t)receiver->sFlowRcvrPort);
sa6->sin6_family = AF_INET6;
sa6->sin6_addr = receiver->sFlowRcvrAddress.address.ip_v6;
}
else {
struct sockaddr_in *sa4 = &receiver->receiver4;
sa4->sin_port = htons((u_int16_t)receiver->sFlowRcvrPort);
sa4->sin_family = AF_INET;
sa4->sin_addr = receiver->sFlowRcvrAddress.address.ip_v4;
}
}
#endif
/*_________________----------------------------------------_____________
_________________ MIB Vars _____________
-----------------________________________________________-------------
*/
char * sfl_receiver_get_sFlowRcvrOwner(SFLReceiver *receiver) {
return receiver->sFlowRcvrOwner;
}
void sfl_receiver_set_sFlowRcvrOwner(SFLReceiver *receiver, char *sFlowRcvrOwner) {
receiver->sFlowRcvrOwner = sFlowRcvrOwner;
if(sFlowRcvrOwner == NULL || sFlowRcvrOwner[0] == '\0') {
// reset condition! owner string was cleared
reset(receiver);
}
}
time_t sfl_receiver_get_sFlowRcvrTimeout(SFLReceiver *receiver) {
return receiver->sFlowRcvrTimeout;
}
void sfl_receiver_set_sFlowRcvrTimeout(SFLReceiver *receiver, time_t sFlowRcvrTimeout) {
receiver->sFlowRcvrTimeout =sFlowRcvrTimeout;
}
u_int32_t sfl_receiver_get_sFlowRcvrMaximumDatagramSize(SFLReceiver *receiver) {
return receiver->sFlowRcvrMaximumDatagramSize;
}
void sfl_receiver_set_sFlowRcvrMaximumDatagramSize(SFLReceiver *receiver, u_int32_t sFlowRcvrMaximumDatagramSize) {
u_int32_t mdz = sFlowRcvrMaximumDatagramSize;
if(mdz < SFL_MIN_DATAGRAM_SIZE) mdz = SFL_MIN_DATAGRAM_SIZE;
receiver->sFlowRcvrMaximumDatagramSize = mdz;
}
SFLAddress *sfl_receiver_get_sFlowRcvrAddress(SFLReceiver *receiver) {
return &receiver->sFlowRcvrAddress;
}
void sfl_receiver_set_sFlowRcvrAddress(SFLReceiver *receiver, SFLAddress *sFlowRcvrAddress) {
if(sFlowRcvrAddress) receiver->sFlowRcvrAddress = *sFlowRcvrAddress; // structure copy
#ifdef SFLOW_DO_SOCKET
initSocket(receiver);
#endif
}
u_int32_t sfl_receiver_get_sFlowRcvrPort(SFLReceiver *receiver) {
return receiver->sFlowRcvrPort;
}
void sfl_receiver_set_sFlowRcvrPort(SFLReceiver *receiver, u_int32_t sFlowRcvrPort) {
receiver->sFlowRcvrPort = sFlowRcvrPort;
// update the socket structure
#ifdef SFLOW_DO_SOCKET
initSocket(receiver);
#endif
}
/*_________________---------------------------__________________
_________________ sfl_receiver_tick __________________
-----------------___________________________------------------
*/
void sfl_receiver_tick(SFLReceiver *receiver, time_t now)
{
// if there are any samples to send, flush them now
if(receiver->sampleCollector.numSamples > 0) sendSample(receiver);
// check the timeout
if(receiver->sFlowRcvrTimeout && (u_int32_t)receiver->sFlowRcvrTimeout != 0xFFFFFFFF) {
// count down one tick and reset if we reach 0
if(--receiver->sFlowRcvrTimeout == 0) reset(receiver);
}
}
/*_________________-----------------------------__________________
_________________ receiver write utilities __________________
-----------------_____________________________------------------
*/
inline static void put32(SFLReceiver *receiver, u_int32_t val)
{
*receiver->sampleCollector.datap++ = val;
}
inline static void putNet32(SFLReceiver *receiver, u_int32_t val)
{
*receiver->sampleCollector.datap++ = htonl(val);
}
inline static void putNet32_run(SFLReceiver *receiver, void *obj, size_t quads)
{
u_int32_t *from = (u_int32_t *)obj;
while(quads--) putNet32(receiver, *from++);
}
inline static void putNet64(SFLReceiver *receiver, u_int64_t val64)
{
u_int32_t *firstQuadPtr = receiver->sampleCollector.datap;
// first copy the bytes in
memcpy((u_char *)firstQuadPtr, &val64, 8);
if(htonl(1) != 1) {
// swap the bytes, and reverse the quads too
u_int32_t tmp = *receiver->sampleCollector.datap++;
*firstQuadPtr = htonl(*receiver->sampleCollector.datap);
*receiver->sampleCollector.datap++ = htonl(tmp);
}
else receiver->sampleCollector.datap += 2;
}
inline static void put128(SFLReceiver *receiver, u_char *val)
{
memcpy(receiver->sampleCollector.datap, val, 16);
receiver->sampleCollector.datap += 4;
}
inline static void putString(SFLReceiver *receiver, SFLString *s)
{
putNet32(receiver, s->len);
memcpy(receiver->sampleCollector.datap, s->str, s->len);
receiver->sampleCollector.datap += (s->len + 3) / 4; /* pad to 4-byte boundary */
}
inline static u_int32_t stringEncodingLength(SFLString *s) {
// answer in bytes, so remember to mulitply by 4 after rounding up to nearest 4-byte boundary
return 4 + (((s->len + 3) / 4) * 4);
}
inline static void putAddress(SFLReceiver *receiver, SFLAddress *addr)
{
// encode unspecified addresses as IPV4:0.0.0.0 - or should we flag this as an error?
if(addr->type == 0) {
putNet32(receiver, SFLADDRESSTYPE_IP_V4);
put32(receiver, 0);
}
else {
putNet32(receiver, addr->type);
if(addr->type == SFLADDRESSTYPE_IP_V4) put32(receiver, addr->address.ip_v4.addr);
else put128(receiver, addr->address.ip_v6.addr);
}
}
inline static u_int32_t addressEncodingLength(SFLAddress *addr) {
return (addr->type == SFLADDRESSTYPE_IP_V6) ? 20 : 8; // type + address (unspecified == IPV4)
}
inline static void putMACAddress(SFLReceiver *receiver, u_int8_t *mac)
{
memcpy(receiver->sampleCollector.datap, mac, 6);
receiver->sampleCollector.datap += 2;
}
inline static void putSwitch(SFLReceiver *receiver, SFLExtended_switch *sw)
{
putNet32(receiver, sw->src_vlan);
putNet32(receiver, sw->src_priority);
putNet32(receiver, sw->dst_vlan);
putNet32(receiver, sw->dst_priority);
}
inline static void putRouter(SFLReceiver *receiver, SFLExtended_router *router)
{
putAddress(receiver, &router->nexthop);
putNet32(receiver, router->src_mask);
putNet32(receiver, router->dst_mask);
}
inline static u_int32_t routerEncodingLength(SFLExtended_router *router) {
return addressEncodingLength(&router->nexthop) + 8;
}
inline static void putGateway(SFLReceiver *receiver, SFLExtended_gateway *gw)
{
putAddress(receiver, &gw->nexthop);
putNet32(receiver, gw->as);
putNet32(receiver, gw->src_as);
putNet32(receiver, gw->src_peer_as);
putNet32(receiver, gw->dst_as_path_segments);
{
u_int32_t seg = 0;
for(; seg < gw->dst_as_path_segments; seg++) {
putNet32(receiver, gw->dst_as_path[seg].type);
putNet32(receiver, gw->dst_as_path[seg].length);
putNet32_run(receiver, gw->dst_as_path[seg].as.seq, gw->dst_as_path[seg].length);
}
}
putNet32(receiver, gw->communities_length);
putNet32_run(receiver, gw->communities, gw->communities_length);
putNet32(receiver, gw->localpref);
}
inline static u_int32_t gatewayEncodingLength(SFLExtended_gateway *gw) {
u_int32_t elemSiz = addressEncodingLength(&gw->nexthop);
u_int32_t seg = 0;
elemSiz += 16; // as, src_as, src_peer_as, dst_as_path_segments
for(; seg < gw->dst_as_path_segments; seg++) {
elemSiz += 8; // type, length
elemSiz += 4 * gw->dst_as_path[seg].length; // set/seq bytes
}
elemSiz += 4; // communities_length
elemSiz += 4 * gw->communities_length; // communities
elemSiz += 4; // localpref
return elemSiz;
}
inline static void putUser(SFLReceiver *receiver, SFLExtended_user *user)
{
putNet32(receiver, user->src_charset);
putString(receiver, &user->src_user);
putNet32(receiver, user->dst_charset);
putString(receiver, &user->dst_user);
}
inline static u_int32_t userEncodingLength(SFLExtended_user *user) {
return 4
+ stringEncodingLength(&user->src_user)
+ 4
+ stringEncodingLength(&user->dst_user);
}
inline static void putUrl(SFLReceiver *receiver, SFLExtended_url *url)
{
putNet32(receiver, url->direction);
putString(receiver, &url->url);
putString(receiver, &url->host);
}
inline static u_int32_t urlEncodingLength(SFLExtended_url *url) {
return 4
+ stringEncodingLength(&url->url)
+ stringEncodingLength(&url->host);
}
inline static void putLabelStack(SFLReceiver *receiver, SFLLabelStack *labelStack)
{
putNet32(receiver, labelStack->depth);
putNet32_run(receiver, labelStack->stack, labelStack->depth);
}
inline static u_int32_t labelStackEncodingLength(SFLLabelStack *labelStack) {
return 4 + (4 * labelStack->depth);
}
inline static void putMpls(SFLReceiver *receiver, SFLExtended_mpls *mpls)
{
putAddress(receiver, &mpls->nextHop);
putLabelStack(receiver, &mpls->in_stack);
putLabelStack(receiver, &mpls->out_stack);
}
inline static u_int32_t mplsEncodingLength(SFLExtended_mpls *mpls) {
return addressEncodingLength(&mpls->nextHop)
+ labelStackEncodingLength(&mpls->in_stack)
+ labelStackEncodingLength(&mpls->out_stack);
}
inline static void putNat(SFLReceiver *receiver, SFLExtended_nat *nat)
{
putAddress(receiver, &nat->src);
putAddress(receiver, &nat->dst);
}
inline static u_int32_t natEncodingLength(SFLExtended_nat *nat) {
return addressEncodingLength(&nat->src)
+ addressEncodingLength(&nat->dst);
}
inline static void putMplsTunnel(SFLReceiver *receiver, SFLExtended_mpls_tunnel *tunnel)
{
putString(receiver, &tunnel->tunnel_lsp_name);
putNet32(receiver, tunnel->tunnel_id);
putNet32(receiver, tunnel->tunnel_cos);
}
inline static u_int32_t mplsTunnelEncodingLength(SFLExtended_mpls_tunnel *tunnel) {
return stringEncodingLength(&tunnel->tunnel_lsp_name) + 8;
}
inline static void putMplsVc(SFLReceiver *receiver, SFLExtended_mpls_vc *vc)
{
putString(receiver, &vc->vc_instance_name);
putNet32(receiver, vc->vll_vc_id);
putNet32(receiver, vc->vc_label_cos);
}
inline static u_int32_t mplsVcEncodingLength(SFLExtended_mpls_vc *vc) {
return stringEncodingLength( &vc->vc_instance_name) + 8;
}
inline static void putMplsFtn(SFLReceiver *receiver, SFLExtended_mpls_FTN *ftn)
{
putString(receiver, &ftn->mplsFTNDescr);
putNet32(receiver, ftn->mplsFTNMask);
}
inline static u_int32_t mplsFtnEncodingLength(SFLExtended_mpls_FTN *ftn) {
return stringEncodingLength( &ftn->mplsFTNDescr) + 4;
}
inline static void putMplsLdpFec(SFLReceiver *receiver, SFLExtended_mpls_LDP_FEC *ldpfec)
{
putNet32(receiver, ldpfec->mplsFecAddrPrefixLength);
}
inline static u_int32_t mplsLdpFecEncodingLength(SFLExtended_mpls_LDP_FEC *ldpfec) {
return 4;
}
inline static void putVlanTunnel(SFLReceiver *receiver, SFLExtended_vlan_tunnel *vlanTunnel)
{
putLabelStack(receiver, &vlanTunnel->stack);
}
inline static u_int32_t vlanTunnelEncodingLength(SFLExtended_vlan_tunnel *vlanTunnel) {
return labelStackEncodingLength(&vlanTunnel->stack);
}
inline static void putGenericCounters(SFLReceiver *receiver, SFLIf_counters *counters)
{
putNet32(receiver, counters->ifIndex);
putNet32(receiver, counters->ifType);
putNet64(receiver, counters->ifSpeed);
putNet32(receiver, counters->ifDirection);
putNet32(receiver, counters->ifStatus);
putNet64(receiver, counters->ifInOctets);
putNet32(receiver, counters->ifInUcastPkts);
putNet32(receiver, counters->ifInMulticastPkts);
putNet32(receiver, counters->ifInBroadcastPkts);
putNet32(receiver, counters->ifInDiscards);
putNet32(receiver, counters->ifInErrors);
putNet32(receiver, counters->ifInUnknownProtos);
putNet64(receiver, counters->ifOutOctets);
putNet32(receiver, counters->ifOutUcastPkts);
putNet32(receiver, counters->ifOutMulticastPkts);
putNet32(receiver, counters->ifOutBroadcastPkts);
putNet32(receiver, counters->ifOutDiscards);
putNet32(receiver, counters->ifOutErrors);
putNet32(receiver, counters->ifPromiscuousMode);
}
/*_________________-----------------------------__________________
_________________ computeFlowSampleSize __________________
-----------------_____________________________------------------
*/
static int computeFlowSampleSize(SFLReceiver *receiver, SFL_FLOW_SAMPLE_TYPE *fs)
{
SFLFlow_sample_element *elem = fs->elements;
#ifdef SFL_USE_32BIT_INDEX
u_int siz = 52; /* tag, length, sequence_number, ds_class, ds_index, sampling_rate,
sample_pool, drops, inputFormat, input, outputFormat, output, number of elements */
#else
u_int siz = 40; /* tag, length, sequence_number, source_id, sampling_rate,
sample_pool, drops, input, output, number of elements */
#endif
fs->num_elements = 0; /* we're going to count them again even if this was set by the client */
for(; elem != NULL; elem = elem->nxt) {
u_int elemSiz = 0;
fs->num_elements++;
siz += 8; /* tag, length */
switch(elem->tag) {
case SFLFLOW_HEADER:
elemSiz = 16; /* header_protocol, frame_length, stripped, header_length */
elemSiz += ((elem->flowType.header.header_length + 3) / 4) * 4; /* header, rounded up to nearest 4 bytes */
break;
case SFLFLOW_ETHERNET: elemSiz = sizeof(SFLSampled_ethernet); break;
case SFLFLOW_IPV4: elemSiz = sizeof(SFLSampled_ipv4); break;
case SFLFLOW_IPV6: elemSiz = sizeof(SFLSampled_ipv6); break;
case SFLFLOW_EX_SWITCH: elemSiz = sizeof(SFLExtended_switch); break;
case SFLFLOW_EX_ROUTER: elemSiz = routerEncodingLength(&elem->flowType.router); break;
case SFLFLOW_EX_GATEWAY: elemSiz = gatewayEncodingLength(&elem->flowType.gateway); break;
case SFLFLOW_EX_USER: elemSiz = userEncodingLength(&elem->flowType.user); break;
case SFLFLOW_EX_URL: elemSiz = urlEncodingLength(&elem->flowType.url); break;
case SFLFLOW_EX_MPLS: elemSiz = mplsEncodingLength(&elem->flowType.mpls); break;
case SFLFLOW_EX_NAT: elemSiz = natEncodingLength(&elem->flowType.nat); break;
case SFLFLOW_EX_MPLS_TUNNEL: elemSiz = mplsTunnelEncodingLength(&elem->flowType.mpls_tunnel); break;
case SFLFLOW_EX_MPLS_VC: elemSiz = mplsVcEncodingLength(&elem->flowType.mpls_vc); break;
case SFLFLOW_EX_MPLS_FTN: elemSiz = mplsFtnEncodingLength(&elem->flowType.mpls_ftn); break;
case SFLFLOW_EX_MPLS_LDP_FEC: elemSiz = mplsLdpFecEncodingLength(&elem->flowType.mpls_ldp_fec); break;
case SFLFLOW_EX_VLAN_TUNNEL: elemSiz = vlanTunnelEncodingLength(&elem->flowType.vlan_tunnel); break;
default:
sflError(receiver, "unexpected packet_data_tag");
return -1;
break;
}
// cache the element size, and accumulate it into the overall FlowSample size
elem->length = elemSiz;
siz += elemSiz;
}
return siz;
}
/*_________________-------------------------------__________________
_________________ sfl_receiver_writeFlowSample __________________
-----------------_______________________________------------------
*/
int sfl_receiver_writeFlowSample(SFLReceiver *receiver, SFL_FLOW_SAMPLE_TYPE *fs)
{
int packedSize;
if(fs == NULL) return -1;
if((packedSize = computeFlowSampleSize(receiver, fs)) == -1) return -1;
// check in case this one sample alone is too big for the datagram
// in fact - if it is even half as big then we should ditch it. Very
// important to avoid overruning the packet buffer.
if(packedSize > (int)(receiver->sFlowRcvrMaximumDatagramSize / 2)) {
sflError(receiver, "flow sample too big for datagram");
return -1;
}
// if the sample pkt is full enough so that this sample might put
// it over the limit, then we should send it now before going on.
if((receiver->sampleCollector.pktlen + packedSize) >= receiver->sFlowRcvrMaximumDatagramSize)
sendSample(receiver);
receiver->sampleCollector.numSamples++;
#ifdef SFL_USE_32BIT_INDEX
putNet32(receiver, SFLFLOW_SAMPLE_EXPANDED);
#else
putNet32(receiver, SFLFLOW_SAMPLE);
#endif
putNet32(receiver, packedSize - 8); // don't include tag and len
putNet32(receiver, fs->sequence_number);
#ifdef SFL_USE_32BIT_INDEX
putNet32(receiver, fs->ds_class);
putNet32(receiver, fs->ds_index);
#else
putNet32(receiver, fs->source_id);
#endif
putNet32(receiver, fs->sampling_rate);
putNet32(receiver, fs->sample_pool);
putNet32(receiver, fs->drops);
#ifdef SFL_USE_32BIT_INDEX
putNet32(receiver, fs->inputFormat);
putNet32(receiver, fs->input);
putNet32(receiver, fs->outputFormat);
putNet32(receiver, fs->output);
#else
putNet32(receiver, fs->input);
putNet32(receiver, fs->output);
#endif
putNet32(receiver, fs->num_elements);
{
SFLFlow_sample_element *elem = fs->elements;
for(; elem != NULL; elem = elem->nxt) {
putNet32(receiver, elem->tag);
putNet32(receiver, elem->length); // length cached in computeFlowSampleSize()
switch(elem->tag) {
case SFLFLOW_HEADER:
putNet32(receiver, elem->flowType.header.header_protocol);
putNet32(receiver, elem->flowType.header.frame_length);
putNet32(receiver, elem->flowType.header.stripped);
putNet32(receiver, elem->flowType.header.header_length);
/* the header */
memcpy(receiver->sampleCollector.datap, elem->flowType.header.header_bytes, elem->flowType.header.header_length);
/* round up to multiple of 4 to preserve alignment */
receiver->sampleCollector.datap += ((elem->flowType.header.header_length + 3) / 4);
break;
case SFLFLOW_ETHERNET:
putNet32(receiver, elem->flowType.ethernet.eth_len);
putMACAddress(receiver, elem->flowType.ethernet.src_mac);
putMACAddress(receiver, elem->flowType.ethernet.dst_mac);
putNet32(receiver, elem->flowType.ethernet.eth_type);
break;
case SFLFLOW_IPV4:
putNet32(receiver, elem->flowType.ipv4.length);
putNet32(receiver, elem->flowType.ipv4.protocol);
put32(receiver, elem->flowType.ipv4.src_ip.addr);
put32(receiver, elem->flowType.ipv4.dst_ip.addr);
putNet32(receiver, elem->flowType.ipv4.src_port);
putNet32(receiver, elem->flowType.ipv4.dst_port);
putNet32(receiver, elem->flowType.ipv4.tcp_flags);
putNet32(receiver, elem->flowType.ipv4.tos);
break;
case SFLFLOW_IPV6:
putNet32(receiver, elem->flowType.ipv6.length);
putNet32(receiver, elem->flowType.ipv6.protocol);
put128(receiver, elem->flowType.ipv6.src_ip.addr);
put128(receiver, elem->flowType.ipv6.dst_ip.addr);
putNet32(receiver, elem->flowType.ipv6.src_port);
putNet32(receiver, elem->flowType.ipv6.dst_port);
putNet32(receiver, elem->flowType.ipv6.tcp_flags);
putNet32(receiver, elem->flowType.ipv6.priority);
break;
case SFLFLOW_EX_SWITCH: putSwitch(receiver, &elem->flowType.sw); break;
case SFLFLOW_EX_ROUTER: putRouter(receiver, &elem->flowType.router); break;
case SFLFLOW_EX_GATEWAY: putGateway(receiver, &elem->flowType.gateway); break;
case SFLFLOW_EX_USER: putUser(receiver, &elem->flowType.user); break;
case SFLFLOW_EX_URL: putUrl(receiver, &elem->flowType.url); break;
case SFLFLOW_EX_MPLS: putMpls(receiver, &elem->flowType.mpls); break;
case SFLFLOW_EX_NAT: putNat(receiver, &elem->flowType.nat); break;
case SFLFLOW_EX_MPLS_TUNNEL: putMplsTunnel(receiver, &elem->flowType.mpls_tunnel); break;
case SFLFLOW_EX_MPLS_VC: putMplsVc(receiver, &elem->flowType.mpls_vc); break;
case SFLFLOW_EX_MPLS_FTN: putMplsFtn(receiver, &elem->flowType.mpls_ftn); break;
case SFLFLOW_EX_MPLS_LDP_FEC: putMplsLdpFec(receiver, &elem->flowType.mpls_ldp_fec); break;
case SFLFLOW_EX_VLAN_TUNNEL: putVlanTunnel(receiver, &elem->flowType.vlan_tunnel); break;
default:
sflError(receiver, "unexpected packet_data_tag");
return -1;
break;
}
}
}
// sanity check
assert(((u_char *)receiver->sampleCollector.datap
- (u_char *)receiver->sampleCollector.data
- receiver->sampleCollector.pktlen) == (u_int32_t)packedSize);
// update the pktlen
receiver->sampleCollector.pktlen = (u_char *)receiver->sampleCollector.datap - (u_char *)receiver->sampleCollector.data;
return packedSize;
}
/*_________________-----------------------------__________________
_________________ computeCountersSampleSize __________________
-----------------_____________________________------------------
*/
static int computeCountersSampleSize(SFLReceiver *receiver, SFL_COUNTERS_SAMPLE_TYPE *cs)
{
SFLCounters_sample_element *elem = cs->elements;
#ifdef SFL_USE_32BIT_INDEX
u_int siz = 24; /* tag, length, sequence_number, ds_class, ds_index, number of elements */
#else
u_int siz = 20; /* tag, length, sequence_number, source_id, number of elements */
#endif
cs->num_elements = 0; /* we're going to count them again even if this was set by the client */
for(; elem != NULL; elem = elem->nxt) {
u_int elemSiz = 0;
cs->num_elements++;
siz += 8; /* tag, length */
switch(elem->tag) {
case SFLCOUNTERS_GENERIC: elemSiz = sizeof(elem->counterBlock.generic); break;
case SFLCOUNTERS_ETHERNET: elemSiz = sizeof(elem->counterBlock.ethernet); break;
case SFLCOUNTERS_TOKENRING: elemSiz = sizeof(elem->counterBlock.tokenring); break;
case SFLCOUNTERS_VG: elemSiz = sizeof(elem->counterBlock.vg); break;
case SFLCOUNTERS_VLAN: elemSiz = sizeof(elem->counterBlock.vlan); break;
default:
sflError(receiver, "unexpected counters_tag");
return -1;
break;
}
// cache the element size, and accumulate it into the overall FlowSample size
elem->length = elemSiz;
siz += elemSiz;
}
return siz;
}
/*_________________----------------------------------__________________
_________________ sfl_receiver_writeCountersSample __________________
-----------------__________________________________------------------
*/
int sfl_receiver_writeCountersSample(SFLReceiver *receiver, SFL_COUNTERS_SAMPLE_TYPE *cs)
{
int packedSize;
if(cs == NULL) return -1;
// if the sample pkt is full enough so that this sample might put
// it over the limit, then we should send it now.
if((packedSize = computeCountersSampleSize(receiver, cs)) == -1) return -1;
// check in case this one sample alone is too big for the datagram
// in fact - if it is even half as big then we should ditch it. Very
// important to avoid overruning the packet buffer.
if(packedSize > (int)(receiver->sFlowRcvrMaximumDatagramSize / 2)) {
sflError(receiver, "counters sample too big for datagram");
return -1;
}
if((receiver->sampleCollector.pktlen + packedSize) >= receiver->sFlowRcvrMaximumDatagramSize)
sendSample(receiver);
receiver->sampleCollector.numSamples++;
#ifdef SFL_USE_32BIT_INDEX
putNet32(receiver, SFLCOUNTERS_SAMPLE_EXPANDED);
#else
putNet32(receiver, SFLCOUNTERS_SAMPLE);
#endif
putNet32(receiver, packedSize - 8); // tag and length not included
putNet32(receiver, cs->sequence_number);
#ifdef SFL_USE_32BIT_INDEX
putNet32(receiver, cs->ds_class);
putNet32(receiver, cs->ds_index);
#else
putNet32(receiver, cs->source_id);
#endif
putNet32(receiver, cs->num_elements);
{
SFLCounters_sample_element *elem = cs->elements;
for(; elem != NULL; elem = elem->nxt) {
putNet32(receiver, elem->tag);
putNet32(receiver, elem->length); // length cached in computeCountersSampleSize()
switch(elem->tag) {
case SFLCOUNTERS_GENERIC:
putGenericCounters(receiver, &(elem->counterBlock.generic));
break;
case SFLCOUNTERS_ETHERNET:
// all these counters are 32-bit
putNet32_run(receiver, &elem->counterBlock.ethernet, sizeof(elem->counterBlock.ethernet) / 4);
break;
case SFLCOUNTERS_TOKENRING:
// all these counters are 32-bit
putNet32_run(receiver, &elem->counterBlock.tokenring, sizeof(elem->counterBlock.tokenring) / 4);
break;
case SFLCOUNTERS_VG:
// mixed sizes
putNet32(receiver, elem->counterBlock.vg.dot12InHighPriorityFrames);
putNet64(receiver, elem->counterBlock.vg.dot12InHighPriorityOctets);
putNet32(receiver, elem->counterBlock.vg.dot12InNormPriorityFrames);
putNet64(receiver, elem->counterBlock.vg.dot12InNormPriorityOctets);
putNet32(receiver, elem->counterBlock.vg.dot12InIPMErrors);
putNet32(receiver, elem->counterBlock.vg.dot12InOversizeFrameErrors);
putNet32(receiver, elem->counterBlock.vg.dot12InDataErrors);
putNet32(receiver, elem->counterBlock.vg.dot12InNullAddressedFrames);
putNet32(receiver, elem->counterBlock.vg.dot12OutHighPriorityFrames);
putNet64(receiver, elem->counterBlock.vg.dot12OutHighPriorityOctets);
putNet32(receiver, elem->counterBlock.vg.dot12TransitionIntoTrainings);
putNet64(receiver, elem->counterBlock.vg.dot12HCInHighPriorityOctets);
putNet64(receiver, elem->counterBlock.vg.dot12HCInNormPriorityOctets);
putNet64(receiver, elem->counterBlock.vg.dot12HCOutHighPriorityOctets);
break;
case SFLCOUNTERS_VLAN:
// mixed sizes
putNet32(receiver, elem->counterBlock.vlan.vlan_id);
putNet64(receiver, elem->counterBlock.vlan.octets);
putNet32(receiver, elem->counterBlock.vlan.ucastPkts);
putNet32(receiver, elem->counterBlock.vlan.multicastPkts);
putNet32(receiver, elem->counterBlock.vlan.broadcastPkts);
putNet32(receiver, elem->counterBlock.vlan.discards);
break;
default:
sflError(receiver, "unexpected counters_tag");
return -1;
break;
}
}
}
// sanity check
assert(((u_char *)receiver->sampleCollector.datap
- (u_char *)receiver->sampleCollector.data
- receiver->sampleCollector.pktlen) == (u_int32_t)packedSize);
// update the pktlen
receiver->sampleCollector.pktlen = (u_char *)receiver->sampleCollector.datap - (u_char *)receiver->sampleCollector.data;
return packedSize;
}
/*_________________---------------------------------__________________
_________________ sfl_receiver_samplePacketsSent __________________
-----------------_________________________________------------------
*/
u_int32_t sfl_receiver_samplePacketsSent(SFLReceiver *receiver)
{
return receiver->sampleCollector.packetSeqNo;
}
/*_________________---------------------------__________________
_________________ sendSample __________________
-----------------___________________________------------------
*/
static void sendSample(SFLReceiver *receiver)
{
/* construct and send out the sample, then reset for the next one... */
/* first fill in the header with the latest values */
/* version, agent_address and sub_agent_id were pre-set. */
u_int32_t hdrIdx = (receiver->agent->myIP.type == SFLADDRESSTYPE_IP_V6) ? 7 : 4;
receiver->sampleCollector.data[hdrIdx++] = htonl(++receiver->sampleCollector.packetSeqNo); /* seq no */
receiver->sampleCollector.data[hdrIdx++] = htonl((receiver->agent->now - receiver->agent->bootTime) * 1000); /* uptime */
receiver->sampleCollector.data[hdrIdx++] = htonl(receiver->sampleCollector.numSamples); /* num samples */
/* send */
if(receiver->agent->sendFn) (*receiver->agent->sendFn)(receiver->agent->magic,
receiver->agent,
receiver,
(u_char *)receiver->sampleCollector.data,
receiver->sampleCollector.pktlen);
else {
#ifdef SFLOW_DO_SOCKET
/* send it myself */
if (receiver->sFlowRcvrAddress.type == SFLADDRESSTYPE_IP_V6) {
u_int32_t soclen = sizeof(struct sockaddr_in6);
int result = sendto(receiver->agent->receiverSocket6,
receiver->sampleCollector.data,
receiver->sampleCollector.pktlen,
0,
(struct sockaddr *)&receiver->receiver6,
soclen);
if(result == -1 && errno != EINTR) sfl_agent_sysError(receiver->agent, "receiver", "IPv6 socket sendto error");
if(result == 0) sfl_agent_error(receiver->agent, "receiver", "IPv6 socket sendto returned 0");
}
else {
u_int32_t soclen = sizeof(struct sockaddr_in);
int result = sendto(receiver->agent->receiverSocket4,
receiver->sampleCollector.data,
receiver->sampleCollector.pktlen,
0,
(struct sockaddr *)&receiver->receiver4,
soclen);
if(result == -1 && errno != EINTR) sfl_agent_sysError(receiver->agent, "receiver", "socket sendto error");
if(result == 0) sfl_agent_error(receiver->agent, "receiver", "socket sendto returned 0");
}
#endif
}
/* reset for the next time */
resetSampleCollector(receiver);
}
/*_________________---------------------------__________________
_________________ resetSampleCollector __________________
-----------------___________________________------------------
*/
static void resetSampleCollector(SFLReceiver *receiver)
{
receiver->sampleCollector.pktlen = 0;
receiver->sampleCollector.numSamples = 0;
/* point the datap to just after the header */
receiver->sampleCollector.datap = (receiver->agent->myIP.type == SFLADDRESSTYPE_IP_V6) ?
(receiver->sampleCollector.data + 10) : (receiver->sampleCollector.data + 7);
receiver->sampleCollector.pktlen = (u_char *)receiver->sampleCollector.datap - (u_char *)receiver->sampleCollector.data;
}
/*_________________---------------------------__________________
_________________ sflError __________________
-----------------___________________________------------------
*/
static void sflError(SFLReceiver *receiver, char *msg)
{
sfl_agent_error(receiver->agent, "receiver", msg);
resetSampleCollector(receiver);
}

183
lib/sflow_sampler.c Normal file
View File

@@ -0,0 +1,183 @@
/* Copyright (c) 2002-2009 InMon Corp. Licensed under the terms of the InMon sFlow licence: */
/* http://www.inmon.com/technology/sflowlicense.txt */
#include "sflow_api.h"
/*_________________--------------------------__________________
_________________ sfl_sampler_init __________________
-----------------__________________________------------------
*/
void sfl_sampler_init(SFLSampler *sampler, SFLAgent *agent, SFLDataSource_instance *pdsi)
{
/* copy the dsi in case it points to sampler->dsi, which we are about to clear.
(Thanks to Jagjit Choudray of Force 10 Networks for pointing out this bug) */
SFLDataSource_instance dsi = *pdsi;
/* preserve the *nxt pointer too, in case we are resetting this poller and it is
already part of the agent's linked list (thanks to Matt Woodly for pointing this out) */
SFLSampler *nxtPtr = sampler->nxt;
/* clear everything */
memset(sampler, 0, sizeof(*sampler));
/* restore the linked list ptr */
sampler->nxt = nxtPtr;
/* now copy in the parameters */
sampler->agent = agent;
sampler->dsi = dsi;
/* set defaults */
sampler->sFlowFsMaximumHeaderSize = SFL_DEFAULT_HEADER_SIZE;
sampler->sFlowFsPacketSamplingRate = SFL_DEFAULT_SAMPLING_RATE;
}
/*_________________--------------------------__________________
_________________ reset __________________
-----------------__________________________------------------
*/
static void reset(SFLSampler *sampler)
{
SFLDataSource_instance dsi = sampler->dsi;
sfl_sampler_init(sampler, sampler->agent, &dsi);
}
/*_________________---------------------------__________________
_________________ MIB access __________________
-----------------___________________________------------------
*/
u_int32_t sfl_sampler_get_sFlowFsReceiver(SFLSampler *sampler) {
return sampler->sFlowFsReceiver;
}
void sfl_sampler_set_sFlowFsReceiver(SFLSampler *sampler, u_int32_t sFlowFsReceiver) {
sampler->sFlowFsReceiver = sFlowFsReceiver;
if(sFlowFsReceiver == 0) reset(sampler);
else {
/* retrieve and cache a direct pointer to my receiver */
sampler->myReceiver = sfl_agent_getReceiver(sampler->agent, sampler->sFlowFsReceiver);
}
}
u_int32_t sfl_sampler_get_sFlowFsPacketSamplingRate(SFLSampler *sampler) {
return sampler->sFlowFsPacketSamplingRate;
}
void sfl_sampler_set_sFlowFsPacketSamplingRate(SFLSampler *sampler, u_int32_t sFlowFsPacketSamplingRate) {
sampler->sFlowFsPacketSamplingRate = sFlowFsPacketSamplingRate;
}
u_int32_t sfl_sampler_get_sFlowFsMaximumHeaderSize(SFLSampler *sampler) {
return sampler->sFlowFsMaximumHeaderSize;
}
void sfl_sampler_set_sFlowFsMaximumHeaderSize(SFLSampler *sampler, u_int32_t sFlowFsMaximumHeaderSize) {
sampler->sFlowFsMaximumHeaderSize = sFlowFsMaximumHeaderSize;
}
/* call this to set a maximum samples-per-second threshold. If the sampler reaches this
threshold it will automatically back off the sampling rate. A value of 0 disables the
mechanism */
void sfl_sampler_set_backoffThreshold(SFLSampler *sampler, u_int32_t samplesPerSecond) {
sampler->backoffThreshold = samplesPerSecond;
}
u_int32_t sfl_sampler_get_backoffThreshold(SFLSampler *sampler) {
return sampler->backoffThreshold;
}
u_int32_t sfl_sampler_get_samplesLastTick(SFLSampler *sampler) {
return sampler->samplesLastTick;
}
/*_________________---------------------------------__________________
_________________ sequence number reset __________________
-----------------_________________________________------------------
Used by the agent to indicate a samplePool discontinuity
so that the sflow collector will know to ignore the next delta.
*/
void sfl_sampler_resetFlowSeqNo(SFLSampler *sampler) { sampler->flowSampleSeqNo = 0; }
/*_________________---------------------------__________________
_________________ sfl_sampler_tick __________________
-----------------___________________________------------------
*/
void sfl_sampler_tick(SFLSampler *sampler, time_t now)
{
if(sampler->backoffThreshold && sampler->samplesThisTick > sampler->backoffThreshold) {
/* automatic backoff. If using hardware sampling then this is where you have to
* call out to change the sampling rate and make sure that any other registers/variables
* that hold this value are updated.
*/
sampler->sFlowFsPacketSamplingRate *= 2;
}
sampler->samplesLastTick = sampler->samplesThisTick;
sampler->samplesThisTick = 0;
}
/*_________________------------------------------__________________
_________________ sfl_sampler_writeFlowSample __________________
-----------------______________________________------------------
*/
void sfl_sampler_writeFlowSample(SFLSampler *sampler, SFL_FLOW_SAMPLE_TYPE *fs)
{
if(fs == NULL) return;
sampler->samplesThisTick++;
/* increment the sequence number */
fs->sequence_number = ++sampler->flowSampleSeqNo;
/* copy the other header fields in */
#ifdef SFL_USE_32BIT_INDEX
fs->ds_class = SFL_DS_CLASS(sampler->dsi);
fs->ds_index = SFL_DS_INDEX(sampler->dsi);
#else
fs->source_id = SFL_DS_DATASOURCE(sampler->dsi);
#endif
/* the sampling rate may have been set already. */
if(fs->sampling_rate == 0) fs->sampling_rate = sampler->sFlowFsPacketSamplingRate;
/* the samplePool may be maintained upstream too. */
if( fs->sample_pool == 0) fs->sample_pool = sampler->samplePool;
/* sent to my receiver */
if(sampler->myReceiver) sfl_receiver_writeFlowSample(sampler->myReceiver, fs);
}
#ifdef SFLOW_SOFTWARE_SAMPLING
/* ================== software sampling ========================*/
/*_________________---------------------------__________________
_________________ nextRandomSkip __________________
-----------------___________________________------------------
*/
inline static u_int32_t nextRandomSkip(u_int32_t mean)
{
if(mean == 0 || mean == 1) return 1;
return ((random() % ((2 * mean) - 1)) + 1);
}
/*_________________---------------------------__________________
_________________ sfl_sampler_takeSample __________________
-----------------___________________________------------------
*/
int sfl_sampler_takeSample(SFLSampler *sampler)
{
if(sampler->skip == 0) {
/* first time - seed the random number generator */
srandom(SFL_DS_INDEX(sampler->dsi));
sampler->skip = nextRandomSkip(sampler->sFlowFsPacketSamplingRate);
}
/* increment the samplePool */
sampler->samplePool++;
if(--sampler->skip == 0) {
/* reached zero. Set the next skip and return true. */
sampler->skip = nextRandomSkip(sampler->sFlowFsPacketSamplingRate);
return 1;
}
return 0;
}
#endif /* SFLOW_SOFTWARE_SAMPLING */