feat: start a unoconv listener by default, but allow stateless mode as before with the --unoconv-disable-listener flag. Also improve the shutdown process

This commit is contained in:
Julien Neuhart
2021-12-03 14:15:59 +01:00
parent c922cc042e
commit 2269c80b02
6 changed files with 575 additions and 95 deletions

View File

@@ -5,7 +5,6 @@ import (
"errors"
"fmt"
"os"
"os/signal"
"path/filepath"
"strings"
"sync"
@@ -202,19 +201,10 @@ func (gc *GarbageCollector) Stop(ctx context.Context) error {
}
// Block until the context is done so that other module may gracefully stop
// before we do a shutdown cleanup. We skip this step if we receive a
// SIGINT in the meantime.
// before we do a shutdown cleanup.
gc.logger.Debug("wait for the end of grace duration")
quit := make(chan os.Signal, 1)
signal.Notify(quit, os.Interrupt)
select {
case <-quit:
return nil
case <-ctx.Done():
break
}
<-ctx.Done()
gc.ticker.Stop()
gc.done <- true

View File

@@ -0,0 +1,31 @@
package unoconv
import (
"fmt"
"net"
"strconv"
"go.uber.org/zap"
)
func freePort(logger *zap.Logger) (int, error) {
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
return 0, fmt.Errorf("listen on the local network address: %w", err)
}
defer func() {
err := listener.Close()
if err != nil {
logger.Error(fmt.Sprintf("close listener: %s", err.Error()))
}
}()
addr := listener.Addr().String()
_, portStr, err := net.SplitHostPort(addr)
if err != nil {
return 0, fmt.Errorf("get free port from host: %w", err)
}
return strconv.Atoi(portStr)
}

View File

