for {
    select {
    case event := <-t.watcher.Events:
        switch event.Op {
        case fsnotify.Create:
            // protect against double Creates.
            if _, ok := t.tails[event.Name]; ok {
                level.Info(t.logger).Log("msg", "got 'create' for existing file", "filename", event.Name)
                continue
            }

            // newTailer 中会启动一个 goroutine 来读目标文件
            tailer := newTailer(t.logger, t.handler, t.positions, t.path, event.Name)
            t.tails[event.Name] = tailer
            
        case fsnotify.Remove:
            tailer, ok := t.tails[event.Name]
            if ok {
                // 关闭 tailer
                helpers.LogError("stopping tailer", tailer.stop)
                delete(t.tails, event.Name)
            }
        }
    case err := <-t.watcher.Errors:
        level.Error(t.logger).Log("msg", "error from fswatch", "error", err)
    case <-t.quit:
        return
    }
}

func newTailer() {
    tail := tail.TailFile(path, tail.Config{
        Follow: true,
        Location: &tail.SeekInfo{
            Offset: positions.Get(path),
            Whence: 0,
        },
    })
   
    tailer := ...
    go tailer.run()
}

func (t *tailer) run() {
    for {
        select {
        case <-positionWait.C:
            // 定时同步当前读取位置
            pos := t.tail.Tell()
            t.positions.Put(t.path, pos)
    
        case line, ok := <-t.tail.Lines:
            // handler.Handle() 中是一些日志行的预处理逻辑, 最后将日志行转化为 `Entry` 对象扔进 channel
            if err := t.handler.Handle(model.LabelSet{}, line.Time, line.Text); err != nil {
                level.Error(t.logger).Log("msg", "error handling line", "error", err)
            }
        }
    }
}

for {
    // 每次发送之后要重置计时器
    maxWait.Reset(c.cfg.BatchWait)
    select {
    case <-c.quit:
        return
    case e := <-c.entries:
        // Batch 足够大之后, 执行发送逻辑
        if batchSize+len(e.Line) > c.cfg.BatchSize {
            c.send(batch)
            // 重置 Batch
            batchSize = 0
            batch = map[model.Fingerprint]*logproto.Stream{}
        }

        // 收到 Entry, 先写进 Batch 当中
        batchSize += len(e.Line)
        
        // 每个 entry 要根据 label 放进对应的日志流(Stream)中
        fp := e.labels.FastFingerprint()
        stream, ok := batch[fp]
        if !ok {
            stream = &logproto.Stream{
                Labels: e.labels.String(),
            }
            batch[fp] = stream
        }
        stream.Entries = append(stream.Entries, e.Entry)
        
    case <-maxWait.C:
        // 到达每个批次的最大等待时间, 同样执行发送
        if len(batch) > 0 {
            c.send(batch);
            batchSize = 0
            batch = map[model.Fingerprint]*logproto.Stream{}
        }
    }
}

// 一次写入请求, 包含多段日志流
type PushRequest struct {
    Streams []*Stream `protobuf:"bytes,1,rep,name=streams" json:"streams,omitempty"`
}
// 一段日志流, 包含它的 label, 以及这段日志流当中的每个日志事件: Entry
type Stream struct {
    Labels  string  `protobuf:"bytes,1,opt,name=labels,proto3" json:"labels,omitempty"`
    Entries []Entry `protobuf:"bytes,2,rep,name=entries" json:"entries"`
}
// 一个日志事件, 包含时间戳与内容
type Entry struct {
    Timestamp time.Time `protobuf:"bytes,1,opt,name=timestamp,stdtime" json:"timestamp"`
    Line      string    `protobuf:"bytes,2,opt,name=line,proto3" json:"line,omitempty"`
}

streams := make([]streamTracker, len(req.Streams))
keys := make([]uint32, 0, len(req.Streams))
for i, stream := range req.Streams {
    // 获取每个 stream 的 label hash
    keys = append(keys, tokenFor(userID, stream.Labels))
    streams[i].stream = stream
}

// 根据 label hash 到 hash ring 上获取对应的 ingester 节点
// 这里的节点指 hash ring 上的节点, 一个节点可能有多个对等的 ingester 副本来做 HA
replicationSets := d.ring.BatchGet(keys, ring.Write)

// 将 Stream 按对应的 ingester 节点进行分组
samplesByIngester := map[string][]*streamTracker{}
ingesterDescs := map[string]ring.IngesterDesc{}
for i, replicationSet := range replicationSets {
    for _, ingester := range replicationSet.Ingesters {
        samplesByIngester[ingester.Addr] = append(samplesByIngester[ingester.Addr], &streams[i])
        ingesterDescs[ingester.Addr] = ingester
    }
}

for ingester, samples := range samplesByIngester {
    // 每组 Stream[] 又作为一个 PushRequest, 下发给对应的 ingester 节点
    d.sendSamples(localCtx, ingester, samples, &tracker)
}

