From 61b67a3094bc0769716f724c2f98d3de70e81f62 Mon Sep 17 00:00:00 2001
From: Max Rottenkolber <max@mr.gy>
Date: Thu, 22 Sep 2022 12:36:50 +0200
Subject: [PATCH] Replace ipfix probe with probe_tree, remove probe_rss

---
 src/program/ipfix/probe/README                | 103 +---
 .../snabbflow.conf => probe/example.conf}     |   0
 src/program/ipfix/probe/probe.lua             | 560 ++++++++++--------
 src/program/ipfix/probe_ptree/probe_ptree.lua | 354 -----------
 src/program/ipfix/probe_rss/README.inc        |  28 -
 src/program/ipfix/probe_rss/probe_rss.lua     | 447 --------------
 src/program/ipfix/stats/README.inc            |   5 -
 src/program/ipfix/stats/stats.lua             | 326 ----------
 8 files changed, 334 insertions(+), 1489 deletions(-)
 rename src/program/ipfix/{probe_ptree/snabbflow.conf => probe/example.conf} (100%)
 delete mode 100644 src/program/ipfix/probe_ptree/probe_ptree.lua
 delete mode 100644 src/program/ipfix/probe_rss/README.inc
 delete mode 100644 src/program/ipfix/probe_rss/probe_rss.lua
 delete mode 100644 src/program/ipfix/stats/README.inc
 delete mode 100644 src/program/ipfix/stats/stats.lua

diff --git a/src/program/ipfix/probe/README b/src/program/ipfix/probe/README
index 7eec821dba..809078e8cb 100644
--- a/src/program/ipfix/probe/README
+++ b/src/program/ipfix/probe/README
@@ -1,100 +1,11 @@
-Usage: snabb ipfix probe [options] <input> <output> [<template> ...]
+Usage: snabb ipfix probe [options] <config>
        snabb ipfix probe --help
 
 Available options:
-       -m, --host-mac <macaddr> The MAC address of the host running the exporter.
-                                Defaults to a randomly generated MAC.
-       -a, --host-ip  <ipaddr>  The IP address of the host running the exporter.
-                                Defaults to 10.0.0.1.
-       -c, --collector <ipaddr> The IP address of the flow collector.
-                                Defaults to 10.0.0.2.
-       -i, --input-type <type>  One of pcap, raw, tap, or pci. Specifies
-                                how to interpret the <input> argument. Interprets
-                                <input> as a pcap file path, device name, or
-                                PCI address. Defaults to pci.
-       -o, --output-type <type> Takes the same values as -i with the addition of
-                                tap_routed, but for <output>.  For tap_routed, <output>
-                                is the name of the device optionally followed by the
-                                MTU (including the Ethernet header), separated by ':'.
-                                With RSS and output type "tap_routed", each instance of
-                                the exporter uses a unique tap interface by appending the
-                                instance's observation domain to the name.
-       -D, --duration <seconds> Duration to run (in seconds).
-       -p, --port <port>        The port on the collector to send to. Defaults
-                                to port 4739.
-       --rss-queues <nqueues>   If <type> is "pci", create <nqueues> instances of the
-                                exporter, each of which attaches to an RSS queue of the
-                                device.  Defaults to 1. Ignored for all other input types.
-       --mtu <mtu>              The MTU in bytes including the 14-byte Ethernet header
-                                used when determining the size of export packets. If the
-                                output type is tap_routed, the MTU is applied to the
-                                tap interface as well.  Defaults to 1514.
-       --active-timeout <secs>  The timeout period for flows that are still receiving
-                                packets (in seconds).
-       --idle-timeout <secs>    The timeout period for flows with no recent activity
-                                (in seconds).
-       --flush-timeout <secs>   The timeout period after which an export packet is
-                                sent even if it is not of maximum possible size (limited
-                                by the MTU). Defaults to 10 seconds.
-       --scan-time <secs>       The number of seconds it takes for a scan of the flow
-                                cache to find flows eligible for export of each configured
-                                template. Defaults to 10.
-       --cache-size <number>    The initial size of the flow caches of all configured
-                                templates in number of flows.  Defaults to 20000.
-       --observation-domain <domain>
-                                The observation domain to be used for exported IPFIX
-                                flows. Defaults to 256.  With RSS, each instance of the
-                                exporter is configured with a unique observation domain by
-                                adding the number of the RSS queue (in the range 0..<nqueues>)
-                                to <domain>.
-       --template-refresh <seconds>
-                                The interval at which template records are sent in
-                                seconds. Defaults to 600.
-       --pfx4-to-as <file>      Name of the file that contains mappings from IPv4 prefixes
-                                to AS numbers from RIR data (used by extended templates)
-       --pfx6-to-as <file>      Name of the file that contains mappings from IPv6 prefixes
-                                to AS numbers from RIR data (used by extended templates)
-       --vlan-to-ifindex <file> Name of the file that contains mappings from VLAN
-                                tags to interface indices (used by extended templates)
-       --mac-to-as <file>       Name of the file that contains mappings from MAC
-                                addresses to adjacent AS numbers (used by extended templates)
-       --maps-log <file>        Name of the file to which extended templates log
-                                map-related messages.  If not specified, logging is
-                                disabled.
-       --netflow-v9             Use the Netflow V9 protocol to communicate to the
-                                flow collector.
-       --ipfix                  Use IPFIX to communicate with the flow collector.
-                                (mutually exclusive with --netflow-v9, --ipfix is
-                                 the default)
-       --transport <proto>      The transport protocol to use to communicate to
-                                the flow collector. Valid values depend on whether
-                                IPFIX or Netflow V9 is used.
-       --cpu <cpu>              Bind to the given CPU.  See the Snabb
-                                performance guide for more details.
-       -b, --busy-wait          Enable polling for arriving packets.  By default, the system
-                                uses sleep(3) during periods of little or no traffic to
-                                reduce CPU usage.
-       -j CMD, --jit CMD        Control LuaJIT behavior. Available commands:
-                                  -jv=FILE, --jit v=FILE
-                                    Write verbose JIT trace output to FILE.
-                                  -jdump=OPTS[,FILE] --jit dump=OPTS[,FILE]
-                                    Output JIT traces, optionally to a file.
-                                  -jp=OPTS[,FILE] --jit p=OPTS[,FILE]
-                                    Profile execution with low-overhead sampling.
-                                  -jopt=OPT
-                                    Sets JIT option OPT.
-                                  -jtprof
-                                    Profile execution with the traceprof library.
-                                See luajit documentation for more information:
-                                  http://luajit.org/running.html
+       -n, --name               Name of the Snabbflow instance.
+       -h, --help               Print this help text and exit.
 
-The mandatory arguments <input> and <output> are followed by an
-ordered list of template specifications in the form
-<name>[:<cache-size>], where <name> must be the name of a valid
-template defined in apps.ipfix.template.lua and <cache-size> is the
-optional value of the initial size of the flow cache for the template
-in units of the number of flows.  The templates are processed in the
-given order when incoming packets are processed.  The packet is passed
-to the first matching template only.
-
-The default is "v4 v6".
+The mandatory argument <config> must be a path to the initial
+configuration file.
+Refer to the schema in lib/yang/snabb-snabbflow-v1.yang.
+There is an example configuration in program/ipfix/probe/example.conf.
\ No newline at end of file
diff --git a/src/program/ipfix/probe_ptree/snabbflow.conf b/src/program/ipfix/probe/example.conf
similarity index 100%
rename from src/program/ipfix/probe_ptree/snabbflow.conf
rename to src/program/ipfix/probe/example.conf
diff --git a/src/program/ipfix/probe/probe.lua b/src/program/ipfix/probe/probe.lua
index b3ed3f938a..5f9e47cc9e 100644
--- a/src/program/ipfix/probe/probe.lua
+++ b/src/program/ipfix/probe/probe.lua
@@ -1,270 +1,364 @@
--- This module implements the `snabb flow_export` command
-
 module(..., package.seeall)
 
-local lib       = require("core.lib")
-local worker    = require("core.worker")
+local yang = require("lib.yang.yang")
+local yang_util = require("lib.yang.util")
+local path_data = require("lib.yang.path_data")
+local mem = require("lib.stream.mem")
+local ptree = require("lib.ptree.ptree")
+local cpuset = require("lib.cpuset")
+local pci = require("lib.hardware.pci")
+local lib = require("core.lib")
 local app_graph = require("core.config")
