From b7a8b9582054a7323dc50663d352c7683fdd746b Mon Sep 17 00:00:00 2001 From: Julien Neuhart Date: Tue, 8 Feb 2022 16:12:41 +0100 Subject: [PATCH] fix(uno): close tcp connexions to the LibreOffice listener, better health check that take the restart into account --- go.sum | 6 - pkg/modules/libreoffice/uno/listener.go | 85 ++++++--- pkg/modules/libreoffice/uno/listener_test.go | 183 ++++++++++++++++--- pkg/modules/libreoffice/uno/uno.go | 10 +- pkg/modules/libreoffice/uno/uno_test.go | 11 +- 5 files changed, 229 insertions(+), 66 deletions(-) diff --git a/go.sum b/go.sum index 153d9680..9849f0d0 100644 --- a/go.sum +++ b/go.sum @@ -212,8 +212,6 @@ github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0j github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= github.com/mholt/archiver/v3 v3.5.1 h1:rDjOBX9JSF5BvoJGvjqK479aL70qh9DIpZCl+k7Clwo= github.com/mholt/archiver/v3 v3.5.1/go.mod h1:e3dqJ7H78uzsRSEACH1joayhuSyhnonssnDhppzS1L4= -github.com/microcosm-cc/bluemonday v1.0.17 h1:Z1a//hgsQ4yjC+8zEkV8IWySkXnsxmdSY642CTFQb5Y= -github.com/microcosm-cc/bluemonday v1.0.17/go.mod h1:Z0r70sCuXHig8YpBzCc5eGHAap2K7e/u082ZUpDRRqM= github.com/microcosm-cc/bluemonday v1.0.18 h1:6HcxvXDAi3ARt3slx6nTesbvorIc3QeTzBNRvWktHBo= github.com/microcosm-cc/bluemonday v1.0.18/go.mod h1:Z0r70sCuXHig8YpBzCc5eGHAap2K7e/u082ZUpDRRqM= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= @@ -231,8 +229,6 @@ github.com/orisano/pixelmatch v0.0.0-20210112091706-4fa4c7ba91d5/go.mod h1:nZgzb github.com/pdfcpu/pdfcpu v0.3.12 h1:B+MdKisilWNSk5OCO58Z9U6H93usH73xqk6hMOaZCls= github.com/pdfcpu/pdfcpu v0.3.12/go.mod h1:8XVBtVxuuIuSZL4Ez15Q4QoC+H8zeAaGnuiOEwAk8jA= github.com/pierrec/lz4/v4 v4.1.2/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= -github.com/pierrec/lz4/v4 v4.1.13 h1:/OvL3gfLjTf7nEATCYFLe4VeorMGI3nhLU5eb8FnEjU= -github.com/pierrec/lz4/v4 v4.1.13/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= github.com/pierrec/lz4/v4 v4.1.14 h1:+fL8AQEZtz/ijeNnpduH0bROTu0O3NZAlPjQxGn8LwE= github.com/pierrec/lz4/v4 v4.1.14/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= @@ -306,8 +302,6 @@ go.uber.org/goleak v1.1.11/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU= go.uber.org/multierr v1.7.0 h1:zaiO/rmgFjbmCXdSYJWQcdvOCsthmdaHfr3Gm2Kx4Ec= go.uber.org/multierr v1.7.0/go.mod h1:7EAYxJLBy9rStEaz58O2t4Uvip6FSURkq8/ppBp95ak= -go.uber.org/zap v1.20.0 h1:N4oPlghZwYG55MlU6LXk/Zp00FVNE9X9wrYO8CEs4lc= -go.uber.org/zap v1.20.0/go.mod h1:wjWOCqI0f2ZZrJF/UufIOkiC8ii6tm1iqIsLo76RfJw= go.uber.org/zap v1.21.0 h1:WefMeulhovoZ2sYXz7st6K0sLj7bBhpiFaud4r4zST8= go.uber.org/zap v1.21.0/go.mod h1:wjWOCqI0f2ZZrJF/UufIOkiC8ii6tm1iqIsLo76RfJw= golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= diff --git a/pkg/modules/libreoffice/uno/listener.go b/pkg/modules/libreoffice/uno/listener.go index 2d789b0b..bdf1e6b9 100644 --- a/pkg/modules/libreoffice/uno/listener.go +++ b/pkg/modules/libreoffice/uno/listener.go @@ -15,6 +15,7 @@ import ( type listener interface { start(logger *zap.Logger) error stop(logger *zap.Logger) error + restart(logger *zap.Logger) error lock(ctx context.Context, logger *zap.Logger) error unlock(logger *zap.Logger) error port() int @@ -33,6 +34,8 @@ type libreOfficeListener struct { cfgMu sync.RWMutex usage int + restarting bool + restartingMu sync.RWMutex queueLength int queueLengthMu sync.RWMutex lockChan chan struct{} @@ -106,8 +109,12 @@ func (listener *libreOfficeListener) start(logger *zap.Logger) error { return fmt.Errorf("waiting for the LibreOffice listener socket to be available: %w", ctx.Err()) } - _, err = net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), time.Duration(1)*time.Second) + conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), time.Duration(1)*time.Second) if err == nil { + err = conn.Close() + if err != nil { + logger.Debug(fmt.Sprintf("close connection after health checking the LibreOffice listener: %v", err)) + } break } } @@ -149,6 +156,32 @@ func (listener *libreOfficeListener) stop(logger *zap.Logger) error { return nil } +func (listener *libreOfficeListener) restart(logger *zap.Logger) error { + listener.restartingMu.Lock() + listener.restarting = true + listener.restartingMu.Unlock() + + defer func() { + listener.restartingMu.Lock() + listener.restarting = false + listener.restartingMu.Unlock() + }() + + err := listener.stop(logger) + if err != nil { + return fmt.Errorf("stop LibreOffice listener: %w", err) + } + + err = listener.start(logger) + if err != nil { + return fmt.Errorf("start LibreOffice listener: %w", err) + } + + listener.usage = 0 + + return nil +} + func (listener *libreOfficeListener) lock(ctx context.Context, logger *zap.Logger) error { listener.queueLengthMu.Lock() listener.queueLength += 1 @@ -162,6 +195,17 @@ func (listener *libreOfficeListener) lock(ctx context.Context, logger *zap.Logge listener.queueLength -= 1 listener.queueLengthMu.Unlock() + if !listener.healthy() { + logger.Debug("LibreOffice listener is unhealthy, restarting it...") + + err := listener.restart(logger) + if err == nil { + return nil + } + + return fmt.Errorf("restart LibreOffice listener: %w", err) + } + return nil case <-ctx.Done(): logger.Debug("failed to acquire LibreOffice listener lock before deadline") @@ -175,22 +219,6 @@ func (listener *libreOfficeListener) lock(ctx context.Context, logger *zap.Logge } func (listener *libreOfficeListener) unlock(logger *zap.Logger) error { - restart := func() error { - err := listener.stop(logger) - if err != nil { - return fmt.Errorf("stop LibreOffice listener: %w", err) - } - - err = listener.start(logger) - if err != nil { - return fmt.Errorf("start LibreOffice listener: %w", err) - } - - listener.usage = 0 - - return nil - } - defer func() { <-listener.lockChan logger.Debug("LibreOffice listener lock released") @@ -199,7 +227,7 @@ func (listener *libreOfficeListener) unlock(logger *zap.Logger) error { if !listener.healthy() { logger.Debug("LibreOffice listener is unhealthy, restarting it...") - err := restart() + err := listener.restart(logger) if err == nil { return nil } @@ -214,7 +242,7 @@ func (listener *libreOfficeListener) unlock(logger *zap.Logger) error { logger.Debug("LibreOffice listener threshold reached, restarting it...") - err := restart() + err := listener.restart(logger) if err == nil { return nil } @@ -237,9 +265,24 @@ func (listener *libreOfficeListener) queue() int { } func (listener *libreOfficeListener) healthy() bool { - _, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", listener.port()), time.Duration(1)*time.Second) + listener.restartingMu.RLock() + defer listener.restartingMu.RUnlock() - return err == nil + if listener.restarting { + return true + } + + conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", listener.port()), time.Duration(1)*time.Second) + if err == nil { + err := conn.Close() + if err != nil { + listener.logger.Debug(fmt.Sprintf("close connection after health checking the LibreOffice listener: %v", err)) + } + + return true + } + + return false } // Interface guards. diff --git a/pkg/modules/libreoffice/uno/listener_test.go b/pkg/modules/libreoffice/uno/listener_test.go index a2153c0f..49c9743f 100644 --- a/pkg/modules/libreoffice/uno/listener_test.go +++ b/pkg/modules/libreoffice/uno/listener_test.go @@ -2,7 +2,6 @@ package uno import ( "context" - "errors" "os" "testing" "time" @@ -74,7 +73,7 @@ func TestListener_stop(t *testing.T) { } } -func TestListener_lock(t *testing.T) { +func TestListener_restart(t *testing.T) { listener := newLibreOfficeListener( zap.NewNop(), os.Getenv("LIBREOFFICE_BIN_PATH"), @@ -82,18 +81,112 @@ func TestListener_lock(t *testing.T) { 10, ) - ctx, cancel := context.WithTimeout(context.Background(), time.Duration(10)*time.Second) - - err := listener.lock(ctx, zap.NewNop()) + err := listener.start(zap.NewNop()) if err != nil { - t.Fatalf("expected no error from listener.lock(), but got: %v", err) + t.Fatalf("expected no error from listener.start(), but got: %v", err) } - cancel() + err = listener.restart(zap.NewNop()) + if err != nil { + t.Errorf("expected no error from listener.stop(), but got: %v", err) + } - err = listener.lock(ctx, zap.NewNop()) - if !errors.Is(err, context.Canceled) { - t.Errorf("expected %v error, but got: %v", context.Canceled, err) + if !listener.healthy() { + t.Error("expected an healthy LibreOffice listener") + } +} + +func TestListener_lock(t *testing.T) { + tests := []struct { + name string + listener listener + ctx context.Context + teardown func(listener listener) error + expectLockErr bool + }{ + { + name: "nominal behavior", + listener: func() listener { + listener := newLibreOfficeListener(zap.NewNop(), os.Getenv("LIBREOFFICE_BIN_PATH"), time.Duration(10)*time.Second, 10) + + err := listener.start(zap.NewNop()) + if err != nil { + t.Fatalf("expected no error from listener.start(), but got: %v", err) + } + + return listener + }(), + ctx: context.Background(), + teardown: func(listener listener) error { + return listener.stop(zap.NewNop()) + }, + }, + { + name: "unhealthy listener", + listener: func() listener { + listener := newLibreOfficeListener(zap.NewNop(), os.Getenv("LIBREOFFICE_BIN_PATH"), time.Duration(10)*time.Second, 10) + + err := listener.start(zap.NewNop()) + if err != nil { + t.Fatalf("expected no error from listener.start(), but got: %v", err) + } + + err = listener.stop(zap.NewNop()) + if err != nil { + t.Fatalf("expected no error from listener.stop(), but got: %v", err) + } + + return listener + }(), + ctx: context.Background(), + teardown: func(listener listener) error { + return listener.stop(zap.NewNop()) + }, + }, + { + name: "context done", + listener: func() listener { + listener := newLibreOfficeListener(zap.NewNop(), os.Getenv("LIBREOFFICE_BIN_PATH"), time.Duration(10)*time.Second, 10) + + err := listener.start(zap.NewNop()) + if err != nil { + t.Fatalf("expected no error from listener.start(), but got: %v", err) + } + + return listener + }(), + ctx: func() context.Context { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + return ctx + }(), + expectLockErr: true, + teardown: func(listener listener) error { + return listener.stop(zap.NewNop()) + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + defer func() { + err := tc.teardown(tc.listener) + if err != nil { + t.Errorf("expected no error from tc.teardown(), but got: %v", err) + } + }() + + err := tc.listener.lock(tc.ctx, zap.NewNop()) + + if tc.expectLockErr && err == nil { + t.Fatalf("expected listener.lock() error, but got none") + } + + if !tc.expectLockErr && err != nil { + t.Fatalf("expected no error from listener.lock(), but got: %v", err) + } + }) } } @@ -112,6 +205,12 @@ func TestListener_unlock(t *testing.T) { if err != nil { t.Fatalf("expected no error from listener.start(), but got: %v", err) } + + err = listener.lock(context.Background(), zap.NewNop()) + if err != nil { + t.Fatalf("expected no error from listener.lock(), but got: %v", err) + } + return listener }(), teardown: func(listener listener) error { @@ -128,6 +227,11 @@ func TestListener_unlock(t *testing.T) { t.Fatalf("expected no error from listener.start(), but got: %v", err) } + err = listener.lock(context.Background(), zap.NewNop()) + if err != nil { + t.Fatalf("expected no error from listener.lock(), but got: %v", err) + } + err = listener.stop(zap.NewNop()) if err != nil { t.Fatalf("expected no error from listener.stop(), but got: %v", err) @@ -148,6 +252,12 @@ func TestListener_unlock(t *testing.T) { if err != nil { t.Fatalf("expected no error from listener.start(), but got: %v", err) } + + err = listener.lock(context.Background(), zap.NewNop()) + if err != nil { + t.Fatalf("expected no error from listener.lock(), but got: %v", err) + } + return listener }(), teardown: func(listener listener) error { @@ -158,23 +268,17 @@ func TestListener_unlock(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - ctx, cancel := context.WithTimeout(context.Background(), time.Duration(10)*time.Second) - defer cancel() + defer func() { + err := tc.teardown(tc.listener) + if err != nil { + t.Errorf("expected no error from tc.teardown(), but got: %v", err) + } + }() - err := tc.listener.lock(ctx, zap.NewNop()) - if err != nil { - t.Fatalf("expected no error from listener.lock(), but got: %v", err) - } - - err = tc.listener.unlock(zap.NewNop()) + err := tc.listener.unlock(zap.NewNop()) if err != nil { t.Errorf("expected no error from listener.unlock(), but got: %v", err) } - - err = tc.teardown(tc.listener) - if err != nil { - t.Errorf("expected no error from tc.teardown(), but got: %v", err) - } }) } } @@ -211,6 +315,18 @@ func TestListener_queue(t *testing.T) { 10, ) + err := listener.start(zap.NewNop()) + if err != nil { + t.Fatalf("expected no error from listener.start(), but got: %v", err) + } + + defer func() { + err := listener.stop(zap.NewNop()) + if err != nil { + t.Errorf("expected no error from listener.stop(), but got: %v", err) + } + }() + queueLength := listener.queue() if queueLength != 0 { t.Fatalf("expected a zero value from listener.queue(), but got %d", queueLength) @@ -218,7 +334,7 @@ func TestListener_queue(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), time.Duration(10)*time.Second) - err := listener.lock(ctx, zap.NewNop()) + err = listener.lock(ctx, zap.NewNop()) if err != nil { t.Fatalf("expected no error from listener.lock(), but got: %v", err) } @@ -261,12 +377,13 @@ func TestListener_queue(t *testing.T) { } func TestListener_healthy(t *testing.T) { - listener := newLibreOfficeListener( - zap.NewNop(), - os.Getenv("LIBREOFFICE_BIN_PATH"), - time.Duration(10)*time.Second, - 10, - ) + listener := &libreOfficeListener{ + binPath: os.Getenv("LIBREOFFICE_BIN_PATH"), + startTimeout: time.Duration(10) * time.Second, + threshold: 10, + lockChan: make(chan struct{}, 1), + logger: zap.NewNop(), + } err := listener.start(zap.NewNop()) if err != nil { @@ -285,4 +402,10 @@ func TestListener_healthy(t *testing.T) { if listener.healthy() { t.Errorf("expected a non-healthy LibreOffice listener") } + + listener.restarting = true + + if !listener.healthy() { + t.Error("expected an healthy LibreOffice listener") + } } diff --git a/pkg/modules/libreoffice/uno/uno.go b/pkg/modules/libreoffice/uno/uno.go index 88de6e96..22f7b932 100644 --- a/pkg/modules/libreoffice/uno/uno.go +++ b/pkg/modules/libreoffice/uno/uno.go @@ -179,10 +179,10 @@ func (mod UNO) Start() error { // StartupMessage returns a custom startup message. func (mod UNO) StartupMessage() string { if mod.libreOfficeRestartThreshold == 0 { - return "Long-running LibreOffice listener disabled" + return "long-running LibreOffice listener disabled" } - return "Long-running LibreOffice listener started" + return "long-running LibreOffice listener started" } // Stop stops the long-running LibreOffice Listener if it exists. @@ -288,10 +288,6 @@ func (mod UNO) Checks() ([]health.CheckerOption, error) { return errors.New("long-running LibreOffice listener unhealthy") }, - // The long-running LibreOffice listener may be restarting, so we - // wait a given amount of time until we consider the module - // unavailable. - MaxTimeInError: mod.libreOfficeStartTimeout, }), }, nil } @@ -339,6 +335,8 @@ func (mod UNO) PDF(ctx context.Context, logger *zap.Logger, inputPath, outputPat err := mod.listener.unlock(logger) if err != nil { mod.logger.Error(fmt.Sprintf("unlock long-running LibreOffice listener: %v", err)) + + return } }() }() diff --git a/pkg/modules/libreoffice/uno/uno_test.go b/pkg/modules/libreoffice/uno/uno_test.go index 73f5b454..3acbbb3f 100644 --- a/pkg/modules/libreoffice/uno/uno_test.go +++ b/pkg/modules/libreoffice/uno/uno_test.go @@ -258,14 +258,14 @@ func TestUNO_StartupMessage(t *testing.T) { mod: UNO{ libreOfficeRestartThreshold: 10, }, - expectMessage: "Long-running LibreOffice listener started", + expectMessage: "long-running LibreOffice listener started", }, { name: "long-running LibreOffice listener disabled", mod: UNO{ libreOfficeRestartThreshold: 0, }, - expectMessage: "Long-running LibreOffice listener disabled", + expectMessage: "long-running LibreOffice listener disabled", }, } @@ -755,7 +755,7 @@ func TestUNO_PDF(t *testing.T) { }, logger: zap.NewNop(), }, - ctx: nil, + ctx: context.Background(), logger: zap.NewNop(), inputPath: "/tests/test/testdata/libreoffice/sample1.docx", expectPDFErr: true, @@ -824,6 +824,7 @@ func TestUNO_UNO(t *testing.T) { type listenerMock struct { startMock func(logger *zap.Logger) error stopMock func(logger *zap.Logger) error + restartMock func(logger *zap.Logger) error lockMock func(ctx context.Context, logger *zap.Logger) error unlockMock func(logger *zap.Logger) error portMock func() int @@ -839,6 +840,10 @@ func (listener listenerMock) stop(logger *zap.Logger) error { return listener.stopMock(logger) } +func (listener listenerMock) restart(logger *zap.Logger) error { + return listener.restartMock(logger) +} + func (listener listenerMock) lock(ctx context.Context, logger *zap.Logger) error { return listener.lockMock(ctx, logger) }