@@ -4,13 +4,12 @@ import (
"context"
"errors"
"fmt"
"net"
"os"
"strconv"
"strings"
"sync"
"github.com/gotenberg/gotenberg/v7/pkg/gotenberg"
flag "github.com/spf13/pflag"
"go.uber.org/zap"
)
@@ -24,7 +23,12 @@ var ErrMalformedPageRanges = errors.New("page ranges are malformed")
// Unoconv is a module which provides an API to interact with unoconv.
type Unoconv struct {
binPath string
binPath string
disableListener bool
listenerCmd gotenberg.Cmd
listenerPort int
logger *zap.Logger
}
// Options gathers available options when converting a document to PDF.
@@ -67,14 +71,23 @@ type Provider interface {
// Descriptor returns a Unoconv's module descriptor.
func (Unoconv) Descriptor() gotenberg.ModuleDescriptor {
return gotenberg.ModuleDescriptor{
ID: "unoconv",
ID: "unoconv",
FlagSet: func() *flag.FlagSet {
fs := flag.NewFlagSet("unoconv", flag.ExitOnError)
fs.Bool("unoconv-disable-listener", false, "Do not start a unoconv listener - save resources in detriment of performance")
return fs
}(),
New: func() gotenberg.Module { return new(Unoconv) },
}
}
// Provision sets the module properties. It returns an error if the environment
// variable UNOCONV_BIN_PATH is not set.
func (mod *Unoconv) Provision(_ *gotenberg.Context) error {
func (mod *Unoconv) Provision(ctx *gotenberg.Context) error {
flags := ctx.ParsedFlags()
mod.disableListener = flags.MustBool("unoconv-disable-listener")
binPath, ok := os.LookupEnv("UNOCONV_BIN_PATH")
if !ok {
return errors.New("UNOCONV_BIN_PATH environment variable is not set")
@@ -82,6 +95,18 @@ func (mod *Unoconv) Provision(_ *gotenberg.Context) error {
mod.binPath = binPath
loggerProvider, err := ctx.Module(new(gotenberg.LoggerProvider))
if err != nil {
return fmt.Errorf("get logger provider: %w", err)
}
logger, err := loggerProvider.(gotenberg.LoggerProvider).Logger(mod)
if err != nil {
return fmt.Errorf("get logger: %w", err)
}
mod.logger = logger
return nil
}
@@ -95,12 +120,92 @@ func (mod Unoconv) Validate() error {
return nil
}
func (mod *Unoconv) Start() error {
if mod.disableListener {
return nil
}
port, err := freePort(mod.logger)
if err != nil {
return fmt.Errorf("get free port: %w", err)
}
mod.listenerPort = port
args := []string{
"--listener",
"--user-profile",
// Just to make sure LibreOffice does not leak files in an unknown
// directory. The directory will be removed anyway by the garbage
// collector.
fmt.Sprintf("//%s", gotenberg.NewDirPath()),
"--port",
fmt.Sprintf("%d", mod.listenerPort),
}
checkedEntry := mod.logger.Check(zap.DebugLevel, "check for debug level before setting high verbosity")
if checkedEntry != nil {
args = append(args, "-vvv")
}
mod.listenerCmd = gotenberg.Command(mod.logger, mod.binPath, args...)
err = mod.listenerCmd.Start()
if err != nil {
return fmt.Errorf("start unoconv listener: %w", err)
}
listenerActiveInstancesCountMu.Lock()
listenerActiveInstancesCount += 1
listenerActiveInstancesCountMu.Unlock()
return nil
}
// StartupMessage returns a custom startup message.
func (mod Unoconv) StartupMessage() string {
if mod.disableListener {
return "listener disabled"
}
return fmt.Sprintf("listener started on port %d", mod.listenerPort)
}
// Stop stops the HTTP server.
func (mod *Unoconv) Stop(ctx context.Context) error {
if mod.disableListener {
return nil
}
_, ok := ctx.Deadline()
if !ok {
return errors.New("no context dead line")
}
// Block until the context is done so that other module may gracefully stop
// before we do a shutdown cleanup.
mod.logger.Debug("wait for the end of grace duration")
<-ctx.Done()
err := mod.listenerCmd.Kill()
if err != nil {
return fmt.Errorf("kill unoconv listener: %w", err)
}
listenerActiveInstancesCountMu.Lock()
listenerActiveInstancesCount -= 1
listenerActiveInstancesCountMu.Unlock()
return nil
}
// Metrics returns the metrics.
func (mod Unoconv) Metrics() ([]gotenberg.Metric, error) {
return []gotenberg.Metric{
{
Name: "unoconv_active_instances_count",
Description: "Current number of active LibreOffice instances.",
Description: "Current number of active unoconv instances.",
Read: func() float64 {
activeInstancesCountMu.RLock()
defer activeInstancesCountMu.RUnlock()
@@ -108,57 +213,70 @@ func (mod Unoconv) Metrics() ([]gotenberg.Metric, error) {
return activeInstancesCount
},
},
{
Name: "unoconv_listener_active_instances_count",
Description: "Current number of active unoconv listener instances.",
Read: func() float64 {
listenerActiveInstancesCountMu.RLock()
defer listenerActiveInstancesCountMu.RUnlock()
return listenerActiveInstancesCount
},
},
{
Name: "unoconv_listener_queue_length",
Description: "Current number of processes in the queue.",
Read: func() float64 {
listenerQueueLengthMu.RLock()
defer listenerQueueLengthMu.RUnlock()
return listenerQueueLength
},
},
}, nil
}
// Unoconv returns an API for interacting with unoconv.
func (mod Unoconv) Unoconv() (API, error) {
func (mod *Unoconv) Unoconv() (API, error) {
return mod, nil
}
// PDF converts a document to PDF. It creates a dedicated LibreOffice instance
// thanks to a custom user profile directory and a free port. Substantial calls
// to this method may increase CPU and memory usage drastically. In such a
// scenario, the given context may also be done before the end of the
// conversion.
// PDF converts a document to PDF.
//
// In stateless mode, it creates a dedicated LibreOffice instance thanks to a
// custom user profile directory and a free port. Substantial calls to this
// method may increase CPU and memory usage drastically. In such a scenario,
// the given context may also be done before the end of the conversion.
//
// In listener mode, it calls the unoconv listener to interact with
// LibreOffice, improving substantially the performance. However, it cannot
// perform parallel operations and have to wait for the lock to be available.
func (mod Unoconv) PDF(ctx context.Context, logger *zap.Logger, inputPath, outputPath string, options Options) error {
port, err := func() (int, error) {
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
return 0, fmt.Errorf("listen on the local network address: %w", err)
}
defer func() {
err := listener.Close()
if err != nil {
logger.Error(fmt.Sprintf("close listener: %s", err.Error()))
}
}()
addr := listener.Addr().String()
_, portStr, err := net.SplitHostPort(addr)
if err != nil {
return 0, fmt.Errorf("get free port from host: %w", err)
}
return strconv.Atoi(portStr)
}()
if err != nil {
return fmt.Errorf("get free port: %w", err)
}
userProfileDirPath := gotenberg.NewDirPath()
args := []string{
"--user-profile",
fmt.Sprintf("//%s", userProfileDirPath),
"--port",
fmt.Sprintf("%d", port),
"--format",
"pdf",
}
var userProfileDirPath string
if mod.disableListener {
port, err := freePort(logger)
if err != nil {
return fmt.Errorf("get free port: %w", err)
}
userProfileDirPath = gotenberg.NewDirPath()
args = append(args,
"--port",
fmt.Sprintf("%d", port),
"--user-profile",
fmt.Sprintf("//%s", userProfileDirPath),
)
} else {
args = append(args, "--port", fmt.Sprintf("%d", mod.listenerPort))
}
checkedEntry := logger.Check(zap.DebugLevel, "check for debug level before setting high verbosity")
if checkedEntry != nil {
args = append(args, "-vvv")
@@ -178,6 +296,36 @@ func (mod Unoconv) PDF(ctx context.Context, logger *zap.Logger, inputPath, outpu
args = append(args, "--output", outputPath, inputPath)
if !mod.disableListener {
listenerQueueLengthMu.Lock()
listenerQueueLength += 1
listenerQueueLengthMu.Unlock()
select {
case listenerLock <- struct{}{}:
logger.Debug("unoconv lock acquired")
listenerQueueLengthMu.Lock()
listenerQueueLength -= 1
listenerQueueLengthMu.Unlock()
break
case <-ctx.Done():
logger.Debug("failed to acquire the unoconv lock before deadline")
listenerQueueLengthMu.Lock()
listenerQueueLength -= 1
listenerQueueLengthMu.Unlock()
return fmt.Errorf("acquire unoconv lock: %w", ctx.Err())
}
defer func() {
<-listenerLock
logger.Debug("unoconv lock released")
}()
}
cmd, err := gotenberg.CommandContext(ctx, logger, mod.binPath, args...)
if err != nil {
return fmt.Errorf("create unoconv command: %w", err)
@@ -195,16 +343,18 @@ func (mod Unoconv) PDF(ctx context.Context, logger *zap.Logger, inputPath, outpu
activeInstancesCount -= 1
activeInstancesCountMu.Unlock()
// Always remove the user profile directory created by LibreOffice.
// See https://github.com/gotenberg/gotenberg/issues/192.
go func() {
logger.Debug(fmt.Sprintf("remove user profile directory '%s'", userProfileDirPath))
if mod.disableListener {
// Always remove the user profile directory created by LibreOffice.
// See https://github.com/gotenberg/gotenberg/issues/192.
go func() {
logger.Debug(fmt.Sprintf("remove user profile directory '%s'", userProfileDirPath))
err := os.RemoveAll(userProfileDirPath)
if err != nil {
logger.Error(fmt.Sprintf("remove user profile directory: %s", err))
}
}()
err := os.RemoveAll(userProfileDirPath)
if err != nil {
logger.Error(fmt.Sprintf("remove user profile directory: %s", err))
}
}()
}
if err == nil {
return nil
@@ -311,8 +461,13 @@ func (mod Unoconv) Extensions() []string {
}
var (
activeInstancesCount float64
activeInstancesCountMu sync.RWMutex
listenerLock = make(chan struct{}, 1)
listenerQueueLength float64
listenerQueueLengthMu sync.RWMutex
listenerActiveInstancesCount float64
listenerActiveInstancesCountMu sync.RWMutex
activeInstancesCount float64
activeInstancesCountMu sync.RWMutex
)
// Interface guards.
@@ -320,6 +475,7 @@ var (
_ gotenberg.Module = (*Unoconv)(nil)
_ gotenberg.Provisioner = (*Unoconv)(nil)
_ gotenberg.Validator = (*Unoconv)(nil)
_ gotenberg.App = (*Unoconv)(nil)
_ gotenberg.MetricsProvider = (*Unoconv)(nil)
_ API = (*Unoconv)(nil)
_ Provider = (*Unoconv)(nil)

View File

@@ -2,14 +2,42 @@ package unoconv
import (
"context"
"errors"
"os"
"reflect"
"testing"
"time"
"github.com/gotenberg/gotenberg/v7/pkg/gotenberg"
"go.uber.org/zap"
)
type ProtoModule struct {
descriptor func() gotenberg.ModuleDescriptor
}
func (mod ProtoModule) Descriptor() gotenberg.ModuleDescriptor {
return mod.descriptor()
}
type ProtoValidator struct {
ProtoModule
validate func() error
}
func (mod ProtoValidator) Validate() error {
return mod.validate()
}
type ProtoLoggerProvider struct {
ProtoModule
logger func(mod gotenberg.Module) (*zap.Logger, error)
}
func (factory ProtoLoggerProvider) Logger(mod gotenberg.Module) (*zap.Logger, error) {
return factory.logger(mod)
}
func TestUnoconv_Descriptor(t *testing.T) {
descriptor := Unoconv{}.Descriptor()
@@ -22,12 +50,69 @@ func TestUnoconv_Descriptor(t *testing.T) {
}
func TestUnoconv_Provision(t *testing.T) {
mod := new(Unoconv)
ctx := gotenberg.NewContext(gotenberg.ParsedFlags{}, nil)
for i, tc := range []struct {
ctx *gotenberg.Context
expectErr bool
}{
{
ctx: gotenberg.NewContext(
gotenberg.ParsedFlags{FlagSet: new(Unoconv).Descriptor().FlagSet},
make([]gotenberg.ModuleDescriptor, 0),
),
expectErr: true,
},
{
ctx: func() *gotenberg.Context {
mod := struct {
ProtoLoggerProvider
}{}
mod.descriptor = func() gotenberg.ModuleDescriptor {
return gotenberg.ModuleDescriptor{ID: "foo", New: func() gotenberg.Module { return mod }}
}
mod.logger = func(mod gotenberg.Module) (*zap.Logger, error) { return nil, errors.New("foo") }
err := mod.Provision(ctx)
if err != nil {
t.Errorf("expected no error but got: %v", err)
return gotenberg.NewContext(
gotenberg.ParsedFlags{
FlagSet: new(Unoconv).Descriptor().FlagSet,
},
[]gotenberg.ModuleDescriptor{
mod.Descriptor(),
},
)
}(),
expectErr: true,
},
{
ctx: func() *gotenberg.Context {
mod := struct {
ProtoLoggerProvider
}{}
mod.descriptor = func() gotenberg.ModuleDescriptor {
return gotenberg.ModuleDescriptor{ID: "foo", New: func() gotenberg.Module { return mod }}
}
mod.logger = func(mod gotenberg.Module) (*zap.Logger, error) { return zap.NewNop(), nil }
return gotenberg.NewContext(
gotenberg.ParsedFlags{
FlagSet: new(Unoconv).Descriptor().FlagSet,
},
[]gotenberg.ModuleDescriptor{
mod.Descriptor(),
},
)
}(),
},
} {
mod := new(Unoconv)
err := mod.Provision(tc.ctx)
if tc.expectErr && err == nil {
t.Errorf("test %d: expected error but got: %v", i, err)
}
if !tc.expectErr && err != nil {
t.Errorf("test %d: expected no error but got: %v", i, err)
}
}
}
@@ -47,8 +132,10 @@ func TestUnoconv_Validate(t *testing.T) {
binPath: os.Getenv("UNOCONV_BIN_PATH"),
},
} {
mod := new(Unoconv)
mod.binPath = tc.binPath
mod := Unoconv{
binPath: tc.binPath,
}
err := mod.Validate()
if tc.expectErr && err == nil {
@@ -61,20 +148,149 @@ func TestUnoconv_Validate(t *testing.T) {
}
}
func TestChromium_Metrics(t *testing.T) {
func TestUnoconv_Start(t *testing.T) {
for i, tc := range []struct {
mod *Unoconv
expectErr bool
}{
{
mod: &Unoconv{
disableListener: true,
logger: zap.NewNop(),
},
},
{
mod: &Unoconv{
binPath: os.Getenv("UNOCONV_BIN_PATH"),
disableListener: false,
logger: zap.NewExample(),
},
},
} {
func() {
err := tc.mod.Start()
if tc.expectErr && err == nil {
t.Errorf("test %d: expected error but got: %v", i, err)
}
if !tc.expectErr && err != nil {
t.Errorf("test %d: expected no error but got: %v", i, err)
}
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(1)*time.Nanosecond)
defer cancel()
err = tc.mod.Stop(ctx)
if err != nil {
t.Errorf("test %d: expected not error but got: %v", i, err)
}
}()
}
}
func TestUnoconv_StartupMessage(t *testing.T) {
for i, tc := range []struct {
disableListener bool
expectMessage string
}{
{
disableListener: true,
expectMessage: "listener disabled",
},
{
expectMessage: "listener started on port 0",
},
} {
mod := Unoconv{
disableListener: tc.disableListener,
}
actual := mod.StartupMessage()
if actual != tc.expectMessage {
t.Errorf("test %d: expected '%s' but got '%s'", i, tc.expectMessage, actual)
}
}
}
func TestUnoconv_Stop(t *testing.T) {
for i, tc := range []struct {
start bool
disableListener bool
timeout time.Duration
expectErr bool
}{
{
disableListener: true,
},
{
expectErr: true,
},
{
start: true,
timeout: time.Duration(1) * time.Nanosecond,
},
} {
func() {
mod := &Unoconv{
binPath: os.Getenv("UNOCONV_BIN_PATH"),
disableListener: tc.disableListener,
logger: zap.NewNop(),
}
if tc.start {
err := mod.Start()
if err != nil {
t.Fatalf("test %d: expected no error but got: %v", i, err)
}
}
var err error
if tc.timeout == 0 {
err = mod.Stop(context.TODO())
} else {
ctx, cancel := context.WithTimeout(context.Background(), tc.timeout)
defer cancel()
err = mod.Stop(ctx)
}
if tc.expectErr && err == nil {
t.Errorf("test %d: expected error but got: %v", i, err)
}
if !tc.expectErr && err != nil {
t.Errorf("test %d: expected no error but got: %v", i, err)
}
}()
}
}
func TestUnoconv_Metrics(t *testing.T) {
metrics, err := new(Unoconv).Metrics()
if err != nil {
t.Errorf("expected no error but got: %v", err)
t.Fatalf("expected no error but got: %v", err)
}
if len(metrics) != 1 {
t.Errorf("expected %d metrics, but got %d", 1, len(metrics))
if len(metrics) != 3 {
t.Fatalf("expected %d metrics, but got %d", 1, len(metrics))
}
actual := metrics[0].Read()
if actual != 0 {
t.Errorf("expected %d unoconv instances, but got %f", 0, actual)
}
actual = metrics[1].Read()
if actual != 0 {
t.Errorf("expected %d unoconv listener instances, but got %f", 0, actual)
}
actual = metrics[2].Read()
if actual != 0 {
t.Errorf("expected %d processes in the queue, but got %f", 0, actual)
}
}
func TestUnoconv_Unoconv(t *testing.T) {
@@ -89,17 +305,26 @@ func TestUnoconv_Unoconv(t *testing.T) {
func TestUnoconv_PDF(t *testing.T) {
for i, tc := range []struct {
ctx context.Context
mod Unoconv
logger *zap.Logger
inputPath string
options Options
expectErr bool
}{
{
mod: Unoconv{
binPath: os.Getenv("UNOCONV_BIN_PATH"),
disableListener: true,
},
logger: zap.NewNop(),
expectErr: true,
},
{
ctx: context.Background(),
ctx: context.Background(),
mod: Unoconv{
binPath: os.Getenv("UNOCONV_BIN_PATH"),
disableListener: true,
},
logger: zap.NewExample(),
inputPath: "/tests/test/testdata/libreoffice/sample1.docx",
options: Options{
@@ -109,7 +334,11 @@ func TestUnoconv_PDF(t *testing.T) {
},
},
{
ctx: context.Background(),
ctx: context.Background(),
mod: Unoconv{
binPath: os.Getenv("UNOCONV_BIN_PATH"),
disableListener: true,
},
logger: zap.NewNop(),
inputPath: "/tests/test/testdata/libreoffice/sample1.docx",
options: Options{
@@ -124,19 +353,40 @@ func TestUnoconv_PDF(t *testing.T) {
return ctx
}(),
mod: Unoconv{
binPath: os.Getenv("UNOCONV_BIN_PATH"),
disableListener: true,
},
logger: zap.NewNop(),
inputPath: "/tests/test/testdata/libreoffice/sample1.docx",
expectErr: true,
},
{
ctx: context.Background(),
mod: Unoconv{
binPath: os.Getenv("UNOCONV_BIN_PATH"),
logger: zap.NewNop(),
},
logger: zap.NewNop(),
inputPath: "/tests/test/testdata/libreoffice/sample1.docx",
},
{
ctx: func() context.Context {
ctx, cancel := context.WithCancel(context.TODO())
defer cancel()
return ctx
}(),
mod: Unoconv{
binPath: os.Getenv("UNOCONV_BIN_PATH"),
logger: zap.NewNop(),
},
logger: zap.NewNop(),
inputPath: "/tests/test/testdata/libreoffice/sample1.docx",
expectErr: true,
},
} {
func() {
mod := new(Unoconv)
err := mod.Provision(nil)
if err != nil {
t.Fatalf("test %d: expected error but got: %v", i, err)
}
outputDir, err := gotenberg.MkdirAll()
if err != nil {
t.Fatalf("test %d: expected error but got: %v", i, err)
@@ -149,7 +399,17 @@ func TestUnoconv_PDF(t *testing.T) {
}
}()
err = mod.PDF(tc.ctx, tc.logger, tc.inputPath, outputDir+"/foo.pdf", tc.options)
if !tc.mod.disableListener {
err = tc.mod.Start()
if err != nil {
t.Fatalf("test %d: expected no error but got: %v", i, err)
}
// Let's give it some room to start.
time.Sleep(time.Duration(1) * time.Second)
}
err = tc.mod.PDF(tc.ctx, tc.logger, tc.inputPath, outputDir+"/foo.pdf", tc.options)
if tc.expectErr && err == nil {
t.Errorf("test %d: expected error but got: %v", i, err)
@@ -158,6 +418,16 @@ func TestUnoconv_PDF(t *testing.T) {
if !tc.expectErr && err != nil {
t.Errorf("test %d: expected no error but got: %v", i, err)
}
if !tc.mod.disableListener {
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(1)*time.Nanosecond)
defer cancel()
err = tc.mod.Stop(ctx)
if err != nil {
t.Fatalf("test %d: expected no error but got: %v", i, err)
}
}
}()
}
}
@@ -173,3 +443,11 @@ func TestUnoconv_Extensions(t *testing.T) {
t.Errorf("expected %d extensions but got %d", expect, actual)
}
}
// Interface guards.
var (
_ gotenberg.Module = (*ProtoModule)(nil)
_ gotenberg.Validator = (*ProtoValidator)(nil)
_ gotenberg.LoggerProvider = (*ProtoLoggerProvider)(nil)
_ gotenberg.Module = (*ProtoLoggerProvider)(nil)
)