VPP-204 Rework and finish IPFIX implementation

Rework flow report registration system - add streams
Add support for IPv6 and src and dst ports for TCP and UDP protocols
Implement binary API for IPFIX classifier module

Change-Id: Id05cc0127a7b95ceaeebf9c79a32c6936449bd63
Signed-off-by: Juraj Sloboda <jsloboda@cisco.com>

4 files changed, 470 insertions, 96 deletions

diff --git a/vnet/vnet/flow/flow_report.c b/vnet/vnet/flow/flow_report.c
index 38ad6136..932613d3 100644
--- a/vnet/vnet/flow/flow_report.c
+++ b/vnet/vnet/flow/flow_report.c
@@ -20,6 +20,52 @@
 
 flow_report_main_t flow_report_main;
 
+static_always_inline u8 stream_index_valid (u32 index)
+{
+  flow_report_main_t * frm = &flow_report_main;
+  return index < vec_len(frm->streams) &&
+         frm->streams[index].domain_id != ~0;
+}
+
+static_always_inline flow_report_stream_t * add_stream (void)
+{
+  flow_report_main_t * frm = &flow_report_main;
+  u32 i;
+  for (i = 0; i < vec_len(frm->streams); i++)
+    if (!stream_index_valid(i))
+      return &frm->streams[i];
+  u32 index = vec_len(frm->streams);
+  vec_validate(frm->streams, index);
+  return &frm->streams[index];
+}
+
+static_always_inline void delete_stream (u32 index)
+{
+  flow_report_main_t * frm = &flow_report_main;
+  ASSERT (index < vec_len(frm->streams));
+  ASSERT (frm->streams[index].domain_id != ~0);
+  frm->streams[index].domain_id = ~0;
+}
+
+static i32 find_stream (u32 domain_id, u16 src_port)
+{
+  flow_report_main_t * frm = &flow_report_main;
+  flow_report_stream_t * stream;
+  u32 i;
+  for (i = 0; i < vec_len(frm->streams); i++)
+    if (stream_index_valid(i)) {
+      stream = &frm->streams[i];
+      if (domain_id == stream->domain_id) {
+        if (src_port != stream->src_port)
+          return -2;
+        return i;
+      } else if (src_port == stream->src_port) {
+        return -2;
+      }
+    }
+  return -1;
+}
+
 int send_template_packet (flow_report_main_t *frm, 
                           flow_report_t *fr,
                           u32 * buffer_indexp)
