src/vnet/flow/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/flow/flow_report.h>
  19 #include <vnet/api_errno.h>
  20
  21 flow_report_main_t flow_report_main;
  22
  23 static_always_inline u8 stream_index_valid (u32 index)
  24 {
  25   flow_report_main_t * frm = &flow_report_main;
  26   return index < vec_len(frm->streams) &&
  27          frm->streams[index].domain_id != ~0;
  28 }
  29
  30 static_always_inline flow_report_stream_t * add_stream (void)
  31 {
  32   flow_report_main_t * frm = &flow_report_main;
  33   u32 i;
  34   for (i = 0; i < vec_len(frm->streams); i++)
  35     if (!stream_index_valid(i))
  36       return &frm->streams[i];
  37   u32 index = vec_len(frm->streams);
  38   vec_validate(frm->streams, index);
  39   return &frm->streams[index];
  40 }
  41
  42 static_always_inline void delete_stream (u32 index)
  43 {
  44   flow_report_main_t * frm = &flow_report_main;
  45   ASSERT (index < vec_len(frm->streams));
  46   ASSERT (frm->streams[index].domain_id != ~0);
  47   frm->streams[index].domain_id = ~0;
  48 }
  49
  50 static i32 find_stream (u32 domain_id, u16 src_port)
  51 {
  52   flow_report_main_t * frm = &flow_report_main;
  53   flow_report_stream_t * stream;
  54   u32 i;
  55   for (i = 0; i < vec_len(frm->streams); i++)
  56     if (stream_index_valid(i)) {
  57       stream = &frm->streams[i];
  58       if (domain_id == stream->domain_id) {
  59         if (src_port != stream->src_port)
  60           return -2;
  61         return i;
  62       } else if (src_port == stream->src_port) {
  63         return -2;
  64       }
  65     }
  66   return -1;
  67 }
  68
  69 int send_template_packet (flow_report_main_t *frm,
  70                           flow_report_t *fr,
  71                           u32 * buffer_indexp)
  72 {
  73   u32 bi0;
  74   vlib_buffer_t * b0;
  75   ip4_ipfix_template_packet_t * tp;
  76   ipfix_message_header_t * h;
  77   ip4_header_t * ip;
  78   udp_header_t * udp;
  79   vlib_main_t * vm = frm->vlib_main;
  80   flow_report_stream_t * stream;
  81   vlib_buffer_free_list_t *fl;
  82
  83   ASSERT (buffer_indexp);
  84
  85   if (fr->update_rewrite || fr->rewrite == 0)
  86     {
  87       if (frm->ipfix_collector.as_u32 == 0
  88           || frm->src_address.as_u32 == 0)
  89         {
  90           vlib_node_set_state (frm->vlib_main, flow_report_process_node.index,
  91                                VLIB_NODE_STATE_DISABLED);
  92           return -1;
  93         }
  94       vec_free (fr->rewrite);
  95       fr->update_rewrite = 1;
  96     }
  97
  98   if (fr->update_rewrite)
  99     {
 100       fr->rewrite = fr->rewrite_callback (frm, fr,
 101                                           &frm->ipfix_collector,
 102                                           &frm->src_address,
 103                                           frm->collector_port);
 104       fr->update_rewrite = 0;
 105     }
 106
 107   if (vlib_buffer_alloc (vm, &bi0, 1) != 1)
 108     return -1;
 109
 110   b0 = vlib_get_buffer (vm, bi0);
 111
 112   /* Initialize the buffer */
 113   fl = vlib_buffer_get_free_list (vm, VLIB_BUFFER_DEFAULT_FREE_LIST_INDEX);
 114   vlib_buffer_init_for_free_list (b0, fl);
 115   VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
 116
 117   ASSERT (vec_len (fr->rewrite) < VLIB_BUFFER_DEFAULT_FREE_LIST_BYTES);
 118
 119   clib_memcpy (b0->data, fr->rewrite, vec_len (fr->rewrite));
 120   b0->current_data = 0;
 121   b0->current_length = vec_len (fr->rewrite);
 122   b0->flags |= (VLIB_BUFFER_TOTAL_LENGTH_VALID | VLIB_BUFFER_FLOW_REPORT);
 123   vnet_buffer (b0)->sw_if_index[VLIB_RX] = 0;
 124   vnet_buffer (b0)->sw_if_index[VLIB_TX] = frm->fib_index;
 125
 126   tp = vlib_buffer_get_current (b0);
 127   ip = (ip4_header_t *) &tp->ip4;
 128   udp = (udp_header_t *) (ip+1);
 129   h = (ipfix_message_header_t *)(udp+1);
 130
 131   /* FIXUP: message header export_time */
 132   h->export_time = (u32)
 133     (((f64)frm->unix_time_0) +
 134      (vlib_time_now(frm->vlib_main) - frm->vlib_time_0));
 135   h->export_time = clib_host_to_net_u32(h->export_time);
 136
 137   stream = &frm->streams[fr->stream_index];
 138
 139   /* FIXUP: message header sequence_number. Templates do not increase it */
 140   h->sequence_number = clib_host_to_net_u32(stream->sequence_number);
 141
 142   /* FIXUP: udp length */
 143   udp->length = clib_host_to_net_u16 (b0->current_length - sizeof (*ip));
 144
 145   if (frm->udp_checksum)
 146     {
 147       /* RFC 7011 section 10.3.2. */
 148       udp->checksum = ip4_tcp_udp_compute_checksum (vm, b0, ip);
 149       if (udp->checksum == 0)
 150         udp->checksum = 0xffff;
 151     }
 152
 153   *buffer_indexp = bi0;
 154
 155   fr->last_template_sent = vlib_time_now (vm);
 156
 157   return 0;
 158 }
 159
 160 static uword
 161 flow_report_process (vlib_main_t * vm,
 162                      vlib_node_runtime_t * rt,
 163                      vlib_frame_t * f)
 164 {
 165   flow_report_main_t * frm = &flow_report_main;
 166   flow_report_t * fr;
 167   u32 ip4_lookup_node_index;
 168   vlib_node_t * ip4_lookup_node;
 169   vlib_frame_t * nf = 0;
 170   u32 template_bi;
 171   u32 * to_next;
 172   int send_template;
 173   f64 now;
 174   int rv;
 175   uword event_type;
 176   uword *event_data = 0;
 177
 178   /* Wait for Godot... */
 179   vlib_process_wait_for_event_or_clock (vm, 1e9);
 180   event_type = vlib_process_get_events (vm, &event_data);
 181   if (event_type != 1)
 182       clib_warning ("bogus kickoff event received, %d", event_type);
 183   vec_reset_length (event_data);
 184
 185   /* Enqueue pkts to ip4-lookup */
 186   ip4_lookup_node = vlib_get_node_by_name (vm, (u8 *) "ip4-lookup");
 187   ip4_lookup_node_index = ip4_lookup_node->index;
 188
 189   while (1)
 190     {
 191       vlib_process_wait_for_event_or_clock (vm, 5.0);
 192       event_type = vlib_process_get_events (vm, &event_data);
 193       vec_reset_length (event_data);
 194
 195       vec_foreach (fr, frm->reports)
 196         {
 197           now = vlib_time_now (vm);
 198
 199           /* Need to send a template packet? */
 200           send_template =
 201               now > (fr->last_template_sent + frm->template_interval);
 202           send_template += fr->last_template_sent == 0;
 203           template_bi = ~0;
 204           rv = 0;
 205
 206           if (send_template)
 207             rv = send_template_packet (frm, fr, &template_bi);
 208
 209           if (rv < 0)
 210             continue;
 211
 212           nf = vlib_get_frame_to_node (vm, ip4_lookup_node_index);
 213           nf->n_vectors = 0;
 214           to_next = vlib_frame_vector_args (nf);
 215
 216           if (template_bi != ~0)
 217             {
 218               to_next[0] = template_bi;
 219               to_next++;
 220               nf->n_vectors++;
 221             }
 222
 223           nf = fr->flow_data_callback (frm, fr,
 224                                        nf, to_next, ip4_lookup_node_index);
 225           if (nf)
 226             vlib_put_frame_to_node (vm, ip4_lookup_node_index, nf);
 227         }
 228     }
 229
 230   return 0; /* not so much */
 231 }
 232
 233 VLIB_REGISTER_NODE (flow_report_process_node) = {
 234     .function = flow_report_process,
 235     .type = VLIB_NODE_TYPE_PROCESS,
 236     .name = "flow-report-process",
 237 };
 238
 239 int vnet_flow_report_add_del (flow_report_main_t *frm,
 240                               vnet_flow_report_add_del_args_t *a,
 241                               u16 *template_id)
 242 {
 243   int i;
 244   int found_index = ~0;
 245   flow_report_t *fr;
 246   flow_report_stream_t * stream;
 247   u32 si;
 248
 249   si = find_stream(a->domain_id, a->src_port);
 250   if (si == -2)
 251     return VNET_API_ERROR_INVALID_VALUE;
 252   if (si == -1 && a->is_add == 0)
 253     return VNET_API_ERROR_NO_SUCH_ENTRY;
 254
 255   for (i = 0; i < vec_len(frm->reports); i++)
 256     {
 257       fr = vec_elt_at_index (frm->reports, i);
 258       if (fr->opaque.as_uword == a->opaque.as_uword
 259           && fr->rewrite_callback == a->rewrite_callback
 260           && fr->flow_data_callback == a->flow_data_callback)
 261         {
 262           found_index = i;
 263           if (template_id)
 264             *template_id = fr->template_id;
 265           break;
 266         }
 267     }
 268
 269   if (a->is_add == 0)
 270     {
 271       if (found_index != ~0)
 272         {
 273           vec_delete (frm->reports, 1, found_index);
 274           stream = &frm->streams[si];
 275           stream->n_reports--;
 276           if (stream->n_reports == 0)
 277             delete_stream(si);
 278           return 0;
 279         }
 280       return VNET_API_ERROR_NO_SUCH_ENTRY;
 281     }
 282
 283   if (found_index != ~0)
 284     return VNET_API_ERROR_VALUE_EXIST;
 285
 286   if (si == -1)
 287     {
 288       stream = add_stream();
 289       stream->domain_id = a->domain_id;
 290       stream->src_port = a->src_port;
 291       stream->sequence_number = 0;
 292       stream->n_reports = 0;
 293       si = stream - frm->streams;
 294     }
 295   else
 296     stream = &frm->streams[si];
 297
 298   stream->n_reports++;
 299
 300   vec_add2 (frm->reports, fr, 1);
 301
 302   fr->stream_index = si;
 303   fr->template_id = 256 + stream->next_template_no;
 304   stream->next_template_no = (stream->next_template_no + 1) % (65536 - 256);
 305   fr->update_rewrite = 1;
 306   fr->opaque = a->opaque;
 307   fr->rewrite_callback = a->rewrite_callback;
 308   fr->flow_data_callback = a->flow_data_callback;
 309
 310   if (template_id)
 311     *template_id = fr->template_id;
 312
 313   return 0;
 314 }
 315
 316 clib_error_t * flow_report_add_del_error_to_clib_error (int error)
 317 {
 318     switch (error)
 319       {
 320       case 0:
 321         return 0;
 322       case VNET_API_ERROR_NO_SUCH_ENTRY:
 323         return clib_error_return (0, "Flow report not found");
 324       case VNET_API_ERROR_VALUE_EXIST:
 325         return clib_error_return (0, "Flow report already exists");
 326       case VNET_API_ERROR_INVALID_VALUE:
 327         return clib_error_return (0, "Expecting either still unused values "
 328                                      "for both domain_id and src_port "
 329                                      "or already used values for both fields");
 330       default:
 331         return clib_error_return (0, "vnet_flow_report_add_del returned %d",
 332                                   error);
 333       }
 334 }
 335
 336 void vnet_flow_reports_reset (flow_report_main_t * frm)
 337 {
 338   flow_report_t *fr;
 339   u32 i;
 340
 341   for (i = 0; i < vec_len(frm->streams); i++)
 342     if (stream_index_valid(i))
 343       frm->streams[i].sequence_number = 0;
 344
 345   vec_foreach (fr, frm->reports)
 346     {
 347       fr->update_rewrite = 1;
 348       fr->last_template_sent = 0;
 349     }
 350 }
 351
 352 void vnet_stream_reset (flow_report_main_t * frm, u32 stream_index)
 353 {
 354   flow_report_t *fr;
 355
 356   frm->streams[stream_index].sequence_number = 0;
 357
 358   vec_foreach (fr, frm->reports)
 359     if (frm->reports->stream_index == stream_index) {
 360       fr->update_rewrite = 1;
 361       fr->last_template_sent = 0;
 362     }
 363 }
 364
 365 int vnet_stream_change (flow_report_main_t * frm,
 366                         u32 old_domain_id, u16 old_src_port,
 367                         u32 new_domain_id, u16 new_src_port)
 368 {
 369   i32 stream_index = find_stream (old_domain_id, old_src_port);
 370   if (stream_index < 0)
 371     return 1;
 372   flow_report_stream_t * stream = &frm->streams[stream_index];
 373   stream->domain_id = new_domain_id;
 374   stream->src_port = new_src_port;
 375   if (old_domain_id != new_domain_id || old_src_port != new_src_port)
 376     vnet_stream_reset (frm, stream_index);
 377   return 0;
 378 }
 379
 380 static clib_error_t *
 381 set_ipfix_exporter_command_fn (vlib_main_t * vm,
 382                                unformat_input_t * input,
 383                                vlib_cli_command_t * cmd)
 384 {
 385   flow_report_main_t * frm = &flow_report_main;
 386   ip4_address_t collector, src;
 387   u16 collector_port = UDP_DST_PORT_ipfix;
 388   u32 fib_id;
 389   u32 fib_index = ~0;
 390
 391   collector.as_u32 = 0;
 392   src.as_u32 = 0;
 393   u32 path_mtu = 512; // RFC 7011 section 10.3.3.
 394   u32 template_interval = 20;
 395   u8 udp_checksum = 0;
 396
 397   while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT) {
 398     if (unformat (input, "collector %U", unformat_ip4_address, &collector))
 399       ;
 400     else if (unformat (input, "port %u", &collector_port))
 401       ;
 402     else if (unformat (input, "src %U", unformat_ip4_address, &src))
 403       ;
 404     else if (unformat (input, "fib-id %u", &fib_id))
 405       {
 406         ip4_main_t * im = &ip4_main;
 407         uword * p = hash_get (im->fib_index_by_table_id, fib_id);
 408         if (! p)
 409           return clib_error_return (0, "fib ID %d doesn't exist\n",
 410                                     fib_id);
 411         fib_index = p[0];
 412       }
 413     else if (unformat (input, "path-mtu %u", &path_mtu))
 414       ;
 415     else if (unformat (input, "template-interval %u", &template_interval))
 416       ;
 417     else if (unformat (input, "udp-checksum"))
 418       udp_checksum = 1;
 419     else
 420       break;
 421   }
 422
 423   if (collector.as_u32 != 0 && src.as_u32 == 0)
 424     return clib_error_return (0, "src address required");
 425
 426   if (path_mtu > 1450 /* vpp does not support fragmentation */)
 427         return clib_error_return (0, "too big path-mtu value, maximum is 1450");
 428
 429   if (path_mtu < 68)
 430         return clib_error_return (0, "too small path-mtu value, minimum is 68");
 431
 432   /* Reset report streams if we are reconfiguring IP addresses */
 433   if (frm->ipfix_collector.as_u32 != collector.as_u32 ||
 434       frm->src_address.as_u32 != src.as_u32 ||
 435       frm->collector_port != collector_port)
 436     vnet_flow_reports_reset(frm);
 437
 438   frm->ipfix_collector.as_u32 = collector.as_u32;
 439   frm->collector_port = collector_port;
 440   frm->src_address.as_u32 = src.as_u32;
 441   frm->fib_index = fib_index;
 442   frm->path_mtu = path_mtu;
 443   frm->template_interval = template_interval;
 444   frm->udp_checksum = udp_checksum;
 445
 446   if (collector.as_u32)
 447     vlib_cli_output (vm, "Collector %U, src address %U, "
 448                            "fib index %d, path MTU %u, "
 449                            "template resend interval %us, "
 450                            "udp checksum %s",
 451                    format_ip4_address, &frm->ipfix_collector,
 452                    format_ip4_address, &frm->src_address,
 453                    fib_index, path_mtu, template_interval,
 454                    udp_checksum ? "enabled" : "disabled");
 455   else
 456     vlib_cli_output (vm, "IPFIX Collector is disabled");
 457
 458   /* Turn on the flow reporting process */
 459   vlib_process_signal_event (vm, flow_report_process_node.index,
 460                              1, 0);
 461   return 0;
 462 }
 463
 464 VLIB_CLI_COMMAND (set_ipfix_exporter_command, static) = {
 465     .path = "set ipfix exporter",
 466     .short_help = "set ipfix exporter "
 467                   "collector <ip4-address> [port <port>] "
 468                   "src <ip4-address> [fib-id <fib-id>] "
 469                   "[path-mtu <path-mtu>] "
 470                   "[template-interval <template-interval>]",
 471                   "[udp-checksum]",
 472     .function = set_ipfix_exporter_command_fn,
 473 };
 474
 475
 476 static clib_error_t *
 477 ipfix_flush_command_fn (vlib_main_t * vm,
 478                         unformat_input_t * input,
 479                         vlib_cli_command_t * cmd)
 480 {
 481   /* poke the flow reporting process */
 482   vlib_process_signal_event (vm, flow_report_process_node.index,
 483                              1, 0);
 484   return 0;
 485 }
 486
 487 VLIB_CLI_COMMAND (ipfix_flush_command, static) = {
 488     .path = "ipfix flush",
 489     .short_help = "flush the current ipfix data [for make test]",
 490     .function = ipfix_flush_command_fn,
 491 };
 492
 493 static clib_error_t *
 494 flow_report_init (vlib_main_t *vm)
 495 {
 496   flow_report_main_t * frm = &flow_report_main;
 497
 498   frm->vlib_main = vm;
 499   frm->vnet_main = vnet_get_main();
 500   frm->unix_time_0 = time(0);
 501   frm->vlib_time_0 = vlib_time_now(frm->vlib_main);
 502   frm->fib_index = ~0;
 503
 504   return 0;
 505 }
 506
 507 VLIB_INIT_FUNCTION (flow_report_init)