文档 Grafana Alloy 参考 组件 loki loki.process

loki.process

loki.process 从其他 loki 组件接收日志条目，应用一个或多个处理 阶段，并将结果转发到组件参数中指定的接收器列表。

阶段是一个多用途的工具，可以在将日志条目传递给下游组件之前解析、转换和过滤它们。这些阶段按照在配置文件中出现的顺序应用于每个日志条目。所有在loki.process块内的阶段都可以访问日志条目的标签集、日志行、日志时间戳，以及一个共享的'提取'值映射，以便一个阶段的结果可以在后续阶段中使用。

可以通过为它们提供不同的标签来指定多个loki.process组件。

用法

alloy 复制

loki.process "LABEL" {
  forward_to = RECEIVER_LIST

  stage.STAGENAME {
    ...
  }
  ...
}

参数

loki.process支持以下参数

块

以下块可以在loki.process的定义内部使用

用户可以在loki.process内部提供任意数量的这些阶段块，它们将按照配置文件中出现的顺序执行。

stage.cri块

stage.cri内块启用了一个预定义的管道，该管道使用CRI日志格式读取日志行。

以下参数被支持

只有当 max_partial_line_size_truncate 设置为 true 时，max_partial_line_size 才会被考虑。

alloy 复制

stage.cri {}

CRI 以空格分隔的值指定日志行，以下为组件：

• time：日志的时间戳字符串
• stream：可以是 stdout 或 stderr
• flags：CRI 标志，包括 F 或 P
• log：日志行的内容

给定以下日志行，将在提取数据的共享映射中创建后续的关键值对

复制

"2019-04-30T02:12:41.8443515Z stdout F message"

content: message
stream: stdout
timestamp: 2019-04-30T02:12:41.8443515

stage.decolorize block

stage.decolorize 从日志行中移除 ANSI 颜色代码，从而使得进一步解析日志更容易。

stage.decolorize 块不支持任何参数或内部块，因此它始终为空。

alloy 复制

stage.decolorize {}

stage.decolorize 将带有颜色代码的每一行转换为无颜色行，例如

复制