@@ -31,6 +77,7 @@ int send_template_packet (flow_report_main_t *frm,
   ip4_header_t * ip;
   udp_header_t * udp;
   vlib_main_t * vm = frm->vlib_main;
+  flow_report_stream_t * stream;
 
   ASSERT (buffer_indexp);
 
@@ -82,12 +129,22 @@ int send_template_packet (flow_report_main_t *frm,
      (vlib_time_now(frm->vlib_main) - frm->vlib_time_0));
   h->export_time = clib_host_to_net_u32(h->export_time);
 
+  stream = &frm->streams[fr->stream_index];
+
   /* FIXUP: message header sequence_number. Templates do not increase it */
-  h->sequence_number = clib_host_to_net_u32(fr->sequence_number);
+  h->sequence_number = clib_host_to_net_u32(stream->sequence_number);
 
   /* FIXUP: udp length */
   udp->length = clib_host_to_net_u16 (b0->current_length - sizeof (*ip));
 
+  if (frm->udp_checksum)
+    {
+      /* RFC 7011 section 10.3.2. */
+      udp->checksum = ip4_tcp_udp_compute_checksum (vm, b0, ip);
+      if (udp->checksum == 0)
+        udp->checksum = 0xffff;
+    }
+
   *buffer_indexp = bi0;
 
   fr->last_template_sent = vlib_time_now (vm);
@@ -178,11 +235,19 @@ int vnet_flow_report_add_del (flow_report_main_t *frm,
   int i;
   int found_index = ~0;
   flow_report_t *fr;
+  flow_report_stream_t * stream;
+  u32 si;
   
+  si = find_stream(a->domain_id, a->src_port);
+  if (si == -2)
+    return VNET_API_ERROR_INVALID_VALUE;
+  if (si == -1 && a->is_add == 0)
+    return VNET_API_ERROR_NO_SUCH_ENTRY;
+
   for (i = 0; i < vec_len(frm->reports); i++)
     {
       fr = vec_elt_at_index (frm->reports, i);
-      if (fr->opaque == a->opaque
+      if (fr->opaque.as_uword == a->opaque.as_uword
           && fr->rewrite_callback == a->rewrite_callback
           && fr->flow_data_callback == a->flow_data_callback)
         {
@@ -196,16 +261,37 @@ int vnet_flow_report_add_del (flow_report_main_t *frm,
       if (found_index != ~0)
         {
           vec_delete (frm->reports, 1, found_index);
+          stream = &frm->streams[si];
+          stream->n_reports--;
+          if (stream->n_reports == 0)
+            delete_stream(si);
           return 0;
         }
       return VNET_API_ERROR_NO_SUCH_ENTRY;
     }
 
+  if (found_index != ~0)
+    return VNET_API_ERROR_VALUE_EXIST;
+
+  if (si == -1)
+    {
+      stream = add_stream();
+      stream->domain_id = a->domain_id;
+      stream->src_port = a->src_port;
+      stream->sequence_number = 0;
+      stream->n_reports = 0;
+      si = stream - frm->streams;
+    }
+  else
+    stream = &frm->streams[si];
+
+  stream->n_reports++;
+
   vec_add2 (frm->reports, fr, 1);
 
-  fr->sequence_number = 0;
-  fr->domain_id = a->domain_id;
-  fr->src_port = a->src_port;
+  fr->stream_index = si;
+  fr->template_id = 256 + stream->next_template_no;
+  stream->next_template_no = (stream->next_template_no + 1) % (65536 - 256);
   fr->update_rewrite = 1;
   fr->opaque = a->opaque;
   fr->rewrite_callback = a->rewrite_callback;
@@ -217,18 +303,51 @@ int vnet_flow_report_add_del (flow_report_main_t *frm,
 void vnet_flow_reports_reset (flow_report_main_t * frm)
 {
   flow_report_t *fr;
+  u32 i;
+
+  for (i = 0; i < vec_len(frm->streams); i++)
+    if (stream_index_valid(i))
+      frm->streams[i].sequence_number = 0;
+
   vec_foreach (fr, frm->reports)
     {
-	  fr->sequence_number = 0;
       fr->update_rewrite = 1;
       fr->last_template_sent = 0;
     }
 }
 
+void vnet_stream_reset (flow_report_main_t * frm, u32 stream_index)
+{
+  flow_report_t *fr;
+
+  frm->streams[stream_index].sequence_number = 0;
+
+  vec_foreach (fr, frm->reports)
+    if (frm->reports->stream_index == stream_index) {
+      fr->update_rewrite = 1;
+      fr->last_template_sent = 0;
+    }
+}
+
+int vnet_stream_change (flow_report_main_t * frm,
+                        u32 old_domain_id, u16 old_src_port,
+                        u32 new_domain_id, u16 new_src_port)
+{
+  i32 stream_index = find_stream (old_domain_id, old_src_port);
+  if (stream_index < 0)
+    return 1;
+  flow_report_stream_t * stream = &frm->streams[stream_index];
+  stream->domain_id = new_domain_id;
+  stream->src_port = new_src_port;
+  if (old_domain_id != new_domain_id || old_src_port != new_src_port)
+    vnet_stream_reset (frm, stream_index);
+  return 0;
+}
+
 static clib_error_t *
-set_ipfix_command_fn (vlib_main_t * vm,
-		 unformat_input_t * input,
-		 vlib_cli_command_t * cmd)
+set_ipfix_exporter_command_fn (vlib_main_t * vm,
+                               unformat_input_t * input,
+                               vlib_cli_command_t * cmd)
 {
   flow_report_main_t * frm = &flow_report_main;
   ip4_address_t collector, src;
@@ -240,6 +359,7 @@ set_ipfix_command_fn (vlib_main_t * vm,
   src.as_u32 = 0;
   u32 path_mtu = 512; // RFC 7011 section 10.3.3.
   u32 template_interval = 20;
+  u8 udp_checksum = 0;
 
   while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT) {
     if (unformat (input, "collector %U", unformat_ip4_address, &collector))
@@ -261,6 +381,8 @@ set_ipfix_command_fn (vlib_main_t * vm,
       ;
     else if (unformat (input, "template-interval %u", &template_interval))
       ;
+    else if (unformat (input, "udp-checksum"))
+      udp_checksum = 1;
     else
       break;
   }
@@ -289,13 +411,16 @@ set_ipfix_command_fn (vlib_main_t * vm,
   frm->fib_index = fib_index;
   frm->path_mtu = path_mtu;
   frm->template_interval = template_interval;
+  frm->udp_checksum = udp_checksum;
   
   vlib_cli_output (vm, "Collector %U, src address %U, "
 		           "fib index %d, path MTU %u, "
-		           "template resend interval %us",
+		           "template resend interval %us, "
+		           "udp checksum %s",
                    format_ip4_address, &frm->ipfix_collector,
                    format_ip4_address, &frm->src_address,
-                   fib_index, path_mtu, template_interval);
+                   fib_index, path_mtu, template_interval,
+                   udp_checksum ? "enabled" : "disabled");
 
   /* Turn on the flow reporting process */
   vlib_process_signal_event (vm, flow_report_process_node.index,
@@ -303,14 +428,15 @@ set_ipfix_command_fn (vlib_main_t * vm,
   return 0;
 }
 
-VLIB_CLI_COMMAND (set_ipfix_command, static) = {
-    .path = "set ipfix",
-    .short_help = "set ipfix collector <ip4-address> "
-                  "[port <port>] "
+VLIB_CLI_COMMAND (set_ipfix_exporter_command, static) = {
+    .path = "set ipfix exporter",
+    .short_help = "set ipfix exporter "
+                  "collector <ip4-address> [port <port>] "
                   "src <ip4-address> [fib-id <fib-id>] "
                   "[path-mtu <path-mtu>] "
                   "[template-interval <template-interval>]",
-    .function = set_ipfix_command_fn,
+                  "[udp-checksum]",
+    .function = set_ipfix_exporter_command_fn,
 };
 
 static clib_error_t * 
@@ -322,6 +448,7 @@ flow_report_init (vlib_main_t *vm)
   frm->vnet_main = vnet_get_main();
   frm->unix_time_0 = time(0);
   frm->vlib_time_0 = vlib_time_now(frm->vlib_main);
+  frm->fib_index = ~0;
 
   return 0;
 }
diff --git a/vnet/vnet/flow/flow_report.h b/vnet/vnet/flow/flow_report.h
index 27aa81ae..7b9fc7b0 100644
--- a/vnet/vnet/flow/flow_report.h
+++ b/vnet/vnet/flow/flow_report.h
@@ -51,12 +51,25 @@ typedef vlib_frame_t * (vnet_flow_data_callback_t) (struct flow_report_main *,
                                                     struct flow_report *,
                                                     vlib_frame_t *, u32 *, 
                                                     u32);
+
+typedef union {
+  void * as_ptr;
+  uword as_uword;
+} opaque_t;
+
+typedef struct {
+  u32 domain_id;
+  u32 sequence_number;
+  u16 src_port;
+  u16 n_reports;
+  u16 next_template_no;
+} flow_report_stream_t;
+
 typedef struct flow_report {
   /* ipfix rewrite, set by callback */
   u8 * rewrite;
-  u32 sequence_number;
-  u32 domain_id;
-  u16 src_port;
+  u16 template_id;
+  u32 stream_index;
   f64 last_template_sent;
   int update_rewrite;
 
@@ -64,7 +77,7 @@ typedef struct flow_report {
   uword * fields_to_send;
 
   /* Opaque data */
-  void * opaque;
+  opaque_t opaque;
 
   /* build-the-rewrite callback */
   vnet_flow_rewrite_callback_t *rewrite_callback;
@@ -75,6 +88,7 @@ typedef struct flow_report {
 
 typedef struct flow_report_main {
   flow_report_t * reports;
+  flow_report_stream_t * streams;
 
   /* ipfix collector ip address, port, our ip address, fib index */
   ip4_address_t ipfix_collector;
@@ -88,6 +102,9 @@ typedef struct flow_report_main {
   /* time interval in seconds after which to resend templates */
   u32 template_interval;
 
+  /* UDP checksum calculation enable flag */
+  u8 udp_checksum;
+
   /* time scale transform. Joy. */
   u32 unix_time_0;
   f64 vlib_time_0;
@@ -106,7 +123,7 @@ int vnet_flow_report_enable_disable (u32 sw_if_index, u32 table_index,
 typedef struct {
   vnet_flow_data_callback_t *flow_data_callback;
   vnet_flow_rewrite_callback_t *rewrite_callback;
-  void * opaque;
+  opaque_t opaque;
   int is_add;
   u32 domain_id;
   u16 src_port;
@@ -117,4 +134,10 @@ int vnet_flow_report_add_del (flow_report_main_t *frm,
 
 void vnet_flow_reports_reset (flow_report_main_t * frm);
 
+void vnet_stream_reset (flow_report_main_t * frm, u32 stream_index);
+
+int vnet_stream_change (flow_report_main_t * frm,
+                        u32 old_domain_id, u16 old_src_port,
+                        u32 new_domain_id, u16 new_src_port);
+
 #endif /* __included_vnet_flow_report_h__ */
diff --git a/vnet/vnet/flow/flow_report_classify.c b/vnet/vnet/flow/flow_report_classify.c
index f13f4fe8..95403793 100644
--- a/vnet/vnet/flow/flow_report_classify.c
+++ b/vnet/vnet/flow/flow_report_classify.c
@@ -16,23 +16,28 @@
 #include <vnet/flow/flow_report_classify.h>
 #include <vnet/api_errno.h>
 
+/* Common prefix of tcp and udp headers
+ * containing only source and destination port fields */
 typedef struct {
-  u32 classify_table_index;
-} flow_report_classify_main_t;
+  u16 src_port, dst_port;
+} tcpudp_header_t;
 
 flow_report_classify_main_t flow_report_classify_main;
 
-static u8 * template_rewrite (flow_report_main_t * frm,
-                              flow_report_t * fr,
-                              ip4_address_t * collector_address,
-                              ip4_address_t * src_address,
-                              u16 collector_port)
+u8 * ipfix_classify_template_rewrite (flow_report_main_t * frm,
+                                      flow_report_t * fr,
+                                      ip4_address_t * collector_address,
+                                      ip4_address_t * src_address,
+                                      u16 collector_port)
 {
+  flow_report_classify_main_t * fcm = &flow_report_classify_main;
   vnet_classify_table_t * tblp;
   vnet_classify_main_t * vcm = &vnet_classify_main;
-  flow_report_classify_main_t *fcm =
-    (flow_report_classify_main_t *) fr->opaque;
+  u32 flow_table_index = fr->opaque.as_uword;
+  u8 * ip_start;
   ip4_header_t * ip;
+  ip6_header_t * ip6;
+  tcpudp_header_t * tcpudp;
   udp_header_t * udp;
   ipfix_message_header_t * h;
   ipfix_set_header_t * s;
@@ -44,20 +49,28 @@ static u8 * template_rewrite (flow_report_main_t * frm,
   i32 l3_offset = -2;  /* sizeof (ethernet_header_t) - sizeof (u32x4) */
   u32 field_count = 0;
   u32 field_index = 0;
-  
-  tblp = pool_elt_at_index (vcm->tables, fcm->classify_table_index);
+  flow_report_stream_t * stream;
+  u8 ip_version;
+  u8 transport_protocol;
+
+  stream = &frm->streams[fr->stream_index];
+
+  ipfix_classify_table_t * table = &fcm->tables[flow_table_index];
+
+  ip_version = table->ip_version;
+  transport_protocol = table->transport_protocol;
+
+  tblp = pool_elt_at_index (vcm->tables, table->classify_table_index);
 
   /* 
    * Mumble, assumes that we're not classifying on L2 or first 2 octets
    * of L3..
    */
 
-  ip = (ip4_header_t *)(((u8 *)(tblp->mask)) + l3_offset);
-  udp = (udp_header_t *)(ip+1);
-
   /* Determine field count */
+  ip_start = ((u8 *)(tblp->mask)) + l3_offset;
 #define _(field,mask,item,length)                                       \
-  if ((field) == (mask))                                                \
+  if (memcmp(&field, &mask, length) == 0)                               \
     {                                                                   \
       field_count++;                                                    \
                                                                         \
@@ -65,9 +78,9 @@ static u8 * template_rewrite (flow_report_main_t * frm,
                                             field_index, 1);            \
     }                                                                   \
   field_index++;
-  
   foreach_ipfix_field;
 #undef _
+
   /* Add packetTotalCount manually */
   field_count += 1;
 
@@ -92,19 +105,18 @@ static u8 * template_rewrite (flow_report_main_t * frm,
   ip->protocol = IP_PROTOCOL_UDP;
   ip->src_address.as_u32 = src_address->as_u32;
   ip->dst_address.as_u32 = collector_address->as_u32;
-  udp->src_port = clib_host_to_net_u16 (fr->src_port);
+  udp->src_port = clib_host_to_net_u16 (stream->src_port);
   udp->dst_port = clib_host_to_net_u16 (collector_port);
   udp->length = clib_host_to_net_u16 (vec_len(rewrite) - sizeof (*ip));
 
   /* FIXUP: message header export_time */ 
   /* FIXUP: message header sequence_number */
-  h->domain_id = clib_host_to_net_u32 (fr->domain_id);
+  h->domain_id = clib_host_to_net_u32 (stream->domain_id);
 
   /* Take another trip through the mask and build the template */
-  ip = (ip4_header_t *)(((u8 *)(tblp->mask)) + l3_offset);
-  udp = (udp_header_t *)(ip+1);
+  ip_start = ((u8 *)(tblp->mask)) + l3_offset;
 #define _(field,mask,item,length)                               \
-  if ((field) == (mask))                                        \
+  if (memcmp(&field, &mask, length) == 0)                       \
     {                                                           \
       f->e_id_length = ipfix_e_id_length (0 /* enterprise */,   \
                                           item, length);        \
@@ -123,7 +135,7 @@ static u8 * template_rewrite (flow_report_main_t * frm,
   
   ASSERT (f - first_field);
   /* Field count in this template */
-  t->id_count = ipfix_id_count (256 /* template_id */, f - first_field);
+  t->id_count = ipfix_id_count (fr->template_id, f - first_field);
 
   /* set length in octets*/
   s->set_id_length = ipfix_set_id_length (2 /* set_id */, (u8 *) f - (u8 *)s);
@@ -137,16 +149,16 @@ static u8 * template_rewrite (flow_report_main_t * frm,
   return rewrite;
 }
 
-static vlib_frame_t * send_flows (flow_report_main_t * frm, 
-                                  flow_report_t * fr,
-                                  vlib_frame_t * f, u32 * to_next, 
-                                  u32 node_index)
+vlib_frame_t * ipfix_classify_send_flows (flow_report_main_t * frm,
+                                          flow_report_t * fr,
+                                          vlib_frame_t * f,
+                                          u32 * to_next,
+                                          u32 node_index)
 {
+  flow_report_classify_main_t * fcm = &flow_report_classify_main;
   vnet_classify_main_t * vcm = &vnet_classify_main;
-  flow_report_classify_main_t * fcm =
-    (flow_report_classify_main_t *) fr->opaque;
-  vnet_classify_table_t * t = 
-    pool_elt_at_index (vcm->tables, fcm->classify_table_index);
+  u32 flow_table_index = fr->opaque.as_uword;
+  vnet_classify_table_t * t;
   vnet_classify_bucket_t * b;
   vnet_classify_entry_t * v, * save_v;
   vlib_buffer_t *b0 = 0;
@@ -157,14 +169,28 @@ static vlib_frame_t * send_flows (flow_report_main_t * frm,
   ip4_ipfix_template_packet_t * tp;
   ipfix_message_header_t * h = 0;
   ipfix_set_header_t * s = 0;
+  u8 * ip_start;
   ip4_header_t * ip;
+  ip6_header_t * ip6;
+  tcpudp_header_t * tcpudp;
   udp_header_t * udp;
   int field_index;
-  ip4_header_t * match;
   u32 records_this_buffer;
   u16 new_l0, old_l0;
   ip_csum_t sum0;
   vlib_main_t * vm = frm->vlib_main;
+  flow_report_stream_t * stream;
+  u8 ip_version;
+  u8 transport_protocol;
+
+  stream = &frm->streams[fr->stream_index];
+
+  ipfix_classify_table_t * table = &fcm->tables[flow_table_index];
+
+  ip_version = table->ip_version;
+  transport_protocol = table->transport_protocol;
+
+  t = pool_elt_at_index (vcm->tables, table->classify_table_index);
   
   while (__sync_lock_test_and_set (t->writer_lock, 1))
     ; 
@@ -216,19 +242,16 @@ static vlib_frame_t * send_flows (flow_report_main_t * frm,
                   h->export_time = clib_host_to_net_u32(h->export_time);
                   
                   /* FIXUP: message header sequence_number */
-                  h->sequence_number = fr->sequence_number;
+                  h->sequence_number = stream->sequence_number;
                   h->sequence_number = clib_host_to_net_u32 (h->sequence_number);
 
                   next_offset = (u32) (((u8 *)(s+1)) - (u8 *)tp);
                   record_offset = next_offset;
                   records_this_buffer = 0;
                 }
-              
+
               field_index = 0;
-              match = (ip4_header_t *) (((u8 *)v->key) - 2);
-              ip = match;
-              udp = (udp_header_t * )(ip+1);
-              
+              ip_start = ((u8 *)v->key) - 2;
 #define _(field,mask,item,length)                                       \
               if (clib_bitmap_get (fr->fields_to_send, field_index))    \
                 {                                                       \
@@ -239,7 +262,7 @@ static vlib_frame_t * send_flows (flow_report_main_t * frm,
               field_index++;
               foreach_ipfix_field;
 #undef _
-              
+
               /* Add packetTotalCount manually */
               {
                 u64 packets = clib_host_to_net_u64 (v->hits);
@@ -247,7 +270,7 @@ static vlib_frame_t * send_flows (flow_report_main_t * frm,
                 next_offset += sizeof (packets);
               }
               records_this_buffer++;
-              fr->sequence_number++;
+              stream->sequence_number++;
               
               /* Next record will have the same size as this record */
               u32 next_record_size = next_offset - record_offset;
@@ -255,7 +278,7 @@ static vlib_frame_t * send_flows (flow_report_main_t * frm,
 
               if (next_offset + next_record_size > frm->path_mtu)
                 {
-                  s->set_id_length = ipfix_set_id_length (256 /* template ID*/, 
+                  s->set_id_length = ipfix_set_id_length (fr->template_id,
                                                           next_offset - 
                                                           (sizeof (*ip) + sizeof (*udp) +
                                                            sizeof (*h)));
@@ -280,7 +303,15 @@ static vlib_frame_t * send_flows (flow_report_main_t * frm,
                   ip->length = new_l0;
                   udp->length = 
                       clib_host_to_net_u16 (b0->current_length - sizeof (*ip));
-                  
+
+                  if (frm->udp_checksum)
+                    {
+                      /* RFC 7011 section 10.3.2. */
+                      udp->checksum = ip4_tcp_udp_compute_checksum (vm, b0, ip);
+                      if (udp->checksum == 0)
+                        udp->checksum = 0xffff;
+                    }
+
                   ASSERT (ip->checksum == ip4_header_checksum (ip));
 
                   to_next[0] = bi0;
@@ -304,7 +335,7 @@ static vlib_frame_t * send_flows (flow_report_main_t * frm,
  flush:
   if (b0)
     {
-        s->set_id_length = ipfix_set_id_length (256 /* template ID*/, 
+        s->set_id_length = ipfix_set_id_length (fr->template_id,
                                                 next_offset - 
                                                 (sizeof (*ip) + sizeof (*udp) +
                                                  sizeof (*h)));
@@ -328,6 +359,14 @@ static vlib_frame_t * send_flows (flow_report_main_t * frm,
       ip->length = new_l0;
       udp->length = clib_host_to_net_u16 (b0->current_length - sizeof (*ip));
 
+      if (frm->udp_checksum)
+        {
+          /* RFC 7011 section 10.3.2. */
+          udp->checksum = ip4_tcp_udp_compute_checksum (vm, b0, ip);
+          if (udp->checksum == 0)
+          udp->checksum = 0xffff;
+        }
+
       ASSERT (ip->checksum == ip4_header_checksum (ip));
       
       to_next[0] = bi0;
@@ -341,47 +380,81 @@ static vlib_frame_t * send_flows (flow_report_main_t * frm,
   return f;
 }
 
-
 static clib_error_t *
-flow_classify_command_fn (vlib_main_t * vm,
-		 unformat_input_t * input,
-		 vlib_cli_command_t * cmd)
+ipfix_classify_table_add_del_command_fn (vlib_main_t * vm,
+                                         unformat_input_t * input,
+                                         vlib_cli_command_t * cmd)
 {
   flow_report_classify_main_t *fcm = &flow_report_classify_main;
   flow_report_main_t *frm = &flow_report_main;
   vnet_flow_report_add_del_args_t args;
+  ipfix_classify_table_t * table;
   int rv;
-  int is_add = 1;
-  u32 domain_id = 0;
-  u32 src_port = UDP_DST_PORT_ipfix;
+  int is_add = -1;
+  u32 classify_table_index;
+  u8 ip_version = 0;
+  u8 transport_protocol = 255;
+
+  if (fcm->src_port == 0)
+    clib_error_return (0, "call 'set ipfix classify stream' first");
 
-  domain_id = 0;
-  fcm->classify_table_index = ~0;
   memset (&args, 0, sizeof (args));
 
   while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT) {
-    if (unformat (input, "table %d", &fcm->classify_table_index))
-      ;
-    else if (unformat (input, "domain %d", &domain_id))
-      ;
-    else if (unformat (input, "src-port %d", &src_port))
-      ;
+    if (unformat (input, "add"))
+      is_add = 1;
     else if (unformat (input, "del"))
       is_add = 0;
+    else if (unformat (input, "%d", &classify_table_index))
+      ;
+    else if (unformat (input, "ip4"))
+      ip_version = 4;
+    else if (unformat (input, "ip6"))
+      ip_version = 6;
+    else if (unformat (input, "tcp"))
+      transport_protocol = 6;
+    else if (unformat (input, "udp"))
+      transport_protocol = 17;
     else
       return clib_error_return (0, "unknown input `%U'",
                                 format_unformat_error, input);
   }
 
-  if (fcm->classify_table_index == ~0)
+  if (is_add == -1)
+    return clib_error_return (0, "expecting: add|del");
+  if (classify_table_index == ~0)
     return clib_error_return (0, "classifier table not specified");
+  if (ip_version == 0)
+    return clib_error_return (0, "IP version not specified");
+
+  table = 0;
+  int i;
+  for (i = 0; i < vec_len(fcm->tables); i++)
+    if (ipfix_classify_table_index_valid(i))
+      if (fcm->tables[i].classify_table_index == classify_table_index) {
+        table = &fcm->tables[i];
+        break;
+      }
+
+  if (is_add) {
+    if (table)
+      return clib_error_return (0, "Specified classifier table already used");
+    table = ipfix_classify_add_table();
+    table->classify_table_index = classify_table_index;
+  } else {
+    if (!table)
+      return clib_error_return (0, "Specified classifier table not registered");
+  }
+
+  table->ip_version = ip_version;
+  table->transport_protocol = transport_protocol;
 
-  args.opaque = (void *) fcm;
-  args.rewrite_callback = template_rewrite;
-  args.flow_data_callback = send_flows;
+  args.opaque.as_uword = table - fcm->tables;
+  args.rewrite_callback = ipfix_classify_template_rewrite;
+  args.flow_data_callback = ipfix_classify_send_flows;
   args.is_add = is_add;
-  args.domain_id = domain_id;
-  args.src_port = (u16)src_port;
+  args.domain_id = fcm->domain_id;
+  args.src_port = fcm->src_port;
 
   rv = vnet_flow_report_add_del (frm, &args);
 
@@ -390,18 +463,73 @@ flow_classify_command_fn (vlib_main_t * vm,
     case 0:
       break;
     case VNET_API_ERROR_NO_SUCH_ENTRY:
-      return clib_error_return (0, "registration not found...");
+      return clib_error_return (0, "Flow report not found");
+    case VNET_API_ERROR_VALUE_EXIST:
+      return clib_error_return (0, "Flow report already exists");
+    case VNET_API_ERROR_INVALID_VALUE:
+      return clib_error_return (0, "Expecting either still unused values "
+                                   "for both domain_id and src_port "
+                                   "or already used values for both fields");
     default:
       return clib_error_return (0, "vnet_flow_report_add_del returned %d", rv);
     }
 
+  if (is_add) {
+    if (rv != 0)
+      ipfix_classify_delete_table(table - fcm->tables);
+  } else {
+    if (rv == 0)
+      ipfix_classify_delete_table(table - fcm->tables);
+  }
+
+  return 0;
+}
+
+VLIB_CLI_COMMAND (ipfix_classify_table_add_del_command, static) = {
+  .path = "ipfix classify table",
+  .short_help = "ipfix classify table add|del <table-index>",
+  .function = ipfix_classify_table_add_del_command_fn,
+};
+
+static clib_error_t *
+set_ipfix_classify_stream_command_fn (vlib_main_t * vm,
+                                      unformat_input_t * input,
+                                      vlib_cli_command_t * cmd)
+{
+  flow_report_classify_main_t *fcm = &flow_report_classify_main;
+  flow_report_main_t *frm = &flow_report_main;
+  u32 domain_id = 1;
+  u32 src_port = UDP_DST_PORT_ipfix;
+
+  while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT) {
+    if (unformat (input, "domain %d", &domain_id))
+      ;
+    else if (unformat (input, "src-port %d", &src_port))
+      ;
+    else
+      return clib_error_return (0, "unknown input `%U'",
+                                format_unformat_error, input);
+  }
+
+  if (fcm->src_port != 0 &&
+      (fcm->domain_id != domain_id ||
+       fcm->src_port != (u16)src_port)) {
+    int rv = vnet_stream_change (frm, fcm->domain_id, fcm->src_port,
+                                 domain_id, (u16)src_port);
+    ASSERT (rv == 0);
+  }
+
+  fcm->domain_id = domain_id;
+  fcm->src_port = (u16)src_port;
+
   return 0;
 }
 
-VLIB_CLI_COMMAND (flow_classify_command, static) = {
-  .path = "flow classify",
-  .short_help = "flow classify",
-  .function = flow_classify_command_fn,
+VLIB_CLI_COMMAND (set_ipfix_classify_stream_command, static) = {
+  .path = "set ipfix classify stream",
+  .short_help = "set ipfix classify stream"
+                "[domain <domain-id>] [src-port <src-port>]",
+  .function = set_ipfix_classify_stream_command_fn,
 };
 
 static clib_error_t *
diff --git a/vnet/vnet/flow/flow_report_classify.h b/vnet/vnet/flow/flow_report_classify.h
index c72ef0eb..77d98b58 100644
--- a/vnet/vnet/flow/flow_report_classify.h
+++ b/vnet/vnet/flow/flow_report_classify.h
@@ -15,12 +15,108 @@
 #ifndef __included_flow_report_classify_h__
 #define __included_flow_report_classify_h__
 
-/* Note: add +2 to udp (src,dst) port enum values to get TCP values */
-#define foreach_ipfix_field                                             \
-_(ip->src_address.as_u32, 0xffffffff, sourceIPv4Address, 4)             \
-_(ip->dst_address.as_u32, 0xffffffff, destinationIPv4Address, 4)        \
-_(ip->protocol, 0xFF, protocolIdentifier, 1)                            \
-_(udp->src_port, 0xFFFF, udpSourcePort, 2)                              \
-_(udp->dst_port, 0xFFFF, udpDestinationPort, 2)
+#define foreach_ipfix_ip4_field                                             \
+_(ip->src_address.as_u32, ((u32[]){0xFFFFFFFF}), sourceIPv4Address, 4)      \
+_(ip->dst_address.as_u32, ((u32[]){0xFFFFFFFF}), destinationIPv4Address, 4) \
+_(ip->protocol, ((u8[]){0xFF}), protocolIdentifier, 1)
+
+#define foreach_ipfix_ip6_field                                             \
+_(ip6->src_address.as_u8,                                                   \
+  ((u32[]){0xFFFFFFFF,0xFFFFFFFF,0xFFFFFFFF,0xFFFFFFFF}),                   \
+  sourceIPv6Address, 16)                                                    \
+_(ip6->dst_address.as_u8,                                                   \
+  ((u32[]){0xFFFFFFFF,0xFFFFFFFF,0xFFFFFFFF,0xFFFFFFFF}),                   \
+  destinationIPv6Address, 16)                                               \
+_(ip6->protocol, ((u8[]){0xFF}), protocolIdentifier, 1)
+
+#define foreach_ipfix_tcpudp_field                                          \
+_(tcpudp->src_port, ((u16[]){0xFFFF}), sourceTransportPort, 2)              \
+_(tcpudp->dst_port, ((u16[]){0xFFFF}), destinationTransportPort, 2)
+
+#define foreach_ipfix_tcp_field                                             \
+_(tcpudp->src_port, ((u16[]){0xFFFF}), tcpSourcePort, 2)                    \
+_(tcpudp->dst_port, ((u16[]){0xFFFF}), tcpDestinationPort, 2)
+
+#define foreach_ipfix_udp_field                                             \
+_(tcpudp->src_port, ((u16[]){0xFFFF}), udpSourcePort, 2)                    \
+_(tcpudp->dst_port, ((u16[]){0xFFFF}), udpDestinationPort, 2)
+
+#define foreach_ipfix_transport_protocol_field                              \
+  switch (transport_protocol) {                                             \
+    case 255:                                                               \
+      foreach_ipfix_tcpudp_field;                                           \
+      break;                                                                \
+    case 6:                                                                 \
+      foreach_ipfix_tcp_field;                                              \
+      break;                                                                \
+    case 17:                                                                \
+      foreach_ipfix_udp_field;                                              \
+      break;                                                                \
+  }
+
+#define foreach_ipfix_field                                                 \
+  if (ip_version == 4) {                                                    \
+    ip = (ip4_header_t *)ip_start;                                          \
+    tcpudp = (tcpudp_header_t *)(ip+1);                                     \
+    foreach_ipfix_ip4_field;                                                \
+  } else {                                                                  \
+    ip6 = (ip6_header_t *)ip_start;                                         \
+    tcpudp = (tcpudp_header_t *)(ip6+1);                                    \
+    foreach_ipfix_ip6_field;                                                \
+  }                                                                         \
+  foreach_ipfix_transport_protocol_field
+
+typedef struct {
+  u32 classify_table_index;
+  u8 ip_version;
+  u8 transport_protocol;
+} ipfix_classify_table_t;
+
+typedef struct {
+  u32 domain_id;
+  u16 src_port;
+  ipfix_classify_table_t * tables;
+} flow_report_classify_main_t;
+
+extern flow_report_classify_main_t flow_report_classify_main;
+
+static_always_inline u8 ipfix_classify_table_index_valid (u32 index)
+{
+  flow_report_classify_main_t * fcm = &flow_report_classify_main;
+  return index < vec_len(fcm->tables) &&
+         fcm->tables[index].classify_table_index != ~0;
+}
+
+static_always_inline ipfix_classify_table_t * ipfix_classify_add_table (void)
+{
+  flow_report_classify_main_t * fcm = &flow_report_classify_main;
+  u32 i;
+  for (i = 0; i < vec_len(fcm->tables); i++)
+    if (!ipfix_classify_table_index_valid(i))
+      return &fcm->tables[i];
+  u32 index = vec_len(fcm->tables);
+  vec_validate(fcm->tables, index);
+  return &fcm->tables[index];
+}
+
+static_always_inline void ipfix_classify_delete_table (u32 index)
+{
+  flow_report_classify_main_t * fcm = &flow_report_classify_main;
+  ASSERT (index < vec_len(fcm->tables));
+  ASSERT (fcm->tables[index].classify_table_index != ~0);
+  fcm->tables[index].classify_table_index = ~0;
+}
+
+u8 * ipfix_classify_template_rewrite (flow_report_main_t * frm,
+                                      flow_report_t * fr,
+                                      ip4_address_t * collector_address,
+                                      ip4_address_t * src_address,
+                                      u16 collector_port);
+
+vlib_frame_t * ipfix_classify_send_flows (flow_report_main_t * frm,
+                                          flow_report_t * fr,
+                                          vlib_frame_t * f,
+                                          u32 * to_next,
+                                          u32 node_index);
 
 #endif /* __included_flow_report_classify_h__ */