Function loggingProcessAdapter

pkg/logging/logging.go:253–391

(ctx context.Context, driver Driver, dataStore, address string, getContainerWait ContainerWaitFunc, config *logging.Config)

type ContainerWaitFunc func(ctx context.Context, address string, config *logging.Config, outputSeen func() bool) (<-chan containerd.ExitStatus, error)

func loggingProcessAdapter(ctx context.Context, driver Driver, dataStore, address string, getContainerWait ContainerWaitFunc, config *logging.Config) error {
	if err := driver.PreProcess(ctx, dataStore, config); err != nil {
		return err
	}

	stdoutR, err := cancelreader.NewReader(config.Stdout)
	if err != nil {
		return err
	}
	stderrR, err := cancelreader.NewReader(config.Stderr)
	if err != nil {
		return err
	}
	go func() {
		<-ctx.Done() // delivered on SIGTERM
		stdoutR.Cancel()
		stderrR.Cancel()
	}()

	// copiedBytes counts how much container output has been read so far. It lets
	// getContainerWait tell "the task has not been created yet" apart from "the
	// task has already exited and been removed" when it sees a missing task.
	var copiedBytes atomic.Int64
	outputSeen := func() bool { return copiedBytes.Load() > 0 }

	stdout := make(chan string, 10000)
	stderr := make(chan string, 10000)

	// If the driver can write synchronously, emit writes each log entry directly
	// from the goroutine that reads the container's stdio. Otherwise it hands the
	// entry to the driver's Process method over a buffered channel, which keeps a
	// slow (e.g. network-backed) driver from blocking the container.
	//
	// The synchronous path matters because, when a container exits, containerd
	// closes the stdio FIFOs and then tears the logging process down almost
	// immediately. Handing the final chunk to another goroutine to write races
	// that teardown and can lose a trailing chunk that has no newline; writing it
	// inline does not. https://github.com/containerd/nerdctl/issues/5006
	syncDriver, isSync := driver.(SyncDriver)
	emit := func(stream, line string) {
		if isSync {
			if err := syncDriver.WriteLogEntry(stream, line); err != nil {
				log.G(ctx).WithError(err).Error("failed to write log entry")
			}
			return
		}
		if stream == streamStdout {
			stdout <- line
		} else {
			stderr <- line
		}
	}

	var wg sync.WaitGroup

	// processStream reads a container stdio FIFO directly and emits its output
	// split into newline-terminated lines. Complete lines are emitted as they are
	// read; a trailing fragment without a newline is buffered until more output