sFlow: Genericize/simplify kernel sFlow implementation

Following patch adds sampling action which takes probability and set
of actions as arguments. When probability is hit, actions are executed for
given packet.
USERSPACE action's userdata (u64) is used to store struct
user_action_cookie as cookie. CONTROLLER action is fixed accordingly.

Now we can remove sFlow code from kernel and implement sFlow generically
as SAMPLE action. sFlow is defined as SAMPLE Action with probability (sFlow
sampling rate) and USERSPACE action as argument. USERSPACE action's data
is used as cookie. sFlow uses this cookie to store output-port, number of
output ports and vlan-id. sample-pool is calculated by using vport
stats.

Signed-off-by: Pravin Shelar <pshelar@nicira.com>
Acked-by: Jesse Gross <jesse@nicira.com>
Acked-by: Ben Pfaff <blp@nicira.com>

3 files changed, 244 insertions, 97 deletions

diff --git a/ofproto/ofproto-dpif-sflow.c b/ofproto/ofproto-dpif-sflow.c
index 68eb804d..d17b0be1 100644
--- a/ofproto/ofproto-dpif-sflow.c
+++ b/ofproto/ofproto-dpif-sflow.c
@@ -46,7 +46,6 @@ struct dpif_sflow_port {
 };
 
 struct dpif_sflow {
-    struct ofproto *ofproto;
     struct collectors *collectors;
     SFLAgent *sflow_agent;
     struct ofproto_sflow_options *options;
@@ -54,6 +53,7 @@ struct dpif_sflow {
     time_t next_tick;
     size_t n_flood, n_all;
     struct hmap ports;          /* Contains "struct dpif_sflow_port"s. */
+    uint32_t probability;
 };
 
 static void dpif_sflow_del_port__(struct dpif_sflow *,
@@ -270,7 +270,7 @@ dpif_sflow_clear(struct dpif_sflow *ds)
     ds->options = NULL;
 
     /* Turn off sampling to save CPU cycles. */
-    dpif_set_sflow_probability(ds->dpif, 0);
+    ds->probability = 0;
 }
 
 bool
@@ -288,9 +288,19 @@ dpif_sflow_create(struct dpif *dpif)
     ds->dpif = dpif;
     ds->next_tick = time_now() + 1;
     hmap_init(&ds->ports);
+    ds->probability = 0;
     return ds;
 }
 
