beehive/util.go

package beehive

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"log"
	"os"
	"os/exec"
	"path"
	"strings"
	"sync"
	"time"

	"sigs.k8s.io/yaml"
)

// Serialize stores data in YAML format at the specified path.
func Serialize(object interface{}, p string) error {
	if p == "" {
		return errors.New("failed to serialize: no path specified")
	}

	out, err := yaml.Marshal(object)
	if err != nil {
		return fmt.Errorf("failed to marshal configuration: %s", err)
	}

	os.MkdirAll(path.Dir(p), 0)

	err = os.WriteFile(p, out, 0600)
	if err != nil {
		return fmt.Errorf("failed to write to %s: %s", p, err)
	}
	return nil
}

// Deserialize loads data from the specified path. If a file does not exist at
// the specified path, no error is returned.
func Deserialize(object interface{}, path string) error {
	if path == "" {
		return errors.New("failed to deserialize: no path specified")
	}

	_, err := os.Stat(path)
	if os.IsNotExist(err) {
		return nil
	}

	configData, err := os.ReadFile(path)
	if err != nil {
		return fmt.Errorf("failed to read file: %s", err)
	}

	err = yaml.Unmarshal(configData, object)
	if err != nil {
		return fmt.Errorf("failed to parse file: %s", err)
	}

	return nil
}

func runCommand(dir string, timeout time.Duration, command string, args []string) (*exec.Cmd, *AsyncBuffer, *AsyncBuffer, error) {
	stdOut := &AsyncBuffer{}
	stdErr := &AsyncBuffer{}

	var cmd *exec.Cmd
	if timeout != 0 {
		ctx, _ := context.WithTimeout(context.Background(), 60*time.Second)
		cmd = exec.CommandContext(ctx, command, args...)
	} else {
		cmd = exec.Command(command, args...)
	}
	cmd.Dir = dir
	cmd.Stdout = stdOut
	cmd.Stderr = stdErr

	err := cmd.Start()
	if err != nil {
		if exitError, ok := err.(*exec.ExitError); ok {
			exitCode := exitError.ExitCode()
			return cmd, stdOut, stdErr, fmt.Errorf("failed to execute %s %+v: return status %d", command, args, exitCode)
		}
		return cmd, stdOut, stdErr, fmt.Errorf("failed to execute %s %+v: %v", command, args, err)
	}
	return cmd, stdOut, stdErr, nil
}

func runCommandAndWait(dir string, timeout time.Duration, command string, args []string) ([]byte, []byte, error) {
	log.Printf("Executing %s: %s %s", dir, command, strings.Join(args, " "))

	cmd, stdOut, stdErr, err := runCommand(dir, timeout, command, args)

	err = cmd.Wait()
	if err != nil {
		var errExtra string
		if stdErr.Len() > 0 {
			errExtra = "\n" + stdErr.String()
		}

		if exitError, ok := err.(*exec.ExitError); ok {
			exitCode := exitError.ExitCode()
			return stdOut.Bytes(), stdErr.Bytes(), fmt.Errorf("command terminated: %s %+v: return status %d%s%s", command, args, exitCode, stdOut.Bytes(), stdErr.Bytes())
		}

		return stdOut.Bytes(), stdErr.Bytes(), fmt.Errorf("failed to execute %s %+v: %v%s", command, args, err, errExtra)
	}

	return stdOut.Bytes(), stdErr.Bytes(), nil
}

func Docker(dir string, args []string) ([]byte, []byte, error) {
	return runCommandAndWait(dir, 1*time.Minute, "docker", args)
}

func DockerCompose(dir string, args []string) ([]byte, []byte, error) {
	return runCommandAndWait(dir, 1*time.Minute, "docker-compose", args)
}

func DockerEvents(container string) (*exec.Cmd, *AsyncBuffer, *AsyncBuffer, error) {
	return runCommand("/", 0, "docker", []string{"events", "--filter", "container=" + container, "--format", "H.net{{ .Status }}H.net"})
}

type AsyncBuffer struct {
	b bytes.Buffer
	l sync.Mutex
}

func (b *AsyncBuffer) Read(p []byte) (n int, err error) {
	b.l.Lock()
	defer b.l.Unlock()
	return b.b.Read(p)
}

func (b *AsyncBuffer) Write(p []byte) (n int, err error) {
	b.l.Lock()
	defer b.l.Unlock()
	return b.b.Write(p)
}

func (b *AsyncBuffer) String() string {
	b.l.Lock()
	defer b.l.Unlock()
	return b.b.String()
}

func (b *AsyncBuffer) Bytes() []byte {
	b.l.Lock()
	defer b.l.Unlock()
	return b.b.Bytes()
}

func (b *AsyncBuffer) Len() int {
	b.l.Lock()
	defer b.l.Unlock()
	return b.b.Len()
}