func (i *instance) Push(ctx context.Context, req *logproto.PushRequest) error {
    for _, s := range req.Streams {
        // 将收到的日志流 Append 到内存中的日志流上, 同样地, 日志流按 label hash 索引
        fp := client.FastFingerprint(req.labels)
        stream, ok := i.streams[fp]
        if !ok {
            stream = newStream(fp, req.labels)
            // 这个过程中, 还会维护日志流的倒排索引(label -> stream)
            i.index.Add(labels, fp)
            i.streams[fp] = stream
        }
        stream.Push(ctx, s.Entries)
    }
    return nil
}

func (s *stream) Push(_ context.Context, entries []logproto.Entry) error {
    for i := range entries {
        // 假如当前 Chunk 已经关闭或者已经到达设定的最大 Chunk 大小, 则再创建一个新的 Chunk
        if s.chunks[0].closed || !s.chunks[0].chunk.SpaceFor(&entries[i]) {
            s.chunks = append(s.chunks, chunkDesc{
                chunk: chunkenc.NewMemChunk(chunkenc.EncGZIP),
            })
        }
        s.chunks[len(s.chunks)-1].chunk.Append(&entries[i])
    }
    return nil
}

// 一个 goroutine 将所有的待存储的 chunks enqueue
func (i *Ingester) sweepStream(instance *instance, stream *stream, immediate bool) {

    // 有一组待存储的队列(默认16个), 取模找一个队列把要存储的 chunk 的引用塞进去
    flushQueueIndex := int(uint64(stream.fp) % uint64(i.cfg.ConcurrentFlushes))
    firstTime, _ := stream.chunks[0].chunk.Bounds()
    i.flushQueues[flushQueueIndex].Enqueue(&flushOp{
        model.TimeFromUnixNano(firstTime.UnixNano()), instance.instanceID,
        stream.fp, immediate,
    })
}

// 每个队列都有一个 goroutine 作为消费者在 dequeue
func (i *Ingester) flushLoop(j int) {
    for {
        op := i.flushQueues[j].Dequeue()
        // 实际的存储操作在这个方法中, 存储完成后, Chunk 会被清理掉
        i.flushUserSeries(op.userID, op.fp, op.immediate)

        // 存储失败的 chunk 会重新塞回队列中
        if op.immediate && err != nil {
            op.from = op.from.Add(flushBackoff)
            i.flushQueues[j].Enqueue(op)
        }
    }
}

// 这部分代码实现了一个简单的二叉堆, MinHeap 和 MaxHeap 实现了相反的 `Less()` 方法
type iteratorHeap []EntryIterator
func (h iteratorHeap) Len() int            { return len(h) }
func (h iteratorHeap) Swap(i, j int)       { h[i], h[j] = h[j], h[i] }
func (h iteratorHeap) Peek() EntryIterator { return h[0] }
func (h *iteratorHeap) Push(x interface{}) {
    *h = append(*h, x.(EntryIterator))
}
func (h *iteratorHeap) Pop() interface{} {
    old := *h
    n := len(old)
    x := old[n-1]
    *h = old[0 : n-1]
    return x
}
type iteratorMinHeap struct {
    iteratorHeap
}
func (h iteratorMinHeap) Less(i, j int) bool {
    return h.iteratorHeap[i].Entry().Timestamp.Before(h.iteratorHeap[j].Entry().Timestamp)
}
type iteratorMaxHeap struct {
    iteratorHeap
}
func (h iteratorMaxHeap) Less(i, j int) bool {
    return h.iteratorHeap[i].Entry().Timestamp.After(h.iteratorHeap[j].Entry().Timestamp)
}

// 将一组 Stream 的 iterator 合并成一个 HeapIterator
func NewHeapIterator(is []EntryIterator, direction logproto.Direction) EntryIterator {
    result := &heapIterator{}
    switch direction {
    case logproto.BACKWARD:
        result.heap = &iteratorMaxHeap{}
    case logproto.FORWARD:
        result.heap = &iteratorMinHeap{}
    default:
        panic("bad direction")
    }
    // pre-next each iterator, drop empty.
    for _, i := range is {
        result.requeue(i)
    }
    return result
}

func (i *heapIterator) requeue(ei EntryIterator) {
    if ei.Next() {
        heap.Push(i.heap, ei)
        return
    }
    if err := ei.Error(); err != nil {
        i.errs = append(i.errs, err)
    }
    helpers.LogError("closing iterator", ei.Close)
}

func (i *heapIterator) Next() bool {
    if i.curr != nil {
        i.requeue(i.curr)
    }
    if i.heap.Len() == 0 {
        return false
    }
    i.curr = heap.Pop(i.heap).(EntryIterator)
    currEntry := i.curr.Entry()
    // keep popping entries off if they match, to dedupe
    for i.heap.Len() > 0 {
        next := i.heap.Peek()
        nextEntry := next.Entry()
        if !currEntry.Equal(nextEntry) {
            break
        }

        next = heap.Pop(i.heap).(EntryIterator)
        i.requeue(next)
    }
    return true
}