func (s *journald) followJournal(logWatcher *logger.LogWatcher, config logger.ReadConfig, j *C.sd_journal, pfd [2]C.int, cursor string) {
	s.readers.mu.Lock()
	s.readers.readers[logWatcher] = logWatcher
	s.readers.mu.Unlock()
	go func() {
		// Keep copying journal data out until we're notified to stop
		// or we hit an error.
		status := C.wait_for_data_or_close(j, pfd[0])
		for status == 1 {
			cursor = s.drainJournal(logWatcher, config, j, cursor)
			status = C.wait_for_data_or_close(j, pfd[0])
		}
		if status < 0 {
			cerrstr := C.strerror(C.int(-status))
			errstr := C.GoString(cerrstr)
			fmtstr := "error %q while attempting to follow journal for container %q"
			logrus.Errorf(fmtstr, errstr, s.vars["CONTAINER_ID_FULL"])
		}
		// Clean up.
		C.close(pfd[0])
		s.readers.mu.Lock()
		delete(s.readers.readers, logWatcher)
		s.readers.mu.Unlock()
		C.sd_journal_close(j)
		close(logWatcher.Msg)
	}()
	// Wait until we're told to stop.
	select {
	case <-logWatcher.WatchClose():
		// Notify the other goroutine that its work is done.
		C.close(pfd[1])
	}
}

func (s *journald) followJournal(logWatcher *logger.LogWatcher, config logger.ReadConfig, j *C.sd_journal, pfd [2]C.int, cursor string) {
	go func() {
		// Keep copying journal data out until we're notified to stop.
		for C.wait_for_data_or_close(j, pfd[0]) == 1 {
			cursor = s.drainJournal(logWatcher, config, j, cursor)
		}
		// Clean up.
		C.close(pfd[0])
		s.readers.mu.Lock()
		delete(s.readers.readers, logWatcher)
		s.readers.mu.Unlock()
	}()
	s.readers.mu.Lock()
	s.readers.readers[logWatcher] = logWatcher
	s.readers.mu.Unlock()
	// Wait until we're told to stop.
	select {
	case <-logWatcher.WatchClose():
		// Notify the other goroutine that its work is done.
		C.close(pfd[1])
	}
}

func followLogs(f *os.File, logWatcher *logger.LogWatcher, notifyRotate chan interface{}, since time.Time) {
	dec := json.NewDecoder(f)
	l := &jsonlog.JSONLog{}
	fileWatcher, err := fsnotify.NewWatcher()
	if err != nil {
		logWatcher.Err <- err
		return
	}
	defer fileWatcher.Close()
	if err := fileWatcher.Add(f.Name()); err != nil {
		logWatcher.Err <- err
		return
	}

	var retries int
	for {
		msg, err := decodeLogLine(dec, l)
		if err != nil {
			if err != io.EOF {
				// try again because this shouldn't happen
				if _, ok := err.(*json.SyntaxError); ok && retries <= maxJSONDecodeRetry {
					dec = json.NewDecoder(f)
					retries += 1
					continue
				}
				logWatcher.Err <- err
				return
			}

			select {
			case <-fileWatcher.Events:
				dec = json.NewDecoder(f)
				continue
			case <-fileWatcher.Errors:
				logWatcher.Err <- err
				return
			case <-logWatcher.WatchClose():
				return
			case <-notifyRotate:
				fileWatcher.Remove(f.Name())

				f, err = os.Open(f.Name())
				if err != nil {
					logWatcher.Err <- err
					return
				}
				if err := fileWatcher.Add(f.Name()); err != nil {
					logWatcher.Err <- err
				}
				dec = json.NewDecoder(f)
				continue
			}
		}

		retries = 0 // reset retries since we've succeeded
		if !since.IsZero() && msg.Timestamp.Before(since) {
			continue
		}
		select {
		case logWatcher.Msg <- msg:
		case <-logWatcher.WatchClose():
			logWatcher.Msg <- msg
			for {
				msg, err := decodeLogLine(dec, l)
				if err != nil {
					return
				}
				if !since.IsZero() && msg.Timestamp.Before(since) {
					continue
				}
				logWatcher.Msg <- msg
			}
		}
	}
}

