ipfix: implement flow caching and aggregation in exporter

Implement a per-exporter flow cache with active timeout expiration.
Add columns "cache_active_timeout" and "cache_max_flows" into table
"IPFIX" to configure each cache.

Add per-flow elements "octetDeltaSumOfSquares",
"minimumIpTotalLength", and "maximumIpTotalLength" to replace
"ethernetTotalLength".  Add per-flow element "flowEndReason" to
indicate whether a flow has expired because of an active timeout, the
cache size limit being reached, or the exporter being stopped.

Signed-off-by: Romain Lenglet <rlenglet@vmware.com>
Signed-off-by: Ben Pfaff <blp@nicira.com>

NEWS

@@ -25,6 +25,7 @@ v1.12.0 - xx xxx xxxx
     - Support for Linux kernels up to 3.10
     - ovs-ofctl:
       * New "ofp-parse" for printing OpenFlow messages read from a file.
+    - Added configurable flow caching support to IPFIX exporter.
 
 
 v1.11.0 - xx xxx xxxx

ofproto/ofproto-dpif-ipfix.h

@@ -41,4 +41,7 @@ void dpif_ipfix_flow_sample(struct dpif_ipfix *, struct ofpbuf *,
                             const struct flow *, uint32_t, uint16_t, uint32_t,
                             uint32_t);
 
+void dpif_ipfix_run(struct dpif_ipfix *);
+void dpif_ipfix_wait(struct dpif_ipfix *);
+
 #endif /* ofproto/ofproto-dpif-ipfix.h */

ofproto/ofproto-dpif.c

@@ -1585,6 +1585,9 @@ run(struct ofproto *ofproto_)
     if (ofproto->sflow) {
         dpif_sflow_run(ofproto->sflow);
     }
