func (w *WALWatcher) watch(segmentNum int, tail bool) error {
	segment, err := wal.OpenReadSegment(wal.SegmentName(w.walDir, segmentNum))
	if err != nil {
		return err
	}
	defer segment.Close()

	reader := wal.NewLiveReader(w.logger, segment)

	readTicker := time.NewTicker(readPeriod)
	defer readTicker.Stop()

	checkpointTicker := time.NewTicker(checkpointPeriod)
	defer checkpointTicker.Stop()

	segmentTicker := time.NewTicker(segmentCheckPeriod)
	defer segmentTicker.Stop()

	// If we're replaying the segment we need to know the size of the file to know
	// when to return from watch and move on to the next segment.
	size := int64(math.MaxInt64)
	if !tail {
		segmentTicker.Stop()
		checkpointTicker.Stop()
		var err error
		size, err = getSegmentSize(w.walDir, segmentNum)
		if err != nil {
			return errors.Wrap(err, "getSegmentSize")
		}
	}

	for {
		select {
		case <-w.quit:
			return nil

		case <-checkpointTicker.C:
			// Periodically check if there is a new checkpoint so we can garbage
			// collect labels. As this is considered an optimisation, we ignore
			// errors during checkpoint processing.
			if err := w.garbageCollectSeries(segmentNum); err != nil {
				level.Warn(w.logger).Log("msg", "error process checkpoint", "err", err)
			}

		case <-segmentTicker.C:
			_, last, err := w.firstAndLast()
			if err != nil {
				return errors.Wrap(err, "segments")
			}

			// Check if new segments exists.
			if last <= segmentNum {
				continue
			}

			err = w.readSegment(reader, segmentNum, tail)

			// Ignore errors reading to end of segment whilst replaying the WAL.
			if !tail {
				if err != nil && err != io.EOF {
					level.Warn(w.logger).Log("msg", "ignoring error reading to end of segment, may have dropped data", "err", err)
				} else if reader.Offset() != size {
					level.Warn(w.logger).Log("msg", "expected to have read whole segment, may have dropped data", "segment", segmentNum, "read", reader.Offset(), "size", size)
				}
				return nil
			}

			// Otherwise, when we are tailing, non-EOFs are fatal.
			if err != io.EOF {
				return err
			}

			return nil

		case <-readTicker.C:
			err = w.readSegment(reader, segmentNum, tail)

			// Ignore all errors reading to end of segment whilst replaying the WAL.
			if !tail {
				if err != nil && err != io.EOF {
					level.Warn(w.logger).Log("msg", "ignoring error reading to end of segment, may have dropped data", "segment", segmentNum, "err", err)
				} else if reader.Offset() != size {
					level.Warn(w.logger).Log("msg", "expected to have read whole segment, may have dropped data", "segment", segmentNum, "read", reader.Offset(), "size", size)
				}
				return nil
			}

			// Otherwise, when we are tailing, non-EOFs are fatal.
			if err != io.EOF {
				return err
			}
		}
	}
}
func (w *WALWatcher) watch(segmentNum int, tail bool) error {
	segment, err := wal.OpenReadSegment(wal.SegmentName(w.walDir, segmentNum))
	if err != nil {
		return err
	}
	defer segment.Close()

	reader := wal.NewLiveReader(w.logger, segment)

	readTicker := time.NewTicker(readPeriod)
	defer readTicker.Stop()

	checkpointTicker := time.NewTicker(checkpointPeriod)
	defer checkpointTicker.Stop()

	segmentTicker := time.NewTicker(segmentCheckPeriod)
	defer segmentTicker.Stop()

	// If we're replaying the segment we need to know the size of the file to know
	// when to return from watch and move on to the next segment.
	size := int64(math.MaxInt64)
	if !tail {
		segmentTicker.Stop()
		checkpointTicker.Stop()
		var err error
		size, err = getSegmentSize(w.walDir, segmentNum)
		if err != nil {
			return errors.Wrap(err, "getSegmentSize")
		}
	}

	for {
		select {
		case <-w.quit:
			return nil

		case <-checkpointTicker.C:
			// Periodically check if there is a new checkpoint so we can garbage
			// collect labels. As this is considered an optimisation, we ignore
			// errors during checkpoint processing.
			if err := w.garbageCollectSeries(segmentNum); err != nil {
				level.Warn(w.logger).Log("msg", "error process checkpoint", "err", err)
			}

		case <-segmentTicker.C:
			_, last, err := w.firstAndLast()
			if err != nil {
				return errors.Wrap(err, "segments")
			}

			// Check if new segments exists.
			if last <= segmentNum {
				continue
			}

			err = w.readSegment(reader, segmentNum, tail)

			// Ignore errors reading to end of segment whilst replaying the WAL.
			if !tail {
				if err != nil && err != io.EOF {
					level.Warn(w.logger).Log("msg", "ignoring error reading to end of segment, may have dropped data", "err", err)
				} else if reader.Offset() != size {
					level.Warn(w.logger).Log("msg", "expected to have read whole segment, may have dropped data", "segment", segmentNum, "read", reader.Offset(), "size", size)
				}
				return nil
			}

			// Otherwise, when we are tailing, non-EOFs are fatal.
			if err != io.EOF {
				return err
			}

			return nil

		case <-readTicker.C:
			err = w.readSegment(reader, segmentNum, tail)

			// Ignore all errors reading to end of segment whilst replaying the WAL.
			if !tail {
				if err != nil && err != io.EOF {
					level.Warn(w.logger).Log("msg", "ignoring error reading to end of segment, may have dropped data", "segment", segmentNum, "err", err)
				} else if reader.Offset() != size {
					level.Warn(w.logger).Log("msg", "expected to have read whole segment, may have dropped data", "segment", segmentNum, "read", reader.Offset(), "size", size)
				}
				return nil
			}

			// Otherwise, when we are tailing, non-EOFs are fatal.
			if err != io.EOF {
				return err
			}
		}
	}
}
func ( w * WALWatcher ) watch ( segmentNum int , tail bool ) error { segment , err := wal . OpenReadSegment ( wal . SegmentName ( w . walDir , segmentNum ) ) if err != nil { return err } defer segment . Close ( ) reader := wal . NewLiveReader ( w . logger , segment ) readTicker := time . NewTicker ( readPeriod ) defer readTicker . Stop ( ) checkpointTicker := time . NewTicker ( checkpointPeriod ) defer checkpointTicker . Stop ( ) segmentTicker := time . NewTicker ( segmentCheckPeriod ) defer segmentTicker . Stop ( ) // If we're replaying the segment we need to know the size of the file to know // when to return from watch and move on to the next segment. size := int64 ( math . MaxInt64 ) if ! tail { segmentTicker . Stop ( ) checkpointTicker . Stop ( ) var err error size , err = getSegmentSize ( w . walDir , segmentNum ) if err != nil { return errors . Wrap ( err , " " ) } } for { select { case <- w . quit : return nil case <- checkpointTicker . C : // Periodically check if there is a new checkpoint so we can garbage // collect labels. As this is considered an optimisation, we ignore // errors during checkpoint processing. if err := w . garbageCollectSeries ( segmentNum ) ; err != nil { level . Warn ( w . logger ) . Log ( " " , " " , " " , err ) } case <- segmentTicker . C : _ , last , err := w . firstAndLast ( ) if err != nil { return errors . Wrap ( err , " " ) } // Check if new segments exists. if last <= segmentNum { continue } err = w . readSegment ( reader , segmentNum , tail ) // Ignore errors reading to end of segment whilst replaying the WAL. if ! tail { if err != nil && err != io . EOF { level . Warn ( w . logger ) . Log ( " " , " " , " " , err ) } else if reader . Offset ( ) != size { level . Warn ( w . logger ) . Log ( " " , " " , " " , segmentNum , " " , reader . Offset ( ) , " " , size ) } return nil } // Otherwise, when we are tailing, non-EOFs are fatal. if err != io . EOF { return err } return nil case <- readTicker . C : err = w . readSegment ( reader , segmentNum , tail ) // Ignore all errors reading to end of segment whilst replaying the WAL. if ! tail { if err != nil && err != io . EOF { level . Warn ( w . logger ) . Log ( " " , " " , " " , segmentNum , " " , err ) } else if reader . Offset ( ) != size { level . Warn ( w . logger ) . Log ( " " , " " , " " , segmentNum , " " , reader . Offset ( ) , " " , size ) } return nil } // Otherwise, when we are tailing, non-EOFs are fatal. if err != io . EOF { return err } } } }
// Use tail true to indicate that the reader is currently on a segment that is
// actively being written to. If false, assume it's a full segment and we're
// replaying it on start to cache the series records.
Use tail true to indicate that the reader is currently on a segment that is actively being written to . If false assume it s a full segment and we re replaying it on start to cache the series records .