From c72e245a0e2c979f4c14279b56183e2200905c6c Mon Sep 17 00:00:00 2001 From: Ben Pfaff Date: Fri, 20 Nov 2009 09:45:26 -0800 Subject: [PATCH] Add InMon's sFlow Agent library to the build system. 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(+) 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 134f02b0f..375efecae 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 e37a316df..e074c7d47 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 300b21e2c..f901f96a9 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 9ba513acb..252abdf15 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 000000000..397ae2dae --- /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 000000000..4b25c25a9 --- /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 000000000..be8d99773 --- /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 +#include +#include +#include +#include +#include +#include /* for htonl */ + +#ifdef SFLOW_DO_SOCKET +#include +#include +#include +#include +#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 000000000..ffd09d3c2 --- /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 000000000..7fccab30f --- /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 +#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 000000000..759b5a22c --- /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 */ -- 2.39.5