This Lua script is used with Fluent Bit to match time and level information in log messages.

filter.lua

local month_table = {
    Jan = 1, Feb = 2, Mar = 3, Apr = 4, May = 5, Jun = 6,
    Jul = 7, Aug = 8, Sep = 9, Oct = 10, Nov = 11, Dec = 12
}
local time_pat_table = {
    redis = "(%d+)%s([JFMASOND]%l%l)%s(%d+)%s(%d+):(%d+):(%d+)%.(%d+)",
    nginx = "%[?(%d+)/([JFMASOND%d][%l%d]+)/(%d+)[%s:](%d+):(%d+):(%d+)[%s%+%d:%]]*",
    mysql = "(%d+)%-(%d+)%-(%d+)T(%d+):(%d+):(%d+)%.(%d+)Z%s+",
    postgres = "(%d+)%-(%d+)%-(%d+)%s(%d+):(%d+):(%d+)%.(%d+)%sGMT%s+",
    rabbitmq = "(%d+)%-(%d+)%-(%d+)%s(%d+):(%d+):(%d+)%.(%d+)[%+%d:]*",
    java = "%[?(%d+)%-(%d+)%-(%d+)[%sT](%d+):(%d+):(%d+)[,%.](%d+)[%+%d:%]]*",
    go = "(%d+)%-(%d+)%-(%d+)T(%d+):(%d+):(%d+)%.(%d+)[%+%d:Z]*"
}
local level_pat_table = {
    redis = "%s([%.%-%*#])%s",
    nginx = "%[([dinweca]%l+[goenrt])%]%s",
    mysql = "%s+%[([SNWE]%l+[megr])%]",
    postgres = "%s+([DIWEFLHSN]%u+[GONRLTE]):",
    rabbitmq = "%s%[([dinwec]%l+[goenrl])%]%s",
    java = "%s+([TDIWEF]%u+[EGONRL])%s+",
    go = "%s+([TDIWEFP]%u+[GONRLC])%s+"
}
local level_table = {
    TRACE = 5, DEBUG = 10, INFO = 20, WARN = 30, ERROR = 40, FATAL = 40,
    -- nginx
    debug = 10, info = 20 , notice = 20, warn = 30, error = 40, crit = 40, alert = 40, emerg = 40,
    -- mysql https://dev.mysql.com/doc/refman/8.4/en/error-log-format.html
    System = 5, Note = 20, Warning = 30, Error = 40,
    -- Postgres https://www.postgresql.org/docs/current/runtime-config-logging.html
    LOG = 5, HINT = 5, DETAIL = 10, STATEMENT = 10, NOTICE = 20, WARNING = 30,
    -- RabbitMQ https://www.rabbitmq.com/docs/logging#log-levels
    warning = 30, critical = 40,
    -- Golang
    DPANIC = 40, PANIC = 40,
}
-- Redis https://build47.com/redis-log-format-levels/
level_table["."] = 10
level_table["-"] = 20
level_table["*"] = 20
level_table["#"] = 30

-- match time in log message
local function match_ts(message, pat)
    local st, ed, year, month, day, hour, min, sec, msec = string.find(message, pat)
    if st then
        local abbr = month_table[month]
        local args = {year=year, month=month, day=day, hour=hour, min=min, sec=sec}
        if abbr then
            args = {year=day, month=abbr, day=year, hour=hour, min=min, sec=sec}
        end

        local precision = 0
        if msec then
            local msec_len = string.len(msec)
            precision = msec / (10 ^ msec_len)
        end
        local status, ts = pcall(os.time, args)
        if status then
            return ts + precision, st, ed
        else
            return 0, st, ed
        end
    end

    return 0, 1, 1
end

-- match level in log message
-- returns level in integer, where it ends
local function match_level(message, pat, init)
    local st, ed, level = string.find(message, pat, init)
    if st then
        local tin = level_table[level]
        if tin then
            return tin, st, ed
        else
            return 0, st, ed
        end
    end

    return 0, init, init
end

local function get_minute()
    local now = os.time()
    local remainder = now % 60
    if remainder == 0 then
        return now
    else
        return now - remainder
    end
end

