diff options
Diffstat (limited to 'src/plugins/flowprobe')
| -rw-r--r-- | src/plugins/flowprobe/flowprobe.api | 40 | ||||
| -rw-r--r-- | src/plugins/flowprobe/flowprobe.c | 1122 | ||||
| -rw-r--r-- | src/plugins/flowprobe/flowprobe.h | 167 | ||||
| -rw-r--r-- | src/plugins/flowprobe/flowprobe_all_api_h.h | 18 | ||||
| -rw-r--r-- | src/plugins/flowprobe/flowprobe_msg_enum.h | 31 | ||||
| -rw-r--r-- | src/plugins/flowprobe/flowprobe_plugin_doc.md | 13 | ||||
| -rw-r--r-- | src/plugins/flowprobe/flowprobe_test.c | 263 | ||||
| -rw-r--r-- | src/plugins/flowprobe/node.c | 998 |
8 files changed, 2652 insertions, 0 deletions
diff --git a/src/plugins/flowprobe/flowprobe.api b/src/plugins/flowprobe/flowprobe.api new file mode 100644 index 00000000000..3f8c583bcf2 --- /dev/null +++ b/src/plugins/flowprobe/flowprobe.api @@ -0,0 +1,40 @@ +/* Define a simple enable-disable binary API to control the feature */ + +/** \file + This file defines the vpp control-plane API messages + used to control the flowprobe plugin +*/ + +/** \brief Enable / disable per-packet IPFIX recording on an interface + @param client_index - opaque cookie to identify the sender + @param context - sender context, to match reply w/ request + @param is_add - add address if non-zero, else delete + @param is_ipv6 - if non-zero the address is ipv6, else ipv4 + @param sw_if_index - index of the interface +*/ +autoreply manual_print define flowprobe_tx_interface_add_del +{ + /* Client identifier, set from api_main.my_client_index */ + u32 client_index; + + /* Arbitrary context, so client can match reply to request */ + u32 context; + + /* Enable / disable the feature */ + u8 is_add; + u8 which; /* 0 = ipv4, 1 = l2, 2 = ipv6 */ + + /* Interface handle */ + u32 sw_if_index; +}; + +autoreply define flowprobe_params +{ + u32 client_index; + u32 context; + u8 record_l2; + u8 record_l3; + u8 record_l4; + u32 active_timer; /* ~0 is off, 0 is default */ + u32 passive_timer; /* ~0 is off, 0 is default */ +}; diff --git a/src/plugins/flowprobe/flowprobe.c b/src/plugins/flowprobe/flowprobe.c new file mode 100644 index 00000000000..8975f89cddf --- /dev/null +++ b/src/plugins/flowprobe/flowprobe.c @@ -0,0 +1,1122 @@ +/* + * flowprobe.c - ipfix probe plugin + * + * Copyright (c) 2016 Cisco and/or its affiliates. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/** + * @file + * @brief Per-packet IPFIX flow record generator plugin + * + * This file implements vpp plugin registration mechanics, + * debug CLI, and binary API handling. + */ + +#include <vnet/vnet.h> +#include <vpp/app/version.h> +#include <vnet/plugin/plugin.h> +#include <flowprobe/flowprobe.h> + +#include <vlibapi/api.h> +#include <vlibmemory/api.h> +#include <vlibsocket/api.h> + +/* define message IDs */ +#include <flowprobe/flowprobe_msg_enum.h> + +/* define message structures */ +#define vl_typedefs +#include <flowprobe/flowprobe_all_api_h.h> +#undef vl_typedefs + +/* define generated endian-swappers */ +#define vl_endianfun +#include <flowprobe/flowprobe_all_api_h.h> +#undef vl_endianfun + +/* instantiate all the print functions we know about */ +#define vl_print(handle, ...) vlib_cli_output (handle, __VA_ARGS__) +#define vl_printfun +#include <flowprobe/flowprobe_all_api_h.h> +#undef vl_printfun + +flowprobe_main_t flowprobe_main; +vlib_node_registration_t flowprobe_walker_node; +static vlib_node_registration_t flowprobe_timer_node; +uword flowprobe_walker_process (vlib_main_t * vm, vlib_node_runtime_t * rt, + vlib_frame_t * f); + +/* Get the API version number */ +#define vl_api_version(n,v) static u32 api_version=(v); +#include <flowprobe/flowprobe_all_api_h.h> +#undef vl_api_version + +#define REPLY_MSG_ID_BASE fm->msg_id_base +#include <vlibapi/api_helper_macros.h> + +/* Define the per-interface configurable features */ +/* *INDENT-OFF* */ +VNET_FEATURE_INIT (flow_perpacket_ip4, static) = +{ + .arc_name = "ip4-output", + .node_name = "flowprobe-ip4", + .runs_before = VNET_FEATURES ("interface-output"), +}; + +VNET_FEATURE_INIT (flow_perpacket_ip6, static) = +{ + .arc_name = "ip6-output", + .node_name = "flowprobe-ip6", + .runs_before = VNET_FEATURES ("interface-output"), +}; + +VNET_FEATURE_INIT (flow_perpacket_l2, static) = +{ + .arc_name = "interface-output", + .node_name = "flowprobe-l2", + .runs_before = VNET_FEATURES ("interface-tx"), +}; +/* *INDENT-ON* */ + +/* Macro to finish up custom dump fns */ +#define FINISH \ + vec_add1 (s, 0); \ + vl_print (handle, (char *)s); \ + vec_free (s); \ + return handle; + +static inline ipfix_field_specifier_t * +flowprobe_template_ip4_fields (ipfix_field_specifier_t * f) +{ +#define flowprobe_template_ip4_field_count() 4 + /* sourceIpv4Address, TLV type 8, u32 */ + f->e_id_length = ipfix_e_id_length (0 /* enterprise */ , + sourceIPv4Address, 4); + f++; + /* destinationIPv4Address, TLV type 12, u32 */ + f->e_id_length = ipfix_e_id_length (0 /* enterprise */ , + destinationIPv4Address, 4); + f++; + /* protocolIdentifier, TLV type 4, u8 */ + f->e_id_length = ipfix_e_id_length (0 /* enterprise */ , + protocolIdentifier, 1); + f++; + /* octetDeltaCount, TLV type 1, u64 */ + f->e_id_length = ipfix_e_id_length (0 /* enterprise */ , + octetDeltaCount, 8); + f++; + return f; +} + +static inline ipfix_field_specifier_t * +flowprobe_template_ip6_fields (ipfix_field_specifier_t * f) +{ +#define flowprobe_template_ip6_field_count() 4 + /* sourceIpv6Address, TLV type 27, 16 octets */ + f->e_id_length = ipfix_e_id_length (0 /* enterprise */ , + sourceIPv6Address, 16); + f++; + /* destinationIPv6Address, TLV type 28, 16 octets */ + f->e_id_length = ipfix_e_id_length (0 /* enterprise */ , + destinationIPv6Address, 16); + f++; + /* protocolIdentifier, TLV type 4, u8 */ + f->e_id_length = ipfix_e_id_length (0 /* enterprise */ , + protocolIdentifier, 1); + f++; + /* octetDeltaCount, TLV type 1, u64 */ + f->e_id_length = ipfix_e_id_length (0 /* enterprise */ , + octetDeltaCount, 8); + f++; + return f; +} + +static inline ipfix_field_specifier_t * +flowprobe_template_l2_fields (ipfix_field_specifier_t * f) +{ +#define flowprobe_template_l2_field_count() 3 + /* sourceMacAddress, TLV type 56, u8[6] we hope */ + f->e_id_length = ipfix_e_id_length (0 /* enterprise */ , + sourceMacAddress, 6); + f++; + /* destinationMacAddress, TLV type 80, u8[6] we hope */ + f->e_id_length = ipfix_e_id_length (0 /* enterprise */ , + destinationMacAddress, 6); + f++; + /* ethernetType, TLV type 256, u16 */ + f->e_id_length = ipfix_e_id_length (0 /* enterprise */ , + ethernetType, 2); + f++; + return f; +} + +static inline ipfix_field_specifier_t * +flowprobe_template_common_fields (ipfix_field_specifier_t * f) +{ +#define flowprobe_template_common_field_count() 3 + /* ingressInterface, TLV type 10, u32 */ + f->e_id_length = ipfix_e_id_length (0 /* enterprise */ , + ingressInterface, 4); + f++; + + /* egressInterface, TLV type 14, u32 */ + f->e_id_length = ipfix_e_id_length (0 /* enterprise */ , + egressInterface, 4); + f++; + + /* packetDeltaCount, TLV type 2, u64 */ + f->e_id_length = ipfix_e_id_length (0 /* enterprise */ , + packetDeltaCount, 8); + f++; + + return f; +} + +static inline ipfix_field_specifier_t * +flowprobe_template_l4_fields (ipfix_field_specifier_t * f) +{ +#define flowprobe_template_l4_field_count() 2 + /* sourceTransportPort, TLV type 7, u16 */ + f->e_id_length = ipfix_e_id_length (0 /* enterprise */ , + sourceTransportPort, 2); + f++; + /* destinationTransportPort, TLV type 11, u16 */ + f->e_id_length = ipfix_e_id_length (0 /* enterprise */ , + destinationTransportPort, 2); + f++; + return f; +} + +/** + * @brief Create an IPFIX template packet rewrite string + * @param frm flow_report_main_t * + * @param fr flow_report_t * + * @param collector_address ip4_address_t * the IPFIX collector address + * @param src_address ip4_address_t * the source address we should use + * @param collector_port u16 the collector port we should use, host byte order + * @returns u8 * vector containing the indicated IPFIX template packet + */ +static inline u8 * +flowprobe_template_rewrite_inline (flow_report_main_t * frm, + flow_report_t * fr, + ip4_address_t * collector_address, + ip4_address_t * src_address, + u16 collector_port, + flowprobe_variant_t which) +{ + ip4_header_t *ip; + udp_header_t *udp; + ipfix_message_header_t *h; + ipfix_set_header_t *s; + ipfix_template_header_t *t; + ipfix_field_specifier_t *f; + ipfix_field_specifier_t *first_field; + u8 *rewrite = 0; + ip4_ipfix_template_packet_t *tp; + u32 field_count = 0; + flow_report_stream_t *stream; + flowprobe_main_t *fm = &flowprobe_main; + flowprobe_record_t flags = fr->opaque.as_uword; + bool collect_ip4 = false, collect_ip6 = false; + + stream = &frm->streams[fr->stream_index]; + + if (flags & FLOW_RECORD_L3) + { + collect_ip4 = which == FLOW_VARIANT_L2_IP4 || which == FLOW_VARIANT_IP4; + collect_ip6 = which == FLOW_VARIANT_L2_IP6 || which == FLOW_VARIANT_IP6; + if (which == FLOW_VARIANT_L2_IP4) + flags |= FLOW_RECORD_L2_IP4; + if (which == FLOW_VARIANT_L2_IP6) + flags |= FLOW_RECORD_L2_IP6; + } + + field_count += flowprobe_template_common_field_count (); + if (flags & FLOW_RECORD_L2) + field_count += flowprobe_template_l2_field_count (); + if (collect_ip4) + field_count += flowprobe_template_ip4_field_count (); + if (collect_ip6) + field_count += flowprobe_template_ip6_field_count (); + if (flags & FLOW_RECORD_L4) + field_count += flowprobe_template_l4_field_count (); + + /* allocate rewrite space */ + vec_validate_aligned + (rewrite, sizeof (ip4_ipfix_template_packet_t) + + field_count * sizeof (ipfix_field_specifier_t) - 1, + CLIB_CACHE_LINE_BYTES); + + tp = (ip4_ipfix_template_packet_t *) rewrite; + ip = (ip4_header_t *) & tp->ip4; + udp = (udp_header_t *) (ip + 1); + h = (ipfix_message_header_t *) (udp + 1); + s = (ipfix_set_header_t *) (h + 1); + t = (ipfix_template_header_t *) (s + 1); + first_field = f = (ipfix_field_specifier_t *) (t + 1); + + ip->ip_version_and_header_length = 0x45; + ip->ttl = 254; + ip->protocol = IP_PROTOCOL_UDP; + ip->src_address.as_u32 = src_address->as_u32; + ip->dst_address.as_u32 = collector_address->as_u32; + udp->src_port = clib_host_to_net_u16 (stream->src_port); + udp->dst_port = clib_host_to_net_u16 (collector_port); + udp->length = clib_host_to_net_u16 (vec_len (rewrite) - sizeof (*ip)); + + /* FIXUP: message header export_time */ + /* FIXUP: message header sequence_number */ + h->domain_id = clib_host_to_net_u32 (stream->domain_id); + + /* Add TLVs to the template */ + f = flowprobe_template_common_fields (f); + + if (flags & FLOW_RECORD_L2) + f = flowprobe_template_l2_fields (f); + if (collect_ip4) + f = flowprobe_template_ip4_fields (f); + if (collect_ip6) + f = flowprobe_template_ip6_fields (f); + if (flags & FLOW_RECORD_L4) + f = flowprobe_template_l4_fields (f); + + /* Back to the template packet... */ + ip = (ip4_header_t *) & tp->ip4; + udp = (udp_header_t *) (ip + 1); + + ASSERT (f - first_field); + /* Field count in this template */ + t->id_count = ipfix_id_count (fr->template_id, f - first_field); + + fm->template_size[flags] = (u8 *) f - (u8 *) s; + + /* set length in octets */ + s->set_id_length = + ipfix_set_id_length (2 /* set_id */ , (u8 *) f - (u8 *) s); + + /* message length in octets */ + h->version_length = version_length ((u8 *) f - (u8 *) h); + + ip->length = clib_host_to_net_u16 ((u8 *) f - (u8 *) ip); + ip->checksum = ip4_header_checksum (ip); + + return rewrite; +} + +static u8 * +flowprobe_template_rewrite_ip6 (flow_report_main_t * frm, + flow_report_t * fr, + ip4_address_t * collector_address, + ip4_address_t * src_address, + u16 collector_port) +{ + return flowprobe_template_rewrite_inline + (frm, fr, collector_address, src_address, collector_port, + FLOW_VARIANT_IP6); +} + +static u8 * +flowprobe_template_rewrite_ip4 (flow_report_main_t * frm, + flow_report_t * fr, + ip4_address_t * collector_address, + ip4_address_t * src_address, + u16 collector_port) +{ + return flowprobe_template_rewrite_inline + (frm, fr, collector_address, src_address, collector_port, + FLOW_VARIANT_IP4); +} + +static u8 * +flowprobe_template_rewrite_l2 (flow_report_main_t * frm, + flow_report_t * fr, + ip4_address_t * collector_address, + ip4_address_t * src_address, + u16 collector_port) +{ + return flowprobe_template_rewrite_inline + (frm, fr, collector_address, src_address, collector_port, + FLOW_VARIANT_L2); +} + +static u8 * +flowprobe_template_rewrite_l2_ip4 (flow_report_main_t * frm, + flow_report_t * fr, + ip4_address_t * collector_address, + ip4_address_t * src_address, + u16 collector_port) +{ + return flowprobe_template_rewrite_inline + (frm, fr, collector_address, src_address, collector_port, + FLOW_VARIANT_L2_IP4); +} + +static u8 * +flowprobe_template_rewrite_l2_ip6 (flow_report_main_t * frm, + flow_report_t * fr, + ip4_address_t * collector_address, + ip4_address_t * src_address, + u16 collector_port) +{ + return flowprobe_template_rewrite_inline + (frm, fr, collector_address, src_address, collector_port, + FLOW_VARIANT_L2_IP6); +} + +/** + * @brief Flush accumulated data + * @param frm flow_report_main_t * + * @param fr flow_report_t * + * @param f vlib_frame_t * + * + * <em>Notes:</em> + * This function must simply return the incoming frame, or no template packets + * will be sent. + */ +vlib_frame_t * +flowprobe_data_callback_ip4 (flow_report_main_t * frm, + flow_report_t * fr, + vlib_frame_t * f, u32 * to_next, u32 node_index) +{ + flowprobe_flush_callback_ip4 (); + return f; +} + +vlib_frame_t * +flowprobe_data_callback_ip6 (flow_report_main_t * frm, + flow_report_t * fr, + vlib_frame_t * f, u32 * to_next, u32 node_index) +{ + flowprobe_flush_callback_ip6 (); + return f; +} + +vlib_frame_t * +flowprobe_data_callback_l2 (flow_report_main_t * frm, + flow_report_t * fr, + vlib_frame_t * f, u32 * to_next, u32 node_index) +{ + flowprobe_flush_callback_l2 (); + return f; +} + +static int +flowprobe_template_add_del (u32 domain_id, u16 src_port, + flowprobe_record_t flags, + vnet_flow_data_callback_t * flow_data_callback, + vnet_flow_rewrite_callback_t * rewrite_callback, + bool is_add, u16 * template_id) +{ + flow_report_main_t *frm = &flow_report_main; + vnet_flow_report_add_del_args_t a = { + .rewrite_callback = rewrite_callback, + .flow_data_callback = flow_data_callback, + .is_add = is_add, + .domain_id = domain_id, + .src_port = src_port, + .opaque.as_uword = flags, + }; + return vnet_flow_report_add_del (frm, &a, template_id); +} + +static void +flowprobe_expired_timer_callback (u32 * expired_timers) +{ + vlib_main_t *vm = vlib_get_main (); + flowprobe_main_t *fm = &flowprobe_main; + u32 my_cpu_number = vm->thread_index; + int i; + u32 poolindex; + + for (i = 0; i < vec_len (expired_timers); i++) + { + poolindex = expired_timers[i] & 0x7FFFFFFF; + vec_add1 (fm->expired_passive_per_worker[my_cpu_number], poolindex); + } +} + +static clib_error_t * +flowprobe_create_state_tables (u32 active_timer) +{ + flowprobe_main_t *fm = &flowprobe_main; + vlib_thread_main_t *tm = &vlib_thread_main; + vlib_main_t *vm = vlib_get_main (); + clib_error_t *error = 0; + u32 num_threads; + int i; + + /* Decide how many worker threads we have */ + num_threads = 1 /* main thread */ + tm->n_threads; + + /* Hash table per worker */ + fm->ht_log2len = FLOWPROBE_LOG2_HASHSIZE; + + /* Init per worker flow state and timer wheels */ + if (active_timer) + { + vec_validate (fm->timers_per_worker, num_threads - 1); + vec_validate (fm->expired_passive_per_worker, num_threads - 1); + vec_validate (fm->hash_per_worker, num_threads - 1); + vec_validate (fm->pool_per_worker, num_threads - 1); + + for (i = 0; i < num_threads; i++) + { + int j; + pool_alloc (fm->pool_per_worker[i], 1 << fm->ht_log2len); + vec_resize (fm->hash_per_worker[i], 1 << fm->ht_log2len); + for (j = 0; j < (1 << fm->ht_log2len); j++) + fm->hash_per_worker[i][j] = ~0; + fm->timers_per_worker[i] = + clib_mem_alloc (sizeof (TWT (tw_timer_wheel))); + tw_timer_wheel_init_2t_1w_2048sl (fm->timers_per_worker[i], + flowprobe_expired_timer_callback, + 1.0, 1024); + } + fm->disabled = true; + } + else + { + f64 now = vlib_time_now (vm); + vec_validate (fm->stateless_entry, num_threads - 1); + for (i = 0; i < num_threads; i++) + fm->stateless_entry[i].last_exported = now; + fm->disabled = false; + } + fm->initialized = true; + return error; +} + +static int +validate_feature_on_interface (flowprobe_main_t * fm, u32 sw_if_index, + u8 which) +{ + vec_validate_init_empty (fm->flow_per_interface, sw_if_index, ~0); + + if (fm->flow_per_interface[sw_if_index] == (u8) ~ 0) + return -1; + else if (fm->flow_per_interface[sw_if_index] != which) + return 0; + else + return 1; +} + +/** + * @brief configure / deconfigure the IPFIX flow-per-packet + * @param fm flowprobe_main_t * fm + * @param sw_if_index u32 the desired interface + * @param is_add int 1 to enable the feature, 0 to disable it + * @returns 0 if successful, non-zero otherwise + */ + +static int +flowprobe_tx_interface_add_del_feature (flowprobe_main_t * fm, + u32 sw_if_index, u8 which, int is_add) +{ + vlib_main_t *vm = vlib_get_main (); + int rv = 0; + u16 template_id = 0; + flowprobe_record_t flags = fm->record; + + fm->flow_per_interface[sw_if_index] = (is_add) ? which : (u8) ~ 0; + fm->template_per_flow[which] += (is_add) ? 1 : -1; + if (is_add && fm->template_per_flow[which] > 1) + template_id = fm->template_reports[flags]; + + if ((is_add && fm->template_per_flow[which] == 1) || + (!is_add && fm->template_per_flow[which] == 0)) + { + if (which == FLOW_VARIANT_L2) + { + if (fm->record & FLOW_RECORD_L2) + { + rv = flowprobe_template_add_del (1, UDP_DST_PORT_ipfix, flags, + flowprobe_data_callback_l2, + flowprobe_template_rewrite_l2, + is_add, &template_id); + } + if (fm->record & FLOW_RECORD_L3 || fm->record & FLOW_RECORD_L4) + { + rv = flowprobe_template_add_del (1, UDP_DST_PORT_ipfix, flags, + flowprobe_data_callback_l2, + flowprobe_template_rewrite_l2_ip4, + is_add, &template_id); + fm->template_reports[flags | FLOW_RECORD_L2_IP4] = + (is_add) ? template_id : 0; + rv = + flowprobe_template_add_del (1, UDP_DST_PORT_ipfix, flags, + flowprobe_data_callback_l2, + flowprobe_template_rewrite_l2_ip6, + is_add, &template_id); + fm->template_reports[flags | FLOW_RECORD_L2_IP6] = + (is_add) ? template_id : 0; + + /* Special case L2 */ + fm->context[FLOW_VARIANT_L2_IP4].flags = + flags | FLOW_RECORD_L2_IP4; + fm->context[FLOW_VARIANT_L2_IP6].flags = + flags | FLOW_RECORD_L2_IP6; + + fm->template_reports[flags] = template_id; + } + } + else if (which == FLOW_VARIANT_IP4) + rv = flowprobe_template_add_del (1, UDP_DST_PORT_ipfix, flags, + flowprobe_data_callback_ip4, + flowprobe_template_rewrite_ip4, + is_add, &template_id); + else if (which == FLOW_VARIANT_IP6) + rv = flowprobe_template_add_del (1, UDP_DST_PORT_ipfix, flags, + flowprobe_data_callback_ip6, + flowprobe_template_rewrite_ip6, + is_add, &template_id); + } + if (rv && rv != VNET_API_ERROR_VALUE_EXIST) + { + clib_warning ("vnet_flow_report_add_del returned %d", rv); + return -1; + } + + if (which != (u8) ~ 0) + { + fm->context[which].flags = fm->record; + fm->template_reports[flags] = (is_add) ? template_id : 0; + } + + if (which == FLOW_VARIANT_IP4) + vnet_feature_enable_disable ("ip4-output", "flowprobe-ip4", + sw_if_index, is_add, 0, 0); + else if (which == FLOW_VARIANT_IP6) + vnet_feature_enable_disable ("ip6-output", "flowprobe-ip6", + sw_if_index, is_add, 0, 0); + else if (which == FLOW_VARIANT_L2) + vnet_feature_enable_disable ("interface-output", "flowprobe-l2", + sw_if_index, is_add, 0, 0); + + /* Stateful flow collection */ + if (is_add && !fm->initialized) + { + flowprobe_create_state_tables (fm->active_timer); + if (fm->active_timer) + vlib_process_signal_event (vm, flowprobe_timer_node.index, 1, 0); + } + + return 0; +} + +/** + * @brief API message handler + * @param mp vl_api_flowprobe_tx_interface_add_del_t * mp the api message + */ +void vl_api_flowprobe_tx_interface_add_del_t_handler + (vl_api_flowprobe_tx_interface_add_del_t * mp) +{ + flowprobe_main_t *fm = &flowprobe_main; + vl_api_flowprobe_tx_interface_add_del_reply_t *rmp; + u32 sw_if_index = ntohl (mp->sw_if_index); + int rv = 0; + + VALIDATE_SW_IF_INDEX (mp); + + if (mp->which != FLOW_VARIANT_IP4 && mp->which != FLOW_VARIANT_L2 + && mp->which != FLOW_VARIANT_IP6) + { + rv = VNET_API_ERROR_UNIMPLEMENTED; + goto out; + } + + if (fm->record == 0) + { + clib_warning ("Please specify flowprobe params record first..."); + rv = VNET_API_ERROR_CANNOT_ENABLE_DISABLE_FEATURE; + goto out; + } + + rv = validate_feature_on_interface (fm, sw_if_index, mp->which); + if ((rv == 1 && mp->is_add == 1) || rv == 0) + { + rv = VNET_API_ERROR_CANNOT_ENABLE_DISABLE_FEATURE; + goto out; + } + + rv = flowprobe_tx_interface_add_del_feature + (fm, sw_if_index, mp->which, mp->is_add); + +out: + BAD_SW_IF_INDEX_LABEL; + + REPLY_MACRO (VL_API_FLOWPROBE_TX_INTERFACE_ADD_DEL_REPLY); +} + +/** + * @brief API message custom-dump function + * @param mp vl_api_flowprobe_tx_interface_add_del_t * mp the api message + * @param handle void * print function handle + * @returns u8 * output string + */ +static void *vl_api_flowprobe_tx_interface_add_del_t_print + (vl_api_flowprobe_tx_interface_add_del_t * mp, void *handle) +{ + u8 *s; + + s = format (0, "SCRIPT: flowprobe_tx_interface_add_del "); + s = format (s, "sw_if_index %d is_add %d which %d ", + clib_host_to_net_u32 (mp->sw_if_index), + (int) mp->is_add, (int) mp->which); + FINISH; +} + +#define vec_neg_search(v,E) \ +({ \ + word _v(i) = 0; \ + while (_v(i) < vec_len(v) && v[_v(i)] == E) \ + { \ + _v(i)++; \ + } \ + if (_v(i) == vec_len(v)) \ + _v(i) = ~0; \ + _v(i); \ +}) + +static int +flowprobe_params (flowprobe_main_t * fm, u8 record_l2, + u8 record_l3, u8 record_l4, + u32 active_timer, u32 passive_timer) +{ + flowprobe_record_t flags = 0; + + if (vec_neg_search (fm->flow_per_interface, (u8) ~ 0) != ~0) + return ~0; + + if (record_l2) + flags |= FLOW_RECORD_L2; + if (record_l3) + flags |= FLOW_RECORD_L3; + if (record_l4) + flags |= FLOW_RECORD_L4; + + fm->record = flags; + + /* + * Timers: ~0 is default, 0 is off + */ + fm->active_timer = + (active_timer == (u32) ~ 0 ? FLOWPROBE_TIMER_ACTIVE : active_timer); + fm->passive_timer = + (passive_timer == (u32) ~ 0 ? FLOWPROBE_TIMER_PASSIVE : passive_timer); + + return 0; +} + +void +vl_api_flowprobe_params_t_handler (vl_api_flowprobe_params_t * mp) +{ + flowprobe_main_t *fm = &flowprobe_main; + vl_api_flowprobe_params_reply_t *rmp; + int rv = 0; + + rv = flowprobe_params + (fm, mp->record_l2, mp->record_l3, mp->record_l4, + clib_net_to_host_u32 (mp->active_timer), + clib_net_to_host_u32 (mp->passive_timer)); + + REPLY_MACRO (VL_API_FLOWPROBE_PARAMS_REPLY); +} + +/* List of message types that this plugin understands */ +#define foreach_flowprobe_plugin_api_msg \ +_(FLOWPROBE_TX_INTERFACE_ADD_DEL, flowprobe_tx_interface_add_del) \ +_(FLOWPROBE_PARAMS, flowprobe_params) + +/* *INDENT-OFF* */ +VLIB_PLUGIN_REGISTER () = { + .version = VPP_BUILD_VER, + .description = "Flow per Packet", +}; +/* *INDENT-ON* */ + +u8 * +format_flowprobe_entry (u8 * s, va_list * args) +{ + flowprobe_entry_t *e = va_arg (*args, flowprobe_entry_t *); + s = format (s, " %d/%d", e->key.rx_sw_if_index, e->key.tx_sw_if_index); + + s = format (s, " %U %U", format_ethernet_address, &e->key.src_mac, + format_ethernet_address, &e->key.dst_mac); + s = format (s, " %U -> %U", + format_ip46_address, &e->key.src_address, IP46_TYPE_ANY, + format_ip46_address, &e->key.dst_address, IP46_TYPE_ANY); + s = format (s, " %d", e->key.protocol); + s = format (s, " %d %d\n", clib_net_to_host_u16 (e->key.src_port), + clib_net_to_host_u16 (e->key.dst_port)); + + return s; +} + +static clib_error_t * +flowprobe_show_table_fn (vlib_main_t * vm, + unformat_input_t * input, vlib_cli_command_t * cm) +{ + flowprobe_main_t *fm = &flowprobe_main; + int i; + flowprobe_entry_t *e; + + vlib_cli_output (vm, "Dumping IPFIX table"); + + for (i = 0; i < vec_len (fm->pool_per_worker); i++) + { + /* *INDENT-OFF* */ + pool_foreach (e, fm->pool_per_worker[i], ( + { + vlib_cli_output (vm, "%U", + format_flowprobe_entry, + e); + })); + /* *INDENT-ON* */ + + } + return 0; +} + +static clib_error_t * +flowprobe_show_stats_fn (vlib_main_t * vm, + unformat_input_t * input, vlib_cli_command_t * cm) +{ + flowprobe_main_t *fm = &flowprobe_main; + int i; + + vlib_cli_output (vm, "IPFIX table statistics"); + vlib_cli_output (vm, "Flow entry size: %d\n", sizeof (flowprobe_entry_t)); + vlib_cli_output (vm, "Flow pool size per thread: %d\n", + 0x1 << FLOWPROBE_LOG2_HASHSIZE); + + for (i = 0; i < vec_len (fm->pool_per_worker); i++) + vlib_cli_output (vm, "Pool utilisation thread %d is %d%%\n", i, + (100 * pool_elts (fm->pool_per_worker[i])) / + (0x1 << FLOWPROBE_LOG2_HASHSIZE)); + return 0; +} + +static clib_error_t * +flowprobe_tx_interface_add_del_feature_command_fn (vlib_main_t * vm, + unformat_input_t * input, + vlib_cli_command_t * cmd) +{ + flowprobe_main_t *fm = &flowprobe_main; + u32 sw_if_index = ~0; + int is_add = 1; + u8 which = FLOW_VARIANT_IP4; + int rv; + + while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT) + { + if (unformat (input, "disable")) + is_add = 0; + else if (unformat (input, "%U", unformat_vnet_sw_interface, + fm->vnet_main, &sw_if_index)); + else if (unformat (input, "ip4")) + which = FLOW_VARIANT_IP4; + else if (unformat (input, "ip6")) + which = FLOW_VARIANT_IP6; + else if (unformat (input, "l2")) + which = FLOW_VARIANT_L2; + else + break; + } + + if (fm->record == 0) + return clib_error_return (0, + "Please specify flowprobe params record first..."); + + if (sw_if_index == ~0) + return clib_error_return (0, "Please specify an interface..."); + + rv = validate_feature_on_interface (fm, sw_if_index, which); + if (rv == 1) + { + if (is_add) + return clib_error_return (0, + "Datapath is already enabled for given interface..."); + } + else if (rv == 0) + return clib_error_return (0, + "Interface has enable different datapath ..."); + + rv = + flowprobe_tx_interface_add_del_feature (fm, sw_if_index, which, is_add); + switch (rv) + { + case 0: + break; + + case VNET_API_ERROR_INVALID_SW_IF_INDEX: + return clib_error_return + (0, "Invalid interface, only works on physical ports"); + break; + + case VNET_API_ERROR_UNIMPLEMENTED: + return clib_error_return (0, "ip6 not supported"); + break; + + default: + return clib_error_return (0, "flowprobe_enable_disable returned %d", + rv); + } + return 0; +} + +static clib_error_t * +flowprobe_params_command_fn (vlib_main_t * vm, + unformat_input_t * input, + vlib_cli_command_t * cmd) +{ + flowprobe_main_t *fm = &flowprobe_main; + bool record_l2 = false, record_l3 = false, record_l4 = false; + u32 active_timer = ~0; + u32 passive_timer = ~0; + + while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT) + { + if (unformat (input, "active %d", &active_timer)) + ; + else if (unformat (input, "passive %d", &passive_timer)) + ; + else if (unformat (input, "record")) + while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT) + { + if (unformat (input, "l2")) + record_l2 = true; + else if (unformat (input, "l3")) + record_l3 = true; + else if (unformat (input, "l4")) + record_l4 = true; + else + break; + } + else + break; + } + + if (passive_timer > 0 && active_timer > passive_timer) + return clib_error_return (0, + "Passive timer has to be greater than active one..."); + + if (flowprobe_params (fm, record_l2, record_l3, record_l4, + active_timer, passive_timer)) + return clib_error_return (0, + "Couldn't change flowperpacket params when feature is enabled on some interface ..."); + return 0; +} + +/*? + * '<em>flowprobe feature add-del</em>' commands to enable/disable + * per-packet IPFIX flow record generation on an interface + * + * @cliexpar + * @parblock + * To enable per-packet IPFIX flow-record generation on an interface: + * @cliexcmd{flowprobe feature add-del GigabitEthernet2/0/0} + * + * To disable per-packet IPFIX flow-record generation on an interface: + * @cliexcmd{flowprobe feature add-del GigabitEthernet2/0/0 disable} + * @cliexend + * @endparblock +?*/ +/* *INDENT-OFF* */ +VLIB_CLI_COMMAND (flowprobe_enable_disable_command, static) = { + .path = "flowprobe feature add-del", + .short_help = + "flowprobe feature add-del <interface-name> <l2|ip4|ip6> disable", + .function = flowprobe_tx_interface_add_del_feature_command_fn, +}; +VLIB_CLI_COMMAND (flowprobe_params_command, static) = { + .path = "flowprobe params", + .short_help = + "flowprobe params record <[l2] [l3] [l4]> [active <timer> passive <timer>]", + .function = flowprobe_params_command_fn, +}; +VLIB_CLI_COMMAND (flowprobe_show_table_command, static) = { + .path = "show flowprobe table", + .short_help = "show flowprobe table", + .function = flowprobe_show_table_fn, +}; +VLIB_CLI_COMMAND (flowprobe_show_stats_command, static) = { + .path = "show flowprobe statistics", + .short_help = "show flowprobe statistics", + .function = flowprobe_show_stats_fn, +}; +/* *INDENT-ON* */ + +/** + * @brief Set up the API message handling tables + * @param vm vlib_main_t * vlib main data structure pointer + * @returns 0 to indicate all is well + */ +static clib_error_t * +flowprobe_plugin_api_hookup (vlib_main_t * vm) +{ + flowprobe_main_t *fm = &flowprobe_main; +#define _(N,n) \ + vl_msg_api_set_handlers((VL_API_##N + fm->msg_id_base), \ + #n, \ + vl_api_##n##_t_handler, \ + vl_noop_handler, \ + vl_api_##n##_t_endian, \ + vl_api_##n##_t_print, \ + sizeof(vl_api_##n##_t), 1); + foreach_flowprobe_plugin_api_msg; +#undef _ + + return 0; +} + +#define vl_msg_name_crc_list +#include <flowprobe/flowprobe_all_api_h.h> +#undef vl_msg_name_crc_list + +static void +setup_message_id_table (flowprobe_main_t * fm, api_main_t * am) +{ +#define _(id,n,crc) \ + vl_msg_api_add_msg_name_crc (am, #n "_" #crc, id + fm->msg_id_base); + foreach_vl_msg_name_crc_flowprobe; +#undef _ +} + +/* + * Main-core process, sending an interrupt to the per worker input + * process that spins the per worker timer wheel. + */ +static uword +timer_process (vlib_main_t * vm, vlib_node_runtime_t * rt, vlib_frame_t * f) +{ + uword *event_data = 0; + vlib_main_t **worker_vms = 0, *worker_vm; + flowprobe_main_t *fm = &flowprobe_main; + + /* Wait for Godot... */ + vlib_process_wait_for_event_or_clock (vm, 1e9); + uword event_type = vlib_process_get_events (vm, &event_data); + if (event_type != 1) + clib_warning ("bogus kickoff event received, %d", event_type); + vec_reset_length (event_data); + + int i; + if (vec_len (vlib_mains) == 0) + vec_add1 (worker_vms, vm); + else + { + for (i = 0; i < vec_len (vlib_mains); i++) + { + worker_vm = vlib_mains[i]; + if (worker_vm) + vec_add1 (worker_vms, worker_vm); + } + } + f64 sleep_duration = 0.1; + + while (1) + { + /* Send an interrupt to each timer input node */ + sleep_duration = 0.1; + for (i = 0; i < vec_len (worker_vms); i++) + { + worker_vm = worker_vms[i]; + if (worker_vm) + { + vlib_node_set_interrupt_pending (worker_vm, + flowprobe_walker_node.index); + sleep_duration = + (fm->expired_passive_per_worker[i] > 0) ? 1e-4 : 0.1; + } + } + vlib_process_suspend (vm, sleep_duration); + } + return 0; /* or not */ +} + +/* *INDENT-OFF* */ +VLIB_REGISTER_NODE (flowprobe_timer_node,static) = { + .function = timer_process, + .name = "flowprobe-timer-process", + .type = VLIB_NODE_TYPE_PROCESS, +}; +/* *INDENT-ON* */ + +/** + * @brief Set up the API message handling tables + * @param vm vlib_main_t * vlib main data structure pointer + * @returns 0 to indicate all is well, or a clib_error_t + */ +static clib_error_t * +flowprobe_init (vlib_main_t * vm) +{ + flowprobe_main_t *fm = &flowprobe_main; + vlib_thread_main_t *tm = &vlib_thread_main; + clib_error_t *error = 0; + u8 *name; + u32 num_threads; + int i; + + fm->vnet_main = vnet_get_main (); + + /* Construct the API name */ + name = format (0, "flowprobe_%08x%c", api_version, 0); + + /* Ask for a correctly-sized block of API message decode slots */ + fm->msg_id_base = vl_msg_api_get_msg_ids + ((char *) name, VL_MSG_FIRST_AVAILABLE); + + /* Hook up message handlers */ + error = flowprobe_plugin_api_hookup (vm); + + /* Add our API messages to the global name_crc hash table */ + setup_message_id_table (fm, &api_main); + + vec_free (name); + + /* Set up time reference pair */ + fm->vlib_time_0 = vlib_time_now (vm); + fm->nanosecond_time_0 = unix_time_now_nsec (); + + memset (fm->template_reports, 0, sizeof (fm->template_reports)); + memset (fm->template_size, 0, sizeof (fm->template_size)); + memset (fm->template_per_flow, 0, sizeof (fm->template_per_flow)); + + /* Decide how many worker threads we have */ + num_threads = 1 /* main thread */ + tm->n_threads; + + /* Allocate per worker thread vectors per flavour */ + for (i = 0; i < FLOW_N_VARIANTS; i++) + { + vec_validate (fm->context[i].buffers_per_worker, num_threads - 1); + vec_validate (fm->context[i].frames_per_worker, num_threads - 1); + vec_validate (fm->context[i].next_record_offset_per_worker, + num_threads - 1); + } + + fm->active_timer = FLOWPROBE_TIMER_ACTIVE; + fm->passive_timer = FLOWPROBE_TIMER_PASSIVE; + + return error; +} + +VLIB_INIT_FUNCTION (flowprobe_init); + +/* + * fd.io coding-style-patch-verification: ON + * + * Local Variables: + * eval: (c-set-style "gnu") + * End: + */ diff --git a/src/plugins/flowprobe/flowprobe.h b/src/plugins/flowprobe/flowprobe.h new file mode 100644 index 00000000000..196c92a7f35 --- /dev/null +++ b/src/plugins/flowprobe/flowprobe.h @@ -0,0 +1,167 @@ +/* + * flowprobe.h - ipfix probe plug-in header file + * + * Copyright (c) 2016 Cisco and/or its affiliates. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#ifndef __included_flowprobe_h__ +#define __included_flowprobe_h__ + +#include <vnet/vnet.h> +#include <vnet/ip/ip.h> +#include <vnet/ethernet/ethernet.h> + +#include <vppinfra/hash.h> +#include <vppinfra/error.h> +#include <vnet/flow/flow_report.h> +#include <vnet/flow/flow_report_classify.h> +#include <vppinfra/tw_timer_2t_1w_2048sl.h> + +/* Default timers in seconds */ +#define FLOWPROBE_TIMER_ACTIVE (15) +#define FLOWPROBE_TIMER_PASSIVE 120 // XXXX: FOR TESTING (30*60) +#define FLOWPROBE_LOG2_HASHSIZE (18) + +typedef enum +{ + FLOW_RECORD_L2 = 1 << 0, + FLOW_RECORD_L3 = 1 << 1, + FLOW_RECORD_L4 = 1 << 2, + FLOW_RECORD_L2_IP4 = 1 << 3, + FLOW_RECORD_L2_IP6 = 1 << 4, + FLOW_N_RECORDS = 1 << 5, +} flowprobe_record_t; + +/* *INDENT-OFF* */ +typedef enum __attribute__ ((__packed__)) +{ + FLOW_VARIANT_IP4, + FLOW_VARIANT_IP6, + FLOW_VARIANT_L2, + FLOW_VARIANT_L2_IP4, + FLOW_VARIANT_L2_IP6, + FLOW_N_VARIANTS, +} flowprobe_variant_t; +/* *INDENT-ON* */ + +STATIC_ASSERT (sizeof (flowprobe_variant_t) == 1, + "flowprobe_variant_t is expected to be 1 byte, " + "revisit padding in flowprobe_key_t"); + +#define FLOW_MAXIMUM_EXPORT_ENTRIES (1024) + +typedef struct +{ + /* what to collect per variant */ + flowprobe_record_t flags; + /** ipfix buffers under construction, per-worker thread */ + vlib_buffer_t **buffers_per_worker; + /** frames containing ipfix buffers, per-worker thread */ + vlib_frame_t **frames_per_worker; + /** next record offset, per worker thread */ + u16 *next_record_offset_per_worker; +} flowprobe_protocol_context_t; + +#define FLOWPROBE_KEY_IN_U32 22 +/* *INDENT-OFF* */ +typedef CLIB_PACKED (union +{ + struct { + u32 rx_sw_if_index; + u32 tx_sw_if_index; + u8 src_mac[6]; + u8 dst_mac[6]; + u16 ethertype; + ip46_address_t src_address; + ip46_address_t dst_address; + u8 protocol; + u16 src_port; + u16 dst_port; + flowprobe_variant_t which; + }; + u32 as_u32[FLOWPROBE_KEY_IN_U32]; +}) flowprobe_key_t; +/* *INDENT-ON* */ + +STATIC_ASSERT (sizeof (flowprobe_key_t) == FLOWPROBE_KEY_IN_U32 * + sizeof (u32), "flowprobe_key_t padding is wrong"); + +typedef struct +{ + flowprobe_key_t key; + u64 packetcount; + u64 octetcount; + f64 last_updated; + f64 last_exported; + u32 passive_timer_handle; +} flowprobe_entry_t; + +/** + * @file + * @brief flow-per-packet plugin header file + */ +typedef struct +{ + /** API message ID base */ + u16 msg_id_base; + + flowprobe_protocol_context_t context[FLOW_N_VARIANTS]; + u16 template_reports[FLOW_N_RECORDS]; + u16 template_size[FLOW_N_RECORDS]; + + /** Time reference pair */ + u64 nanosecond_time_0; + f64 vlib_time_0; + + /** Per CPU flow-state */ + u8 ht_log2len; /* Hash table size is 2^log2len */ + u32 **hash_per_worker; + flowprobe_entry_t **pool_per_worker; + /* *INDENT-OFF* */ + TWT (tw_timer_wheel) ** timers_per_worker; + /* *INDENT-ON* */ + u32 **expired_passive_per_worker; + + flowprobe_record_t record; + u32 active_timer; + u32 passive_timer; + flowprobe_entry_t *stateless_entry; + + bool initialized; + bool disabled; + + u16 template_per_flow[FLOW_N_VARIANTS]; + u8 *flow_per_interface; + + /** convenience vlib_main_t pointer */ + vlib_main_t *vlib_main; + /** convenience vnet_main_t pointer */ + vnet_main_t *vnet_main; +} flowprobe_main_t; + +extern flowprobe_main_t flowprobe_main; + +void flowprobe_flush_callback_ip4 (void); +void flowprobe_flush_callback_ip6 (void); +void flowprobe_flush_callback_l2 (void); +u8 *format_flowprobe_entry (u8 * s, va_list * args); + +#endif + +/* + * fd.io coding-style-patch-verification: ON + * + * Local Variables: + * eval: (c-set-style "gnu") + * End: + */ diff --git a/src/plugins/flowprobe/flowprobe_all_api_h.h b/src/plugins/flowprobe/flowprobe_all_api_h.h new file mode 100644 index 00000000000..1f30ecccf98 --- /dev/null +++ b/src/plugins/flowprobe/flowprobe_all_api_h.h @@ -0,0 +1,18 @@ +/* + * flowprobe_all_api_h.h - plug-in api #include file + * + * Copyright (c) <current-year> <your-organization> + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +/* Include the generated file, see BUILT_SOURCES in Makefile.am */ +#include <flowprobe/flowprobe.api.h> diff --git a/src/plugins/flowprobe/flowprobe_msg_enum.h b/src/plugins/flowprobe/flowprobe_msg_enum.h new file mode 100644 index 00000000000..bc0b21c96c1 --- /dev/null +++ b/src/plugins/flowprobe/flowprobe_msg_enum.h @@ -0,0 +1,31 @@ +/* + * flowprobe_msg_enum.h - vpp engine plug-in message enumeration + * + * Copyright (c) <current-year> <your-organization> + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#ifndef included_flowprobe_msg_enum_h +#define included_flowprobe_msg_enum_h + +#include <vppinfra/byte_order.h> + +#define vl_msg_id(n,h) n, +typedef enum +{ +#include <flowprobe/flowprobe_all_api_h.h> + /* We'll want to know how many messages IDs we need... */ + VL_MSG_FIRST_AVAILABLE, +} vl_msg_id_t; +#undef vl_msg_id + +#endif /* included_flowprobe_msg_enum_h */ diff --git a/src/plugins/flowprobe/flowprobe_plugin_doc.md b/src/plugins/flowprobe/flowprobe_plugin_doc.md new file mode 100644 index 00000000000..4c9b2342a83 --- /dev/null +++ b/src/plugins/flowprobe/flowprobe_plugin_doc.md @@ -0,0 +1,13 @@ +IPFIX flow record plugin {#flowprobe_plugin_doc} +======================== + +## Introduction + +This plugin generates ipfix flow records on interfaces which have the feature enabled + +## Sample configuration + +set ipfix exporter collector 192.168.6.2 src 192.168.6.1 template-interval 20 port 4739 path-mtu 1500 + +flowprobe params record l3 active 20 passive 120 +flowprobe feature add-del GigabitEthernet2/3/0 l2
\ No newline at end of file diff --git a/src/plugins/flowprobe/flowprobe_test.c b/src/plugins/flowprobe/flowprobe_test.c new file mode 100644 index 00000000000..91793f552c9 --- /dev/null +++ b/src/plugins/flowprobe/flowprobe_test.c @@ -0,0 +1,263 @@ +/* + * flowprobe.c - skeleton vpp-api-test plug-in + * + * Copyright (c) <current-year> <your-organization> + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include <vat/vat.h> +#include <vlibapi/api.h> +#include <vlibmemory/api.h> +#include <vlibsocket/api.h> +#include <vppinfra/error.h> +#include <flowprobe/flowprobe.h> + +#define __plugin_msg_base flowprobe_test_main.msg_id_base +#include <vlibapi/vat_helper_macros.h> + +/** + * @file vpp_api_test plugin + */ + +uword unformat_sw_if_index (unformat_input_t * input, va_list * args); + +/* Declare message IDs */ +#include <flowprobe/flowprobe_msg_enum.h> + +/* define message structures */ +#define vl_typedefs +#include <flowprobe/flowprobe_all_api_h.h> +#undef vl_typedefs + +/* declare message handlers for each api */ + +#define vl_endianfun /* define message structures */ +#include <flowprobe/flowprobe_all_api_h.h> +#undef vl_endianfun + +/* instantiate all the print functions we know about */ +#define vl_print(handle, ...) +#define vl_printfun +#include <flowprobe/flowprobe_all_api_h.h> +#undef vl_printfun + +/* Get the API version number. */ +#define vl_api_version(n,v) static u32 api_version=(v); +#include <flowprobe/flowprobe_all_api_h.h> +#undef vl_api_version + +typedef struct +{ + /** API message ID base */ + u16 msg_id_base; + /** vat_main_t pointer */ + vat_main_t *vat_main; +} flowprobe_test_main_t; + +flowprobe_test_main_t flowprobe_test_main; + +#define foreach_standard_reply_retval_handler \ +_(flowprobe_tx_interface_add_del_reply) \ +_(flowprobe_params_reply) + +#define _(n) \ + static void vl_api_##n##_t_handler \ + (vl_api_##n##_t * mp) \ + { \ + vat_main_t * vam = flowprobe_test_main.vat_main; \ + i32 retval = ntohl(mp->retval); \ + if (vam->async_mode) { \ + vam->async_errors += (retval < 0); \ + } else { \ + vam->retval = retval; \ + vam->result_ready = 1; \ + } \ + } +foreach_standard_reply_retval_handler; +#undef _ + +/* + * Table of message reply handlers, must include boilerplate handlers + * we just generated + */ +#define foreach_vpe_api_reply_msg \ +_(FLOWPROBE_TX_INTERFACE_ADD_DEL_REPLY, \ + flowprobe_tx_interface_add_del_reply) \ +_(FLOWPROBE_PARAMS_REPLY, flowprobe_params_reply) + +static int +api_flowprobe_tx_interface_add_del (vat_main_t * vam) +{ + unformat_input_t *i = vam->input; + int enable_disable = 1; + u8 which = FLOW_VARIANT_IP4; + u32 sw_if_index = ~0; + vl_api_flowprobe_tx_interface_add_del_t *mp; + int ret; + + /* Parse args required to build the message */ + while (unformat_check_input (i) != UNFORMAT_END_OF_INPUT) + { + if (unformat (i, "%U", unformat_sw_if_index, vam, &sw_if_index)) + ; + else if (unformat (i, "sw_if_index %d", &sw_if_index)) + ; + else if (unformat (i, "disable")) + enable_disable = 0; + else if (unformat (i, "ip4")) + which = FLOW_VARIANT_IP4; + else if (unformat (i, "ip6")) + which = FLOW_VARIANT_IP6; + else if (unformat (i, "l2")) + which = FLOW_VARIANT_L2; + else + break; + } + + if (sw_if_index == ~0) + { + errmsg ("missing interface name / explicit sw_if_index number \n"); + return -99; + } + + /* Construct the API message */ + M (FLOWPROBE_TX_INTERFACE_ADD_DEL, mp); + mp->sw_if_index = ntohl (sw_if_index); + mp->is_add = enable_disable; + mp->which = which; + + /* send it... */ + S (mp); + + /* Wait for a reply... */ + W (ret); + return ret; +} + +static int +api_flowprobe_params (vat_main_t * vam) +{ + unformat_input_t *i = vam->input; + u8 record_l2 = 0, record_l3 = 0, record_l4 = 0; + u32 active_timer = ~0; + u32 passive_timer = ~0; + vl_api_flowprobe_params_t *mp; + int ret; + + while (unformat_check_input (i) != UNFORMAT_END_OF_INPUT) + { + if (unformat (i, "active %d", &active_timer)) + ; + else if (unformat (i, "passive %d", &passive_timer)) + ; + else if (unformat (i, "record")) + while (unformat_check_input (i) != UNFORMAT_END_OF_INPUT) + { + if (unformat (i, "l2")) + record_l2 = 1; + else if (unformat (i, "l3")) + record_l3 = 1; + else if (unformat (i, "l4")) + record_l4 = 1; + else + break; + } + else + break; + } + + if (passive_timer > 0 && active_timer > passive_timer) + { + errmsg ("Passive timer has to be greater than active one...\n"); + return -99; + } + + /* Construct the API message */ + M (FLOWPROBE_PARAMS, mp); + mp->record_l2 = record_l2; + mp->record_l3 = record_l3; + mp->record_l4 = record_l4; + mp->active_timer = ntohl (active_timer); + mp->passive_timer = ntohl (passive_timer); + + /* send it... */ + S (mp); + + /* Wait for a reply... */ + W (ret); + + return ret; +} + +/* + * List of messages that the api test plugin sends, + * and that the data plane plugin processes + */ +#define foreach_vpe_api_msg \ +_(flowprobe_tx_interface_add_del, "<intfc> [disable]") \ +_(flowprobe_params, "record <[l2] [l3] [l4]> [active <timer> passive <timer>]") + +static void +flowprobe_vat_api_hookup (vat_main_t * vam) +{ + flowprobe_test_main_t *sm = &flowprobe_test_main; + /* Hook up handlers for replies from the data plane plug-in */ +#define _(N,n) \ + vl_msg_api_set_handlers((VL_API_##N + sm->msg_id_base), \ + #n, \ + vl_api_##n##_t_handler, \ + vl_noop_handler, \ + vl_api_##n##_t_endian, \ + vl_api_##n##_t_print, \ + sizeof(vl_api_##n##_t), 1); + foreach_vpe_api_reply_msg; +#undef _ + + /* API messages we can send */ +#define _(n,h) hash_set_mem (vam->function_by_name, #n, api_##n); + foreach_vpe_api_msg; +#undef _ + + /* Help strings */ +#define _(n,h) hash_set_mem (vam->help_by_name, #n, h); + foreach_vpe_api_msg; +#undef _ +} + +clib_error_t * +vat_plugin_register (vat_main_t * vam) +{ + flowprobe_test_main_t *sm = &flowprobe_test_main; + u8 *name; + + sm->vat_main = vam; + + /* Ask the vpp engine for the first assigned message-id */ + name = format (0, "flowprobe_%08x%c", api_version, 0); + sm->msg_id_base = vl_client_get_first_plugin_msg_id ((char *) name); + + /* Don't attempt to hook up API messages if the data plane plugin is AWOL */ + if (sm->msg_id_base != (u16) ~ 0) + flowprobe_vat_api_hookup (vam); + + vec_free (name); + + return 0; +} + +/* + * fd.io coding-style-patch-verification: ON + * + * Local Variables: + * eval: (c-set-style "gnu") + * End: + */ diff --git a/src/plugins/flowprobe/node.c b/src/plugins/flowprobe/node.c new file mode 100644 index 00000000000..6a539db9bbf --- /dev/null +++ b/src/plugins/flowprobe/node.c @@ -0,0 +1,998 @@ +/* + * node.c - ipfix probe graph node + * + * Copyright (c) 2017 Cisco and/or its affiliates. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include <vlib/vlib.h> +#include <vnet/vnet.h> +#include <vnet/pg/pg.h> +#include <vppinfra/error.h> +#include <flowprobe/flowprobe.h> +#include <vnet/ip/ip6_packet.h> + +static void flowprobe_export_entry (vlib_main_t * vm, flowprobe_entry_t * e); + +/** + * @file flow record generator graph node + */ + +typedef struct +{ + /** interface handle */ + u32 rx_sw_if_index; + u32 tx_sw_if_index; + /** packet timestamp */ + u64 timestamp; + /** size of the buffer */ + u16 buffer_size; + + /** L2 information */ + u8 src_mac[6]; + u8 dst_mac[6]; + /** Ethertype */ + u16 ethertype; + + /** L3 information */ + ip46_address_t src_address; + ip46_address_t dst_address; + u8 protocol; + u8 tos; + + /** L4 information */ + u16 src_port; + u16 dst_port; + + flowprobe_variant_t which; +} flowprobe_trace_t; + +static char *flowprobe_variant_strings[] = { + [FLOW_VARIANT_IP4] = "IP4", + [FLOW_VARIANT_IP6] = "IP6", + [FLOW_VARIANT_L2] = "L2", + [FLOW_VARIANT_L2_IP4] = "L2-IP4", + [FLOW_VARIANT_L2_IP6] = "L2-IP6", +}; + +/* packet trace format function */ +static u8 * +format_flowprobe_trace (u8 * s, va_list * args) +{ + CLIB_UNUSED (vlib_main_t * vm) = va_arg (*args, vlib_main_t *); + CLIB_UNUSED (vlib_node_t * node) = va_arg (*args, vlib_node_t *); + flowprobe_trace_t *t = va_arg (*args, flowprobe_trace_t *); + uword indent = format_get_indent (s); + + s = format (s, + "FLOWPROBE[%s]: rx_sw_if_index %d, tx_sw_if_index %d, " + "timestamp %lld, size %d", flowprobe_variant_strings[t->which], + t->rx_sw_if_index, t->tx_sw_if_index, + t->timestamp, t->buffer_size); + + if (t->which == FLOW_VARIANT_L2) + s = format (s, "\n%U -> %U", format_white_space, indent, + format_ethernet_address, &t->src_mac, + format_ethernet_address, &t->dst_mac); + + if (t->protocol > 0 + && (t->which == FLOW_VARIANT_L2_IP4 || t->which == FLOW_VARIANT_IP4 + || t->which == FLOW_VARIANT_L2_IP6 || t->which == FLOW_VARIANT_IP6)) + s = + format (s, "\n%U%U: %U -> %U", format_white_space, indent, + format_ip_protocol, t->protocol, format_ip46_address, + &t->src_address, IP46_TYPE_ANY, format_ip46_address, + &t->dst_address, IP46_TYPE_ANY); + return s; +} + +vlib_node_registration_t flowprobe_ip4_node; +vlib_node_registration_t flowprobe_ip6_node; +vlib_node_registration_t flowprobe_l2_node; + +/* No counters at the moment */ +#define foreach_flowprobe_error \ +_(COLLISION, "Hash table collisions") \ +_(BUFFER, "Buffer allocation error") \ +_(EXPORTED_PACKETS, "Exported packets") \ +_(INPATH, "Exported packets in path") + +typedef enum +{ +#define _(sym,str) FLOWPROBE_ERROR_##sym, + foreach_flowprobe_error +#undef _ + FLOWPROBE_N_ERROR, +} flowprobe_error_t; + +static char *flowprobe_error_strings[] = { +#define _(sym,string) string, + foreach_flowprobe_error +#undef _ +}; + +typedef enum +{ + FLOWPROBE_NEXT_DROP, + FLOWPROBE_NEXT_IP4_LOOKUP, + FLOWPROBE_N_NEXT, +} flowprobe_next_t; + +#define FLOWPROBE_NEXT_NODES { \ + [FLOWPROBE_NEXT_DROP] = "error-drop", \ + [FLOWPROBE_NEXT_IP4_LOOKUP] = "ip4-lookup", \ +} + +static inline flowprobe_variant_t +flowprobe_get_variant (flowprobe_variant_t which, + flowprobe_record_t flags, u16 ethertype) +{ + if (which == FLOW_VARIANT_L2 + && (flags & FLOW_RECORD_L3 || flags & FLOW_RECORD_L4)) + return ethertype == ETHERNET_TYPE_IP6 ? FLOW_VARIANT_L2_IP6 : ethertype == + ETHERNET_TYPE_IP4 ? FLOW_VARIANT_L2_IP4 : FLOW_VARIANT_L2; + return which; +} + +static inline u32 +flowprobe_common_add (vlib_buffer_t * to_b, flowprobe_entry_t * e, u16 offset) +{ + u16 start = offset; + + /* Ingress interface */ + u32 rx_if = clib_host_to_net_u32 (e->key.rx_sw_if_index); + clib_memcpy (to_b->data + offset, &rx_if, sizeof (rx_if)); + offset += sizeof (rx_if); + + /* Egress interface */ + u32 tx_if = clib_host_to_net_u32 (e->key.tx_sw_if_index); + clib_memcpy (to_b->data + offset, &tx_if, sizeof (tx_if)); + offset += sizeof (tx_if); + + /* packet delta count */ + u64 packetdelta = clib_host_to_net_u64 (e->packetcount); + clib_memcpy (to_b->data + offset, &packetdelta, sizeof (u64)); + offset += sizeof (u64); + + return offset - start; +} + +static inline u32 +flowprobe_l2_add (vlib_buffer_t * to_b, flowprobe_entry_t * e, u16 offset) +{ + u16 start = offset; + + /* src mac address */ + clib_memcpy (to_b->data + offset, &e->key.src_mac, 6); + offset += 6; + + /* dst mac address */ + clib_memcpy (to_b->data + offset, &e->key.dst_mac, 6); + offset += 6; + + /* ethertype */ + clib_memcpy (to_b->data + offset, &e->key.ethertype, 2); + offset += 2; + + return offset - start; +} + +static inline u32 +flowprobe_l3_ip6_add (vlib_buffer_t * to_b, flowprobe_entry_t * e, u16 offset) +{ + u16 start = offset; + + /* ip6 src address */ + clib_memcpy (to_b->data + offset, &e->key.src_address, + sizeof (ip6_address_t)); + offset += sizeof (ip6_address_t); + + /* ip6 dst address */ + clib_memcpy (to_b->data + offset, &e->key.dst_address, + sizeof (ip6_address_t)); + offset += sizeof (ip6_address_t); + + /* Protocol */ + to_b->data[offset++] = e->key.protocol; + + /* octetDeltaCount */ + u64 octetdelta = clib_host_to_net_u64 (e->octetcount); + clib_memcpy (to_b->data + offset, &octetdelta, sizeof (u64)); + offset += sizeof (u64); + + return offset - start; +} + +static inline u32 +flowprobe_l3_ip4_add (vlib_buffer_t * to_b, flowprobe_entry_t * e, u16 offset) +{ + u16 start = offset; + + /* ip4 src address */ + clib_memcpy (to_b->data + offset, &e->key.src_address.ip4, + sizeof (ip4_address_t)); + offset += sizeof (ip4_address_t); + + /* ip4 dst address */ + clib_memcpy (to_b->data + offset, &e->key.dst_address.ip4, + sizeof (ip4_address_t)); + offset += sizeof (ip4_address_t); + + /* Protocol */ + to_b->data[offset++] = e->key.protocol; + + /* octetDeltaCount */ + u64 octetdelta = clib_host_to_net_u64 (e->octetcount); + clib_memcpy (to_b->data + offset, &octetdelta, sizeof (u64)); + offset += sizeof (u64); + + return offset - start; +} + +static inline u32 +flowprobe_l4_add (vlib_buffer_t * to_b, flowprobe_entry_t * e, u16 offset) +{ + u16 start = offset; + + /* src port */ + clib_memcpy (to_b->data + offset, &e->key.src_port, 2); + offset += 2; + + /* dst port */ + clib_memcpy (to_b->data + offset, &e->key.dst_port, 2); + offset += 2; + + return offset - start; +} + +static inline u32 +flowprobe_hash (flowprobe_key_t * k) +{ + flowprobe_main_t *fm = &flowprobe_main; + int i; + u32 h = 0; + for (i = 0; i < sizeof (k->as_u32) / sizeof (u32); i++) + h = crc_u32 (k->as_u32[i], h); + return h >> (32 - fm->ht_log2len); +} + +flowprobe_entry_t * +flowprobe_lookup (u32 my_cpu_number, flowprobe_key_t * k, u32 * poolindex, + bool * collision) +{ + flowprobe_main_t *fm = &flowprobe_main; + flowprobe_entry_t *e; + u32 h; + + h = (fm->active_timer) ? flowprobe_hash (k) : 0; + + /* Lookup in the flow state pool */ + *poolindex = fm->hash_per_worker[my_cpu_number][h]; + if (*poolindex != ~0) + { + e = pool_elt_at_index (fm->pool_per_worker[my_cpu_number], *poolindex); + if (e) + { + /* Verify key or report collision */ + if (memcmp (k, &e->key, sizeof (flowprobe_key_t))) + *collision = true; + return e; + } + } + + return 0; +} + +flowprobe_entry_t * +flowprobe_create (u32 my_cpu_number, flowprobe_key_t * k, u32 * poolindex) +{ + flowprobe_main_t *fm = &flowprobe_main; + u32 h; + + flowprobe_entry_t *e; + + /* Get my index */ + h = (fm->active_timer) ? flowprobe_hash (k) : 0; + + pool_get (fm->pool_per_worker[my_cpu_number], e); + *poolindex = e - fm->pool_per_worker[my_cpu_number]; + fm->hash_per_worker[my_cpu_number][h] = *poolindex; + + e->key = *k; + + if (fm->passive_timer > 0) + { + e->passive_timer_handle = tw_timer_start_2t_1w_2048sl + (fm->timers_per_worker[my_cpu_number], *poolindex, 0, + fm->passive_timer); + } + return e; +} + +static inline void +add_to_flow_record_state (vlib_main_t * vm, vlib_node_runtime_t * node, + flowprobe_main_t * fm, vlib_buffer_t * b, + u64 timestamp, u16 length, + flowprobe_variant_t which, flowprobe_trace_t * t) +{ + if (fm->disabled) + return; + + u32 my_cpu_number = vm->thread_index; + u16 octets = 0; + + flowprobe_record_t flags = fm->context[which].flags; + bool collect_ip4 = false, collect_ip6 = false; + ASSERT (b); + ethernet_header_t *eth = vlib_buffer_get_current (b); + u16 ethertype = clib_net_to_host_u16 (eth->type); + /* *INDENT-OFF* */ + flowprobe_key_t k = { {0} }; + /* *INDENT-ON* */ + ip4_header_t *ip4 = 0; + ip6_header_t *ip6 = 0; + udp_header_t *udp = 0; + + if (flags & FLOW_RECORD_L3 || flags & FLOW_RECORD_L4) + { + collect_ip4 = which == FLOW_VARIANT_L2_IP4 || which == FLOW_VARIANT_IP4; + collect_ip6 = which == FLOW_VARIANT_L2_IP6 || which == FLOW_VARIANT_IP6; + } + + k.rx_sw_if_index = vnet_buffer (b)->sw_if_index[VLIB_RX]; + k.tx_sw_if_index = vnet_buffer (b)->sw_if_index[VLIB_TX]; + + k.which = which; + + if (flags & FLOW_RECORD_L2) + { + clib_memcpy (k.src_mac, eth->src_address, 6); + clib_memcpy (k.dst_mac, eth->dst_address, 6); + k.ethertype = ethertype; + } + if (collect_ip6 && ethertype == ETHERNET_TYPE_IP6) + { + ip6 = (ip6_header_t *) (eth + 1); + udp = (udp_header_t *) (ip6 + 1); + if (flags & FLOW_RECORD_L3) + { + k.src_address.as_u64[0] = ip6->src_address.as_u64[0]; + k.src_address.as_u64[1] = ip6->src_address.as_u64[1]; + k.dst_address.as_u64[0] = ip6->dst_address.as_u64[0]; + k.dst_address.as_u64[1] = ip6->dst_address.as_u64[1]; + } + k.protocol = ip6->protocol; + octets = clib_net_to_host_u16 (ip6->payload_length) + + sizeof (ip6_header_t); + } + if (collect_ip4 && ethertype == ETHERNET_TYPE_IP4) + { + ip4 = (ip4_header_t *) (eth + 1); + udp = (udp_header_t *) (ip4 + 1); + if (flags & FLOW_RECORD_L3) + { + k.src_address.ip4.as_u32 = ip4->src_address.as_u32; + k.dst_address.ip4.as_u32 = ip4->dst_address.as_u32; + } + k.protocol = ip4->protocol; + octets = clib_net_to_host_u16 (ip4->length); + } + if ((flags & FLOW_RECORD_L4) && udp && + (k.protocol == IP_PROTOCOL_TCP || k.protocol == IP_PROTOCOL_UDP)) + { + k.src_port = udp->src_port; + k.dst_port = udp->dst_port; + } + + if (t) + { + t->rx_sw_if_index = k.rx_sw_if_index; + t->tx_sw_if_index = k.tx_sw_if_index; + clib_memcpy (t->src_mac, k.src_mac, 6); + clib_memcpy (t->dst_mac, k.dst_mac, 6); + t->ethertype = k.ethertype; + t->src_address.ip4.as_u32 = k.src_address.ip4.as_u32; + t->dst_address.ip4.as_u32 = k.dst_address.ip4.as_u32; + t->protocol = k.protocol; + t->src_port = k.src_port; + t->dst_port = k.dst_port; + t->which = k.which; + } + + flowprobe_entry_t *e = 0; + f64 now = vlib_time_now (vm); + if (fm->active_timer > 0) + { + u32 poolindex = ~0; + bool collision = false; + + e = flowprobe_lookup (my_cpu_number, &k, &poolindex, &collision); + if (collision) + { + /* Flush data and clean up entry for reuse. */ + if (e->packetcount) + flowprobe_export_entry (vm, e); + e->key = k; + vlib_node_increment_counter (vm, node->node_index, + FLOWPROBE_ERROR_COLLISION, 1); + } + if (!e) /* Create new entry */ + { + e = flowprobe_create (my_cpu_number, &k, &poolindex); + e->last_exported = now; + } + } + else + { + e = &fm->stateless_entry[my_cpu_number]; + e->key = k; + } + + if (e) + { + /* Updating entry */ + e->packetcount++; + e->octetcount += octets; + e->last_updated = now; + + if (fm->active_timer == 0 + || (now > e->last_exported + fm->active_timer)) + flowprobe_export_entry (vm, e); + } +} + +static u16 +flowprobe_get_headersize (void) +{ + return sizeof (ip4_header_t) + sizeof (udp_header_t) + + sizeof (ipfix_message_header_t) + sizeof (ipfix_set_header_t); +} + +static void +flowprobe_export_send (vlib_main_t * vm, vlib_buffer_t * b0, + flowprobe_variant_t which) +{ + flowprobe_main_t *fm = &flowprobe_main; + flow_report_main_t *frm = &flow_report_main; + vlib_frame_t *f; + ip4_ipfix_template_packet_t *tp; + ipfix_set_header_t *s; + ipfix_message_header_t *h; + ip4_header_t *ip; + udp_header_t *udp; + flowprobe_record_t flags = fm->context[which].flags; + u32 my_cpu_number = vm->thread_index; + + /* Fill in header */ + flow_report_stream_t *stream; + + /* Nothing to send */ + if (fm->context[which].next_record_offset_per_worker[my_cpu_number] <= + flowprobe_get_headersize ()) + return; + + u32 i, index = vec_len (frm->streams); + for (i = 0; i < index; i++) + if (frm->streams[i].domain_id == 1) + { + index = i; + break; + } + if (i == vec_len (frm->streams)) + { + vec_validate (frm->streams, index); + frm->streams[index].domain_id = 1; + } + stream = &frm->streams[index]; + + tp = vlib_buffer_get_current (b0); + ip = (ip4_header_t *) & tp->ip4; + udp = (udp_header_t *) (ip + 1); + h = (ipfix_message_header_t *) (udp + 1); + s = (ipfix_set_header_t *) (h + 1); + + ip->ip_version_and_header_length = 0x45; + ip->ttl = 254; + ip->protocol = IP_PROTOCOL_UDP; + ip->flags_and_fragment_offset = 0; + ip->src_address.as_u32 = frm->src_address.as_u32; + ip->dst_address.as_u32 = frm->ipfix_collector.as_u32; + udp->src_port = clib_host_to_net_u16 (UDP_DST_PORT_ipfix); + udp->dst_port = clib_host_to_net_u16 (UDP_DST_PORT_ipfix); + udp->checksum = 0; + + /* FIXUP: message header export_time */ + h->export_time = (u32) + (((f64) frm->unix_time_0) + + (vlib_time_now (frm->vlib_main) - frm->vlib_time_0)); + h->export_time = clib_host_to_net_u32 (h->export_time); + h->domain_id = clib_host_to_net_u32 (stream->domain_id); + + /* FIXUP: message header sequence_number */ + h->sequence_number = stream->sequence_number++; + h->sequence_number = clib_host_to_net_u32 (h->sequence_number); + + s->set_id_length = ipfix_set_id_length (fm->template_reports[flags], + b0->current_length - + (sizeof (*ip) + sizeof (*udp) + + sizeof (*h))); + h->version_length = version_length (b0->current_length - + (sizeof (*ip) + sizeof (*udp))); + + ip->length = clib_host_to_net_u16 (b0->current_length); + + ip->checksum = ip4_header_checksum (ip); + udp->length = clib_host_to_net_u16 (b0->current_length - sizeof (*ip)); + + if (frm->udp_checksum) + { + /* RFC 7011 section 10.3.2. */ + udp->checksum = ip4_tcp_udp_compute_checksum (vm, b0, ip); + if (udp->checksum == 0) + udp->checksum = 0xffff; + } + + ASSERT (ip->checksum == ip4_header_checksum (ip)); + + /* Find or allocate a frame */ + f = fm->context[which].frames_per_worker[my_cpu_number]; + if (PREDICT_FALSE (f == 0)) + { + u32 *to_next; + f = vlib_get_frame_to_node (vm, ip4_lookup_node.index); + fm->context[which].frames_per_worker[my_cpu_number] = f; + u32 bi0 = vlib_get_buffer_index (vm, b0); + + /* Enqueue the buffer */ + to_next = vlib_frame_vector_args (f); + to_next[0] = bi0; + f->n_vectors = 1; + } + + vlib_put_frame_to_node (vm, ip4_lookup_node.index, f); + vlib_node_increment_counter (vm, flowprobe_l2_node.index, + FLOWPROBE_ERROR_EXPORTED_PACKETS, 1); + + fm->context[which].frames_per_worker[my_cpu_number] = 0; + fm->context[which].buffers_per_worker[my_cpu_number] = 0; + fm->context[which].next_record_offset_per_worker[my_cpu_number] = + flowprobe_get_headersize (); +} + +static vlib_buffer_t * +flowprobe_get_buffer (vlib_main_t * vm, flowprobe_variant_t which) +{ + flowprobe_main_t *fm = &flowprobe_main; + flow_report_main_t *frm = &flow_report_main; + vlib_buffer_t *b0; + u32 bi0; + vlib_buffer_free_list_t *fl; + u32 my_cpu_number = vm->thread_index; + + /* Find or allocate a buffer */ + b0 = fm->context[which].buffers_per_worker[my_cpu_number]; + + /* Need to allocate a buffer? */ + if (PREDICT_FALSE (b0 == 0)) + { + if (vlib_buffer_alloc (vm, &bi0, 1) != 1) + { + vlib_node_increment_counter (vm, flowprobe_l2_node.index, + FLOWPROBE_ERROR_BUFFER, 1); + return 0; + } + + /* Initialize the buffer */ + b0 = fm->context[which].buffers_per_worker[my_cpu_number] = + vlib_get_buffer (vm, bi0); + fl = + vlib_buffer_get_free_list (vm, VLIB_BUFFER_DEFAULT_FREE_LIST_INDEX); + vlib_buffer_init_for_free_list (b0, fl); + VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0); + + b0->current_data = 0; + b0->current_length = flowprobe_get_headersize (); + b0->flags |= (VLIB_BUFFER_TOTAL_LENGTH_VALID | VLIB_BUFFER_FLOW_REPORT); + vnet_buffer (b0)->sw_if_index[VLIB_RX] = 0; + vnet_buffer (b0)->sw_if_index[VLIB_TX] = frm->fib_index; + fm->context[which].next_record_offset_per_worker[my_cpu_number] = + b0->current_length; + } + + return b0; +} + +static void +flowprobe_export_entry (vlib_main_t * vm, flowprobe_entry_t * e) +{ + u32 my_cpu_number = vm->thread_index; + flowprobe_main_t *fm = &flowprobe_main; + flow_report_main_t *frm = &flow_report_main; + vlib_buffer_t *b0; + bool collect_ip4 = false, collect_ip6 = false; + flowprobe_variant_t which = e->key.which; + flowprobe_record_t flags = fm->context[which].flags; + u16 offset = + fm->context[which].next_record_offset_per_worker[my_cpu_number]; + + if (offset < flowprobe_get_headersize ()) + offset = flowprobe_get_headersize (); + + b0 = flowprobe_get_buffer (vm, which); + /* No available buffer, what to do... */ + if (b0 == 0) + return; + + if (flags & FLOW_RECORD_L3) + { + collect_ip4 = which == FLOW_VARIANT_L2_IP4 || which == FLOW_VARIANT_IP4; + collect_ip6 = which == FLOW_VARIANT_L2_IP6 || which == FLOW_VARIANT_IP6; + } + + offset += flowprobe_common_add (b0, e, offset); + + if (flags & FLOW_RECORD_L2) + offset += flowprobe_l2_add (b0, e, offset); + if (collect_ip6) + offset += flowprobe_l3_ip6_add (b0, e, offset); + if (collect_ip4) + offset += flowprobe_l3_ip4_add (b0, e, offset); + if (flags & FLOW_RECORD_L4) + offset += flowprobe_l4_add (b0, e, offset); + + /* Reset per flow-export counters */ + e->packetcount = 0; + e->octetcount = 0; + e->last_exported = vlib_time_now (vm); + + b0->current_length = offset; + + fm->context[which].next_record_offset_per_worker[my_cpu_number] = offset; + /* Time to flush the buffer? */ + if (offset + fm->template_size[flags] > frm->path_mtu) + flowprobe_export_send (vm, b0, which); +} + +uword +flowprobe_node_fn (vlib_main_t * vm, + vlib_node_runtime_t * node, vlib_frame_t * frame, + flowprobe_variant_t which) +{ + u32 n_left_from, *from, *to_next; + flowprobe_next_t next_index; + flowprobe_main_t *fm = &flowprobe_main; + u64 now; + + now = (u64) ((vlib_time_now (vm) - fm->vlib_time_0) * 1e9); + now += fm->nanosecond_time_0; + + from = vlib_frame_vector_args (frame); + n_left_from = frame->n_vectors; + next_index = node->cached_next_index; + + while (n_left_from > 0) + { + u32 n_left_to_next; + + vlib_get_next_frame (vm, node, next_index, to_next, n_left_to_next); + + while (n_left_from >= 4 && n_left_to_next >= 2) + { + u32 next0 = FLOWPROBE_NEXT_DROP; + u32 next1 = FLOWPROBE_NEXT_DROP; + u16 len0, len1; + u32 bi0, bi1; + vlib_buffer_t *b0, *b1; + + /* Prefetch next iteration. */ + { + vlib_buffer_t *p2, *p3; + + p2 = vlib_get_buffer (vm, from[2]); + p3 = vlib_get_buffer (vm, from[3]); + + vlib_prefetch_buffer_header (p2, LOAD); + vlib_prefetch_buffer_header (p3, LOAD); + + CLIB_PREFETCH (p2->data, CLIB_CACHE_LINE_BYTES, STORE); + CLIB_PREFETCH (p3->data, CLIB_CACHE_LINE_BYTES, STORE); + } + + /* speculatively enqueue b0 and b1 to the current next frame */ + to_next[0] = bi0 = from[0]; + to_next[1] = bi1 = from[1]; + from += 2; + to_next += 2; + n_left_from -= 2; + n_left_to_next -= 2; + + b0 = vlib_get_buffer (vm, bi0); + b1 = vlib_get_buffer (vm, bi1); + + vnet_feature_next (vnet_buffer (b0)->sw_if_index[VLIB_TX], + &next0, b0); + vnet_feature_next (vnet_buffer (b1)->sw_if_index[VLIB_TX], + &next1, b1); + + len0 = vlib_buffer_length_in_chain (vm, b0); + ethernet_header_t *eh0 = vlib_buffer_get_current (b0); + u16 ethertype0 = clib_net_to_host_u16 (eh0->type); + + if (PREDICT_TRUE ((b0->flags & VLIB_BUFFER_FLOW_REPORT) == 0)) + add_to_flow_record_state (vm, node, fm, b0, now, len0, + flowprobe_get_variant + (which, fm->context[which].flags, + ethertype0), 0); + + len1 = vlib_buffer_length_in_chain (vm, b1); + ethernet_header_t *eh1 = vlib_buffer_get_current (b1); + u16 ethertype1 = clib_net_to_host_u16 (eh1->type); + + if (PREDICT_TRUE ((b1->flags & VLIB_BUFFER_FLOW_REPORT) == 0)) + add_to_flow_record_state (vm, node, fm, b1, now, len1, + flowprobe_get_variant + (which, fm->context[which].flags, + ethertype1), 0); + + /* verify speculative enqueues, maybe switch current next frame */ + vlib_validate_buffer_enqueue_x2 (vm, node, next_index, + to_next, n_left_to_next, + bi0, bi1, next0, next1); + } + + while (n_left_from > 0 && n_left_to_next > 0) + { + u32 bi0; + vlib_buffer_t *b0; + u32 next0 = FLOWPROBE_NEXT_DROP; + u16 len0; + + /* speculatively enqueue b0 to the current next frame */ + bi0 = from[0]; + to_next[0] = bi0; + from += 1; + to_next += 1; + n_left_from -= 1; + n_left_to_next -= 1; + + b0 = vlib_get_buffer (vm, bi0); + + vnet_feature_next (vnet_buffer (b0)->sw_if_index[VLIB_TX], + &next0, b0); + + len0 = vlib_buffer_length_in_chain (vm, b0); + ethernet_header_t *eh0 = vlib_buffer_get_current (b0); + u16 ethertype0 = clib_net_to_host_u16 (eh0->type); + + if (PREDICT_TRUE ((b0->flags & VLIB_BUFFER_FLOW_REPORT) == 0)) + { + flowprobe_trace_t *t = 0; + if (PREDICT_FALSE ((node->flags & VLIB_NODE_FLAG_TRACE) + && (b0->flags & VLIB_BUFFER_IS_TRACED))) + t = vlib_add_trace (vm, node, b0, sizeof (*t)); + + add_to_flow_record_state (vm, node, fm, b0, now, len0, + flowprobe_get_variant + (which, fm->context[which].flags, + ethertype0), t); + } + + /* verify speculative enqueue, maybe switch current next frame */ + vlib_validate_buffer_enqueue_x1 (vm, node, next_index, + to_next, n_left_to_next, + bi0, next0); + } + + vlib_put_next_frame (vm, node, next_index, n_left_to_next); + } + return frame->n_vectors; +} + +static uword +flowprobe_ip4_node_fn (vlib_main_t * vm, + vlib_node_runtime_t * node, vlib_frame_t * frame) +{ + return flowprobe_node_fn (vm, node, frame, FLOW_VARIANT_IP4); +} + +static uword +flowprobe_ip6_node_fn (vlib_main_t * vm, + vlib_node_runtime_t * node, vlib_frame_t * frame) +{ + return flowprobe_node_fn (vm, node, frame, FLOW_VARIANT_IP6); +} + +static uword +flowprobe_l2_node_fn (vlib_main_t * vm, + vlib_node_runtime_t * node, vlib_frame_t * frame) +{ + return flowprobe_node_fn (vm, node, frame, FLOW_VARIANT_L2); +} + +static inline void +flush_record (flowprobe_variant_t which) +{ + vlib_main_t *vm = vlib_get_main (); + vlib_buffer_t *b = flowprobe_get_buffer (vm, which); + if (b) + flowprobe_export_send (vm, b, which); +} + +void +flowprobe_flush_callback_ip4 (void) +{ + flush_record (FLOW_VARIANT_IP4); +} + +void +flowprobe_flush_callback_ip6 (void) +{ + flush_record (FLOW_VARIANT_IP6); +} + +void +flowprobe_flush_callback_l2 (void) +{ + flush_record (FLOW_VARIANT_L2); + flush_record (FLOW_VARIANT_L2_IP4); + flush_record (FLOW_VARIANT_L2_IP6); +} + + +static void +flowprobe_delete_by_index (u32 my_cpu_number, u32 poolindex) +{ + flowprobe_main_t *fm = &flowprobe_main; + flowprobe_entry_t *e; + u32 h; + + e = pool_elt_at_index (fm->pool_per_worker[my_cpu_number], poolindex); + + /* Get my index */ + h = flowprobe_hash (&e->key); + + /* Reset hash */ + fm->hash_per_worker[my_cpu_number][h] = ~0; + + pool_put_index (fm->pool_per_worker[my_cpu_number], poolindex); +} + + +/* Per worker process processing the active/passive expired entries */ +static uword +flowprobe_walker_process (vlib_main_t * vm, + vlib_node_runtime_t * rt, vlib_frame_t * f) +{ + flowprobe_main_t *fm = &flowprobe_main; + flow_report_main_t *frm = &flow_report_main; + flowprobe_entry_t *e; + + /* + * $$$$ Remove this check from here and track FRM status and disable + * this process if required. + */ + if (frm->ipfix_collector.as_u32 == 0 || frm->src_address.as_u32 == 0) + { + fm->disabled = true; + return 0; + } + fm->disabled = false; + + u32 cpu_index = os_get_thread_index (); + u32 *to_be_removed = 0, *i; + + /* + * Tick the timer when required and process the vector of expired + * timers + */ + f64 start_time = vlib_time_now (vm); + u32 count = 0; + + tw_timer_expire_timers_2t_1w_2048sl (fm->timers_per_worker[cpu_index], + start_time); + + vec_foreach (i, fm->expired_passive_per_worker[cpu_index]) + { + u32 exported = 0; + f64 now = vlib_time_now (vm); + if (now > start_time + 100e-6 + || exported > FLOW_MAXIMUM_EXPORT_ENTRIES - 1) + break; + + if (pool_is_free_index (fm->pool_per_worker[cpu_index], *i)) + { + clib_warning ("Element is %d is freed already\n", *i); + continue; + } + else + e = pool_elt_at_index (fm->pool_per_worker[cpu_index], *i); + + /* Check last update timestamp. If it is longer than passive time nuke + * entry. Otherwise restart timer with what's left + * Premature passive timer by more than 10% + */ + if ((now - e->last_updated) < (fm->passive_timer * 0.9)) + { + f64 delta = fm->passive_timer - (now - e->last_updated); + e->passive_timer_handle = tw_timer_start_2t_1w_2048sl + (fm->timers_per_worker[cpu_index], *i, 0, delta); + } + else /* Nuke entry */ + { + vec_add1 (to_be_removed, *i); + } + /* If anything to report send it to the exporter */ + if (e->packetcount && now > e->last_exported + fm->active_timer) + { + exported++; + flowprobe_export_entry (vm, e); + } + count++; + } + if (count) + vec_delete (fm->expired_passive_per_worker[cpu_index], count, 0); + + vec_foreach (i, to_be_removed) flowprobe_delete_by_index (cpu_index, *i); + vec_free (to_be_removed); + + return 0; +} + +/* *INDENT-OFF* */ +VLIB_REGISTER_NODE (flowprobe_ip4_node) = { + .function = flowprobe_ip4_node_fn, + .name = "flowprobe-ip4", + .vector_size = sizeof (u32), + .format_trace = format_flowprobe_trace, + .type = VLIB_NODE_TYPE_INTERNAL, + .n_errors = ARRAY_LEN(flowprobe_error_strings), + .error_strings = flowprobe_error_strings, + .n_next_nodes = FLOWPROBE_N_NEXT, + .next_nodes = FLOWPROBE_NEXT_NODES, +}; +VLIB_REGISTER_NODE (flowprobe_ip6_node) = { + .function = flowprobe_ip6_node_fn, + .name = "flowprobe-ip6", + .vector_size = sizeof (u32), + .format_trace = format_flowprobe_trace, + .type = VLIB_NODE_TYPE_INTERNAL, + .n_errors = ARRAY_LEN(flowprobe_error_strings), + .error_strings = flowprobe_error_strings, + .n_next_nodes = FLOWPROBE_N_NEXT, + .next_nodes = FLOWPROBE_NEXT_NODES, +}; +VLIB_REGISTER_NODE (flowprobe_l2_node) = { + .function = flowprobe_l2_node_fn, + .name = "flowprobe-l2", + .vector_size = sizeof (u32), + .format_trace = format_flowprobe_trace, + .type = VLIB_NODE_TYPE_INTERNAL, + .n_errors = ARRAY_LEN(flowprobe_error_strings), + .error_strings = flowprobe_error_strings, + .n_next_nodes = FLOWPROBE_N_NEXT, + .next_nodes = FLOWPROBE_NEXT_NODES, +}; +VLIB_REGISTER_NODE (flowprobe_walker_node) = { + .function = flowprobe_walker_process, + .name = "flowprobe-walker", + .type = VLIB_NODE_TYPE_INPUT, + .state = VLIB_NODE_STATE_INTERRUPT, +}; +/* *INDENT-ON* */ + +/* + * fd.io coding-style-patch-verification: ON + * + * Local Variables: + * eval: (c-set-style "gnu") + * End: + */ |