<commit_before>package gateway

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"sync"

	"github.com/moby/buildkit/cache"
	"github.com/moby/buildkit/executor"
	"github.com/moby/buildkit/frontend/gateway/client"
	"github.com/moby/buildkit/session"
	"github.com/moby/buildkit/solver"
	"github.com/moby/buildkit/solver/llbsolver/mounts"
	opspb "github.com/moby/buildkit/solver/pb"
	"github.com/moby/buildkit/util/stack"
	utilsystem "github.com/moby/buildkit/util/system"
	"github.com/moby/buildkit/worker"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
	"golang.org/x/sync/errgroup"
)

type NewContainerRequest struct {
	ContainerID string
	NetMode     opspb.NetMode
	Mounts      []Mount
}

// Mount used for the gateway.Container is nearly identical to the client.Mount
// except is has a RefProxy instead of Ref to allow for a common abstraction
// between gateway clients.
type Mount struct {
	Dest      string
	Selector  string
	Readonly  bool
	MountType opspb.MountType
	RefProxy  solver.ResultProxy
	CacheOpt  *opspb.CacheOpt
	SecretOpt *opspb.SecretOpt
	SSHOpt    *opspb.SSHOpt
}

func toProtoMount(m Mount) *opspb.Mount {
	return &opspb.Mount{
		Selector:  m.Selector,
		Dest:      m.Dest,
		Readonly:  m.Readonly,
		MountType: m.MountType,
		CacheOpt:  m.CacheOpt,
		SecretOpt: m.SecretOpt,
		SSHOpt:    m.SSHOpt,
	}
}

func NewContainer(ctx context.Context, e executor.Executor, sm *session.Manager, g session.Group, req NewContainerRequest) (client.Container, error) {
	ctx, cancel := context.WithCancel(ctx)
	eg, ctx := errgroup.WithContext(ctx)
	ctr := &gatewayContainer{
		id:       req.ContainerID,
		netMode:  req.NetMode,
		executor: e,
		errGroup: eg,
		ctx:      ctx,
		cancel:   cancel,
	}

	makeMutable := func(worker worker.Worker, ref cache.ImmutableRef) (cache.MutableRef, error) {
		mRef, err := worker.CacheManager().New(ctx, ref)
		if err != nil {
			return nil, stack.Enable(err)
		}
		ctr.cleanup = append(ctr.cleanup, func() error {
			return stack.Enable(mRef.Release(context.TODO()))
		})
		return mRef, nil
	}

	var mm *mounts.MountManager
	mnts := req.Mounts

	for i, m := range mnts {
		if m.Dest == opspb.RootMount && m.RefProxy != nil {
			res, err := m.RefProxy.Result(ctx)
			if err != nil {
				return nil, stack.Enable(err)
			}
			workerRef, ok := res.Sys().(*worker.WorkerRef)
			if !ok {
				return nil, errors.Errorf("invalid reference for exec %T", res.Sys())
			}

			name := fmt.Sprintf("container %s", req.ContainerID)
			mm = mounts.NewMountManager(name, workerRef.Worker.CacheManager(), sm, workerRef.Worker.MetadataStore())

			ctr.rootFS = workerRef.ImmutableRef
			if !m.Readonly {
				ctr.rootFS, err = makeMutable(workerRef.Worker, workerRef.ImmutableRef)
				if err != nil {
					return nil, stack.Enable(err)
				}
			}

			// delete root mount from list, handled here
			mnts = append(mnts[:i], mnts[i+1:]...)
			break
		}
	}

	if ctr.rootFS == nil {
		return nil, errors.Errorf("root mount required")
	}

	for _, m := range mnts {
		var ref cache.ImmutableRef
		var mountable cache.Mountable
		if m.RefProxy != nil {
			res, err := m.RefProxy.Result(ctx)
			if err != nil {
				return nil, stack.Enable(err)
			}
			workerRef, ok := res.Sys().(*worker.WorkerRef)
			if !ok {
				return nil, errors.Errorf("invalid reference for exec %T", res.Sys())
			}
			ref = workerRef.ImmutableRef
			mountable = ref

			if !m.Readonly {
				mountable, err = makeMutable(workerRef.Worker, ref)
				if err != nil {
					return nil, stack.Enable(err)
				}
			}
		}
		switch m.MountType {
		case opspb.MountType_BIND:
			// nothing to do here
		case opspb.MountType_CACHE:
			mRef, err := mm.MountableCache(ctx, toProtoMount(m), ref)
			if err != nil {
				return nil, err
			}
			mountable = mRef
			ctr.cleanup = append(ctr.cleanup, func() error {
				return stack.Enable(mRef.Release(context.TODO()))
			})
		case opspb.MountType_TMPFS:
			mountable = mm.MountableTmpFS()
		case opspb.MountType_SECRET:
			var err error
			mountable, err = mm.MountableSecret(ctx, toProtoMount(m), g)
			if err != nil {
				return nil, err
			}
			if mountable == nil {
				continue
			}
		case opspb.MountType_SSH:
			var err error
			mountable, err = mm.MountableSSH(ctx, toProtoMount(m), g)
			if err != nil {
				return nil, err
			}
			if mountable == nil {
				continue
			}
		default:
			return nil, errors.Errorf("mount type %s not implemented", m.MountType)
		}

		// validate that there is a mount
		if mountable == nil {
			return nil, errors.Errorf("mount %s has no input", m.Dest)
		}

		execMount := executor.Mount{
			Src:      mountable,
			Selector: m.Selector,
			Dest:     m.Dest,
			Readonly: m.Readonly,
		}

		ctr.mounts = append(ctr.mounts, execMount)
	}

	// sort mounts so parents are mounted first
	sort.Slice(ctr.mounts, func(i, j int) bool {
		return ctr.mounts[i].Dest < ctr.mounts[j].Dest
	})

	return ctr, nil
}