-- counter reference https://github.com/fluent/fluent-bit/blob/master/scripts/rate_limit.lua
local last_min = 0
local counter = 0
function modify(tag, timestamp, record)
    local current_min = get_minute()
    if current_min == last_min then -- the same minute
        counter = counter + 1
    else
        last_min = current_min
        counter = 1 -- reset the counter within a minute
    end
    record["tie_breaker_id"] = counter

    if record["message"] == nil then
        return -1, timestamp, record
    end
    if record["host"] then
        if string.sub(record["host"], 1, 1) == "/" then
            record["host"] = string.sub(record["host"], 2)
        end
    end

    local lang = string.match(tag, "logs%-(%l+)")
    if not lang then
        record["level"] = level_table["INFO"]
        return 2, timestamp, record
    end

    local time_pat = time_pat_table[lang]
    local level_pat = level_pat_table[lang]
    if not time_pat or not level_pat then
        return 2, timestamp, record
    end

    local ts, tst, ted = match_ts(record["message"], time_pat)
    local level, lst, led = match_level(record["message"], level_pat, ted)

    if level > 0 then
        record["level"] = level
    else
        if ts > 0 then
            record["level"] = level_table["INFO"]
        else
            record["level"] = level_table["ERROR"]
        end
    end

    local head = ""
    if tst > 1 then -- there is message ahead of time part.
        head = string.sub(record["message"], 1, tst - 1)
    end
    if lst > ted + 1 then -- there is message between time and level.
        if ted > 1 then -- time matched
            ted = ted + 1
        end

        head = head .. string.sub(record["message"], ted, lst - 1)
    end
    if led > 1 then -- level matched
        led = led + 1
    end
    record["message"] = head .. string.sub(record["message"], led)
    if ts > 0 then
        return 1, ts, record
    else
        return 2, timestamp, record
    end
end

filter_test.lua

require("filter")

function test_redis()
    local code, ts, record = modify("docker-logs-redis", 123.456, {
        message = "577:C 29 Sep 2025 12:35:29.032 * RDB: 0 MB of memory used by copy-on-write"
    })
    print(record["level"])
    print(record["message"])
    print(ts)

    code, ts, record = modify("docker-logs-redis", 1.23, {
        message = "1:M 15 Oct 2025 11:34:39.971 # WARNING Memory overcommit must be enabled! "
    })
    print(record["level"])
    print(record["message"])
    print(ts)
end

print("Test: Redis")
test_redis()

function test_mysql()
    local code, ts, record = modify("docker-logs-mysql", 123.456, {
        -- message = "2024-07-23T10:45:19.765700Z 0 [System] [MY-011323] [Server] X Plugin ready for connections. Bind-address: '::' port: 33060, socket: /var/lib/mysql/mysqlx.sock"
        -- message = "mbind: Operation not permitted"
        message = "2024-07-23T10:45:19.548020Z 0 [Warning] [MY-013829] [Server] Missing data directory for ICU regular expressions: /usr/lib64/mysql/private/."
    })
    print(record["level"])
    print(record["message"])
    print(ts)

    code, ts, record = modify("docker-logs-mysql", 1.23, {
        -- message = "# Time: 2025-10-13T17:21:13.326218Z"
        message = "# Query_time: 71.790903  Lock_time: 0.000002  Rows_sent: 0  Rows_examined: 1  Rows_affected: 0  Bytes_sent: 52"
        -- message = "UPDATE enterprise_qualification SET `choice` = 1, `preliminary` = 1, `reexamine` = 0, `date_created` = '2024-12-04 14:53:15', `date_updated` = '2025-01-16 16:51:25' WHERE clause_id = 148335581905092608 AND enterprise_id = 67059297074282496 LIMIT 1;"
    })
    print(record["level"])
    print(record["message"])
    print(ts)
end

print("\nTest: MySQL")
test_mysql()

function test_postgres()
    local code, ts, record = modify("docker-logs-postgres", 123.456, {
        -- message = "2025-10-11 06:32:08.050 GMT [4943] DETAIL:  Key (identifier_hash)=(mW-xk2ARhvoQWPw4dc6mSQ) already exists."
        message = "2025-10-16 02:48:26.805 GMT [28487] STATEMENT:  insert into cwd_token (directory_id, entity_name, random_number, identifier_hash, random_hash, created_date, last_accessed_date, last_accessed_time, duration, id) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)"
    })
    print(record["level"])
    print(record["message"])
    print(ts)

    code, ts, record = modify("docker-logs-postgres", 1.23, {
        message = "2025-09-29 13:25:59.398 GMT [1] HINT:  Future log output will appear in directory /var/log/postgresql."
    })
    print(record["level"])
    print(record["message"])
    print(ts)