-local pci       = require("lib.hardware.pci")
-local probe     = require("program.ipfix.lib")
-local S         = require("syscall")
+
+local probe = require("program.ipfix.lib")
+
+local probe_schema = 'snabb-snabbflow-v1'
+
+local usage = require("program.ipfix.probe.README_inc")
 
 local long_opts = {
    help = "h",
-   jit = "j",
-   duration = "D",
-   port = "p",
-   transport = 1,
-   ["host-ip"] = "a",
-   ["input-type"] = "i",
-   ["output-type"] = "o",
-   ["mtu"] = 1,
-   ["rss-queues"] = 1,
-   ["netflow-v9"] = 0,
-   ["ipfix"] = 0,
-   ["active-timeout"] = 1,
-   ["idle-timeout"] = 1,
-   ["observation-domain"] = 1,
-   ["template-refresh"] = 1,
-   ["flush-timeout"] = 1,
-   ["cache-size"] = 1,
-   ["scan-time"] = 1,
-   ["pfx4-to-as"] = 1,
-   ["pfx6-to-as"] = 1,
-   ["maps-log"] = 1,
-   ["vlan-to-ifindex"] = 1,
-   ["mac-to-as"] = 1,
-   ["cpu"] = 1,
-   ["busy-wait"] = "b"
+   name = "n"
 }
+local opt = "hn:"
+local opt_handler = {}
+local name
+function opt_handler.n (arg) name = arg end
+function opt_handler.h () print(usage) main.exit(0) end
 
 function run (args)
