database_observability: add caching to Postgres schema details collector (#4565)

Author: fridgepoet

docs/sources/reference/components/database_observability/database_observability.postgres.md

@@ -73,7 +73,13 @@ You can use the following blocks with `database_observability.postgres`:
 
 ### `schema_details`
 
-This collector has no config options.
+| Name               | Type       | Description                                                           | Default | Required |
+|--------------------|------------|-----------------------------------------------------------------------|---------|----------|
+| `collect_interval` | `duration` | How frequently to collect information from database.                  | `"1m"`  | no       |
+| `cache_enabled`    | `boolean`  | Whether to enable caching of table definitions.                       | `true`  | no       |
+| `cache_size`       | `integer`  | Cache size.                                                           | `256`   | no       |
+| `cache_ttl`        | `duration` | Cache TTL.                                                            | `"10m"` | no       |
+
 
 ### `explain_plans`
 

internal/component/database_observability/postgres/collector/schema_details.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/go-kit/log"
+	"github.com/hashicorp/golang-lru/v2/expirable"
 	"github.com/lib/pq"
 	"go.uber.org/atomic"
 
@@ -200,8 +201,13 @@ type foreignKey struct {
 }
 
 type SchemaDetailsArguments struct {
-	DB           *sql.DB
-	EntryHandler loki.EntryHandler
+	DB              *sql.DB
+	CollectInterval time.Duration
+	EntryHandler    loki.EntryHandler
+
+	CacheEnabled bool
+	CacheSize    int
+	CacheTTL     time.Duration
 
 	Logger log.Logger
 }
@@ -211,6 +217,11 @@ type SchemaDetails struct {
 	collectInterval time.Duration
 	entryHandler    loki.EntryHandler
 
+	// Cache of table definitions. Entries are removed after a configurable TTL.
+	// Key is a string of the form "database.schema.table".
+	// (unlike MySQL) no create/update timestamp available for detecting immediately when a table schema is changed; relying on TTL only
+	cache *expirable.LRU[string, *tableInfo]
+
 	logger  log.Logger
 	running *atomic.Bool
 	ctx     context.Context
@@ -220,12 +231,16 @@ type SchemaDetails struct {
 func NewSchemaDetails(args SchemaDetailsArguments) (*SchemaDetails, error) {
 	c := &SchemaDetails{
 		dbConnection:    args.DB,
-		collectInterval: 10 * time.Minute, // TODO: make it configurable again once caching is implemented
+		collectInterval: args.CollectInterval,
 		entryHandler:    args.EntryHandler,
 		logger:          log.With(args.Logger, "collector", SchemaDetailsCollector),
 		running:         &atomic.Bool{},
 	}
 
+	if args.CacheEnabled {
+		c.cache = expirable.NewLRU[string, *tableInfo](args.CacheSize, nil, args.CacheTTL)
+	}
+
 	return c, nil
 }
 
@@ -354,10 +369,25 @@ func (c *SchemaDetails) extractNames(ctx context.Context) error {
 	}
 
 	for _, table := range tables {
-		table, err = c.fetchTableDefinitions(ctx, table)
-		if err != nil {
-			level.Error(c.logger).Log("msg", "failed to get table definitions", "datname", dbName, "schema", table.schema, "err", err)
-			continue
+		cacheKey := fmt.Sprintf("%s.%s.%s", table.database, table.schema, table.tableName)
+
+		cacheHit := false
+		if c.cache != nil {
+			if cached, ok := c.cache.Get(cacheKey); ok {
+				table = cached
+				cacheHit = true
+			}
+		}
+
+		if !cacheHit {
+			table, err = c.fetchTableDefinitions(ctx, table)
+			if err != nil {
+				level.Error(c.logger).Log("msg", "failed to get table definitions", "datname", dbName, "schema", table.schema, "err", err)
+				continue
+			}
+			if c.cache != nil {
+				c.cache.Add(cacheKey, table)
+			}
 		}
 
 		c.entryHandler.Chan() <- database_observability.BuildLokiEntry(