process load balancing: broken but in progress

internal/pkg/process/chrome.go

@@ -0,0 +1,106 @@
+package process
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/mafredri/cdp/devtool"
+	"github.com/thecodingmachine/gotenberg/internal/pkg/xlog"
+)
+
+const ChromeKey Key = "chrome"
+
+type chromeProcess struct {
+	id   string
+	host string
+	port int
+}
+
+// NewChromeProcess returns a Google Chrome
+// headless process.
+func NewChromeProcess(id, host string, port int) Process {
+	return chromeProcess{
+		id:   id,
+		host: host,
+		port: port,
+	}
+}
+
+func (p chromeProcess) ID() string {
+	return p.id
+}
+
+func (p chromeProcess) Host() string {
+	return p.host
+}
+
+func (p chromeProcess) Port() int {
+	return p.port
+}
+
+func (p chromeProcess) binary() string {
+	return "google-chrome-stable"
+}
+
+func (p chromeProcess) args() []string {
+	return []string{
+		"--no-sandbox",
+		"--headless",
+		// see https://github.com/GoogleChrome/puppeteer/issues/2410.
+		"--font-render-hinting=medium",
+		fmt.Sprintf("--remote-debugging-port=%d", p.port),
+		"--disable-gpu",
+		"--disable-translate",
+		"--disable-extensions",
+		"--disable-background-networking",
+		"--safebrowsing-disable-auto-update",
+		"--disable-sync",
+		"--disable-default-apps",
+		"--hide-scrollbars",
+		"--metrics-recording-only",
+		"--mute-audio",
+		"--no-first-run",
+	}
+}
+
+func (p chromeProcess) warmupTime() time.Duration {
+	return 10 * time.Second
+}
+
+func (p chromeProcess) viabilityFunc() func(logger xlog.Logger) bool {
+	const op string = "process.chromeProcess.viabilityFunc"
+	return func(logger xlog.Logger) bool {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		endpoint := fmt.Sprintf("http://%s:%d" /*p.host*/, "localhost", p.port)
+		logger.DebugfOp(
+			op,
+			"checking '%s' viability via endpoint '%s/json/version'",
+			p.ID(),
+			endpoint,
+		)
+		v, err := devtool.New(endpoint).Version(ctx)
+		if err != nil {
+			logger.ErrorfOp(
+				op,
+				"'%s' is not viable as endpoint returned '%v'",
+				p.ID(),
+				err,
+			)
+			return false
+		}
+		logger.DebugfOp(
+			op,
+			"'%s' is viable as endpoint returned '%v'",
+			p.ID(),
+			v,
+		)
+		return true
+	}
+}
+
+// Compile-time checks to ensure type implements desired interfaces.
+var (
+	_ = Process(new(chromeProcess))
+)

internal/pkg/process/doc.go

@@ -0,0 +1 @@
+package process

internal/pkg/process/pm2.go