+    if (ofproto->ipfix) {
+        dpif_ipfix_run(ofproto->ipfix);
+    }
 
     HMAP_FOR_EACH (ofport, up.hmap_node, &ofproto->up.ports) {
         port_run(ofport);
@@ -1644,6 +1647,9 @@ wait(struct ofproto *ofproto_)
     if (ofproto->sflow) {
         dpif_sflow_wait(ofproto->sflow);
     }
+    if (ofproto->ipfix) {
+        dpif_ipfix_wait(ofproto->ipfix);
+    }
     HMAP_FOR_EACH (ofport, up.hmap_node, &ofproto->up.ports) {
         port_wait(ofport);
     }
@@ -1973,20 +1979,25 @@ set_ipfix(
 {
     struct ofproto_dpif *ofproto = ofproto_dpif_cast(ofproto_);
     struct dpif_ipfix *di = ofproto->ipfix;
+    bool has_options = bridge_exporter_options || flow_exporters_options;
 
-    if (bridge_exporter_options || flow_exporters_options) {
-        if (!di) {
-            di = ofproto->ipfix = dpif_ipfix_create();
-        }
+    if (has_options && !di) {
+        di = ofproto->ipfix = dpif_ipfix_create();
+    }
+
+    if (di) {
+        /* Call set_options in any case to cleanly flush the flow
+         * caches in the last exporters that are to be destroyed. */
         dpif_ipfix_set_options(
             di, bridge_exporter_options, flow_exporters_options,
             n_flow_exporters_options);
-    } else {
-        if (di) {
+
+        if (!has_options) {
             dpif_ipfix_unref(di);
             ofproto->ipfix = NULL;
         }
     }
+
     return 0;
 }
 

ofproto/ofproto.h

@@ -76,11 +76,15 @@ struct ofproto_ipfix_bridge_exporter_options {
     uint32_t sampling_rate;
     uint32_t obs_domain_id;  /* Bridge-wide Observation Domain ID. */
     uint32_t obs_point_id;  /* Bridge-wide Observation Point ID. */
+    uint32_t cache_active_timeout;
+    uint32_t cache_max_flows;
 };
 
 struct ofproto_ipfix_flow_exporter_options {
     uint32_t collector_set_id;
     struct sset targets;
+    uint32_t cache_active_timeout;
+    uint32_t cache_max_flows;
 };
 
 struct ofproto_stp_settings {

utilities/ovs-vsctl.8.in

@@ -942,11 +942,12 @@ Deconfigure sFlow from \fBbr0\fR, which also destroys the sFlow record
 .PP
 Configure bridge \fBbr0\fR to send one IPFIX flow record per packet
 sample to UDP port 4739 on host 192.168.0.34, with Observation Domain
-ID 123 and Observation Point ID 456:
+ID 123 and Observation Point ID 456, a flow cache active timeout of 1
+minute (60 seconds), and a maximum flow cache size of 13 flows:
 .IP
 .B "ovs\-vsctl \-\- set Bridge br0 ipfix=@i \(rs"
 .IP
-.B "\-\- \-\-id=@i create IPFIX targets=\(rs\(dq192.168.0.34:4739\(rs\(dq obs_domain_id=123 obs_point_id=456"
+.B "\-\- \-\-id=@i create IPFIX targets=\(rs\(dq192.168.0.34:4739\(rs\(dq obs_domain_id=123 obs_point_id=456 cache_active_timeout=60 cache_max_flows=13"
 .PP
 Deconfigure the IPFIX settings from \fBbr0\fR, which also destroys the
 IPFIX record (since it is now unreferenced):

vswitchd/bridge.c

@@ -1025,6 +1025,12 @@ bridge_configure_ipfix(struct bridge *br)
         if (be_cfg->obs_point_id) {
             be_opts.obs_point_id = *be_cfg->obs_point_id;
         }
+        if (be_cfg->cache_active_timeout) {
+            be_opts.cache_active_timeout = *be_cfg->cache_active_timeout;
+        }
+        if (be_cfg->cache_max_flows) {
+            be_opts.cache_max_flows = *be_cfg->cache_max_flows;
+        }
     }
 
     if (n_fe_opts > 0) {
@@ -1037,6 +1043,10 @@ bridge_configure_ipfix(struct bridge *br)
                 sset_init(&opts->targets);
                 sset_add_array(&opts->targets, fe_cfg->ipfix->targets,
                                fe_cfg->ipfix->n_targets);
+                opts->cache_active_timeout = fe_cfg->ipfix->cache_active_timeout
+                    ? *fe_cfg->ipfix->cache_active_timeout : 0;
+                opts->cache_max_flows = fe_cfg->ipfix->cache_max_flows
+                    ? *fe_cfg->ipfix->cache_max_flows : 0;
                 opts++;
             }
         }

vswitchd/vswitch.ovsschema

@@ -1,6 +1,6 @@
 {"name": "Open_vSwitch",
  "version": "7.3.0",
- "cksum": "1081379034 19765",
+ "cksum": "2483452374 20182",
  "tables": {
    "Open_vSwitch": {
      "columns": {
@@ -423,6 +423,16 @@
                           "minInteger": 0,
                           "maxInteger": 4294967295},
                   "min": 0, "max": 1}},
+       "cache_active_timeout": {
+         "type": {"key": {"type": "integer",
+                          "minInteger": 0,
+                          "maxInteger": 4200},
+                  "min": 0, "max": 1}},
+       "cache_max_flows": {
+         "type": {"key": {"type": "integer",
+                          "minInteger": 0,
+                          "maxInteger": 4294967295},
+                  "min": 0, "max": 1}},
        "external_ids": {
          "type": {"key": "string", "value": "string",
                   "min": 0, "max": "unlimited"}}}},

vswitchd/vswitch.xml

@@ -3513,6 +3513,18 @@
       referenced from a <ref table="Flow_Sample_Collector_Set"/>.
     </column>
 
+    <column name="cache_active_timeout">
+      The maximum period in seconds for which an IPFIX flow record is
+      cached and aggregated before being sent.  If not specified,
+      defaults to 0.  If 0, caching is disabled.
+    </column>
+
+    <column name="cache_max_flows">
+      The maximum number of IPFIX flow records that can be cached at a
+      time.  If not specified, defaults to 0.  If 0, caching is
+      disabled.
+    </column>
+
     <group title="Common Columns">
       The overall purpose of these columns is described under <code>Common
       Columns</code> at the beginning of this document.