type gatewayContainer struct {
	id           string
	netMode      opspb.NetMode
	securityMode opspb.SecurityMode
	rootFS       cache.Mountable
	mounts       []executor.Mount
	executor     executor.Executor
	started      bool
	errGroup     *errgroup.Group
	mu           sync.Mutex
	cleanup      []func() error
	ctx          context.Context
	cancel       func()
}

func (gwCtr *gatewayContainer) Start(ctx context.Context, req client.StartRequest) (client.ContainerProcess, error) {
	resize := make(chan executor.WinSize)
	procInfo := executor.ProcessInfo{
		Meta: executor.Meta{
			Args:         req.Args,
			Env:          req.Env,
			User:         req.User,
			Cwd:          req.Cwd,
			Tty:          req.Tty,
			NetMode:      gwCtr.netMode,
			SecurityMode: req.SecurityMode,
		},
		Stdin:  req.Stdin,
		Stdout: req.Stdout,
		Stderr: req.Stderr,
		Resize: resize,
	}<commit_msg>fix linting error<commit_after>package gateway

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"sync"

	"github.com/moby/buildkit/cache"
	"github.com/moby/buildkit/executor"
	"github.com/moby/buildkit/frontend/gateway/client"
	"github.com/moby/buildkit/session"
	"github.com/moby/buildkit/solver"
	"github.com/moby/buildkit/solver/llbsolver/mounts"
	opspb "github.com/moby/buildkit/solver/pb"
	"github.com/moby/buildkit/util/stack"
	utilsystem "github.com/moby/buildkit/util/system"
	"github.com/moby/buildkit/worker"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
	"golang.org/x/sync/errgroup"
)

type NewContainerRequest struct {
	ContainerID string
	NetMode     opspb.NetMode
	Mounts      []Mount
}

// Mount used for the gateway.Container is nearly identical to the client.Mount
// except is has a RefProxy instead of Ref to allow for a common abstraction
// between gateway clients.
type Mount struct {
	Dest      string
	Selector  string
	Readonly  bool
	MountType opspb.MountType
	RefProxy  solver.ResultProxy
	CacheOpt  *opspb.CacheOpt
	SecretOpt *opspb.SecretOpt
	SSHOpt    *opspb.SSHOpt
}

func toProtoMount(m Mount) *opspb.Mount {
	return &opspb.Mount{
		Selector:  m.Selector,
		Dest:      m.Dest,
		Readonly:  m.Readonly,
		MountType: m.MountType,
		CacheOpt:  m.CacheOpt,
		SecretOpt: m.SecretOpt,
		SSHOpt:    m.SSHOpt,
	}
}

