ntopng/scripts/lua/modules/recording_utils.lua

--
-- (C) 2014-18 - ntop.org
--

local dirs = ntop.getDirs()
require "lua_utils"
--require "prefs_utils"
local json = require("dkjson")
local os_utils = require("os_utils")

prefs = ntop.getPrefs()

local extraction_queue_key = "ntopng.traffic_recording.extraction_queue"
local extraction_stop_queue_key = "ntopng.traffic_recording.extraction_stop_queue"
local extraction_seqnum_key = "ntopng.traffic_recording.extraction_seqnum"
local extraction_jobs_key = "ntopng.traffic_recording.extraction_jobs"
local IS_AVAILABLE_KEY = "ntopng.cache.traffic_recording_available"

local recording_utils = {}

recording_utils.default_disk_space = 10*1024

local function n2diskctl(...)
  return os_utils.ntopctlCmd("n2disk", ...)
end

-- #################################

--! @brief Check if an interface is a ZMQ interface that can be used with external interfaces for traffic recording and flow import
--! @param ifid the interface identifier 
--! @return true if supported, false otherwise
function recording_utils.isSupportedZMQInterface(ifid)
  local ifname = getInterfaceName(ifid)
  -- localhost in collectore mode is accepted
  local zmq_prefix_any = "tcp://*"
  local zmq_prefix_loh = "tcp://127.0.0.1"
  if (ifname:sub(1, #zmq_prefix_any) == zmq_prefix_any or
      ifname:sub(1, #zmq_prefix_loh) == zmq_prefix_loh) and
      ifname:sub(-1) == "c" then
    return true
  end
  return false
end

local function getZMQPort(addr)
  local values = split(addr, ':')
  if #values == 3 then
    local port = split(values[3], 'c')
    return port[1]
  end
  return nil
end

--! @brief Return the ZMQ endpoint that should be used by an external process (n2disk) to deliver flows to ntopng
--! @param ifid the interface identifier 
--! @return the endpoint
function recording_utils.getZMQProbeAddr(ifid)
  local port = getZMQPort(getInterfaceName(ifid))
  if port == nil then
    port = "5556"
  end
  return "tcp://127.0.0.1:"..port
end

--! @brief Check if an interface is supported for recording (packet interface, or ZMQ interface that can be used with external interfaces for traffic dump and flow import)
--! @param ifid the interface identifier 
--! @return true if supported, false otherwise
function recording_utils.isSupportedInterface(ifid)
  if interface.isPacketInterface() or 
     recording_utils.isSupportedZMQInterface(ifid) then
    return true
  end
  return false
end

-- only called during boot
function recording_utils.checkAvailable()
  local is_available = false

  if(not ntop.isWindows())
    and (not ntop.isnEdge())
    and os_utils.hasService("n2disk", getInterfaceName(getFirstInterfaceId())) then
    is_available = true
  end

  ntop.setCache(IS_AVAILABLE_KEY, ternary(is_available, "1", "0"))
end

--! @brief Check if traffic recording is available and allowed for the current user on an interface
--! @return true if recording is available, false otherwise
function recording_utils.isAvailable()
  if isAdministrator() and (ntop.getCache(IS_AVAILABLE_KEY) == "1") then
    return true
  end
  return false
end

--! @brief Return information about the recording service (n2disk) including systemid and version
--! @return a table with the information
function recording_utils.getN2diskInfo()
  local info = {}

  local n2disk_version = os_utils.execWithOutput("n2disk --version")
  local lines = split(n2disk_version, "\n")
  for i = 1, #lines do
    local line = lines[i]
    line = string.gsub(line, "%s+", " ")
    local pair = split(line, " ")
    if pair[1] ~= nil and pair[2] ~= nil then
      if pair[1] == "n2disk" then
        info.version = trimString(pair[2])
      elseif pair[1] == "SystemID:" then
        info.systemid = trimString(pair[2])
      end
    end
  end

  local license_file = io.open("/etc/n2disk.license", "r")
  if license_file ~= nil then
    local license = license_file:read "*l"
    info.license = license
    license_file:close()
  end

  return info
end

--! @brief Install a license for n2disk
--! @param key The license key
--! @return true if the license is installed, false in case it is not possible
function recording_utils.setLicense(key)
  n2diskctl("set-license", key)
  return true
end

local function setLicenseFromRedis()
  local n2disk_license = ntop.getCache('ntopng.prefs.n2disk_license')
  if not isEmptyString(n2disk_license) then
    recording_utils.setLicense(n2disk_license)
  end
end

local function isZCInterface(ifname)
  local proc_info = io.open("/proc/net/pf_ring/dev/"..ifname.."/info", "r")
  if proc_info ~= nil then
    local info = proc_info:read "*a"
    proc_info:close()
    if string.match(info, "ZC") then
      return true
    end
  end
  -- return true -- DEBUG
  return false
end

local function getInUseExtInterfaces(current_ifid)
  local inuse_ext_interfaces = {}
  local ntopng_interfaces = interface.getIfNames()
  for other_ifid,other_ifname in pairs(ntopng_interfaces) do
    if other_ifid ~= current_ifid then
      local enabled = ntop.getCache('ntopng.prefs.ifid_'..other_ifid..'.traffic_recording.enabled')
      if not isEmptyString(enabled) and enabled == "true" then
        local other_ext_ifname = ntop.getCache('ntopng.prefs.ifid_'..other_ifid..'.traffic_recording.ext_ifname')
        if not isEmptyString(other_ext_ifname) then
          inuse_ext_interfaces[other_ext_ifname] = true
        end
      end
    end
  end
  return inuse_ext_interfaces
end

--! @brief Return external interfaces, not in use by ntopng, that can be used through ZMQ interface for traffic recording and flow import
--! @param ifid the interface identifier 
--! @return a table with external interfaces information
function recording_utils.getExtInterfaces(ifid)
  local ext_interfaces = {}
  local all_interfaces = ntop.listInterfaces()
  local ntopng_interfaces = swapKeysValues(interface.getIfNames()) 
  local inuse_ext_interfaces = getInUseExtInterfaces(ifid)
 
  for ifname,_ in pairs(all_interfaces) do
    if ntopng_interfaces[ifname] == nil and -- not in use as packet interface by ntopng 
       inuse_ext_interfaces[ifname] == nil and -- not in use by other zmq interfaces
       all_interfaces[ifname].module ~= nil and -- detected by pf_ring
       all_interfaces[ifname].module ~= "pf_ring" -- ('pf_ring-zc', 'napatech', ..)
      then
      local prefix = ""
      if all_interfaces[ifname].module == "pf_ring-zc" then
        prefix = "zc:"
      end
      ext_interfaces[ifname] = {
        ifdesc = prefix..ifname,
        module = all_interfaces[ifname].module
      }
    end
  end

  return ext_interfaces
end

local function nextFreeCore(num_cores, busy_cores, start)
  local busy_map = swapKeysValues(busy_cores) 
  for i=start,num_cores-1 do
    if busy_map[i] == nil then
      return i
    end
  end
  return start
end

-- Return memory information (values in MB)
local function memInfo()
  local mem_info = {}

  for line in io.lines("/proc/meminfo") do 
    local pair = split(line, ':')
    local k = pair[1]
    if pair[2] ~= nil then
      local value = split(trimString(pair[2]), ' ')
      local v = tonumber(value[1])
      if value[2] ~= nil and value[2] == 'kB' then
        v = v/1024 -- kB to MB
      end
      mem_info[k] = math.floor(v)
    end
  end

  if mem_info['MemAvailable'] == nil and 
     mem_info['MemFree'] ~= nil and 
     mem_info['SReclaimable'] ~= nil then
    mem_info['MemAvailable'] = mem_info['MemFree'] + mem_info['SReclaimable']
  end

  return mem_info
end

local function dirname(s)
  s = s:gsub('/$', '')
  local s, n = s:gsub('/[^/]*$', '')
  if n == 1 then
    return ternary(string.len(s) > 0, s, "/")
  else 
    return '.' 
  end
end

--! @brief Return the root path for recorded pcap data
--! @param ifid the interface identifier 
--! @return the path
function recording_utils.getPcapPath(ifid)
  local storage_path = dirs.pcapdir
  return storage_path.."/"..ifid.."/pcap"
end

local function getPcapExtractionPath(ifid)
  local storage_path = dirs.pcapdir
  return storage_path.."/"..ifid.."/extr_pcap"
end

local function getTimelinePath(ifid)
  local storage_path = dirs.pcapdir
  return storage_path.."/"..ifid.."/timeline"
end

local function getPcapFileDir(job_id, ifid)
  local dir_path = getPcapExtractionPath(ifid)
  return dir_path.."/"..job_id
end

local function getPcapFilePath(job_id, ifid, file_id)
  local dir_path = getPcapFileDir(job_id, ifid)
  return dir_path.."/"..file_id..".pcap"
end

-- Read information about used disk space for an interface dump
local function interfaceStorageUsed(ifid)
  local pcap_path = recording_utils.getPcapPath(ifid)
  local line = os_utils.execWithOutput("du -s "..pcap_path.." 2>/dev/null")
  local values = split(line, '\t')
  if #values >= 1 then
    local if_used = tonumber(values[1])
    if if_used ~= nil then
      if_used = if_used/1024
      return math.ceil(if_used)
    end
  end
  return 0
end

--! @brief Read information about the storage, including storage size and available space
--! @param ifid the interface identifier 
--! @return a table containing storage information
function recording_utils.storageInfo(ifid)
  local storage_info = {
    path = dirs.pcapdir, dev = "", mount = "",
    total = 0, used = 0, avail = 0, used_perc = 0,
    if_used = 0, extraction_used = 0,
  }

  -- Global storage info
  local root_path = storage_info.path
  while not ntop.isdir(root_path) and string.len(root_path) > 1 do
    root_path = dirname(root_path) 
  end
  local line = os_utils.execWithOutput("df "..root_path.." 2>/dev/null|tail -n1")
  line = line:gsub('%s+', ' ')
  local values = split(line, ' ')
  if #values >= 6 then
    storage_info.dev = values[1]
    storage_info.total = tonumber(values[2])/1024
    storage_info.used = tonumber(values[3])/1024
    storage_info.avail = tonumber(values[4])/1024
    storage_info.used_perc = values[5]
    storage_info.mount = values[6]
  end

  -- Interface storage info
  storage_info.if_used = interfaceStorageUsed(ifid)

  -- PCAP Extraction storage info
  local extraction_path = getPcapExtractionPath(ifid)
  local line = os_utils.execWithOutput("du -s "..extraction_path.." 2>/dev/null")
  local values = split(line, '\t')
  if #values >= 1 then
    local extraction_used = tonumber(values[1])
    if extraction_used ~= nil then
      extraction_used = extraction_used/1024
      storage_info.extraction_used = extraction_used
    end
  end

  return storage_info
end

local function getN2diskInterfaceName(ifid)
  if interface.isPacketInterface() then
    return getInterfaceName(ifid)
  else
    return ntop.getCache('ntopng.prefs.ifid_'..ifid..'.traffic_recording.ext_ifname')
  end
end

-- Encode an interface name in a string that can be used in the n2disk configuration file name
local function getConfigInterfaceName(ifid)
  local ifname = getN2diskInterfaceName(ifid)
  return ifname:gsub("%,", "_")
end

--! @brief Generate a configuration for the traffic recording service (n2disk)
--! @param ifid the interface identifier 
--! @param params the traffic recording settings
function recording_utils.createConfig(ifid, params)
  local ifname = getN2diskInterfaceName(ifid)

  setLicenseFromRedis()

  if isEmptyString(ifname) then
    return false
  end

  local real_ifname = ifname
  if not interface.isPacketInterface() and isZCInterface(ifname) then
    -- real_ifname = ifname -- DEBUG
    real_ifname = "zc:"..ifname
  end

  local conf_dir = dirs.workingdir.."/n2disk"
  local filename = conf_dir.."/n2disk-" .. getConfigInterfaceName(ifid) .. ".conf"
  local storage_path = dirs.pcapdir

  if isEmptyString(storage_path) then
    return false
  end

  local defaults = {
    buffer_size = 1024,       -- Buffer size (MB)
    max_file_size = 256,      -- Max file length (MB)
    max_file_duration = 60,   -- Max file duration (sec)
    max_disk_space = recording_utils.default_disk_space, -- Max disk space (MB)
    snaplen = 1536,           -- Capture length
    writer_core = 0,          -- Writer thread affinity
    reader_core = 1,          -- Reader thread affinity
    indexer_cores = { 2 },    -- Indexer threads affinity
    -- Optional parameters
    -- zmq_endpoint = "tcp://*:5556" -- ZMQ endpoint for stats/flows
  }

  local ifspeed = (getInterfaceSpeed(getInterfaceId(ifname)) or 1000)

  -- Reading system memory info

  local mem_info = memInfo()
  local mem_available_mb = mem_info['MemAvailable']

  -- Computing file and buffer size

  local num_buffered_files = 4
  local min_file_size = 16 
  if ifspeed > 10000 then -- 40/100G
    defaults.max_file_size = 4*1024
    min_file_size = 1024
  elseif ifspeed > 1000 then -- 10G
    defaults.max_file_size = 1*1024
    min_file_size = 256
  elseif ifspeed > 100 then -- 1G
    defaults.max_file_size = 256
  else -- 10/100M
    defaults.max_file_size = 64
    num_buffered_files = 2
  end
  defaults.buffer_size = num_buffered_files * defaults.max_file_size

  local total_n2disk_mem = defaults.buffer_size * 2 -- pcap + index buffer

  if mem_available_mb ~= nil and mem_available_mb < total_n2disk_mem then
    local min_n2disk_buffer_size = (min_file_size * num_buffered_files) -- min memory for n2disk to work
    local min_n2disk_mem = min_n2disk_buffer_size * 2 -- pcap + index buffer
    if mem_available_mb < min_n2disk_mem then
      traceError(TRACE_ERROR, TRACE_CONSOLE, "Not enough memory available ("..mem_available_mb.."MB available, min required is "..min_n2disk_mem.."MB)") 
      return false
    end
    defaults.buffer_size = (mem_available_mb/2) -- leave some room for index memory and other processes
    defaults.max_file_size = math.floor(defaults.buffer_size/num_buffered_files)
  end

  -- Computing core affinity

  local indexing_threads = 1 -- 1G
  if ifspeed > 10000 then    -- 40/100G
    indexing_threads = 4
  elseif ifspeed > 1000 then -- 10G
    indexing_threads = 2
  end
  local n2disk_threads = indexing_threads + 2

  local line = os_utils.execWithOutput("nproc")
  local cores = tonumber(line)

  local ntopng_affinity = split(prefs.cpu_affinity, ',')
  local busy_cores = {}
  if cores - (#ntopng_affinity) >= n2disk_threads then
    -- enough cores to isolate all threads, skipping ntopng threads
    busy_cores = ntopng_affinity
  end

  local first_core = 0

  defaults.writer_core = nextFreeCore(cores, busy_cores, first_core)
  table.insert(busy_cores, defaults.writer_core)
  first_core = (defaults.writer_core + 1) % cores

  defaults.reader_core = nextFreeCore(cores, busy_cores, first_core)
  table.insert(busy_cores, defaults.reader_core)
  first_core = (defaults.reader_core + 1) % cores

  defaults.indexer_cores = {}
  for i=1,indexing_threads do
    local indexer_core = nextFreeCore(cores, busy_cores, first_core)
    table.insert(defaults.indexer_cores, indexer_core)
    table.insert(busy_cores, indexer_core)
    first_core = (indexer_core + 1) % cores
  end 

  local config = table.merge(defaults, params)

  -- Writing configuration file

  local ret = ntop.mkdir(conf_dir)

  if not ret then
    return false
  end

  local f = io.open(filename, "w")

  if not f then
    return false
  end

  local pcap_path = recording_utils.getPcapPath(ifid)
  local timeline_path = getTimelinePath(ifid)

  f:write("--interface="..real_ifname.."\n")
  f:write("--dump-directory="..pcap_path.."\n")
  f:write("--index\n")
  f:write("--timeline-dir="..timeline_path.."\n")
  f:write("--buffer-len="..config.buffer_size.."\n")
  f:write("--max-file-len="..config.max_file_size.."\n")
  f:write("--max-file-duration="..config.max_file_duration.."\n")
  f:write("--disk-limit="..config.max_disk_space.."\n")
  f:write("--snaplen="..config.snaplen.."\n")
  f:write("--writer-cpu-affinity="..config.writer_core.."\n")
  f:write("--reader-cpu-affinity="..config.reader_core.."\n")
  f:write("--compressor-cpu-affinity=")
  for i, v in ipairs(config.indexer_cores) do
    f:write(v..ternary(i == #config.indexer_cores, "", ","))
  end
  f:write("\n")
  f:write("--index-on-compressor-threads\n")
  if not isEmptyString(prefs.user) then
    f:write("-u="..prefs.user.."\n");
  else
    f:write("--dont-change-user\n");
  end
  if interface.isPacketInterface() then
    if prefs.capture_direction == "in" then
      f:write("--capture-direction=1\n")
    elseif prefs.capture_direction == "out" then
      f:write("--capture-direction=2\n")
    else
      f:write("--capture-direction=0\n")
    end
  end
  if config.zmq_endpoint ~= nil then
    f:write("--zmq="..config.zmq_endpoint.."\n")
    f:write("--zmq-probe-mode\n")
    f:write("--zmq-export-flows\n")
  end
  -- Ignored by systemd, required by init.d
  f:write("--daemon\n")
  f:write("-P=/var/run/n2disk-"..ifname..".pid\n")

  f:close()

  return true
end

--! @brief Check if traffic recording is available and enabled on an interface
--! @param ifid the interface identifier 
--! @return true if recording is enabled, false otherwise
function recording_utils.isEnabled(ifid)
  if recording_utils.isAvailable() then
    local record_traffic = ntop.getCache('ntopng.prefs.ifid_'..ifid..'.traffic_recording.enabled')
    if record_traffic == "true" then
      return true
    end
  end
  return false
end

--! @brief Check if the traffic recording service is running
--! @param ifid the interface identifier 
--! @return true if the service is running, false otherwise
function recording_utils.isActive(ifid)
  if not recording_utils.isAvailable() then return false end

  local confifname = getConfigInterfaceName(ifid)
  return os_utils.isActive("n2disk", confifname)
end

--! @brief Start (or restart) the traffic recording service
--! @param ifid the interface identifier 
function recording_utils.restart(ifid)
  local confifname = getConfigInterfaceName(ifid)
  os_utils.enableService("n2disk", confifname)
  os_utils.restartService("n2disk", confifname)
end

--! @brief Stop the traffic recording service
--! @param ifid the interface identifier 
function recording_utils.stop(ifid)
  local confifname = getConfigInterfaceName(ifid)
  os_utils.stopService("n2disk", confifname)
  os_utils.disableService("n2disk", confifname)
end

--! @brief Return the log trace of the traffic recording service (n2disk)
--! @param ifid the interface identifier 
--! @param rows the number of lines to return
--| @note lines are retuned in reverse order (most recent line first)
--! @return the log trace
function recording_utils.log(ifid, rows)
  local confifname = getConfigInterfaceName(ifid)
  local output = n2diskctl("log", confifname, "|tail -n", rows, "|tac")
  
  return output
end

--! @brief Return statistics from the traffic recording service (n2disk)
--! @param ifid the interface identifier 
--! @return the statistics
function recording_utils.stats(ifid)
  local confifname = getConfigInterfaceName(ifid)
  local stats = {}
  local proc_stats = n2diskctl("stats", confifname)
  local lines = split(proc_stats, "\n")

  for i = 1, #lines do
    local pair = split(lines[i], ": ")
    if pair[1] ~= nil and pair[2] ~= nil then
      stats[pair[1]] = trimString(pair[2])
    end
  end
  return stats
end

-- Read information about used disk space for all ntopng interfaces 
-- (skipping the provided ifid if not nil)
local function allInterfacesStorageUsage(ifid)
  local info = { reserved_disk_space = 0, used_disk_space = 0, delta_disk_space = 0 }
  local ntopng_interfaces = interface.getIfNames()
  for id,name in pairs(ntopng_interfaces) do
    if ifid == nil or id ~= ifid then
      local enabled = ntop.getCache('ntopng.prefs.ifid_'..id..'.traffic_recording.enabled')
      if enabled == "true" then
        local disk_space = ntop.getCache('ntopng.prefs.ifid_'..id..'.traffic_recording.disk_space')
        local if_disk_space = 0
        if not isEmptyString(disk_space) then
          if_disk_space = tonumber(disk_space)
        end
        info.reserved_disk_space = info.reserved_disk_space + if_disk_space
        info.used_disk_space = info.used_disk_space + interfaceStorageUsed(id)
      end
    end
  end
  if info.reserved_disk_space > info.used_disk_space then
    info.delta_disk_space = info.reserved_disk_space - info.used_disk_space -- available, to be used
  end
  return info
end

function recording_utils.recommendedSpace(ifid, storage_info)
  -- available disk space
  local avail = storage_info.avail + storage_info.if_used

  -- compute available disk space based on space reserved by other interfaces
  local current = allInterfacesStorageUsage(ifid)
  if avail > current.delta_disk_space then
    avail = avail - current.delta_disk_space
  else
    avail = 0
  end

  local recommended = avail - (avail*0.2)
  return math.floor(recommended)
end

--! @brief Check if there is pcap data for a specified time interval (fully included in the dump window) 
--! @param ifid the interface identifier 
--! @param epoch_begin the begin time (epoch)
--! @param epoch_end the end time (epoch)
--! @return a table with 'available' = true if the specified interval is included in the dump window, 'epoch_begin'/'epoch_end' are also returned with the actual available window.
function recording_utils.isDataAvailable(ifid, epoch_begin, epoch_end)
  local info = {}
  info.available = false
  if recording_utils.isEnabled(ifid) then
    local stats = recording_utils.stats(ifid)
    if stats['FirstDumpedEpoch'] ~= nil and stats['LastDumpedEpoch'] ~= nil then
      local first_epoch = tonumber(stats['FirstDumpedEpoch'])
      local last_epoch = tonumber(stats['LastDumpedEpoch'])
      if first_epoch > 0 and last_epoch > 0 and 
         epoch_end > first_epoch and epoch_begin < last_epoch then
        info.epoch_begin = epoch_begin
        info.epoch_end = epoch_end
        if first_epoch > epoch_begin then
          info.epoch_begin = first_epoch
        end
        if last_epoch < epoch_end then
          info.epoch_end = last_epoch
        end
        info.available = true
      end
    end
  end
  return info
end

--! @brief Return the list of pcap files extracted for a job
--! @param job_id the job identifier 
--! @return the list of pcap files (paths)
function recording_utils.getJobFiles(id)
  local job_json = ntop.getHashCache(extraction_jobs_key, id)
  local files = {}
  if not isEmptyString(job_json) then
    local job = json.decode(job_json)
    local file_id = 1
    local file = getPcapFilePath(job.id, job.ifid, file_id)
    while ntop.exists(file) do
      table.insert(files, file)
      file_id = file_id + 1
      file = getPcapFilePath(job.id, job.ifid, file_id) 
    end
  end
  return files
end

--! @brief Delete an extraction job and its pcap data on disk, if any
--! @param job_id the job identifier 
function recording_utils.deleteJob(job_id)
  local job_json = ntop.getHashCache(extraction_jobs_key, job_id)
  if not isEmptyString(job_json) then
    ntop.delHashCache(extraction_jobs_key, tostring(job_id))
    local job = json.decode(job_json)
    local dir_path = getPcapFileDir(job.id, job.ifid)
    ntop.rmdir(dir_path)
  end
end

--! @brief Delete and stop all the extraction jobs for the specified interface.
--! @param ifid the interface identifier
function recording_utils.deleteAndStopAllJobs(ifid)
  for _, job in pairs(recording_utils.getExtractionJobs(ifid)) do
    if job.status == "completed" then
      recording_utils.deleteJob(job.id)
    else
      recording_utils.stopJob(job.id)
    end
  end
end

--! @brief Return statistics about the extraction jobs.
--! @param ifid the interface identifier 
--! @return the jobs statistics (ready, total)
function recording_utils.extractionJobsInfo(ifid)
  local job_ids = ntop.getHashKeysCache(extraction_jobs_key) or {}
  local jobs_info = { total = 0, ready = 0 }

  for id,_ in pairs(job_ids) do
    local job_json = ntop.getHashCache(extraction_jobs_key, id)
    local job = json.decode(job_json)
    if ifid == nil or job.ifid == ifid then
      if job.status == "completed" or job.status == "failed" then
        jobs_info.ready = jobs_info.ready + 1
      end
      jobs_info.total = jobs_info.total + 1
    end
  end

  return jobs_info
end

--! @brief Return the list of scheduled extraction jobs.
--! @param ifid the interface identifier
--! @return the list of jobs
function recording_utils.getExtractionJobs(ifid)
  local jobs = {}
  local job_ids = ntop.getHashKeysCache(extraction_jobs_key) or {}

  for id,_ in pairs(job_ids) do
    local job_json = ntop.getHashCache(extraction_jobs_key, id)
    local job = json.decode(job_json)
    if ifid == nil or job.ifid == ifid then
      jobs[tonumber(id)] = job
    end
  end

  return jobs
end

--! @brief Stop a running extraction job.
--! @param job_id the job identifier 
function recording_utils.stopJob(job_id)
  ntop.rpushCache(extraction_stop_queue_key, job_id)
end

--! @brief Schedule a new extraction job.
--! @param ifid the interface identifier 
--! @param params the extraction parameters. time_from/time_to (epoch) are mandatory. filter (nBPF format) is optional.
--! @return the newly created job 
function recording_utils.scheduleExtraction(ifid, params)

  if params.time_from == nil or params.time_to == nil then
    return nil
  end
  if params.filter == nil then
    params.filter = ""
  end

  local id = ntop.incrCache(extraction_seqnum_key)

  local job = {
    id = id,
    ifid = tonumber(ifid),
    time = os.time(),
    status = 'waiting',
    time_from = tonumber(params.time_from),
    time_to = tonumber(params.time_to),
    filter = params.filter,
    chart_url = params.chart_url,
  }

  ntop.setHashCache(extraction_jobs_key, job.id, json.encode(job))

  ntop.rpushCache(extraction_queue_key, tostring(job.id))
  
  local job_info = { id = job.id }
  return job_info
end

local function setStuckJobsAsFailed()
  local jobs = {}
  local job_ids = ntop.getHashKeysCache(extraction_jobs_key) or {}

  for id,_ in pairs(job_ids) do
    local job_json = ntop.getHashCache(extraction_jobs_key, id)
    local job = json.decode(job_json)
    if job.status == "processing" then
      job.status = "failed"
      job.error_code = 9 -- stuck
      ntop.setHashCache(extraction_jobs_key, job.id, json.encode(job)) 
    end
  end
end

local function setJobAsCompleted()
  local datapath_extractions = ntop.getExtractionStatus()
  for id,status in pairs(datapath_extractions) do
    local job_json = ntop.getHashCache(extraction_jobs_key, id)
    if not isEmptyString(job_json) then
      local job = json.decode(job_json)
      if job.status == "processing" then
        if status.status == 0 then
          job.status = "completed"
        else
          job.status = "failed"
          job.error_code = status.status
        end
        job.extracted_pkts = status.extracted_pkts
        job.extracted_bytes = status.extracted_bytes
        ntop.setHashCache(extraction_jobs_key, job.id, json.encode(job)) 
      end
    end
  end
  setStuckJobsAsFailed()
end

-- Manages extraction jobs. This is called from a single, periodic script (housekeeping.lua) 
function recording_utils.checkExtractionJobs()

  -- stop extractions for stopped jobs, if any
  if ntop.isExtractionRunning() then
    local id = ntop.lpopCache(extraction_stop_queue_key)
    if not isEmptyString(id) then
      local job_json = ntop.getHashCache(extraction_jobs_key, id)
      if not isEmptyString(job_json) then
        local job = json.decode(job_json)

        ntop.stopExtraction(job.id)

        job.status = 'stopped'
        ntop.setHashCache(extraction_jobs_key, job.id, json.encode(job))
      end
    end
  end

  if not ntop.isExtractionRunning() then
    -- set the previous job as completed, if any
    setJobAsCompleted()

    -- run a new extraction job
    local id = ntop.lpopCache(extraction_queue_key)
    if not isEmptyString(id) then

      local job_json = ntop.getHashCache(extraction_jobs_key, id)
      if not isEmptyString(job_json) then
        local job = json.decode(job_json)

        -- computing available space as safety check
        local extraction_limit = 0
        local storage_info = recording_utils.storageInfo(job.ifid)
        local usage = allInterfacesStorageUsage(nil)
        if storage_info.avail > usage.delta_disk_space then
          local avail = storage_info.avail - usage.delta_disk_space
          extraction_limit = math.floor(avail*1024*1024)
        else
          extraction_limit = nil
        end 

        if extraction_limit ~= nil then
          -- running extraction
          ntop.runExtraction(job.id, 
            tonumber(job.ifid), 
            tonumber(job.time_from), 
            tonumber(job.time_to), 
            job.filter, 
            extraction_limit)

          job.status = 'processing'
          ntop.setHashCache(extraction_jobs_key, job.id, json.encode(job))
        else
          -- no space available - delay this job
          job.status = 'waiting_nospace'
          ntop.setHashCache(extraction_jobs_key, job.id, json.encode(job))
          ntop.rpushCache(extraction_queue_key, tostring(job.id))
        end
      end
    end
  end
end

-- #################################

return recording_utils