Loki 源码分析之日志写入

时间:2021-07-31 02:36:38

 Loki 源码分析之日志写入

前面我们介绍了 Loki 的一些基本使用配置,但是对 Loki 还是了解不够深入,官方文档写得较为凌乱,而且没有跟上新版本,为了能够对 Loki 有一个更深入的认识,做到有的放矢,这里面我们尝试对 Loki 的源码进行一些简单的分析,由于有很多模块和实现细节,这里我们主要是对核心功能进行分析,希望对大家有所帮助。本文首先对日志的写入过程进行简单分析。

Loki 源码分析之日志写入

Distributor Push API

 

Promtail 通过 Loki 的 Push API 接口推送日志数据,该接口在初始化 Distributor 的时候进行初始化,在控制器基础上包装了两个中间件,其中的 HTTPAuthMiddleware 就是获取租户 ID,如果开启了认证配置,则从 X-Scope-OrgID 这个请求 Header 头里面获取,如果没有配置则用默认的 fake 代替。

  1. // pkg/loki/modules.go 
  2. func (t *Loki) initDistributor() (services.Service, error) { 
  3.  ...... 
  4.  if t.cfg.Target != All { 
  5.   logproto.RegisterPusherServer(t.Server.GRPC, t.distributor) 
  6.  } 
  7.  
  8.  pushHandler := middleware.Merge( 
  9.   serverutil.RecoveryHTTPMiddleware, 
  10.   t.HTTPAuthMiddleware, 
  11.  ).Wrap(http.HandlerFunc(t.distributor.PushHandler)) 
  12.  
  13.  t.Server.HTTP.Handle("/api/prom/push", pushHandler) 
  14.  t.Server.HTTP.Handle("/loki/api/v1/push", pushHandler) 
  15.  return t.distributor, nil 

Push API 处理器实现如下所示,首先通过 ParseRequest 函数将 Http 请求转换成 logproto.PushRequest,然后直接调用 Distributor 下面的 Push 函数来推送日志数据:

  1. // pkg/distributor/http.go 
  2.  
  3. // PushHandler 从 HTTP body 中读取一个 snappy 压缩的 proto 
  4. func (d *Distributor) PushHandler(w http.ResponseWriter, r *http.Request) { 
  5.  logger := util_log.WithContext(r.Context(), util_log.Logger) 
  6.  userID, _ := user.ExtractOrgID(r.Context()) 
  7.  req, err := ParseRequest(logger, userID, r) 
  8.  ...... 
  9.  _, err = d.Push(r.Context(), req) 
  10.  ...... 
  11.  
  12. func ParseRequest(logger gokit.Logger, userID string, r *http.Request) (*logproto.PushRequest, error) { 
  13.  var body lokiutil.SizeReader 
  14.  contentEncoding := r.Header.Get(contentEnc) 
  15.  switch contentEncoding { 
  16.  case ""
  17.   body = lokiutil.NewSizeReader(r.Body) 
  18.  case "snappy"
  19.   body = lokiutil.NewSizeReader(r.Body) 
  20.  case "gzip"
  21.   gzipReader, err := gzip.NewReader(r.Body) 
  22.   if err != nil { 
  23.    return nil, err 
  24.   } 
  25.   defer gzipReader.Close() 
  26.   body = lokiutil.NewSizeReader(gzipReader) 
  27.  default
  28.   return nil, fmt.Errorf("Content-Encoding %q not supported", contentEncoding) 
  29.  } 
  30.  
  31.  contentType := r.Header.Get(contentType) 
  32.  var req logproto.PushRequest 
  33.  ...... 
  34.  switch contentType { 
  35.  case applicationJSON: 
  36.   var err error 
  37.   if loghttp.GetVersion(r.RequestURI) == loghttp.VersionV1 { 
  38.    err = unmarshal.DecodePushRequest(body, &req) 
  39.   } else { 
  40.    err = unmarshal_legacy.DecodePushRequest(body, &req) 
  41.   } 
  42.   if err != nil { 
  43.    return nil, err 
  44.   } 
  45.  default
  46.   // When no content-type header is set or when it is set to 
  47.   // `application/x-protobuf`: expect snappy compression. 
  48.   if err := util.ParseProtoReader(r.Context(), body, int(r.ContentLength), math.MaxInt32, &req, util.RawSnappy); err != nil { 
  49.    return nil, err 
  50.   } 
  51.  } 
  52.  return &req, nil 

首先我们先了解下 PushRequest 的结构,PushRequest 就是一个 Stream 集合:

  1. // pkg/logproto/logproto.pb.go 
  2. type PushRequest struct { 
  3.  Streams []Stream `protobuf:"bytes,1,rep,name=streams,proto3,customtype=Stream" json:"streams"
  4.  
  5. // pkg/logproto/types.go 
  6. // Stream 流包含一个唯一的标签集,作为一个字符串,然后还包含一组日志条目 
  7. type Stream struct { 
  8.  Labels  string  `protobuf:"bytes,1,opt,name=labels,proto3" json:"labels"
  9.  Entries []Entry `protobuf:"bytes,2,rep,name=entries,proto3,customtype=EntryAdapter" json:"entries"
  10.  
  11. // Entry 是一个带有时间戳的日志条目 
  12. type Entry struct { 
  13.  Timestamp time.Time `protobuf:"bytes,1,opt,name=timestamp,proto3,stdtime" json:"ts"
  14.  Line      string    `protobuf:"bytes,2,opt,name=line,proto3" json:"line"

 Loki 源码分析之日志写入

Loki 源码分析之日志写入

然后查看 Distributor 下的 Push 函数实现:

  1. // pkg/distributor/distributor.go 
  2. // Push 日志流集合 
  3. func (d *Distributor) Push(ctx context.Context, req *logproto.PushRequest) (*logproto.PushResponse, error) { 
  4.  // 获取租户ID 
  5.  userID, err := user.ExtractOrgID(ctx) 
  6.  ...... 
  7.  
  8.  // 首先把请求平铺成一个样本的列表 
  9.  streams := make([]streamTracker, 0, len(req.Streams)) 
  10.  keys := make([]uint32, 0, len(req.Streams)) 
  11.  var validationErr error 
  12.  validatedSamplesSize := 0 
  13.  validatedSamplesCount := 0 
  14.  
  15.  validationContext := d.validator.getValidationContextFor(userID) 
  16.  
  17.  for _, stream := range req.Streams { 
  18.   // 解析日志流标签 
  19.   stream.Labels, err = d.parseStreamLabels(validationContext, stream.Labels, &stream) 
  20.   ...... 
  21.   n := 0 
  22.   for _, entry := range stream.Entries { 
  23.    // 校验一个日志Entry实体 
  24.    if err := d.validator.ValidateEntry(validationContext, stream.Labels, entry); err != nil { 
  25.     validationErr = err 
  26.     continue 
  27.    } 
  28.    stream.Entries[n] = entry 
  29.    n++ 
  30.    // 校验成功的样本大小和个数 
  31.    validatedSamplesSize += len(entry.Line) 
  32.    validatedSamplesCount++ 
  33.   } 
  34.   // 去掉校验失败的实体 
  35.   stream.Entries = stream.Entries[:n] 
  36.  
  37.   if len(stream.Entries) == 0 { 
  38.    continue 
  39.   } 
  40.   // 为当前日志流生成用于hash换的token值 
  41.   keys = append(keys, util.TokenFor(userID, stream.Labels)) 
  42.   streams = append(streams, streamTracker{ 
  43.    stream: stream, 
  44.   }) 
  45.  } 
  46.  
  47.  if len(streams) == 0 { 
  48.   return &logproto.PushResponse{}, validationErr 
  49.  } 
  50.  
  51.  now := time.Now() 
  52.  // 每个租户有一个限速器,判断可以正常传输的日志大小是否应该被限制 
  53.  if !d.ingestionRateLimiter.AllowN(now, userID, validatedSamplesSize) { 
  54.   // 返回429表明客户端被限速了 
  55.   ...... 
  56.   return nil, httpgrpc.Errorf(http.StatusTooManyRequests, validation.RateLimitedErrorMsg, int(d.ingestionRateLimiter.Limit(now, userID)), validatedSamplesCount, validatedSamplesSize) 
  57.  } 
  58.  
  59.  const maxExpectedReplicationSet = 5 // typical replication factor 3 plus one for inactive plus one for luck 
  60.  var descs [maxExpectedReplicationSet]ring.InstanceDesc 
  61.  
  62.  samplesByIngester := map[string][]*streamTracker{} 
  63.  ingesterDescs := map[string]ring.InstanceDesc{} 
  64.  for i, key := range keys { 
  65.   // ReplicationSet 描述了一个指定的键与哪些 Ingesters 进行对话,以及可以容忍多少个错误 
  66.   // 根据 label hash 到 hash 环上获取对应的 ingester 节点,一个节点可能有多个对等的 ingester 副本来做 HA 
  67.   replicationSet, err := d.ingestersRing.Get(key, ring.Write, descs[:0], nil, nil) 
  68.   ...... 
  69.   // 最小成功的实例树 
  70.   streams[i].minSuccess = len(replicationSet.Ingesters) - replicationSet.MaxErrors 
  71.   // 可容忍的最大故障实例数 
  72.   streams[i].maxFailures = replicationSet.MaxErrors 
  73.   // 将 Stream 按对应的 ingester 进行分组 
  74.   for _, ingester := range replicationSet.Ingesters { 
  75.    // 配置每个 ingester 副本对应的日志流数据 
  76.    samplesByIngester[ingester.Addr] = append(samplesByIngester[ingester.Addr], &streams[i]) 
  77.    ingesterDescs[ingester.Addr] = ingester 
  78.   } 
  79.  } 
  80.  
  81.  tracker := pushTracker{ 
  82.   done: make(chan struct{}), 
  83.   err:  make(chan error), 
  84.  } 
  85.  tracker.samplesPending.Store(int32(len(streams))) 
  86.  // 循环Ingesters 
  87.  for ingester, samples := range samplesByIngester { 
  88.   // 让ingester并行处理通过hash环对应的日志流列表 
  89.   go func(ingester ring.InstanceDesc, samples []*streamTracker) { 
  90.    ...... 
  91.    // 将日志流样本数据下发给对应的 ingester 节点 
  92.    d.sendSamples(localCtx, ingester, samples, &tracker) 
  93.   }(ingesterDescs[ingester], samples) 
  94.  } 
  95.  ...... 

Push 函数的核心就是根据日志流的标签来计算一个 Token 值,根据这个 Token 值去哈希环上获取对应的处理日志的 Ingester 实例,然后并行通过 Ingester 处理日志流数据,通过 sendSamples 函数为单个 ingester 去发送日志样本数据:

  1. // pkg/distributor/distributor.go 
  2.  
  3. func (d *Distributor) sendSamples(ctx context.Context, ingester ring.InstanceDesc, streamTrackers []*streamTracker, pushTracker *pushTracker) { 
  4.  err := d.sendSamplesErr(ctx, ingester, streamTrackers) 
  5.  ...... 
  6.  
  7. func (d *Distributor) sendSamplesErr(ctx context.Context, ingester ring.InstanceDesc, streams []*streamTracker) error { 
  8.  // 根据 ingester 地址获取 client 
  9.  c, err := d.pool.GetClientFor(ingester.Addr) 
  10.  ...... 
  11.  // 重新构造 PushRequest 
  12.  req := &logproto.PushRequest{ 
  13.   Streams: make([]logproto.Stream, len(streams)), 
  14.  } 
  15.  for i, s := range streams { 
  16.   req.Streams[i] = s.stream 
  17.  } 
  18.  // 通过 Ingester 客户端请求数据 
  19.  _, err = c.(logproto.PusherClient).Push(ctx, req) 
  20.  ...... 

Ingester 写入日志

 

Ingester 客户端中的 Push 函数实际上就是一个 gRPC 服务的客户端:

  1. // pkg/ingester/ingester.go 
  2.  
  3. // Push 实现 logproto.Pusher. 
  4. func (i *Ingester) Push(ctx context.Context, req *logproto.PushRequest) (*logproto.PushResponse, error) { 
  5.  // 获取租户ID 
  6.  instanceID, err := user.ExtractOrgID(ctx) 
  7.  ...... 
  8.  // 根据租户ID获取 instance 对象 
  9.  instance := i.getOrCreateInstance(instanceID) 
  10.  // 直接调用 instance 对象 Push 数据 
  11.  err = instance.Push(ctx, req) 
  12.  return &logproto.PushResponse{}, err 

instance 下的 Push 函数:

  1. // pkg/ingester/instance.go 
  2.  
  3. func (i *instance) Push(ctx context.Context, req *logproto.PushRequest) error { 
  4.  record := recordPool.GetRecord() 
  5.  record.UserID = i.instanceID 
  6.  defer recordPool.PutRecord(record) 
  7.  
  8.  i.streamsMtx.Lock() 
  9.  defer i.streamsMtx.Unlock() 
  10.  
  11.  var appendErr error 
  12.  for _, s := range req.Streams { 
  13.   // 获取一个 stream 对象 
  14.   stream, err := i.getOrCreateStream(s, false, record) 
  15.   if err != nil { 
  16.    appendErr = err 
  17.    continue 
  18.   } 
  19.   // 真正用于数据处理的是 stream 对象中的 Push 函数 
  20.   if _, err := stream.Push(ctx, s.Entries, record); err != nil { 
  21.    appendErr = err 
  22.    continue 
  23.   } 
  24.  } 
  25.  ...... 
  26.  return appendErr 
  27.  
  28. func (i *instance) getOrCreateStream(pushReqStream logproto.Stream, lock bool, record *WALRecord) (*stream, error) { 
  29.  if lock { 
  30.   i.streamsMtx.Lock() 
  31.   defer i.streamsMtx.Unlock() 
  32.  } 
  33.  // 如果 streams 中包含当前标签列表对应的 stream 对象,则直接返回 
  34.  stream, ok := i.streams[pushReqStream.Labels] 
  35.  if ok { 
  36.   return stream, nil 
  37.  } 
  38.  // record 只在重放 WAL 时为 nil 
  39.  // 我们不希望在重放 WAL 后丢掉数据 
  40.  // 为 instance 降低 stream 流限制 
  41.  var err error 
  42.  if record != nil { 
  43.   // 限流器判断 
  44.   // AssertMaxStreamsPerUser 确保与当前输入的流数量没有达到限制 
  45.   err = i.limiter.AssertMaxStreamsPerUser(i.instanceID, len(i.streams)) 
  46.  } 
  47.  ...... 
  48.  // 解析日志流标签集 
  49.  labels, err := logql.ParseLabels(pushReqStream.Labels) 
  50.  ...... 
  51.  // 获取对应标签集的指纹 
  52.  fp := i.getHashForLabels(labels) 
  53.  // 重新实例化一个 stream 对象,这里还会维护日志流的倒排索引 
  54.  sortedLabels := i.index.Add(client.FromLabelsToLabelAdapters(labels), fp) 
  55.  stream = newStream(i.cfg, fp, sortedLabels, i.metrics) 
  56.  // 将stream设置到streams中去 
  57.  i.streams[pushReqStream.Labels] = stream 
  58.  i.streamsByFP[fp] = stream 
  59.  
  60.  // 当重放 wal 的时候 record 是 nil (我们不希望在重放时重写 wal entries). 
  61.  if record != nil { 
  62.   record.Series = append(record.Series, tsdb_record.RefSeries{ 
  63.    Ref:    uint64(fp), 
  64.    Labels: sortedLabels, 
  65.   }) 
  66.  } else { 
  67.   // 如果 record 为 nil,这就是一个 WAL 恢复 
  68.   i.metrics.recoveredStreamsTotal.Inc() 
  69.  } 
  70.  ...... 
  71.  i.addTailersToNewStream(stream) 
  72.  return stream, nil 

这个里面涉及到 WAL 这一块的设计,比较复杂,我们可以先看 stream 下面的 Push 函数实现,主要就是将收到的 []Entry 先 Append 到内存中的 Chunk 流([]chunkDesc) 中:

  1. // pkg/ingester/stream.go 
  2. func (s *stream) Push(ctx context.Context, entries []logproto.Entry, record *WALRecord) (int, error) { 
  3.  s.chunkMtx.Lock() 
  4.  defer s.chunkMtx.Unlock() 
  5.  var bytesAdded int 
  6.  prevNumChunks := len(s.chunks) 
  7.  var lastChunkTimestamp time.Time 
  8.  // 如果之前的 chunks 列表为空,则创建一个新的 chunk 
  9.  if prevNumChunks == 0 { 
  10.   s.chunks = append(s.chunks, chunkDesc{ 
  11.    chunk: s.NewChunk(), 
  12.   }) 
  13.   chunksCreatedTotal.Inc() 
  14.  } else { 
  15.   // 获取最新一个chunk的日志时间戳 
  16.   _, lastChunkTimestamp = s.chunks[len(s.chunks)-1].chunk.Bounds() 
  17.  } 
  18.  
  19.  var storedEntries []logproto.Entry 
  20.  failedEntriesWithError := []entryWithError{} 
  21.  
  22.  for i := range entries { 
  23.   // 如果这个日志条目与我们最后 append 的一行的时间戳和内容相匹配,则忽略它 
  24.   if entries[i].Timestamp.Equal(s.lastLine.ts) && entries[i].Line == s.lastLine.content { 
  25.    continue 
  26.   } 
  27.  
  28.   // 最新的一个 chunk 
  29.   chunk := &s.chunks[len(s.chunks)-1] 
  30.   // 如果当前chunk已经关闭 或者 已经达到设置的最大 Chunk 大小 
  31.   if chunk.closed || !chunk.chunk.SpaceFor(&entries[i]) || s.cutChunkForSynchronization(entries[i].Timestamp, lastChunkTimestamp, chunk, s.cfg.SyncPeriod, s.cfg.SyncMinUtilization) { 
  32.    // 如果 chunk 没有更多的空间,则调用 Close 来以确保 head block 中的数据都被切割和压缩。 
  33.    err := chunk.chunk.Close() 
  34.    ...... 
  35.    chunk.closed = true 
  36.    ...... 
  37.    // Append 一个新的 Chunk 
  38.    s.chunks = append(s.chunks, chunkDesc{ 
  39.     chunk: s.NewChunk(), 
  40.    }) 
  41.    chunk = &s.chunks[len(s.chunks)-1] 
  42.    lastChunkTimestamp = time.Time{} 
  43.   } 
  44.   // 往 chunk 里面 Append 日志数据 
  45.   if err := chunk.chunk.Append(&entries[i]); err != nil { 
  46.    failedEntriesWithError = append(failedEntriesWithError, entryWithError{&entries[i], err}) 
  47.   } else { 
  48.    // 存储添加到 chunk 中的日志数据 
  49.    storedEntries = append(storedEntries, entries[i]) 
  50.    // 配置最后日志行的数据 
  51.    lastChunkTimestamp = entries[i].Timestamp 
  52.    s.lastLine.ts = lastChunkTimestamp 
  53.    s.lastLine.content = entries[i].Line 
  54.    // 累计大小 
  55.    bytesAdded += len(entries[i].Line) 
  56.   } 
  57.   chunk.lastUpdated = time.Now() 
  58.  } 
  59.  
  60.  if len(storedEntries) != 0 { 
  61.   // 当重放 wal 的时候 record 将为 nil(我们不希望在重放的时候重写wal日志条目) 
  62.   if record != nil { 
  63.    record.AddEntries(uint64(s.fp), storedEntries...) 
  64.   } 
  65.   // 后续是用与tail日志的处理 
  66.   ...... 
  67.  } 
  68.  ...... 
  69.  // 如果新增了chunks 
  70.  if len(s.chunks) != prevNumChunks { 
  71.   memoryChunks.Add(float64(len(s.chunks) - prevNumChunks)) 
  72.  } 
  73.  return bytesAdded, nil 

Chunk 其实就是多条日志构成的压缩包,将日志压成 Chunk 的可以直接存入对象存储, 一个 Chunk 到达指定大小之前会不断 Append 新的日志到里面,而在达到大小之后, Chunk 就会关闭等待持久化(强制持久化也会关闭 Chunk, 比如关闭 ingester 实例时就会关闭所有的 Chunk 并持久化)。Chunk 的大小控制很重要:

  • 假如 Chunk 容量过小: 首先是导致压缩效率不高,同时也会增加整体的 Chunk 数量, 导致倒排索引过大,最后, 对象存储的操作次数也会变多, 带来额外的性能开销
  • 假如 Chunk 过大: 一个 Chunk 的 open 时间会更长, 占用额外的内存空间, 同时, 也增加了丢数据的风险,Chunk 过大也会导致查询读放大

Loki 源码分析之日志写入

(图片来源: https://aleiwu.com/post/grafana-loki/)

在将日志流追加到 Chunk 中过后,在 Ingester 初始化时会启动两个循环去处理 Chunk 数据,分别从 chunks 数据取出存入优先级队列,另外一个循环定期检查从内存中删除已经持久化过后的数据。

首先是 Ingester 中定义了一个 flushQueues 属性,是一个优先级队列数组,该队列中存放的是 flushOp:

  1. // pkg/ingester/ingester.go 
  2. type Ingester struct { 
  3.  services.Service 
  4.  ...... 
  5.  // 每个 flush 线程一个队列,指纹用来选择队列 
  6.  flushQueues     []*util.PriorityQueue  // 优先级队列数组 
  7.  flushQueuesDone sync.WaitGroup 
  8.  ...... 
  9.  
  10. // pkg/ingester/flush.go 
  11. // 优先级队列中存放的数据 
  12. type flushOp struct { 
  13.  from      model.Time 
  14.  userID    string 
  15.  fp        model.Fingerprint 
  16.  immediate bool 

在初始化 Ingester 的时候会根据传递的 ConcurrentFlushes 参数来实例化 flushQueues的大小:

  1. // pkg/ingester/ingester.go 
  2. func New(cfg Config, clientConfig client.Config, store ChunkStore, limits *validation.Overrides, configs *runtime.TenantConfigs, registerer prometheus.Registerer) (*Ingester, error) { 
  3.  ...... 
  4.  i := &Ingester{ 
  5.   ...... 
  6.   flushQueues:           make([]*util.PriorityQueue, cfg.ConcurrentFlushes), 
  7.   ...... 
  8.  } 
  9.  ...... 
  10.  i.Service = services.NewBasicService(i.starting, i.running, i.stopping) 
  11.  return i, nil 

然后通过 services.NewBasicService 实例化 Service 的时候指定了服务的 Starting、Running、Stopping 3 个状态,在其中的 staring 状态函数中会启动协程去消费优先级队列中的数据

  1. // pkg/ingester/ingester.go 
  2. func (i *Ingester) starting(ctx context.Context) error { 
  3.  // todo,如果开启了 WAL 的处理 
  4.  ...... 
  5.  // 初始化 flushQueues 
  6.  i.InitFlushQueues() 
  7.  ...... 
  8.  // 启动循环检查chunk数据 
  9.  i.loopDone.Add(1) 
  10.  go i.loop() 
  11.  return nil 

初始化 flushQueues 实现如下所示,其中 flushQueuesDone 是一个 WaitGroup,根据配置的并发数量并发执行 flushLoop 操作:

  1. // pkg/ingester/flush.go 
  2. func (i *Ingester) InitFlushQueues() { 
  3.  i.flushQueuesDone.Add(i.cfg.ConcurrentFlushes) 
  4.  for j := 0; j < i.cfg.ConcurrentFlushes; j++ { 
  5.   // 为每个协程构造一个优先级队列 
  6.   i.flushQueues[j] = util.NewPriorityQueue(flushQueueLength) 
  7.   go i.flushLoop(j) 
  8.  } 

每一个优先级队列循环消费数据:

  1. // pkg/ingester/flush.go 
  2. func (i *Ingester) flushLoop(j int) { 
  3.  ...... 
  4.  for { 
  5.   // 从队列中根据优先级取出数据 
  6.   o := i.flushQueues[j].Dequeue() 
  7.   if o == nil { 
  8.    return 
  9.   } 
  10.   op := o.(*flushOp) 
  11.   // 执行真正的刷新用户序列数据 
  12.   err := i.flushUserSeries(op.userID, op.fp, op.immediate) 
  13.   ...... 
  14.   // 如果退出时刷新失败了,把失败的操作放回到队列中去。 
  15.   if op.immediate && err != nil { 
  16.    op.from = op.from.Add(flushBackoff) 
  17.    i.flushQueues[j].Enqueue(op) 
  18.   } 
  19.  } 

刷新用户的序列操作,也就是要保存到存储中去:

  1. // pkg/ingester/flush.go 
  2. // 根据用户ID刷新用户日志序列 
  3. func (i *Ingester) flushUserSeries(userID string, fp model.Fingerprint, immediate bool) error { 
  4.  instance, ok := i.getInstanceByID(userID) 
  5.  ...... 
  6.  // 根据instance和fp指纹数据获取需要刷新的chunks 
  7.  chunks, labels, chunkMtx := i.collectChunksToFlush(instance, fp, immediate) 
  8.  ...... 
  9.  // 执行真正的刷新 chunks 操作 
  10.  err := i.flushChunks(ctx, fp, labels, chunks, chunkMtx) 
  11.  ...... 
  12.  
  13. // 收集需要刷新的 chunks 
  14. func (i *Ingester) collectChunksToFlush(instance *instance, fp model.Fingerprint, immediate bool) ([]*chunkDesc, labels.Labels, *sync.RWMutex) { 
  15.  instance.streamsMtx.Lock() 
  16.  // 根据指纹数据获取 stream 
  17.  stream, ok := instance.streamsByFP[fp] 
  18.  instance.streamsMtx.Unlock() 
  19.  if !ok { 
  20.   return nil, nil, nil 
  21.  } 
  22.  
  23.  var result []*chunkDesc 
  24.  stream.chunkMtx.Lock() 
  25.  defer stream.chunkMtx.Unlock() 
  26.  // 循环所有chunks 
  27.  for j := range stream.chunks { 
  28.   // 判断是否应该刷新当前chunk 
  29.   shouldFlush, reason := i.shouldFlushChunk(&stream.chunks[j]) 
  30.   if immediate || shouldFlush { 
  31.    // 确保不再对该块进行写操作(如果没有关闭,则设置为关闭状态) 
  32.    if !stream.chunks[j].closed { 
  33.     stream.chunks[j].closed = true 
  34.    } 
  35.    // 如果该 chunk 还没有被成功刷新,则刷新这个块 
  36.    if stream.chunks[j].flushed.IsZero() { 
  37.     result = append(result, &stream.chunks[j]) 
  38.     ...... 
  39.    } 
  40.   } 
  41.  } 
  42.  return result, stream.labels, &stream.chunkMtx 

下面是判断一个具体的 chunk 是否应该被刷新的逻辑:

  1. // pkg/ingester/flush.go 
  2. func (i *Ingester) shouldFlushChunk(chunk *chunkDesc) (bool, string) { 
  3.  // chunk关闭了也应该刷新了 
  4.  if chunk.closed { 
  5.   if chunk.synced { 
  6.    return true, flushReasonSynced 
  7.   } 
  8.   return true, flushReasonFull 
  9.  } 
  10.  // chunk最后更新的时间超过了配置的 chunk 空闲时间 MaxChunkIdle 
  11.  if time.Since(chunk.lastUpdated) > i.cfg.MaxChunkIdle { 
  12.   return true, flushReasonIdle 
  13.  } 
  14.  
  15.  // chunk的边界时间操过了配置的 chunk  最大时间 MaxChunkAge 
  16.  if fromto := chunk.chunk.Bounds(); to.Sub(from) > i.cfg.MaxChunkAge { 
  17.   return true, flushReasonMaxAge 
  18.  } 
  19.  return false"" 

真正将 chunks 数据刷新保存到存储中是 flushChunks 函数实现的:

  1. // pkg/ingester/flush.go 
  2. func (i *Ingester) flushChunks(ctx context.Context, fp model.Fingerprint, labelPairs labels.Labels, cs []*chunkDesc, chunkMtx sync.Locker) error { 
  3.  ...... 
  4.  wireChunks := make([]chunk.Chunk, len(cs)) 
  5.  // 下面的匿名函数用于生成保存到存储中的chunk数据 
  6.  err = func() error { 
  7.   chunkMtx.Lock() 
  8.   defer chunkMtx.Unlock() 
  9.  
  10.   for j, c := range cs { 
  11.    if err := c.chunk.Close(); err != nil { 
  12.     return err 
  13.    } 
  14.    firstTime, lastTime := loki_util.RoundToMilliseconds(c.chunk.Bounds()) 
  15.    ch := chunk.NewChunk( 
  16.     userID, fp, metric, 
  17.     chunkenc.NewFacade(c.chunk, i.cfg.BlockSize, i.cfg.TargetChunkSize), 
  18.     firstTime, 
  19.     lastTime, 
  20.    ) 
  21.  
  22.    chunkSize := c.chunk.BytesSize() + 4*1024 // size + 4kB should be enough room for cortex header 
  23.    start := time.Now() 
  24.    if err := ch.EncodeTo(bytes.NewBuffer(make([]byte, 0, chunkSize))); err != nil { 
  25.     return err 
  26.    } 
  27.    wireChunks[j] = ch 
  28.   } 
  29.   return nil 
  30.  }() 
  31.  
  32.  
  33.  // 通过 store 接口保存 chunk 数据 
  34.  if err := i.store.Put(ctx, wireChunks); err != nil { 
  35.   return err 
  36.  } 
  37.  
  38.  ...... 
  39.  
  40.  chunkMtx.Lock() 
  41.  defer chunkMtx.Unlock() 
  42.  for i, wc := range wireChunks { 
  43.   // flush 成功,写入刷新时间 
  44.   cs[i].flushed = time.Now() 
  45.   // 下是一些监控数据更新 
  46.   ...... 
  47.  } 
  48.  
  49.  return nil 

chunk 数据被写入到存储后,还有有一个协程会去定时清理本地的这些 chunk 数据,在上面的 Ingester 的 staring 函数中最后有一个 go i.loop(),在这个 loop() 函数中会每隔 FlushCheckPeriod(默认 30s,可以通过 --ingester.flush-check-period 进行配置)时间就会去去调用 sweepUsers 函数进行垃圾回收:

  1. // pkg/ingester/ingester.go 
  2. func (i *Ingester) loop() { 
  3.  defer i.loopDone.Done() 
  4.  
  5.  flushTicker := time.NewTicker(i.cfg.FlushCheckPeriod) 
  6.  defer flushTicker.Stop() 
  7.  
  8.  for { 
  9.   select { 
  10.   case <-flushTicker.C: 
  11.    i.sweepUsers(falsetrue
  12.   case <-i.loopQuit: 
  13.    return 
  14.   } 
  15.  } 

sweepUsers 函数用于执行将日志流数据加入到优先级队列中,并对没有序列的用户进行垃圾回收:

  1. // pkg/ingester/flush.go 
  2. // sweepUsers 定期执行 flush 操作,并对没有序列的用户进行垃圾回收 
  3. func (i *Ingester) sweepUsers(immediate, mayRemoveStreams bool) { 
  4.  instances := i.getInstances() 
  5.  for _, instance := range instances { 
  6.   i.sweepInstance(instance, immediate, mayRemoveStreams) 
  7.  } 
  8.  
  9. func (i *Ingester) sweepInstance(instance *instance, immediate, mayRemoveStreams bool) { 
  10.  instance.streamsMtx.Lock() 
  11.  defer instance.streamsMtx.Unlock() 
  12.  for _, stream := range instance.streams { 
  13.   i.sweepStream(instance, stream, immediate) 
  14.   i.removeFlushedChunks(instance, stream, mayRemoveStreams) 
  15.  } 
  16.  
  17. // must hold streamsMtx 
  18. func (i *Ingester) sweepStream(instance *instance, stream *stream, immediate bool) { 
  19.  stream.chunkMtx.RLock() 
  20.  defer stream.chunkMtx.RUnlock() 
  21.  if len(stream.chunks) == 0 { 
  22.   return 
  23.  } 
  24.  // 最新的chunk 
  25.  lastChunk := stream.chunks[len(stream.chunks)-1] 
  26.  // 判断是否应该被flush 
  27.  shouldFlush, _ := i.shouldFlushChunk(&lastChunk) 
  28.  // 如果只有一个chunk并且不是强制持久化切最新的chunk还不应该被flush,则直接返回 
  29.  if len(stream.chunks) == 1 && !immediate && !shouldFlush { 
  30.   return 
  31.  } 
  32.  // 根据指纹获取用与处理的优先级队列索引 
  33.  flushQueueIndex := int(uint64(stream.fp) % uint64(i.cfg.ConcurrentFlushes)) 
  34.  firstTime, _ := stream.chunks[0].chunk.Bounds() 
  35.  // 加入到优先级队列中去 
  36.  i.flushQueues[flushQueueIndex].Enqueue(&flushOp{ 
  37.   model.TimeFromUnixNano(firstTime.UnixNano()), instance.instanceID, 
  38.   stream.fp, immediate, 
  39.  }) 
  40.  
  41. // 移除已经flush过后的chunks数据 
  42. func (i *Ingester) removeFlushedChunks(instance *instance, stream *stream, mayRemoveStream bool) { 
  43.  now := time.Now() 
  44.  
  45.  stream.chunkMtx.Lock() 
  46.  defer stream.chunkMtx.Unlock() 
  47.  prevNumChunks := len(stream.chunks) 
  48.  var subtracted int 
  49.  for len(stream.chunks) > 0 { 
  50.   // 如果chunk还没有被刷新到存储 或者 chunk被刷新到存储到现在的时间还没操过 RetainPeriod(默认15分钟,可以通过--ingester.chunks-retain-period 进行配置)则忽略 
  51.   if stream.chunks[0].flushed.IsZero() || now.Sub(stream.chunks[0].flushed) < i.cfg.RetainPeriod { 
  52.    break 
  53.   } 
  54.   subtracted += stream.chunks[0].chunk.UncompressedSize() 
  55.   // 删除引用,以便该块可以被垃圾回收起来 
  56.   stream.chunks[0].chunk = nil 
  57.   // 移除chunk 
  58.   stream.chunks = stream.chunks[1:] 
  59.  } 
  60.  ...... 
  61.  // 如果stream中的所有chunk都被清空了,则清空该 stream 的相关数据 
  62.  if mayRemoveStream && len(stream.chunks) == 0 { 
  63.   delete(instance.streamsByFP, stream.fp) 
  64.   delete(instance.streams, stream.labelsString) 
  65.   instance.index.Delete(stream.labels, stream.fp) 
  66.   ...... 
  67.  } 

关于存储或者查询等模块的实现在后文再继续探索,包括 WAL 的实现也较为复杂。

原文链接:https://mp.weixin.qq.com/s/9BKZvNANqGJwziygytJ1ag