/*
* flow_report.c
*/
+#include <vppinfra/atomics.h>
#include <vnet/ipfix-export/flow_report.h>
#include <vnet/api_errno.h>
+#include <vnet/udp/udp.h>
flow_report_main_t flow_report_main;
static_always_inline u8
-stream_index_valid (u32 index)
+stream_index_valid (ipfix_exporter_t *exp, u32 index)
{
- flow_report_main_t *frm = &flow_report_main;
- return index < vec_len (frm->streams) &&
- frm->streams[index].domain_id != ~0;
+ return index < vec_len (exp->streams) && exp->streams[index].domain_id != ~0;
}
static_always_inline flow_report_stream_t *
-add_stream (void)
+add_stream (ipfix_exporter_t *exp)
{
- flow_report_main_t *frm = &flow_report_main;
u32 i;
- for (i = 0; i < vec_len (frm->streams); i++)
- if (!stream_index_valid (i))
- return &frm->streams[i];
- u32 index = vec_len (frm->streams);
- vec_validate (frm->streams, index);
- return &frm->streams[index];
+ for (i = 0; i < vec_len (exp->streams); i++)
+ if (!stream_index_valid (exp, i))
+ return &exp->streams[i];
+ u32 index = vec_len (exp->streams);
+ vec_validate (exp->streams, index);
+ return &exp->streams[index];
}
static_always_inline void
-delete_stream (u32 index)
+delete_stream (ipfix_exporter_t *exp, u32 index)
{
- flow_report_main_t *frm = &flow_report_main;
- ASSERT (index < vec_len (frm->streams));
- ASSERT (frm->streams[index].domain_id != ~0);
- frm->streams[index].domain_id = ~0;
+ ASSERT (index < vec_len (exp->streams));
+ ASSERT (exp->streams[index].domain_id != ~0);
+ exp->streams[index].domain_id = ~0;
}
static i32
-find_stream (u32 domain_id, u16 src_port)
+find_stream (ipfix_exporter_t *exp, u32 domain_id, u16 src_port)
{
- flow_report_main_t *frm = &flow_report_main;
flow_report_stream_t *stream;
u32 i;
- for (i = 0; i < vec_len (frm->streams); i++)
- if (stream_index_valid (i))
+ for (i = 0; i < vec_len (exp->streams); i++)
+ if (stream_index_valid (exp, i))
{
- stream = &frm->streams[i];
+ stream = &exp->streams[i];
if (domain_id == stream->domain_id)
{
if (src_port != stream->src_port)
}
int
-send_template_packet (flow_report_main_t * frm,
- flow_report_t * fr, u32 * buffer_indexp)
+send_template_packet (flow_report_main_t *frm, ipfix_exporter_t *exp,
+ flow_report_t *fr, u32 *buffer_indexp)
{
u32 bi0;
vlib_buffer_t *b0;
- ip4_ipfix_template_packet_t *tp;
+ ip4_ipfix_template_packet_t *tp4;
+ ip6_ipfix_template_packet_t *tp6;
ipfix_message_header_t *h;
- ip4_header_t *ip;
+ ip4_header_t *ip4;
+ ip6_header_t *ip6;
+ void *ip;
udp_header_t *udp;
vlib_main_t *vm = frm->vlib_main;
flow_report_stream_t *stream;
if (fr->update_rewrite || fr->rewrite == 0)
{
- if (frm->ipfix_collector.as_u32 == 0 || frm->src_address.as_u32 == 0)
+ if (ip_address_is_zero (&exp->ipfix_collector) ||
+ ip_address_is_zero (&exp->src_address))
{
vlib_node_set_state (frm->vlib_main, flow_report_process_node.index,
VLIB_NODE_STATE_DISABLED);
if (fr->update_rewrite)
{
- fr->rewrite = fr->rewrite_callback (frm, fr,
- &frm->ipfix_collector,
- &frm->src_address,
- frm->collector_port,
- fr->report_elements,
- fr->n_report_elements,
- fr->stream_indexp);
+ fr->rewrite = fr->rewrite_callback (
+ exp, fr, exp->collector_port, fr->report_elements,
+ fr->n_report_elements, fr->stream_indexp);
fr->update_rewrite = 0;
}
b0 = vlib_get_buffer (vm, bi0);
- /* Initialize the buffer */
- VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
-
ASSERT (vec_len (fr->rewrite) < vlib_buffer_get_default_data_size (vm));
clib_memcpy_fast (b0->data, fr->rewrite, vec_len (fr->rewrite));
b0->current_length = vec_len (fr->rewrite);
b0->flags |= (VLIB_BUFFER_TOTAL_LENGTH_VALID | VNET_BUFFER_F_FLOW_REPORT);
vnet_buffer (b0)->sw_if_index[VLIB_RX] = 0;
- vnet_buffer (b0)->sw_if_index[VLIB_TX] = frm->fib_index;
+ vnet_buffer (b0)->sw_if_index[VLIB_TX] = exp->fib_index;
- tp = vlib_buffer_get_current (b0);
- ip = (ip4_header_t *) & tp->ip4;
- udp = (udp_header_t *) (ip + 1);
+ if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
+ {
+ tp4 = vlib_buffer_get_current (b0);
+ ip4 = (ip4_header_t *) &tp4->ip4;
+ ip = ip4;
+ udp = (udp_header_t *) (ip4 + 1);
+ }
+ else
+ {
+ tp6 = vlib_buffer_get_current (b0);
+ ip6 = (ip6_header_t *) &tp6->ip6;
+ ip = ip6;
+ udp = (udp_header_t *) (ip6 + 1);
+ }
h = (ipfix_message_header_t *) (udp + 1);
/* FIXUP: message header export_time */
(vlib_time_now (frm->vlib_main) - frm->vlib_time_0));
h->export_time = clib_host_to_net_u32 (h->export_time);
- stream = &frm->streams[fr->stream_index];
+ stream = &exp->streams[fr->stream_index];
/* FIXUP: message header sequence_number. Templates do not increase it */
h->sequence_number = clib_host_to_net_u32 (stream->sequence_number);
/* FIXUP: udp length */
- udp->length = clib_host_to_net_u16 (b0->current_length - sizeof (*ip));
+ if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
+ udp->length = clib_host_to_net_u16 (b0->current_length - sizeof (*ip4));
+ else
+ udp->length = clib_host_to_net_u16 (b0->current_length - sizeof (*ip6));
- if (frm->udp_checksum)
+ if (exp->udp_checksum || ip_addr_version (&exp->ipfix_collector) == AF_IP6)
{
/* RFC 7011 section 10.3.2. */
- udp->checksum = ip4_tcp_udp_compute_checksum (vm, b0, ip);
+
+ if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
+ udp->checksum = ip4_tcp_udp_compute_checksum (vm, b0, ip);
+ else
+ {
+ int bogus = 0;
+ udp->checksum =
+ ip6_tcp_udp_icmp_compute_checksum (vm, b0, ip, &bogus);
+ }
+
if (udp->checksum == 0)
udp->checksum = 0xffff;
}
return 0;
}
+u32 always_inline
+ipfix_write_headers (ipfix_exporter_t *exp, void *data, void **ip,
+ udp_header_t **udp, u32 len)
+{
+ if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
+ {
+ ip4_ipfix_template_packet_t *tp4;
+ ip4_header_t *ip4;
+
+ tp4 = (ip4_ipfix_template_packet_t *) data;
+ ip4 = (ip4_header_t *) &tp4->ip4;
+ ip4->ip_version_and_header_length = 0x45;
+ ip4->ttl = 254;
+ ip4->protocol = IP_PROTOCOL_UDP;
+ ip4->flags_and_fragment_offset = 0;
+ ip4->src_address.as_u32 = exp->src_address.ip.ip4.as_u32;
+ ip4->dst_address.as_u32 = exp->ipfix_collector.ip.ip4.as_u32;
+ *ip = ip4;
+ *udp = (udp_header_t *) (ip4 + 1);
+
+ (*udp)->length = clib_host_to_net_u16 (len - sizeof (*ip4));
+ return sizeof (*ip4);
+ }
+ else
+ {
+ ip6_ipfix_template_packet_t *tp6;
+ ip6_header_t *ip6;
+
+ tp6 = (ip6_ipfix_template_packet_t *) data;
+ ip6 = (ip6_header_t *) &tp6->ip6;
+ ip6->ip_version_traffic_class_and_flow_label =
+ clib_host_to_net_u32 (6 << 28);
+ ip6->hop_limit = 254;
+ ip6->protocol = IP_PROTOCOL_UDP;
+ ip6->src_address = exp->src_address.ip.ip6;
+ ip6->dst_address = exp->ipfix_collector.ip.ip6;
+ *ip = ip6;
+ *udp = (udp_header_t *) (ip6 + 1);
+ (*udp)->length = clib_host_to_net_u16 (len - sizeof (*ip6));
+ return sizeof (*ip6);
+ }
+}
+
u8 *
-vnet_flow_rewrite_generic_callback (flow_report_main_t * frm,
- flow_report_t * fr,
- ip4_address_t * collector_address,
- ip4_address_t * src_address,
+vnet_flow_rewrite_generic_callback (ipfix_exporter_t *exp, flow_report_t *fr,
u16 collector_port,
- ipfix_report_element_t * report_elts,
- u32 n_elts, u32 * stream_indexp)
+ ipfix_report_element_t *report_elts,
+ u32 n_elts, u32 *stream_indexp)
{
- ip4_header_t *ip;
+ ip4_header_t *ip4;
+ ip6_header_t *ip6;
+ void *ip;
udp_header_t *udp;
ipfix_message_header_t *h;
ipfix_set_header_t *s;
ipfix_field_specifier_t *f;
ipfix_field_specifier_t *first_field;
u8 *rewrite = 0;
- ip4_ipfix_template_packet_t *tp;
flow_report_stream_t *stream;
int i;
ipfix_report_element_t *ep;
+ u32 size;
ASSERT (stream_indexp);
ASSERT (n_elts);
ASSERT (report_elts);
- stream = &frm->streams[fr->stream_index];
+ stream = &exp->streams[fr->stream_index];
*stream_indexp = fr->stream_index;
+ if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
+ size = sizeof (ip4_ipfix_template_packet_t);
+ else
+ size = sizeof (ip6_ipfix_template_packet_t);
/* allocate rewrite space */
vec_validate_aligned (rewrite,
- sizeof (ip4_ipfix_template_packet_t)
- + n_elts * sizeof (ipfix_field_specifier_t) - 1,
+ size + n_elts * sizeof (ipfix_field_specifier_t) - 1,
CLIB_CACHE_LINE_BYTES);
/* create the packet rewrite string */
- tp = (ip4_ipfix_template_packet_t *) rewrite;
- ip = (ip4_header_t *) & tp->ip4;
- udp = (udp_header_t *) (ip + 1);
+ ipfix_write_headers (exp, rewrite, &ip, &udp, vec_len (rewrite));
+
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 LATER: message header export_time */
h->domain_id = clib_host_to_net_u32 (stream->domain_id);
ep++;
}
- /* 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);
/* 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);
+ if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
+ {
+ ip4 = (ip4_header_t *) ip;
+ ip4->length = clib_host_to_net_u16 ((u8 *) f - (u8 *) ip4);
+ ip4->checksum = ip4_header_checksum (ip4);
+ }
+ else
+ {
+ ip6 = (ip6_header_t *) ip;
+ /* IPv6 payload length does not include the IPv6 header */
+ ip6->payload_length = clib_host_to_net_u16 ((u8 *) f - (u8 *) udp);
+ }
return rewrite;
}
+vlib_buffer_t *
+vnet_ipfix_exp_get_buffer (vlib_main_t *vm, ipfix_exporter_t *exp,
+ flow_report_t *fr, u32 thread_index)
+{
+ u32 bi0;
+ vlib_buffer_t *b0;
+
+ if (fr->per_thread_data[thread_index].buffer)
+ return fr->per_thread_data[thread_index].buffer;
+
+ if (vlib_buffer_alloc (vm, &bi0, 1) != 1)
+ return NULL;
+
+ /* Initialize the buffer */
+ b0 = fr->per_thread_data[thread_index].buffer = vlib_get_buffer (vm, bi0);
+
+ b0->current_data = 0;
+ b0->current_length = exp->all_headers_size;
+ b0->flags |= (VLIB_BUFFER_TOTAL_LENGTH_VALID | VNET_BUFFER_F_FLOW_REPORT);
+ vnet_buffer (b0)->sw_if_index[VLIB_RX] = 0;
+ vnet_buffer (b0)->sw_if_index[VLIB_TX] = exp->fib_index;
+ fr->per_thread_data[thread_index].next_data_offset = b0->current_length;
+
+ return b0;
+}
+
+/*
+ * Send a buffer that is mostly populated. Has flow records but needs some
+ * header fields updated.
+ */
+void
+vnet_ipfix_exp_send_buffer (vlib_main_t *vm, ipfix_exporter_t *exp,
+ flow_report_t *fr, flow_report_stream_t *stream,
+ u32 thread_index, vlib_buffer_t *b0)
+{
+ flow_report_main_t *frm = &flow_report_main;
+ vlib_frame_t *f;
+ ipfix_set_header_t *s;
+ ipfix_message_header_t *h;
+ ip4_header_t *ip4 = 0;
+ ip6_header_t *ip6 = 0;
+ void *ip;
+ udp_header_t *udp;
+ int ip_len;
+
+ /* nothing to send */
+ if (fr->per_thread_data[thread_index].next_data_offset <=
+ exp->all_headers_size)
+ return;
+
+ ip_len = ipfix_write_headers (exp, (void *) vlib_buffer_get_current (b0),
+ &ip, &udp, b0->current_length);
+
+ h = (ipfix_message_header_t *) (udp + 1);
+ s = (ipfix_set_header_t *) (h + 1);
+
+ udp->src_port = clib_host_to_net_u16 (stream->src_port);
+ udp->dst_port = clib_host_to_net_u16 (exp->collector_port);
+ udp->checksum = 0;
+
+ /* FIXUP: message header export_time */
+ h->export_time =
+ (u32) (((f64) frm->unix_time_0) + (vlib_time_now (vm) - 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);
+
+ /*
+ * RFC 7011: Section 3.2
+ *
+ * Incremental sequence counter modulo 2^32 of all IPFIX Data Records
+ * sent in the current stream from the current Observation Domain by
+ * the Exporting Process
+ */
+ h->sequence_number =
+ clib_atomic_fetch_add (&stream->sequence_number,
+ fr->per_thread_data[thread_index].n_data_records);
+ h->sequence_number = clib_host_to_net_u32 (h->sequence_number);
+
+ /*
+ * For data records we use the template ID as the set ID.
+ * RFC 7011: 3.4.3
+ */
+ s->set_id_length = ipfix_set_id_length (
+ fr->template_id,
+ b0->current_length - (ip_len + sizeof (*udp) + sizeof (*h)));
+ h->version_length =
+ version_length (b0->current_length - (ip_len + sizeof (*udp)));
+
+ if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
+ {
+ ip4 = (ip4_header_t *) ip;
+ ip4->length = clib_host_to_net_u16 (b0->current_length);
+ ip4->checksum = ip4_header_checksum (ip4);
+ udp->length = clib_host_to_net_u16 (b0->current_length - sizeof (*ip4));
+ ASSERT (ip4_header_checksum_is_valid (ip4));
+ }
+ else
+ {
+ ip6 = (ip6_header_t *) ip;
+ /* Ipv6 payload length does not include the IPv6 header */
+ ip6->payload_length =
+ clib_host_to_net_u16 (b0->current_length - sizeof (*ip6));
+ udp->length = clib_host_to_net_u16 (b0->current_length - sizeof (*ip6));
+ }
+
+ if (exp->udp_checksum || ip_addr_version (&exp->ipfix_collector) == AF_IP6)
+ {
+ /* RFC 7011 section 10.3.2. */
+ if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
+ udp->checksum = ip4_tcp_udp_compute_checksum (vm, b0, ip4);
+ else
+ {
+ int bogus = 0;
+ udp->checksum =
+ ip6_tcp_udp_icmp_compute_checksum (vm, b0, ip6, &bogus);
+ }
+ if (udp->checksum == 0)
+ udp->checksum = 0xffff;
+ }
+
+ /* Find or allocate a frame */
+ f = fr->per_thread_data[thread_index].frame;
+ if (PREDICT_FALSE (f == 0))
+ {
+ u32 *to_next;
+ if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
+ f = vlib_get_frame_to_node (vm, ip4_lookup_node.index);
+ else
+ f = vlib_get_frame_to_node (vm, ip6_lookup_node.index);
+ fr->per_thread_data[thread_index].frame = 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;
+ }
+
+ if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
+ vlib_put_frame_to_node (vm, ip4_lookup_node.index, f);
+ else
+ vlib_put_frame_to_node (vm, ip6_lookup_node.index, f);
+
+ fr->per_thread_data[thread_index].frame = NULL;
+ fr->per_thread_data[thread_index].buffer = NULL;
+ fr->per_thread_data[thread_index].next_data_offset = 0;
+}
+
+static void
+flow_report_process_send (vlib_main_t *vm, flow_report_main_t *frm,
+ ipfix_exporter_t *exp, flow_report_t *fr,
+ u32 next_node, u32 template_bi)
+{
+ vlib_frame_t *nf = 0;
+ u32 *to_next;
+
+ nf = vlib_get_frame_to_node (vm, next_node);
+ nf->n_vectors = 0;
+ to_next = vlib_frame_vector_args (nf);
+
+ if (template_bi != ~0)
+ {
+ to_next[0] = template_bi;
+ to_next++;
+ nf->n_vectors++;
+ }
+
+ nf = fr->flow_data_callback (frm, exp, fr, nf, to_next, next_node);
+ if (nf)
+ {
+ if (nf->n_vectors)
+ vlib_put_frame_to_node (vm, next_node, nf);
+ else
+ {
+ vlib_frame_free (vm, nf);
+ }
+ }
+}
+
static uword
flow_report_process (vlib_main_t * vm,
vlib_node_runtime_t * rt, vlib_frame_t * f)
flow_report_t *fr;
u32 ip4_lookup_node_index;
vlib_node_t *ip4_lookup_node;
- vlib_frame_t *nf = 0;
+ u32 ip6_lookup_node_index;
+ vlib_node_t *ip6_lookup_node;
u32 template_bi;
- u32 *to_next;
int send_template;
- f64 now;
+ f64 now, wait_time;
+ f64 def_wait_time = 5.0;
int rv;
uword event_type;
uword *event_data = 0;
ip4_lookup_node = vlib_get_node_by_name (vm, (u8 *) "ip4-lookup");
ip4_lookup_node_index = ip4_lookup_node->index;
+ /* Enqueue pkts to ip6-lookup */
+ ip6_lookup_node = vlib_get_node_by_name (vm, (u8 *) "ip6-lookup");
+ ip6_lookup_node_index = ip6_lookup_node->index;
+
+ wait_time = def_wait_time;
+
while (1)
{
- vlib_process_wait_for_event_or_clock (vm, 5.0);
+ vlib_process_wait_for_event_or_clock (vm, wait_time);
event_type = vlib_process_get_events (vm, &event_data);
vec_reset_length (event_data);
+ ipfix_exporter_t *exp;
+ pool_foreach (exp, frm->exporters)
+ {
- vec_foreach (fr, frm->reports)
- {
- now = vlib_time_now (vm);
-
- /* Need to send a template packet? */
- send_template =
- now > (fr->last_template_sent + frm->template_interval);
- send_template += fr->last_template_sent == 0;
- template_bi = ~0;
- rv = 0;
-
- if (send_template)
- rv = send_template_packet (frm, fr, &template_bi);
-
- if (rv < 0)
- continue;
-
- nf = vlib_get_frame_to_node (vm, ip4_lookup_node_index);
- nf->n_vectors = 0;
- to_next = vlib_frame_vector_args (nf);
-
- if (template_bi != ~0)
- {
- to_next[0] = template_bi;
- to_next++;
- nf->n_vectors++;
- }
-
- nf = fr->flow_data_callback (frm, fr,
- nf, to_next, ip4_lookup_node_index);
- if (nf)
- vlib_put_frame_to_node (vm, ip4_lookup_node_index, nf);
- }
+ /* 5s delay by default, possibly reduced by template intervals */
+ wait_time = def_wait_time;
+
+ vec_foreach (fr, exp->reports)
+ {
+ f64 next_template;
+ now = vlib_time_now (vm);
+
+ /* Need to send a template packet? */
+ send_template =
+ now > (fr->last_template_sent + exp->template_interval);
+ send_template += fr->last_template_sent == 0;
+ template_bi = ~0;
+ rv = 0;
+
+ if (send_template)
+ rv = send_template_packet (frm, exp, fr, &template_bi);
+
+ if (rv < 0)
+ continue;
+
+ /*
+ * decide if template should be sent sooner than current wait
+ * time
+ */
+ next_template =
+ (fr->last_template_sent + exp->template_interval) - now;
+ wait_time = clib_min (wait_time, next_template);
+
+ if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
+ {
+ flow_report_process_send (
+ vm, frm, exp, fr, ip4_lookup_node_index, template_bi);
+ }
+ else
+ {
+ flow_report_process_send (
+ vm, frm, exp, fr, ip6_lookup_node_index, template_bi);
+ }
+ }
+ }
}
return 0; /* not so much */
}
-/* *INDENT-OFF* */
VLIB_REGISTER_NODE (flow_report_process_node) = {
.function = flow_report_process,
.type = VLIB_NODE_TYPE_PROCESS,
.name = "flow-report-process",
};
-/* *INDENT-ON* */
int
-vnet_flow_report_add_del (flow_report_main_t * frm,
- vnet_flow_report_add_del_args_t * a,
- u16 * template_id)
+vnet_flow_report_add_del (ipfix_exporter_t *exp,
+ vnet_flow_report_add_del_args_t *a, u16 *template_id)
{
int i;
int found_index = ~0;
flow_report_t *fr;
flow_report_stream_t *stream;
u32 si;
+ vlib_thread_main_t *tm = &vlib_thread_main;
+ flow_report_main_t *frm = &flow_report_main;
+ vlib_main_t *vm = frm->vlib_main;
+ int size;
- si = find_stream (a->domain_id, a->src_port);
+ si = find_stream (exp, a->domain_id, a->src_port);
if (si == -2)
return VNET_API_ERROR_INVALID_VALUE;
if (si == -1 && a->is_add == 0)
return VNET_API_ERROR_NO_SUCH_ENTRY;
- for (i = 0; i < vec_len (frm->reports); i++)
+ for (i = 0; i < vec_len (exp->reports); i++)
{
- fr = vec_elt_at_index (frm->reports, i);
+ fr = vec_elt_at_index (exp->reports, i);
if (fr->opaque.as_uword == a->opaque.as_uword
&& fr->rewrite_callback == a->rewrite_callback
&& fr->flow_data_callback == a->flow_data_callback)
{
if (found_index != ~0)
{
- vec_delete (frm->reports, 1, found_index);
- stream = &frm->streams[si];
+ for (int i = 0;
+ i < vec_len (exp->reports[found_index].per_thread_data); i++)
+ {
+ u32 bi;
+ if (exp->reports[found_index].per_thread_data[i].buffer)
+ {
+ bi = vlib_get_buffer_index (
+ vm, exp->reports[found_index].per_thread_data[i].buffer);
+ vlib_buffer_free (vm, &bi, 1);
+ }
+ }
+ vec_free (exp->reports[found_index].per_thread_data);
+
+ vec_delete (exp->reports, 1, found_index);
+ stream = &exp->streams[si];
stream->n_reports--;
if (stream->n_reports == 0)
- delete_stream (si);
+ delete_stream (exp, si);
return 0;
}
return VNET_API_ERROR_NO_SUCH_ENTRY;
if (si == -1)
{
- stream = add_stream ();
+ stream = add_stream (exp);
stream->domain_id = a->domain_id;
stream->src_port = a->src_port;
stream->sequence_number = 0;
stream->n_reports = 0;
- si = stream - frm->streams;
+ si = stream - exp->streams;
}
else
- stream = &frm->streams[si];
+ stream = &exp->streams[si];
stream->n_reports++;
- vec_add2 (frm->reports, fr, 1);
+ vec_add2 (exp->reports, fr, 1);
fr->stream_index = si;
fr->template_id = 256 + stream->next_template_no;
fr->report_elements = a->report_elements;
fr->n_report_elements = a->n_report_elements;
fr->stream_indexp = a->stream_indexp;
+ vec_validate (fr->per_thread_data, tm->n_threads);
+ /* Store the flow_report index back in the args struct */
+ a->flow_report_index = fr - exp->reports;
+
+ size = 0;
+ for (int i = 0; i < fr->n_report_elements; i++)
+ size += fr->report_elements[i].size;
+ fr->data_record_size = size;
if (template_id)
*template_id = fr->template_id;
}
void
-vnet_flow_reports_reset (flow_report_main_t * frm)
+vnet_flow_reports_reset (ipfix_exporter_t *exp)
{
flow_report_t *fr;
u32 i;
- for (i = 0; i < vec_len (frm->streams); i++)
- if (stream_index_valid (i))
- frm->streams[i].sequence_number = 0;
+ for (i = 0; i < vec_len (exp->streams); i++)
+ if (stream_index_valid (exp, i))
+ exp->streams[i].sequence_number = 0;
- vec_foreach (fr, frm->reports)
- {
- fr->update_rewrite = 1;
- fr->last_template_sent = 0;
- }
+ vec_foreach (fr, exp->reports)
+ {
+ fr->update_rewrite = 1;
+ fr->last_template_sent = 0;
+ }
}
void
-vnet_stream_reset (flow_report_main_t * frm, u32 stream_index)
+vnet_stream_reset (ipfix_exporter_t *exp, u32 stream_index)
{
flow_report_t *fr;
- frm->streams[stream_index].sequence_number = 0;
+ exp->streams[stream_index].sequence_number = 0;
- vec_foreach (fr, frm->reports)
- if (frm->reports->stream_index == stream_index)
- {
- fr->update_rewrite = 1;
- fr->last_template_sent = 0;
- }
+ vec_foreach (fr, exp->reports)
+ if (exp->reports->stream_index == stream_index)
+ {
+ fr->update_rewrite = 1;
+ fr->last_template_sent = 0;
+ }
}
int
-vnet_stream_change (flow_report_main_t * frm,
- u32 old_domain_id, u16 old_src_port,
+vnet_stream_change (ipfix_exporter_t *exp, u32 old_domain_id, u16 old_src_port,
u32 new_domain_id, u16 new_src_port)
{
- i32 stream_index = find_stream (old_domain_id, old_src_port);
+ i32 stream_index = find_stream (exp, old_domain_id, old_src_port);
+
if (stream_index < 0)
return 1;
- flow_report_stream_t *stream = &frm->streams[stream_index];
+ flow_report_stream_t *stream = &exp->streams[stream_index];
stream->domain_id = new_domain_id;
stream->src_port = new_src_port;
if (old_domain_id != new_domain_id || old_src_port != new_src_port)
- vnet_stream_reset (frm, stream_index);
+ vnet_stream_reset (exp, stream_index);
return 0;
}
vlib_cli_command_t * cmd)
{
flow_report_main_t *frm = &flow_report_main;
- ip4_address_t collector, src;
+ ip_address_t collector = IP_ADDRESS_V4_ALL_0S, src = IP_ADDRESS_V4_ALL_0S;
u16 collector_port = UDP_DST_PORT_ipfix;
u32 fib_id;
u32 fib_index = ~0;
- collector.as_u32 = 0;
- src.as_u32 = 0;
u32 path_mtu = 512; // RFC 7011 section 10.3.3.
u32 template_interval = 20;
u8 udp_checksum = 0;
+ ipfix_exporter_t *exp = pool_elt_at_index (frm->exporters, 0);
+ u32 ip_header_size;
while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
{
- if (unformat (input, "collector %U", unformat_ip4_address, &collector))
+ if (unformat (input, "collector %U", unformat_ip4_address,
+ &collector.ip.ip4))
;
- else if (unformat (input, "port %u", &collector_port))
+ else if (unformat (input, "port %U", unformat_udp_port,
+ &collector_port))
;
- else if (unformat (input, "src %U", unformat_ip4_address, &src))
+ else if (unformat (input, "src %U", unformat_ip4_address, &src.ip.ip4))
;
else if (unformat (input, "fib-id %u", &fib_id))
{
break;
}
- if (collector.as_u32 != 0 && src.as_u32 == 0)
+ /*
+ * If the collector address is set then the src must be too.
+ * Collector address can be set to 0 to disable exporter
+ */
+ if (!ip_address_is_zero (&collector) && ip_address_is_zero (&src))
return clib_error_return (0, "src address required");
+ if (collector.version != src.version)
+ return clib_error_return (
+ 0, "src address and dest address must use same IP version");
if (path_mtu > 1450 /* vpp does not support fragmentation */ )
return clib_error_return (0, "too big path-mtu value, maximum is 1450");
if (path_mtu < 68)
return clib_error_return (0, "too small path-mtu value, minimum is 68");
+ /* Calculate how much header data we need. */
+ if (collector.version == AF_IP4)
+ ip_header_size = sizeof (ip4_header_t);
+ else
+ ip_header_size = sizeof (ip6_header_t);
+ exp->all_headers_size = ip_header_size + sizeof (udp_header_t) +
+ sizeof (ipfix_message_header_t) +
+ sizeof (ipfix_set_header_t);
+
/* Reset report streams if we are reconfiguring IP addresses */
- if (frm->ipfix_collector.as_u32 != collector.as_u32 ||
- frm->src_address.as_u32 != src.as_u32 ||
- frm->collector_port != collector_port)
- vnet_flow_reports_reset (frm);
-
- frm->ipfix_collector.as_u32 = collector.as_u32;
- frm->collector_port = collector_port;
- frm->src_address.as_u32 = src.as_u32;
- frm->fib_index = fib_index;
- frm->path_mtu = path_mtu;
- frm->template_interval = template_interval;
- frm->udp_checksum = udp_checksum;
-
- if (collector.as_u32)
- vlib_cli_output (vm, "Collector %U, src address %U, "
+ if (ip_address_cmp (&exp->ipfix_collector, &collector) ||
+ ip_address_cmp (&exp->src_address, &src) ||
+ exp->collector_port != collector_port)
+ vnet_flow_reports_reset (exp);
+
+ exp->ipfix_collector = collector;
+ exp->collector_port = collector_port;
+ exp->src_address = src;
+ exp->fib_index = fib_index;
+ exp->path_mtu = path_mtu;
+ exp->template_interval = template_interval;
+ exp->udp_checksum = udp_checksum;
+
+ if (collector.ip.ip4.as_u32)
+ vlib_cli_output (vm,
+ "Collector %U, src address %U, "
"fib index %d, path MTU %u, "
"template resend interval %us, "
"udp checksum %s",
- format_ip4_address, &frm->ipfix_collector,
- format_ip4_address, &frm->src_address,
- fib_index, path_mtu, template_interval,
+ format_ip4_address, &exp->ipfix_collector.ip.ip4,
+ format_ip4_address, &exp->src_address.ip.ip4, fib_index,
+ path_mtu, template_interval,
udp_checksum ? "enabled" : "disabled");
else
vlib_cli_output (vm, "IPFIX Collector is disabled");
return 0;
}
-/* *INDENT-OFF* */
VLIB_CLI_COMMAND (set_ipfix_exporter_command, static) = {
.path = "set ipfix exporter",
.short_help = "set ipfix exporter "
"collector <ip4-address> [port <port>] "
"src <ip4-address> [fib-id <fib-id>] "
"[path-mtu <path-mtu>] "
- "[template-interval <template-interval>]",
+ "[template-interval <template-interval>] "
"[udp-checksum]",
.function = set_ipfix_exporter_command_fn,
};
-/* *INDENT-ON* */
static clib_error_t *
return 0;
}
-/* *INDENT-OFF* */
VLIB_CLI_COMMAND (ipfix_flush_command, static) = {
.path = "ipfix flush",
.short_help = "flush the current ipfix data [for make test]",
.function = ipfix_flush_command_fn,
};
-/* *INDENT-ON* */
static clib_error_t *
flow_report_init (vlib_main_t * vm)
{
flow_report_main_t *frm = &flow_report_main;
+ ipfix_exporter_t *exp;
frm->vlib_main = vm;
frm->vnet_main = vnet_get_main ();
frm->unix_time_0 = time (0);
frm->vlib_time_0 = vlib_time_now (frm->vlib_main);
- frm->fib_index = ~0;
-
+ /*
+ * Make sure that we can always access the first exporter for
+ * backwards compatibility reasons.
+ */
+ pool_alloc (frm->exporters, IPFIX_EXPORTERS_MAX);
+ pool_get (frm->exporters, exp);
+ /* Verify that this is at index 0 */
+ ASSERT (frm->exporters == exp);
+ exp->fib_index = ~0;
return 0;
}
Code Review / vpp.git / blobdiff