[2022-11-04 22:17:57.811] \033[0;32http\033[0m: GET /_health (0 ms) 204

转换为

复制

[2022-11-04 22:17:57.811] http: GET /_health (0 ms) 204

stage.docker block

stage.docker 内部块启用一个预定义的管道，该管道以 Docker 日志文件的规范格式读取日志行。

stage.docker 块不支持任何参数或内部块，因此它始终为空。

alloy 复制

stage.docker {}

Docker 日志条目格式为 JSON，包含以下键

• log：日志行的内容
• stream：可以是 stdout 或 stderr
• time：日志行的日期时间字符串

给定以下日志行，将在提取数据的共享映射中创建后续的关键值对

复制

{"log":"log message\n","stream":"stderr","time":"2019-04-30T02:12:41.8443515Z"}

output: log message\n
stream: stderr
timestamp: 2019-04-30T02:12:41.8443515

stage.drop block

stage.drop 内部块配置了一个基于几个选项的过滤阶段，根据这些选项丢弃日志条目。如果提供了多个选项，则将它们视为 AND 子句，并且必须 全部 为真，日志条目才被丢弃。要使用 OR 子句丢弃条目，请连续指定多个 drop 块。

以下参数被支持

expression 字段必须是 RE2 正则表达式字符串。

• 如果 source 为空或未提供，则正则表达式尝试匹配日志行本身。
• 如果 source 是单个名称，则正则表达式尝试匹配从提取映射中获取的相应值。
• 如果 source 是以逗号分隔的名称列表，则使用 separator 将提取映射中的相应值连接起来，然后正则表达式尝试匹配连接的字符串。

value 字段只能与来自提取映射的值一起使用，并且必须与 source 一起指定。

• 如果 source 是单个名称，则当提取映射中相应值与 value 完全匹配时丢弃条目。
• 如果 source 是以逗号分隔的名称列表，则当 value 匹配使用 separator 连接的 source 值时丢弃条目。

每次丢弃条目时，都会增加指标 loki_process_dropped_lines_total。默认情况下，原因标签是 "drop_stage"，但您可以使用 drop_counter_reason 参数提供自定义标签。

以下阶段会丢弃包含单词debug并且长度超过1KB的日志条目。

alloy 复制

stage.drop {
    expression  = ".*debug.*"
    longer_than = "1KB"
}

以下示例中，我们定义了多个drop块，以便loki.process丢弃24小时或更早、长度超过8KB，或提取的“app”值为foo的条目。

alloy 复制

stage.drop {
    older_than          = "24h"
    drop_counter_reason = "too old"
}

stage.drop {
    longer_than         = "8KB"
    drop_counter_reason = "too long"
}

stage.drop {
    source = "app"
    value  = "foo"
}

stage.eventlogmessage块

eventlogmessage阶段从Windows事件日志中出现的消息字符串中提取数据。

以下参数被支持

当overwrite_existing设置为true时，该阶段将用相同名称覆盖现有提取数据字段。如果设置为false，则将_extracted后缀追加到已存在的字段名称。

当drop_invalid_labels设置为true时，该阶段将丢弃无效标签名称的字段。如果设置为false，阶段将自动将它们转换为有效标签，将无效字符替换为下划线。

与stage.json结合的示例

alloy 复制

stage.json {
    expressions = {
        message = "",
        Overwritten = "",
    }
}

stage.eventlogmessage {
    source = "message"
    overwrite_existing = true
}

给定以下日志行

复制

{"event_id": 1, "Overwritten": "old", "message": "Message type:\r\nOverwritten: new\r\nImage: C:\\Users\\User\\alloy.exe"}

第一阶段会在提取数据的集合中创建以下键值对

• message: 消息类型:\r\n已覆盖: 新\r\n映像: C:\Users\User\alloy.exe
• Overwritten: 旧

第二阶段将从提取数据中解析message的值，并将以下键值对追加/覆盖到提取数据的集合中

• Image: C:\\Users\\User\\alloy.exe
• Message_type: (空字符串)
• Overwritten: 新

stage.json块

stage.json内部块配置一个JSON处理阶段，该阶段解析传入的日志行或先前提取的值作为JSON，并使用JMESPath表达式从它们中提取新的值。

以下参数被支持

配置JSON阶段时，source字段定义要解析为JSON的数据源。默认情况下，这是日志行本身，但它也可以是先前提取的值。

expressions字段是运行的表达式键值对集合。映射键定义了提取数据时使用的名称，而映射值是用于填充值的表达式。

以下是一个日志行和两个要运行的JSON阶段的示例。

alloy 复制

{"log":"log message\n","extra":"{\"user\":\"alloy\"}"}

loki.process "username" {
  stage.json {
      expressions = {output = "log", extra = ""}
  }

  stage.json {
      source      = "extra"
      expressions = {username = "user"}
  }
}

在此示例中，第一阶段使用日志行作为源，并在共享映射中填充这些值。空表达式表示使用与键相同的值（因此extra="extra"）。

复制

output: log message\n
extra: {"user": "alloy"}

第二阶段使用extra中的值作为输入，并将以下键值对追加到提取数据的集合中。

复制

username: alloy

stage.label_drop块

内联块 stage.label_drop 配置了一个处理阶段，该阶段从传入的日志条目中删除标签。

以下参数被支持

alloy 复制

stage.label_drop {
    values = [ "kubernetes_node_name", "kubernetes_namespace" ]
}

stage.label_keep 块

内联块 stage.label_keep 配置了一个处理阶段，该阶段将传入日志条目的标签集合过滤到子集。

以下参数被支持

alloy 复制

stage.label_keep {
    values = [ "kubernetes_pod_name", "kubernetes_pod_container_name" ]
}

stage.labels 块

内联块 stage.labels 配置了一个标签处理阶段，该阶段可以读取从提取的值映射中提取的数据，并在传入的日志条目上设置新的标签。

以下参数被支持

在标签阶段中，映射的键定义要设置的标签，值定义如何查找它们。如果值为空，则推断它与键相同。

alloy 复制

stage.labels {
    values = {
      env  = "",         // Sets up an 'env' label, based on the 'env' extracted value.
      user = "username", // Sets up a 'user' label, based on the 'username' extracted value.
    }
}

stage.structured_metadata 块

内联块 stage.structured_metadata 配置了一个阶段，该阶段可以读取从提取的值映射中提取的数据，并将它们作为结构化元数据添加到日志条目中。

以下参数被支持

在 structured_metadata 阶段中，映射的键定义要设置的标签，值定义如何查找它们。如果值为空，则推断它与键相同。

alloy 复制

stage.structured_metadata {
    values = {
      env  = "",         // Sets up an 'env' property to structured metadata, based on the 'env' extracted value.
      user = "username", // Sets up a 'user' property to structured metadata, based on the 'username' extracted value.
    }
}

stage.limit 块

内联块 stage.limit 配置了一个基于多个选项节流的速率限制阶段。

以下参数被支持

速率限制以大小为 burst 的“令牌桶”实现，最初满载，并以每秒 rate 令牌的速度补充。每个接收到的日志条目从桶中消耗一个令牌。当 drop 设置为 true 时，超出速率限制的传入条目将被丢弃，否则它们将排队，直到有更多令牌可用。

alloy 复制

stage.limit {
    rate  = 5
    burst = 10
}

如果设置 by_label_name，则 drop 必须设置为 true。这使阶段能够根据标签的数量而不是行数进行速率限制。

以下示例独立地对每个独特的 namespace 值的条目进行速率限制。没有 namespace 标签的条目不受速率限制。阶段跟踪多达 max_distinct_labels 个独特的值，默认为 10000。

alloy 复制

stage.limit {
    rate  = 10
    burst = 10
    drop  = true

    by_label_name = "namespace"
}

stage.logfmt 块

内联块 stage.logfmt 配置了一个处理阶段，该阶段读取传入的日志行作为 logfmt，并从中提取值。

以下参数被支持

source 字段定义要解析为 logfmt 的数据源。当 source 缺失或为空时，阶段解析日志行本身，但它也可以用于解析之前提取的值。

此阶段使用 go-logfmt 解包器，因此数值或布尔类型将解包到其正确的形式。阶段不执行任何其他类型转换。如果提取的值是复杂类型，则将其视为字符串。

让我们看看以下日志行和阶段是如何工作的。

复制

time=2012-11-01T22:08:41+00:00 app=loki level=WARN duration=125 message="this is a log line" extra="user=foo"

stage.logfmt {
    mapping = { "extra" = "" }
}

stage.logfmt {
    mapping = { "username" = "user" }
    source  = "extra"
}

第一个阶段解析日志行本身，并在提取的数据集中插入 extra 键，其值为 user=foo。

第二个阶段解析 extra 的内容，并将 username: foo 键值对追加到提取的数据集中。

stage.luhn 块

内联块 stage.luhn 配置了一个处理阶段，该阶段读取传入的日志行，并删除与 Luhn 算法匹配的字符串。

卢恩算法（Luhn algorithm）是一种简单的校验和公式，用于验证各种识别号，例如信用卡号、IMEI号、美国国家提供者识别号和加拿大社会保险号。许多支付卡行业环境要求这些号码被隐藏。

以下参数被支持

source字段定义了要搜索的数据源。当source不存在或为空时，该阶段将解析日志行本身，但它也可以用于解析之前提取的值。

以下示例日志行包含一个批准的信用卡号。

复制

time=2012-11-01T22:08:41+00:00 app=loki level=WARN duration=125 message="credit card approved 4032032513548443" extra="user=foo"

stage.luhn {
    replacement = "**DELETED**"
}

该阶段解析日志行，隐藏信用卡号，并生成以下更新后的日志行

复制

time=2012-11-01T22:08:41+00:00 app=loki level=INFO duration=125 message="credit card approved **DELETED**" extra="user=foo"

stage.match 块

stage.match 内部块配置了一个过滤阶段，可以根据条件应用嵌套的多个处理阶段或丢弃一个条目，当日志条目匹配可配置的 LogQL 流选择器和过滤器表达式时。

以下参数被支持

stage.match 块支持多个 stage.* 内部块，如顶级块。这些用于构建当选择器匹配日志条目的标签和内容时运行的嵌套阶段集。它支持与 loki.process 组件顶级块相同的所有 stage.NAME 块。

如果指定的操作是 "drop"，则每丢弃一行，指标 loki_process_dropped_lines_total 就会增加。默认情况下，原因标签是 "match_stage"，但可以使用 drop_counter_reason 参数提供自定义原因。

让我们看看以下日志行和阶段来实际应用它

复制

{ "time":"2023-01-18T17:08:41+00:00", "app":"foo", "component": ["parser","type"], "level" : "WARN", "message" : "app1 log line" }
{ "time":"2023-01-18T17:08:42+00:00", "app":"bar", "component": ["parser","type"], "level" : "ERROR", "message" : "foo noisy error" }

stage.json {
    expressions = { "appname" = "app" }
}

stage.labels {
    values = { "applbl" = "appname" }
}

stage.match {
    selector = "{applbl=\"foo\"}"

    stage.json {
        expressions = { "msg" = "message" }
    }
}

stage.match {
    selector = "{applbl=\"qux\"}"
    stage.json {
        expressions = { "msg" = "msg" }
    }
}

stage.match {
    selector = "{applbl=\"bar\"} |~ \".*noisy error.*\""
    action   = "drop"

    drop_counter_reason = "discard_noisy_errors"
}

stage.output {
    source = "msg"
}

前两个阶段将日志行解析为 JSON，将 app 值解码到共享提取映射中作为 appname，并使用其值作为 applbl 标签。

第三个阶段使用 LogQL 选择器仅对 applbl="foo" 的行执行嵌套阶段。因此，对于第一行，嵌套 JSON 阶段将 msg="app1 log line" 添加到提取映射中。

第四个阶段使用 LogQL 选择器仅对 applbl="qux" 的行执行；这意味着它不会与任何输入匹配，嵌套 JSON 阶段不会运行。

第五个阶段从将 applbl 设置为 'bar' 且行内容与正则表达式 .*noisy error.* 匹配的行中删除条目。它还使用标签 drop_counter_reason="discard_noisy_errors" 增加指标 loki_process_dropped_lines_total。

最终输出阶段将日志行的内容更改为提取映射中 msg 的值。在这种情况下，第一日志条目的内容更改为 app1 log line。

stage.metrics 块

stage.metrics 内部块配置了一个阶段，允许基于共享提取映射中的值定义和更新指标。创建的指标在 Alloy 的根 /metrics 端点可用。

stage.metrics 块不支持任何参数，并且仅通过多个嵌套内部 metric.* 块进行配置，每个块对应应生成的每个指标。

以下块可以在 stage.metrics 定义内部使用

metric.counter 块

定义一个仅递增的指标。

以下参数被支持

计数器不能同时将match_all设置为true和value。计数器在没有将match_all=true或action=add设置的情况下不能设置count_entry_bytes。有效的action值是inc和add。inc操作对于每个通过过滤器的日志行，将指标值增加1。add操作将提取的值转换为正浮点数，并将其添加到指标中。

metric.gauge块

定义一个值可以上升或下降的仪表指标。

以下参数被支持

有效的action值是inc、dec、set、add或sub。inc和dec分别通过1递增和递减指标的值。如果选择set、add或sub，则提取的值必须可转换为正浮点数，并将其设置为、添加到或从指标的值中减去。

metric.histogram块

定义一个值记录在预定义桶中的直方图指标。

以下参数被支持

指标行为

如果不存在value，则所有传入的日志条目都匹配。

创建的指标上的标签值可以是动态的，这可能导致导出的指标在基数或过时时爆炸，例如，当流停止接收新的日志时。为了避免/metrics端点的无界增长，任何在max_idle_duration内未更新的指标都将被删除。max_idle_duration必须大于或等于"1s"，默认为"5m"。

从日志数据中提取的指标值在内部转换为浮点数。支持以下值

• 整数
• 浮点数
• 字符串 - 支持两种字符串格式
  • 表示浮点数的字符串，例如，“0.804”转换为0.804。
  • 持续时间格式字符串。有效的时间单位是“ns”、“us”、“ms”、“s”、“m”、“h”。该格式的值转换为秒的浮点数，例如，“0.5ms”转换为0.0005。
• 布尔值
  • true转换为1。
  • false转换为0。

以下管道通过使用match_all参数创建一个计数器，该计数器在收到任何日志行时递增。该管道通过使用count_entry_bytes参数创建第二个计数器，该计数器将这些日志行的字节数加起来。

如果24小时内没有接收到新条目，这两个指标将消失，以避免构建不再有用的指标。这两个指标是跟踪日志流数量和字节数的好起点，以识别高流量或高基数数据源。

alloy 复制

stage.metrics {
    metric.counter {
        name        = "log_lines_total"
        description = "total number of log lines"
        prefix      = "my_custom_tracking_"

        match_all         = true
        action            = "inc"
        max_idle_duration = "24h"
    }
}
stage.metrics {
    metric.counter {
        name        = "log_bytes_total"
        description = "total bytes of log lines"
        prefix      = "my_custom_tracking_"

        match_all         = true
        count_entry_bytes = true
        action            = "add"
        max_idle_duration = "24h"
    }
}

在这里，第一阶段使用正则表达式从日志行中提取格式为order_status=<string>的文本。第二阶段，定义一个计数器，根据先前提取的值递增successful_orders_total和failed_orders_total。

alloy 复制

stage.regex {
    expression = "^.* order_status=(?P<order_status>.*?) .*$"
}
stage.metrics {
    metric.counter {
        name        = "successful_orders_total"
        description = "successful orders"
        source      = "order_status"
        value       = "success"
        action      = "inc"
    }
}
stage.metrics {
    metric.counter {
        name        = "failed_orders_total"
        description = "failed orders"
        source      = "order_status"
        value       = "fail"
        action      = "inc"
    }
}

在这个例子中，第一阶段从日志行中提取格式为 retries=<value> 的文本。第二阶段创建一个仪表，其当前指标值会增加从重试字段中提取的数字。

alloy 复制

stage.regex {
    expression = "^.* retries=(?P<retries>\\d+) .*$"
}
stage.metrics {
    metric.gauge {
        name        = "retries_total"
        description = "total_retries"
        source      = "retries"
        action      = "add"
    }
}

以下示例显示了读取 response_time 的直方图，并将其放入一个桶中，同时增加该桶的计数和特定桶的总和。

alloy 复制

stage.metrics {
    metric.histogram {
        name        = "http_response_time_seconds"
        description = "recorded response times"
        source      = "response_time"
        buckets     = [0.001,0.0025,0.005,0.010,0.025,0.050]
    }
}

阶段的多行块

stage.multiline 内部块在将其传递到管道中的下一个阶段之前，将多行合并为一个块。

以下参数被支持

新的块通过在 firstline 中传递的 RE2 正则表达式来识别。

任何不匹配表达式的行都视为上一个匹配的块的一部分。如果没有新的日志在 max_wait_time 内到达，则将块发送出去。如果超出 max_lines，则开始一个新的块。

让我们通过一个示例阶段和来自 Flask 网络服务的日志条目流来实际看看它是如何工作的。

复制

stage.multiline {
    firstline     = "^\\[\\d{4}-\\d{2}-\\d{2} \\d{1,2}:\\d{2}:\\d{2}\\]"
    max_wait_time = "10s"
}

[2023-01-18 17:41:21] "GET /hello HTTP/1.1" 200 -
[2023-01-18 17:41:25] ERROR in app: Exception on /error [GET]
Traceback (most recent call last):
  File "/home/pallets/.pyenv/versions/3.8.5/lib/python3.8/site-packages/flask/app.py", line 2447, in wsgi_app
    response = self.full_dispatch_request()
  File "/home/pallets/.pyenv/versions/3.8.5/lib/python3.8/site-packages/flask/app.py", line 1952, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/home/pallets/.pyenv/versions/3.8.5/lib/python3.8/site-packages/flask/app.py", line 1821, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/home/pallets/.pyenv/versions/3.8.5/lib/python3.8/site-packages/flask/_compat.py", line 39, in reraise
    raise value
  File "/home/pallets/.pyenv/versions/3.8.5/lib/python3.8/site-packages/flask/app.py", line 1950, in full_dispatch_request
    rv = self.dispatch_request()
  File "/home/pallets/.pyenv/versions/3.8.5/lib/python3.8/site-packages/flask/app.py", line 1936, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/home/pallets/src/deployment_tools/hello.py", line 10, in error
    raise Exception("Sorry, this route always breaks")
Exception: Sorry, this route always breaks
[2023-01-18 17:42:24] "GET /error HTTP/1.1" 500 -
[2023-01-18 17:42:29] "GET /hello HTTP/1.1" 200 -

所有形成单独网络请求日志条目的“块”都以方括号中的时间戳开始。该阶段使用 firstline 中的正则表达式来检测这一点，以将跟踪的所有行合并为一个块，从而形成一个单独的 Loki 日志条目。

阶段输出块

stage.output 内部块配置了一个读取提取映射并更改转发给下一个组件的日志条目内容的处理阶段。

以下参数被支持

让我们看看以下日志行和三阶段管道是如何工作的。

复制

{"user": "John Doe", "message": "hello, world!"}

stage.json {
    expressions = { "user" = "user", "message" = "message" }
}

stage.labels {
    values = { "user" = "user" }
}

stage.output {
    source = "message"
}

第一阶段将以下键值对提取到共享映射中

复制

user: John Doe
message: hello, world!

然后，第二阶段将 user="John Doe" 添加到日志条目的标签集中，最终输出阶段将日志行从原始 JSON 更改为 hello, world!。

阶段打包块

stage.pack 内部块配置了一个转换阶段，该阶段用包含提取值和标签的 JSON 对象替换日志条目。

以下参数被支持

此阶段允许您通过将它们打包到 JSON 对象中，将提取的值和标签与日志行一起嵌入。原始消息存储在 _entry 键下，所有其他键保持其值。这在您确实想要保留某些标签或元数据，但又不希望由于高基数而将其作为标签索引的情况下很有用。

Loki 的查询能力使得在查询时仍然可以轻松访问这些数据，以便在查询时进行过滤和聚合。

例如，考虑以下日志条目

复制

log_line: "something went wrong"
labels:   { "level" = "error", "env" = "dev", "user_id" = "f8fas0r" }

和此处理阶段

alloy 复制

stage.pack {
    labels = ["env", "user_id"]
}

该阶段将日志条目转换成以下 JSON 对象，其中删除了原始日志条目中的两个嵌入的标签

json Copy

{
  "_entry": "something went wrong",
  "env": "dev",
  "user_id": "f8fas0r"
}

在查询时，可以使用Loki的unpack解析器来访问这些嵌套标签，并自动将日志行替换为存储在_entry字段中的原始日志行。

当将多个日志流组合在一起使用pack阶段时，可以将ingest_timestamp设置为true以避免交错的时间戳和乱序摄取问题。

stage.regex块

stage.regex内部块配置了一个处理阶段，该阶段使用正则表达式解析日志行，并使用命名捕获组将数据添加到共享提取的值映射中。

以下参数被支持

expression字段需要是一个RE2正则表达式字符串。每个匹配的捕获组都添加到提取映射中，所以它必须命名为：(?Pre)。捕获组的名称随后用作提取映射中匹配值的键。

由于Alloy语法字符串的工作方式，expression中的任何反斜杠都必须用双反斜杠转义；例如"\\w"或"\\S+"。

如果source为空或不存在，则阶段本身解析日志行。如果设置了，阶段将解析具有相同名称的先前提取值。

给定以下日志行和正则表达式阶段，以下显示了提取的值

复制

2019-01-01T01:00:00.000000001Z stderr P i'm a log message!

stage.regex {
    expression = "^(?s)(?P<time>\\S+?) (?P<stream>stdout|stderr) (?P<flags>\\S+?) (?P<content>.*)$"
}

time: 2019-01-01T01:00:00.000000001Z,
stream: stderr,
flags: P,
content: i'm a log message

另一方面，如果设置了source值，则正则表达式应用于存储在该名称下共享映射中的值。

让我们看看当以下日志行通过这个两阶段管道时会发生什么

复制

{"timestamp":"2022-01-01T01:00:00.000000001Z"}

stage.json {
    expressions = { time = "timestamp" }
}
stage.regex {
    expression = "^(?P<year>\\d+)"
    source     = "time"
}

第一阶段将以下键值对添加到提取映射中

复制

time: 2022-01-01T01:00:00.000000001Z

然后，正则表达式阶段从共享值中解析时间值，并将后续键值对追加回提取值映射

复制

year: 2022

stage.replace块

stage.replace内部块配置了一个阶段，该阶段使用正则表达式解析日志行并替换日志行内容。正则表达式中的命名捕获组也支持将数据添加到共享提取的值映射中。

以下参数被支持

expression中的每个捕获组和命名捕获组都替换为replace中给出的值。

expression必须包含有效的RE2正则表达式捕获组。您还可以使用类似于(?Pre)的语法命名一些组。如果任何捕获组被命名，它们的值将设置在共享提取映射中，其名称为正则表达式组。

source字段定义了使用expression解析的数据源。当source缺失或为空时，阶段本身解析日志行，但它也可以用于解析先前提取的值。替换的值分配回source键。

由于Alloy语法处理双引号字符串中的反斜杠的方式，请注意，正则表达式中的所有反斜杠都必须像"\\w*"一样转义。

让我们看看以下日志行和阶段是如何工作的。由于省略了source，替换发生在日志行本身上。

复制

2023-01-01T01:00:00.000000001Z stderr P i'm a log message who has sensitive information with password xyz!

stage.replace {
    expression = "password (\\S+)"
    replace    = "*****"
}

日志行被转换成：

复制

2023-01-01T01:00:00.000000001Z stderr P i'm a log message who has sensitive information with password *****!

如果 replace 为空，则捕获的值将被省略。

以下示例中，source 已定义。

复制

{"time":"2023-01-01T01:00:00.000000001Z", "level": "info", "msg":"11.11.11.11 - \"POST /loki/api/push/ HTTP/1.1\" 200 932 \"-\" \"Mozilla/5.0\"}

stage.json {
    expressions = { "level" = "", "msg" = "" }
}

stage.replace {
    expression = "\\S+ - \"POST (\\S+) .*"
    source     = "msg"
    replace    = "redacted_url"
}

JSON 阶段将以下键值对添加到提取的映射中：

复制

time: 2023-01-01T01:00:00.000000001Z
level: info
msg: "11.11.11.11 - "POST /loki/api/push/ HTTP/1.1" 200 932 "-" "Mozilla/5.0"

replace 阶段对 msg 值进行操作。捕获组与 /loki/api/push 匹配，并被替换为 redacted_url。

msg 值最终被转换成：

复制

msg: "11.11.11.11 - "POST redacted_url HTTP/1.1" 200 932 "-" "Mozilla/5.0"

replace 字段可以使用一组模板函数，通过使用 Go 的 text/template 包。

让我们通过一个示例日志行和阶段来查看它是如何工作的。

复制

11.11.11.11 - alloy [01/Jan/2023:00:00:01 +0200]

stage.replace {
    expression = "^(?P<ip>\\S+) (?P<identd>\\S+) (?P<user>\\S+) \\[(?P<timestamp>[\\w:/]+\\s[+\\-]\\d{4})\\]"
    replace    = "{{ .Value | ToUpper }}"
}

由于 source 为空，正则表达式解析日志行本身并将命名捕获组提取到共享值映射中。 replace 字段作用于这些提取的值并将它们转换为大写

复制

ip: 11.11.11.11
identd: -
user: FRANK
timestamp: 01/JAN/2023:00:00:01 +0200

然后日志行变为：

复制

11.11.11.11 - FRANK [01/JAN/2023:00:00:01 +0200]

以下列表包含可用的函数，以及更复杂 replace 字段的示例。

复制

ToLower, ToUpper, Replace, Trim, TrimLeftTrimRight, TrimPrefix, TrimSuffix, TrimSpace, Hash, Sha2Hash, regexReplaceAll, regexReplaceAllLiteral

"{{ if eq .Value \"200\" }}{{ Replace .Value \"200\" \"HttpStatusOk\" -1 }}{{ else }}{{ .Value | ToUpper }}{{ end }}"
"*IP4*{{ .Value | Hash "salt" }}*"

stage.sampling 块

使用 sampling 阶段来采样日志。配置值 rate = 0.1 表示 10% 的日志将继续处理。其余 90% 的日志将被丢弃。

以下参数被支持

例如，下面的配置将采样 25% 的日志并丢弃剩余的 75%。当日志被丢弃时，带有 reason=logs_sampling 标签的 loki_process_dropped_lines_total 指标将增加。

alloy 复制

stage.sampling {
    rate = 0.25
    drop_counter_reason = "logs_sampling"
}

stage.static_labels 块

内嵌的 stage.static_labels 块配置了一个静态标签处理阶段，该阶段将一组静态标签添加到传入的日志条目中。

以下参数被支持

alloy 复制

stage.static_labels {
    values = {
      foo = "fooval",
      bar = "barval",
    }
}

stage.template 块

内嵌的 stage.template 块配置了一个转换阶段，该阶段允许用户通过使用 Go 的 text/template 包 语法来操作提取映射中的值。此阶段主要用于在后续阶段设置标签之前操作和标准化之前阶段的 数据。示例用例包括用下划线替换空格、将大写字符串转换为小写或对值进行哈希处理。

模板阶段还可以在提取映射中创建新的键。

以下参数被支持

模板字符串可以是 Go 的 text/template 可以使用的任何有效模板。它支持来自 sprig 包 的所有函数，以及以下列表中的自定义函数：

复制

ToLower, ToUpper, Replace, Trim, TrimLeftTrimRight, TrimPrefix, TrimSuffix, TrimSpace, Hash, Sha2Hash, regexReplaceAll, regexReplaceAllLiteral

有关每个函数的更多详细信息，请参阅以下 支持的功能 部分。

假设提取映射中没有数据，以下阶段仅将 new_key: "hello_world" 键值对添加到共享映射中。

alloy 复制

stage.template {
    source   = "new_key"
    template = "hello_world"
}

如果提取字段中存在 source 值，则可以在模板中将该值引用为 .Value。下一阶段获取提取映射中 app 的当前值，将其转换为小写，并添加一个后缀到其值

alloy 复制

stage.template {
    source   = "app"
    template = "{{ ToLower .Value }}_some_suffix"
}

任何之前提取的键都可用于 template 进行展开和使用。下一阶段获取 level、app 和 module 的当前值并创建一个名为 output_message 的新键

alloy 复制

stage.template {
    source   = "output_msg"
    template = "{{ .level }} for app {{ ToUpper .app }} in module {{.module}}"
}

可以使用特殊键 Entry 来引用当前行；这在你需要将某些内容追加/附加到日志行时非常有用，例如以下片段：

alloy 复制

stage.template {
    source   = "message"
    template = "{{.app }}: {{ .Entry }}"
}
stage.output {
    source = "message"
}

支持的功能

除了支持来自 sprig 包 的所有函数外，template 阶段还支持以下自定义函数。

ToLower 和 ToUpper

ToLower 和 ToUpper 分别将整个字符串转换为小写和大写。

示例

alloy 复制

stage.template {
    source   = "out"
    template = "{{ ToLower .app }}"
}
stage.template {
    source   = "out"
    template = "{{ .app | ToUpper }}"
}

替换

Replace 函数的语法定义为 {{ Replace <string> <old> <new> <n> }}。

该函数返回输入字符串的副本，将 <old> 参数的实例替换为 <new>。该函数最多替换第二个参数的 <n> 个非重叠实例。如果 <n> 小于零，则没有替换实例的数量限制。最后，如果 <old> 为空，则匹配字符串中的每个 UTF-8 字符之前和之后。

以下示例将 loki 单词的前两个实例替换为 Loki。

alloy 复制

stage.template {
    source   = "output"
    template = `{{ Replace .Value "loki" "Loki" 2 }}`
}

Trim, TrimLeft, TrimRight, TrimSpace, TrimPrefix, TrimSuffix

• Trim 返回一个字符串 s 的切片，其中删除了所有在 cutset 中包含的 Unicode 代码点。
• TrimLeft 和 TrimRight 与 Trim 相同，不同之处在于它们分别仅删除前导和尾随字符。
• TrimSpace 返回一个字符串 s 的切片，其中删除了所有由 Unicode 定义的 leading 和 trailing 空白字符。
• TrimPrefix 和 TrimSuffix 分别修剪提供的前缀或后缀。

示例

alloy 复制

stage.template {
    source   = "output"
    template = `{{ Trim .Value ",. " }}`
}
stage.template {
    source   = "output"
    template = "{{ TrimSpace .Value }}"
}
stage.template {
    source   = "output"
    template = `{{ TrimPrefix .Value "--" }}`
}

正则表达式

regexReplaceAll 返回输入字符串的副本，用替换字符串替换正则表达式的匹配项。在替换字符串内部，$ 字符的解释与 Expand 函数中的解释相同，例如，$1 代表第一个捕获的子匹配。

regexReplaceAllLiteral 返回输入字符串的副本，用替换字符串替换正则表达式的匹配项。替换字符串直接替换，不使用 Expand。

alloy 复制

stage.template {
    source   = "output"
    template = `{{ regexReplaceAll "(a*)bc" .Value "${1}a" }}`
}
stage.template {
    source   = "output"
    template = `{{ regexReplaceAllLiteral "(ts=)" .Value "timestamp=" }}`
}

哈希和 Sha2Hash

Hash 返回字符串的 Sha3_256 哈希，表示为 64 位十六进制数字。您可以使用它来混淆日志中的敏感数据和 PII。它需要一个（固定的）盐值，以增加低输入域的复杂性（例如，所有可能的社会保险号）。Sha2Hash 返回字符串的 Sha2_256，比 Hash 更快、更节省 CPU，但安全性较低。

示例

alloy 复制

stage.template {
    source   = "output"
    template = `{{ Hash .Value "salt" }}`
}
stage.template {
    source   = "output"
    template = `{{ Sha2Hash .Value "salt" }}`
}

我们建议使用 Hash，因为它具有更强的哈希算法。

stage.tenant 块

stage.tenant 内部块通过从提取的数据映射、标签或提供的值中获取它来设置日志条目的租户 ID。

以下参数被支持

该块期望只提供 label、source 或 value 中的一个。

以下阶段将固定值 team-a 作为租户 ID

alloy 复制

stage.tenant {
    value = "team-a"
}

此阶段在将日志条目解析为共享提取映射中的 JSON 后从 customer_id 字段中提取租户 ID

alloy 复制

stage.json {
    expressions = { "customer_id" = "" }
}
stage.tenant {
    source = "customer_id"
}

以下示例从先前阶段设置的标签中提取租户 ID

alloy 复制

stage.labels {
    "namespace" = "k8s_namespace"
}
stage.tenant {
    label = "namespace"
}

stage.timestamp 块

stage.timestamp 内部块配置一个处理阶段，在将日志条目转发到下一个组件之前设置日志条目的时间戳。如果没有设置时间戳阶段，则日志条目的时间戳默认为日志条目被抓取的时间。

以下参数被支持

source字段定义了阶段应该尝试解析为时间戳的共享提取值映射中的哪个值。

format字段定义了如何解析该源。

format可以设置为以下常用形式的缩写值之一

复制

ANSIC: Mon Jan _2 15:04:05 2006
UnixDate: Mon Jan _2 15:04:05 MST 2006
RubyDate: Mon Jan 02 15:04:05 -0700 2006
RFC822: 02 Jan 06 15:04 MST
RFC822Z: 02 Jan 06 15:04 -0700
RFC850: Monday, 02-Jan-06 15:04:05 MST
RFC1123: Mon, 02 Jan 2006 15:04:05 MST
RFC1123Z: Mon, 02 Jan 2006 15:04:05 -0700
RFC3339: 2006-01-02T15:04:05-07:00
RFC3339Nano: 2006-01-02T15:04:05.999999999-07:00

此外，还支持以下格式值以支持常用的Unix时间戳

复制

Unix: 1562708916 or with fractions 1562708916.000000123
UnixMs: 1562708916414
UnixUs: 1562708916414123
UnixNs: 1562708916000000123

否则，该字段接受一个自定义格式字符串，该字符串定义了阶段应该如何解释历史中的任意参考点。任意的参考点是Mon Jan 2 15:04:05 -0700 MST 2006。

该字段的字符串值直接传递到Go的time.Parse函数中的布局参数。time.Parse

如果自定义格式没有年份组件，阶段使用系统时钟的当前年份。

以下表格显示了在定义自定义格式时使用的支持参考值。

fallback_formats字段定义了一个或多个要尝试解析时间戳的格式字段，如果使用format解析失败。

location字段必须是有效的IANA时区数据库位置，并确定时间戳值解释为哪个时区。

action_on_failure字段定义了当源字段不存在于共享提取映射中或时间戳解析失败时应该发生什么。

支持的操作包括

• fudge (默认): 将时间戳更改为最后已知的时间戳，加上1纳秒（以保证日志条目排序）。
• skip: 不更改时间戳并保持日志条目抓取时的时间。

以下阶段从共享值映射中获取time值，将其解析为RFC3339格式，并将其设置为日志条目的时间戳。

alloy 复制

stage.timestamp {
    source = "time"
    format = "RFC3339"
}

以下示例会解析类似于2024-12-20T09:14:58,381+02:00的时间戳。

alloy 复制

stage.timestamp {
    source = "time"
    format = "2006-01-02T15:04:05,000-07:00"
}

stage.geoip块

stage.geoip内部块配置了一个处理阶段，该阶段读取IP地址并将geoip字段填充到共享映射中。使用Maxmind的GeoIP2数据库进行查找。

以下参数被支持

使用城市数据库的GeoIP示例

复制

{"log":"log message","client_ip":"34.120.177.193"}

loki.process "example" {
    stage.json {
        expressions = {ip = "client_ip"}
    }

    stage.geoip {
        source  = "ip"
        db      = "/path/to/db/GeoLite2-City.mmdb"
        db_type = "city"
    }

    stage.labels {
        values = {
            geoip_city_name          = "",
            geoip_country_name       = "",
            geoip_country_code       = "",
            geoip_continent_name     = "",
            geoip_continent_code     = "",
            geoip_location_latitude  = "",
            geoip_location_longitude = "",
            geoip_postal_code        = "",
            geoip_timezone           = "",
            geoip_subdivision_name   = "",
            geoip_subdivision_code   = "",
        }
    }
}

json阶段从日志行的client_ip键中提取IP地址。然后，提取的ip值作为源提供给geoip阶段。geoip阶段在IP上进行查找，并在共享映射中填充以下字段，这些字段通过labels阶段添加为标签。

本例中使用的IP地址提取的数据

• geoip_city_name：堪萨斯城
• geoip_country_name：美国
• geoip_country_code：US
• geoip_continent_name：北美洲
• geoip_continent_code：NA
• geoip_location_latitude：39.1027
• geoip_location_longitude：-94.5778
• geoip_postal_code：64184
• geoip_timezone：America/Chicago
• geoip_subdivision_name：密苏里州
• geoip_subdivision_code：MO

使用ASN（自治系统编号）数据库的GeoIP示例

复制

loki.process "example" {
    stage.json {
        expressions = {ip = "client_ip"}
    }

    stage.geoip {
        source  = "ip"
        db      = "/path/to/db/GeoIP2-ASN.mmdb"
        db_type = "asn"
    }

    stage.labels {
        values = {
            geoip_autonomous_system_number       = "",
            geoip_autonomous_system_organization = "",
        }
    }
}

json阶段从日志行的client_ip键中提取IP地址。然后，提取的ip值作为源提供给geoip阶段。geoip阶段在IP上进行查找，并在共享映射中填充数据。

本例中使用的IP地址提取的数据

• geoip_autonomous_system_number：396982
• geoip_autonomous_system_organization：GOOGLE-CLOUD-PLATFORM

使用国家数据库的GeoIP示例

复制

{"log":"log message","client_ip":"34.120.177.193"}

loki.process "example" {
    stage.json {
        expressions = {ip = "client_ip"}
    }

    stage.geoip {
        source  = "ip"
        db      = "/path/to/db/GeoLite2-Country.mmdb"
        db_type = "country"
    }

    stage.labels {
        values = {
            geoip_country_name       = "",
            geoip_country_code       = "",
            geoip_continent_name     = "",
            geoip_continent_code     = "",
        }
    }
}

json阶段从日志行的client_ip键中提取IP地址。然后，提取的ip值作为源提供给geoip阶段。geoip阶段在IP上进行查找，并在共享映射中填充以下字段，这些字段通过labels阶段添加为标签。

本例中使用的IP地址提取的数据

• geoip_country_name：美国
• geoip_country_code：US
• geoip_continent_name：北美洲
• geoip_continent_code：NA

使用自定义字段进行GeoIP示例

如果使用的MMDB文件包含自定义数据，例如，如Maxmind博客文章中所述的私有IP地址，则可以使用custom_lookups属性从记录中提取。

复制

loki.process "example" {
    stage.json {
        expressions = {ip = "client_ip"}
    }

    stage.geoip {
        source         = "ip"
        db             = "/path/to/db/GeoIP2-Enriched.mmdb"
        db_type        = "city"
        custom_lookups = {
            "department"  = "MyCompany.DeptName",
            "parent_vnet" = "MyCompany.ParentVNet",
            "subnet"      = "MyCompany.Subnet",
        }
    }

    stage.labels {
        values = {
            department  = "",
            parent_vnet = "",
            subnet      = "",
        }
    }
}

json阶段从日志行的client_ip键中提取IP地址。然后，提取的ip值作为源提供给geoip阶段。geoip阶段在IP上进行查找，并在共享映射中填充来自城市数据库的结果数据以及自定义查找数据。最后，将共享映射中的自定义查找字段添加为标签。

导出字段

以下字段被导出，可以被其他组件引用

组件健康状态

loki.process只有在给定无效配置时才报告为不健康。

调试信息

loki.process不公开任何特定组件的调试信息。

调试指标

• loki_process_dropped_lines_total（计数器）：作为处理阶段的一部分丢弃的行数。
• loki_process_dropped_lines_by_label_total（计数器）：当在stage.limit中的by_label_name非空时丢弃的行数。

示例

此示例创建了一个loki.process组件，该组件从JSON日志行中提取environment值并将其设置为名为’env’的标签。

alloy 复制

loki.process "local" {
  forward_to = [loki.write.onprem.receiver]

  stage.json {
      expressions = { "extracted_env" = "environment" }
  }

  stage.labels {
      values = { "env" = "extracted_env" }
  }
}

兼容组件

loki.process可以接受以下组件的参数

• 导出Loki LogsReceiver的组件

loki.process的导出可以被以下组件消费

• 消费Loki LogsReceiver的组件