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