Loki作为一个新兴的日志解决方案，现在越来越受到关注。经过调研比较，京东智联云正在将云翼的日志服务底层逐步从ES替换为Loki 。本文基于我们对Loki的使用和理解，从它产生的背景、解决的问题、采用的方案、系统架构、实现逻辑等做一些剖析，希望对关注Loki的小伙伴们提供一些帮助。

背景

在日常的系统可视化监控过程中，当监控探知到指标异常时，我们往往需要对问题的根因做出定位。但监控数据所暴露的信息是提前预设、高度提炼的，在信息量上存在着很大的不足，它需要结合能够承载丰富信息的日志系统一起使用。

当监控系统探知到异常告警，我们通常在Dashboard上根据异常指标所属的集群、主机、实例、应用、时间等信息圈定问题的大致方向，然后跳转到日志系统做更精细的查询，获取更丰富的信息来最终判断问题根因。

在如上流程中，监控系统和日志系统往往是独立的，使用方式具有很大差异。比如监控系统Prometheus比较受欢迎，日志系统多采用ES+Kibana 。他们具有完全不同的概念、不同的搜索语法和界面，这不仅给使用者增加了学习成本，也使得在使用时需在两套系统中频繁做上下文切换，对问题的定位迟滞。

此外，日志系统多采用全文索引来支撑搜索服务，它需要为日志的原文建立反向索引，这会导致最终存储数据相较原始内容成倍增长，产生不可小觑的存储成本。并且，不管数据将来是否会被搜索，都会在写入时因为索引操作而占用大量的计算资源，这对于日志这种写多读少的服务无疑也是一种计算资源的浪费。

Loki则是为了应对上述问题而产生的解决方案，它的目标是打造能够与监控深度集成、成本极度低廉的日志系统。

Loki日志方案

低使用成本

数据模型

在数据模型上Loki参考了Prometheus 。数据由标签、时间戳、内容组成，所有标签相同的数据属于同一日志流，具有如下结构：

在数据模型上Loki参考了Prometheus 。数据由标签、时间戳、内容组成，所有标签相同的数据属于同一日志流，具有如下结构：

{
  "stream": { 
    "label1": "value1",
    "label1": "value2"
  }, # 标签
  "values": [
    ["<timestamp nanoseconds>","log content"], # 时间戳，内容
    ["<timestamp nanoseconds>","log content"]
  ]
}

<左右滑动以查看完整代码>

• 标签，描述日志所属集群、服务、主机、应用、类型等元信息， 用于后期搜索服务；

• 时间戳，日志的产生时间；

• 内容，日志的原始内容。

Loki还支持多租户，同一租户下具有完全相同标签的日志所组成的集合称为一个日志流。

在日志的采集端使用和监控时序数据一致的标签，这样在可以后续与监控系统结合时使用相同的标签，也为在UI界面中与监控结合使用做快速上下文切换提供数据基础。

LogQL

Loki使用类似Prometheus的PromQL的查询语句logQL ，语法简单并贴近社区使用习惯，降低用户学习和使用成本。语法例子如下：