+/* 32-bit fraction of packets to sample with.  A value of 0 samples no packets,
+ * a value of %UINT32_MAX samples all packets and intermediate values sample
+ * intermediate fractions of packets. */
+uint32_t
+dpif_sflow_get_probability(const struct dpif_sflow *ds)
+{
+    return ds->probability;
+}
+
 void
 dpif_sflow_destroy(struct dpif_sflow *ds)
 {
@@ -457,8 +467,7 @@ dpif_sflow_set_options(struct dpif_sflow *ds,
     sfl_receiver_set_sFlowRcvrTimeout(receiver, 0xffffffff);
 
     /* Set the sampling_rate down in the datapath. */
-    dpif_set_sflow_probability(ds->dpif,
-                               MAX(1, UINT32_MAX / options->sampling_rate));
+    ds->probability = MAX(1, UINT32_MAX / ds->options->sampling_rate);
 
     /* Add samplers and pollers for the currently known ports. */
     HMAP_FOR_EACH (dsp, hmap_node, &ds->ports) {
@@ -467,7 +476,7 @@ dpif_sflow_set_options(struct dpif_sflow *ds,
     }
 }
 
-static int
+int
 dpif_sflow_odp_port_to_ifindex(const struct dpif_sflow *ds,
                                uint16_t odp_port)
 {
@@ -476,24 +485,34 @@ dpif_sflow_odp_port_to_ifindex(const struct dpif_sflow *ds,
 }
 
 void
-dpif_sflow_received(struct dpif_sflow *ds, const struct dpif_upcall *upcall,
-                    const struct flow *flow)
+dpif_sflow_received(struct dpif_sflow *ds, struct ofpbuf *packet,
+                    const struct flow *flow,
+                    const struct user_action_cookie *cookie)
 {
     SFL_FLOW_SAMPLE_TYPE fs;
     SFLFlow_sample_element hdrElem;
     SFLSampled_header *header;
     SFLFlow_sample_element switchElem;
     SFLSampler *sampler;
-    unsigned int left;
-    struct nlattr *a;
-    size_t n_outputs;
+    struct dpif_sflow_port *in_dsp;
+    struct netdev_stats stats;
+    int error;
 
     /* Build a flow sample */
     memset(&fs, 0, sizeof fs);
-    fs.input = dpif_sflow_odp_port_to_ifindex(ds,
-				 ofp_port_to_odp_port(flow->in_port));
-    fs.output = 0;              /* Filled in correctly below. */
-    fs.sample_pool = upcall->sample_pool;
+
+    in_dsp = dpif_sflow_find_port(ds, ofp_port_to_odp_port(flow->in_port));
+    if (!in_dsp) {
+        return;
+    }
+    fs.input = SFL_DS_INDEX(in_dsp->dsi);
+
+    error = netdev_get_stats(in_dsp->netdev, &stats);
+    if (error) {
+        VLOG_WARN_RL(&rl, "netdev get-stats error %s", strerror(error));
+        return;
+    }
+    fs.sample_pool = stats.rx_packets;
 
     /* We are going to give it to the sampler that represents this input port.
      * By implementing "ingress-only" sampling like this we ensure that we
@@ -512,54 +531,34 @@ dpif_sflow_received(struct dpif_sflow *ds, const struct dpif_upcall *upcall,
     header->header_protocol = SFLHEADER_ETHERNET_ISO8023;
     /* The frame_length should include the Ethernet FCS (4 bytes),
        but it has already been stripped,  so we need to add 4 here. */
-    header->frame_length = upcall->packet->size + 4;
+    header->frame_length = packet->size + 4;
     /* Ethernet FCS stripped off. */
     header->stripped = 4;
-    header->header_length = MIN(upcall->packet->size,
+    header->header_length = MIN(packet->size,
                                 sampler->sFlowFsMaximumHeaderSize);
-    header->header_bytes = upcall->packet->data;
+    header->header_bytes = packet->data;
 
     /* Add extended switch element. */
     memset(&switchElem, 0, sizeof(switchElem));
     switchElem.tag = SFLFLOW_EX_SWITCH;
     switchElem.flowType.sw.src_vlan = vlan_tci_to_vid(flow->vlan_tci);
     switchElem.flowType.sw.src_priority = vlan_tci_to_pcp(flow->vlan_tci);
-    /* Initialize the output VLAN and priority to be the same as the input,
-       but these fields can be overriden below if affected by an action. */
-    switchElem.flowType.sw.dst_vlan = switchElem.flowType.sw.src_vlan;
-    switchElem.flowType.sw.dst_priority = switchElem.flowType.sw.src_priority;
-
-    /* Figure out the output ports. */
-    n_outputs = 0;
-    NL_ATTR_FOR_EACH_UNSAFE (a, left, upcall->actions, upcall->actions_len) {
-        ovs_be16 tci;
-
-        switch (nl_attr_type(a)) {
-        case OVS_ACTION_ATTR_OUTPUT:
-            fs.output = dpif_sflow_odp_port_to_ifindex(ds, nl_attr_get_u32(a));
-            n_outputs++;
-            break;
-
-        case OVS_ACTION_ATTR_PUSH_VLAN:
-            tci = nl_attr_get_be16(a);
-            switchElem.flowType.sw.dst_vlan = vlan_tci_to_vid(tci);
-            switchElem.flowType.sw.dst_priority = vlan_tci_to_pcp(tci);
-            break;
-
-        default:
-            break;
-        }
-    }
+
+    /* Retrieve data from user_action_cookie. */
+    switchElem.flowType.sw.dst_vlan = vlan_tci_to_vid(cookie->vlan_tci);
+    switchElem.flowType.sw.dst_priority = vlan_tci_to_pcp(cookie->vlan_tci);
 
     /* Set output port, as defined by http://www.sflow.org/sflow_version_5.txt
        (search for "Input/output port information"). */
-    if (!n_outputs) {
+    if (!cookie->n_output) {
         /* This value indicates that the packet was dropped for an unknown
          * reason. */
         fs.output = 0x40000000 | 256;
-    } else if (n_outputs > 1 || !fs.output) {
+    } else if (cookie->n_output > 1 || !cookie->data) {
         /* Setting the high bit means "multiple output ports". */
-        fs.output = 0x80000000 | n_outputs;
+        fs.output = 0x80000000 | cookie->n_output;
+    } else {
+        fs.output = cookie->data;
     }
 
     /* Submit the flow sample to be encoded into the next datagram. */
diff --git a/ofproto/ofproto-dpif-sflow.h b/ofproto/ofproto-dpif-sflow.h
index c7c8b5af..439f4104 100644
--- a/ofproto/ofproto-dpif-sflow.h
+++ b/ofproto/ofproto-dpif-sflow.h
@@ -20,6 +20,7 @@
 
 #include <stdint.h>
 #include "svec.h"
+#include "lib/odp-util.h"
 
 struct dpif;
 struct dpif_upcall;
@@ -27,6 +28,8 @@ struct flow;
 struct ofproto_sflow_options;
 
 struct dpif_sflow *dpif_sflow_create(struct dpif *);
+uint32_t dpif_sflow_get_probability(const struct dpif_sflow *);
+
 void dpif_sflow_destroy(struct dpif_sflow *);
 void dpif_sflow_set_options(struct dpif_sflow *,
                             const struct ofproto_sflow_options *);
@@ -40,7 +43,11 @@ void dpif_sflow_del_port(struct dpif_sflow *, uint16_t ovs_port);
 void dpif_sflow_run(struct dpif_sflow *);
 void dpif_sflow_wait(struct dpif_sflow *);
 
-void dpif_sflow_received(struct dpif_sflow *, const struct dpif_upcall *,
-                         const struct flow *);
+void dpif_sflow_received(struct dpif_sflow *,
+                         struct ofpbuf *,
+                         const struct flow *,
+                         const struct user_action_cookie *);
+
+int dpif_sflow_odp_port_to_ifindex(const struct dpif_sflow *, uint16_t);
 
 #endif /* ofproto/ofproto-dpif-sflow.h */
diff --git a/ofproto/ofproto-dpif.c b/ofproto/ofproto-dpif.c
index ecad4890..f9738e74 100644
--- a/ofproto/ofproto-dpif.c
+++ b/ofproto/ofproto-dpif.c
@@ -205,6 +205,9 @@ struct action_xlate_ctx {
     struct flow base_flow;      /* Flow at the last commit. */
     uint32_t base_priority;     /* Priority at the last commit. */
     uint8_t table_id;           /* OpenFlow table ID where flow was found. */
+    uint32_t sflow_n_outputs;   /* Number of output ports. */
+    uint16_t sflow_odp_port;    /* Output port for composing sFlow action. */
+    uint16_t user_cookie_offset;/* Used for user_action_cookie fixup. */
 };
 
 static void action_xlate_ctx_init(struct action_xlate_ctx *,
@@ -401,7 +404,9 @@ static int expire(struct ofproto_dpif *);
 /* Utilities. */
 static int send_packet(struct ofproto_dpif *, uint32_t odp_port,
                        const struct ofpbuf *packet);
-
+static size_t
+compose_sflow_action(const struct ofproto_dpif *, struct ofpbuf *odp_actions,
+                     const struct flow *, uint32_t odp_port);
 /* Global variables. */
 static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
 
@@ -471,8 +476,7 @@ construct(struct ofproto *ofproto_, int *n_tablesp)
 
     error = dpif_recv_set_mask(ofproto->dpif,
                                ((1u << DPIF_UC_MISS) |
-                                (1u << DPIF_UC_ACTION) |
-                                (1u << DPIF_UC_SAMPLE)));
+                                (1u << DPIF_UC_ACTION)));
     if (error) {
         VLOG_ERR("failed to listen on datapath %s: %s", name, strerror(error));
         dpif_close(ofproto->dpif);
@@ -810,6 +814,7 @@ set_sflow(struct ofproto *ofproto_,
 {
     struct ofproto_dpif *ofproto = ofproto_dpif_cast(ofproto_);
     struct dpif_sflow *ds = ofproto->sflow;
+
     if (sflow_options) {
         if (!ds) {
             struct ofport_dpif *ofport;
@@ -819,11 +824,15 @@ set_sflow(struct ofproto *ofproto_,
                 dpif_sflow_add_port(ds, ofport->odp_port,
                                     netdev_get_name(ofport->up.netdev));
             }
+            ofproto->need_revalidate = true;
         }
         dpif_sflow_set_options(ds, sflow_options);
     } else {
-        dpif_sflow_destroy(ds);
-        ofproto->sflow = NULL;
+        if (ds) {
+            dpif_sflow_destroy(ds);
+            ofproto->need_revalidate = true;
+            ofproto->sflow = NULL;
+        }
     }
     return 0;
 }
@@ -1713,12 +1722,15 @@ send_packet_in(struct ofproto_dpif *ofproto, struct dpif_upcall *upcall,
                const struct flow *flow, bool clone)
 {
     struct ofputil_packet_in pin;
+    struct user_action_cookie cookie;
 
     pin.packet = upcall->packet;
     pin.in_port = flow->in_port;
     pin.reason = upcall->type == DPIF_UC_MISS ? OFPR_NO_MATCH : OFPR_ACTION;
     pin.buffer_id = 0;          /* not yet known */
-    pin.send_len = upcall->userdata;
+
+    memcpy(&cookie, &upcall->userdata, sizeof(cookie));
+    pin.send_len = cookie.data;
     connmgr_send_packet_in(ofproto->up.connmgr, &pin, flow,
                            clone ? NULL : upcall->packet);
 }
@@ -1822,23 +1834,36 @@ handle_miss_upcall(struct ofproto_dpif *ofproto, struct dpif_upcall *upcall)
 }
 
 static void
-handle_upcall(struct ofproto_dpif *ofproto, struct dpif_upcall *upcall)
+handle_userspace_upcall(struct ofproto_dpif *ofproto,
+                        struct dpif_upcall *upcall)
 {
     struct flow flow;
+    struct user_action_cookie cookie;
 
-    switch (upcall->type) {
-    case DPIF_UC_ACTION:
-        COVERAGE_INC(ofproto_dpif_ctlr_action);
-        odp_flow_key_to_flow(upcall->key, upcall->key_len, &flow);
-        send_packet_in(ofproto, upcall, &flow, false);
-        break;
+    memcpy(&cookie, &upcall->userdata, sizeof(cookie));
 
-    case DPIF_UC_SAMPLE:
+    if (cookie.type == USER_ACTION_COOKIE_SFLOW) {
         if (ofproto->sflow) {
             odp_flow_key_to_flow(upcall->key, upcall->key_len, &flow);
-            dpif_sflow_received(ofproto->sflow, upcall, &flow);
+            dpif_sflow_received(ofproto->sflow, upcall->packet, &flow, &cookie);
         }
         ofpbuf_delete(upcall->packet);
+
+    } else if (cookie.type == USER_ACTION_COOKIE_CONTROLLER) {
+        COVERAGE_INC(ofproto_dpif_ctlr_action);
+        odp_flow_key_to_flow(upcall->key, upcall->key_len, &flow);
+        send_packet_in(ofproto, upcall, &flow, false);
+    } else {
+        VLOG_WARN_RL(&rl, "invalid user cookie : 0x%"PRIx64, upcall->userdata);
+    }
+}
+
+static void
+handle_upcall(struct ofproto_dpif *ofproto, struct dpif_upcall *upcall)
+{
+    switch (upcall->type) {
+    case DPIF_UC_ACTION:
+        handle_userspace_upcall(ofproto, upcall);
         break;
 
     case DPIF_UC_MISS:
@@ -2174,39 +2199,43 @@ execute_odp_actions(struct ofproto_dpif *ofproto, const struct flow *flow,
                     const struct nlattr *odp_actions, size_t actions_len,
                     struct ofpbuf *packet)
 {
+    struct odputil_keybuf keybuf;
+    struct ofpbuf key;
+    int error;
+
     if (actions_len == NLA_ALIGN(NLA_HDRLEN + sizeof(uint64_t))
         && odp_actions->nla_type == OVS_ACTION_ATTR_USERSPACE) {
-        /* As an optimization, avoid a round-trip from userspace to kernel to
-         * userspace.  This also avoids possibly filling up kernel packet
-         * buffers along the way. */
+        const struct user_action_cookie *cookie;
         struct dpif_upcall upcall;
 
-        upcall.type = DPIF_UC_ACTION;
-        upcall.packet = packet;
-        upcall.key = NULL;
-        upcall.key_len = 0;
-        upcall.userdata = nl_attr_get_u64(odp_actions);
-        upcall.sample_pool = 0;
-        upcall.actions = NULL;
-        upcall.actions_len = 0;
-
-        send_packet_in(ofproto, &upcall, flow, false);
-
-        return true;
-    } else {
-        struct odputil_keybuf keybuf;
-        struct ofpbuf key;
-        int error;
+        cookie = nl_attr_get_unspec(odp_actions, sizeof(*cookie));
+        if (cookie->type == USER_ACTION_COOKIE_CONTROLLER) {
+            /* As an optimization, avoid a round-trip from userspace to kernel
+             * to userspace.  This also avoids possibly filling up kernel packet
+             * buffers along the way.
+             * This optimization does not work in case of sFlow is turned ON.
+             * Since first action would be sFlow SAMPLE action followed by
+             * Controller action. */
+
+            upcall.type = DPIF_UC_ACTION;
+            upcall.packet = packet;
+            upcall.key = NULL;
+            upcall.key_len = 0;
+            upcall.userdata = nl_attr_get_u64(odp_actions);
+
+            send_packet_in(ofproto, &upcall, flow, false);
+            return true;
+        }
+    }
 
-        ofpbuf_use_stack(&key, &keybuf, sizeof keybuf);
-        odp_flow_key_from_flow(&key, flow);
+    ofpbuf_use_stack(&key, &keybuf, sizeof keybuf);
+    odp_flow_key_from_flow(&key, flow);
 
-        error = dpif_execute(ofproto->dpif, key.data, key.size,
-                             odp_actions, actions_len, packet);
+    error = dpif_execute(ofproto->dpif, key.data, key.size,
+                         odp_actions, actions_len, packet);
 
-        ofpbuf_delete(packet);
-        return !error;
-    }
+    ofpbuf_delete(packet);
+    return !error;
 }
 
 /* Executes the actions indicated by 'facet' on 'packet' and credits 'facet''s
@@ -2910,6 +2939,8 @@ send_packet(struct ofproto_dpif *ofproto, uint32_t odp_port,
     odp_flow_key_from_flow(&key, &flow);
 
     ofpbuf_init(&odp_actions, 32);
+    compose_sflow_action(ofproto, &odp_actions, &flow, odp_port);
+
     nl_msg_put_u32(&odp_actions, OVS_ACTION_ATTR_OUTPUT, odp_port);
     error = dpif_execute(ofproto->dpif,
                          key.data, key.size,
@@ -2930,6 +2961,95 @@ static void do_xlate_actions(const union ofp_action *in, size_t n_in,
                              struct action_xlate_ctx *ctx);
 static void xlate_normal(struct action_xlate_ctx *);
 
+/* Compose SAMPLE action for sFlow. */
+static size_t
+compose_sflow_action(const struct ofproto_dpif *ofproto,
+                     struct ofpbuf *odp_actions,
+                     const struct flow *flow,
+                     uint32_t odp_port)
+{
+    uint32_t port_ifindex;
+    uint32_t probability;
+    struct user_action_cookie *cookie;
+    size_t sample_offset, actions_offset;
+    int user_cookie_offset, n_output;
+
+    if (!ofproto->sflow || flow->in_port == OFPP_NONE) {
+        return 0;
+    }
+
+    if (odp_port == OVSP_NONE) {
+        port_ifindex = 0;
+        n_output = 0;
+    } else {
+        port_ifindex = dpif_sflow_odp_port_to_ifindex(ofproto->sflow, odp_port);
+        n_output = 1;
+    }
+
+    sample_offset = nl_msg_start_nested(odp_actions, OVS_ACTION_ATTR_SAMPLE);
+
+    /* Number of packets out of UINT_MAX to sample. */
+    probability = dpif_sflow_get_probability(ofproto->sflow);
+    nl_msg_put_u32(odp_actions, OVS_SAMPLE_ATTR_PROBABILITY, probability);
+
+    actions_offset = nl_msg_start_nested(odp_actions, OVS_SAMPLE_ATTR_ACTIONS);
+
+    cookie = nl_msg_put_unspec_uninit(odp_actions, OVS_ACTION_ATTR_USERSPACE,
+						 sizeof(*cookie));
+    cookie->type = USER_ACTION_COOKIE_SFLOW;
+    cookie->data = port_ifindex;
+    cookie->n_output = n_output;
+    cookie->vlan_tci = 0;
+    user_cookie_offset = (char *) cookie - (char *) odp_actions->data;
+
+    nl_msg_end_nested(odp_actions, actions_offset);
+    nl_msg_end_nested(odp_actions, sample_offset);
+    return user_cookie_offset;
+}
+
+/* SAMPLE action must be first action in any given list of actions.
+ * At this point we do not have all information required to build it. So try to
+ * build sample action as complete as possible. */
+static void
+add_sflow_action(struct action_xlate_ctx *ctx)
+{
+    ctx->user_cookie_offset = compose_sflow_action(ctx->ofproto,
+                                                   ctx->odp_actions,
+                                                   &ctx->flow, OVSP_NONE);
+    ctx->sflow_odp_port = 0;
+    ctx->sflow_n_outputs = 0;
+}
+
+/* Fix SAMPLE action according to data collected while composing ODP actions.
+ * We need to fix SAMPLE actions OVS_SAMPLE_ATTR_ACTIONS attribute, i.e. nested
+ * USERSPACE action's user-cookie which is required for sflow. */
+static void
+fix_sflow_action(struct action_xlate_ctx *ctx)
+{
+    const struct flow *base = &ctx->base_flow;
+    struct user_action_cookie *cookie;
+
+    if (!ctx->user_cookie_offset) {
+        return;
+    }
+
+    cookie = ofpbuf_at(ctx->odp_actions, ctx->user_cookie_offset,
+                     sizeof(*cookie));
+    assert(cookie != NULL);
+    assert(cookie->type == USER_ACTION_COOKIE_SFLOW);
+
+    if (ctx->sflow_n_outputs) {
+        cookie->data = dpif_sflow_odp_port_to_ifindex(ctx->ofproto->sflow,
+                                                    ctx->sflow_odp_port);
+    }
+    if (ctx->sflow_n_outputs >= 255) {
+        cookie->n_output = 255;
+    } else {
+        cookie->n_output = ctx->sflow_n_outputs;
+    }
+    cookie->vlan_tci = base->vlan_tci;
+}
+
 static void
 commit_vlan_tci(struct action_xlate_ctx *ctx, ovs_be16 vlan_tci)
 {
@@ -3013,6 +3133,14 @@ commit_odp_actions(struct action_xlate_ctx *ctx)
 }
 
 static void
+compose_output_action(struct action_xlate_ctx *ctx, uint16_t odp_port)
+{
+    nl_msg_put_u32(ctx->odp_actions, OVS_ACTION_ATTR_OUTPUT, odp_port);
+    ctx->sflow_odp_port = odp_port;
+    ctx->sflow_n_outputs++;
+}
+
+static void
 add_output_action(struct action_xlate_ctx *ctx, uint16_t ofp_port)
 {
     const struct ofport_dpif *ofport = get_ofp_port(ctx->ofproto, ofp_port);
@@ -3032,7 +3160,7 @@ add_output_action(struct action_xlate_ctx *ctx, uint16_t ofp_port)
     }
 
     commit_odp_actions(ctx);
-    nl_msg_put_u32(ctx->odp_actions, OVS_ACTION_ATTR_OUTPUT, odp_port);
+    compose_output_action(ctx, odp_port);
     ctx->nf_output_iface = ofp_port;
 }
 
@@ -3113,8 +3241,7 @@ flood_packets(struct action_xlate_ctx *ctx, ovs_be32 mask)
     HMAP_FOR_EACH (ofport, up.hmap_node, &ctx->ofproto->up.ports) {
         uint16_t ofp_port = ofport->up.ofp_port;
         if (ofp_port != ctx->flow.in_port && !(ofport->up.opp.config & mask)) {
-            nl_msg_put_u32(ctx->odp_actions, OVS_ACTION_ATTR_OUTPUT,
-                           ofport->odp_port);
+            compose_output_action(ctx, ofport->odp_port);
         }
     }
 
@@ -3122,6 +3249,20 @@ flood_packets(struct action_xlate_ctx *ctx, ovs_be32 mask)
 }
 
 static void
+compose_controller_action(struct ofpbuf *odp_actions, int len)
+{
+    struct user_action_cookie cookie;
+
+    cookie.type = USER_ACTION_COOKIE_CONTROLLER;
+    cookie.data = len;
+    cookie.n_output = 0;
+    cookie.vlan_tci = 0;
+
+    nl_msg_put_unspec(odp_actions, OVS_ACTION_ATTR_USERSPACE,
+                                       &cookie, sizeof(cookie));
+}
+
+static void
 xlate_output_action__(struct action_xlate_ctx *ctx,
                       uint16_t port, uint16_t max_len)
 {
@@ -3147,7 +3288,7 @@ xlate_output_action__(struct action_xlate_ctx *ctx,
         break;
     case OFPP_CONTROLLER:
         commit_odp_actions(ctx);
-        nl_msg_put_u64(ctx->odp_actions, OVS_ACTION_ATTR_USERSPACE, max_len);
+        compose_controller_action(ctx->odp_actions, max_len);
         break;
     case OFPP_LOCAL:
         add_output_action(ctx, OFPP_LOCAL);
@@ -3515,7 +3656,9 @@ xlate_actions(struct action_xlate_ctx *ctx,
     if (process_special(ctx->ofproto, &ctx->flow, ctx->packet)) {
         ctx->may_set_up_flow = false;
     } else {
+        add_sflow_action(ctx);
         do_xlate_actions(in, n_in, ctx);
+        fix_sflow_action(ctx);
     }
 
     /* Check with in-band control to see if we're allowed to set up this
@@ -3847,8 +3990,7 @@ compose_actions(struct action_xlate_ctx *ctx, uint16_t vlan,
         if (dst->vid != initial_vid) {
             continue;
         }
-        nl_msg_put_u32(ctx->odp_actions,
-                       OVS_ACTION_ATTR_OUTPUT, dst->port->odp_port);
+        compose_output_action(ctx, dst->port->odp_port);
     }
 
     /* Then output the rest. */
@@ -3869,8 +4011,7 @@ compose_actions(struct action_xlate_ctx *ctx, uint16_t vlan,
 
             cur_vid = dst->vid;
         }
-        nl_msg_put_u32(ctx->odp_actions,
-                       OVS_ACTION_ATTR_OUTPUT, dst->port->odp_port);
+        compose_output_action(ctx, dst->port->odp_port);
     }
 
     dst_set_free(&set);