func followLogs(f *os.File, logWatcher *logger.LogWatcher, notifyRotate chan interface{}, since time.Time) {
	dec := json.NewDecoder(f)
	l := &jsonlog.JSONLog{}

	name := f.Name()
	fileWatcher, err := watchFile(name)
	if err != nil {
		logWatcher.Err <- err
		return
	}
	defer func() {
		f.Close()
		fileWatcher.Remove(name)
		fileWatcher.Close()
	}()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() {
		select {
		case <-logWatcher.WatchClose():
			fileWatcher.Remove(name)
			cancel()
		case <-ctx.Done():
			return
		}
	}()

	var retries int
	handleRotate := func() error {
		f.Close()
		fileWatcher.Remove(name)

		// retry when the file doesn't exist
		for retries := 0; retries <= 5; retries++ {
			f, err = os.Open(name)
			if err == nil || !os.IsNotExist(err) {
				break
			}
		}
		if err != nil {
			return err
		}
		if err := fileWatcher.Add(name); err != nil {
			return err
		}
		dec = json.NewDecoder(f)
		return nil
	}

	errRetry := errors.New("retry")
	errDone := errors.New("done")
	waitRead := func() error {
		select {
		case e := <-fileWatcher.Events():
			switch e.Op {
			case fsnotify.Write:
				dec = json.NewDecoder(f)
				return nil
			case fsnotify.Rename, fsnotify.Remove:
				select {
				case <-notifyRotate:
				case <-ctx.Done():
					return errDone
				}
				if err := handleRotate(); err != nil {
					return err
				}
				return nil
			}
			return errRetry
		case err := <-fileWatcher.Errors():
			logrus.Debug("logger got error watching file: %v", err)
			// Something happened, let's try and stay alive and create a new watcher
			if retries <= 5 {
				fileWatcher.Close()
				fileWatcher, err = watchFile(name)
				if err != nil {
					return err
				}
				retries++
				return errRetry
			}
			return err
		case <-ctx.Done():
			return errDone
		}
	}

	handleDecodeErr := func(err error) error {
		if err == io.EOF {
			for err := waitRead(); err != nil; {
				if err == errRetry {
					// retry the waitRead
					continue
				}
				return err
			}
			return nil
		}
//.........这里部分代码省略.........

func followLogs(f *os.File, logWatcher *logger.LogWatcher, notifyRotate chan interface{}, since time.Time) {
	dec := json.NewDecoder(f)
	l := &jsonlog.JSONLog{}

	fileWatcher, err := filenotify.New()
	if err != nil {
		logWatcher.Err <- err
	}
	defer func() {
		f.Close()
		fileWatcher.Close()
	}()
	name := f.Name()

	if err := fileWatcher.Add(name); err != nil {
		logrus.WithField("logger", "json-file").Warnf("falling back to file poller due to error: %v", err)
		fileWatcher.Close()
		fileWatcher = filenotify.NewPollingWatcher()

		if err := fileWatcher.Add(name); err != nil {
			logrus.Debugf("error watching log file for modifications: %v", err)
			logWatcher.Err <- err
			return
		}
	}

	var retries int
	for {
		msg, err := decodeLogLine(dec, l)
		if err != nil {
			if err != io.EOF {
				// try again because this shouldn't happen
				if _, ok := err.(*json.SyntaxError); ok && retries <= maxJSONDecodeRetry {
					dec = json.NewDecoder(f)
					retries++
					continue
				}

				// io.ErrUnexpectedEOF is returned from json.Decoder when there is
				// remaining data in the parser's buffer while an io.EOF occurs.
				// If the json logger writes a partial json log entry to the disk
				// while at the same time the decoder tries to decode it, the race condition happens.
				if err == io.ErrUnexpectedEOF && retries <= maxJSONDecodeRetry {
					reader := io.MultiReader(dec.Buffered(), f)
					dec = json.NewDecoder(reader)
					retries++
					continue
				}

				return
			}

			select {
			case <-fileWatcher.Events():
				dec = json.NewDecoder(f)
				continue
			case <-fileWatcher.Errors():
				logWatcher.Err <- err
				return
			case <-logWatcher.WatchClose():
				fileWatcher.Remove(name)
				return
			case <-notifyRotate:
				f.Close()
				fileWatcher.Remove(name)

				// retry when the file doesn't exist
				for retries := 0; retries <= 5; retries++ {
					f, err = os.Open(name)
					if err == nil || !os.IsNotExist(err) {
						break
					}
				}

				if err = fileWatcher.Add(name); err != nil {
					logWatcher.Err <- err
					return
				}
				if err != nil {
					logWatcher.Err <- err
					return
				}

				dec = json.NewDecoder(f)
				continue
			}
		}

		retries = 0 // reset retries since we've succeeded
		if !since.IsZero() && msg.Timestamp.Before(since) {
			continue
		}
		select {
		case logWatcher.Msg <- msg:
		case <-logWatcher.WatchClose():
			logWatcher.Msg <- msg
			for {
				msg, err := decodeLogLine(dec, l)
				if err != nil {
					return
//.........这里部分代码省略.........

func (l *JSONFileLogger) followLogs(f *os.File, logWatcher *logger.LogWatcher, notifyRotate chan interface{}, since time.Time) {
	var (
		rotated bool

		dec         = json.NewDecoder(f)
		log         = &jsonlog.JSONLog{}
		writeNotify = l.writeNotifier.Subscribe()
		watchClose  = logWatcher.WatchClose()
	)

	reopenLogFile := func() error {
		f.Close()
		f, err := os.Open(f.Name())
		if err != nil {
			return err
		}
		dec = json.NewDecoder(f)
		rotated = true
		return nil
	}

	readToEnd := func() error {
		for {
			msg, err := decodeLogLine(dec, log)
			if err != nil {
				return err
			}
			if !since.IsZero() && msg.Timestamp.Before(since) {
				continue
			}
			logWatcher.Msg <- msg
		}
	}

	defer func() {
		l.writeNotifier.Evict(writeNotify)
		if rotated {
			f.Close()
		}
	}()

	for {
		select {
		case <-watchClose:
			readToEnd()
			return
		case <-notifyRotate:
			readToEnd()
			if err := reopenLogFile(); err != nil {
				logWatcher.Err <- err
				return
			}
		case _, ok := <-writeNotify:
			if err := readToEnd(); err == io.EOF {
				if !ok {
					// The writer is closed, no new logs will be generated.
					return
				}

				select {
				case <-notifyRotate:
					if err := reopenLogFile(); err != nil {
						logWatcher.Err <- err
						return
					}
				default:
					dec = json.NewDecoder(f)
				}

			} else if err == io.ErrUnexpectedEOF {
				dec = json.NewDecoder(io.MultiReader(dec.Buffered(), f))
			} else {
				logrus.Errorf("Failed to decode json log %s: %v", f.Name(), err)
				logWatcher.Err <- err
				return
			}
		}
	}
}