peer/http3/worker.go

@@ -20,7 +20,9 @@ var getReqCh = make(chan *blobRequest, 20000)
 func InitWorkers(server *Server, workers int) {
 	stopper := stop.New(server.grp)
 	for i := 0; i < workers; i++ {
+		metrics.RoutinesQueue.WithLabelValues("http3", "worker").Inc()
 		go func(worker int) {
+			defer metrics.RoutinesQueue.WithLabelValues("http3", "worker").Dec()
 			for {
 				select {
 				case <-stopper.Ch():

peer/server.go

@@ -89,7 +89,9 @@ func (s *Server) listenAndServe(listener net.Listener) {
 			log.Error(errors.Prefix("accepting conn", err))
 		} else {
 			s.grp.Add(1)
+			metrics.RoutinesQueue.WithLabelValues("peer", "server-handleconn").Inc()
 			go func() {
+				defer metrics.RoutinesQueue.WithLabelValues("peer", "server-handleconn").Dec()
 				s.handleConnection(conn)
 				s.grp.Done()
 			}()

store/caching.go

@@ -66,7 +66,9 @@ func (c *CachingStore) Get(hash string) (stream.Blob, shared.BlobTrace, error) {
 	}
 	// there is no need to wait for the blob to be stored before we return it
 	// TODO: however this should be refactored to limit the amount of routines that the process can spawn to avoid a possible DoS
+	metrics.RoutinesQueue.WithLabelValues("store", "cache-put").Inc()
 	go func() {
+		defer metrics.RoutinesQueue.WithLabelValues("store", "cache-put").Dec()
 		err = c.cache.Put(hash, blob)
 		if err != nil {
 			log.Errorf("error saving blob to underlying cache: %s", errors.FullTrace(err))

store/speedwalk/speedwalk.go

@@ -6,6 +6,8 @@ import (
 	"runtime"
 	"sync"
 
+	"github.com/lbryio/reflector.go/internal/metrics"
+
 	"github.com/lbryio/lbry.go/v2/extras/errors"
 
 	"github.com/karrick/godirwalk"
@@ -24,6 +26,7 @@ func AllFiles(startDir string, basename bool) ([]string, error) {
 	paths := make([]string, 0, 1000)
 	pathWG := &sync.WaitGroup{}
 	pathWG.Add(1)
+	metrics.RoutinesQueue.WithLabelValues("speedwalk", "worker").Inc()
 	go func() {
 		defer pathWG.Done()
 		for {