@@ -0,0 +1,324 @@
+package process
+
+import (
+	"encoding/json"
+	"fmt"
+	"os/exec"
+	"sync"
+	"time"
+
+	"github.com/thecodingmachine/gotenberg/internal/pkg/conf"
+	"github.com/thecodingmachine/gotenberg/internal/pkg/xerror"
+	"github.com/thecodingmachine/gotenberg/internal/pkg/xexec"
+	"github.com/thecodingmachine/gotenberg/internal/pkg/xlog"
+	"github.com/thecodingmachine/gotenberg/internal/pkg/xtime"
+)
+
+type jlistItem struct {
+	Name   string `json:"name"`
+	PM2Env struct {
+		Status      string `json:"status"`
+		RestartTime int64  `json:"restart_time"`
+	} `json:"pm2_env"`
+	Monit struct {
+		Memory int64   `json:"memory"`
+		CPU    float64 `json:"cpu"`
+	} `json:"monit"`
+}
+
+type jlist []jlistItem
+
+func (list jlist) toList() List {
+	var result List
+	for _, current := range list {
+		item := ListItem{
+			Name:    current.Name,
+			Status:  current.PM2Env.Status,
+			Restart: current.PM2Env.RestartTime,
+			Memory:  current.Monit.Memory, // TODO humanize?
+			CPU:     current.Monit.CPU,
+		}
+		result = append(result, item)
+	}
+	return result
+}
+
+func (list jlist) isOnline(p Process) bool {
+	const onlineStatus string = "online"
+	for _, item := range list {
+		if item.Name == p.ID() {
+			return item.PM2Env.Status == onlineStatus
+		}
+	}
+	return false
+}
+
+func (list jlist) memory(p Process) (int64, error) {
+	const op string = "process.jlist.memory"
+	for _, item := range list {
+		if item.Name == p.ID() {
+			return item.Monit.Memory, nil
+		}
+	}
+	return 0, xerror.New(
+		op,
+		fmt.Errorf("'%s' does not exist in the list of PM2 processes", p.ID()),
+	)
+}
+
+type command string
+
+const (
+	startCommand   command = "start"
+	restartCommand command = "restart"
+	stopCommand    command = "stop"
+	logsCommand    command = "logs"
+	jlistCommand   command = "jlist"
+)
+
+const maximumRestartAttempts uint = 3
+
+type pm2Manager struct {
+	logger   xlog.Logger
+	config   conf.Config
+	pool     map[Key][]Process
+	list     *jlist
+	listLock *sync.Mutex
+}
+
+// NewPM2Manager returns a PM2 manager.
+func NewPM2Manager(logger xlog.Logger, config conf.Config) Manager {
+	const op string = "process.NewPM2Manager"
+	m := &pm2Manager{
+		logger:   logger,
+		config:   config,
+		pool:     make(map[Key][]Process),
+		listLock: &sync.Mutex{},
+	}
+	if !config.DisableGoogleChrome() {
+		processes := make([]Process, 2)
+		availablePort := 9222
+		// TODO from config
+		for i := 0; i < 2; i++ {
+			proc := chromeProcess{
+				host: "127.0.0.1",
+				port: availablePort,
+			}
+			proc.id = fmt.Sprintf("%s-%d", proc.binary(), proc.port)
+			processes[i] = proc
+			logger.DebugfOp(op, "added new process %v", proc)
+			availablePort++
+		}
+		m.pool[ChromeKey] = processes
+	}
+	if !config.DisableUnoconv() {
+		processes := make([]Process, 2)
+		availablePort := 2002
+		// TODO from config
+		for i := 0; i < 2; i++ {
+			proc := sofficeProcess{
+				host: "127.0.0.1",
+				port: availablePort,
+			}
+			proc.id = fmt.Sprintf("%s-%d", proc.binary(), proc.port)
+			processes[i] = proc
+			logger.DebugfOp(op, "added new process %v", proc)
+			availablePort++
+		}
+		m.pool[SofficeKey] = processes
+	}
+	// update the manager processes list
+	// only if there are processes.
+	if !config.DisableGoogleChrome() || !config.DisableUnoconv() {
+		go m.jlistTimer()
+	}
+	return m
+}
+
+func (m *pm2Manager) jlistTimer() {
+	const op string = "process.pm2Manager.jlistTimer"
+	duration := xtime.Duration(10)
+	resolver := func() error {
+		m.listLock.Lock()
+		defer m.listLock.Unlock()
+		out, err := exec.
+			Command("pm2", string(jlistCommand)).
+			Output()
+		if err != nil {
+			return err
+		}
+		data := &jlist{}
+		if err := json.Unmarshal(out, data); err != nil {
+			return err
+		}
+		m.list = data
+		return nil
+	}
+	// update every x seconds the
+	// list from the manager.
+	for range time.Tick(duration) {
+		if err := resolver(); err != nil {
+			m.logger.ErrorOp(op, xerror.New(op, err))
+		}
+	}
+}
+
+func (m *pm2Manager) Start() error {
+	const op string = "process.pm2Manager.Start"
+	for _, processes := range m.pool {
+		for _, proc := range processes {
+			if err := m.start(proc); err != nil {
+				return xerror.New(op, err)
+			}
+		}
+	}
+	return nil
+}
+
+func (m *pm2Manager) start(p Process) error {
+	const op string = "process.pm2Manager.start"
+	resolver := func() error {
+		// first, we try to start the process.
+		if err := m.run(startCommand, p); err != nil {
+			return err
+		}
+		// we wait the process to be ready.
+		m.warmup(p)
+		// if the process failed to start correctly,
+		// we have to restart it.
+		if !m.IsViable(p) && maximumRestartAttempts > 0 {
+			return m.Restart(p)
+		}
+		// the process is viable, let's log its
+		// output.
+		return m.logs(p)
+	}
+	if err := resolver(); err != nil {
+		return xerror.New(op, err)
+	}
+	return nil
+}
+
+func (m *pm2Manager) List() List {
+	m.listLock.Lock()
+	defer m.listLock.Unlock()
+	return m.list.toList()
+}
+
+func (m *pm2Manager) All() []Process {
+	var result []Process
+	for _, processes := range m.pool {
+		result = append(result, processes...)
+	}
+	return result
+}
+
+func (m *pm2Manager) Processes(key Key) []Process {
+	return m.pool[key]
+}
+
+func (m *pm2Manager) Restart(p Process) error {
+	const op string = "process.pm2Manager.Restart"
+	resolver := func() error {
+		var attempts uint
+		for attempts < maximumRestartAttempts {
+			// we restart the process.
+			if err := m.run(restartCommand, p); err != nil {
+				return err
+			}
+			// we wait the process to be ready.
+			m.warmup(p)
+			attempts++
+			// if the process is viable, we
+			// leave.
+			if m.IsViable(p) {
+				return m.logs(p)
+			}
+		}
+		return fmt.Errorf("failed to start '%s'", p.ID())
+	}
+	if err := resolver(); err != nil {
+		return xerror.New(op, err)
+	}
+	return nil
+}
+
+func (m *pm2Manager) Stop(p Process) error {
+	const op string = "process.pm2Manager.Stop"
+	if err := m.run(stopCommand, p); err != nil {
+		return xerror.New(op, err)
+	}
+	return nil
+}
+
+func (m *pm2Manager) IsViable(p Process) bool {
+	if !p.viabilityFunc()(m.logger) {
+		return false
+	}
+	m.listLock.Lock()
+	defer m.listLock.Unlock()
+	return m.list.isOnline(p)
+}
+
+func (m *pm2Manager) Memory(p Process) (int64, error) {
+	const op string = "process.pm2Manager.Memory"
+	m.listLock.Lock()
+	defer m.listLock.Unlock()
+	memory, err := m.list.memory(p)
+	if err != nil {
+		return 0, xerror.New(op, err)
+	}
+	return memory, nil
+}
+
+func (m *pm2Manager) warmup(p Process) {
+	const op string = "process.pm2Manager.warmup"
+	warmupTime := p.warmupTime()
+	m.logger.DebugfOp(
+		op,
+		"waiting '%v' for allowing '%s' to warmup",
+		warmupTime,
+		p.ID(),
+	)
+	time.Sleep(warmupTime)
+}
+
+func (m *pm2Manager) logs(p Process) error {
+	const op string = "process.pm2Manager.logs"
+	if m.config.LogLevel() == xlog.DebugLevel {
+		if err := m.run(logsCommand, p); err != nil {
+			return xerror.New(op, err)
+		}
+	}
+	return nil
+}
+
+func (m *pm2Manager) run(pm2Cmd command, p Process) error {
+	const op string = "process.pm2Manager.run"
+	resolver := func() error {
+		args := []string{
+			string(pm2Cmd),
+			p.binary(),
+		}
+		if pm2Cmd == startCommand {
+			args = append(args, fmt.Sprintf("--name=%s", p.ID()))
+			args = append(args, "--interpreter=none", "--")
+			args = append(args, p.args()...)
+		}
+		cmd, err := xexec.Command(m.logger, "pm2", args...)
+		if err != nil {
+			return err
+		}
+		xexec.LogBeforeExecute(m.logger, cmd)
+		return cmd.Start()
+	}
+	if err := resolver(); err != nil {
+		return xerror.New(op, err)
+	}
+	return nil
+}
+
+// Compile-time checks to ensure type implements desired interfaces.
+var (
+	_ = Manager(new(pm2Manager))
+)

