feat: flush metrics in separate goroutine (#2424)

pkg/topology/kademlia/kademlia.go

@@ -506,18 +506,33 @@ func (k *Kad) manage() {
 	peerConnChan2 := make(chan *peerConnInfo)
 	go k.connectionAttemptsHandler(ctx, &wg, peerConnChan, peerConnChan2)
 
+	k.wg.Add(1)
+	go func() {
+		defer k.wg.Done()
 		for {
 			select {
+			case <-k.halt:
+				return
 			case <-k.quit:
 				return
-		case <-time.After(15 * time.Second):
+			case <-time.After(5 * time.Minute):
 				start := time.Now()
 				if err := k.collector.Flush(); err != nil {
 					k.metrics.InternalMetricsFlushTotalErrors.Inc()
-				k.logger.Debugf("kademlia: unable to flush metrics counters to the persistent store: %v", err)
+					k.logger.Debugf("kademlia: took %s unable to flush metrics counters to the persistent store: %v", time.Since(start), err)
 				} else {
 					k.metrics.InternalMetricsFlushTime.Observe(float64(time.Since(start).Nanoseconds()))
+					k.logger.Tracef("kademlia took %s to flush", time.Since(start))
+				}
+			}
 		}
+	}()
+
+	for {
+		select {
+		case <-k.quit:
+			return
+		case <-time.After(15 * time.Second):
 			k.notifyManageLoop()
 		case <-k.manageC:
 			start := time.Now()