-   local duration
-   local busywait = false
-   local jit = { opts = {} }
-
-   local input_type, output_type = "pci", "pci"
-
-   local host_mac
-   local host_ip = '10.0.0.1' -- Just to have a default.
-   local collector_ip = '10.0.0.2' -- Likewise.
-   local port = 4739
-   local mtu = 1514
-   local nqueues = 1
-
-   local active_timeout, idle_timeout, flush_timeout, scan_time
-   local observation_domain, template_refresh_interval
-   local cache_size
-   local ipfix_version = 10
-   local templates = {}
-   local maps = {}
-   local maps_logfile
-
-   local cpu
-
-   -- TODO: better input validation
-   local opt = {
-      h = function (arg)
-         print(require("program.ipfix.probe.README_inc"))
-         main.exit(0)
-      end,
-      D = function (arg)
-         duration = assert(tonumber(arg), "expected number for duration")
-      end,
-      i = function (arg)
-         input_type = arg
-      end,
-      o = function (arg)
-         output_type = arg
-      end,
-      p = function (arg)
-         port = assert(tonumber(arg), "expected number for port")
-      end,
-      m = function (arg)
-         host_mac = arg
-      end,
-      a = function (arg)
-         host_ip = arg
-      end,
-      c = function (arg)
-         collector_ip = arg
-      end,
-      b = function (arg)
-         busywait = true
-      end,
-      ["active-timeout"] = function (arg)
-         active_timeout =
-            assert(tonumber(arg), "expected number for active timeout")
-      end,
-      ["idle-timeout"] = function (arg)
-         idle_timeout =
-            assert(tonumber(arg), "expected number for idle timeout")
-      end,
-      ["flush-timeout"] = function (arg)
-         flush_timeout =
-            assert(tonumber(arg), "expected number for flush timeout")
-      end,
-      ["scan-time"] = function (arg)
-         scan_time =
-            assert(tonumber(arg), "expected number for scan time")
-      end,
-      ["observation-domain"] = function (arg)
-         observation_domain =
-            assert(tonumber(arg), "expected number for observation domain")
-      end,
-      ["template-refresh"] = function (arg)
-         template_refresh_interval =
-            assert(tonumber(arg), "expected number for template refresh interval")
-      end,
-      ["cache-size"] = function (arg)
-         cache_size =
-            assert(tonumber(arg), "expected number for cache size")
-      end,
-      ["pfx4-to-as"] = function (arg)
-         if arg then
-            maps.pfx4_to_as = arg
-         end
-      end,
-      ["pfx6-to-as"] = function (arg)
-         if arg then
-            maps.pfx6_to_as = arg
-         end
-      end,
-      ["vlan-to-ifindex"] = function (arg)
-         if arg then
-            maps.vlan_to_ifindex = arg
-         end
-      end,
-      ["mac-to-as"] = function (arg)
-         if arg then
-            maps.mac_to_as = arg
+   args = lib.dogetopt(args, opt_handler, opt, long_opts)
+   if #args ~= 1 then
+      print(usage)
+      main.exit(1)
+   end
+   local confpath = args[1]
+   local manager = start(name, confpath)
+   manager:main()
+end
+
+local probe_cpuset = cpuset.new()
+
+local function update_cpuset (cpu_pool)
+   local cpu_set = {}
+   if cpu_pool then
+      for _, cpu in ipairs(cpu_pool.cpu) do
+         if not probe_cpuset:contains(cpu) then
+            probe_cpuset:add(cpu)
          end
-      end,
-      ["maps-log"] = function (arg)
-         if arg ~= "" then
-            maps_logfile = arg
+         cpu_set[cpu] = true
+      end
+      for _, cpu in ipairs(probe_cpuset:list()) do
+         if not cpu_set[cpu] then
+            probe_cpuset:remove(cpu)
          end
-      end,
-      ipfix = function (arg)
-         ipfix_version = 10
-      end,
-      ["netflow-v9"] = function (arg)
-         ipfix_version = 9
-      end,
-      ["mtu"] = function (arg)
-         mtu = tonumber(arg)
-      end,
-      ["rss-queues"] = function (arg)
-         nqueues = tonumber(arg)
-      end,
-      -- TODO: not implemented
-      ["transport"] = function (arg) end,
-      ["cpu"] = function (arg)
-         cpu = tonumber(arg)
-      end,
-      j = require("program.ipfix.lib").parse_jit_option_fn(jit)
+      end
+   end
+end
+
+local function warn (msg, ...)
+   io.stderr:write("Warning: "..msg:format(...).."\n")
+   io.stderr:flush()
+end
+
+function start (name, confpath)
+   local conf = yang.load_configuration(confpath, {schema_name=probe_schema})
+   update_cpuset(conf.snabbflow_config.rss.cpu_pool)
+   return ptree.new_manager{
+      log_level = 'INFO',
+      setup_fn = setup_workers,
+      initial_configuration = conf,
+      schema_name = probe_schema,
+      cpuset = probe_cpuset,
+      name = name,
+      worker_default_scheduling = {
+         busywait = false,
+         jit_opt = {
+            sizemcode=256,
+            maxmcode=8192,
+            maxtrace=8000,
+            maxrecord=50000,
+            maxsnap=20000,
+            maxside=10000
+         }
+      },
    }
+end
 
-   args = lib.dogetopt(args, opt, "hD:i:o:p:m:a:c:j:b", long_opts)
-   if #args < 2 then
-      print(require("program.ipfix.probe.README_inc"))
-      main.exit(1)
-   elseif #args == 2 then
-      table.insert(args, 'v4')
-      table.insert(args, 'v6')
+local ipfix_default_config = lib.deepcopy(probe.probe_config)
+for _, key in ipairs({
+      "collector_ip",
+      "collector_port",
+      "observation_domain",
+      "exporter_mac",
+      "templates",
+      "output_type",
+      "output",
+      "input_type",
+      "input",
+      "instance"
+}) do
+   ipfix_default_config[key] = nil
+end
+
+local software_scaling_parser = path_data.parser_for_schema_by_name(
+   probe_schema, '/snabbflow-config/rss/software-scaling/exporter[name=""]'
+)
+local default_software_scaling =
+   software_scaling_parser(mem.open_input_string(''))
+
+function setup_workers (config)
+   local interfaces = config.snabbflow_config.interface
+   local rss = config.snabbflow_config.rss
+   local flow_director = config.snabbflow_config.flow_director
+   local ipfix = config.snabbflow_config.ipfix
+
+   local collector_pools = {}
+   for name, p in pairs(ipfix.collector_pool) do
+      local collectors = {}
+      for entry in p.collector:iterate() do
+         table.insert(collectors, {
+            ip = yang_util.ipv4_ntop(entry.key.ip),
+            port = entry.key.port
+         })
+      end
+      collector_pools[name] = collectors
    end
 
-   local input, output = args[1], args[2]
+   local function select_collector (pool)
+      -- Select the collector ip and port from the front of the
+      -- pool and rotate the pool's elements by one
+      assert(collector_pools[pool] and #collector_pools[pool] > 0,
+               "Undefined or empty collector pool: "..pool)
+      local collector = table.remove(collector_pools[pool], 1)
+      table.insert(collector_pools[pool], collector)
+      return collector
+   end
 
-   for i = 3, #args do
-      table.insert(templates, args[i])
+   if flow_director.default_class.exporter then
+      assert(not flow_director.class[flow_director.default_class.exporter],
+         "Exporter for the default traffic class can not be the exporter for a defined class.")
    end
 
-   local probe_config = {
-      active_timeout = active_timeout,
-      idle_timeout = idle_timeout,
-      flush_timeout = flush_timeout,
-      cache_size = cache_size,
-      scan_time = scan_time,
-      observation_domain = observation_domain,
-      template_refresh_interval = template_refresh_interval,
-      ipfix_version = ipfix_version,
-      exporter_ip = host_ip,
-      exporter_mac = host_mac,
-      collector_ip = collector_ip,
-      collector_port = port,
-      mtu = mtu,
-      templates = templates,
-      maps = maps,
-      maps_logfile = maps_logfile,
-      output_type = output_type,
-      output = output,
-      input_type = input_type,
-      input = input
-   }
+   local class_order = {}
+   for exporter in pairs(flow_director.class) do
+      table.insert(class_order, exporter)
+   end
+   table.sort(class_order, function (x, y)
+      return flow_director.class[x].order < flow_director.class[y].order
+   end)
+
+   local function class_name (exporter, class)
+      -- Including order in name to avoid name collision with 'default' class
+      return ("%s_%d"):format(exporter, class.order)
+   end
 
-   local mellanox_qs
-   if input_type == "pci" then
-      local device_info = pci.device_info(input)
-      if device_info.driver == 'apps.mellanox.connectx' then
-         mellanox_qs = {}
+   local rss_links = {}
+   local function rss_link_name (class)
+      if not rss_links[class] then
+         rss_links[class] = 0
       end
+      local qno = rss_links[class] + 1
+      rss_links[class] = qno
+      return class.."_"..qno
+   end
+
+   local observation_domain = ipfix.observation_domain_base
+   local function next_observation_domain ()
+      local ret = observation_domain
+      observation_domain = observation_domain + 1
+      return ret
    end
+   
+   local workers = {}
+   local worker_opts = {}
 
-   for rssq = 0, nqueues - 1 do
+   local mellanox = {}
 
-      local jit_c = lib.deepcopy(jit)
+   update_cpuset(rss.cpu_pool)
 
-      probe_config.input = input.."/"..rssq
-      if mellanox_qs then
-         table.insert(mellanox_qs, { id = rssq })
+   local function ensure_device_unique (device, interfaces)
+      for other in pairs(interfaces) do
+         if device ~= other then
+            if pci.qualified(device) == pci.qualified(other) then
+               error("Duplicate interfaces: "..device..", "..other..
+                     "\nNot applying configuration. Remove one of them via"..
+                     ("\n  snabb config remove <snabbflow> /snabbflow-config/interface[device=%q]")
+                        :format(other))
+            end
+         end
       end
+   end
 
-      if nqueues > 1 then
-         -- Create unique identifiers for config options that need to
-         -- be unique per instance
-         probe_config.observation_domain = observation_domain + rssq
-         if output_type == "tap_routed" then
-            probe_config.output = output..probe_config.observation_domain
-         end
-         if jit_c.dump and #jit_c.dump == 2 then
-            jit_c.dump[2] = jit_c.dump[2]..rssq
+   for rss_group = 1, rss.hardware_scaling.rss_groups do
+      local inputs, outputs = {}, {}
+      for device, opt in pairs(interfaces) do
+         ensure_device_unique(device, interfaces)
+         local input = lib.deepcopy(opt)
+         input.device = device
+         input.rxq = rss_group - 1
+         table.insert(inputs, input)
+
+         -- The mellanox driver requires a master process that sets up
+         -- all queues for the interface. We collect all queues per
+         -- device of this type here.
+         local device_info = pci.device_info(device)
+         if device_info.driver == 'apps.mellanox.connectx' then
+            local spec = mellanox[device]
+            if not spec then
+               spec = { ifName = input.name,
+                        ifAlias = input.description,
+                        queues = {},
+                        recvq_size = input.receive_queue_size }
+               mellanox[device] = spec
+            end
+            table.insert(spec.queues, { id = input.rxq })
+         else
+            -- Silently truncate receive-queue-size for other drivers.
+            -- (We are not sure what they can handle.)
+            input.receive_queue_size = math.min(input.receive_queue_size, 8192)
          end
       end
 
-      local worker_expr = string.format(
-         'require("program.ipfix.lib").run(%s, %s, %s, nil, %s)',
-         probe.value_to_string(probe_config), tostring(duration),
-         tostring(busywait), probe.value_to_string(jit_c)
-      )
-      local child_pid = worker.start("ipfix"..rssq, worker_expr)
-      print("Launched IPFIX worker process #"..child_pid)
-   end
+      for name, exporter in pairs(ipfix.exporter) do
+         local config = {}
+         for key in pairs(ipfix_default_config) do
+            config[key] = ipfix[key]
+         end
+         config.exporter_ip = yang_util.ipv4_ntop(ipfix.exporter_ip)
+
+         config.collector_pool = exporter.collector_pool
+         config.templates = exporter.template
+
+         config.output_type = "tap_routed"
+         config.add_packet_metadata = false
+
+         config.maps = {}
+         for name, map in pairs(ipfix.maps) do
+            config.maps[name] = map.file
+         end
+
+         local software_scaling = (rss.software_scaling.exporter and
+                                    rss.software_scaling.exporter[name])
+                               or default_software_scaling
+         local num_instances = 1
+         if not software_scaling.embed then
+            num_instances = software_scaling.instances
+         end
+         for i = 1, num_instances do
+            -- Create a clone of the configuration for parameters
+            -- specific to the instance
+            local iconfig = lib.deepcopy(config)
+
+            -- This is used to disambiguate multiple instances of the
+            -- ipfix app (possibly using multiple instances of the same
+            -- template) within a single worker.
+            iconfig.instance = name.."_"..i
+
+            local rss_link
+            local class = flow_director.class[name]
+            if class then
+               rss_link = rss_link_name(class_name(name, class))
+            elseif name == flow_director.default_class.exporter then
+               rss_link = rss_link_name('default')
+            else
+               -- No traffic class configured for exporter, do not create
+               -- instances.
+               warn("No traffic class configured for exporter '%s'.", name)
+               break
+            end
 
-   local ctrl_graph
-   if mellanox_qs then
-      ctrl_graph = probe.configure_mlx_ctrl_graph{
-         [input] = {
-            queues = mellanox_qs,
-            recvq_size = 8192,
-            ifName = (input:gsub("[:%.]", "_"))
+            local collector = select_collector(config.collector_pool)
+            iconfig.collector_ip = collector.ip
+            iconfig.collector_port = collector.port
+            iconfig.collector_pool = nil
+
+            iconfig.log_date = ipfix.log_date
+            local od = next_observation_domain()
+            iconfig.observation_domain = od
+            iconfig.output = "ipfixexport"..od
+            if ipfix.maps.log_directory then
+               iconfig.maps_logfile =
+                  ipfix.maps.log_directory.."/"..od..".log"
+            end
+
+            -- Scale the scan protection parameters by the number of
+            -- ipfix instances in this RSS class
+            local scale_factor = rss.hardware_scaling.rss_groups * num_instances
+            iconfig.scan_protection = {
+               enable = ipfix.scan_protection.enable,
+               threshold_rate = ipfix.scan_protection.threshold_rate / scale_factor,
+               export_rate = ipfix.scan_protection.export_rate / scale_factor,
+            }
+
+            local output
+            if software_scaling.embed then
+               output = {
+                  link_name = rss_link,
+                  args = iconfig
+               }
+            else
+               output = { type = "interlink", link_name = rss_link }
+               iconfig.input_type = "interlink"
+               iconfig.input = rss_link
+               
+
+               workers[rss_link] = probe.configure_graph(iconfig)
+               -- Dedicated exporter processes are restartable
+               worker_opts[rss_link] = {
+                  restart_intensity = software_scaling.restart.intensity,
+                  restart_period = software_scaling.restart.period
+               }
+            end
+            table.insert(outputs, output)
+         end
+      end
+
+      local rss_config = {
+         default_class = flow_director.default_class.exporter ~= nil,
+         classes = {},
+         remove_extension_headers = flow_director.remove_ipv6_extension_headers
       }
-   }
+      for _, exporter in ipairs(class_order) do
+         local class = flow_director.class[exporter]
+         if not ipfix.exporter[exporter] then
+            error(("Exporter '%s' referenced in traffic class %d is not defined.")
+               :format(exporter, class.order))
+         end
+         table.insert(rss_config.classes, {
+            name = class_name(exporter, class),
+            filter = class.filter,
+            continue = class.continue
+         })
+      end
+      workers["rss"..rss_group] = probe.configure_rss_graph(
+         rss_config, inputs, outputs, ipfix.log_date, rss_group
+      )
    end
 
-   engine.busywait = false
-   engine.Hz = 10
-   engine.configure(ctrl_graph)
+   -- Create a trivial app graph that only contains the control apps
+   -- for the Mellanox driver, which sets up the queues and
+   -- maintains interface counters.
+   local ctrl_graph, need_ctrl = probe.configure_mlx_ctrl_graph(mellanox, ipfix.log_date)
 
-   engine.main({ duration = duration })
+   if need_ctrl then
+      workers["mlx_ctrl"] = ctrl_graph
+      worker_opts["mlx_ctrl"] = {acquire_cpu=false}
+   end
 
-   print("Waiting for workers to finish")
-   local alive
-   repeat
-      alive = false
-      for _, s in pairs(worker.status()) do
-         if s.alive then alive = true end
+   if false then -- enable to debug
+      for name, graph in pairs(workers) do
+         print("worker", name)
+         print("", "apps:")
+         for name, _ in pairs(graph.apps) do
+            print("", "", name)
+         end
+         print("", "links:")
+         for spec in pairs(graph.links) do
+            print("", "", spec)
+         end
       end
-   until not alive
+   end
 
-end
+   return workers, worker_opts
+end
\ No newline at end of file
diff --git a/src/program/ipfix/probe_ptree/probe_ptree.lua b/src/program/ipfix/probe_ptree/probe_ptree.lua
deleted file mode 100644
index cbc9cfb975..0000000000
--- a/src/program/ipfix/probe_ptree/probe_ptree.lua
+++ /dev/null
@@ -1,354 +0,0 @@
-module(..., package.seeall)
-
-local yang = require("lib.yang.yang")
-local yang_util = require("lib.yang.util")
-local path_data = require("lib.yang.path_data")
-local mem = require("lib.stream.mem")
-local ptree = require("lib.ptree.ptree")
-local cpuset = require("lib.cpuset")
-local pci = require("lib.hardware.pci")
-local lib = require("core.lib")
-local app_graph = require("core.config")
-
-local probe = require("program.ipfix.lib")
-
-local probe_schema = 'snabb-snabbflow-v1'
-
-local probe_cpuset = cpuset.new()
-
-local function warn (msg, ...)
-   io.stderr:write("Warning: "..msg:format(...).."\n")
-   io.stderr:flush()
-end
-
-function setup_ipfix (conf)
-   -- yang.print_config_for_schema_by_name(probe_schema, conf, io.stdout)
-   return setup_workers(conf)
-end
-
-function start (name, confpath)
-   local conf = yang.load_configuration(confpath, {schema_name=probe_schema})
-   update_cpuset(conf.snabbflow_config.rss.cpu_pool)
-   return ptree.new_manager{
-      log_level = 'INFO',
-      setup_fn = setup_ipfix,
-      initial_configuration = conf,
-      schema_name = probe_schema,
-      cpuset = probe_cpuset,
-      name = name,
-      worker_default_scheduling = {
-         busywait = false,
-         jit_opt = {
-            sizemcode=256,
-            maxmcode=8192,
-            maxtrace=8000,
-            maxrecord=50000,
-            maxsnap=20000,
-            maxside=10000
-         }
-      },
-   }
-end
-
-function run (args)
-   local name = assert(args[1])
-   local confpath = assert(args[2])
-   -- print("Confpath is:", confpath)
-   local manager = start(name, confpath)
-   manager:main()
-end
-
-local ipfix_default_config = lib.deepcopy(probe.probe_config)
-for _, key in ipairs({
-      "collector_ip",
-      "collector_port",
-      "observation_domain",
-      "exporter_mac",
-      "templates",
-      "output_type",
-      "output",
-      "input_type",
-      "input",
-      "instance"
-}) do
-   ipfix_default_config[key] = nil
-end
-
-function update_cpuset (cpu_pool)
-   local cpu_set = {}
-   if cpu_pool then
-      for _, cpu in ipairs(cpu_pool.cpu) do
-         if not probe_cpuset:contains(cpu) then
-            probe_cpuset:add(cpu)
-         end
-         cpu_set[cpu] = true
-      end
-      for _, cpu in ipairs(probe_cpuset:list()) do
-         if not cpu_set[cpu] then
-            probe_cpuset:remove(cpu)
-         end
-      end
-   end
-end
-
-local software_scaling_parser = path_data.parser_for_schema_by_name(
-   probe_schema, '/snabbflow-config/rss/software-scaling/exporter[name=""]'
-)
-local default_software_scaling =
-   software_scaling_parser(mem.open_input_string(''))
-
-function setup_workers (config)
-   local interfaces = config.snabbflow_config.interface
-   local rss = config.snabbflow_config.rss
-   local flow_director = config.snabbflow_config.flow_director
-   local ipfix = config.snabbflow_config.ipfix
-
-   local collector_pools = {}
-   for name, p in pairs(ipfix.collector_pool) do
-      local collectors = {}
-      for entry in p.collector:iterate() do
-         table.insert(collectors, {
-            ip = yang_util.ipv4_ntop(entry.key.ip),
-            port = entry.key.port
-         })
-      end
-      collector_pools[name] = collectors
-   end
-
-   local function select_collector (pool)
-      -- Select the collector ip and port from the front of the
-      -- pool and rotate the pool's elements by one
-      assert(collector_pools[pool] and #collector_pools[pool] > 0,
-               "Undefined or empty collector pool: "..pool)
-      local collector = table.remove(collector_pools[pool], 1)
-      table.insert(collector_pools[pool], collector)
-      return collector
-   end
-
-   if flow_director.default_class.exporter then
-      assert(not flow_director.class[flow_director.default_class.exporter],
-         "Exporter for the default traffic class can not be the exporter for a defined class.")
-   end
-
-   local class_order = {}
-   for exporter in pairs(flow_director.class) do
-      table.insert(class_order, exporter)
-   end
-   table.sort(class_order, function (x, y)
-      return flow_director.class[x].order < flow_director.class[y].order
-   end)
-
-   local function class_name (exporter, class)
-      -- Including order in name to avoid name collision with 'default' class
-      return ("%s_%d"):format(exporter, class.order)
-   end
-
-   local rss_links = {}
-   local function rss_link_name (class)
-      if not rss_links[class] then
-         rss_links[class] = 0
-      end
-      local qno = rss_links[class] + 1
-      rss_links[class] = qno
-      return class.."_"..qno
-   end
-
-   local observation_domain = ipfix.observation_domain_base
-   local function next_observation_domain ()
-      local ret = observation_domain
-      observation_domain = observation_domain + 1
-      return ret
-   end
-   
-   local workers = {}
-   local worker_opts = {}
-
-   local mellanox = {}
-
-   update_cpuset(rss.cpu_pool)
-
-   local function ensure_device_unique (device, interfaces)
-      for other in pairs(interfaces) do
-         if device ~= other then
-            if pci.qualified(device) == pci.qualified(other) then
-               error("Duplicate interfaces: "..device..", "..other..
-                     "\nNot applying configuration. Remove one of them via"..
-                     ("\n  snabb config remove <snabbflow> /snabbflow-config/interface[device=%q]")
-                        :format(other))
-            end
-         end
-      end
-   end
-
-   for rss_group = 1, rss.hardware_scaling.rss_groups do
-      local inputs, outputs = {}, {}
-      for device, opt in pairs(interfaces) do
-         ensure_device_unique(device, interfaces)
-         local input = lib.deepcopy(opt)
-         input.device = device
-         input.rxq = rss_group - 1
-         table.insert(inputs, input)
-
-         -- The mellanox driver requires a master process that sets up
-         -- all queues for the interface. We collect all queues per
-         -- device of this type here.
-         local device_info = pci.device_info(device)
-         if device_info.driver == 'apps.mellanox.connectx' then
-            local spec = mellanox[device]
-            if not spec then
-               spec = { ifName = input.name,
-                        ifAlias = input.description,
-                        queues = {},
-                        recvq_size = input.receive_queue_size }
-               mellanox[device] = spec
-            end
-            table.insert(spec.queues, { id = input.rxq })
-         else
-            -- Silently truncate receive-queue-size for other drivers.
-            -- (We are not sure what they can handle.)
-            input.receive_queue_size = math.min(input.receive_queue_size, 8192)
-         end
-      end
-
-      for name, exporter in pairs(ipfix.exporter) do
-         local config = {}
-         for key in pairs(ipfix_default_config) do
-            config[key] = ipfix[key]
-         end
-         config.exporter_ip = yang_util.ipv4_ntop(ipfix.exporter_ip)
-
-         config.collector_pool = exporter.collector_pool
-         config.templates = exporter.template
-
-         config.output_type = "tap_routed"
-         config.add_packet_metadata = false
-
-         config.maps = {}
-         for name, map in pairs(ipfix.maps) do
-            config.maps[name] = map.file
-         end
-
-         local software_scaling = (rss.software_scaling.exporter and
-                                    rss.software_scaling.exporter[name])
-                               or default_software_scaling
-         local num_instances = 1
-         if not software_scaling.embed then
-            num_instances = software_scaling.instances
-         end
-         for i = 1, num_instances do
-            -- Create a clone of the configuration for parameters
-            -- specific to the instance
-            local iconfig = lib.deepcopy(config)
-
-            -- This is used to disambiguate multiple instances of the
-            -- ipfix app (possibly using multiple instances of the same
-            -- template) within a single worker.
-            iconfig.instance = name.."_"..i
-
-            local rss_link
-            local class = flow_director.class[name]
-            if class then
-               rss_link = rss_link_name(class_name(name, class))
-            elseif name == flow_director.default_class.exporter then
-               rss_link = rss_link_name('default')
-            else
-               -- No traffic class configured for exporter, do not create
-               -- instances.
-               warn("No traffic class configured for exporter '%s'.", name)
-               break
-            end
-
-            local collector = select_collector(config.collector_pool)
-            iconfig.collector_ip = collector.ip
-            iconfig.collector_port = collector.port
-            iconfig.collector_pool = nil
-
-            iconfig.log_date = ipfix.log_date
-            local od = next_observation_domain()
-            iconfig.observation_domain = od
-            iconfig.output = "ipfixexport"..od
-            if ipfix.maps.log_directory then
-               iconfig.maps_logfile =
-                  ipfix.maps.log_directory.."/"..od..".log"
-            end
-
-            -- Scale the scan protection parameters by the number of
-            -- ipfix instances in this RSS class
-            local scale_factor = rss.hardware_scaling.rss_groups * num_instances
-            iconfig.scan_protection = {
-               enable = ipfix.scan_protection.enable,
-               threshold_rate = ipfix.scan_protection.threshold_rate / scale_factor,
-               export_rate = ipfix.scan_protection.export_rate / scale_factor,
-            }
-
-            local output
-            if software_scaling.embed then
-               output = {
-                  link_name = rss_link,
-                  args = iconfig
-               }
-            else
-               output = { type = "interlink", link_name = rss_link }
-               iconfig.input_type = "interlink"
-               iconfig.input = rss_link
-               
-
-               workers[rss_link] = probe.configure_graph(iconfig)
-               -- Dedicated exporter processes are restartable
-               worker_opts[rss_link] = {
-                  restart_intensity = software_scaling.restart.intensity,
-                  restart_period = software_scaling.restart.period
-               }
-            end
-            table.insert(outputs, output)
-         end
-      end
-
-      local rss_config = {
-         default_class = flow_director.default_class.exporter ~= nil,
-         classes = {},
-         remove_extension_headers = flow_director.remove_ipv6_extension_headers
-      }
-      for _, exporter in ipairs(class_order) do
-         local class = flow_director.class[exporter]
-         if not ipfix.exporter[exporter] then
-            error(("Exporter '%s' referenced in traffic class %d is not defined.")
-               :format(exporter, class.order))
-         end
-         table.insert(rss_config.classes, {
-            name = class_name(exporter, class),
-            filter = class.filter,
-            continue = class.continue
-         })
-      end
-      workers["rss"..rss_group] = probe.configure_rss_graph(
-         rss_config, inputs, outputs, ipfix.log_date, rss_group
-      )
-   end
-
-   -- Create a trivial app graph that only contains the control apps
-   -- for the Mellanox driver, which sets up the queues and
-   -- maintains interface counters.
-   local ctrl_graph, need_ctrl = probe.configure_mlx_ctrl_graph(mellanox, ipfix.log_date)
-
-   if need_ctrl then
-      workers["mlx_ctrl"] = ctrl_graph
-      worker_opts["mlx_ctrl"] = {acquire_cpu=false}
-   end
-
-   if false then -- enable to debug
-      for name, graph in pairs(workers) do
-         print("worker", name)
-         print("", "apps:")
-         for name, _ in pairs(graph.apps) do
-            print("", "", name)
-         end
-         print("", "links:")
-         for spec in pairs(graph.links) do
-            print("", "", spec)
-         end
-      end
-   end
-
-   return workers, worker_opts
-end
\ No newline at end of file
diff --git a/src/program/ipfix/probe_rss/README.inc b/src/program/ipfix/probe_rss/README.inc
deleted file mode 100644
index c6e70cf483..0000000000
--- a/src/program/ipfix/probe_rss/README.inc
+++ /dev/null
@@ -1,28 +0,0 @@
-Usage:
-  snabb ipfix probe_rss [options] <config-file>
-
-Available options:
-       -h, --help               Display this help message
-       -D, --duration <seconds> Duration to run (in seconds).
-       -b, --busy-wait          Enable polling for arriving packets.  By default, the system
-                                uses sleep(3) during periods of little or no traffic to
-                                reduce CPU usage.
-       -L, --log-date           Include date and time in log messages. Off by default.
-       -c, --worker-check-interval
-                                Interval in seconds at which the master process checks if
-				its worker processes are still alive.  Dead workers are
-				restarted.
-       -j CMD, --jit CMD        Control LuaJIT behavior. Available commands:
-                                  -jv=FILE, --jit v=FILE
-                                    Write verbose JIT trace output to FILE.
-                                  -jdump=OPTS[,FILE] --jit dump=OPTS[,FILE]
-                                    Output JIT traces, optionally to a file.
-                                  -jp=OPTS[,FILE] --jit p=OPTS[,FILE]
-                                    Profile execution with low-overhead sampling.
-                                  -jopt=OPT
-                                    Sets JIT option OPT.
-                                  -jtprof
-                                    Profile execution with the traceprof library.
-                                 See luajit documentation for more information:
-                                   http://luajit.org/running.html
-       -d, --debug               Enable developer debug
diff --git a/src/program/ipfix/probe_rss/probe_rss.lua b/src/program/ipfix/probe_rss/probe_rss.lua
deleted file mode 100644
index 792c9e2dec..0000000000
--- a/src/program/ipfix/probe_rss/probe_rss.lua
+++ /dev/null
@@ -1,447 +0,0 @@
-module(...,package.seeall)
-
-local S           = require("syscall")
-local lib         = require("core.lib")
-local app_graph   = require("core.config")
-local worker      = require("core.worker")
-local shm         = require("core.shm")
-local timer       = require("core.timer")
-local pci         = require("lib.hardware.pci")
-local logger      = require("lib.logger")
-local numa        = require("lib.numa")
-local probe       = require("program.ipfix.lib")
-
-local main_config = {
-   interfaces = { required = true },
-   hw_rss_scaling = { default = 1 },
-   rss = { required = true },
-   cpu_pool = { default = {} },
-   rss_pin_cpu = { default = false },
-   rss_jit = { default = nil },
-   ipfix = { required = true }
-}
-local interface_config = {
-   device = { required = true },
-   name = { default = nil},
-   description = { default = nil },
-   tag = { default = nil },
-   config = { default = {} }
-}
-local ipfix_config = {
-   default = { default = {} },
-   collector_pools = { default = {} },
-   maps = { default = {} },
-   observation_domain_base = { default = 256 },
-   exporters = { required = true }
-}
-local collector_pool_config = {
-   ip = { required = true },
-   port = { required = true }
-}
-local maps_config = {
-   pfx4_to_as = { default = nil },
-   pfx6_to_as = { default = nil },
-   vlan_to_ifindex = { default = nil },
-   mac_to_as = { default = nil }
-}
-local exporter_config = {
-   templates = { required = true },
-   rss_class = { default = "default" },
-   collector_pool = { default = nil },
-   use_maps = { default = false },
-   maps_log_dir = { default = nil },
-   instances = { default = {} }
-}
-local instance_config = {
-   embed = { default = true },
-   weight = { default = 1 },
-   jit = { default = nil },
-   busywait = { default = nil },
-   pin_cpu = { default = false },
-}
-local jit_config = {
-   v = { default = nil },
-   p = { default = {} },
-   dump = { default = {} },
-   opts = { default = {} },
-   traceprof = { default = false }
-}
-local ipfix_default_config = lib.deepcopy(probe.probe_config)
-for _, key in ipairs({
-      "collector_ip",
-      "collector_port",
-      "observation_domain",
-      "exporter_mac",
-      "templates",
-      "output_type",
-      "output",
-      "input_type",
-      "input",
-      "instance"
-}) do
-   ipfix_default_config[key] = nil
-end
-
-local function override_jit(default, config_in, dump_suffix)
-   local jit = lib.deepcopy(default)
-   if config_in then
-      local config = lib.parse(config_in, jit_config)
-      if config.opts then
-         jit.opts = config.opts
-      end
-      if config.dump then
-         local dump = lib.deepcopy(config.dump)
-         local file = dump[2]
-         if file then
-            file = file.."_"..dump_suffix
-            dump[2] = file
-         end
-         jit.dump = dump
-      end
-   end
-   return jit
-end
-
-local function create_workers (probe_config, duration, busywait, jit, logger, log_date)
-   local main = lib.parse(probe_config, main_config)
-   local maps = lib.parse(main.ipfix.maps, maps_config)
-   local ipfix = lib.parse(main.ipfix, ipfix_config)
-   local ipfix_default = lib.parse(ipfix.default, ipfix_default_config)
-   local collector_pools = {}
-   for name, list in pairs(ipfix.collector_pools) do
-      pool = {}
-      for _, member in ipairs(list) do
-         table.insert(pool, lib.parse(member, collector_pool_config))
-      end
-      collector_pools[name] = pool
-   end
-
-   local function merge_with_default (config)
-      local merged = lib.deepcopy(ipfix_default)
-      for _, key in ipairs({ "collector_pool", "templates" }) do
-         if config[key] then
-            merged[key] = config[key]
-         end
-      end
-      return merged
-   end
-
-   local classes = {}
-   local function rss_link_name (class, weight)
-      if not classes[class] then
-         classes[class] = 0
-      end
-      local instance = classes[class] + 1
-      classes[class] = instance
-      return class.."_"..instance..(weight > 1 and "_"..weight or '')
-   end
-
-   local workers = {}
-   local function add_worker (name, expr, create_fn, remove_fn)
-      table.insert(workers,
-                   {
-                      name = name,
-                      expr = expr,
-                      create_fn = create_fn,
-                      remove_fn = remove_fn,
-                      restarts = 0
-                   }
-      )
-   end
-
-   assert(type(main.interfaces) == "table")
-
-   local mellanox = {}
-   local observation_domain = ipfix.observation_domain_base
-
-   -- Determine NUMA affinity for the input interfaces
-   local pci_addrs = {}
-   for _, interface in ipairs(main.interfaces) do
-      local input = lib.parse(interface, interface_config)
-      table.insert(pci_addrs, input.device)
-   end
-   local node = numa.choose_numa_node_for_pci_addresses(pci_addrs)
-   local cpu_pool_size = #main.cpu_pool
-   local function cpu_for_node (activate)
-      if not activate then return nil end
-      for n, cpu in ipairs(main.cpu_pool) do
-         local cpu_node =  numa.cpu_get_numa_node(cpu)
-         if cpu_node == node then
-            return table.remove(main.cpu_pool, n)
-         end
-      end
-      return nil
-   end
-   local function log_cpu_choice (pid, cpu, activate)
-      if cpu_pool_size == 0 or not activate then return end
-      if cpu then
-         logger:log(string.format("Binding #%d to CPU %d, "
-                                     .."NUMA node %d",
-                                  pid, cpu, node))
-      else
-         logger:log(string.format("Not binding #%d to any CPU "
-                                  .."(no match found in pool for "
-                                     .."NUMA node %d)", pid, node))
-      end
-   end
-
-   for rssq = 0, main.hw_rss_scaling - 1 do
-      local inputs, outputs = {}, {}
-      for i, interface in ipairs(main.interfaces) do
-         local input = lib.parse(interface, interface_config)
-         input.rxq = rssq
-         input.receive_queue_size = input.config.recvq_size
-         table.insert(inputs, input)
-
-         -- The mellanox driver requires a master process that sets up
-         -- all queues for the interface. We collect all queues per
-         -- device of this type here.
-         local device_info = pci.device_info(input.device)
-         if device_info.driver == 'apps.mellanox.connectx' then
-            local spec = mellanox[input.device]
-            if not spec then
-               spec = { ifName = input.name,
-                        ifAlias = input.description,
-                        queues = {},
-                        recvq_size = input.config.recvq_size or 8192 }
-               mellanox[input.device] = spec
-            end
-            table.insert(spec.queues, { id = rssq })
-         end
-      end
-
-      local embedded_instance = 1
-      for _, exporter in ipairs(main.ipfix.exporters) do
-         local exporter = lib.parse(exporter, exporter_config)
-         local config = merge_with_default(exporter)
-
-         config.output_type = "tap_routed"
-         config.instance = nil
-         config.add_packet_metadata = false
-         if exporter.use_maps then
-            config.maps = maps
-         end
-
-         for i, instance in ipairs(exporter.instances) do
-            -- Create a clone of the configuration for parameters
-            -- specific to the instance
-            local iconfig = lib.deepcopy(config)
-            local instance = lib.parse(instance, instance_config)
-            local rss_link = rss_link_name(exporter.rss_class, instance.weight)
-            local od = observation_domain
-
-            -- Select the collector ip and port from the front of the
-            -- pool and rotate the pool's elements by one
-            local pool = config.collector_pool
-            assert(collector_pools[pool] and #collector_pools[pool] > 0,
-                   "Undefined or empty collector pool: "..pool)
-            collector = table.remove(collector_pools[pool], 1)
-            table.insert(collector_pools[pool], collector)
-            iconfig.collector_ip = collector.ip
-            iconfig.collector_port = collector.port
-            iconfig.collector_pool = nil
-
-            iconfig.log_date = log_date
-            observation_domain = observation_domain + 1
-            iconfig.observation_domain = od
-            iconfig.output = "ipfixexport"..od
-            if exporter.maps_log_dir then
-               iconfig.maps_logfile =
-                  exporter.maps_log_dir.."/"..od..".log"
-            end
-
-            -- Scale the scan protection parameters by the number of
-            -- ipfix instances in this RSS class
-            iconfig.scan_protection = lib.deepcopy(config.scan_protection)
-            local scale_factor = main.hw_rss_scaling * #exporter.instances
-            for _, field in ipairs({ "threshold_rate", "export_rate" }) do
-               iconfig.scan_protection[field] =
-                  iconfig.scan_protection[field]/scale_factor
-            end
-
-            local output
-            if instance.embed then
-               output = {
-                  link_name = rss_link,
-                  args = iconfig,
-                  instance = embedded_instance
-               }
-               embedded_instance = embedded_instance + 1
-            else
-               output = { type = "interlink", link_name = rss_link }
-               iconfig.input_type = "interlink"
-               iconfig.input = rss_link
-
-               local jit =  override_jit(jit, instance.jit, od)
-
-               local cpu = cpu_for_node(instance.pin_cpu)
-               local worker_expr = string.format(
-                  'require("program.ipfix.lib").run(%s, %s, %s, %s, %s)',
-                  probe.value_to_string(iconfig), tostring(duration),
-                  tostring(instance.busywait), tostring(cpu),
-                  probe.value_to_string(jit)
-               )
-               add_worker(rss_link, worker_expr,
-                          function(pid)
-                             logger:log(string.format("Launched IPFIX worker process #%d, "..
-                                                         "observation domain %d",
-                                                      pid, od))
-                             logger:log(string.format("Selected collector %s:%d from pool %s "
-                                                      .."for process #%d ",
-                                                      collector.ip, collector.port, pool,
-                                                      pid))
-                             log_cpu_choice(pid, cpu, instance.pin_cpu)
-                             shm.create("ipfix_workers/"..pid, "uint64_t")
-                          end,
-                          function(pid)
-                             shm.unlink("ipfix_workers/"..pid)
-                          end
-               )
-            end
-            table.insert(outputs, output)
-         end
-      end
-
-      local cpu = cpu_for_node(main.rss_pin_cpu)
-      local worker_expr = string.format(
-         'require("program.ipfix.lib").run_rss(%s, %s, %s, %s, %s, %s, %s)',
-         probe.value_to_string(main.rss), probe.value_to_string(inputs),
-         probe.value_to_string(outputs), tostring(duration),
-         tostring(busywait), tostring(cpu),
-         probe.value_to_string(override_jit(jit, main.rss_jit, rssq)),
-         log_date
-      )
-      add_worker("rss"..rssq, worker_expr,
-                 function(pid)
-                    logger:log(string.format("Launched RSS worker process #%d",
-                                             pid))
-                    log_cpu_choice(pid, cpu, main.rss_pin_cpu)
-                    shm.create("rss_workers/"..pid, "uint64_t")
-                 end,
-                 function(pid)
-                    shm.unlink("rss_workers/"..pid)
-                 end
-      )
-
-   end
-
-   local ctrl_graph = probe.configure_mlx_ctrl_graph(mellanox, log_date)
-
-   for _, spec in ipairs(workers) do
-      local child_pid = worker.start(spec.name, spec.expr)
-      spec.create_fn(child_pid)
-   end
-
-   return workers, ctrl_graph, mellanox
-end
-
-local long_opts = {
-   duration = "D",
-   debug = "d",
-   jit = "j",
-   help = "h",
-   ["busy-wait"] = "b",
-   ["log-date"] = 'L',
-   ["worker-check-interval"] = 'c'
-}
-
-local function usage(exit_code)
-   print(require("program.ipfix.probe_rss.README_inc"))
-   main.exit(exit_code)
-end
-
-function run (parameters)
-   -- Limit for the number of restarts of a worker.  If exceeded, the
-   -- entire probe is terminated.  Currently not configurable.
-   local restart_limit = 5
-
-   local duration
-   local busywait = false
-   local log_date = false
-   local worker_check_interval = 5
-   local profiling, traceprofiling
-   local jit = { opts = {} }
-   local log_pid = string.format("[%5d]", S.getpid())
-   local opt = {
-      D = function (arg)
-         if arg:match("^[0-9]+$") then
-            duration = tonumber(arg)
-         else
-            usage(1)
-         end
-      end,
-      h = function (arg) usage(0) end,
-      d = function (arg) _G.developer_debug = true end,
-      b = function (arg)
-         busywait = true
-      end,
-      j = probe.parse_jit_option_fn(jit),
-      L = function(arg)
-         log_date = true
-      end,
-      c = function(arg)
-         if arg:match("^[0-9]+$") then
-            worker_check_interval = tonumber(arg)
-         end
-      end
-   }
-
-   -- Parse command line arguments
-   parameters = lib.dogetopt(parameters, opt, "hdj:D:l:bLc:", long_opts)
-   if #parameters ~= 1 then usage (1) end
-
-   local logger = logger.new({ rate = 200, date = log_date,
-                               module = log_pid.." RSS master" })
-   local file = table.remove(parameters, 1)
-   local probe_config = assert(loadfile(file))()
-   local workers, ctrl_graph, mellanox =
-      create_workers(probe_config, duration, busywait, jit, logger, log_date)
-
-   if worker_check_interval > 0 then
-      local workers_by_name = {}
-      for _, spec in ipairs(workers) do
-         workers_by_name[spec.name] = spec
-      end
-
-      local worker_check = function()
-         for n, s in pairs(worker.status()) do
-            if not s.alive then
-               logger:log(string.format("Worker process %d died (status %d)",
-                                        s.pid, s.status))
-               local spec = workers_by_name[n]
-               spec.restarts = spec.restarts + 1
-               if spec.restarts > restart_limit then
-                  logger:log(string.format("Too many restarts (>%d), "
-                                              .."terminating", restart_limit))
-                  S.exit(1)
-               end
-               logger:log(string.format("Restarting process (attempt #%d)",
-                                        spec.restarts))
-               spec.remove_fn(s.pid)
-               local new_pid = worker.start(n, spec.expr)
-               spec.create_fn(new_pid)
-            else
-               workers_by_name[n].restarts = 0
-            end
-         end
-      end
-      timer.activate(timer.new("workermon", worker_check,
-                               worker_check_interval * 10e8, "repeating"))
-   end
-
-   engine.busywait = false
-   engine.Hz = 10
-   engine.configure(ctrl_graph)
-
-   engine.main({ duration = duration })
-
-   logger:log("waiting for workers to finish")
-   local alive
-   repeat
-      alive = false
-      for _, s in pairs(worker.status()) do
-         if s.alive then alive = true end
-      end
-   until not alive
-   logger:log("done")
-end
diff --git a/src/program/ipfix/stats/README.inc b/src/program/ipfix/stats/README.inc
deleted file mode 100644
index 68725c4a76..0000000000
--- a/src/program/ipfix/stats/README.inc
+++ /dev/null
@@ -1,5 +0,0 @@
-Usage: snabb ipfix stats [options] [ <pid> ... ]
-       snabb ipfix stats --help
-
-Available options:
-       -n    Don't pretty-print numbers in "comma value" form
diff --git a/src/program/ipfix/stats/stats.lua b/src/program/ipfix/stats/stats.lua
deleted file mode 100644
index 8f74d61e43..0000000000
--- a/src/program/ipfix/stats/stats.lua
+++ /dev/null
@@ -1,326 +0,0 @@
--- Display statistics for IPFIX processes
-
--- XXX Re-write needed
-module(..., package.seeall)
-
-local shm = require("core.shm")
-local counter = require("core.counter")
-local lib = require("core.lib")
-local S = require("syscall")
-
-local pretty = true
-local function comma_value(value)
-   return pretty and lib.comma_value(value) or tostring(value)
-end
-
-local function read(c, cv)
-   local n = tonumber(counter.read(c))
-   return cv and comma_value(n) or n
-end
-
-local function format(indent, format, ...)
-   for i = 1, indent do
-      io.stdout:write(" ")
-   end
-   print(string.format(format, ...))
-end
-
-local function ppid_from_group (pid)
-   local ppid
-   local group = shm.root.."/"..shm.resolve("/"..pid.."/group")
-   local stat = assert(S.lstat(group))
-   if stat.islnk then
-      ppid = S.readlink(group):match("/(%d+)/group")
-   end
-   return ppid
-end
-
-local function pci_stats(indent, path)
-   local indent = indent or 0
-   for _, pciaddr in ipairs(shm.children(path.."/pci")) do
-      local pci = shm.open_frame(path.."/pci/"..pciaddr)
-      local rxdrops_total = 0
-      print()
-      format(indent, "NIC %s", pciaddr)
-      local qids = {}
-      for _, file in ipairs(shm.children(path.."/pci/"..pciaddr)) do
-         local base = file:match('(q[%d]+_rx_enabled)')
-         if base then
-            if read(pci[base]) == 1 then
-               local i = tonumber(file:match('q([%d]+)_'))
-               table.insert(qids, i)
-            end
-         end
-      end
-      table.sort(qids)
-
-      if #qids == 0 then
-         rxdrops_total = read(pci['rxdrop'])
-      else
-         for _, i in ipairs(qids) do
-            local name = "q"..i.."_"
-            local rxdrops = read(pci[name..'rxdrops'])
-            format(indent+2, "Queue #"..i)
-            format(indent+4, "rxpackets         %s", read(pci[name..'rxpackets'], true))
-            format(indent+4, "rxdrops           %s", comma_value(rxdrops))
-            rxdrops_total = rxdrops_total + rxdrops
-         end
-      end
-      local rxpackets = read(pci.rxpackets)
-      format(indent+2, "Total")
-      format(indent+4, "rxpackets         %s", comma_value(rxpackets))
-      format(indent+4, "rxdrops           %s (%1.4F%%)",
-             comma_value(rxdrops_total),
-             100*rxdrops_total/rxpackets)
-      shm.delete_frame(pci)
-   end
-end
-
-local function ipfix_info(pid)
-   local path = '/'..pid
-   local link, err = S.readlink(shm.root.."/"..shm.resolve(path..'/group'))
-   if err then
-      return { pid = pid }
-   end
-   local link
-   for _, name in ipairs(shm.children(path
-                                      .."/interlink/receiver")) do
-      link = name:match("(.*).interlink")
-      break
-   end
-   return { pid = pid, link = link }
-end
-
-local function rss_info(pid)
-   local path = '/'..pid
-   local interlinks = {}
-   if shm.exists(path..'/interlink') then
-      for _, name in ipairs(shm.children(path
-                                         .."/interlink/transmitter")) do
-         local link = name:match("(.*).interlink")
-         interlinks[link] = true
-      end
-   end
-   local receivers = {}
-   for _, link in ipairs(shm.children(path..'/links')) do
-      repeat
-         if not link:match("^rss%.") then break end
-         local id, type
-         local receiver = link:match(".* -> +([%w_]+)%.")
-         if interlinks[receiver] then
-            -- Receiver is connected through a
-            -- multi-process link
-            type = 'interlink'
-            -- Will be translated into a process ID later
-            id = receiver
-         else
-            -- Receiver is running in the same process
-            type = 'instance'
-            id = tonumber(receiver:match("ipfix(%d+)"))
-         end
-         table.insert(receivers,
-                      { id = id, type = type, link = link })
-      until true
-   end
-   return { pid = pid, receivers = receivers }
-end
-
-local function usage()
-   print(require("program.ipfix.stats.README_inc"))
-   main.exit(0)
-end
-
-local long_opts = {
-   help = "h",
-}
-
-function run (args)
-   local opt = {
-      h = function (arg)
-         usage()
-      end,
-      n = function (arg)
-         pretty = false
-      end
-   }
-   args = lib.dogetopt(args, opt, "hn", long_opts)
-
-   local pids = {}
-   if #args == 0 then
-      for _, pid in ipairs(shm.children('/')) do
-         if tonumber(pid) then
-            table.insert(pids, pid)
-         end
-      end
-   else
-      pids = args
-   end
-
-   local ipfix = {}
-   local rss = {}
-   local ctrls = {}
-
-   for _, pid in ipairs(pids) do
-      if not shm.exists('/'..pid.."/config-worker-channel") then
-	goto continue
-      end
-      for _, dir in ipairs(shm.children('/'..pid)) do
-         if dir == 'ipfix_templates' then
-            table.insert(ipfix, ipfix_info(tonumber(pid)))
-            local ppid = ppid_from_group(pid)
-            if ppid then
-               ctrls[ppid] = true
-            end
-            break
-         end
-      end
-      for _, dir in ipairs(shm.children('/'..pid..'/apps')) do
-         if dir == 'rss' then
-            table.insert(rss, rss_info(tonumber(pid)))
-            local ppid = ppid_from_group(pid)
-            if ppid then
-               ctrls[ppid] = true
-            end
-         end
-      end
-      ::continue::
-   end
-
-   if #ipfix == 0 then
-      print("No IPFIX processes found")
-      os.exit(0)
-   end
-
-   local function by_pid(a, b)
-      return a.pid < b.pid
-   end
-   table.sort(ipfix, by_pid)
-   table.sort(rss, by_pid)
-
-   -- Get PIDs of interlink receivers
-   local function pid_from_link(link, ppid)
-      for _, ipfix in ipairs(ipfix) do
-         if ipfix.link == link then
-            return ipfix.pid
-         end
-      end
-   end
-   for _, rss in ipairs(rss) do
-      for _, r in ipairs(rss.receivers) do
-         if r.type == "interlink" then
-            r.id = pid_from_link(r.id, rss.pid)
-         end
-      end
-   end
-
-   for ppid, _ in pairs(ctrls) do
-      print()
-      format(0, "Control process #%d", ppid)
-      pci_stats(2, '/'..ppid)
-      print()
-   end
-
-   for _, rss in ipairs(rss) do
-      print()
-      format(0, "RSS process #%d", rss.pid)
-      pci_stats(2, '/'..rss.pid)
-      print()
-      format(2, "IPFIX receivers")
-      table.sort(rss.receivers, function (a, b) return a.id < b.id end)
-      for _, rcv in ipairs(rss.receivers) do
-         if rcv.type == 'interlink' then
-            format(4, "Process #%d", rcv.id)
-         else
-            format(4, "Embedded instance #%d", rcv.id)
-         end
-         local stats = shm.open_frame('/'..rss.pid..'/links/'..rcv.link)
-         local txpackets = read(stats.txpackets)
-         local txdrop = read(stats.txdrop)
-         format(6, "txpackets         %s", comma_value(txpackets))
-         format(6, "txdrop            %s (%1.4F%%)",
-                comma_value(txdrop),
-                100*txdrop/txpackets)
-         shm.delete_frame(stats)
-      end
-   end
-
-   for _, ipfix in ipairs(ipfix) do
-      print()
-      format(0, "IPFIX process #%d", ipfix.pid)
-      local base_path = "/"..ipfix.pid
-
-      if #rss == 0 then
-         pci_stats(2, base_path)
-      end
-
-      local instances = {}
-      for _, app in ipairs(shm.children(base_path.."/apps")) do
-         local instance =  app:match("^ipfix(%d)$")
-         if instance then
-            table.insert(instances, tonumber(instance))
-         end
-      end
-      table.sort(instances)
-      for _, instance in ipairs(instances) do
-         local ipfix = shm.open_frame(base_path.."/apps/ipfix"..instance)
-         print()
-         format(2,"Instance #"..instance)
-         format(4,"Version             %d", read(ipfix.version))
-         format(4,"Observation domain  %d", read(ipfix.observation_domain))
-         format(4,"Received packets    %s", read(ipfix.received_packets, true))
-         format(4,"Ignored packets     %s", read(ipfix.ignored_packets, true))
-         format(4,"Template packets    %s", read(ipfix.template_packets, true))
-         format(4, "Sequence number     %s", read(ipfix.sequence_number, true))
-         shm.delete_frame(ipfix)
-
-         local templates = {}
-         local path = base_path.."/ipfix_templates/"..instance
-         for _, id in ipairs(shm.children(path)) do
-            table.insert(templates, tonumber(id))
-         end
-         table.sort(templates)
-         for _, id in ipairs(templates) do
-            print()
-            format(4,"Template #%d", id)
-            local template = shm.open_frame(path.."/"..id)
-            format(6, "Processed packets    %s",
-                   read(template.packets_in, true))
-            format(6, "Exported flows       %s",
-                   read(template.exported_flows, true))
-            format(6, "Flow export packets  %s",
-                   read(template.flow_export_packets, true))
-
-            local function table_stats(prefix, title)
-               local function get(name)
-                  return  template[prefix..'_'..name]
-               end
-               local size = read(get('size'))
-               local occupancy = read(get('occupancy'))
-               local max_disp = read(get('max_displacement'))
-               format(6,title )
-               format(8,"Occupancy          %s", comma_value(occupancy))
-               format(8,"Size               %s", comma_value(size))
-               format(8,"Byte size          %s", read(get('byte_size'), true))
-               format(8,"Load-factor        %1.2f", occupancy/size)
-               format(8,"Max displacement   %d", max_disp)
-               if template[prefix..'_scan_time'] then
-                  format(8,"Last scan time     %d", read(get('scan_time')))
-               end
-            end
-            table_stats('table', 'Table stats')
-            if read(template.rate_table_size) > 0 then
-               table_stats('rate_table', 'Export Flow-Rate Table stats')
-            end
-            shm.delete_frame(template)
-            if shm.exists(path.."/"..id.."/stats") then
-               format(6, "Template-specific stats")
-               local stats = shm.open_frame(path.."/"..id.."/stats")
-               for name, _ in pairs(stats.specs) do
-                  format(8, "%-25s %s", name, read(stats[name], true))
-               end
-               shm.delete_frame(stats)
-            end
-         end
-      end
-   end
-end