func NewContainer(ctx context.Context, e executor.Executor, sm *session.Manager, g session.Group, req NewContainerRequest) (client.Container, error) {
	ctx, cancel := context.WithCancel(ctx)
	eg, ctx := errgroup.WithContext(ctx)
	ctr := &gatewayContainer{
		id:       req.ContainerID,
		netMode:  req.NetMode,
		executor: e,
		errGroup: eg,
		ctx:      ctx,
		cancel:   cancel,
	}

	makeMutable := func(worker worker.Worker, ref cache.ImmutableRef) (cache.MutableRef, error) {
		mRef, err := worker.CacheManager().New(ctx, ref)
		if err != nil {
			return nil, stack.Enable(err)
		}
		ctr.cleanup = append(ctr.cleanup, func() error {
			return stack.Enable(mRef.Release(context.TODO()))
		})
		return mRef, nil
	}

	var mm *mounts.MountManager
	mnts := req.Mounts

	for i, m := range mnts {
		if m.Dest == opspb.RootMount && m.RefProxy != nil {
			res, err := m.RefProxy.Result(ctx)
			if err != nil {
				return nil, stack.Enable(err)
			}
			workerRef, ok := res.Sys().(*worker.WorkerRef)
			if !ok {
				return nil, errors.Errorf("invalid reference for exec %T", res.Sys())
			}

			name := fmt.Sprintf("container %s", req.ContainerID)
			mm = mounts.NewMountManager(name, workerRef.Worker.CacheManager(), sm, workerRef.Worker.MetadataStore())

			ctr.rootFS = workerRef.ImmutableRef
			if !m.Readonly {
				ctr.rootFS, err = makeMutable(workerRef.Worker, workerRef.ImmutableRef)
				if err != nil {
					return nil, stack.Enable(err)
				}
			}

			// delete root mount from list, handled here
			mnts = append(mnts[:i], mnts[i+1:]...)
			break
		}
	}

	if ctr.rootFS == nil {
		return nil, errors.Errorf("root mount required")
	}

	for _, m := range mnts {
		var ref cache.ImmutableRef
		var mountable cache.Mountable
		if m.RefProxy != nil {
			res, err := m.RefProxy.Result(ctx)
			if err != nil {
				return nil, stack.Enable(err)
			}
			workerRef, ok := res.Sys().(*worker.WorkerRef)
			if !ok {
				return nil, errors.Errorf("invalid reference for exec %T", res.Sys())
			}
			ref = workerRef.ImmutableRef
			mountable = ref

			if !m.Readonly {
				mountable, err = makeMutable(workerRef.Worker, ref)
				if err != nil {
					return nil, stack.Enable(err)
				}
			}
		}
		switch m.MountType {
		case opspb.MountType_BIND:
			// nothing to do here
		case opspb.MountType_CACHE:
			mRef, err := mm.MountableCache(ctx, toProtoMount(m), ref)
			if err != nil {
				return nil, err
			}
			mountable = mRef
			ctr.cleanup = append(ctr.cleanup, func() error {
				return stack.Enable(mRef.Release(context.TODO()))
			})
		case opspb.MountType_TMPFS:
			mountable = mm.MountableTmpFS()
		case opspb.MountType_SECRET:
			var err error
			mountable, err = mm.MountableSecret(ctx, toProtoMount(m), g)
			if err != nil {
				return nil, err
			}
			if mountable == nil {
				continue
			}
		case opspb.MountType_SSH:
			var err error
			mountable, err = mm.MountableSSH(ctx, toProtoMount(m), g)
			if err != nil {
				return nil, err
			}
			if mountable == nil {
				continue
			}
		default:
			return nil, errors.Errorf("mount type %s not implemented", m.MountType)
		}

		// validate that there is a mount
		if mountable == nil {
			return nil, errors.Errorf("mount %s has no input", m.Dest)
		}

		execMount := executor.Mount{
			Src:      mountable,
			Selector: m.Selector,
			Dest:     m.Dest,
			Readonly: m.Readonly,
		}

		ctr.mounts = append(ctr.mounts, execMount)
	}

	// sort mounts so parents are mounted first
	sort.Slice(ctr.mounts, func(i, j int) bool {
		return ctr.mounts[i].Dest < ctr.mounts[j].Dest
	})

	return ctr, nil
}

type gatewayContainer struct {
	id       string
	netMode  opspb.NetMode
	rootFS   cache.Mountable
	mounts   []executor.Mount
	executor executor.Executor
	started  bool
	errGroup *errgroup.Group
	mu       sync.Mutex
	cleanup  []func() error
	ctx      context.Context
	cancel   func()
}

func (gwCtr *gatewayContainer) Start(ctx context.Context, req client.StartRequest) (client.ContainerProcess, error) {
	resize := make(chan executor.WinSize)
	procInfo := executor.ProcessInfo{
		Meta: executor.Meta{
			Args:         req.Args,
			Env:          req.Env,
			User:         req.User,
			Cwd:          req.Cwd,
			Tty:          req.Tty,
			NetMode:      gwCtr.netMode,
			SecurityMode: req.SecurityMode,
		},
		Stdin:  req.Stdin,
		Stdout: req.Stdout,
		Stderr: req.Stderr,
		Resize: resize,
	}
{'fraction_non_alphanumeric': 0.12223678462210573, 'fraction_numerical': 0.00017474879860200962, 'mean_word_length': 3.086397715101749, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6174863', 'n_tokens_mistral': 4588, 'n_tokens_neox': 4104, 'n_words': 968}