Add Goroutine stack inspector to admin/monitor (#19207)

Continues on from #19202.

Following the addition of pprof labels we can now more easily understand the relationship between a goroutine and the requests that spawn them. 

This PR takes advantage of the labels and adds a few others, then provides a mechanism for the monitoring page to query the pprof goroutine profile.

The binary profile that results from this profile is immediately piped in to the google library for parsing this and then stack traces are formed for the goroutines.

If the goroutine is within a context or has been created from a goroutine within a process context it will acquire the process description labels for that process. 

The goroutines are mapped with there associate pids and any that do not have an associated pid are placed in a group at the bottom as unbound.

In this way we should be able to more easily examine goroutines that have been stuck.

A manager command `gitea manager processes` is also provided that can export the processes (with or without stacktraces) to the command line.

Signed-off-by: Andrew Thornton <art27@cantab.net>

cmd/web.go

@@ -16,6 +16,7 @@ import (
 
 	"code.gitea.io/gitea/modules/graceful"
 	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/process"
 	"code.gitea.io/gitea/modules/setting"
 	"code.gitea.io/gitea/routers"
 	"code.gitea.io/gitea/routers/install"
@@ -59,6 +60,9 @@ and it takes care of all the other things for you`,
 }
 
 func runHTTPRedirector() {
+	_, _, finished := process.GetManager().AddTypedContext(graceful.GetManager().HammerContext(), "Web: HTTP Redirector", process.SystemProcessType, true)
+	defer finished()
+
 	source := fmt.Sprintf("%s:%s", setting.HTTPAddr, setting.PortToRedirect)
 	dest := strings.TrimSuffix(setting.AppURL, "/")
 	log.Info("Redirecting: %s to %s", source, dest)
@@ -141,8 +145,10 @@ func runWeb(ctx *cli.Context) error {
 
 	if setting.EnablePprof {
 		go func() {
+			_, _, finished := process.GetManager().AddTypedContext(context.Background(), "Web: PProf Server", process.SystemProcessType, true)
 			log.Info("Starting pprof server on localhost:6060")
 			log.Info("%v", http.ListenAndServe("localhost:6060", nil))
+			finished()
 		}()
 	}
 
@@ -204,6 +210,8 @@ func listen(m http.Handler, handleRedirector bool) error {
 	if setting.Protocol != setting.HTTPUnix && setting.Protocol != setting.FCGIUnix {
 		listenAddr = net.JoinHostPort(listenAddr, setting.HTTPPort)
 	}
+	_, _, finished := process.GetManager().AddTypedContext(graceful.GetManager().HammerContext(), "Web: Gitea Server", process.SystemProcessType, true)
+	defer finished()
 	log.Info("Listen: %v://%s%s", setting.Protocol, listenAddr, setting.AppSubURL)
 	// This can be useful for users, many users do wrong to their config and get strange behaviors behind a reverse-proxy.
 	// A user may fix the configuration mistake when he sees this log.

cmd/web_acme.go

@@ -15,6 +15,7 @@ import (
 
 	"code.gitea.io/gitea/modules/graceful"
 	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/process"
 	"code.gitea.io/gitea/modules/setting"
 
 	"github.com/caddyserver/certmagic"
@@ -107,6 +108,9 @@ func runACME(listenAddr string, m http.Handler) error {
 
 	if enableHTTPChallenge {
 		go func() {
+			_, _, finished := process.GetManager().AddTypedContext(graceful.GetManager().HammerContext(), "Web: ACME HTTP challenge server", process.SystemProcessType, true)
+			defer finished()
+
 			log.Info("Running Let's Encrypt handler on %s", setting.HTTPAddr+":"+setting.PortToRedirect)
 			// all traffic coming into HTTP will be redirect to HTTPS automatically (LE HTTP-01 validation happens here)
 			err := runHTTP("tcp", setting.HTTPAddr+":"+setting.PortToRedirect, "Let's Encrypt HTTP Challenge", myACME.HTTPChallengeHandler(http.HandlerFunc(runLetsEncryptFallbackHandler)))

docs/content/doc/usage/command-line.en-us.md

@@ -503,6 +503,13 @@ Manage running server operations:
               - `--host value`, `-H value`: Mail server host (defaults to: 127.0.0.1:25)
               - `--send-to value`, `-s value`: Email address(es) to send to
               - `--subject value`, `-S value`: Subject header of sent emails
+  - `processes`: Display Gitea processes and goroutine information
+    - Options:
+      - `--flat`: Show processes as flat table rather than as tree
+      - `--no-system`: Do not show system processes
+      - `--stacktraces`: Show stacktraces for goroutines associated with processes
+      - `--json`: Output as json
+      - `--cancel PID`: Send cancel to process with PID. (Only for non-system processes.)
 
 ### dump-repo
 

go.mod

@@ -42,6 +42,7 @@ require (
 	github.com/gogs/go-gogs-client v0.0.0-20210131175652-1d7215cd8d85
 	github.com/golang-jwt/jwt/v4 v4.3.0
 	github.com/google/go-github/v39 v39.2.0
+	github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1
 	github.com/google/uuid v1.3.0
 	github.com/gorilla/feeds v1.1.1
 	github.com/gorilla/sessions v1.2.1

go.sum

@@ -761,6 +761,7 @@ github.com/google/pprof v0.0.0-20210226084205-cbba55b83ad5/go.mod h1:kpwsk12EmLe
 github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/google/pprof v0.0.0-20210601050228-01bbb1931b22/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/google/pprof v0.0.0-20210609004039-a478d1d731e9/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
+github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1 h1:K6RDEckDVWvDI9JAJYCmNdQXq6neHJOYx3V6jnqNEec=
 github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/google/rpmpack v0.0.0-20191226140753-aa36bfddb3a0/go.mod h1:RaTPr0KUf2K7fnZYLNDrr8rxAamWs3iNywJLtQ2AzBg=

modules/context/private.go

@@ -79,6 +79,6 @@ func PrivateContexter() func(http.Handler) http.Handler {
 // the underlying request has timed out from the ssh/http push
 func OverrideContext(ctx *PrivateContext) (cancel context.CancelFunc) {
 	// We now need to override the request context as the base for our work because even if the request is cancelled we have to continue this work
-	ctx.Override, _, cancel = process.GetManager().AddContext(graceful.GetManager().HammerContext(), fmt.Sprintf("PrivateContext: %s", ctx.Req.RequestURI))
+	ctx.Override, _, cancel = process.GetManager().AddTypedContext(graceful.GetManager().HammerContext(), fmt.Sprintf("PrivateContext: %s", ctx.Req.RequestURI), process.RequestProcessType, true)
 	return
 }

modules/eventsource/manager_run.go

@@ -11,6 +11,7 @@ import (
 	"code.gitea.io/gitea/models"
 	"code.gitea.io/gitea/modules/graceful"
 	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/process"
 	"code.gitea.io/gitea/modules/setting"
 	"code.gitea.io/gitea/modules/timeutil"
 )
@@ -25,6 +26,9 @@ func (m *Manager) Init() {
 
 // Run runs the manager within a provided context
 func (m *Manager) Run(ctx context.Context) {
+	ctx, _, finished := process.GetManager().AddTypedContext(ctx, "Service: EventSource", process.SystemProcessType, true)
+	defer finished()
+
 	then := timeutil.TimeStampNow().Add(-2)
 	timer := time.NewTicker(setting.UI.Notification.EventSourceUpdateTime)
 loop:

modules/graceful/manager_unix.go

@@ -18,6 +18,7 @@ import (
 	"time"
 
 	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/process"
 	"code.gitea.io/gitea/modules/setting"
 )
 
@@ -73,7 +74,7 @@ func (g *Manager) start(ctx context.Context) {
 
 	// Set the running state & handle signals
 	g.setState(stateRunning)
-	go g.handleSignals(ctx)
+	go g.handleSignals(g.managerCtx)
 
 	// Handle clean up of unused provided listeners	and delayed start-up
 	startupDone := make(chan struct{})
@@ -112,6 +113,9 @@ func (g *Manager) start(ctx context.Context) {
 }
 
 func (g *Manager) handleSignals(ctx context.Context) {
+	ctx, _, finished := process.GetManager().AddTypedContext(ctx, "Graceful: HandleSignals", process.SystemProcessType, true)
+	defer finished()
+
 	signalChannel := make(chan os.Signal, 1)
 
 	signal.Notify(

modules/indexer/code/indexer.go

@@ -7,6 +7,7 @@ package code
 import (
 	"context"
 	"os"
+	"runtime/pprof"
 	"strconv"
 	"strings"
 	"time"
@@ -15,6 +16,7 @@ import (
 	repo_model "code.gitea.io/gitea/models/repo"
 	"code.gitea.io/gitea/modules/graceful"
 	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/process"
 	"code.gitea.io/gitea/modules/queue"
 	"code.gitea.io/gitea/modules/setting"
 	"code.gitea.io/gitea/modules/timeutil"
@@ -116,7 +118,7 @@ func Init() {
 		return
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel, finished := process.GetManager().AddTypedContext(context.Background(), "Service: CodeIndexer", process.SystemProcessType, false)
 
 	graceful.GetManager().RunAtTerminate(func() {
 		select {
@@ -128,6 +130,7 @@ func Init() {
 		log.Debug("Closing repository indexer")
 		indexer.Close()
 		log.Info("PID: %d Repository Indexer closed", os.Getpid())
+		finished()
 	})
 
 	waitChannel := make(chan time.Duration)
@@ -172,6 +175,7 @@ func Init() {
 	}
 
 	go func() {
+		pprof.SetGoroutineLabels(ctx)
 		start := time.Now()
 		var (
 			rIndexer Indexer
@@ -247,6 +251,7 @@ func Init() {
 
 	if setting.Indexer.StartupTimeout > 0 {
 		go func() {
+			pprof.SetGoroutineLabels(ctx)
 			timeout := setting.Indexer.StartupTimeout
 			if graceful.GetManager().IsChild() && setting.GracefulHammerTime > 0 {
 				timeout += setting.GracefulHammerTime

modules/indexer/issues/indexer.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"runtime/pprof"
 	"sync"
 	"time"
 
@@ -16,6 +17,7 @@ import (
 	repo_model "code.gitea.io/gitea/models/repo"
 	"code.gitea.io/gitea/modules/graceful"
 	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/process"
 	"code.gitea.io/gitea/modules/queue"
 	"code.gitea.io/gitea/modules/setting"
 	"code.gitea.io/gitea/modules/util"
@@ -100,6 +102,8 @@ var (
 // InitIssueIndexer initialize issue indexer, syncReindex is true then reindex until
 // all issue index done.
 func InitIssueIndexer(syncReindex bool) {
+	ctx, _, finished := process.GetManager().AddTypedContext(context.Background(), "Service: IssueIndexer", process.SystemProcessType, false)
+
 	waitChannel := make(chan time.Duration)
 
 	// Create the Queue
@@ -165,6 +169,7 @@ func InitIssueIndexer(syncReindex bool) {
 
 	// Create the Indexer
 	go func() {
+		pprof.SetGoroutineLabels(ctx)
 		start := time.Now()
 		log.Info("PID %d: Initializing Issue Indexer: %s", os.Getpid(), setting.Indexer.IssueType)
 		var populate bool
@@ -193,11 +198,13 @@ func InitIssueIndexer(syncReindex bool) {
 				if issueIndexer != nil {
 					issueIndexer.Close()
 				}
+				finished()
 				log.Info("PID: %d Issue Indexer closed", os.Getpid())
 			})
 			log.Debug("Created Bleve Indexer")
 		case "elasticsearch":
 			graceful.GetManager().RunWithShutdownFns(func(_, atTerminate func(func())) {
+				pprof.SetGoroutineLabels(ctx)
 				issueIndexer, err := NewElasticSearchIndexer(setting.Indexer.IssueConnStr, setting.Indexer.IssueIndexerName)
 				if err != nil {
 					log.Fatal("Unable to initialize Elastic Search Issue Indexer at connection: %s Error: %v", setting.Indexer.IssueConnStr, err)
@@ -208,10 +215,12 @@ func InitIssueIndexer(syncReindex bool) {
 				}
 				populate = !exist
 				holder.set(issueIndexer)
+				atTerminate(finished)
 			})
 		case "db":
 			issueIndexer := &DBIndexer{}
 			holder.set(issueIndexer)
+			graceful.GetManager().RunAtTerminate(finished)
 		default:
 			holder.cancel()
 			log.Fatal("Unknown issue indexer type: %s", setting.Indexer.IssueType)
@@ -251,6 +260,7 @@ func InitIssueIndexer(syncReindex bool) {
 		}
 	} else if setting.Indexer.StartupTimeout > 0 {
 		go func() {
+			pprof.SetGoroutineLabels(ctx)
 			timeout := setting.Indexer.StartupTimeout
 			if graceful.GetManager().IsChild() && setting.GracefulHammerTime > 0 {
 				timeout += setting.GracefulHammerTime
@@ -272,6 +282,8 @@ func InitIssueIndexer(syncReindex bool) {
 
 // populateIssueIndexer populate the issue indexer with issue data
 func populateIssueIndexer(ctx context.Context) {
+	ctx, _, finished := process.GetManager().AddTypedContext(ctx, "Service: PopulateIssueIndexer", process.SystemProcessType, true)
+	defer finished()
 	for page := 1; ; page++ {
 		select {
 		case <-ctx.Done():

modules/log/event.go

@@ -5,9 +5,13 @@
 package log
 
 import (
+	"context"
 	"fmt"
+	"runtime/pprof"
 	"sync"
 	"time"
+
+	"code.gitea.io/gitea/modules/process"
 )
 
 // Event represents a logging event
@@ -34,6 +38,8 @@ type EventLogger interface {
 
 // ChannelledLog represents a cached channel to a LoggerProvider
 type ChannelledLog struct {
+	ctx            context.Context
+	finished       context.CancelFunc
 	name           string
 	provider       string
 	queue          chan *Event
@@ -44,8 +50,9 @@ type ChannelledLog struct {
 }
 
 // NewChannelledLog a new logger instance with given logger provider and config.
-func NewChannelledLog(name, provider, config string, bufferLength int64) (*ChannelledLog, error) {
+func NewChannelledLog(parent context.Context, name, provider, config string, bufferLength int64) (*ChannelledLog, error) {
 	if log, ok := providers[provider]; ok {
+
 		l := &ChannelledLog{
 			queue:  make(chan *Event, bufferLength),
 			flush:  make(chan bool),
@@ -58,6 +65,7 @@ func NewChannelledLog(name, provider, config string, bufferLength int64) (*Chann
 		}
 		l.name = name
 		l.provider = provider
+		l.ctx, _, l.finished = process.GetManager().AddTypedContext(parent, fmt.Sprintf("Logger: %s(%s)", l.name, l.provider), process.SystemProcessType, false)
 		go l.Start()
 		return l, nil
 	}
@@ -66,6 +74,8 @@ func NewChannelledLog(name, provider, config string, bufferLength int64) (*Chann
 
 // Start processing the ChannelledLog
 func (l *ChannelledLog) Start() {
+	pprof.SetGoroutineLabels(l.ctx)
+	defer l.finished()
 	for {
 		select {
 		case event, ok := <-l.queue:
@@ -140,6 +150,8 @@ func (l *ChannelledLog) GetName() string {
 
 // MultiChannelledLog represents a cached channel to a LoggerProvider
 type MultiChannelledLog struct {
+	ctx             context.Context
+	finished        context.CancelFunc
 	name            string
 	bufferLength    int64
 	queue           chan *Event
@@ -156,7 +168,11 @@ type MultiChannelledLog struct {
 
 // NewMultiChannelledLog a new logger instance with given logger provider and config.
 func NewMultiChannelledLog(name string, bufferLength int64) *MultiChannelledLog {
+	ctx, _, finished := process.GetManager().AddTypedContext(context.Background(), fmt.Sprintf("Logger: %s", name), process.SystemProcessType, false)
+
 	m := &MultiChannelledLog{
+		ctx:             ctx,
+		finished:        finished,
 		name:            name,
 		queue:           make(chan *Event, bufferLength),
 		flush:           make(chan bool),
@@ -277,6 +293,9 @@ func (m *MultiChannelledLog) Start() {
 		m.rwmutex.Unlock()
 		return
 	}
+	pprof.SetGoroutineLabels(m.ctx)
+	defer m.finished()
+
 	m.started = true
 	m.rwmutex.Unlock()
 	paused := false

modules/log/multichannel.go

@@ -31,7 +31,7 @@ func newLogger(name string, buffer int64) *MultiChannelledLogger {
 
 // SetLogger sets new logger instance with given logger provider and config.
 func (l *MultiChannelledLogger) SetLogger(name, provider, config string) error {
-	eventLogger, err := NewChannelledLog(name, provider, config, l.bufferLength)
+	eventLogger, err := NewChannelledLog(l.ctx, name, provider, config, l.bufferLength)
 	if err != nil {
 		return fmt.Errorf("Failed to create sublogger (%s): %v", name, err)
 	}

modules/nosql/manager.go

@@ -5,10 +5,12 @@
 package nosql
 
 import (
+	"context"
 	"strconv"
 	"sync"
 	"time"
 
+	"code.gitea.io/gitea/modules/process"
 	"github.com/go-redis/redis/v8"
 	"github.com/syndtr/goleveldb/leveldb"
 )
@@ -17,6 +19,8 @@ var manager *Manager
 
 // Manager is the nosql connection manager
 type Manager struct {
+	ctx      context.Context
+	finished context.CancelFunc
 	mutex    sync.Mutex
 
 	RedisConnections   map[string]*redisClientHolder
@@ -46,7 +50,10 @@ func init() {
 // GetManager returns a Manager and initializes one as singleton is there's none yet
 func GetManager() *Manager {
 	if manager == nil {
+		ctx, _, finished := process.GetManager().AddTypedContext(context.Background(), "Service: NoSQL", process.SystemProcessType, false)
 		manager = &Manager{
+			ctx:                ctx,
+			finished:           finished,
 			RedisConnections:   make(map[string]*redisClientHolder),
 			LevelDBConnections: make(map[string]*levelDBHolder),
 		}

modules/nosql/manager_leveldb.go

@@ -7,6 +7,7 @@ package nosql
 import (
 	"fmt"
 	"path"
+	"runtime/pprof"
 	"strconv"
 	"strings"
 
@@ -50,7 +51,31 @@ func (m *Manager) CloseLevelDB(connection string) error {
 }
 
 // GetLevelDB gets a levelDB for a particular connection
-func (m *Manager) GetLevelDB(connection string) (*leveldb.DB, error) {
+func (m *Manager) GetLevelDB(connection string) (db *leveldb.DB, err error) {
+	// Because we want associate any goroutines created by this call to the main nosqldb context we need to
+	// wrap this in a goroutine labelled with the nosqldb context
+	done := make(chan struct{})
+	var recovered interface{}
+	go func() {
+		defer func() {
+			recovered = recover()
+			if recovered != nil {
+				log.Critical("PANIC during GetLevelDB: %v\nStacktrace: %s", recovered, log.Stack(2))
+			}
+			close(done)
+		}()
+		pprof.SetGoroutineLabels(m.ctx)
+
+		db, err = m.getLevelDB(connection)
+	}()
+	<-done
+	if recovered != nil {
+		panic(recovered)
+	}
+	return
+}
+
+func (m *Manager) getLevelDB(connection string) (*leveldb.DB, error) {
 	// Convert the provided connection description to the common format
 	uri := ToLevelDBURI(connection)
 
@@ -168,15 +193,18 @@ func (m *Manager) GetLevelDB(connection string) (*leveldb.DB, error) {
 	if err != nil {
 		if !errors.IsCorrupted(err) {
 			if strings.Contains(err.Error(), "resource temporarily unavailable") {
-				return nil, fmt.Errorf("unable to lock level db at %s: %w", dataDir, err)
-			}
-
-			return nil, fmt.Errorf("unable to open level db at %s: %w", dataDir, err)
-		}
-		db.db, err = leveldb.RecoverFile(dataDir, opts)
-		if err != nil {
+				err = fmt.Errorf("unable to lock level db at %s: %w", dataDir, err)
 				return nil, err
 			}
+
+			err = fmt.Errorf("unable to open level db at %s: %w", dataDir, err)
+			return nil, err
+		}
+		db.db, err = leveldb.RecoverFile(dataDir, opts)
+	}
+
+	if err != nil {
+		return nil, err
 	}
 
 	for _, name := range db.name {

modules/nosql/manager_redis.go

@@ -8,6 +8,7 @@ import (
 	"crypto/tls"
 	"net/url"
 	"path"
+	"runtime/pprof"
 	"strconv"
 	"strings"
 
@@ -43,7 +44,31 @@ func (m *Manager) CloseRedisClient(connection string) error {
 }
 
 // GetRedisClient gets a redis client for a particular connection
-func (m *Manager) GetRedisClient(connection string) redis.UniversalClient {
+func (m *Manager) GetRedisClient(connection string) (client redis.UniversalClient) {
+	// Because we want associate any goroutines created by this call to the main nosqldb context we need to
+	// wrap this in a goroutine labelled with the nosqldb context
+	done := make(chan struct{})
+	var recovered interface{}
+	go func() {
+		defer func() {
+			recovered = recover()
+			if recovered != nil {
+				log.Critical("PANIC during GetRedisClient: %v\nStacktrace: %s", recovered, log.Stack(2))
+			}
+			close(done)
+		}()
+		pprof.SetGoroutineLabels(m.ctx)
+
+		client = m.getRedisClient(connection)
+	}()
+	<-done
+	if recovered != nil {
+		panic(recovered)
+	}
+	return
+}
+
+func (m *Manager) getRedisClient(connection string) redis.UniversalClient {
 	m.mutex.Lock()
 	defer m.mutex.Unlock()
 	client, ok := m.RedisConnections[connection]

modules/private/manager.go

@@ -7,6 +7,7 @@ package private
 import (
 	"context"
 	"fmt"
+	"io"
 	"net/http"
 	"net/url"
 	"time"
@@ -189,3 +190,25 @@ func RemoveLogger(ctx context.Context, group, name string) (int, string) {
 
 	return http.StatusOK, "Removed"
 }
+
+// Processes return the current processes from this gitea instance
+func Processes(ctx context.Context, out io.Writer, flat, noSystem, stacktraces, json bool, cancel string) (int, string) {
+	reqURL := setting.LocalURL + fmt.Sprintf("api/internal/manager/processes?flat=%t&no-system=%t&stacktraces=%t&json=%t&cancel-pid=%s", flat, noSystem, stacktraces, json, url.QueryEscape(cancel))
+
+	req := newInternalRequest(ctx, reqURL, "GET")
+	resp, err := req.Response()
+	if err != nil {
+		return http.StatusInternalServerError, fmt.Sprintf("Unable to contact gitea: %v", err.Error())
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return resp.StatusCode, decodeJSONError(resp).Err
+	}
+
+	_, err = io.Copy(out, resp.Body)
+	if err != nil {
+		return http.StatusInternalServerError, err.Error()
+	}
+	return http.StatusOK, ""
+}

modules/process/error.go

@@ -0,0 +1,26 @@
+// Copyright 2022 The Gitea Authors. All rights reserved.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package process
+
+import "fmt"
+
+// Error is a wrapped error describing the error results of Process Execution
+type Error struct {
+	PID         IDType
+	Description string
+	Err         error
+	CtxErr      error
+	Stdout      string
+	Stderr      string
+}
+
+func (err *Error) Error() string {
+	return fmt.Sprintf("exec(%s:%s) failed: %v(%v) stdout: %s stderr: %s", err.PID, err.Description, err.Err, err.CtxErr, err.Stdout, err.Stderr)
+}
+
+// Unwrap implements the unwrappable implicit interface for go1.13 Unwrap()
+func (err *Error) Unwrap() error {
+	return err.Err
+}

modules/process/manager_exec.go

@@ -0,0 +1,79 @@
+// Copyright 2022 The Gitea Authors. All rights reserved.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package process
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"os/exec"
+	"time"
+)
+
+// Exec a command and use the default timeout.
+func (pm *Manager) Exec(desc, cmdName string, args ...string) (string, string, error) {
+	return pm.ExecDir(DefaultContext, -1, "", desc, cmdName, args...)
+}
+
+// ExecTimeout a command and use a specific timeout duration.
+func (pm *Manager) ExecTimeout(timeout time.Duration, desc, cmdName string, args ...string) (string, string, error) {
+	return pm.ExecDir(DefaultContext, timeout, "", desc, cmdName, args...)
+}
+
+// ExecDir a command and use the default timeout.
+func (pm *Manager) ExecDir(ctx context.Context, timeout time.Duration, dir, desc, cmdName string, args ...string) (string, string, error) {
+	return pm.ExecDirEnv(ctx, timeout, dir, desc, nil, cmdName, args...)
+}
+
+// ExecDirEnv runs a command in given path and environment variables, and waits for its completion
+// up to the given timeout (or DefaultTimeout if -1 is given).
+// Returns its complete stdout and stderr
+// outputs and an error, if any (including timeout)
+func (pm *Manager) ExecDirEnv(ctx context.Context, timeout time.Duration, dir, desc string, env []string, cmdName string, args ...string) (string, string, error) {
+	return pm.ExecDirEnvStdIn(ctx, timeout, dir, desc, env, nil, cmdName, args...)
+}
+
+// ExecDirEnvStdIn runs a command in given path and environment variables with provided stdIN, and waits for its completion
+// up to the given timeout (or DefaultTimeout if timeout <= 0 is given).
+// Returns its complete stdout and stderr
+// outputs and an error, if any (including timeout)
+func (pm *Manager) ExecDirEnvStdIn(ctx context.Context, timeout time.Duration, dir, desc string, env []string, stdIn io.Reader, cmdName string, args ...string) (string, string, error) {
+	if timeout <= 0 {
+		timeout = 60 * time.Second
+	}
+
+	stdOut := new(bytes.Buffer)
+	stdErr := new(bytes.Buffer)
+
+	ctx, _, finished := pm.AddContextTimeout(ctx, timeout, desc)
+	defer finished()
+
+	cmd := exec.CommandContext(ctx, cmdName, args...)
+	cmd.Dir = dir
+	cmd.Env = env
+	cmd.Stdout = stdOut
+	cmd.Stderr = stdErr
+	if stdIn != nil {
+		cmd.Stdin = stdIn
+	}
+
+	if err := cmd.Start(); err != nil {
+		return "", "", err
+	}
+
+	err := cmd.Wait()
+	if err != nil {
+		err = &Error{
+			PID:         GetPID(ctx),
+			Description: desc,
+			Err:         err,
+			CtxErr:      ctx.Err(),
+			Stdout:      stdOut.String(),
+			Stderr:      stdErr.String(),
+		}
+	}
+
+	return stdOut.String(), stdErr.String(), err
+}

modules/process/manager_test.go

@@ -22,7 +22,7 @@ func TestGetManager(t *testing.T) {
 }
 
 func TestManager_AddContext(t *testing.T) {
-	pm := Manager{processes: make(map[IDType]*Process), next: 1}
+	pm := Manager{processMap: make(map[IDType]*process), next: 1}
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -41,7 +41,7 @@ func TestManager_AddContext(t *testing.T) {
 }
 
 func TestManager_Cancel(t *testing.T) {
-	pm := Manager{processes: make(map[IDType]*Process), next: 1}
+	pm := Manager{processMap: make(map[IDType]*process), next: 1}
 
 	ctx, _, finished := pm.AddContext(context.Background(), "foo")
 	defer finished()
@@ -69,7 +69,7 @@ func TestManager_Cancel(t *testing.T) {
 }
 
 func TestManager_Remove(t *testing.T) {
-	pm := Manager{processes: make(map[IDType]*Process), next: 1}
+	pm := Manager{processMap: make(map[IDType]*process), next: 1}
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -85,7 +85,7 @@ func TestManager_Remove(t *testing.T) {
 
 	pm.Remove(GetPID(p2Ctx))
 
-	_, exists := pm.processes[GetPID(p2Ctx)]
+	_, exists := pm.processMap[GetPID(p2Ctx)]
 	assert.False(t, exists, "PID %d is in the list but shouldn't", GetPID(p2Ctx))
 }
 

modules/process/process.go

@@ -6,61 +6,34 @@ package process
 
 import (
 	"context"
-	"sync"
 	"time"
 )
 
-// Process represents a working process inheriting from Gitea.
-type Process struct {
+var (
+	SystemProcessType  = "system"
+	RequestProcessType = "request"
+	NormalProcessType  = "normal"
+	NoneProcessType    = "none"
+)
+
+// process represents a working process inheriting from Gitea.
+type process struct {
 	PID         IDType // Process ID, not system one.
 	ParentPID   IDType
 	Description string
 	Start       time.Time
 	Cancel      context.CancelFunc
-
-	lock     sync.Mutex
-	children []*Process
+	Type        string
 }
 
-// Children gets the children of the process
-// Note: this function will behave nicely even if p is nil
-func (p *Process) Children() (children []*Process) {
-	if p == nil {
-		return
-	}
-
-	p.lock.Lock()
-	defer p.lock.Unlock()
-	children = make([]*Process, len(p.children))
-	copy(children, p.children)
-	return children
-}
-
-// AddChild adds a child process
-// Note: this function will behave nicely even if p is nil
-func (p *Process) AddChild(child *Process) {
-	if p == nil {
-		return
-	}
-
-	p.lock.Lock()
-	defer p.lock.Unlock()
-	p.children = append(p.children, child)
-}
-
-// RemoveChild removes a child process
-// Note: this function will behave nicely even if p is nil
-func (p *Process) RemoveChild(process *Process) {
-	if p == nil {
-		return
-	}
-
-	p.lock.Lock()
-	defer p.lock.Unlock()
-	for i, child := range p.children {
-		if child == process {
-			p.children = append(p.children[:i], p.children[i+1:]...)
-			return
-		}
+// ToProcess converts a process to a externally usable Process
+func (p *process) toProcess() *Process {
+	process := &Process{
+		PID:         p.PID,
+		ParentPID:   p.ParentPID,
+		Description: p.Description,
+		Start:       p.Start,
+		Type:        p.Type,
 	}
+	return process
 }

modules/queue/queue_bytefifo.go

@@ -7,6 +7,7 @@ package queue
 import (
 	"context"
 	"fmt"
+	"runtime/pprof"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -20,7 +21,6 @@ import (
 type ByteFIFOQueueConfiguration struct {
 	WorkerPoolConfiguration
 	Workers     int
-	Name        string
 	WaitOnEmpty bool
 }
 
@@ -153,6 +153,7 @@ func (q *ByteFIFOQueue) Flush(timeout time.Duration) error {
 
 // Run runs the bytefifo queue
 func (q *ByteFIFOQueue) Run(atShutdown, atTerminate func(func())) {
+	pprof.SetGoroutineLabels(q.baseCtx)
 	atShutdown(q.Shutdown)
 	atTerminate(q.Terminate)
 	log.Debug("%s: %s Starting", q.typ, q.name)
@@ -355,6 +356,7 @@ func (q *ByteFIFOQueue) Terminate() {
 	if err := q.byteFIFO.Close(); err != nil {
 		log.Error("Error whilst closing internal byte fifo in %s: %s: %v", q.typ, q.name, err)
 	}
+	q.baseCtxFinished()
 	log.Debug("%s: %s Terminated", q.typ, q.name)
 }
 

modules/queue/queue_channel.go

@@ -7,6 +7,7 @@ package queue
 import (
 	"context"
 	"fmt"
+	"runtime/pprof"
 	"sync/atomic"
 	"time"
 
@@ -20,7 +21,6 @@ const ChannelQueueType Type = "channel"
 type ChannelQueueConfiguration struct {
 	WorkerPoolConfiguration
 	Workers int
-	Name    string
 }
 
 // ChannelQueue implements Queue
@@ -84,6 +84,7 @@ func NewChannelQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue, erro
 
 // Run starts to run the queue
 func (q *ChannelQueue) Run(atShutdown, atTerminate func(func())) {
+	pprof.SetGoroutineLabels(q.baseCtx)
 	atShutdown(q.Shutdown)
 	atTerminate(q.Terminate)
 	log.Debug("ChannelQueue: %s Starting", q.name)
@@ -169,6 +170,7 @@ func (q *ChannelQueue) Terminate() {
 	default:
 	}
 	q.terminateCtxCancel()
+	q.baseCtxFinished()
 	log.Debug("ChannelQueue: %s Terminated", q.name)
 }
 

modules/queue/queue_channel_test.go

@@ -34,9 +34,9 @@ func TestChannelQueue(t *testing.T) {
 				BlockTimeout: 1 * time.Second,
 				BoostTimeout: 5 * time.Minute,
 				BoostWorkers: 5,
+				Name:         "TestChannelQueue",
 			},
 			Workers: 0,
-			Name:    "TestChannelQueue",
 		}, &testData{})
 	assert.NoError(t, err)
 

modules/queue/queue_disk_channel.go

@@ -7,6 +7,7 @@ package queue
 import (
 	"context"
 	"fmt"
+	"runtime/pprof"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -72,9 +73,9 @@ func NewPersistableChannelQueue(handle HandlerFunc, cfg, exemplar interface{}) (
 			BoostTimeout: config.BoostTimeout,
 			BoostWorkers: config.BoostWorkers,
 			MaxWorkers:   config.MaxWorkers,
+			Name:         config.Name + "-channel",
 		},
 		Workers: config.Workers,
-		Name:    config.Name + "-channel",
 	}, exemplar)
 	if err != nil {
 		return nil, err
@@ -90,9 +91,9 @@ func NewPersistableChannelQueue(handle HandlerFunc, cfg, exemplar interface{}) (
 				BoostTimeout: 5 * time.Minute,
 				BoostWorkers: 1,
 				MaxWorkers:   5,
+				Name:         config.Name + "-level",
 			},
 			Workers: 0,
-			Name:    config.Name + "-level",
 		},
 		DataDir: config.DataDir,
 	}
@@ -154,6 +155,7 @@ func (q *PersistableChannelQueue) PushBack(data Data) error {
 
 // Run starts to run the queue
 func (q *PersistableChannelQueue) Run(atShutdown, atTerminate func(func())) {
+	pprof.SetGoroutineLabels(q.channelQueue.baseCtx)
 	log.Debug("PersistableChannelQueue: %s Starting", q.delayedStarter.name)
 	_ = q.channelQueue.AddWorkers(q.channelQueue.workers, 0)
 

modules/queue/unique_queue_channel.go

@@ -7,6 +7,7 @@ package queue
 import (
 	"context"
 	"fmt"
+	"runtime/pprof"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -97,6 +98,7 @@ func NewChannelUniqueQueue(handle HandlerFunc, cfg, exemplar interface{}) (Queue
 
 // Run starts to run the queue
 func (q *ChannelUniqueQueue) Run(atShutdown, atTerminate func(func())) {
+	pprof.SetGoroutineLabels(q.baseCtx)
 	atShutdown(q.Shutdown)
 	atTerminate(q.Terminate)
 	log.Debug("ChannelUniqueQueue: %s Starting", q.name)
@@ -226,6 +228,7 @@ func (q *ChannelUniqueQueue) Terminate() {
 	default:
 	}
 	q.terminateCtxCancel()
+	q.baseCtxFinished()
 	log.Debug("ChannelUniqueQueue: %s Terminated", q.name)
 }
 

modules/queue/unique_queue_channel_test.go

@@ -32,9 +32,9 @@ func TestChannelUniqueQueue(t *testing.T) {
 				BlockTimeout: 1 * time.Second,
 				BoostTimeout: 5 * time.Minute,
 				BoostWorkers: 5,
+				Name:         "TestChannelQueue",
 			},
 			Workers: 0,
-			Name:    "TestChannelQueue",
 		}, &testData{})
 	assert.NoError(t, err)