opensandbox-group--opensandbox/components/execd/pkg/runtime/command_common.go

// Copyright 2025 Alibaba Group Holding Ltd.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package runtime

import (
	"bufio"
	"bytes"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sync"
	"time"
)

// tailStdPipe streams appended log data until the process finishes.
func (c *Controller) tailStdPipe(file string, onExecute func(text string), done <-chan struct{}) {
	lastPos := int64(0)
	ticker := time.NewTicker(100 * time.Millisecond)
	defer ticker.Stop()

	mutex := &sync.Mutex{}
	var lastWasCR bool
	for {
		select {
		case <-done:
			c.readFromPos(mutex, file, lastPos, onExecute, true, &lastWasCR)
			return
		case <-ticker.C:
			newPos := c.readFromPos(mutex, file, lastPos, onExecute, false, &lastWasCR)
			lastPos = newPos
		}
	}
}

// getCommandKernel retrieves a command execution context.
func (c *Controller) getCommandKernel(sessionID string) *commandKernel {
	if v, ok := c.commandClientMap.Load(sessionID); ok {
		if kernel, ok := v.(*commandKernel); ok {
			return kernel
		}
	}
	return nil
}

// storeCommandKernel registers a command execution context.
func (c *Controller) storeCommandKernel(sessionID string, kernel *commandKernel) {
	c.commandClientMap.Store(sessionID, kernel)
}

// stdLogDescriptor creates temporary files for capturing command output.
// It ensures the temp directory exists before opening files, so that commands
// continue to work even after the /tmp directory has been removed and recreated.
func (c *Controller) stdLogDescriptor(session string) (io.WriteCloser, io.WriteCloser, error) {
	logDir := os.TempDir()
	if err := os.MkdirAll(logDir, 0o755); err != nil {
		return nil, nil, fmt.Errorf("failed to create temp dir %s: %w", logDir, err)
	}

	stdout, err := os.OpenFile(c.stdoutFileName(session), os.O_RDWR|os.O_CREATE|os.O_TRUNC, os.ModePerm)
	if err != nil {
		return nil, nil, err
	}
	stderr, err := os.OpenFile(c.stderrFileName(session), os.O_RDWR|os.O_CREATE|os.O_TRUNC, os.ModePerm)
	if err != nil {
		stdout.Close()
		return nil, nil, err
	}

	return stdout, stderr, nil
}

func (c *Controller) combinedOutputDescriptor(session string) (io.WriteCloser, error) {
	logDir := os.TempDir()
	if err := os.MkdirAll(logDir, 0o755); err != nil {
		return nil, fmt.Errorf("failed to create temp dir %s: %w", logDir, err)
	}
	return os.OpenFile(c.combinedOutputFileName(session), os.O_RDWR|os.O_CREATE|os.O_TRUNC, os.ModePerm)
}

// stdoutFileName constructs the stdout log path.
func (c *Controller) stdoutFileName(session string) string {
	return filepath.Join(os.TempDir(), session+".stdout")
}

// stderrFileName constructs the stderr log path.
func (c *Controller) stderrFileName(session string) string {
	return filepath.Join(os.TempDir(), session+".stderr")
}

func (c *Controller) combinedOutputFileName(session string) string {
	return filepath.Join(os.TempDir(), session+".output")
}

// readFromPos streams new content from a file starting at startPos.
// lastWasCR persists CRLF detection across calls so a \r\n pair split between
// two polls does not surface a spurious blank line for the trailing \n.
func (c *Controller) readFromPos(mutex *sync.Mutex, filepath string, startPos int64, onExecute func(string), flushIncomplete bool, lastWasCR *bool) int64 {
	if !mutex.TryLock() {
		return -1
	}
	defer mutex.Unlock()

	file, err := os.Open(filepath)
	if err != nil {
		return startPos
	}
	defer file.Close()

	_, _ = file.Seek(startPos, 0) //nolint:errcheck

	reader := bufio.NewReader(file)
	var buffer bytes.Buffer
	var currentPos int64 = startPos
	cr := false
	if lastWasCR != nil {
		cr = *lastWasCR
	}
	defer func() {
		if lastWasCR != nil {
			*lastWasCR = cr
		}
	}()

	for {
		b, err := reader.ReadByte()
		if err != nil {
			if err == io.EOF {
				// If buffer has content but no newline, flush if needed, otherwise wait for next read
				if flushIncomplete && buffer.Len() > 0 {
					onExecute(buffer.String())
					buffer.Reset()
				}
			}
			break
		}
		currentPos++

		// Check if it's a line terminator (\n or \r)
		if b == '\n' || b == '\r' {
			switch {
			case buffer.Len() > 0:
				// Flush the line content without the terminator
				onExecute(buffer.String())
				buffer.Reset()
			case b == '\n' && cr:
				// Second half of a \r\n pair; already emitted on \r
			default:
				// Standalone blank line; surface it so callers see the gap
				onExecute("\n")
			}
			cr = (b == '\r')
			continue
		}

		cr = false
		buffer.WriteByte(b)
	}

	endPos, _ := file.Seek(0, 1)
	// If the last read position doesn't end with a newline, return buffer start position and wait for next flush
	if !flushIncomplete && buffer.Len() > 0 {
		return currentPos - int64(buffer.Len())
	}
	return endPos
}