flowprobe: add support for reporting on inbound packets
[vpp.git] / src / plugins / flowprobe / node.c
1 /*
2  * node.c - ipfix probe graph node
3  *
4  * Copyright (c) 2017 Cisco and/or its affiliates.
5  * Licensed under the Apache License, Version 2.0 (the "License");
6  * you may not use this file except in compliance with the License.
7  * You may obtain a copy of the License at:
8  *
9  *     http://www.apache.org/licenses/LICENSE-2.0
10  *
11  * Unless required by applicable law or agreed to in writing, software
12  * distributed under the License is distributed on an "AS IS" BASIS,
13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14  * See the License for the specific language governing permissions and
15  * limitations under the License.
16  */
17 #include <vlib/vlib.h>
18 #include <vnet/vnet.h>
19 #include <vppinfra/crc32.h>
20 #include <vppinfra/xxhash.h>
21 #include <vppinfra/error.h>
22 #include <flowprobe/flowprobe.h>
23 #include <vnet/ip/ip6_packet.h>
24 #include <vnet/udp/udp_local.h>
25 #include <vlibmemory/api.h>
26
27 static void flowprobe_export_entry (vlib_main_t * vm, flowprobe_entry_t * e);
28
29 /**
30  * @file node.c
31  * flow record generator graph node
32  */
33
34 typedef struct
35 {
36   /** interface handle */
37   u32 rx_sw_if_index;
38   u32 tx_sw_if_index;
39   /** packet timestamp */
40   u64 timestamp;
41   /** size of the buffer */
42   u16 buffer_size;
43
44   /** L2 information */
45   u8 src_mac[6];
46   u8 dst_mac[6];
47   /** Ethertype */
48   u16 ethertype;
49
50   /** L3 information */
51   ip46_address_t src_address;
52   ip46_address_t dst_address;
53   u8 protocol;
54   u8 tos;
55
56   /** L4 information */
57   u16 src_port;
58   u16 dst_port;
59
60   flowprobe_variant_t which;
61 } flowprobe_trace_t;
62
63 static char *flowprobe_variant_strings[] = {
64   [FLOW_VARIANT_IP4] = "IP4",
65   [FLOW_VARIANT_IP6] = "IP6",
66   [FLOW_VARIANT_L2] = "L2",
67   [FLOW_VARIANT_L2_IP4] = "L2-IP4",
68   [FLOW_VARIANT_L2_IP6] = "L2-IP6",
69 };
70
71 /* packet trace format function */
72 static u8 *
73 format_flowprobe_trace (u8 * s, va_list * args)
74 {
75   CLIB_UNUSED (vlib_main_t * vm) = va_arg (*args, vlib_main_t *);
76   CLIB_UNUSED (vlib_node_t * node) = va_arg (*args, vlib_node_t *);
77   flowprobe_trace_t *t = va_arg (*args, flowprobe_trace_t *);
78   u32 indent = format_get_indent (s);
79
80   s = format (s,
81               "FLOWPROBE[%s]: rx_sw_if_index %d, tx_sw_if_index %d, "
82               "timestamp %lld, size %d", flowprobe_variant_strings[t->which],
83               t->rx_sw_if_index, t->tx_sw_if_index,
84               t->timestamp, t->buffer_size);
85
86   if (t->which == FLOW_VARIANT_L2)
87     s = format (s, "\n%U -> %U", format_white_space, indent,
88                 format_ethernet_address, &t->src_mac,
89                 format_ethernet_address, &t->dst_mac);
90
91   if (t->protocol > 0
92       && (t->which == FLOW_VARIANT_L2_IP4 || t->which == FLOW_VARIANT_IP4
93           || t->which == FLOW_VARIANT_L2_IP6 || t->which == FLOW_VARIANT_IP6))
94     s =
95       format (s, "\n%U%U: %U -> %U", format_white_space, indent,
96               format_ip_protocol, t->protocol, format_ip46_address,
97               &t->src_address, IP46_TYPE_ANY, format_ip46_address,
98               &t->dst_address, IP46_TYPE_ANY);
99   return s;
100 }
101
102 vlib_node_registration_t flowprobe_input_ip4_node;
103 vlib_node_registration_t flowprobe_input_ip6_node;
104 vlib_node_registration_t flowprobe_input_l2_node;
105 vlib_node_registration_t flowprobe_output_ip4_node;
106 vlib_node_registration_t flowprobe_output_ip6_node;
107 vlib_node_registration_t flowprobe_output_l2_node;
108
109 /* No counters at the moment */
110 #define foreach_flowprobe_error                 \
111 _(COLLISION, "Hash table collisions")           \
112 _(BUFFER, "Buffer allocation error")            \
113 _(EXPORTED_PACKETS, "Exported packets")         \
114 _(INPATH, "Exported packets in path")
115
116 typedef enum
117 {
118 #define _(sym,str) FLOWPROBE_ERROR_##sym,
119   foreach_flowprobe_error
120 #undef _
121     FLOWPROBE_N_ERROR,
122 } flowprobe_error_t;
123
124 static char *flowprobe_error_strings[] = {
125 #define _(sym,string) string,
126   foreach_flowprobe_error
127 #undef _
128 };
129
130 typedef enum
131 {
132   FLOWPROBE_NEXT_DROP,
133   FLOWPROBE_NEXT_IP4_LOOKUP,
134   FLOWPROBE_N_NEXT,
135 } flowprobe_next_t;
136
137 #define FLOWPROBE_NEXT_NODES {                                  \
138     [FLOWPROBE_NEXT_DROP] = "error-drop",                       \
139     [FLOWPROBE_NEXT_IP4_LOOKUP] = "ip4-lookup",         \
140 }
141
142 static inline flowprobe_variant_t
143 flowprobe_get_variant (flowprobe_variant_t which,
144                        flowprobe_record_t flags, u16 ethertype)
145 {
146   if (which == FLOW_VARIANT_L2
147       && (flags & FLOW_RECORD_L3 || flags & FLOW_RECORD_L4))
148     return ethertype == ETHERNET_TYPE_IP6 ? FLOW_VARIANT_L2_IP6 : ethertype ==
149       ETHERNET_TYPE_IP4 ? FLOW_VARIANT_L2_IP4 : FLOW_VARIANT_L2;
150   return which;
151 }
152
153 /*
154  * NTP rfc868 : 2 208 988 800 corresponds to 00:00  1 Jan 1970 GMT
155  */
156 #define NTP_TIMESTAMP 2208988800LU
157
158 static inline u32
159 flowprobe_common_add (vlib_buffer_t * to_b, flowprobe_entry_t * e, u16 offset)
160 {
161   u16 start = offset;
162
163   /* Ingress interface */
164   u32 rx_if = clib_host_to_net_u32 (e->key.rx_sw_if_index);
165   clib_memcpy_fast (to_b->data + offset, &rx_if, sizeof (rx_if));
166   offset += sizeof (rx_if);
167
168   /* Egress interface */
169   u32 tx_if = clib_host_to_net_u32 (e->key.tx_sw_if_index);
170   clib_memcpy_fast (to_b->data + offset, &tx_if, sizeof (tx_if));
171   offset += sizeof (tx_if);
172
173   /* Flow direction
174      0x00: ingress flow
175      0x01: egress flow */
176   to_b->data[offset++] = (e->key.direction == FLOW_DIRECTION_TX);
177
178   /* packet delta count */
179   u64 packetdelta = clib_host_to_net_u64 (e->packetcount);
180   clib_memcpy_fast (to_b->data + offset, &packetdelta, sizeof (u64));
181   offset += sizeof (u64);
182
183   /* flowStartNanoseconds */
184   u32 t = clib_host_to_net_u32 (e->flow_start.sec + NTP_TIMESTAMP);
185   clib_memcpy_fast (to_b->data + offset, &t, sizeof (u32));
186   offset += sizeof (u32);
187   t = clib_host_to_net_u32 (e->flow_start.nsec);
188   clib_memcpy_fast (to_b->data + offset, &t, sizeof (u32));
189   offset += sizeof (u32);
190
191   /* flowEndNanoseconds */
192   t = clib_host_to_net_u32 (e->flow_end.sec + NTP_TIMESTAMP);
193   clib_memcpy_fast (to_b->data + offset, &t, sizeof (u32));
194   offset += sizeof (u32);
195   t = clib_host_to_net_u32 (e->flow_end.nsec);
196   clib_memcpy_fast (to_b->data + offset, &t, sizeof (u32));
197   offset += sizeof (u32);
198
199   return offset - start;
200 }
201
202 static inline u32
203 flowprobe_l2_add (vlib_buffer_t * to_b, flowprobe_entry_t * e, u16 offset)
204 {
205   u16 start = offset;
206
207   /* src mac address */
208   clib_memcpy_fast (to_b->data + offset, &e->key.src_mac, 6);
209   offset += 6;
210
211   /* dst mac address */
212   clib_memcpy_fast (to_b->data + offset, &e->key.dst_mac, 6);
213   offset += 6;
214
215   /* ethertype */
216   clib_memcpy_fast (to_b->data + offset, &e->key.ethertype, 2);
217   offset += 2;
218
219   return offset - start;
220 }
221
222 static inline u32
223 flowprobe_l3_ip6_add (vlib_buffer_t * to_b, flowprobe_entry_t * e, u16 offset)
224 {
225   u16 start = offset;
226
227   /* ip6 src address */
228   clib_memcpy_fast (to_b->data + offset, &e->key.src_address,
229                     sizeof (ip6_address_t));
230   offset += sizeof (ip6_address_t);
231
232   /* ip6 dst address */
233   clib_memcpy_fast (to_b->data + offset, &e->key.dst_address,
234                     sizeof (ip6_address_t));
235   offset += sizeof (ip6_address_t);
236
237   /* Protocol */
238   to_b->data[offset++] = e->key.protocol;
239
240   /* octetDeltaCount */
241   u64 octetdelta = clib_host_to_net_u64 (e->octetcount);
242   clib_memcpy_fast (to_b->data + offset, &octetdelta, sizeof (u64));
243   offset += sizeof (u64);
244
245   return offset - start;
246 }
247
248 static inline u32
249 flowprobe_l3_ip4_add (vlib_buffer_t * to_b, flowprobe_entry_t * e, u16 offset)
250 {
251   u16 start = offset;
252
253   /* ip4 src address */
254   clib_memcpy_fast (to_b->data + offset, &e->key.src_address.ip4,
255                     sizeof (ip4_address_t));
256   offset += sizeof (ip4_address_t);
257
258   /* ip4 dst address */
259   clib_memcpy_fast (to_b->data + offset, &e->key.dst_address.ip4,
260                     sizeof (ip4_address_t));
261   offset += sizeof (ip4_address_t);
262
263   /* Protocol */
264   to_b->data[offset++] = e->key.protocol;
265
266   /* octetDeltaCount */
267   u64 octetdelta = clib_host_to_net_u64 (e->octetcount);
268   clib_memcpy_fast (to_b->data + offset, &octetdelta, sizeof (u64));
269   offset += sizeof (u64);
270
271   return offset - start;
272 }
273
274 static inline u32
275 flowprobe_l4_add (vlib_buffer_t * to_b, flowprobe_entry_t * e, u16 offset)
276 {
277   u16 start = offset;
278
279   /* src port */
280   clib_memcpy_fast (to_b->data + offset, &e->key.src_port, 2);
281   offset += 2;
282
283   /* dst port */
284   clib_memcpy_fast (to_b->data + offset, &e->key.dst_port, 2);
285   offset += 2;
286
287   /* tcp control bits */
288   u16 control_bits = htons (e->prot.tcp.flags);
289   clib_memcpy_fast (to_b->data + offset, &control_bits, 2);
290   offset += 2;
291
292   return offset - start;
293 }
294
295 static inline u32
296 flowprobe_hash (flowprobe_key_t * k)
297 {
298   flowprobe_main_t *fm = &flowprobe_main;
299   u32 h = 0;
300
301 #ifdef clib_crc32c_uses_intrinsics
302   h = clib_crc32c ((u8 *) k, sizeof (*k));
303 #else
304   int i;
305   u64 tmp = 0;
306   for (i = 0; i < sizeof (*k) / 8; i++)
307     tmp ^= ((u64 *) k)[i];
308
309   h = clib_xxhash (tmp);
310 #endif
311
312   return h >> (32 - fm->ht_log2len);
313 }
314
315 flowprobe_entry_t *
316 flowprobe_lookup (u32 my_cpu_number, flowprobe_key_t * k, u32 * poolindex,
317                   bool * collision)
318 {
319   flowprobe_main_t *fm = &flowprobe_main;
320   flowprobe_entry_t *e;
321   u32 h;
322
323   h = (fm->active_timer) ? flowprobe_hash (k) : 0;
324
325   /* Lookup in the flow state pool */
326   *poolindex = fm->hash_per_worker[my_cpu_number][h];
327   if (*poolindex != ~0)
328     {
329       e = pool_elt_at_index (fm->pool_per_worker[my_cpu_number], *poolindex);
330       if (e)
331         {
332           /* Verify key or report collision */
333           if (memcmp (k, &e->key, sizeof (flowprobe_key_t)))
334             *collision = true;
335           return e;
336         }
337     }
338
339   return 0;
340 }
341
342 flowprobe_entry_t *
343 flowprobe_create (u32 my_cpu_number, flowprobe_key_t * k, u32 * poolindex)
344 {
345   flowprobe_main_t *fm = &flowprobe_main;
346   u32 h;
347
348   flowprobe_entry_t *e;
349
350   /* Get my index */
351   h = (fm->active_timer) ? flowprobe_hash (k) : 0;
352
353   pool_get (fm->pool_per_worker[my_cpu_number], e);
354   *poolindex = e - fm->pool_per_worker[my_cpu_number];
355   fm->hash_per_worker[my_cpu_number][h] = *poolindex;
356
357   e->key = *k;
358
359   if (fm->passive_timer > 0)
360     {
361       e->passive_timer_handle = tw_timer_start_2t_1w_2048sl
362         (fm->timers_per_worker[my_cpu_number], *poolindex, 0,
363          fm->passive_timer);
364     }
365   return e;
366 }
367
368 static inline void
369 add_to_flow_record_state (vlib_main_t *vm, vlib_node_runtime_t *node,
370                           flowprobe_main_t *fm, vlib_buffer_t *b,
371                           timestamp_nsec_t timestamp, u16 length,
372                           flowprobe_variant_t which,
373                           flowprobe_direction_t direction,
374                           flowprobe_trace_t *t)
375 {
376   if (fm->disabled)
377     return;
378
379   ASSERT (direction == FLOW_DIRECTION_RX || direction == FLOW_DIRECTION_TX);
380
381   u32 my_cpu_number = vm->thread_index;
382   u16 octets = 0;
383
384   flowprobe_record_t flags = fm->context[which].flags;
385   bool collect_ip4 = false, collect_ip6 = false;
386   ASSERT (b);
387   ethernet_header_t *eth = ethernet_buffer_get_header (b);
388   u16 ethertype = clib_net_to_host_u16 (eth->type);
389   u16 l2_hdr_sz = sizeof (ethernet_header_t);
390   /* *INDENT-OFF* */
391   flowprobe_key_t k = {};
392   /* *INDENT-ON* */
393   ip4_header_t *ip4 = 0;
394   ip6_header_t *ip6 = 0;
395   udp_header_t *udp = 0;
396   tcp_header_t *tcp = 0;
397   u8 tcp_flags = 0;
398
399   if (flags & FLOW_RECORD_L3 || flags & FLOW_RECORD_L4)
400     {
401       collect_ip4 = which == FLOW_VARIANT_L2_IP4 || which == FLOW_VARIANT_IP4;
402       collect_ip6 = which == FLOW_VARIANT_L2_IP6 || which == FLOW_VARIANT_IP6;
403     }
404
405   k.rx_sw_if_index = vnet_buffer (b)->sw_if_index[VLIB_RX];
406   k.tx_sw_if_index = vnet_buffer (b)->sw_if_index[VLIB_TX];
407
408   k.which = which;
409   k.direction = direction;
410
411   if (flags & FLOW_RECORD_L2)
412     {
413       clib_memcpy_fast (k.src_mac, eth->src_address, 6);
414       clib_memcpy_fast (k.dst_mac, eth->dst_address, 6);
415       k.ethertype = ethertype;
416     }
417   if (ethertype == ETHERNET_TYPE_VLAN)
418     {
419       /*VLAN TAG*/
420       ethernet_vlan_header_tv_t *ethv =
421         (ethernet_vlan_header_tv_t *) (&(eth->type));
422       /*Q in Q possibility */
423       while (clib_net_to_host_u16 (ethv->type) == ETHERNET_TYPE_VLAN)
424         {
425           ethv++;
426           l2_hdr_sz += sizeof (ethernet_vlan_header_tv_t);
427         }
428       k.ethertype = ethertype = clib_net_to_host_u16 ((ethv)->type);
429     }
430   if (collect_ip6 && ethertype == ETHERNET_TYPE_IP6)
431     {
432       ip6 = (ip6_header_t *) (b->data + l2_hdr_sz);
433       if (flags & FLOW_RECORD_L3)
434         {
435           k.src_address.as_u64[0] = ip6->src_address.as_u64[0];
436           k.src_address.as_u64[1] = ip6->src_address.as_u64[1];
437           k.dst_address.as_u64[0] = ip6->dst_address.as_u64[0];
438           k.dst_address.as_u64[1] = ip6->dst_address.as_u64[1];
439         }
440       k.protocol = ip6->protocol;
441       if (k.protocol == IP_PROTOCOL_UDP)
442         udp = (udp_header_t *) (ip6 + 1);
443       else if (k.protocol == IP_PROTOCOL_TCP)
444         tcp = (tcp_header_t *) (ip6 + 1);
445
446       octets = clib_net_to_host_u16 (ip6->payload_length)
447         + sizeof (ip6_header_t);
448     }
449   if (collect_ip4 && ethertype == ETHERNET_TYPE_IP4)
450     {
451       ip4 = (ip4_header_t *) (b->data + l2_hdr_sz);
452       if (flags & FLOW_RECORD_L3)
453         {
454           k.src_address.ip4.as_u32 = ip4->src_address.as_u32;
455           k.dst_address.ip4.as_u32 = ip4->dst_address.as_u32;
456         }
457       k.protocol = ip4->protocol;
458       if ((flags & FLOW_RECORD_L4) && k.protocol == IP_PROTOCOL_UDP)
459         udp = (udp_header_t *) (ip4 + 1);
460       else if ((flags & FLOW_RECORD_L4) && k.protocol == IP_PROTOCOL_TCP)
461         tcp = (tcp_header_t *) (ip4 + 1);
462
463       octets = clib_net_to_host_u16 (ip4->length);
464     }
465
466   if (udp)
467     {
468       k.src_port = udp->src_port;
469       k.dst_port = udp->dst_port;
470     }
471   else if (tcp)
472     {
473       k.src_port = tcp->src_port;
474       k.dst_port = tcp->dst_port;
475       tcp_flags = tcp->flags;
476     }
477
478   if (t)
479     {
480       t->rx_sw_if_index = k.rx_sw_if_index;
481       t->tx_sw_if_index = k.tx_sw_if_index;
482       clib_memcpy_fast (t->src_mac, k.src_mac, 6);
483       clib_memcpy_fast (t->dst_mac, k.dst_mac, 6);
484       t->ethertype = k.ethertype;
485       t->src_address.ip4.as_u32 = k.src_address.ip4.as_u32;
486       t->dst_address.ip4.as_u32 = k.dst_address.ip4.as_u32;
487       t->protocol = k.protocol;
488       t->src_port = k.src_port;
489       t->dst_port = k.dst_port;
490       t->which = k.which;
491     }
492
493   flowprobe_entry_t *e = 0;
494   f64 now = vlib_time_now (vm);
495   if (fm->active_timer > 0)
496     {
497       u32 poolindex = ~0;
498       bool collision = false;
499
500       e = flowprobe_lookup (my_cpu_number, &k, &poolindex, &collision);
501       if (collision)
502         {
503           /* Flush data and clean up entry for reuse. */
504           if (e->packetcount)
505             flowprobe_export_entry (vm, e);
506           e->key = k;
507           e->flow_start = timestamp;
508           vlib_node_increment_counter (vm, node->node_index,
509                                        FLOWPROBE_ERROR_COLLISION, 1);
510         }
511       if (!e)                   /* Create new entry */
512         {
513           e = flowprobe_create (my_cpu_number, &k, &poolindex);
514           e->last_exported = now;
515           e->flow_start = timestamp;
516         }
517     }
518   else
519     {
520       e = &fm->stateless_entry[my_cpu_number];
521       e->key = k;
522     }
523
524   if (e)
525     {
526       /* Updating entry */
527       e->packetcount++;
528       e->octetcount += octets;
529       e->last_updated = now;
530       e->flow_end = timestamp;
531       e->prot.tcp.flags |= tcp_flags;
532       if (fm->active_timer == 0
533           || (now > e->last_exported + fm->active_timer))
534         flowprobe_export_entry (vm, e);
535     }
536 }
537
538 static u16
539 flowprobe_get_headersize (void)
540 {
541   return sizeof (ip4_header_t) + sizeof (udp_header_t) +
542     sizeof (ipfix_message_header_t) + sizeof (ipfix_set_header_t);
543 }
544
545 static void
546 flowprobe_export_send (vlib_main_t * vm, vlib_buffer_t * b0,
547                        flowprobe_variant_t which)
548 {
549   flowprobe_main_t *fm = &flowprobe_main;
550   flow_report_main_t *frm = &flow_report_main;
551   ipfix_exporter_t *exp = pool_elt_at_index (frm->exporters, 0);
552   vlib_frame_t *f;
553   ip4_ipfix_template_packet_t *tp;
554   ipfix_set_header_t *s;
555   ipfix_message_header_t *h;
556   ip4_header_t *ip;
557   udp_header_t *udp;
558   flowprobe_record_t flags = fm->context[which].flags;
559   u32 my_cpu_number = vm->thread_index;
560
561   /* Fill in header */
562   flow_report_stream_t *stream;
563
564   /* Nothing to send */
565   if (fm->context[which].next_record_offset_per_worker[my_cpu_number] <=
566       flowprobe_get_headersize ())
567     return;
568
569   u32 i, index = vec_len (exp->streams);
570   for (i = 0; i < index; i++)
571     if (exp->streams[i].domain_id == 1)
572       {
573         index = i;
574         break;
575       }
576   if (i == vec_len (exp->streams))
577     {
578       vec_validate (exp->streams, index);
579       exp->streams[index].domain_id = 1;
580     }
581   stream = &exp->streams[index];
582
583   tp = vlib_buffer_get_current (b0);
584   ip = (ip4_header_t *) & tp->ip4;
585   udp = (udp_header_t *) (ip + 1);
586   h = (ipfix_message_header_t *) (udp + 1);
587   s = (ipfix_set_header_t *) (h + 1);
588
589   ip->ip_version_and_header_length = 0x45;
590   ip->ttl = 254;
591   ip->protocol = IP_PROTOCOL_UDP;
592   ip->flags_and_fragment_offset = 0;
593   ip->src_address.as_u32 = exp->src_address.ip.ip4.as_u32;
594   ip->dst_address.as_u32 = exp->ipfix_collector.ip.ip4.as_u32;
595   udp->src_port = clib_host_to_net_u16 (stream->src_port);
596   udp->dst_port = clib_host_to_net_u16 (exp->collector_port);
597   udp->checksum = 0;
598
599   /* FIXUP: message header export_time */
600   h->export_time = (u32)
601     (((f64) frm->unix_time_0) +
602      (vlib_time_now (frm->vlib_main) - frm->vlib_time_0));
603   h->export_time = clib_host_to_net_u32 (h->export_time);
604   h->domain_id = clib_host_to_net_u32 (stream->domain_id);
605
606   /* FIXUP: message header sequence_number */
607   h->sequence_number = stream->sequence_number++;
608   h->sequence_number = clib_host_to_net_u32 (h->sequence_number);
609
610   s->set_id_length = ipfix_set_id_length (fm->template_reports[flags],
611                                           b0->current_length -
612                                           (sizeof (*ip) + sizeof (*udp) +
613                                            sizeof (*h)));
614   h->version_length = version_length (b0->current_length -
615                                       (sizeof (*ip) + sizeof (*udp)));
616
617   ip->length = clib_host_to_net_u16 (b0->current_length);
618
619   ip->checksum = ip4_header_checksum (ip);
620   udp->length = clib_host_to_net_u16 (b0->current_length - sizeof (*ip));
621
622   if (exp->udp_checksum)
623     {
624       /* RFC 7011 section 10.3.2. */
625       udp->checksum = ip4_tcp_udp_compute_checksum (vm, b0, ip);
626       if (udp->checksum == 0)
627         udp->checksum = 0xffff;
628     }
629
630   ASSERT (ip4_header_checksum_is_valid (ip));
631
632   /* Find or allocate a frame */
633   f = fm->context[which].frames_per_worker[my_cpu_number];
634   if (PREDICT_FALSE (f == 0))
635     {
636       u32 *to_next;
637       f = vlib_get_frame_to_node (vm, ip4_lookup_node.index);
638       fm->context[which].frames_per_worker[my_cpu_number] = f;
639       u32 bi0 = vlib_get_buffer_index (vm, b0);
640
641       /* Enqueue the buffer */
642       to_next = vlib_frame_vector_args (f);
643       to_next[0] = bi0;
644       f->n_vectors = 1;
645     }
646
647   vlib_put_frame_to_node (vm, ip4_lookup_node.index, f);
648   vlib_node_increment_counter (vm, flowprobe_output_l2_node.index,
649                                FLOWPROBE_ERROR_EXPORTED_PACKETS, 1);
650
651   fm->context[which].frames_per_worker[my_cpu_number] = 0;
652   fm->context[which].buffers_per_worker[my_cpu_number] = 0;
653   fm->context[which].next_record_offset_per_worker[my_cpu_number] =
654     flowprobe_get_headersize ();
655 }
656
657 static vlib_buffer_t *
658 flowprobe_get_buffer (vlib_main_t * vm, flowprobe_variant_t which)
659 {
660   flowprobe_main_t *fm = &flowprobe_main;
661   ipfix_exporter_t *exp = pool_elt_at_index (flow_report_main.exporters, 0);
662   vlib_buffer_t *b0;
663   u32 bi0;
664   u32 my_cpu_number = vm->thread_index;
665
666   /* Find or allocate a buffer */
667   b0 = fm->context[which].buffers_per_worker[my_cpu_number];
668
669   /* Need to allocate a buffer? */
670   if (PREDICT_FALSE (b0 == 0))
671     {
672       if (vlib_buffer_alloc (vm, &bi0, 1) != 1)
673         {
674           vlib_node_increment_counter (vm, flowprobe_output_l2_node.index,
675                                        FLOWPROBE_ERROR_BUFFER, 1);
676           return 0;
677         }
678
679       /* Initialize the buffer */
680       b0 = fm->context[which].buffers_per_worker[my_cpu_number] =
681         vlib_get_buffer (vm, bi0);
682
683       b0->current_data = 0;
684       b0->current_length = flowprobe_get_headersize ();
685       b0->flags |=
686         (VLIB_BUFFER_TOTAL_LENGTH_VALID | VNET_BUFFER_F_FLOW_REPORT);
687       vnet_buffer (b0)->sw_if_index[VLIB_RX] = 0;
688       vnet_buffer (b0)->sw_if_index[VLIB_TX] = exp->fib_index;
689       fm->context[which].next_record_offset_per_worker[my_cpu_number] =
690         b0->current_length;
691     }
692
693   return b0;
694 }
695
696 static void
697 flowprobe_export_entry (vlib_main_t * vm, flowprobe_entry_t * e)
698 {
699   u32 my_cpu_number = vm->thread_index;
700   flowprobe_main_t *fm = &flowprobe_main;
701   ipfix_exporter_t *exp = pool_elt_at_index (flow_report_main.exporters, 0);
702   vlib_buffer_t *b0;
703   bool collect_ip4 = false, collect_ip6 = false;
704   flowprobe_variant_t which = e->key.which;
705   flowprobe_record_t flags = fm->context[which].flags;
706   u16 offset =
707     fm->context[which].next_record_offset_per_worker[my_cpu_number];
708
709   if (offset < flowprobe_get_headersize ())
710     offset = flowprobe_get_headersize ();
711
712   b0 = flowprobe_get_buffer (vm, which);
713   /* No available buffer, what to do... */
714   if (b0 == 0)
715     return;
716
717   if (flags & FLOW_RECORD_L3)
718     {
719       collect_ip4 = which == FLOW_VARIANT_L2_IP4 || which == FLOW_VARIANT_IP4;
720       collect_ip6 = which == FLOW_VARIANT_L2_IP6 || which == FLOW_VARIANT_IP6;
721     }
722
723   offset += flowprobe_common_add (b0, e, offset);
724
725   if (flags & FLOW_RECORD_L2)
726     offset += flowprobe_l2_add (b0, e, offset);
727   if (collect_ip6)
728     offset += flowprobe_l3_ip6_add (b0, e, offset);
729   if (collect_ip4)
730     offset += flowprobe_l3_ip4_add (b0, e, offset);
731   if (flags & FLOW_RECORD_L4)
732     offset += flowprobe_l4_add (b0, e, offset);
733
734   /* Reset per flow-export counters */
735   e->packetcount = 0;
736   e->octetcount = 0;
737   e->last_exported = vlib_time_now (vm);
738
739   b0->current_length = offset;
740
741   fm->context[which].next_record_offset_per_worker[my_cpu_number] = offset;
742   /* Time to flush the buffer? */
743   if (offset + fm->template_size[flags] > exp->path_mtu)
744     flowprobe_export_send (vm, b0, which);
745 }
746
747 uword
748 flowprobe_node_fn (vlib_main_t *vm, vlib_node_runtime_t *node,
749                    vlib_frame_t *frame, flowprobe_variant_t which,
750                    flowprobe_direction_t direction)
751 {
752   u32 n_left_from, *from, *to_next;
753   flowprobe_next_t next_index;
754   flowprobe_main_t *fm = &flowprobe_main;
755   timestamp_nsec_t timestamp;
756
757   unix_time_now_nsec_fraction (&timestamp.sec, &timestamp.nsec);
758
759   from = vlib_frame_vector_args (frame);
760   n_left_from = frame->n_vectors;
761   next_index = node->cached_next_index;
762
763   while (n_left_from > 0)
764     {
765       u32 n_left_to_next;
766
767       vlib_get_next_frame (vm, node, next_index, to_next, n_left_to_next);
768
769       while (n_left_from >= 4 && n_left_to_next >= 2)
770         {
771           u32 next0 = FLOWPROBE_NEXT_DROP;
772           u32 next1 = FLOWPROBE_NEXT_DROP;
773           u16 len0, len1;
774           u32 bi0, bi1;
775           vlib_buffer_t *b0, *b1;
776
777           /* Prefetch next iteration. */
778           {
779             vlib_buffer_t *p2, *p3;
780
781             p2 = vlib_get_buffer (vm, from[2]);
782             p3 = vlib_get_buffer (vm, from[3]);
783
784             vlib_prefetch_buffer_header (p2, LOAD);
785             vlib_prefetch_buffer_header (p3, LOAD);
786
787             clib_prefetch_store (p2->data);
788             clib_prefetch_store (p3->data);
789           }
790
791           /* speculatively enqueue b0 and b1 to the current next frame */
792           to_next[0] = bi0 = from[0];
793           to_next[1] = bi1 = from[1];
794           from += 2;
795           to_next += 2;
796           n_left_from -= 2;
797           n_left_to_next -= 2;
798
799           b0 = vlib_get_buffer (vm, bi0);
800           b1 = vlib_get_buffer (vm, bi1);
801
802           vnet_feature_next (&next0, b0);
803           vnet_feature_next (&next1, b1);
804
805           len0 = vlib_buffer_length_in_chain (vm, b0);
806           ethernet_header_t *eh0 = vlib_buffer_get_current (b0);
807           u16 ethertype0 = clib_net_to_host_u16 (eh0->type);
808
809           if (PREDICT_TRUE ((b0->flags & VNET_BUFFER_F_FLOW_REPORT) == 0))
810             add_to_flow_record_state (
811               vm, node, fm, b0, timestamp, len0,
812               flowprobe_get_variant (which, fm->context[which].flags,
813                                      ethertype0),
814               direction, 0);
815
816           len1 = vlib_buffer_length_in_chain (vm, b1);
817           ethernet_header_t *eh1 = vlib_buffer_get_current (b1);
818           u16 ethertype1 = clib_net_to_host_u16 (eh1->type);
819
820           if (PREDICT_TRUE ((b1->flags & VNET_BUFFER_F_FLOW_REPORT) == 0))
821             add_to_flow_record_state (
822               vm, node, fm, b1, timestamp, len1,
823               flowprobe_get_variant (which, fm->context[which].flags,
824                                      ethertype1),
825               direction, 0);
826
827           /* verify speculative enqueues, maybe switch current next frame */
828           vlib_validate_buffer_enqueue_x2 (vm, node, next_index,
829                                            to_next, n_left_to_next,
830                                            bi0, bi1, next0, next1);
831         }
832
833       while (n_left_from > 0 && n_left_to_next > 0)
834         {
835           u32 bi0;
836           vlib_buffer_t *b0;
837           u32 next0 = FLOWPROBE_NEXT_DROP;
838           u16 len0;
839
840           /* speculatively enqueue b0 to the current next frame */
841           bi0 = from[0];
842           to_next[0] = bi0;
843           from += 1;
844           to_next += 1;
845           n_left_from -= 1;
846           n_left_to_next -= 1;
847
848           b0 = vlib_get_buffer (vm, bi0);
849
850           vnet_feature_next (&next0, b0);
851
852           len0 = vlib_buffer_length_in_chain (vm, b0);
853           ethernet_header_t *eh0 = vlib_buffer_get_current (b0);
854           u16 ethertype0 = clib_net_to_host_u16 (eh0->type);
855
856           if (PREDICT_TRUE ((b0->flags & VNET_BUFFER_F_FLOW_REPORT) == 0))
857             {
858               flowprobe_trace_t *t = 0;
859               if (PREDICT_FALSE ((node->flags & VLIB_NODE_FLAG_TRACE)
860                                  && (b0->flags & VLIB_BUFFER_IS_TRACED)))
861                 t = vlib_add_trace (vm, node, b0, sizeof (*t));
862
863               add_to_flow_record_state (
864                 vm, node, fm, b0, timestamp, len0,
865                 flowprobe_get_variant (which, fm->context[which].flags,
866                                        ethertype0),
867                 direction, t);
868             }
869
870           /* verify speculative enqueue, maybe switch current next frame */
871           vlib_validate_buffer_enqueue_x1 (vm, node, next_index,
872                                            to_next, n_left_to_next,
873                                            bi0, next0);
874         }
875
876       vlib_put_next_frame (vm, node, next_index, n_left_to_next);
877     }
878   return frame->n_vectors;
879 }
880
881 static uword
882 flowprobe_input_ip4_node_fn (vlib_main_t *vm, vlib_node_runtime_t *node,
883                              vlib_frame_t *frame)
884 {
885   return flowprobe_node_fn (vm, node, frame, FLOW_VARIANT_IP4,
886                             FLOW_DIRECTION_RX);
887 }
888
889 static uword
890 flowprobe_input_ip6_node_fn (vlib_main_t *vm, vlib_node_runtime_t *node,
891                              vlib_frame_t *frame)
892 {
893   return flowprobe_node_fn (vm, node, frame, FLOW_VARIANT_IP6,
894                             FLOW_DIRECTION_RX);
895 }
896
897 static uword
898 flowprobe_input_l2_node_fn (vlib_main_t *vm, vlib_node_runtime_t *node,
899                             vlib_frame_t *frame)
900 {
901   return flowprobe_node_fn (vm, node, frame, FLOW_VARIANT_L2,
902                             FLOW_DIRECTION_RX);
903 }
904
905 static uword
906 flowprobe_output_ip4_node_fn (vlib_main_t *vm, vlib_node_runtime_t *node,
907                               vlib_frame_t *frame)
908 {
909   return flowprobe_node_fn (vm, node, frame, FLOW_VARIANT_IP4,
910                             FLOW_DIRECTION_TX);
911 }
912
913 static uword
914 flowprobe_output_ip6_node_fn (vlib_main_t *vm, vlib_node_runtime_t *node,
915                               vlib_frame_t *frame)
916 {
917   return flowprobe_node_fn (vm, node, frame, FLOW_VARIANT_IP6,
918                             FLOW_DIRECTION_TX);
919 }
920
921 static uword
922 flowprobe_output_l2_node_fn (vlib_main_t *vm, vlib_node_runtime_t *node,
923                              vlib_frame_t *frame)
924 {
925   return flowprobe_node_fn (vm, node, frame, FLOW_VARIANT_L2,
926                             FLOW_DIRECTION_TX);
927 }
928
929 static inline void
930 flush_record (flowprobe_variant_t which)
931 {
932   vlib_main_t *vm = vlib_get_main ();
933   vlib_buffer_t *b = flowprobe_get_buffer (vm, which);
934   if (b)
935     flowprobe_export_send (vm, b, which);
936 }
937
938 void
939 flowprobe_flush_callback_ip4 (void)
940 {
941   flush_record (FLOW_VARIANT_IP4);
942 }
943
944 void
945 flowprobe_flush_callback_ip6 (void)
946 {
947   flush_record (FLOW_VARIANT_IP6);
948 }
949
950 void
951 flowprobe_flush_callback_l2 (void)
952 {
953   flush_record (FLOW_VARIANT_L2);
954   flush_record (FLOW_VARIANT_L2_IP4);
955   flush_record (FLOW_VARIANT_L2_IP6);
956 }
957
958
959 static void
960 flowprobe_delete_by_index (u32 my_cpu_number, u32 poolindex)
961 {
962   flowprobe_main_t *fm = &flowprobe_main;
963   flowprobe_entry_t *e;
964   u32 h;
965
966   e = pool_elt_at_index (fm->pool_per_worker[my_cpu_number], poolindex);
967
968   /* Get my index */
969   h = flowprobe_hash (&e->key);
970
971   /* Reset hash */
972   fm->hash_per_worker[my_cpu_number][h] = ~0;
973
974   pool_put_index (fm->pool_per_worker[my_cpu_number], poolindex);
975 }
976
977
978 /* Per worker process processing the active/passive expired entries */
979 static uword
980 flowprobe_walker_process (vlib_main_t * vm,
981                           vlib_node_runtime_t * rt, vlib_frame_t * f)
982 {
983   flowprobe_main_t *fm = &flowprobe_main;
984   flowprobe_entry_t *e;
985   ipfix_exporter_t *exp = pool_elt_at_index (flow_report_main.exporters, 0);
986
987   /*
988    * $$$$ Remove this check from here and track FRM status and disable
989    * this process if required.
990    */
991   if (ip_address_is_zero (&exp->ipfix_collector) ||
992       ip_address_is_zero (&exp->src_address))
993     {
994       fm->disabled = true;
995       return 0;
996     }
997   fm->disabled = false;
998
999   u32 cpu_index = os_get_thread_index ();
1000   u32 *to_be_removed = 0, *i;
1001
1002   /*
1003    * Tick the timer when required and process the vector of expired
1004    * timers
1005    */
1006   f64 start_time = vlib_time_now (vm);
1007   u32 count = 0;
1008
1009   tw_timer_expire_timers_2t_1w_2048sl (fm->timers_per_worker[cpu_index],
1010                                        start_time);
1011
1012   vec_foreach (i, fm->expired_passive_per_worker[cpu_index])
1013   {
1014     u32 exported = 0;
1015     f64 now = vlib_time_now (vm);
1016     if (now > start_time + 100e-6
1017         || exported > FLOW_MAXIMUM_EXPORT_ENTRIES - 1)
1018       break;
1019
1020     if (pool_is_free_index (fm->pool_per_worker[cpu_index], *i))
1021       {
1022         clib_warning ("Element is %d is freed already\n", *i);
1023         continue;
1024       }
1025     else
1026       e = pool_elt_at_index (fm->pool_per_worker[cpu_index], *i);
1027
1028     /* Check last update timestamp. If it is longer than passive time nuke
1029      * entry. Otherwise restart timer with what's left
1030      * Premature passive timer by more than 10%
1031      */
1032     if ((now - e->last_updated) < (u64) (fm->passive_timer * 0.9))
1033       {
1034         u64 delta = fm->passive_timer - (now - e->last_updated);
1035         e->passive_timer_handle = tw_timer_start_2t_1w_2048sl
1036           (fm->timers_per_worker[cpu_index], *i, 0, delta);
1037       }
1038     else                        /* Nuke entry */
1039       {
1040         vec_add1 (to_be_removed, *i);
1041       }
1042     /* If anything to report send it to the exporter */
1043     if (e->packetcount && now > e->last_exported + fm->active_timer)
1044       {
1045         exported++;
1046         flowprobe_export_entry (vm, e);
1047       }
1048     count++;
1049   }
1050   if (count)
1051     vec_delete (fm->expired_passive_per_worker[cpu_index], count, 0);
1052
1053   vec_foreach (i, to_be_removed) flowprobe_delete_by_index (cpu_index, *i);
1054   vec_free (to_be_removed);
1055
1056   return 0;
1057 }
1058
1059 /* *INDENT-OFF* */
1060 VLIB_REGISTER_NODE (flowprobe_input_ip4_node) = {
1061   .function = flowprobe_input_ip4_node_fn,
1062   .name = "flowprobe-input-ip4",
1063   .vector_size = sizeof (u32),
1064   .format_trace = format_flowprobe_trace,
1065   .type = VLIB_NODE_TYPE_INTERNAL,
1066   .n_errors = ARRAY_LEN (flowprobe_error_strings),
1067   .error_strings = flowprobe_error_strings,
1068   .n_next_nodes = FLOWPROBE_N_NEXT,
1069   .next_nodes = FLOWPROBE_NEXT_NODES,
1070 };
1071 VLIB_REGISTER_NODE (flowprobe_input_ip6_node) = {
1072   .function = flowprobe_input_ip6_node_fn,
1073   .name = "flowprobe-input-ip6",
1074   .vector_size = sizeof (u32),
1075   .format_trace = format_flowprobe_trace,
1076   .type = VLIB_NODE_TYPE_INTERNAL,
1077   .n_errors = ARRAY_LEN (flowprobe_error_strings),
1078   .error_strings = flowprobe_error_strings,
1079   .n_next_nodes = FLOWPROBE_N_NEXT,
1080   .next_nodes = FLOWPROBE_NEXT_NODES,
1081 };
1082 VLIB_REGISTER_NODE (flowprobe_input_l2_node) = {
1083   .function = flowprobe_input_l2_node_fn,
1084   .name = "flowprobe-input-l2",
1085   .vector_size = sizeof (u32),
1086   .format_trace = format_flowprobe_trace,
1087   .type = VLIB_NODE_TYPE_INTERNAL,
1088   .n_errors = ARRAY_LEN (flowprobe_error_strings),
1089   .error_strings = flowprobe_error_strings,
1090   .n_next_nodes = FLOWPROBE_N_NEXT,
1091   .next_nodes = FLOWPROBE_NEXT_NODES,
1092 };
1093 VLIB_REGISTER_NODE (flowprobe_output_ip4_node) = {
1094   .function = flowprobe_output_ip4_node_fn,
1095   .name = "flowprobe-output-ip4",
1096   .vector_size = sizeof (u32),
1097   .format_trace = format_flowprobe_trace,
1098   .type = VLIB_NODE_TYPE_INTERNAL,
1099   .n_errors = ARRAY_LEN (flowprobe_error_strings),
1100   .error_strings = flowprobe_error_strings,
1101   .n_next_nodes = FLOWPROBE_N_NEXT,
1102   .next_nodes = FLOWPROBE_NEXT_NODES,
1103 };
1104 VLIB_REGISTER_NODE (flowprobe_output_ip6_node) = {
1105   .function = flowprobe_output_ip6_node_fn,
1106   .name = "flowprobe-output-ip6",
1107   .vector_size = sizeof (u32),
1108   .format_trace = format_flowprobe_trace,
1109   .type = VLIB_NODE_TYPE_INTERNAL,
1110   .n_errors = ARRAY_LEN (flowprobe_error_strings),
1111   .error_strings = flowprobe_error_strings,
1112   .n_next_nodes = FLOWPROBE_N_NEXT,
1113   .next_nodes = FLOWPROBE_NEXT_NODES,
1114 };
1115 VLIB_REGISTER_NODE (flowprobe_output_l2_node) = {
1116   .function = flowprobe_output_l2_node_fn,
1117   .name = "flowprobe-output-l2",
1118   .vector_size = sizeof (u32),
1119   .format_trace = format_flowprobe_trace,
1120   .type = VLIB_NODE_TYPE_INTERNAL,
1121   .n_errors = ARRAY_LEN (flowprobe_error_strings),
1122   .error_strings = flowprobe_error_strings,
1123   .n_next_nodes = FLOWPROBE_N_NEXT,
1124   .next_nodes = FLOWPROBE_NEXT_NODES,
1125 };
1126 VLIB_REGISTER_NODE (flowprobe_walker_node) = {
1127   .function = flowprobe_walker_process,
1128   .name = "flowprobe-walker",
1129   .type = VLIB_NODE_TYPE_INPUT,
1130   .state = VLIB_NODE_STATE_INTERRUPT,
1131 };
1132 /* *INDENT-ON* */
1133
1134 /*
1135  * fd.io coding-style-patch-verification: ON
1136  *
1137  * Local Variables:
1138  * eval: (c-set-style "gnu")
1139  * End:
1140  */