internal/pkg/process/process.go

@@ -0,0 +1,40 @@
+package process
+
+import (
+	"time"
+
+	"github.com/thecodingmachine/gotenberg/internal/pkg/xlog"
+)
+
+type Process interface {
+	ID() string
+	Host() string
+	Port() int
+	binary() string
+	args() []string
+	warmupTime() time.Duration
+	viabilityFunc() func(logger xlog.Logger) bool
+}
+
+type ListItem struct {
+	Name    string  `json:"name"`
+	Status  string  `json:"status"`
+	Restart int64   `json:"restart"`
+	Memory  int64   `json:"memory"`
+	CPU     float64 `json:"cpu"`
+}
+
+type List []ListItem
+
+type Key string
+
+type Manager interface {
+	Start() error
+	List() List
+	All() []Process
+	Processes(key Key) []Process
+	Restart(proc Process) error
+	Stop(proc Process) error
+	IsViable(proc Process) bool
+	Memory(proc Process) (int64, error)
+}

internal/pkg/process/soffice.go

@@ -0,0 +1,74 @@
+package process
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/thecodingmachine/gotenberg/internal/pkg/xlog"
+)
+
+const SofficeKey Key = "soffice"
+
+type sofficeProcess struct {
+	id   string
+	host string
+	port int
+}
+
+// NewSofficeProcess returns a LibreOffice
+// headless process.
+func NewSofficeProcess(id, host string, port int) Process {
+	return sofficeProcess{
+		id:   id,
+		host: host,
+		port: port,
+	}
+}
+
+func (p sofficeProcess) ID() string {
+	return p.id
+}
+
+func (p sofficeProcess) Host() string {
+	return p.host
+}
+
+func (p sofficeProcess) Port() int {
+	return p.port
+}
+
+func (p sofficeProcess) binary() string {
+	return "soffice"
+}
+
+func (p sofficeProcess) args() []string {
+	return []string{
+		// see https://ask.libreoffice.org/en/question/42975/how-can-i-run-multiple-instances-of-sofficebin-at-a-time/.
+		fmt.Sprintf("-env:UserInstallation=file:///tmp/%d", p.port),
+		"--headless",
+		"--invisible",
+		"--nocrashreport",
+		"--nodefault",
+		"--nofirststartwizard",
+		"--nologo",
+		"--norestore",
+		fmt.Sprintf("--accept=socket,host=%s,port=%d,tcpNoDelay=1;urp;StarOffice.ComponentContext", p.host, p.port),
+	}
+}
+
+func (p sofficeProcess) warmupTime() time.Duration {
+	return 3 * time.Second
+}
+
+func (p sofficeProcess) viabilityFunc() func(logger xlog.Logger) bool {
+	const op string = "process.sofficeProcess.viabilityFunc"
+	return func(logger xlog.Logger) bool {
+		// TODO find a way to check.
+		return true
+	}
+}
+
+// Compile-time checks to ensure type implements desired interfaces.
+var (
+	_ = Process(new(sofficeProcess))
+)