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 <vnet/ipfix-export/flow_report.h>
  19 #include <vnet/api_errno.h>
  20 #include <vnet/udp/udp.h>
  21
  22 flow_report_main_t flow_report_main;
  23
  24 static_always_inline u8
  25 stream_index_valid (ipfix_exporter_t *exp, u32 index)
  26 {
  27   return index < vec_len (exp->streams) && exp->streams[index].domain_id != ~0;
  28 }
  29
  30 static_always_inline flow_report_stream_t *
  31 add_stream (ipfix_exporter_t *exp)
  32 {
  33   u32 i;
  34   for (i = 0; i < vec_len (exp->streams); i++)
  35     if (!stream_index_valid (exp, i))
  36       return &exp->streams[i];
  37   u32 index = vec_len (exp->streams);
  38   vec_validate (exp->streams, index);
  39   return &exp->streams[index];
  40 }
  41
  42 static_always_inline void
  43 delete_stream (ipfix_exporter_t *exp, u32 index)
  44 {
  45   ASSERT (index < vec_len (exp->streams));
  46   ASSERT (exp->streams[index].domain_id != ~0);
  47   exp->streams[index].domain_id = ~0;
  48 }
  49
  50 static i32
  51 find_stream (ipfix_exporter_t *exp, u32 domain_id, u16 src_port)
  52 {
  53   flow_report_stream_t *stream;
  54   u32 i;
  55   for (i = 0; i < vec_len (exp->streams); i++)
  56     if (stream_index_valid (exp, i))
  57       {
  58         stream = &exp->streams[i];
  59         if (domain_id == stream->domain_id)
  60           {
  61             if (src_port != stream->src_port)
  62               return -2;
  63             return i;
  64           }
  65         else if (src_port == stream->src_port)
  66           {
  67             return -2;
  68           }
  69       }
  70   return -1;
  71 }
  72
  73 int
  74 send_template_packet (flow_report_main_t *frm, ipfix_exporter_t *exp,
  75                       flow_report_t *fr, u32 *buffer_indexp)
  76 {
  77   u32 bi0;
  78   vlib_buffer_t *b0;
  79   ip4_ipfix_template_packet_t *tp;
  80   ipfix_message_header_t *h;
  81   ip4_header_t *ip;
  82   udp_header_t *udp;
  83   vlib_main_t *vm = frm->vlib_main;
  84   flow_report_stream_t *stream;
  85
  86   ASSERT (buffer_indexp);
  87
  88   if (fr->update_rewrite || fr->rewrite == 0)
  89     {
  90       if (exp->ipfix_collector.as_u32 == 0 || exp->src_address.as_u32 == 0)
  91         {
  92           vlib_node_set_state (frm->vlib_main, flow_report_process_node.index,
  93                                VLIB_NODE_STATE_DISABLED);
  94           return -1;
  95         }
  96       vec_free (fr->rewrite);
  97       fr->update_rewrite = 1;
  98     }
  99
 100   if (fr->update_rewrite)
 101     {
 102       fr->rewrite = fr->rewrite_callback (
 103         exp, fr, &exp->ipfix_collector, &exp->src_address, exp->collector_port,
 104         fr->report_elements, fr->n_report_elements, fr->stream_indexp);
 105       fr->update_rewrite = 0;
 106     }
 107
 108   if (vlib_buffer_alloc (vm, &bi0, 1) != 1)
 109     return -1;
 110
 111   b0 = vlib_get_buffer (vm, bi0);
 112
 113   ASSERT (vec_len (fr->rewrite) < vlib_buffer_get_default_data_size (vm));
 114
 115   clib_memcpy_fast (b0->data, fr->rewrite, vec_len (fr->rewrite));
 116   b0->current_data = 0;
 117   b0->current_length = vec_len (fr->rewrite);
 118   b0->flags |= (VLIB_BUFFER_TOTAL_LENGTH_VALID | VNET_BUFFER_F_FLOW_REPORT);
 119   vnet_buffer (b0)->sw_if_index[VLIB_RX] = 0;
 120   vnet_buffer (b0)->sw_if_index[VLIB_TX] = exp->fib_index;
 121
 122   tp = vlib_buffer_get_current (b0);
 123   ip = (ip4_header_t *) & tp->ip4;
 124   udp = (udp_header_t *) (ip + 1);
 125   h = (ipfix_message_header_t *) (udp + 1);
 126
 127   /* FIXUP: message header export_time */
 128   h->export_time = (u32)
 129     (((f64) frm->unix_time_0) +
 130      (vlib_time_now (frm->vlib_main) - frm->vlib_time_0));
 131   h->export_time = clib_host_to_net_u32 (h->export_time);
 132
 133   stream = &exp->streams[fr->stream_index];
 134
 135   /* FIXUP: message header sequence_number. Templates do not increase it */
 136   h->sequence_number = clib_host_to_net_u32 (stream->sequence_number);
 137
 138   /* FIXUP: udp length */
 139   udp->length = clib_host_to_net_u16 (b0->current_length - sizeof (*ip));
 140
 141   if (exp->udp_checksum)
 142     {
 143       /* RFC 7011 section 10.3.2. */
 144       udp->checksum = ip4_tcp_udp_compute_checksum (vm, b0, ip);
 145       if (udp->checksum == 0)
 146         udp->checksum = 0xffff;
 147     }
 148
 149   *buffer_indexp = bi0;
 150
 151   fr->last_template_sent = vlib_time_now (vm);
 152
 153   return 0;
 154 }
 155
 156 u8 *
 157 vnet_flow_rewrite_generic_callback (ipfix_exporter_t *exp, flow_report_t *fr,
 158                                     ip4_address_t *collector_address,
 159                                     ip4_address_t *src_address,
 160                                     u16 collector_port,
 161                                     ipfix_report_element_t *report_elts,
 162                                     u32 n_elts, u32 *stream_indexp)
 163 {
 164   ip4_header_t *ip;
 165   udp_header_t *udp;
 166   ipfix_message_header_t *h;
 167   ipfix_set_header_t *s;
 168   ipfix_template_header_t *t;
 169   ipfix_field_specifier_t *f;
 170   ipfix_field_specifier_t *first_field;
 171   u8 *rewrite = 0;
 172   ip4_ipfix_template_packet_t *tp;
 173   flow_report_stream_t *stream;
 174   int i;
 175   ipfix_report_element_t *ep;
 176
 177   ASSERT (stream_indexp);
 178   ASSERT (n_elts);
 179   ASSERT (report_elts);
 180
 181   stream = &exp->streams[fr->stream_index];
 182   *stream_indexp = fr->stream_index;
 183
 184   /* allocate rewrite space */
 185   vec_validate_aligned (rewrite,
 186                         sizeof (ip4_ipfix_template_packet_t)
 187                         + n_elts * sizeof (ipfix_field_specifier_t) - 1,
 188                         CLIB_CACHE_LINE_BYTES);
 189
 190   /* create the packet rewrite string */
 191   tp = (ip4_ipfix_template_packet_t *) rewrite;
 192   ip = (ip4_header_t *) & tp->ip4;
 193   udp = (udp_header_t *) (ip + 1);
 194   h = (ipfix_message_header_t *) (udp + 1);
 195   s = (ipfix_set_header_t *) (h + 1);
 196   t = (ipfix_template_header_t *) (s + 1);
 197   first_field = f = (ipfix_field_specifier_t *) (t + 1);
 198
 199   ip->ip_version_and_header_length = 0x45;
 200   ip->ttl = 254;
 201   ip->protocol = IP_PROTOCOL_UDP;
 202   ip->src_address.as_u32 = src_address->as_u32;
 203   ip->dst_address.as_u32 = collector_address->as_u32;
 204   udp->src_port = clib_host_to_net_u16 (stream->src_port);
 205   udp->dst_port = clib_host_to_net_u16 (collector_port);
 206   udp->length = clib_host_to_net_u16 (vec_len (rewrite) - sizeof (*ip));
 207
 208   /* FIXUP LATER: message header export_time */
 209   h->domain_id = clib_host_to_net_u32 (stream->domain_id);
 210
 211   ep = report_elts;
 212
 213   for (i = 0; i < n_elts; i++)
 214     {
 215       f->e_id_length = ipfix_e_id_length (0, ep->info_element, ep->size);
 216       f++;
 217       ep++;
 218     }
 219
 220   /* Back to the template packet... */
 221   ip = (ip4_header_t *) & tp->ip4;
 222   udp = (udp_header_t *) (ip + 1);
 223
 224   ASSERT (f - first_field);
 225   /* Field count in this template */
 226   t->id_count = ipfix_id_count (fr->template_id, f - first_field);
 227
 228   /* set length in octets */
 229   s->set_id_length =
 230     ipfix_set_id_length (2 /* set_id */ , (u8 *) f - (u8 *) s);
 231
 232   /* message length in octets */
 233   h->version_length = version_length ((u8 *) f - (u8 *) h);
 234
 235   ip->length = clib_host_to_net_u16 ((u8 *) f - (u8 *) ip);
 236   ip->checksum = ip4_header_checksum (ip);
 237
 238   return rewrite;
 239 }
 240
 241 static uword
 242 flow_report_process (vlib_main_t * vm,
 243                      vlib_node_runtime_t * rt, vlib_frame_t * f)
 244 {
 245   flow_report_main_t *frm = &flow_report_main;
 246   flow_report_t *fr;
 247   u32 ip4_lookup_node_index;
 248   vlib_node_t *ip4_lookup_node;
 249   vlib_frame_t *nf = 0;
 250   u32 template_bi;
 251   u32 *to_next;
 252   int send_template;
 253   f64 now, wait_time;
 254   f64 def_wait_time = 5.0;
 255   int rv;
 256   uword event_type;
 257   uword *event_data = 0;
 258
 259   /* Wait for Godot... */
 260   vlib_process_wait_for_event_or_clock (vm, 1e9);
 261   event_type = vlib_process_get_events (vm, &event_data);
 262   if (event_type != 1)
 263     clib_warning ("bogus kickoff event received, %d", event_type);
 264   vec_reset_length (event_data);
 265
 266   /* Enqueue pkts to ip4-lookup */
 267   ip4_lookup_node = vlib_get_node_by_name (vm, (u8 *) "ip4-lookup");
 268   ip4_lookup_node_index = ip4_lookup_node->index;
 269
 270   wait_time = def_wait_time;
 271
 272   while (1)
 273     {
 274       vlib_process_wait_for_event_or_clock (vm, wait_time);
 275       event_type = vlib_process_get_events (vm, &event_data);
 276       vec_reset_length (event_data);
 277       ipfix_exporter_t *exp;
 278       pool_foreach (exp, frm->exporters)
 279         {
 280
 281           /* 5s delay by default, possibly reduced by template intervals */
 282           wait_time = def_wait_time;
 283
 284           vec_foreach (fr, exp->reports)
 285             {
 286               f64 next_template;
 287               now = vlib_time_now (vm);
 288
 289               /* Need to send a template packet? */
 290               send_template =
 291                 now > (fr->last_template_sent + exp->template_interval);
 292               send_template += fr->last_template_sent == 0;
 293               template_bi = ~0;
 294               rv = 0;
 295
 296               if (send_template)
 297                 rv = send_template_packet (frm, exp, fr, &template_bi);
 298
 299               if (rv < 0)
 300                 continue;
 301
 302               /*
 303                * decide if template should be sent sooner than current wait
 304                * time
 305                */
 306               next_template =
 307                 (fr->last_template_sent + exp->template_interval) - now;
 308               wait_time = clib_min (wait_time, next_template);
 309
 310               nf = vlib_get_frame_to_node (vm, ip4_lookup_node_index);
 311               nf->n_vectors = 0;
 312               to_next = vlib_frame_vector_args (nf);
 313
 314               if (template_bi != ~0)
 315                 {
 316                   to_next[0] = template_bi;
 317                   to_next++;
 318                   nf->n_vectors++;
 319                 }
 320
 321               nf = fr->flow_data_callback (frm, exp, fr, nf, to_next,
 322                                            ip4_lookup_node_index);
 323               if (nf)
 324                 vlib_put_frame_to_node (vm, ip4_lookup_node_index, nf);
 325             }
 326         }
 327     }
 328
 329   return 0;                     /* not so much */
 330 }
 331
 332 /* *INDENT-OFF* */
 333 VLIB_REGISTER_NODE (flow_report_process_node) = {
 334     .function = flow_report_process,
 335     .type = VLIB_NODE_TYPE_PROCESS,
 336     .name = "flow-report-process",
 337 };
 338 /* *INDENT-ON* */
 339
 340 int
 341 vnet_flow_report_add_del (ipfix_exporter_t *exp,
 342                           vnet_flow_report_add_del_args_t *a, u16 *template_id)
 343 {
 344   int i;
 345   int found_index = ~0;
 346   flow_report_t *fr;
 347   flow_report_stream_t *stream;
 348   u32 si;
 349
 350   si = find_stream (exp, a->domain_id, a->src_port);
 351   if (si == -2)
 352     return VNET_API_ERROR_INVALID_VALUE;
 353   if (si == -1 && a->is_add == 0)
 354     return VNET_API_ERROR_NO_SUCH_ENTRY;
 355
 356   for (i = 0; i < vec_len (exp->reports); i++)
 357     {
 358       fr = vec_elt_at_index (exp->reports, i);
 359       if (fr->opaque.as_uword == a->opaque.as_uword
 360           && fr->rewrite_callback == a->rewrite_callback
 361           && fr->flow_data_callback == a->flow_data_callback)
 362         {
 363           found_index = i;
 364           if (template_id)
 365             *template_id = fr->template_id;
 366           break;
 367         }
 368     }
 369
 370   if (a->is_add == 0)
 371     {
 372       if (found_index != ~0)
 373         {
 374           vec_delete (exp->reports, 1, found_index);
 375           stream = &exp->streams[si];
 376           stream->n_reports--;
 377           if (stream->n_reports == 0)
 378             delete_stream (exp, si);
 379           return 0;
 380         }
 381       return VNET_API_ERROR_NO_SUCH_ENTRY;
 382     }
 383
 384   if (found_index != ~0)
 385     return VNET_API_ERROR_VALUE_EXIST;
 386
 387   if (si == -1)
 388     {
 389       stream = add_stream (exp);
 390       stream->domain_id = a->domain_id;
 391       stream->src_port = a->src_port;
 392       stream->sequence_number = 0;
 393       stream->n_reports = 0;
 394       si = stream - exp->streams;
 395     }
 396   else
 397     stream = &exp->streams[si];
 398
 399   stream->n_reports++;
 400
 401   vec_add2 (exp->reports, fr, 1);
 402
 403   fr->stream_index = si;
 404   fr->template_id = 256 + stream->next_template_no;
 405   stream->next_template_no = (stream->next_template_no + 1) % (65536 - 256);
 406   fr->update_rewrite = 1;
 407   fr->opaque = a->opaque;
 408   fr->rewrite_callback = a->rewrite_callback;
 409   fr->flow_data_callback = a->flow_data_callback;
 410   fr->report_elements = a->report_elements;
 411   fr->n_report_elements = a->n_report_elements;
 412   fr->stream_indexp = a->stream_indexp;
 413   if (template_id)
 414     *template_id = fr->template_id;
 415
 416   return 0;
 417 }
 418
 419 clib_error_t *
 420 flow_report_add_del_error_to_clib_error (int error)
 421 {
 422   switch (error)
 423     {
 424     case 0:
 425       return 0;
 426     case VNET_API_ERROR_NO_SUCH_ENTRY:
 427       return clib_error_return (0, "Flow report not found");
 428     case VNET_API_ERROR_VALUE_EXIST:
 429       return clib_error_return (0, "Flow report already exists");
 430     case VNET_API_ERROR_INVALID_VALUE:
 431       return clib_error_return (0, "Expecting either still unused values "
 432                                 "for both domain_id and src_port "
 433                                 "or already used values for both fields");
 434     default:
 435       return clib_error_return (0, "vnet_flow_report_add_del returned %d",
 436                                 error);
 437     }
 438 }
 439
 440 void
 441 vnet_flow_reports_reset (ipfix_exporter_t *exp)
 442 {
 443   flow_report_t *fr;
 444   u32 i;
 445
 446   for (i = 0; i < vec_len (exp->streams); i++)
 447     if (stream_index_valid (exp, i))
 448       exp->streams[i].sequence_number = 0;
 449
 450   vec_foreach (fr, exp->reports)
 451     {
 452       fr->update_rewrite = 1;
 453       fr->last_template_sent = 0;
 454     }
 455 }
 456
 457 void
 458 vnet_stream_reset (ipfix_exporter_t *exp, u32 stream_index)
 459 {
 460   flow_report_t *fr;
 461
 462   exp->streams[stream_index].sequence_number = 0;
 463
 464   vec_foreach (fr, exp->reports)
 465     if (exp->reports->stream_index == stream_index)
 466       {
 467         fr->update_rewrite = 1;
 468         fr->last_template_sent = 0;
 469       }
 470 }
 471
 472 int
 473 vnet_stream_change (ipfix_exporter_t *exp, u32 old_domain_id, u16 old_src_port,
 474                     u32 new_domain_id, u16 new_src_port)
 475 {
 476   i32 stream_index = find_stream (exp, old_domain_id, old_src_port);
 477
 478   if (stream_index < 0)
 479     return 1;
 480   flow_report_stream_t *stream = &exp->streams[stream_index];
 481   stream->domain_id = new_domain_id;
 482   stream->src_port = new_src_port;
 483   if (old_domain_id != new_domain_id || old_src_port != new_src_port)
 484     vnet_stream_reset (exp, stream_index);
 485   return 0;
 486 }
 487
 488 static clib_error_t *
 489 set_ipfix_exporter_command_fn (vlib_main_t * vm,
 490                                unformat_input_t * input,
 491                                vlib_cli_command_t * cmd)
 492 {
 493   flow_report_main_t *frm = &flow_report_main;
 494   ip4_address_t collector, src;
 495   u16 collector_port = UDP_DST_PORT_ipfix;
 496   u32 fib_id;
 497   u32 fib_index = ~0;
 498
 499   collector.as_u32 = 0;
 500   src.as_u32 = 0;
 501   u32 path_mtu = 512;           // RFC 7011 section 10.3.3.
 502   u32 template_interval = 20;
 503   u8 udp_checksum = 0;
 504   ipfix_exporter_t *exp = pool_elt_at_index (frm->exporters, 0);
 505
 506   while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
 507     {
 508       if (unformat (input, "collector %U", unformat_ip4_address, &collector))
 509         ;
 510       else if (unformat (input, "port %U", unformat_udp_port,
 511                          &collector_port))
 512         ;
 513       else if (unformat (input, "src %U", unformat_ip4_address, &src))
 514         ;
 515       else if (unformat (input, "fib-id %u", &fib_id))
 516         {
 517           ip4_main_t *im = &ip4_main;
 518           uword *p = hash_get (im->fib_index_by_table_id, fib_id);
 519           if (!p)
 520             return clib_error_return (0, "fib ID %d doesn't exist\n", fib_id);
 521           fib_index = p[0];
 522         }
 523       else if (unformat (input, "path-mtu %u", &path_mtu))
 524         ;
 525       else if (unformat (input, "template-interval %u", &template_interval))
 526         ;
 527       else if (unformat (input, "udp-checksum"))
 528         udp_checksum = 1;
 529       else
 530         break;
 531     }
 532
 533   if (collector.as_u32 != 0 && src.as_u32 == 0)
 534     return clib_error_return (0, "src address required");
 535
 536   if (path_mtu > 1450 /* vpp does not support fragmentation */ )
 537     return clib_error_return (0, "too big path-mtu value, maximum is 1450");
 538
 539   if (path_mtu < 68)
 540     return clib_error_return (0, "too small path-mtu value, minimum is 68");
 541
 542   /* Reset report streams if we are reconfiguring IP addresses */
 543   if (exp->ipfix_collector.as_u32 != collector.as_u32 ||
 544       exp->src_address.as_u32 != src.as_u32 ||
 545       exp->collector_port != collector_port)
 546     vnet_flow_reports_reset (exp);
 547
 548   exp->ipfix_collector.as_u32 = collector.as_u32;
 549   exp->collector_port = collector_port;
 550   exp->src_address.as_u32 = src.as_u32;
 551   exp->fib_index = fib_index;
 552   exp->path_mtu = path_mtu;
 553   exp->template_interval = template_interval;
 554   exp->udp_checksum = udp_checksum;
 555
 556   if (collector.as_u32)
 557     vlib_cli_output (vm,
 558                      "Collector %U, src address %U, "
 559                      "fib index %d, path MTU %u, "
 560                      "template resend interval %us, "
 561                      "udp checksum %s",
 562                      format_ip4_address, exp->ipfix_collector,
 563                      format_ip4_address, exp->src_address, fib_index, path_mtu,
 564                      template_interval, udp_checksum ? "enabled" : "disabled");
 565   else
 566     vlib_cli_output (vm, "IPFIX Collector is disabled");
 567
 568   /* Turn on the flow reporting process */
 569   vlib_process_signal_event (vm, flow_report_process_node.index, 1, 0);
 570   return 0;
 571 }
 572
 573 /* *INDENT-OFF* */
 574 VLIB_CLI_COMMAND (set_ipfix_exporter_command, static) = {
 575     .path = "set ipfix exporter",
 576     .short_help = "set ipfix exporter "
 577                   "collector <ip4-address> [port <port>] "
 578                   "src <ip4-address> [fib-id <fib-id>] "
 579                   "[path-mtu <path-mtu>] "
 580                   "[template-interval <template-interval>] "
 581                   "[udp-checksum]",
 582     .function = set_ipfix_exporter_command_fn,
 583 };
 584 /* *INDENT-ON* */
 585
 586
 587 static clib_error_t *
 588 ipfix_flush_command_fn (vlib_main_t * vm,
 589                         unformat_input_t * input, vlib_cli_command_t * cmd)
 590 {
 591   /* poke the flow reporting process */
 592   vlib_process_signal_event (vm, flow_report_process_node.index, 1, 0);
 593   return 0;
 594 }
 595
 596 /* *INDENT-OFF* */
 597 VLIB_CLI_COMMAND (ipfix_flush_command, static) = {
 598     .path = "ipfix flush",
 599     .short_help = "flush the current ipfix data [for make test]",
 600     .function = ipfix_flush_command_fn,
 601 };
 602 /* *INDENT-ON* */
 603
 604 static clib_error_t *
 605 flow_report_init (vlib_main_t * vm)
 606 {
 607   flow_report_main_t *frm = &flow_report_main;
 608   ipfix_exporter_t *exp;
 609
 610   frm->vlib_main = vm;
 611   frm->vnet_main = vnet_get_main ();
 612   frm->unix_time_0 = time (0);
 613   frm->vlib_time_0 = vlib_time_now (frm->vlib_main);
 614   /*
 615    * Make sure that we can always access the first exporter for
 616    * backwards compatibility reasons.
 617    */
 618   pool_alloc (frm->exporters, IPFIX_EXPORTERS_MAX);
 619   pool_get (frm->exporters, exp);
 620   /* Verify that this is at index 0 */
 621   ASSERT (frm->exporters == exp);
 622   exp->fib_index = ~0;
 623   return 0;
 624 }
 625
 626 VLIB_INIT_FUNCTION (flow_report_init);
 627 /*
 628  * fd.io coding-style-patch-verification: ON
 629  *
 630  * Local Variables:
 631  * eval: (c-set-style "gnu")
 632  * End:
 633  */