csvlua

local dquote = ('"'):byte()
local function extract_simple(record, start)
    local index = record:find(',', start)
    local stop_index
    local next_offset
    if index ~= nil then
        stop_index = index - 1
        next_offset = index + 1
    end
    return start, stop_index, next_offset
end
local function extract_quoted(record, start)
    start = start + 1
    local offset = start
    while true do
        local index = record:find('"', offset)
        local next_index = index + 1
        local next_byte = record:byte(next_index)
        if next_byte ~= dquote then
            -- found the end index of the field, return it
            return start, index - 1, index + 2
        end 
        offset = index + 2 -- advance both dquotes
    end
end
local function split_csv(record)
    local rv = {}
    local offset = 1
    while offset ~= nil do
        local start_idx
        local stop_idx
        if record:byte(offset) == dquote then
            start_idx, stop_idx, offset = extract_quoted(record, offset)
        else
            start_idx, stop_idx, offset = extract_simple(record, offset)
        end
        table.insert(rv, record:sub(start_idx, stop_idx))
    end
    return rv
end
-- declare a module variable to hold header
local headers = {}
local function parse_message(message, target)
  if not message then
    return
  end
  for  k, v in message:gmatch('(%S+)=(%S+),?') do
    target[k:gsub('^{', '')] = v:gsub('}$', ''):gsub(',$', '')
  end
end
local function parse_ms(record, key)
  if record[key] ~= nil then
    record[key] = tonumber(record[key]:gsub('ms$',''), 10)  
  end
end

local function blank_url(record)
  local key = 'BO-API-URL'
  if record[key] == nil then
    return
  end
  record[key] = record[key]:gsub('/[^/]+(/?)', '/-%1', 1)
end

function cb_filter(tag, timestamp, record)
    if headers[tag] == nil then
      headers[tag] = split_csv(record._raw)
      -- return -1 to drop the header
      return -1, timestamp, record
    end
    local data = {}
    local fields = split_csv(record._raw)
    for index, header in ipairs(headers[tag]) do
      data[header] = fields[index]
    end
    data.accountid = nil
    parse_message(data.message, data)
    blank_url(data)
    data['date'] = nil
    data.view = nil
    data.encoding = nil
    data.accountId = nil
    data.region = nil
    data.sourceName = nil
    data.sourceHost = nil
    data.sourceCategory = nil
    data.format = nil
    data.sourceId = nil
    data.instanceid = nil
    data.receiptTime = nil
    data['instana.trace.id'] = nil
    data.collectorId = nil
    data['RESPONSE-LENGTH'] = nil
    data.count = nil
    data.timestamp = data.messageTime/1000
    -- data.message = nil
    parse_ms(data, 'PROCESSING-TIME')
    -- return "1" to modify timestamp
    return 1, data.timestamp, data
end