loki.process

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

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

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

用法

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。

stage.cri {}

CRI将日志行指定为用单个空格分隔的值，以下为其组件

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

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

复制

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

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

stage.decolorize块

stage.decolorize从日志行中去除ANSI颜色代码，从而使解析日志变得更简单。

stage.decolorize块不支持任何参数或内嵌块，因此总是为空。

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块

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

stage.docker块不支持任何参数或内嵌块，因此总是为空。

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块

内嵌的stage.drop块配置了一个过滤器阶段，根据几个选项删除日志条目。如果提供了多个选项，将作为AND子句处理，并且日志条目 必须满足所有条件才会被删除。如果要使用OR子句删除条目，请按顺序指定多个drop块。

支持以下参数

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

• 如果source为空或未提供，则正则表达式尝试匹配整个日志行。
• 如果source是单个名字，则正则表达式尝试匹配提取映射中相应的值。
• 如果 source 是由逗号分隔的名称列表，则使用 separator 连接提取出来的映射中的对应的值，并使用正则表达式尝试匹配拼接后的字符串。

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

• 如果 source 是单个名称，当提取出来的映射中相应的值与 value 完全匹配时，条目将被丢弃。
• 如果 source 是由逗号分隔的名称列表，当 value 与使用 separator 拼接的提取数据中的 source 值匹配时，条目将被丢弃。

whenever an entry is dropped, the metric loki_process_dropped_lines_total is incremented. By default, the reason label is "drop_stage", but you can provide a custom label using the drop_counter_reason argument.

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

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

在以下示例中，我们定义了多个 drop 块，以便 loki.process 丢弃 24h 或更早的，长度超过 8KB，或特定值为 'foo' 的条目。

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 事件日志中的 Message 字符串中提取数据。

支持以下参数

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

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

结合 stage.json 的示例

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：Message type:\r\nOverwritten: new\r\nImage: C:\Users\User\alloy.exe
• Overwritten：old

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

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

stage.json 块

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

支持以下参数

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

expressions 字段是一组要运行的键值对 JMESPath 表达式。映射键定义了用哪个名称从数据提取数据，而映射值是用于填充值的表达式。

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

{"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 嵌套块配置了一个可以丢弃传入日志条目中标签的处理阶段。

支持以下参数

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

stage.label_keep 块

stage.label_keep 嵌套块配置了一个可以过滤传入日志条目的标签集到子集的处理阶段。

支持以下参数

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

stage.labels 块

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

支持以下参数

在一个标签阶段，映射的键定义要设置的标签，值是查找的方式。如果值为空，则假定它与键相同。

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阶段，映射的键定义要设置的标签，值是如何查找的。如果值为空，则假定它与键相同。

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时，超过速率限制的传入条目将被丢弃，否则它们将在更多令牌可用之前排队。

stage.limit {
    rate  = 5
    burst = 10
}

如果by_label_name被设置，那么drop必须被设置为true。这允许阶段根据标签的数量而不是线条的数量进行速率限制。

以下示例独立地对每个唯一的namespace值进行速率限制。任何没有namespace标签的条目都不会进行速率限制。该阶段会跟踪最多max_distinct_labels个唯一值，默认为10000。

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"
}

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

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

阶段.luhn 块

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

卢恩算法 是一个简单的校验和公式，用于验证各种识别号码，例如信用卡号码、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"

阶段.match 块

stage.match 内部块配置一个过滤阶段，当日志条目匹配可配置的 LogQL 流选择器和过滤器表达式时，有条件地应用嵌套的一系列处理阶段或删除条目。

支持以下参数

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

如果指定的操作是 "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。

阶段.metrics 块

stage.metrics 内部块配置了阶段，允许根据共享提取的地图中的值定义和更新指标。创建的指标在Alloy的根目录 `/metrics` 端点可用。

stage.metrics 块不支持任何参数，只能通过一系列嵌套的内部 metric.* 块进行配置，每个块对应一个应生成的指标。

以下是在 stage.metrics 定义内部支持的块：

metric.counter 块

定义一个值仅递增的指标。

支持以下参数

计数器不能同时设置 match_all 为 true 和 value。计数器在没有设置 match_all=true、action=add 或 count_entry_bytes 时不能设置 value。有效的 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"。

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

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

以下管道通过使用match_all参数创建了一个计数器，每当接收到任何日志行时，计数器都会递增。该管道还使用了count_entry_bytes参数创建了一个第二个计数器，该计数器会累加这些日志行的字节数。

