src/vnet/ipfix-export/flow_report.c

   1 /*
   2  * Copyright (c) 2015 Cisco and/or its affiliates.
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at:
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15 /*
  16  * flow_report.c
  17  */
  18 #include <vppinfra/atomics.h>
  19 #include <vnet/ipfix-export/flow_report.h>
  20 #include <vnet/api_errno.h>
  21 #include <vnet/udp/udp.h>
  22
  23 flow_report_main_t flow_report_main;
  24
  25 static_always_inline u8
  26 stream_index_valid (ipfix_exporter_t *exp, u32 index)
  27 {
  28   return index < vec_len (exp->streams) && exp->streams[index].domain_id != ~0;
  29 }
  30
  31 static_always_inline flow_report_stream_t *
  32 add_stream (ipfix_exporter_t *exp)
  33 {
  34   u32 i;
  35   for (i = 0; i < vec_len (exp->streams); i++)
  36     if (!stream_index_valid (exp, i))
  37       return &exp->streams[i];
  38   u32 index = vec_len (exp->streams);
  39   vec_validate (exp->streams, index);
  40   return &exp->streams[index];
  41 }
  42
  43 static_always_inline void
  44 delete_stream (ipfix_exporter_t *exp, u32 index)
  45 {
  46   ASSERT (index < vec_len (exp->streams));
  47   ASSERT (exp->streams[index].domain_id != ~0);
  48   exp->streams[index].domain_id = ~0;
  49 }
  50
  51 static i32
  52 find_stream (ipfix_exporter_t *exp, u32 domain_id, u16 src_port)
  53 {
  54   flow_report_stream_t *stream;
  55   u32 i;
  56   for (i = 0; i < vec_len (exp->streams); i++)
  57     if (stream_index_valid (exp, i))
  58       {
  59         stream = &exp->streams[i];
  60         if (domain_id == stream->domain_id)
  61           {
  62             if (src_port != stream->src_port)
  63               return -2;
  64             return i;
  65           }
  66         else if (src_port == stream->src_port)
  67           {
  68             return -2;
  69           }
  70       }
  71   return -1;
  72 }
  73
  74 int
  75 send_template_packet (flow_report_main_t *frm, ipfix_exporter_t *exp,
  76                       flow_report_t *fr, u32 *buffer_indexp)
  77 {
  78   u32 bi0;
  79   vlib_buffer_t *b0;
  80   ip4_ipfix_template_packet_t *tp4;
  81   ip6_ipfix_template_packet_t *tp6;
  82   ipfix_message_header_t *h;
  83   ip4_header_t *ip4;
  84   ip6_header_t *ip6;
  85   void *ip;
  86   udp_header_t *udp;
  87   vlib_main_t *vm = frm->vlib_main;
  88   flow_report_stream_t *stream;
  89
  90   ASSERT (buffer_indexp);
  91
  92   if (fr->update_rewrite || fr->rewrite == 0)
  93     {
  94       if (ip_address_is_zero (&exp->ipfix_collector) ||
  95           ip_address_is_zero (&exp->src_address))
  96         {
  97           vlib_node_set_state (frm->vlib_main, flow_report_process_node.index,
  98                                VLIB_NODE_STATE_DISABLED);
  99           return -1;
 100         }
 101       vec_free (fr->rewrite);
 102       fr->update_rewrite = 1;
 103     }
 104
 105   if (fr->update_rewrite)
 106     {
 107       fr->rewrite = fr->rewrite_callback (
 108         exp, fr, exp->collector_port, fr->report_elements,
 109         fr->n_report_elements, fr->stream_indexp);
 110       fr->update_rewrite = 0;
 111     }
 112
 113   if (vlib_buffer_alloc (vm, &bi0, 1) != 1)
 114     return -1;
 115
 116   b0 = vlib_get_buffer (vm, bi0);
 117
 118   ASSERT (vec_len (fr->rewrite) < vlib_buffer_get_default_data_size (vm));
 119
 120   clib_memcpy_fast (b0->data, fr->rewrite, vec_len (fr->rewrite));
 121   b0->current_data = 0;
 122   b0->current_length = vec_len (fr->rewrite);
 123   b0->flags |= (VLIB_BUFFER_TOTAL_LENGTH_VALID | VNET_BUFFER_F_FLOW_REPORT);
 124   vnet_buffer (b0)->sw_if_index[VLIB_RX] = 0;
 125   vnet_buffer (b0)->sw_if_index[VLIB_TX] = exp->fib_index;
 126
 127   if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
 128     {
 129       tp4 = vlib_buffer_get_current (b0);
 130       ip4 = (ip4_header_t *) &tp4->ip4;
 131       ip = ip4;
 132       udp = (udp_header_t *) (ip4 + 1);
 133     }
 134   else
 135     {
 136       tp6 = vlib_buffer_get_current (b0);
 137       ip6 = (ip6_header_t *) &tp6->ip6;
 138       ip = ip6;
 139       udp = (udp_header_t *) (ip6 + 1);
 140     }
 141   h = (ipfix_message_header_t *) (udp + 1);
 142
 143   /* FIXUP: message header export_time */
 144   h->export_time = (u32)
 145     (((f64) frm->unix_time_0) +
 146      (vlib_time_now (frm->vlib_main) - frm->vlib_time_0));
 147   h->export_time = clib_host_to_net_u32 (h->export_time);
 148
 149   stream = &exp->streams[fr->stream_index];
 150
 151   /* FIXUP: message header sequence_number. Templates do not increase it */
 152   h->sequence_number = clib_host_to_net_u32 (stream->sequence_number);
 153
 154   /* FIXUP: udp length */
 155   if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
 156     udp->length = clib_host_to_net_u16 (b0->current_length - sizeof (*ip4));
 157   else
 158     udp->length = clib_host_to_net_u16 (b0->current_length - sizeof (*ip6));
 159
 160   if (exp->udp_checksum || ip_addr_version (&exp->ipfix_collector) == AF_IP6)
 161     {
 162       /* RFC 7011 section 10.3.2. */
 163
 164       if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
 165         udp->checksum = ip4_tcp_udp_compute_checksum (vm, b0, ip);
 166       else
 167         {
 168           int bogus = 0;
 169           udp->checksum =
 170             ip6_tcp_udp_icmp_compute_checksum (vm, b0, ip, &bogus);
 171         }
 172
 173       if (udp->checksum == 0)
 174         udp->checksum = 0xffff;
 175     }
 176
 177   *buffer_indexp = bi0;
 178
 179   fr->last_template_sent = vlib_time_now (vm);
 180
 181   return 0;
 182 }
 183
 184 u32 always_inline
 185 ipfix_write_headers (ipfix_exporter_t *exp, void *data, void **ip,
 186                      udp_header_t **udp, u32 len)
 187 {
 188   if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
 189     {
 190       ip4_ipfix_template_packet_t *tp4;
 191       ip4_header_t *ip4;
 192
 193       tp4 = (ip4_ipfix_template_packet_t *) data;
 194       ip4 = (ip4_header_t *) &tp4->ip4;
 195       ip4->ip_version_and_header_length = 0x45;
 196       ip4->ttl = 254;
 197       ip4->protocol = IP_PROTOCOL_UDP;
 198       ip4->flags_and_fragment_offset = 0;
 199       ip4->src_address.as_u32 = exp->src_address.ip.ip4.as_u32;
 200       ip4->dst_address.as_u32 = exp->ipfix_collector.ip.ip4.as_u32;
 201       *ip = ip4;
 202       *udp = (udp_header_t *) (ip4 + 1);
 203
 204       (*udp)->length = clib_host_to_net_u16 (len - sizeof (*ip4));
 205       return sizeof (*ip4);
 206     }
 207   else
 208     {
 209       ip6_ipfix_template_packet_t *tp6;
 210       ip6_header_t *ip6;
 211
 212       tp6 = (ip6_ipfix_template_packet_t *) data;
 213       ip6 = (ip6_header_t *) &tp6->ip6;
 214       ip6->ip_version_traffic_class_and_flow_label =
 215         clib_host_to_net_u32 (6 << 28);
 216       ip6->hop_limit = 254;
 217       ip6->protocol = IP_PROTOCOL_UDP;
 218       ip6->src_address = exp->src_address.ip.ip6;
 219       ip6->dst_address = exp->ipfix_collector.ip.ip6;
 220       *ip = ip6;
 221       *udp = (udp_header_t *) (ip6 + 1);
 222       (*udp)->length = clib_host_to_net_u16 (len - sizeof (*ip6));
 223       return sizeof (*ip6);
 224     }
 225 }
 226
 227 u8 *
 228 vnet_flow_rewrite_generic_callback (ipfix_exporter_t *exp, flow_report_t *fr,
 229                                     u16 collector_port,
 230                                     ipfix_report_element_t *report_elts,
 231                                     u32 n_elts, u32 *stream_indexp)
 232 {
 233   ip4_header_t *ip4;
 234   ip6_header_t *ip6;
 235   void *ip;
 236   udp_header_t *udp;
 237   ipfix_message_header_t *h;
 238   ipfix_set_header_t *s;
 239   ipfix_template_header_t *t;
 240   ipfix_field_specifier_t *f;
 241   ipfix_field_specifier_t *first_field;
 242   u8 *rewrite = 0;
 243   flow_report_stream_t *stream;
 244   int i;
 245   ipfix_report_element_t *ep;
 246   u32 size;
 247
 248   ASSERT (stream_indexp);
 249   ASSERT (n_elts);
 250   ASSERT (report_elts);
 251
 252   stream = &exp->streams[fr->stream_index];
 253   *stream_indexp = fr->stream_index;
 254
 255   if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
 256     size = sizeof (ip4_ipfix_template_packet_t);
 257   else
 258     size = sizeof (ip6_ipfix_template_packet_t);
 259   /* allocate rewrite space */
 260   vec_validate_aligned (rewrite,
 261                         size + n_elts * sizeof (ipfix_field_specifier_t) - 1,
 262                         CLIB_CACHE_LINE_BYTES);
 263
 264   /* create the packet rewrite string */
 265   ipfix_write_headers (exp, rewrite, &ip, &udp, vec_len (rewrite));
 266
 267   h = (ipfix_message_header_t *) (udp + 1);
 268   s = (ipfix_set_header_t *) (h + 1);
 269   t = (ipfix_template_header_t *) (s + 1);
 270   first_field = f = (ipfix_field_specifier_t *) (t + 1);
 271   udp->src_port = clib_host_to_net_u16 (stream->src_port);
 272   udp->dst_port = clib_host_to_net_u16 (collector_port);
 273
 274   /* FIXUP LATER: message header export_time */
 275   h->domain_id = clib_host_to_net_u32 (stream->domain_id);
 276
 277   ep = report_elts;
 278
 279   for (i = 0; i < n_elts; i++)
 280     {
 281       f->e_id_length = ipfix_e_id_length (0, ep->info_element, ep->size);
 282       f++;
 283       ep++;
 284     }
 285
 286   ASSERT (f - first_field);
 287   /* Field count in this template */
 288   t->id_count = ipfix_id_count (fr->template_id, f - first_field);
 289
 290   /* set length in octets */
 291   s->set_id_length =
 292     ipfix_set_id_length (2 /* set_id */ , (u8 *) f - (u8 *) s);
 293
 294   /* message length in octets */
 295   h->version_length = version_length ((u8 *) f - (u8 *) h);
 296
 297   if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
 298     {
 299       ip4 = (ip4_header_t *) ip;
 300       ip4->length = clib_host_to_net_u16 ((u8 *) f - (u8 *) ip4);
 301       ip4->checksum = ip4_header_checksum (ip4);
 302     }
 303   else
 304     {
 305       ip6 = (ip6_header_t *) ip;
 306       /* IPv6 payload length does not include the IPv6 header */
 307       ip6->payload_length = clib_host_to_net_u16 ((u8 *) f - (u8 *) udp);
 308     }
 309
 310   return rewrite;
 311 }
 312
 313 vlib_buffer_t *
 314 vnet_ipfix_exp_get_buffer (vlib_main_t *vm, ipfix_exporter_t *exp,
 315                            flow_report_t *fr, u32 thread_index)
 316 {
 317   u32 bi0;
 318   vlib_buffer_t *b0;
 319
 320   if (fr->per_thread_data[thread_index].buffer)
 321     return fr->per_thread_data[thread_index].buffer;
 322
 323   if (vlib_buffer_alloc (vm, &bi0, 1) != 1)
 324     return NULL;
 325
 326   /* Initialize the buffer */
 327   b0 = fr->per_thread_data[thread_index].buffer = vlib_get_buffer (vm, bi0);
 328
 329   b0->current_data = 0;
 330   b0->current_length = exp->all_headers_size;
 331   b0->flags |= (VLIB_BUFFER_TOTAL_LENGTH_VALID | VNET_BUFFER_F_FLOW_REPORT);
 332   vnet_buffer (b0)->sw_if_index[VLIB_RX] = 0;
 333   vnet_buffer (b0)->sw_if_index[VLIB_TX] = exp->fib_index;
 334   fr->per_thread_data[thread_index].next_data_offset = b0->current_length;
 335
 336   return b0;
 337 }
 338
 339 /*
 340  * Send a buffer that is mostly populated. Has flow records but needs some
 341  * header fields updated.
 342  */
 343 void
 344 vnet_ipfix_exp_send_buffer (vlib_main_t *vm, ipfix_exporter_t *exp,
 345                             flow_report_t *fr, flow_report_stream_t *stream,
 346                             u32 thread_index, vlib_buffer_t *b0)
 347 {
 348   flow_report_main_t *frm = &flow_report_main;
 349   vlib_frame_t *f;
 350   ipfix_set_header_t *s;
 351   ipfix_message_header_t *h;
 352   ip4_header_t *ip4 = 0;
 353   ip6_header_t *ip6 = 0;
 354   void *ip;
 355   udp_header_t *udp;
 356   int ip_len;
 357
 358   /* nothing to send */
 359   if (fr->per_thread_data[thread_index].next_data_offset <=
 360       exp->all_headers_size)
 361     return;
 362
 363   ip_len = ipfix_write_headers (exp, (void *) vlib_buffer_get_current (b0),
 364                                 &ip, &udp, b0->current_length);
 365
 366   h = (ipfix_message_header_t *) (udp + 1);
 367   s = (ipfix_set_header_t *) (h + 1);
 368
 369   udp->src_port = clib_host_to_net_u16 (stream->src_port);
 370   udp->dst_port = clib_host_to_net_u16 (exp->collector_port);
 371   udp->checksum = 0;
 372
 373   /* FIXUP: message header export_time */
 374   h->export_time =
 375     (u32) (((f64) frm->unix_time_0) + (vlib_time_now (vm) - frm->vlib_time_0));
 376   h->export_time = clib_host_to_net_u32 (h->export_time);
 377   h->domain_id = clib_host_to_net_u32 (stream->domain_id);
 378
 379   /*
 380    * RFC 7011: Section 3.2
 381    *
 382    * Incremental sequence counter modulo 2^32 of all IPFIX Data Records
 383    * sent in the current stream from the current Observation Domain by
 384    * the Exporting Process
 385    */
 386   h->sequence_number =
 387     clib_atomic_fetch_add (&stream->sequence_number,
 388                            fr->per_thread_data[thread_index].n_data_records);
 389   h->sequence_number = clib_host_to_net_u32 (h->sequence_number);
 390
 391   /*
 392    * For data records we use the template ID as the set ID.
 393    * RFC 7011: 3.4.3
 394    */
 395   s->set_id_length = ipfix_set_id_length (
 396     fr->template_id,
 397     b0->current_length - (ip_len + sizeof (*udp) + sizeof (*h)));
 398   h->version_length =
 399     version_length (b0->current_length - (ip_len + sizeof (*udp)));
 400
 401   if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
 402     {
 403       ip4 = (ip4_header_t *) ip;
 404       ip4->length = clib_host_to_net_u16 (b0->current_length);
 405       ip4->checksum = ip4_header_checksum (ip4);
 406       udp->length = clib_host_to_net_u16 (b0->current_length - sizeof (*ip4));
 407       ASSERT (ip4_header_checksum_is_valid (ip4));
 408     }
 409   else
 410     {
 411       ip6 = (ip6_header_t *) ip;
 412       /* Ipv6 payload length does not include the IPv6 header */
 413       ip6->payload_length =
 414         clib_host_to_net_u16 (b0->current_length - sizeof (*ip6));
 415       udp->length = clib_host_to_net_u16 (b0->current_length - sizeof (*ip6));
 416     }
 417
 418   if (exp->udp_checksum || ip_addr_version (&exp->ipfix_collector) == AF_IP6)
 419     {
 420       /* RFC 7011 section 10.3.2. */
 421       if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
 422         udp->checksum = ip4_tcp_udp_compute_checksum (vm, b0, ip4);
 423       else
 424         {
 425           int bogus = 0;
 426           udp->checksum =
 427             ip6_tcp_udp_icmp_compute_checksum (vm, b0, ip6, &bogus);
 428         }
 429       if (udp->checksum == 0)
 430         udp->checksum = 0xffff;
 431     }
 432
 433   /* Find or allocate a frame */
 434   f = fr->per_thread_data[thread_index].frame;
 435   if (PREDICT_FALSE (f == 0))
 436     {
 437       u32 *to_next;
 438       if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
 439         f = vlib_get_frame_to_node (vm, ip4_lookup_node.index);
 440       else
 441         f = vlib_get_frame_to_node (vm, ip6_lookup_node.index);
 442       fr->per_thread_data[thread_index].frame = f;
 443       u32 bi0 = vlib_get_buffer_index (vm, b0);
 444
 445       /* Enqueue the buffer */
 446       to_next = vlib_frame_vector_args (f);
 447       to_next[0] = bi0;
 448       f->n_vectors = 1;
 449     }
 450
 451   if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
 452     vlib_put_frame_to_node (vm, ip4_lookup_node.index, f);
 453   else
 454     vlib_put_frame_to_node (vm, ip6_lookup_node.index, f);
 455
 456   fr->per_thread_data[thread_index].frame = NULL;
 457   fr->per_thread_data[thread_index].buffer = NULL;
 458   fr->per_thread_data[thread_index].next_data_offset = 0;
 459 }
 460
 461 static void
 462 flow_report_process_send (vlib_main_t *vm, flow_report_main_t *frm,
 463                           ipfix_exporter_t *exp, flow_report_t *fr,
 464                           u32 next_node, u32 template_bi)
 465 {
 466   vlib_frame_t *nf = 0;
 467   u32 *to_next;
 468
 469   nf = vlib_get_frame_to_node (vm, next_node);
 470   nf->n_vectors = 0;
 471   to_next = vlib_frame_vector_args (nf);
 472
 473   if (template_bi != ~0)
 474     {
 475       to_next[0] = template_bi;
 476       to_next++;
 477       nf->n_vectors++;
 478     }
 479
 480   nf = fr->flow_data_callback (frm, exp, fr, nf, to_next, next_node);
 481   if (nf)
 482     {
 483       if (nf->n_vectors)
 484         vlib_put_frame_to_node (vm, next_node, nf);
 485       else
 486         {
 487           vlib_frame_free (vm, nf);
 488         }
 489     }
 490 }
 491
 492 static uword
 493 flow_report_process (vlib_main_t * vm,
 494                      vlib_node_runtime_t * rt, vlib_frame_t * f)
 495 {
 496   flow_report_main_t *frm = &flow_report_main;
 497   flow_report_t *fr;
 498   u32 ip4_lookup_node_index;
 499   vlib_node_t *ip4_lookup_node;
 500   u32 ip6_lookup_node_index;
 501   vlib_node_t *ip6_lookup_node;
 502   u32 template_bi;
 503   int send_template;
 504   f64 now, wait_time;
 505   f64 def_wait_time = 5.0;
 506   int rv;
 507   uword event_type;
 508   uword *event_data = 0;
 509
 510   /* Wait for Godot... */
 511   vlib_process_wait_for_event_or_clock (vm, 1e9);
 512   event_type = vlib_process_get_events (vm, &event_data);
 513   if (event_type != 1)
 514     clib_warning ("bogus kickoff event received, %d", event_type);
 515   vec_reset_length (event_data);
 516
 517   /* Enqueue pkts to ip4-lookup */
 518   ip4_lookup_node = vlib_get_node_by_name (vm, (u8 *) "ip4-lookup");
 519   ip4_lookup_node_index = ip4_lookup_node->index;
 520
 521   /* Enqueue pkts to ip6-lookup */
 522   ip6_lookup_node = vlib_get_node_by_name (vm, (u8 *) "ip6-lookup");
 523   ip6_lookup_node_index = ip6_lookup_node->index;
 524
 525   wait_time = def_wait_time;
 526
 527   while (1)
 528     {
 529       vlib_process_wait_for_event_or_clock (vm, wait_time);
 530       event_type = vlib_process_get_events (vm, &event_data);
 531       vec_reset_length (event_data);
 532       ipfix_exporter_t *exp;
 533       pool_foreach (exp, frm->exporters)
 534         {
 535
 536           /* 5s delay by default, possibly reduced by template intervals */
 537           wait_time = def_wait_time;
 538
 539           vec_foreach (fr, exp->reports)
 540             {
 541               f64 next_template;
 542               now = vlib_time_now (vm);
 543
 544               /* Need to send a template packet? */
 545               send_template =
 546                 now > (fr->last_template_sent + exp->template_interval);
 547               send_template += fr->last_template_sent == 0;
 548               template_bi = ~0;
 549               rv = 0;
 550
 551               if (send_template)
 552                 rv = send_template_packet (frm, exp, fr, &template_bi);
 553
 554               if (rv < 0)
 555                 continue;
 556
 557               /*
 558                * decide if template should be sent sooner than current wait
 559                * time
 560                */
 561               next_template =
 562                 (fr->last_template_sent + exp->template_interval) - now;
 563               wait_time = clib_min (wait_time, next_template);
 564
 565               if (ip_addr_version (&exp->ipfix_collector) == AF_IP4)
 566                 {
 567                   flow_report_process_send (
 568                     vm, frm, exp, fr, ip4_lookup_node_index, template_bi);
 569                 }
 570               else
 571                 {
 572                   flow_report_process_send (
 573                     vm, frm, exp, fr, ip6_lookup_node_index, template_bi);
 574                 }
 575             }
 576         }
 577     }
 578
 579   return 0;                     /* not so much */
 580 }
 581
 582 VLIB_REGISTER_NODE (flow_report_process_node) = {
 583     .function = flow_report_process,
 584     .type = VLIB_NODE_TYPE_PROCESS,
 585     .name = "flow-report-process",
 586 };
 587
 588 int
 589 vnet_flow_report_add_del (ipfix_exporter_t *exp,
 590                           vnet_flow_report_add_del_args_t *a, u16 *template_id)
 591 {
 592   int i;
 593   int found_index = ~0;
 594   flow_report_t *fr;
 595   flow_report_stream_t *stream;
 596   u32 si;
 597   vlib_thread_main_t *tm = &vlib_thread_main;
 598   flow_report_main_t *frm = &flow_report_main;
 599   vlib_main_t *vm = frm->vlib_main;
 600   int size;
 601
 602   si = find_stream (exp, a->domain_id, a->src_port);
 603   if (si == -2)
 604     return VNET_API_ERROR_INVALID_VALUE;
 605   if (si == -1 && a->is_add == 0)
 606     return VNET_API_ERROR_NO_SUCH_ENTRY;
 607
 608   for (i = 0; i < vec_len (exp->reports); i++)
 609     {
 610       fr = vec_elt_at_index (exp->reports, i);
 611       if (fr->opaque.as_uword == a->opaque.as_uword
 612           && fr->rewrite_callback == a->rewrite_callback
 613           && fr->flow_data_callback == a->flow_data_callback)
 614         {
 615           found_index = i;
 616           if (template_id)
 617             *template_id = fr->template_id;
 618           break;
 619         }
 620     }
 621
 622   if (a->is_add == 0)
 623     {
 624       if (found_index != ~0)
 625         {
 626           for (int i = 0;
 627                i < vec_len (exp->reports[found_index].per_thread_data); i++)
 628             {
 629               u32 bi;
 630               if (exp->reports[found_index].per_thread_data[i].buffer)
 631                 {
 632                   bi = vlib_get_buffer_index (
 633                     vm, exp->reports[found_index].per_thread_data[i].buffer);
 634                   vlib_buffer_free (vm, &bi, 1);
 635                 }
 636             }
 637           vec_free (exp->reports[found_index].per_thread_data);
 638
 639           vec_delete (exp->reports, 1, found_index);
 640           stream = &exp->streams[si];
 641           stream->n_reports--;
 642           if (stream->n_reports == 0)
 643             delete_stream (exp, si);
 644           return 0;
 645         }
 646       return VNET_API_ERROR_NO_SUCH_ENTRY;
 647     }
 648
 649   if (found_index != ~0)
 650     return VNET_API_ERROR_VALUE_EXIST;
 651
 652   if (si == -1)
 653     {
 654       stream = add_stream (exp);
 655       stream->domain_id = a->domain_id;
 656       stream->src_port = a->src_port;
 657       stream->sequence_number = 0;
 658       stream->n_reports = 0;
 659       si = stream - exp->streams;
 660     }
 661   else
 662     stream = &exp->streams[si];
 663
 664   stream->n_reports++;
 665
 666   vec_add2 (exp->reports, fr, 1);
 667
 668   fr->stream_index = si;
 669   fr->template_id = 256 + stream->next_template_no;
 670   stream->next_template_no = (stream->next_template_no + 1) % (65536 - 256);
 671   fr->update_rewrite = 1;
 672   fr->opaque = a->opaque;
 673   fr->rewrite_callback = a->rewrite_callback;
 674   fr->flow_data_callback = a->flow_data_callback;
 675   fr->report_elements = a->report_elements;
 676   fr->n_report_elements = a->n_report_elements;
 677   fr->stream_indexp = a->stream_indexp;
 678   vec_validate (fr->per_thread_data, tm->n_threads);
 679   /* Store the flow_report index back in the args struct */
 680   a->flow_report_index = fr - exp->reports;
 681
 682   size = 0;
 683   for (int i = 0; i < fr->n_report_elements; i++)
 684     size += fr->report_elements[i].size;
 685   fr->data_record_size = size;
 686   if (template_id)
 687     *template_id = fr->template_id;
 688
 689   return 0;
 690 }
 691
 692 clib_error_t *
 693 flow_report_add_del_error_to_clib_error (int error)
 694 {
 695   switch (error)
 696     {
 697     case 0:
 698       return 0;
 699     case VNET_API_ERROR_NO_SUCH_ENTRY:
 700       return clib_error_return (0, "Flow report not found");
 701     case VNET_API_ERROR_VALUE_EXIST:
 702       return clib_error_return (0, "Flow report already exists");
 703     case VNET_API_ERROR_INVALID_VALUE:
 704       return clib_error_return (0, "Expecting either still unused values "
 705                                 "for both domain_id and src_port "
 706                                 "or already used values for both fields");
 707     default:
 708       return clib_error_return (0, "vnet_flow_report_add_del returned %d",
 709                                 error);
 710     }
 711 }
 712
 713 void
 714 vnet_flow_reports_reset (ipfix_exporter_t *exp)
 715 {
 716   flow_report_t *fr;
 717   u32 i;
 718
 719   for (i = 0; i < vec_len (exp->streams); i++)
 720     if (stream_index_valid (exp, i))
 721       exp->streams[i].sequence_number = 0;
 722
 723   vec_foreach (fr, exp->reports)
 724     {
 725       fr->update_rewrite = 1;
 726       fr->last_template_sent = 0;
 727     }
 728 }
 729
 730 void
 731 vnet_stream_reset (ipfix_exporter_t *exp, u32 stream_index)
 732 {
 733   flow_report_t *fr;
 734
 735   exp->streams[stream_index].sequence_number = 0;
 736
 737   vec_foreach (fr, exp->reports)
 738     if (exp->reports->stream_index == stream_index)
 739       {
 740         fr->update_rewrite = 1;
 741         fr->last_template_sent = 0;
 742       }
 743 }
 744
 745 int
 746 vnet_stream_change (ipfix_exporter_t *exp, u32 old_domain_id, u16 old_src_port,
 747                     u32 new_domain_id, u16 new_src_port)
 748 {
 749   i32 stream_index = find_stream (exp, old_domain_id, old_src_port);
 750
 751   if (stream_index < 0)
 752     return 1;
 753   flow_report_stream_t *stream = &exp->streams[stream_index];
 754   stream->domain_id = new_domain_id;
 755   stream->src_port = new_src_port;
 756   if (old_domain_id != new_domain_id || old_src_port != new_src_port)
 757     vnet_stream_reset (exp, stream_index);
 758   return 0;
 759 }
 760
 761 static clib_error_t *
 762 set_ipfix_exporter_command_fn (vlib_main_t * vm,
 763                                unformat_input_t * input,
 764                                vlib_cli_command_t * cmd)
 765 {
 766   flow_report_main_t *frm = &flow_report_main;
 767   ip_address_t collector = IP_ADDRESS_V4_ALL_0S, src = IP_ADDRESS_V4_ALL_0S;
 768   u16 collector_port = UDP_DST_PORT_ipfix;
 769   u32 fib_id;
 770   u32 fib_index = ~0;
 771
 772   u32 path_mtu = 512;           // RFC 7011 section 10.3.3.
 773   u32 template_interval = 20;
 774   u8 udp_checksum = 0;
 775   ipfix_exporter_t *exp = pool_elt_at_index (frm->exporters, 0);
 776   u32 ip_header_size;
 777
 778   while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
 779     {
 780       if (unformat (input, "collector %U", unformat_ip4_address,
 781                     &collector.ip.ip4))
 782         ;
 783       else if (unformat (input, "port %U", unformat_udp_port,
 784                          &collector_port))
 785         ;
 786       else if (unformat (input, "src %U", unformat_ip4_address, &src.ip.ip4))
 787         ;
 788       else if (unformat (input, "fib-id %u", &fib_id))
 789         {
 790           ip4_main_t *im = &ip4_main;
 791           uword *p = hash_get (im->fib_index_by_table_id, fib_id);
 792           if (!p)
 793             return clib_error_return (0, "fib ID %d doesn't exist\n", fib_id);
 794           fib_index = p[0];
 795         }
 796       else if (unformat (input, "path-mtu %u", &path_mtu))
 797         ;
 798       else if (unformat (input, "template-interval %u", &template_interval))
 799         ;
 800       else if (unformat (input, "udp-checksum"))
 801         udp_checksum = 1;
 802       else
 803         break;
 804     }
 805
 806   /*
 807    * If the collector address is set then the src must be too.
 808    * Collector address can be set to 0 to disable exporter
 809    */
 810   if (!ip_address_is_zero (&collector) && ip_address_is_zero (&src))
 811     return clib_error_return (0, "src address required");
 812   if (collector.version != src.version)
 813     return clib_error_return (
 814       0, "src address and dest address must use same IP version");
 815
 816   if (path_mtu > 1450 /* vpp does not support fragmentation */ )
 817     return clib_error_return (0, "too big path-mtu value, maximum is 1450");
 818
 819   if (path_mtu < 68)
 820     return clib_error_return (0, "too small path-mtu value, minimum is 68");
 821
 822   /* Calculate how much header data we need. */
 823   if (collector.version == AF_IP4)
 824     ip_header_size = sizeof (ip4_header_t);
 825   else
 826     ip_header_size = sizeof (ip6_header_t);
 827   exp->all_headers_size = ip_header_size + sizeof (udp_header_t) +
 828                           sizeof (ipfix_message_header_t) +
 829                           sizeof (ipfix_set_header_t);
 830
 831   /* Reset report streams if we are reconfiguring IP addresses */
 832   if (ip_address_cmp (&exp->ipfix_collector, &collector) ||
 833       ip_address_cmp (&exp->src_address, &src) ||
 834       exp->collector_port != collector_port)
 835     vnet_flow_reports_reset (exp);
 836
 837   exp->ipfix_collector = collector;
 838   exp->collector_port = collector_port;
 839   exp->src_address = src;
 840   exp->fib_index = fib_index;
 841   exp->path_mtu = path_mtu;
 842   exp->template_interval = template_interval;
 843   exp->udp_checksum = udp_checksum;
 844
 845   if (collector.ip.ip4.as_u32)
 846     vlib_cli_output (vm,
 847                      "Collector %U, src address %U, "
 848                      "fib index %d, path MTU %u, "
 849                      "template resend interval %us, "
 850                      "udp checksum %s",
 851                      format_ip4_address, &exp->ipfix_collector.ip.ip4,
 852                      format_ip4_address, &exp->src_address.ip.ip4, fib_index,
 853                      path_mtu, template_interval,
 854                      udp_checksum ? "enabled" : "disabled");
 855   else
 856     vlib_cli_output (vm, "IPFIX Collector is disabled");
 857
 858   /* Turn on the flow reporting process */
 859   vlib_process_signal_event (vm, flow_report_process_node.index, 1, 0);
 860   return 0;
 861 }
 862
 863 VLIB_CLI_COMMAND (set_ipfix_exporter_command, static) = {
 864     .path = "set ipfix exporter",
 865     .short_help = "set ipfix exporter "
 866                   "collector <ip4-address> [port <port>] "
 867                   "src <ip4-address> [fib-id <fib-id>] "
 868                   "[path-mtu <path-mtu>] "
 869                   "[template-interval <template-interval>] "
 870                   "[udp-checksum]",
 871     .function = set_ipfix_exporter_command_fn,
 872 };
 873
 874
 875 static clib_error_t *
 876 ipfix_flush_command_fn (vlib_main_t * vm,
 877                         unformat_input_t * input, vlib_cli_command_t * cmd)
 878 {
 879   /* poke the flow reporting process */
 880   vlib_process_signal_event (vm, flow_report_process_node.index, 1, 0);
 881   return 0;
 882 }
 883
 884 VLIB_CLI_COMMAND (ipfix_flush_command, static) = {
 885     .path = "ipfix flush",
 886     .short_help = "flush the current ipfix data [for make test]",
 887     .function = ipfix_flush_command_fn,
 888 };
 889
 890 static clib_error_t *
 891 flow_report_init (vlib_main_t * vm)
 892 {
 893   flow_report_main_t *frm = &flow_report_main;
 894   ipfix_exporter_t *exp;
 895
 896   frm->vlib_main = vm;
 897   frm->vnet_main = vnet_get_main ();
 898   frm->unix_time_0 = time (0);
 899   frm->vlib_time_0 = vlib_time_now (frm->vlib_main);
 900   /*
 901    * Make sure that we can always access the first exporter for
 902    * backwards compatibility reasons.
 903    */
 904   pool_alloc (frm->exporters, IPFIX_EXPORTERS_MAX);
 905   pool_get (frm->exporters, exp);
 906   /* Verify that this is at index 0 */
 907   ASSERT (frm->exporters == exp);
 908   exp->fib_index = ~0;
 909   return 0;
 910 }
 911
 912 VLIB_INIT_FUNCTION (flow_report_init);
 913 /*
 914  * fd.io coding-style-patch-verification: ON
 915  *
 916  * Local Variables:
 917  * eval: (c-set-style "gnu")
 918  * End:
 919  */