end

print("\nTest: Postgres")
test_postgres()

function test_rabbitmq()
    local code, ts, record = modify("docker-logs-rabbitmq", 123.456, {
        -- message = "2025-05-20 05:54:11.273089+00:00 [notice] <0.342.0> WAL: ra_coordination_log_wal init, open tbls: ra_coordination_log_open_mem_tables, closed tbls: ra_coordination_log_closed_mem_tables"
        message = "2025-05-06 12:55:20.181907+00:00 [info] <0.435.0> Making sure data directory '/var/lib/rabbitmq/mnesia/rabbit@rabbitmq/msg_stores/vhosts/628WB79CIFDYO9LJI6DKMI09L' for vhost '/' exists"
    })
    print(record["level"])
    print(record["message"])
    print(ts)

    code, ts, record = modify("docker-logs-rabbitmq", 1.23, {
        message = "2025-05-06 12:55:20.188295+00:00 [warning] <0.444.0> Message store 628WB79CIFDYO9LJI6DKMI09L/msg_store_persistent: rebuilding indices from scratch"
    })
    print(record["level"])
    print(record["message"])
    print(ts)
end

print("\nTest: RabbitMQ")
test_rabbitmq()

function test_nginx()
    local code, ts, record = modify("docker-logs-nginx", 123.456, {
        message = "127.0.0.1 - - [16/Oct/2025:00:00:31 +0800] GET /nginx_status HTTP/1.1 200 113 - Netdata go.d.plugin/v2.0.1"
    })
    print(record["level"])
    print(record["message"])
    print(ts)

    code, ts, record = modify("docker-logs-nginx", 1.23, {
        message = "2025/10/16 07:41:56 [crit] 1082#1082: *946885 SSL_do_handshake() failed (SSL: error:0A00006C:SSL routines::bad key share) while SSL handshaking, client: 195.178.110.15, server: 0.0.0.0:443"
    })
    print(record["level"])
    print(record["message"])
    print(ts)
end

print("\nTest: nginx")
test_nginx()

function test_java()
    code, ts, record = modify("docker-logs-java", 1.23, {
        host = "/zookeeper.test",
        message = "[2025-10-15 10:12:44,476] TRACE [Controller id=1] Leader imbalance ratio for broker 1 is 0.0 (kafka.controller.KafkaController)"
    })
    print(record["host"])
    print(record["level"])
    print(record["message"])
    print(ts)

    code, ts, record = modify("docker-logs-java", 1.23, {
        message = "2025-09-25 21:45:10.000  WARN 1 --- [           main] ConfigServletWebServerApplicationContext : Exception encountered during context initialization - cancelling refresh attempt: org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'entityManagerFactory' defined in class path resource [org/springframework/boot/autoconfigure/orm/jpa/HibernateJpaConfiguration.class]: Unable to create requested service [org.hibernate.engine.jdbc.env.spi.JdbcEnvironment]"
        -- message = "2025-09-29 13:27:22,640 [myid:1] - WARN  [main:ConstraintSecurityHandler@759] - ServletContext@o.e.j.s.ServletContextHandler@f1da57d{/,null,STARTING} has uncovered http methods for path: /*"
    })
    print(record["level"])
    print(record["message"])
    print(ts)

    code, ts, record = modify("docker-logs-java", 1.23, {
        message = "2025-09-29T21:28:23.920+08:00  INFO 1 --- [           main] o.youxin.identity.IdentityApplication    : Started IdentityApplication in 7.525 seconds (process running for 9.075)"
    })
    print(record["level"])
    print(record["message"])
    print(ts)
end

print("\nTest: Java")
test_java()

function test_go()
    local code, ts, record = modify("docker-logs-go", 123.456, {
        message = "2025-08-13T18:03:30.344+0800	 FATAL	event/voter.go:319"
    })
    print(record["level"])
    print(record["message"])
    print(os.date("%Y-%m-%d %H:%M:%S", math.floor(ts)))

    code, ts, record = modify("docker-logs-go", 1.23, {
        message = "2025-05-20T05:54:34.674Z	ERROR	livekit	service/server.go:211"
    })
    print(record["level"])
    print(record["message"])
    print("tie_breaker_id", record["tie_breaker_id"])
    print(ts)
    print(os.date("%Y-%m-%d %H:%M:%S", math.floor(ts)))
end

print("\nTest: Golang")
test_go()