[+] implement parallel source discovery

Improves dead-source handling with parallel resolution and instance_up=0
on discovery failure.

`Sources.ResolveDatabases()` previously resolved each source
sequentially. A single slow or unresponsive source
(e.g. a continuous-discovery endpoint behind a firewall) would block
discovery of all subsequent sources for the full connection timeout
duration.

Sources are now resolved concurrently using `sync.WaitGroup.Go()`.
Results are collected into a pre-allocated indexed slice to preserve
deterministic ordering. Per-source error logging with source name is
included in the resolver itself.

When a `SourcePostgresContinuous` or `SourcePatroni` source fails to
resolve any databases, `LoadSources()` now emits `instance_up=0` to
the configured sinks. This makes the failure visible in dashboards and
alerting, consistent with how unreachable directly-monitored sources are
handled.

Author: pashagolub

internal/reaper/reaper.go

@@ -407,8 +407,19 @@ func (r *Reaper) LoadSources(ctx context.Context) (err error) {
 	srcs = slices.DeleteFunc(srcs, func(s sources.Source) bool {
 		return !s.IsEnabled || len(r.Sources.Groups) > 0 && !slices.Contains(r.Sources.Groups, s.Group)
 	})
-	if newSrcs, err = srcs.ResolveDatabases(); err != nil {
+	newSrcs, err = srcs.ResolveDatabases()
+	if err != nil {
 		r.logger.WithError(err).Error("could not resolve databases from sources")
+		for _, s := range srcs {
+			if s.Kind != sources.SourcePostgresContinuous && s.Kind != sources.SourcePatroni {
+				continue
+			}
+			if !slices.ContainsFunc(newSrcs, func(sc *sources.SourceConn) bool {
+				return sc.Name == s.Name || strings.HasPrefix(sc.Name, s.Name+"_")
+			}) {
+				r.WriteInstanceDown(sources.NewSourceConn(s))
+			}
+		}
 	}
 
 	for i, newMD := range newSrcs {

internal/sources/resolver.go

@@ -14,6 +14,7 @@ import (
 	"net/url"
 	"os"
 	"strings"
+	"sync"
 	"time"
 
 	jsoniter "github.com/json-iterator/go"
@@ -25,16 +26,29 @@ import (
 	"go.uber.org/zap"
 )
 
-// ResolveDatabases() updates list of monitored objects from continuous monitoring sources, e.g. patroni
+// ResolveDatabases() updates list of monitored objects from continuous monitoring sources, e.g. patroni.
+// Each source is resolved concurrently so that a slow or unreachable source does not block the others.
 func (srcs Sources) ResolveDatabases() (_ SourceConns, err error) {
+	type result struct {
+		dbs SourceConns
+		err error
+	}
+	results := make([]result, len(srcs))
+	var wg sync.WaitGroup
+	for i, s := range srcs {
+		wg.Go(func() {
+			dbs, e := s.ResolveDatabases()
+			results[i] = result{dbs, e}
+		})
+	}
+	wg.Wait()
 	resolvedDbs := make(SourceConns, 0, len(srcs))
-	for _, s := range srcs {
-		if !s.IsEnabled {
-			continue
+	for i, res := range results {
+		if res.err != nil {
+			logger.WithField("source", srcs[i].Name).WithError(res.err).Error("could not resolve databases from source")
+			err = errors.Join(err, res.err)
 		}
-		dbs, e := s.ResolveDatabases()
-		err = errors.Join(err, e)
-		resolvedDbs = append(resolvedDbs, dbs...)
+		resolvedDbs = append(resolvedDbs, res.dbs...)
 	}
 	return resolvedDbs, err
 }