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