Add skeleton database_observability.postgres component (#3945)

This adds a new database observability component for postgres. The purpose of this component will be to collect postgres query analysis data; replicating that of the mysql component.

The required collectors for that functionality will be built over time. This change introduces the initial component so that the collectors can subsequently be added.

The most basic connection_info collector is added, so we can begin sending the database_observability_connection_info metric for pg databases.

A docs page is added for the new component but the component README is not updated to reference postgres things, while we work out the small details on configuring the DB etc.

The changelog isn't updated to include this new addition, as we don't expect adoption of this component yet - however happy to add an entry if it makes sense to do so.

Author: matthewnolf

docs/sources/reference/compatibility/_index.md

@@ -46,6 +46,7 @@ The following components, grouped by namespace, _export_ Targets.
 
 {{< collapse title="database_observability" >}}
 - [database_observability.mysql](../components/database_observability/database_observability.mysql)
+- [database_observability.postgres](../components/database_observability/database_observability.postgres)
 {{< /collapse >}}
 
 {{< collapse title="discovery" >}}
@@ -245,6 +246,7 @@ The following components, grouped by namespace, _consume_ Loki `LogsReceiver`.
 
 {{< collapse title="database_observability" >}}
 - [database_observability.mysql](../components/database_observability/database_observability.mysql)
+- [database_observability.postgres](../components/database_observability/database_observability.postgres)
 {{< /collapse >}}
 
 {{< collapse title="faro" >}}

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

@@ -0,0 +1,99 @@
+---
+canonical: https://grafana.com/docs/alloy/latest/reference/components/database_observability.postgres/
+description: Learn about database_observability.postgres
+title: database_observability.postgres
+labels:
+  stage: experimental
+  products:
+    - oss
+---
+
+# `database_observability.postgres`
+
+{{< docs/shared lookup="stability/experimental.md" source="alloy" version="<ALLOY_VERSION>" >}}
+
+## Usage
+
+```alloy
+database_observability.postgres "<LABEL>" {
+  data_source_name = <DATA_SOURCE_NAME>
+  forward_to       = [<LOKI_RECEIVERS>]
+}
+```
+
+## Arguments
+
+You can use the following arguments with `database_observability.postgres`:
+
+| Name                               | Type                 | Description                                                                                    | Default | Required |
+|------------------------------------|----------------------|------------------------------------------------------------------------------------------------|---------|----------|
+| `data_source_name`                 | `secret`             | [Data Source Name][] for the Postgres server to connect to.                                       |         | yes      |
+| `forward_to`                       | `list(LogsReceiver)` | Where to forward log entries after processing.                                                 |         | yes      |
+
+## Blocks
+
+The `database_observability.postgres` component doesn't support any blocks. You can configure this component with arguments.
+
+## Example
+
+```alloy
+database_observability.postgres "orders_db" {
+  data_source_name = "postgres://user:pass@localhost:5432/mydb"
+  forward_to = [loki.write.logs_service.receiver]
+}
+
+prometheus.scrape "orders_db" {
+  targets = database_observability.postgres.orders_db.targets
+  honor_labels = true // required to keep job and instance labels
+  forward_to = [prometheus.remote_write.metrics_service.receiver]
+}
+
+prometheus.remote_write "metrics_service" {
+  endpoint {
+    url = sys.env("<GRAFANA_CLOUD_HOSTED_METRICS_URL>")
+    basic_auth {
+      username = sys.env("<GRAFANA_CLOUD_HOSTED_METRICS_ID>")
+      password = sys.env("<GRAFANA_CLOUD_RW_API_KEY>")
+    }
+  }
+}
+
+loki.write "logs_service" {
+  endpoint {
+    url = sys.env("<GRAFANA_CLOUD_HOSTED_LOGS_URL>")
+    basic_auth {
+      username = sys.env("<GRAFANA_CLOUD_HOSTED_LOGS_ID>")
+      password = sys.env("<GRAFANA_CLOUD_RW_API_KEY>")
+    }
+  }
+}
+```
+
+Replace the following:
+
+* _`<GRAFANA_CLOUD_HOSTED_METRICS_URL>`_: The URL for your Grafana Cloud hosted metrics.
+* _`<GRAFANA_CLOUD_HOSTED_METRICS_ID>`_: The user ID for your Grafana Cloud hosted metrics.
+* _`<GRAFANA_CLOUD_RW_API_KEY>`_: Your Grafana Cloud API key.
+* _`<GRAFANA_CLOUD_HOSTED_LOGS_URL>`_: The URL for your Grafana Cloud hosted logs.
+* _`<GRAFANA_CLOUD_HOSTED_LOGS_ID>`_: The user ID for your Grafana Cloud hosted logs.
+
+[Data Source Name]: https://pkg.go.dev/github.com/lib/pq#hdr-Connection_String_Parameters
+
+<!-- START GENERATED COMPATIBLE COMPONENTS -->
+
+## Compatible components
+
+`database_observability.postgres` can accept arguments from the following components:
+
+- Components that export [Loki `LogsReceiver`](../../../compatibility/#loki-logsreceiver-exporters)
+
+`database_observability.postgres` has exports that can be consumed by the following components:
+
+- Components that consume [Targets](../../../compatibility/#targets-consumers)
+
+{{< admonition type="note" >}}
+Connecting some components may not be sensible or components may require further configuration to make the connection work correctly.
+Refer to the linked documentation for more details.
+{{< /admonition >}}
+
+<!-- END GENERATED COMPATIBLE COMPONENTS -->

internal/component/all/all.go

@@ -4,6 +4,7 @@ package all
 import (
 	_ "github.com/grafana/alloy/internal/component/beyla/ebpf"                               // Import beyla.ebpf
 	_ "github.com/grafana/alloy/internal/component/database_observability/mysql"             // Import database_observability.mysql
+	_ "github.com/grafana/alloy/internal/component/database_observability/postgres"          // Import database_observability.postgres
 	_ "github.com/grafana/alloy/internal/component/discovery/aws"                            // Import discovery.aws.ec2 and discovery.aws.lightsail
 	_ "github.com/grafana/alloy/internal/component/discovery/azure"                          // Import discovery.azure
 	_ "github.com/grafana/alloy/internal/component/discovery/consul"                         // Import discovery.consul

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

@@ -0,0 +1,95 @@
+package collector
+
+import (
+	"context"
+	"regexp"
+	"strings"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"go.uber.org/atomic"
+)
+
+const ConnectionInfoName = "connection_info"
+
+var (
+	rdsRegex   = regexp.MustCompile(`(?P<identifier>[^\.]+)\.([^\.]+)\.(?P<region>[^\.]+)\.rds\.amazonaws\.com`)
+	azureRegex = regexp.MustCompile(`(?P<identifier>[^\.]+)\.postgres\.database\.azure\.com`)
+)
+
+type ConnectionInfoArguments struct {
+	DSN      string
+	Registry *prometheus.Registry
+}
+
+type ConnectionInfo struct {
+	DSN        string
+	Registry   *prometheus.Registry
+	InfoMetric *prometheus.GaugeVec
+
+	running *atomic.Bool
+}
+
+func NewConnectionInfo(args ConnectionInfoArguments) (*ConnectionInfo, error) {
+	infoMetric := prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: "database_observability",
+		Name:      "connection_info",
+		Help:      "Information about the connection",
+	}, []string{"provider_name", "provider_region", "db_instance_identifier", "engine"})
+
+	args.Registry.MustRegister(infoMetric)
+
+	return &ConnectionInfo{
+		DSN:        args.DSN,
+		Registry:   args.Registry,
+		InfoMetric: infoMetric,
+		running:    &atomic.Bool{},
+	}, nil
+}
+
+func (c *ConnectionInfo) Name() string {
+	return ConnectionInfoName
+}
+
+func (c *ConnectionInfo) Start(ctx context.Context) error {
+	c.running.Store(true)
+
+	var (
+		providerName         = "unknown"
+		providerRegion       = "unknown"
+		dbInstanceIdentifier = "unknown"
+		engine               = "postgres"
+	)
+
+	parts, err := ParseURL(c.DSN)
+	if err != nil {
+		return err
+	}
+
+	if host, ok := parts["host"]; ok {
+		if strings.HasSuffix(host, "rds.amazonaws.com") {
+			providerName = "aws"
+			matches := rdsRegex.FindStringSubmatch(host)
+			if len(matches) > 3 {
+				dbInstanceIdentifier = matches[1]
+				providerRegion = matches[3]
+			}
+		} else if strings.HasSuffix(host, "postgres.database.azure.com") {
+			providerName = "azure"
+			matches := azureRegex.FindStringSubmatch(host)
+			if len(matches) > 1 {
+				dbInstanceIdentifier = matches[1]
+			}
+		}
+	}
+	c.InfoMetric.WithLabelValues(providerName, providerRegion, dbInstanceIdentifier, engine).Set(1)
+	return nil
+}
+
+func (c *ConnectionInfo) Stopped() bool {
+	return !c.running.Load()
+}
+
+func (c *ConnectionInfo) Stop() {
+	c.Registry.Unregister(c.InfoMetric)
+	c.running.Store(false)
+}

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

@@ -0,0 +1,61 @@
+package collector
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+)
+
+func TestConnectionInfo(t *testing.T) {
+	defer goleak.VerifyNone(t)
+
+	const baseExpectedMetrics = `
+	# HELP database_observability_connection_info Information about the connection
+	# TYPE database_observability_connection_info gauge
+	database_observability_connection_info{db_instance_identifier="%s",engine="%s",provider_name="%s",provider_region="%s"} 1
+`
+
+	testCases := []struct {
+		name            string
+		dsn             string
+		expectedMetrics string
+	}{
+		{
+			name:            "generic dsn",
+			dsn:             "postgres://user:pass@localhost:5432/mydb",
+			expectedMetrics: fmt.Sprintf(baseExpectedMetrics, "unknown", "postgres", "unknown", "unknown"),
+		},
+		{
+			name:            "AWS/RDS dsn",
+			dsn:             "postgres://user:[email protected]:5432/mydb",
+			expectedMetrics: fmt.Sprintf(baseExpectedMetrics, "products-db", "postgres", "aws", "us-east-1"),
+		},
+		{
+			name:            "Azure flexibleservers dsn",
+			dsn:             "postgres://user:[email protected]:5432/mydb",
+			expectedMetrics: fmt.Sprintf(baseExpectedMetrics, "products-db", "postgres", "azure", "unknown"),
+		},
+	}
+
+	for _, tc := range testCases {
+		reg := prometheus.NewRegistry()
+
+		collector, err := NewConnectionInfo(ConnectionInfoArguments{
+			DSN:      tc.dsn,
+			Registry: reg,
+		})
+		require.NoError(t, err)
+		require.NotNil(t, collector)
+
+		err = collector.Start(t.Context())
+		require.NoError(t, err)
+
+		err = testutil.GatherAndCompare(reg, strings.NewReader(tc.expectedMetrics))
+		require.NoError(t, err)
+	}
+}

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

@@ -0,0 +1,43 @@
+package collector
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/lib/pq"
+)
+
+func ParseURL(url string) (map[string]string, error) {
+	if url == "postgresql://" || url == "postgres://" {
+		return map[string]string{}, nil
+	}
+
+	raw, err := pq.ParseURL(url)
+	if err != nil {
+		return nil, err
+	}
+
+	res := map[string]string{}
+
+	unescaper := strings.NewReplacer(`\'`, `'`, `\\`, `\`)
+
+	for keypair := range strings.SplitSeq(raw, " ") {
+		parts := strings.SplitN(keypair, "=", 2)
+		if len(parts) != 2 {
+			return nil, fmt.Errorf("unexpected keypair %s from pq", keypair)
+		}
+
+		key := parts[0]
+		value := parts[1]
+
+		// Undo all the transformations ParseURL did: remove wrapping
+		// quotes and then unescape the escaped characters.
+		value = strings.TrimPrefix(value, "'")
+		value = strings.TrimSuffix(value, "'")
+		value = unescaper.Replace(value)
+
+		res[key] = value
+	}
+
+	return res, nil
+}