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