[wip] snabb top --yang: dump RFC7223 interface stats as JSON

src/README.md

@@ -543,10 +543,18 @@ Returns a table that acts as a bounds checked wrapper around a C array of
 ctype and the caller must ensure that the allocated memory region at
 *base*/*offset* is at least `sizeof(type)*size` bytes long.
 
-— Function **lib.timer** *s*
+— Function **lib.timer** *duration*, *mode*, *timefun*
+
+Returns a closure that will return `false` until *duration* has elapsed. If
+*mode* is `'repeating'` the timer will reset itself after returning `true`,
+thus implementing an interval timer. *Timefun* is used to get a monotonic time.
+*Timefun* defaults to `C.get_time_ns`.
+
+The “deadline” for a given *duration* is computed by adding *duration* to the
+result of calling *timefun*, and is saved in the resulting closure. A
+*duration* has elapsed when its deadline is less than or equal the value
+obtained using *timefun* when calling the closure.
 
-Returns a function that accepts no parameters and acts as a predicate to
-test if *ns* nanoseconds have elapsed.
 
 — Function **lib.waitfor** *condition*
 

src/apps/intel/intel_app.lua

@@ -6,8 +6,11 @@ local zone = require("jit.zone")
 local basic_apps = require("apps.basic.basic_apps")
 local ffi      = require("ffi")
 local lib      = require("core.lib")
+local shm      = require("core.shm")
+local counter  = require("core.counter")
 local pci      = require("lib.hardware.pci")
 local register = require("lib.hardware.register")
+local macaddress = require("lib.macaddress")
 local intel10g = require("apps.intel.intel10g")
 local receive, transmit, full, empty = link.receive, link.transmit, link.full, link.empty
 Intel82599 = {}
@@ -35,21 +38,49 @@ end
 -- Create an Intel82599 App for the device with 'pciaddress'.
 function Intel82599:new (arg)
    local conf = config.parse_app_arg(arg)
+   local self = {}
 
    if conf.vmdq then
       if devices[conf.pciaddr] == nil then
-         devices[conf.pciaddr] = {pf=intel10g.new_pf(conf):open(), vflist={}}
+         local pf = intel10g.new_pf(conf):open()
+         devices[conf.pciaddr] = {pf=pf, vflist={}, stats={register=pf.s}}
       end
       local dev = devices[conf.pciaddr]
       local poolnum = firsthole(dev.vflist)-1
       local vf = dev.pf:new_vf(poolnum)
       dev.vflist[poolnum+1] = vf
-      return setmetatable({dev=vf:open(conf)}, Intel82599)
+      self.dev = vf:open(conf)
+      self.stats = devices[conf.pciaddr].stats
    else
-      local dev = intel10g.new_sf(conf):open()
-      if not dev then return null end
-      return setmetatable({dev=dev, zone="intel"}, Intel82599)
+      self.dev = assert(intel10g.new_sf(conf):open(), "Can not open device.")
+      self.stats = { register = self.dev.s }
+      self.zone = "intel"
    end
+   if not self.stats.counters then
+      local counters = {}
+      local path = "/counters/"..conf.pciaddr.."/"
+      counters['type']               = counter.open(path..'type')
+      counters['discontinuity-time'] = counter.open(path..'discontinuity-time')
+      counters['in-octets']          = counter.open(path..'in-octets')
+      counters['in-multicast']       = counter.open(path..'in-multicast')
+      counters['in-broadcast']       = counter.open(path..'in-broadcast')
+      counters['in-discards']        = counter.open(path..'in-discards')
+      counters['out-octets']         = counter.open(path..'out-octets')
+      counters['out-multicast']      = counter.open(path..'out-multicast')
+      counters['out-broadcast']      = counter.open(path..'out-broadcast')
+      counters['out-discards']       = counter.open(path..'out-discards')
+      counter.set(counters['type'], 0x1000) -- Hardware interface
+      counter.set(counters['discontinuity-time'], C.get_unix_time())
+      if not conf.vmdq and conf.macaddr then
+         counters['phys-address'] = counter.open(path..'phys-address')
+         counter.set(counters['phys-address'],
+                     macaddress:new(conf.macaddr):int())
+      end
+      self.stats.counters = counters
+      self.stats.path = path
+      self.stats.sync_timer = lib.timer(0.001, 'repeating', engine.now)
+   end
+   return setmetatable(self, Intel82599)
 end
 
 function Intel82599:stop()
@@ -70,6 +101,12 @@ function Intel82599:stop()
    if close_pf then
       close_pf:close()
    end
+   if not self.dev.pf or close_pf then
+      for name, _ in pairs(self.stats.counters) do
+         counter.delete(self.stats.path..name)
+      end
+      shm.unlink(self.stats.path)
+   end
 end
 
 
@@ -78,6 +115,11 @@ function Intel82599:reconfig(arg)
    assert((not not self.dev.pf) == (not not conf.vmdq), "Can't reconfig from VMDQ to single-port or viceversa")
 
    self.dev:reconfig(conf)
+
+   if not self.dev.pf and conf.macaddr then
+      counter.set(self.stats.counters['phys-address'],
+                  macaddress:new(conf.macaddr):int())
+   end
 end
 
 -- Allocate receive buffers from the given freelist.
@@ -95,6 +137,9 @@ function Intel82599:pull ()
       transmit(l, self.dev:receive())
    end
    self:add_receive_buffers()
+   if self.stats.sync_timer() then
+      self:sync_stats()
+   end
 end
 
 function Intel82599:add_receive_buffers ()
@@ -104,6 +149,21 @@ function Intel82599:add_receive_buffers ()
    end
 end
 
+-- Synchronize self.stats.register a and self.stats.counters.
+function Intel82599:sync_stats ()
+   --- XXX - it is questionable if I choose the right register to counter
+   --- mapping
+   local counters, register = self.stats.counters, self.stats.register
+   counter.set(counters['in-octets'],     register.GORC64())
+   counter.set(counters['in-multicast'],  register.MPRC())
+   counter.set(counters['in-broadcast'],  register.BPRC())
+   counter.set(counters['in-discards'],   register.TPR() - register.GPRC())
+   counter.set(counters['out-octets'],    register.GOTC64())
+   counter.set(counters['out-multicast'], register.MPTC())
+   counter.set(counters['out-broadcast'], register.BPTC())
+   counter.set(counters['out-discards'],  register.TPT() - register.GPTC())
+end
+
 -- Push packets from our 'rx' link onto the network.
 function Intel82599:push ()
    local l = self.input.rx

src/apps/vhost/vhost_user.lua

@@ -13,7 +13,9 @@ local lib       = require("core.lib")
 local link      = require("core.link")
 local main      = require("core.main")
 local memory    = require("core.memory")
+local counter   = require("core.counter")
 local pci       = require("lib.hardware.pci")
+local ethernet  = require("lib.protocol.ethernet")
 local net_device= require("lib.virtio.net_device")
 local timer     = require("core.timer")
 local ffi       = require("ffi")
@@ -52,6 +54,15 @@ function VhostUser:new (args)
    else
       self.qemu_connect = self.client_connect
    end
+   -- initialize counters
+   self.counters = {}
+   for _, name in ipairs({'type', 'discontinuity-time',
+                          'in-octets', 'in-unicast', 'in-multicast', 'in-discards',
+                          'out-octets', 'out-unicast', 'out-multicast'}) do
+      self.counters[name] = counter.open(name)
+   end
+   counter.set(self.counters['type'], 0x1001) -- Virtual interface
+   counter.set(self.counters['discontinuity-time'], C.get_unix_time())
    return self
 end
 
@@ -68,6 +79,9 @@ function VhostUser:stop()
    self:free_mem_table()
 
    if self.link_down_proc then self.link_down_proc() end
+
+   -- delete counters
+   for name, _ in pairs(self.counters) do counter.delete(name) end
 end
 
 function VhostUser:pull ()
@@ -86,6 +100,24 @@ function VhostUser:push ()
    end
 end
 
+function VhostUser:tx_callback (p)
+   counter.add(self.counters['out-octets'], packet.length(p))
+   local mcast = ethernet:n_mcast(packet.data(p))
+   counter.add(self.counters['out-multicast'], mcast)
+   counter.add(self.counters['out-unicast'], 1 - mcast)
+end
+
+function VhostUser:rx_callback (p)
+   counter.add(self.counters['in-octets'], packet.length(p))
+   local mcast = ethernet:n_mcast(packet.data(p))
+   counter.add(self.counters['in-multicast'], mcast)
+   counter.add(self.counters['in-unicast'], 1 - mcast)
+end
+
+function VhostUser:rxdrop_callback (p)
+   counter.add(self.counters['in-discards'])
+end
+
 -- Try to connect to QEMU.
 function VhostUser:client_connect ()
    return C.vhost_user_connect(self.socket_path)

src/core/app.lua

@@ -7,6 +7,7 @@ local lib       = require("core.lib")
 local link      = require("core.link")
 local config    = require("core.config")
 local timer     = require("core.timer")
+local shm     = require("core.shm")
 local histogram = require('core.histogram')
 local counter   = require("core.counter")
 local zone      = require("jit.zone")
@@ -68,6 +69,8 @@ end
 -- Run app:methodname() in protected mode (pcall). If it throws an
 -- error app will be marked as dead and restarted eventually.
 local function with_restart (app, method)
+   local oldshm = shm.path
+   shm.path = app.shmpath
    if use_restart then
       -- Run fn in protected mode using pcall.
       local status, err = pcall(method, app)
@@ -78,6 +81,7 @@ local function with_restart (app, method)
    else
       method(app)
    end
+   shm.path = oldshm
 end
 
 -- Restart dead apps.
@@ -155,7 +159,13 @@ function apply_config_actions (actions, conf)
    -- Table of functions that execute config actions
    local ops = {}
    function ops.stop (name)
-      if app_table[name].stop then app_table[name]:stop() end
+      if app_table[name].stop then
+         local shmorig = shm.path
+         shm.path = app_table[name].shmpath
+         app_table[name]:stop()
+         shm.path = shmorig
+         shm.unlink(app_table[name].shmpath)
+      end
    end
    function ops.keep (name)
       new_app_table[name] = app_table[name]
@@ -165,7 +175,10 @@ function apply_config_actions (actions, conf)
    function ops.start (name)
       local class = conf.apps[name].class
       local arg = conf.apps[name].arg
+      local shmpath, shmorig = "counters/"..name, shm.path
+      shm.path = shmpath
       local app = class:new(arg)
+      shm.path = shmorig
       if type(app) ~= 'table' then
          error(("bad return value from app '%s' start() method: %s"):format(
                   name, tostring(app)))
@@ -174,6 +187,7 @@ function apply_config_actions (actions, conf)
       app.appname = name
       app.output = {}
       app.input = {}
+      app.shmpath = shmpath
       new_app_table[name] = app
       table.insert(new_app_array, app)
       app_name_to_index[name] = #new_app_array
@@ -493,6 +507,34 @@ function selftest ()
    assert(app_table.app3 == orig_app3) -- should be the same
    main({duration = 4, report = {showapps = true}})
    assert(app_table.app3 ~= orig_app3) -- should be restarted
+   -- Test shm.path management
+   print("shm.path management")
+   local S = require("syscall")
+   local App4 = {zone="test"}
+   function App4:new ()
+      local c = counter.open('test')
+      counter.set(c, 42)
+      counter.commit()
+      return setmetatable({test_counter = c},
+                          {__index = App4})
+   end
+   function App4:pull ()
+      assert(counter.read(self.test_counter) == 42, "Invalid counter value")
+      counter.add(self.test_counter)
+   end
+   function App4:stop ()
+      assert(counter.read(self.test_counter) == 43, "Invalid counter value")
+      counter.delete('test')
+   end
+   local c_counter = config.new()
+   config.app(c_counter, "App4", App4)
+   configure(c_counter)
+   main({done = function () return app_table.App4.test_counter end})
+   assert(S.stat(shm.root.."/"..shm.resolve("counters/App4/test")),
+          "Missing : counters/App4/test")
+   configure(config.new())
+   assert(not S.stat(shm.root.."/"..shm.resolve("counters/App4")),
+          "Failed to unlink counters/App4")
    print("OK")
 end
 

src/core/counter.lua

@@ -44,7 +44,8 @@ local private = {}
 local numbers = {} -- name -> number
 
 function open (name, readonly)
-   if numbers[name] then return private[numbers[name]] end
+   local qname = shm.resolve(name)
+   if numbers[qname] then return private[numbers[qname]] end
    local n = #public+1
    if readonly then
       public[n] = shm.open(name, counter_t, readonly)
@@ -53,21 +54,22 @@ function open (name, readonly)
       public[n] = shm.create(name, counter_t)
       private[n] = ffi.new(counter_t)
    end
-   numbers[name] = n
+   numbers[qname] = n
    return private[n]
 end
 
 function delete (name)
-   local number = numbers[name]
-   if not number then error("counter not found for deletion: " .. name) end
+   local qname = shm.resolve(name)
+   local number = numbers[qname]
+   if not number then error("counter not found for deletion: " .. qname) end
    -- Free shm object
    shm.unmap(public[number])
    -- If we "own" the counter for writing then we unlink it too.
    if public[number] ~= private[number] then
       shm.unlink(name)
    end
    -- Free local state
-   numbers[name] = false
+   numbers[qname] = false
    public[number] = false
    private[number] = false
 end

src/core/lib.lua

@@ -246,10 +246,23 @@ function bounds_checked (type, base, offset, size)
    return wrap(ffi.cast(tptr, ffi.cast("uint8_t *", base) + offset))
 end
 
--- Return a function that will return false until NS nanoseconds have elapsed.
-function timer (ns)
-   local deadline = C.get_time_ns() + ns
-   return function () return C.get_time_ns() >= deadline end
+-- Return a function that will return false until duration has elapsed.
+-- If mode is 'repeating' the timer will reset itself after returning true,
+-- thus implementing an interval timer. Timefun defaults to `C.get_time_ns'.
+function timer (duration, mode, timefun)
+   timefun = timefun or C.get_time_ns
+   local deadline = timefun() + duration
+   local function oneshot ()
+      return timefun() >= deadline
+   end
+   local function repeating ()
+      if timefun() >= deadline then
+         deadline = deadline + duration
+         return true
+      else return false end
+   end
+   if mode == 'repeating' then return repeating
+   else return oneshot end
 end
 
 -- Loop until the function `condition` returns true.

src/core/link.lua

@@ -28,13 +28,16 @@ function new (name)
    for _, c in ipairs(counternames) do
       r.stats[c] = counter.open("counters/"..name.."/"..c)
    end
+   counter.set(counter.open("counters/"..name.."/discontinuity-time"),
+               C.get_unix_time())
    return r
 end
 
 function free (r, name)
    for _, c in ipairs(counternames) do
       counter.delete("counters/"..name.."/"..c)
    end
+   counter.delete("counters/"..name.."/discontinuity-time")
    shm.unmap(r)
    shm.unlink("links/"..name)
 end