Bound document content extraction time and decouple it from uploads (MM-69098) (#36856) (#37043)

Automatic Merge

server/channels/app/file.go

@@ -858,12 +858,14 @@ func (a *App) UploadFileX(c request.CTX, channelID, name string, input io.Reader
 
 	if *a.Config().FileSettings.ExtractContent && t.ExtractContent {
 		infoCopy := *t.fileinfo
-		a.Srv().GoBuffered(func() {
+		if !a.Srv().GoExtraction(func() {
 			err := a.ExtractContentFromFileInfo(c, &infoCopy)
 			if err != nil {
 				c.Logger().Error("Failed to extract file content", mlog.Err(err), mlog.String("fileInfoId", infoCopy.Id))
 			}
-		})
+		}) {
+			c.Logger().Warn("Content extraction queue is full, skipping inline extraction; this file's content will not be searchable until an admin runs a content extraction job (e.g. mmctl extract)", mlog.String("fileInfoId", infoCopy.Id))
+		}
 	}
 
 	return t.fileinfo, nil
@@ -1125,12 +1127,14 @@ func (a *App) DoUploadFileExpectModification(c request.CTX, now time.Time, rawTe
 	// and something we can do without.
 	if *a.Config().FileSettings.ExtractContent && extractContent {
 		infoCopy := *info
-		a.Srv().GoBuffered(func() {
+		if !a.Srv().GoExtraction(func() {
 			err := a.ExtractContentFromFileInfo(c, &infoCopy)
 			if err != nil {
 				c.Logger().Error("Failed to extract file content", mlog.Err(err), mlog.String("fileInfoId", infoCopy.Id))
 			}
-		})
+		}) {
+			c.Logger().Warn("Content extraction queue is full, skipping inline extraction; this file's content will not be searchable until an admin runs a content extraction job (e.g. mmctl extract)", mlog.String("fileInfoId", infoCopy.Id))
+		}
 	}
 
 	return info, data, nil
@@ -1584,10 +1588,15 @@ func (a *App) ExtractContentFromFileInfo(rctx request.CTX, fileInfo *model.FileI
 	if aerr != nil {
 		return errors.Wrap(aerr, "failed to open file for extract file content")
 	}
-	defer file.Close()
+	// Ownership of closing the file is handed to docextractor.Extract via
+	// ReaderCloser: with a timeout configured, extraction may continue on a
+	// detached goroutine after Extract returns, so closing the file here would
+	// race with that goroutine still reading it.
 	text, err := docextractor.Extract(rctx.Logger(), fileInfo.Name, file, docextractor.ExtractSettings{
 		ArchiveRecursion: *a.Config().FileSettings.ArchiveRecursion,
 		MaxFileSize:      *a.Config().FileSettings.MaxFileSize,
+		Timeout:          time.Duration(*a.Config().FileSettings.ExtractContentTimeout) * time.Second,
+		ReaderCloser:     file,
 	})
 	if err != nil {
 		return errors.Wrap(err, "failed to extract file content")

server/channels/app/platform/goroutines.go

@@ -3,7 +3,10 @@
 
 package platform
 
-import "sync/atomic"
+import (
+	"runtime"
+	"sync/atomic"
+)
 
 // Go creates a goroutine, but maintains a record of it to ensure that execution completes before
 // the server is shutdown.
@@ -45,3 +48,55 @@ func (ps *PlatformService) GoBuffered(f func()) {
 		<-ps.goroutineBuffered
 	}()
 }
+
+// startExtractionWorkers launches the fixed-size pool of workers that run
+// document extraction tasks submitted through GoExtraction.
+func (ps *PlatformService) startExtractionWorkers() {
+	numWorkers := runtime.NumCPU()
+	for range numWorkers {
+		ps.extractionWG.Go(func() {
+			for {
+				select {
+				case <-ps.extractionStop:
+					return
+				case f := <-ps.extractionQueue:
+					f()
+				}
+			}
+		})
+	}
+}
+
+// stopExtractionWorkers signals the extraction workers to exit and waits for
+// any in-flight extraction to finish. Queued-but-not-started tasks are drained
+// and discarded so a worker cannot dequeue and run them after shutdown has been
+// signaled.
+func (ps *PlatformService) stopExtractionWorkers() {
+	close(ps.extractionStop)
+
+drain:
+	for {
+		select {
+		case <-ps.extractionQueue:
+		default:
+			break drain
+		}
+	}
+
+	ps.extractionWG.Wait()
+}
+
+// GoExtraction submits f to the bounded document extraction worker pool. It
+// never blocks the caller: if every worker is busy and the queue is full it
+// returns false without running f. Skipped files stay unextracted until an
+// admin runs a content extraction job (e.g. mmctl extract); there is no
+// scheduler that picks them up automatically. This keeps expensive extractions
+// from stalling the request goroutines that dispatch them.
+func (ps *PlatformService) GoExtraction(f func()) bool {
+	select {
+	case ps.extractionQueue <- f:
+		return true
+	default:
+		return false
+	}
+}

server/channels/app/platform/goroutines_test.go

@@ -0,0 +1,79 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package platform
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestGoExtraction(t *testing.T) {
+	mainHelper.Parallel(t)
+
+	t.Run("runs submitted work on the pool", func(t *testing.T) {
+		const tasks = 5
+		ps := &PlatformService{
+			extractionQueue: make(chan func(), tasks),
+			extractionStop:  make(chan struct{}),
+		}
+		ps.startExtractionWorkers()
+		defer ps.stopExtractionWorkers()
+
+		var wg sync.WaitGroup
+		wg.Add(tasks)
+		for range tasks {
+			require.True(t, ps.GoExtraction(func() {
+				wg.Done()
+			}))
+		}
+
+		done := make(chan struct{})
+		go func() {
+			wg.Wait()
+			close(done)
+		}()
+		select {
+		case <-done:
+		case <-time.After(5 * time.Second):
+			require.Fail(t, "submitted extraction tasks did not run")
+		}
+	})
+
+	t.Run("never blocks and skips work once the queue is saturated", func(t *testing.T) {
+		// No workers are started, so nothing drains the queue.
+		ps := &PlatformService{
+			extractionQueue: make(chan func(), 2),
+			extractionStop:  make(chan struct{}),
+		}
+
+		require.True(t, ps.GoExtraction(func() {}))
+		require.True(t, ps.GoExtraction(func() {}))
+		// The queue is now full; further submissions must be rejected rather
+		// than block the caller.
+		require.False(t, ps.GoExtraction(func() {}))
+	})
+
+	t.Run("stop waits for in-flight extraction to finish", func(t *testing.T) {
+		ps := &PlatformService{
+			extractionQueue: make(chan func(), 1),
+			extractionStop:  make(chan struct{}),
+		}
+		ps.startExtractionWorkers()
+
+		var finished bool
+		started := make(chan struct{})
+		require.True(t, ps.GoExtraction(func() {
+			close(started)
+			time.Sleep(100 * time.Millisecond)
+			finished = true
+		}))
+
+		<-started
+		ps.stopExtractionWorkers()
+		require.True(t, finished, "stopExtractionWorkers should wait for the running task to complete")
+	})
+}

server/channels/app/platform/service.go

@@ -107,6 +107,13 @@ type PlatformService struct {
 	goroutineExitSignal chan struct{}
 	goroutineBuffered   chan struct{}
 
+	// Document content extraction runs on a dedicated, bounded worker pool so
+	// that expensive extractions cannot saturate the generic worker pool and
+	// block the request goroutines that dispatch them.
+	extractionQueue chan func()
+	extractionStop  chan struct{}
+	extractionWG    sync.WaitGroup
+
 	additionalClusterHandlers map[model.ClusterEvent]einterfaces.ClusterMessageHandler
 
 	shareChannelServiceMux sync.RWMutex
@@ -136,6 +143,8 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) {
 		hashSeed:            maphash.MakeSeed(),
 		goroutineExitSignal: make(chan struct{}, 1),
 		goroutineBuffered:   make(chan struct{}, runtime.NumCPU()),
+		extractionQueue:     make(chan func(), runtime.NumCPU()),
+		extractionStop:      make(chan struct{}),
 		WebSocketRouter: &WebSocketRouter{
 			handlers: make(map[string]webSocketHandler),
 		},
@@ -401,6 +410,8 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) {
 	ps.searchConfigListenerId = searchConfigListenerId
 	ps.searchLicenseListenerId = searchLicenseListenerId
 
+	ps.startExtractionWorkers()
+
 	return ps, nil
 }
 
@@ -517,6 +528,10 @@ func (ps *PlatformService) Shutdown() error {
 
 	ps.RemoveLicenseListener(ps.licenseListenerId)
 
+	// Stop the document extraction workers and wait for any in-flight
+	// extraction to finish before closing the store it depends on.
+	ps.stopExtractionWorkers()
+
 	// we need to wait the goroutines to finish before closing the store
 	// and this needs to be called after hub stop because hub generates goroutines
 	// when it is active. If we wait first we have no mechanism to prevent adding

server/channels/app/server.go

@@ -813,6 +813,14 @@ func (s *Server) GoBuffered(f func()) {
 	s.platform.GoBuffered(f)
 }
 
+// GoExtraction submits f to the bounded document extraction worker pool without
+// blocking the caller. It returns false if the pool is saturated and f was not
+// run; skipped files stay unextracted until an admin runs a content extraction
+// job (e.g. mmctl extract).
+func (s *Server) GoExtraction(f func()) bool {
+	return s.platform.GoExtraction(f)
+}
+
 var corsAllowedMethods = []string{
 	"POST",
 	"GET",

server/channels/app/upload.go

@@ -339,12 +339,14 @@ func (a *App) UploadData(c request.CTX, us *model.UploadSession, rd io.Reader) (
 
 	if *a.Config().FileSettings.ExtractContent {
 		infoCopy := *info
-		a.Srv().Go(func() {
+		if !a.Srv().GoExtraction(func() {
 			err := a.ExtractContentFromFileInfo(c, &infoCopy)
 			if err != nil {
 				c.Logger().Error("Failed to extract file content", mlog.Err(err), mlog.String("fileInfoId", infoCopy.Id))
 			}
-		})
+		}) {
+			c.Logger().Warn("Content extraction queue is full, skipping inline extraction; this file's content will not be searchable until an admin runs a content extraction job (e.g. mmctl extract)", mlog.String("fileInfoId", infoCopy.Id))
+		}
 	}
 
 	// delete upload session