{file="debug.log""} |= "err"

<左右滑动以查看完整代码>

• 流选择器：{label1="value1", label2="value2"}， 通过标签选择日志流， 支持等、不等、匹配、不匹配等选择方式；

• 过滤器：|= "err"，过滤日志内容，支持包含、不包含、匹配、不匹配等过滤方式。

这种工作方式类似于find+grep，find找出文件，grep从文件中逐行匹配：

find . -name "debug.log" | grep err

<左右滑动以查看完整代码>

logQL除支持日志内容查询外，还支持对日志总量、频率等聚合计算。

Grafana

在Grafana中原生支持Loki插件，将监控和日志查询集成在一起，在同一UI界面中可以对监控数据和日志进行side-by-side的下钻查询探索，比使用不同系统反复进行切换更直观、更便捷。

此外，在Dashboard中可以将监控和日志查询配置在一起，这样可同时查看监控数据走势和日志内容，为捕捉可能存在的问题提供更直观的途径。

低存储成本

只索引与日志相关的元数据标签，而日志内容则以压缩方式存储于对象存储中, 不做任何索引。相较于ES这种全文索引的系统，数据可在十倍量级上降低，加上使用对象存储，最终存储成本可降低数十倍甚至更低。方案不解决复杂的存储系统问题，而是直接应用现有成熟的分布式存储系统，比如S3、GCS、Cassandra、BigTable 。

架构

整体上Loki采用了读写分离的架构，由多个模块组成。其主体结构如下图所示：

* Promtail、Fluent-bit、Fluentd、Rsyslog等开源客户端负责采集并上报日志；

* Distributor：日志写入入口，将数据转发到Ingester；

* Ingester：日志的写入服务，缓存并写入日志内容和索引到底层存储；

* Querier：日志读取服务，执行搜索请求；

* QueryFrontend：日志读取入口，分发读取请求到Querier并返回结果；

* Cassandra/BigTable/DnyamoDB/S3/GCS：索引、日志内容底层存储；

* Cache：缓存，支持Redis/Memcache/本地Cache。

Distributor

作为日志写入的入口服务，其负责对上报数据进行解析、校验与转发。它将接收到的上报数解析完成后会进行大小、条目、频率、标签、租户等参数校验，然后将合法数据转发到Ingester 服务，其在转发之前最重要的任务是确保同一日志流的数据必须转发到相同Ingester上，以确保数据的顺序性。

Hash环

Distributor采用一致性哈希与副本因子相结合的办法来决定数据转发到哪些Ingester上。

Ingester在启动后，会生成一系列的32位随机数作为自己的Token ，然后与这一组Token一起将自己注册到Hash环中。在选择数据转发目的地时， Distributor根据日志的标签和租户ID生成Hash，然后在Hash环中按Token的升序查找第一个大于这个Hash的Token ，这个Token所对应的Ingester即为这条日志需要转发的目的地。如果设置了副本因子，顺序的在之后的token中查找不同的Ingester做为副本的目的地。

Hash环可存储于etcd、consul中。另外Loki使用Memberlist实现了集群内部的KV存储，如不想依赖etcd或consul ，可采用此方案。

输入输出

Distributor的输入主要是以HTTP协议批量的方式接受上报日志，日志封装格式支持JSON和PB ，数据封装结构：

[
  {
   "stream": { 
     "label1": "value1",
     "label1": "value2"
   },
   "values": [
     ["<timestamp nanoseconds>","log content"],
     ["<timestamp nanoseconds>","log content"]
   ]
  },
  ......
]

<左右滑动以查看完整代码>

Distributor以grpc方式向ingester发送数据，数据封装结构：

{
  "streams": [
    {
      "labels": "{label1=value1, label2=value2}",
      "entries": [
          {"ts": <unix epoch in nanoseconds>, "line:":"<log line>" },
          {"ts": <unix epoch in nanoseconds>, "line:":"<log line>" },
      ]
    }
    ....
   ]
}

<左右滑动以查看完整代码>

Ingester

作为Loki的写入模块，Ingester主要任务是缓存并写入数据到底层存储。根据写入数据在模块中的生命周期，ingester大体上分为校验、缓存、存储适配三层结构。

校验

Loki有个重要的特性是它不整理数据乱序，要求同一日志流的数据必须严格遵守时间戳单调递增顺序写入。所以除对数据的长度、频率等做校验外，至关重要的是日志顺序检查。 Ingester对每个日志流里每一条日志都会和上一条进行时间戳和内容的对比，策略如下：

• 与上一条日志相比，本条日志时间戳更新，接收本条日志；

• 与上一条日志相比，时间戳相同内容不同，接收本条日志；

• 与上一条日志相比，时间戳和内容都相同，忽略本条日志；

• 与上一条日志相比，本条日志时间戳更老，返回乱序错误。

缓存

日志在内存中的缓存采用多层树形结构对不同租户、日志流做出隔离。同一日志流采用顺序追加方式写入分块，整体结构如下：

* Instances：以租户的userID为键Instance为值的Map结构；

* Instance：一个租户下所有日志流 (stream) 的容器；

* Streams：以日志流的指纹 (streamFP) 为键，Stream为值的Map结构；

* Stream：一个日志流所有Chunk的容器；

* Chunks：Chunk的列表；

* Chunk：持久存储读写最小单元在内存态的结构；

* Block：Chunk的分块，为已压缩归档的数据；

* HeadBlock：尚在开放写入的分块；

* Entry： 单条日志单元，包含时间戳 (timestamp) 和日志内容 (line) 。

Chunks

在向内存写入数据前，ingester首先会根据租户ID(userID)和由标签计算的指纹(streamPF) 定位到日志流(stream)及Chunks。

Chunks由按时间升序排列的chunk组成，最后一个chunk接收最新写入的数据，其他则等刷写到底层存储。当最后一个chunk的存活时间或数据大小超过指定阈值时，Chunks尾部追加新的chunk 。

Chunk 

Chunk为Loki在底层存储上读写的最小单元在内存态下的结构。其由若干block组成，其中headBlock为正在开放写入的block ，而其他Block则已经归档压缩的数据。

Block

Block为数据的压缩单元，目的是为了在读取操作那里避免因为每次解压整个Chunk 而浪费计算资源，因为很多情况下是读取一个chunk的部分数据就满足所需数据量而返回结果了。

Block存储的是日志的压缩数据，其结构为按时间顺序的日志时间戳和原始内容，压缩可采用gzip、snappy 、lz4等方式。

HeadBlock

正在接收写入的特殊block ，它在满足一定大小后会被压缩归档为Block ，然后新headBlock会被创建。

存储适配

由于底层存储要支持S3、Cassandra、BigTable、DnyamoDB等系统，适配层将各种系统的读写操作抽象成统一接口，负责与他们进行数据交互。

输出

Chunk

Loki以Chunk为单位在存储系统中读写数据。在持久存储态下的Chunk具有如下结构：

* meta：封装chunk所属stream的指纹、租户ID，开始截止时间等元信息；

* data：封装日志内容，其中一些重要字段；

* encode保存数据的压缩方式；

* block-N bytes保存一个block的日志数据；

* #blocks section byte offset单元记录#block单元的偏移量；

* #block单元记录一共有多少个block；

* #entries和block-N bytes一一对应，记录每个block里有日式行数、时间起始点，blokc-N bytes的开始位置和长度等元信息。

Chunk数据的解析顺序：

1. 根据尾部的#blocks section byte offset单元得到#block单元的位置；

2. 根据#block单元记录得出chunk里block数量；

3. 从#block单元所在位置开始读取所有block的entries、mint、maxt、offset、len等元信息；

4. 顺序的根据每个block元信息解析出block的数据

索引

Loki只索引了标签数据，用于实现标签→日志流→Chunk的索引映射， 以分表形式在存储层存储。

1.  表结构

CREATE TABLE IF NOT EXISTS Table_N (
    hash text,
    range blob,
    value blob,
    PRIMARY KEY (hash, range)
 )

<左右滑动以查看完整代码>

• Table_N，根据时间周期分表名；

• hash， 不同查询类型时使用的索引；

• range，范围查询字段；

• value，日志标签的值

2.  数据类型

Loki保存了不同类型的索引数据用以实现不同映射场景，对于每种类型的映射数据，Hash/Range/Value三个字段的数据组成如下图所示：

seriesID为日志流ID， shard为分片，userID为租户ID，labelName为标签名，labelValueHash为标签值hash，chunkID为chunk的ID，chunkThrough为chunk里最后一条数据的时间这些数据元素在映射过程中的作用在Querier环节的[查询流程]((null))做详细介绍。

上图中三种颜色标识的索引类型从上到下分别为：

• 数据类型1：用于根据用户ID搜索查询所有日志流的ID；

• 数据类型2：用于根据用户ID和标签查询日志流的ID；

• 数据类型3：用于根据日志流ID查询底层存储Chunk的ID；

除了采用分表外，Loki还采用分桶、分片的方式优化索引查询速度。

• 分桶

以天分割：

bucketID = timestamp / secondsInDay。

以小时分割：

bucketID = timestamp / secondsInHour。

• 分片

将不同日志流的索引分散到不同分片，shard = seriesID%分片数。

Chunk状态

Chunk作为在Ingester中重要的数据单元，其在内存中的生命周期内分如下四种状态：

• Writing：正在写入新数据；

• Waiting flush：停止写入新数据，等待写入到存储；

• Retain：已经写入存储，等待销毁；

• Destroy：已经销毁。

四种状态之间的转换以writing -> waiting flush -> retain -> destroy顺序进行。

1.  状态转换时机

• 协作触发：有新的数据写入请求；

• 定时触发：刷写周期触发将chunk写入存储，回收周期触发将chunk销毁。

2.  writing转为waiting flush

chunk初始状态为writing，标识正在接受数据的写入，满足如下条件则进入到等待刷写状态：

• chunk空间满（协作触发）；

• chunk的存活时间（首末两条数据时间差）超过阈值 （定时触发）；

• chunk的空闲时间（连续未写入数据时长)超过设置 （定时触发）。

