[prev in list] [next in list] [prev in thread] [next in thread]
List: openvswitch-dev
Subject: [ovs-dev] [PATCH 1/3] Add InMon's sFlow Agent library to the build system.
From: blp () nicira ! com (Ben Pfaff)
Date: 2009-12-22 0:29:23
Message-ID: 1261441765-11381-1-git-send-email-blp () nicira ! com
[Download RAW message or body]
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.
---
COPYING | 4 +
acinclude.m4 | 10 +
configure.ac | 1 +
lib/automake.mk | 13 +
lib/sflow.h | 548 +++++++++++++++++++++++++++++++++
lib/sflow_agent.c | 492 +++++++++++++++++++++++++++++
lib/sflow_api.h | 340 ++++++++++++++++++++
lib/sflow_poller.c | 142 +++++++++
lib/sflow_receiver.c | 832 ++++++++++++++++++++++++++++++++++++++++++++++++++
lib/sflow_sampler.c | 183 +++++++++++
10 files changed, 2565 insertions(+), 0 deletions(-)
create mode 100644 lib/sflow.h
create mode 100644 lib/sflow_agent.c
create mode 100644 lib/sflow_api.h
create mode 100644 lib/sflow_poller.c
create mode 100644 lib/sflow_receiver.c
create mode 100644 lib/sflow_sampler.c
diff --git a/COPYING b/COPYING
index 134f02b..375efec 100644
--- a/COPYING
+++ b/COPYING
@@ -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
diff --git a/acinclude.m4 b/acinclude.m4
index e37a316..e074c7d 100644
--- a/acinclude.m4
+++ b/acinclude.m4
@@ -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 ----------------------------------------------------------------------
diff --git a/configure.ac b/configure.ac
index 300b21e..f901f96 100644
--- a/configure.ac
+++ b/configure.ac
@@ -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)
diff --git a/lib/automake.mk b/lib/automake.mk
index 9ba513a..252abdf 100644
--- a/lib/automake.mk
+++ b/lib/automake.mk
@@ -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 \
diff --git a/lib/sflow.h b/lib/sflow.h
new file mode 100644
index 0000000..397ae2d
--- /dev/null
+++ b/lib/sflow.h
@@ -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 */
diff --git a/lib/sflow_agent.c b/lib/sflow_agent.c
new file mode 100644
index 0000000..4b25c25
--- /dev/null
+++ b/lib/sflow_agent.c
@@ -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);
+}
diff --git a/lib/sflow_api.h b/lib/sflow_api.h
new file mode 100644
index 0000000..be8d997
--- /dev/null
+++ b/lib/sflow_api.h
@@ -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 */
+
+
diff --git a/lib/sflow_poller.c b/lib/sflow_poller.c
new file mode 100644
index 0000000..ffd09d3
--- /dev/null
+++ b/lib/sflow_poller.c
@@ -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);
+}
+
diff --git a/lib/sflow_receiver.c b/lib/sflow_receiver.c
new file mode 100644
index 0000000..7fccab3
--- /dev/null
+++ b/lib/sflow_receiver.c
@@ -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);
+}
diff --git a/lib/sflow_sampler.c b/lib/sflow_sampler.c
new file mode 100644
index 0000000..759b5a2
--- /dev/null
+++ b/lib/sflow_sampler.c
@@ -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 */
--
1.6.3.3
[prev in list] [next in list] [prev in thread] [next in thread]
Configure |
About |
News |
Add a list |
Sponsored by KoreLogic