cep/executor/container_oneshot.go

package executor

import (
	"context"
	"errors"
	"fmt"
	"time"

	dockerpkg "git.buran.team/main/cep/docker"
	schemepkg "git.buran.team/main/cep/scheme"
	fw "git.buran.team/main/fairwind"
)

var ErrContainerContextCancelled = errors.New("container context cancelled")
var ErrContainerContextTimeout = errors.New("container context timeout")
var ErrContainerResultInvalid = errors.New("container result invalid")

// TODO: add start/stop conditions
type OneshotContainer struct {
	global    *Global
	ticket    *Ticket
	id        string
	container *dockerpkg.Container
	timeout   int
	result    *Result
}

func NewOneshotContainer(
	global *Global,
	ticket *Ticket,
	network *Network,
	containerIndex int,
	containerScheme schemepkg.Container,
) (*OneshotContainer, error) {
	varialbles := []dockerpkg.Variable{}
	for _, variable := range containerScheme.Variables {
		varialbles = append(
			varialbles,
			dockerpkg.Variable{
				Key:   variable.Key,
				Value: variable.Value,
			},
		)
	}

	hosts := []dockerpkg.ContainerSettingsHost{}
	for _, host := range network.Hosts() {
		hosts = append(
			hosts,
			dockerpkg.ContainerSettingsHost{
				Name: host.Name,
				IP:   host.IP,
			},
		)
	}

	container := dockerpkg.NewContainer(
		global.Ctx,
		global.Log,
		global.Docker,
		network.Network(),
		dockerpkg.NewImage(
			global.Ctx,
			global.Log,
			global.Docker,
			global.Registry,
			containerScheme.Image.Tag,
			containerScheme.Image.Version,
			containerScheme.Image.Pull,
		),
		dockerpkg.ContainerSettings{
			Name: containerName(
				ticket.Index,
				containerIndex,
			),
			Command:   containerScheme.Command,
			Variables: varialbles,
			Hosts:     hosts,
			Binds:     containerScheme.Binds,
			Permissions: dockerpkg.ContainerSettingsPermissions{
				Privileged:   containerScheme.Permissions.Privileged,
				Capabilities: containerScheme.Permissions.Capabilities,
			},
			Network: dockerpkg.ContainerSettingsNetwork{
				IP: containerIP(
					ticket.Index,
					containerIndex,
				),
			},
			Resources: dockerpkg.ContainerResources{},
		},
	)

	result, err := NewResult(containerScheme.KindOneshot.Result)
	if err != nil {
		return nil, fmt.Errorf("can't create oneshot container: %w", err)
	}

	return &OneshotContainer{
		global:    global,
		ticket:    ticket,
		id:        containerScheme.ID,
		container: container,
		timeout:   containerScheme.KindOneshot.Timeout,
		result:    result,
	}, nil
}

func (this *OneshotContainer) ID() string {
	return this.id
}

func (this *OneshotContainer) Start() *ContainerResult {
	var err error
	this.global.Log.Debug("starting container", fw.LogValue("uuid", this.ticket.UUID), fw.LogValue("id", this.id))
	defer func() {
		if err != nil {
			this.global.Log.Debug("container start failed", fw.LogValue("uuid", this.ticket.UUID), fw.LogValue("id", this.id), fw.LogError(err))
		} else {
			this.global.Log.Debug("container started", fw.LogValue("uuid", this.ticket.UUID), fw.LogValue("id", this.id))
		}
	}()

	// Start
	err = this.container.Start()
	if err != nil {
		return NewContainerResultError(
			fmt.Errorf("can't start container: %w", err),
		)
	}

	// Wait
	ctx, cancel := context.WithTimeout(
		context.Background(),
		time.Duration(this.timeout)*time.Millisecond,
	)
	defer cancel()

	var status dockerpkg.ExecutionResult
	for {
		select {
		case <-this.global.Ctx.Done():
			err = ErrContainerContextCancelled
			return NewContainerResultError(err)

		case <-ctx.Done():
			err = ErrContainerContextTimeout
			return NewContainerResultError(err)

		case <-time.After(100 * time.Millisecond): // NOTE: basic 100-millisecond startup delay
		}

		this.global.Log.Debug("check container status", fw.LogValue("uuid", this.ticket.UUID), fw.LogValue("id", this.id))
		status = this.container.Status(true)
		if status.Error != nil {
			this.global.Log.Information("container startup error", fw.LogError(status.Error))
			continue
		}

		this.global.Log.Debug("check container result", fw.LogValue("uuid", this.ticket.UUID), fw.LogValue("id", this.id))
		if !this.result.Check(
			status.Code,
			status.Stdout,
			status.Stderr,
		) {
			this.global.Log.Information("container startup check failed", fw.LogValue("uuid", this.ticket.UUID), fw.LogValue("id", this.id))
			err = ErrContainerResultInvalid
			return NewContainerResultError(err)
		}

		this.global.Log.Debug("container checks succeed", fw.LogValue("uuid", this.ticket.UUID), fw.LogValue("id", this.id))
		break
	}

	// Done
	return NewContainerResultSuccess(
		map[string]bool{},
		status.Code,
		status.Stdout,
		status.Stderr,
	)
}

func (this *OneshotContainer) Stop() *ContainerResult {
	var err error
	this.global.Log.Debug("stopping container", fw.LogValue("uuid", this.ticket.UUID), fw.LogValue("id", this.id))
	defer func() {
		if err != nil {
			this.global.Log.Debug("container stop failed", fw.LogValue("uuid", this.ticket.UUID), fw.LogValue("id", this.id), fw.LogError(err))
		} else {
			this.global.Log.Debug("container stopped", fw.LogValue("uuid", this.ticket.UUID), fw.LogValue("id", this.id))
		}
	}()

	err = this.container.Stop()
	if err != nil {
		return NewContainerResultError(
			fmt.Errorf("can't stop container: %w", err),
		)
	}

	return NewContainerResultSuccess(
		map[string]bool{},
		0,
		[]byte{},
		[]byte{},
	)
}

func (this *OneshotContainer) Execute(timeout int, command string) ExecutionResult {
	result := this.container.Execute(timeout, command)
	return ExecutionResult{
		Error:  result.Error,
		Code:   result.Code,
		Stdout: result.Stdout,
		Stderr: result.Stderr,
	}
}

func (this *OneshotContainer) Read(timeout int, path string) ([]byte, error) {
	return this.container.Read(timeout, path)
}