3. waiting flush转为etain

Ingester会定时的将等待刷写的chunk写到底层存储，之后这些chunk会处于”retain“状态，这是因为ingester提供了对最新数据的搜索服务，需要在内存里保留一段时间，retain状态则解耦了数据的刷写时间以及在内存中的保留时间，方便视不同选项优化内存配置。

4.  destroy，被回收等待GC销毁

总体上，Loki由于针对日志的使用场景，采用了顺序追加方式写入，只索引元信息，极大程度上简化了它的数据结构和处理逻辑，这也为Ingester能够应对高速写入提供了基础。

Querier

查询服务的执行组件，其负责从底层存储拉取数据并按照LogQL语言所描述的筛选条件过滤。它可以直接通过API提供查询服务，也可以与queryFrontend结合使用实现分布式并发查询。

查询类型

• 范围日志查询

• 单日志查询

• 统计查询

• 元信息查询

在这些查询类型中，范围日志查询应用最为广泛，所以下文只对范围日志查询做详细介绍。

并发查询 

对于单个查询请求，虽然可以直接调用Querier的API进行查询，但很容易会由于大查询导致OOM，为应对此种问题querier与queryFrontend结合一起实现查询分解与多querier并发执行。

每个querier都与所有queryFrontend建立grpc双向流式连接，实时从queryFrontend中获取已经分割的子查询求，执行后将结果发送回queryFrontend。具体如何分割查询及在querier间调度子查询将在queryFrontend环节介绍。

