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