<commit_before>-- This Source Code Form is subject to the terms of the Mozilla Public
-- License, v. 2.0. If a copy of the MPL was not distributed with this
-- file, You can obtain one at http://mozilla.org/MPL/2.0/.

-- See https://wiki.mozilla.org/CloudServices/DataPipeline/HTTPEdgeServerSpecification

require "cjson"
require "lpeg"
require "string"
require "table"
require 'geoip.city'

-- Split a path into components. Multiple consecutive separators do not
-- result in empty path components.
-- Examples:
--   /foo/bar      ->   {"foo", "bar"}
--   ///foo//bar/  ->   {"foo", "bar"}
--   foo/bar/      ->   {"foo", "bar"}
--   /             ->   {}
local sep = lpeg.P("/")
local elem = lpeg.C((1 - sep)^1)
local path_grammar = lpeg.Ct(elem^0 * (sep^0 * elem)^0)

local function split_path(s)
    if not s then
        return {}
    end
    return lpeg.match(path_grammar, s)
end

local city_db = assert(geoip.city.open(read_config("geoip_city_db")))
local UNK_GEO = "??"

function get_geo_country()
    local country
    local ipaddr = read_message("Fields[X-Forwarded-For]")
    if ipaddr then
        country = city_db:query_by_addr(ipaddr, "country_code")
    end
    if country then return country end
    ipaddr = read_message("Fields[RemoteAddr]")
    if ipaddr then
        country = city_db:query_by_addr(ipaddr, "country_code")
    end
    return country or UNK_GEO
end

-- Load the namespace configuration externally.
-- Note that if the config contains invalid JSON, we will discard any messages
-- we receive with the following error:
--    FATAL: process_message() function was not found
local ok, ns_config = pcall(cjson.decode, read_config("namespace_config"))
if not ok then return -1, ns_config end


local msg = {
    Timestamp   = nil,
    Type        = "http_edge_incoming",
    Payload     = nil,
    EnvVersion  = nil,
    Hostname    = nil,
    Fields      = {}
}

function process_message()
    -- Reset Fields, since different namespaces may use different fields.
    msg.Fields = {}

    -- Carry forward payload some incoming fields.
    msg.Payload = read_message("Payload")
    msg.Timestamp = read_message("Timestamp")
    msg.EnvVersion = read_message("EnvVersion")

    -- Hostname is the host name of the server that received the message.
    msg.Hostname = read_message("Hostname")

    -- Host is the name of the HTTP endpoint the client used (such as
    -- "incoming.telemetry.mozilla.org").
    msg.Fields.Host = read_message("Fields[Host]")

    -- Path should be of the form:<commit_msg>Remove extra line<commit_after>-- This Source Code Form is subject to the terms of the Mozilla Public
-- License, v. 2.0. If a copy of the MPL was not distributed with this
-- file, You can obtain one at http://mozilla.org/MPL/2.0/.

-- See https://wiki.mozilla.org/CloudServices/DataPipeline/HTTPEdgeServerSpecification

require "cjson"
require "lpeg"
require "string"
require "table"
require 'geoip.city'

-- Split a path into components. Multiple consecutive separators do not
-- result in empty path components.
-- Examples:
--   /foo/bar      ->   {"foo", "bar"}
--   ///foo//bar/  ->   {"foo", "bar"}
--   foo/bar/      ->   {"foo", "bar"}
--   /             ->   {}
local sep = lpeg.P("/")
local elem = lpeg.C((1 - sep)^1)
local path_grammar = lpeg.Ct(elem^0 * (sep^0 * elem)^0)

local function split_path(s)
    if not s then
        return {}
    end
    return lpeg.match(path_grammar, s)
end

local city_db = assert(geoip.city.open(read_config("geoip_city_db")))
local UNK_GEO = "??"

function get_geo_country()
    local country
    local ipaddr = read_message("Fields[X-Forwarded-For]")
    if ipaddr then
        country = city_db:query_by_addr(ipaddr, "country_code")
    end
    if country then return country end
    ipaddr = read_message("Fields[RemoteAddr]")
    if ipaddr then
        country = city_db:query_by_addr(ipaddr, "country_code")
    end
    return country or UNK_GEO
end

-- Load the namespace configuration externally.
-- Note that if the config contains invalid JSON, we will discard any messages
-- we receive with the following error:
--    FATAL: process_message() function was not found
local ok, ns_config = pcall(cjson.decode, read_config("namespace_config"))
if not ok then return -1, ns_config end

local msg = {
    Timestamp   = nil,
    Type        = "http_edge_incoming",
    Payload     = nil,
    EnvVersion  = nil,
    Hostname    = nil,
    Fields      = {}
}

function process_message()
    -- Reset Fields, since different namespaces may use different fields.
    msg.Fields = {}

    -- Carry forward payload some incoming fields.
    msg.Payload = read_message("Payload")
    msg.Timestamp = read_message("Timestamp")
    msg.EnvVersion = read_message("EnvVersion")

    -- Hostname is the host name of the server that received the message.
    msg.Hostname = read_message("Hostname")

    -- Host is the name of the HTTP endpoint the client used (such as
    -- "incoming.telemetry.mozilla.org").
    msg.Fields.Host = read_message("Fields[Host]")

    -- Path should be of the form:
{'fraction_non_alphanumeric': 0.11551350285826927, 'fraction_numerical': 0.003942440370589394, 'mean_word_length': 2.9827315541601256, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3620340', 'n_tokens_mistral': 1592, 'n_tokens_neox': 1552, 'n_words': 505}