查询流程

1.  解析logQL指令

2.  查询日志流ID列表

Loki根据不同的标签选择器语法使用了不同的索引查询逻辑，大体分为两种：

• =，或多值的正则匹配=~ ， 工作过程如下：

1. 以类似下SQL所描述的语义查询出标签选择器里引用的每个标签键值对所对应的日志流ID(seriesID)的集合。

SELECT * FROM Table_N WHERE hash=？ AND range>=？    AND value=labelValue

<左右滑动以查看完整代码>

◆ hash为租户ID(userID)、分桶(bucketID)、标签名(labelName)组合计算的哈希值；
◆ range为标签值(labelValue)计算的哈希值。

2. 将根据标签键值对所查询的多个seriesID集合取并集或交集求最终集合。

比如，标签选择器{file="app.log", level=~"debug|error"}的工作过程如下：

1. 查询出file="app.log"，level="debug", level="error" 三个标签键值所对应的seriesID集合，S1 、S2、S3；
2. 根据三个集合计算最终seriesID集合S = S1∩cap (S2∪S3)。

• !=，=~，!~，工作过程如下：

1. 以如下SQL所描述的语义查询出标签选择器里引用的每个标签所对应seriesID集合。

SELECT * FROM Table_N WHERE hash = ?

<左右滑动以查看完整代码>

◆ hash为租户ID(userID)、分桶(bucketID)、标签名(labelName)。

2. 根据标签选择语法对每个seriesID集合进行过滤。

3. 将过滤后的集合进行并集、交集等操作求最终集合。

比如，{file~="mysql*", level!="error"}的工作过程如下：

1. 查询出标签“file”和标签"level"对应的seriesID的集合，S1、S2；