如果24小时内没有接收到新条目，这两个度量将消失，以避免构建不再有任何用途的度量。这两个度量是跟踪日志流数量和字节大小的良好起点，以确定高流量或高基数数据的来源。

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。

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>的文本。第二阶段创建了一个仪表，其当前度量值将增加从重试字段中提取的数字。

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

以下示例显示了读取提取映射中的response_time并将其放入桶中的直方图，同时增加该桶的计数和特定桶的求和

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]
    }
}

阶段多行块

阶段的多行内部块将多行合并为单个块，然后将其传递到管道中的下一阶段。

支持以下参数

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

任何不匹配该表达式的行都被认为是上一个匹配的块的组成部分。如果没有新的日志在max_wait_time内到达，则发送该块。如果超过了max_lines字段的定义，则开始一个新的块。

让我们用一个示例阶段和一个从Flask Web服务流出的日志条目流来实际看看这是如何工作的。

复制

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 -

所有形成单独Web请求日志条目的'块'都以方括号中的时间戳开始。该阶段使用firstline中的正则表达式检测此时间戳，将所有跟踪行折叠成单个块，从而形成一个Loki日志条目。

阶段输出块

阶段输出内部块配置了一个处理阶段，该阶段从提取的映射中读取并更改转发到下一组件的日志条目的内容。

支持以下参数

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

复制

{"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!。

阶段包装块

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

支持以下参数

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

Loki的查询功能使您仍然可以轻松访问此数据，以便在查询时间进行筛选和聚合。

例如，考虑以下日志条目：

复制

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

以及这个处理阶段：

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

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

json 复制

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

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

在结合多个日志流与使用pack阶段时，可以将ingest_timestamp设置为true以避免交错时间戳和不按顺序摄取问题。

阶段.regex块

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

支持以下参数

需要解析的数据提取名字。如果为空，则使用日志消息。

因为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

阶段.replace块

《阶段.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" }}*"

阶段采样块

使用sampling阶段对日志进行采样。配置值rate = 0.1意味着10%的日志将继续处理。其余90%的日志将被丢弃。

支持以下参数

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

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

阶段静态标签块

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

支持以下参数

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

阶段模板块

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"添加到共享映射中。

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

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

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

任何先前提取的键都可用于在template中展开和使用。下一个阶段会将当前值level、app和module用于创建一个名为output_message的新键

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

可以使用名为Entry的特殊键来引用当前行；当需要附加/预处理某些内容到日志行时，这非常有用，如下面的代码片段所示

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

支持函数

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

ToLower 和 ToUpper

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

示例

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

Replace

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

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

以下示例将第一个两个 loki 单词替换为 Loki

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

Trim, TrimLeft, TrimRight, TrimSpace, TrimPrefix, TrimSuffix

• Trim 返回一个字符串切片，其中去除了 cutset 中包含的所有前导和尾部 Unicode 代码点。
• TrimLeft 和 TrimRight 与 Trim 相同，除了它们分别只去除前导和尾部字符。
• TrimSpace 返回一个字符串切片，其中去除了 Unicode 定义的所有前导和尾部空白。
• TrimPrefix 和 TrimSuffix 分别去除提供的前缀或后缀。

示例

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

Regex

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

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

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

Hash 和 Sha2Hash

Hash 返回字符串的 Sha3_256 哈希，表示为 64 位十六进制数。您可以使用它来混淆日志中的敏感数据和个人信息 (PII)。它需要一个（固定的）盐值，以便在输入域中添加复杂度（例如，所有可能的社会安全号码）的过程中。 Sha2Hash 返回字符串的 Sha2_256 哈希，比 Hash 快且更节能，但不如 Hash 安全。

示例

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 分配

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

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

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

以下示例从上一个阶段设置的标签中提取租户 ID

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 函数中的布局参数。

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

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

fallback_formats 字段定义了一个或多个格式字段，用于在 format 解析失败时尝试解析时间戳。

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

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

支持的操作有

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

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

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

以下示例将解析如 2024-12-20T09:14:58,381+02:00 的时间戳

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

stage.geoip block

下面的 stage.geoip 嵌套块配置了一个处理阶段，它会读取一个 IP 地址并将地理信息字段填充到共享映射中。查找使用 Maxmind 的 GeoIP2 数据库进行。

支持以下参数

带有 City 数据库的 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’的标签。

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》的组件