2. 求出S1中file的值匹配mysql*的子集SS1，S2中level的值!="error"的子集SS2；
3. 计算最终seriesID集合S = SS1∩SS2。

3.  以如下SQL所描述的语义查询出所有日志流所包含的chunk的ID

SELECT * FROM Table_N Where hash = ?

<左右滑动以查看完整代码>

• hash为分桶(bucketID)和日志流(seriesID)计算的哈希值。

4.  根据chunkID列表生成遍历器来顺序读取日志行

遍历器作为数据读取的组件，其主要功能为从存储系统中拉取chunk并从中读取日志行。其采用多层树形结构，自顶向下逐层递归触发方式弹出数据。具体结构如下图所示：

* batch Iterator：以批量的方式从存储中下载chunk原始数据，并生成iterator树；

* stream Iterator：多个stream数据的遍历器，其采用堆排序确保多个stream之间数据的保序；

* chunks Iterator：多个chunk数据的遍历器，同样采用堆排序确保多个chunk之间保序及多副本之间的去重；

* blocks Iterator：多个block数据的遍历器；

* block bytes Iterator：block里日志行的遍历器。

5. 从Ingester查询在内存中尚未写入到存储中的数据

由于Ingester是定时的将缓存数据写入到存储中，所以Querier在查询时间范围较新的数据时，还会通过grpc协议从每个ingester中查询出内存数据。需要在ingester中查询的时间范围是可配置的，视ingester缓存数据时长而定。

上面是日志内容查询的主要流程。至于指标查询的流程与其大同小异，只是增加了指标计算的遍历器层用于从查询出的日志计算指标数据。其他两种则更为简单，这里不再详细展开。

QueryFrontend

Loki对查询采用了计算后置的方式，类似于在大量原始数据上做grep，所以查询势必会消耗比较多的计算和内存资源。如果以单节点执行一个查询请求的话很容易因为大查询造成OOM、速度慢等性能瓶颈。为解决此问题，Loki采用了将单个查询分解在多个querier上并发执行方式，其中查询请求的分解和调度则由queryFrontend完成。

queryFrontend在Loki的整体架构上处于querier的前端，它作为数据读取操作的入口服务，其主要的组件及工作流程如下图所示：

1. 分割Request：将单个查询分割成子查询subReq的列表；

2. Feeder: 将子查询顺序注入到缓存队列 Buf Queue；

3. Runner: 多个并发的运行器将Buf Queue中的查询并注入到子查询队列，并等待返回查询结果；

4. Querier通过grpc协议实时从子查询队列弹出子查询，执行后将结果返回给相应的Runner；

5. 所有子请求在Runner执行完毕后汇总结果返回API响应。

查询分割

queryFrontend按照固定时间跨度将查询请求分割成多个子查询。比如，一个查询的时间范围是6小时，分割跨度为15分钟，则查询会被分为6*60/15=24个子查询

查询调度

Feeder

Feeder负责将分割好的子查询逐一的写入到缓存队列Buf Queue，以生产者/消费者模式与下游的Runner实现可控的子查询并发。

Runner

从Buf Queue中竞争方式读取子查询并写入到下游的请求队列中，并处理来自Querier的返回结果。Runner的并发个数通过全局配置控制，避免因为一次分解过多子查询而对Querier造成巨大的徒流量，影响其稳定性。

子查询队列

队列是一个二维结构，第一维存储的是不同租户的队列，第二维存储同一租户子查询列表，它们都是以FIFO的顺序组织里面的元素的入队出队

分配请求

queryFrontend是以被动方式分配查询请求，后端Querier与queryFrontend实时的通过grpc监听子查询队列，当有新请求时以如下顺序在队列中弹出下一个请求：

1. 以循环的方式遍历队列中的租户列表，寻找下一个有数据的租户队列；

2. 弹出该租户队列中的最老的请求。

总结

Loki作为一个正在快速发展的项目，最新版本已到2.0，相较1.6增强了诸如日志解析、Ruler、Boltdb-shipper等新功能，不过基本的模块、架构、数据模型、工作原理上已处于稳定状态，希望本文的这些尝试性的剖析能够能够为大家提供一些帮助，如文中有理解错误之处，欢迎批评指正。