PostgreSQL state store v2 (dapr#3250)

Signed-off-by: ItalyPaleAle <[email protected]>
Co-authored-by: Bernd Verst <[email protected]>

Author: ItalyPaleAle

.build-tools/component-folders.yaml

@@ -40,4 +40,5 @@ excludeFolders:
   - state/gcp
   - state/hashicorp
   - state/oci
+  - state/postgresql
   - state/utils

.github/scripts/test-info.mjs

@@ -615,14 +615,17 @@ const components = {
         conformanceSetup: 'conformance-state.cloudflare.workerskv-setup.sh',
         conformanceDestroy: 'conformance-state.cloudflare.workerskv-destroy.sh',
     },
-    'state.cockroachdb': {
+    'state.cockroachdb.v1': {
         conformance: true,
         certification: true,
         conformanceSetup: 'docker-compose.sh cockroachdb',
         sourcePkg: [
             'state/cockroachdb',
-            'common/component/postgresql',
+            'common/component/postgresql/interfaces',
+            'common/component/postgresql/transactions',
+            'common/component/postgresql/v1',
             'common/component/sql',
+            'common/component/sql/migrations',
         ],
     },
     'state.etcd.v1': {
@@ -664,26 +667,32 @@ const components = {
         conformance: true,
         conformanceSetup: 'docker-compose.sh oracledatabase',
     },
-    'state.postgresql': {
+    'state.postgresql.v1': {
         certification: true,
         sourcePkg: [
-            'state/postgresql',
+            'state/postgresql/v1',
             'common/authentication/postgresql',
-            'common/component/postgresql',
+            'common/component/postgresql/interfaces',
+            'common/component/postgresql/transactions',
+            'common/component/postgresql/v1',
             'common/component/sql',
+            'common/component/sql/migrations',
         ],
     },
-    'state.postgresql.docker': {
+    'state.postgresql.v1.docker': {
         conformance: true,
         conformanceSetup: 'docker-compose.sh postgresql',
         sourcePkg: [
-            'state/postgresql',
+            'state/postgresql/v1',
             'common/authentication/postgresql',
-            'common/component/postgresql',
+            'common/component/postgresql/interfaces',
+            'common/component/postgresql/transactions',
+            'common/component/postgresql/v1',
             'common/component/sql',
+            'common/component/sql/migrations',
         ],
     },
-    'state.postgresql.azure': {
+    'state.postgresql.v1.azure': {
         conformance: true,
         requiredSecrets: [
             'AzureDBPostgresConnectionString',
@@ -692,10 +701,53 @@ const components = {
             'AzureDBPostgresTenantId',
         ],
         sourcePkg: [
-            'state/postgresql',
+            'state/postgresql/v1',
             'common/authentication/postgresql',
-            'common/component/postgresql',
+            'common/component/postgresql/interfaces',
+            'common/component/postgresql/transactions',
+            'common/component/postgresql/v1',
             'common/component/sql',
+            'common/component/sql/migrations',
+        ],
+    },
+    'state.postgresql.v2': {
+        certification: true,
+        sourcePkg: [
+            'state/postgresql/v2',
+            'common/authentication/postgresql',
+            'common/component/postgresql/interfaces',
+            'common/component/postgresql/transactions',
+            'common/component/sql',
+            'common/component/sql/migrations',
+        ],
+    },
+    'state.postgresql.v2.docker': {
+        conformance: true,
+        conformanceSetup: 'docker-compose.sh postgresql',
+        sourcePkg: [
+            'state/postgresql/v2',
+            'common/authentication/postgresql',
+            'common/component/postgresql/interfaces',
+            'common/component/postgresql/transactions',
+            'common/component/sql',
+            'common/component/sql/migrations',
+        ],
+    },
+    'state.postgresql.v2.azure': {
+        conformance: true,
+        requiredSecrets: [
+            'AzureDBPostgresConnectionString',
+            'AzureDBPostgresClientId',
+            'AzureDBPostgresClientSecret',
+            'AzureDBPostgresTenantId',
+        ],
+        sourcePkg: [
+            'state/postgresql/v2',
+            'common/authentication/postgresql',
+            'common/component/postgresql/interfaces',
+            'common/component/postgresql/transactions',
+            'common/component/sql',
+            'common/component/sql/migrations',
         ],
     },
     'state.redis': {

bindings/postgres/metadata.go

@@ -20,27 +20,17 @@ import (
 
 type psqlMetadata struct {
 	pgauth.PostgresAuthMetadata `mapstructure:",squash"`
-
-	// URL is the connection string to connect to the database.
-	// Deprecated alias: use connectionString instead.
-	URL string `mapstructure:"url"`
 }
 
 func (m *psqlMetadata) InitWithMetadata(meta map[string]string) error {
 	// Reset the object
 	m.PostgresAuthMetadata.Reset()
-	m.URL = ""
 
 	err := kitmd.DecodeMetadata(meta, &m)
 	if err != nil {
 		return err
 	}
 
-	// Legacy options
-	if m.ConnectionString == "" && m.URL != "" {
-		m.ConnectionString = m.URL
-	}
-
 	// Validate and sanitize input
 	// Azure AD auth is supported for this component
 	err = m.PostgresAuthMetadata.InitWithMetadata(meta, true)

bindings/postgres/metadata.yaml

@@ -70,15 +70,6 @@ metadata:
       database driver to choose.
     example:  "5m"
     type: duration
-  - name: url
-    deprecated: true
-    required: false
-    description: |
-      Deprecated alias for "connectionString"
-    type: string
-    sensitive: true
-    example: |
-      "user=dapr password=secret host=dapr.example.com port=5432 dbname=dapr sslmode=verify-ca"
   - name: queryExecMode
     required: false
     description: |

common/authentication/postgresql/metadata.go

@@ -28,7 +28,7 @@ import (
 
 // PostgresAuthMetadata contains authentication metadata for PostgreSQL components.
 type PostgresAuthMetadata struct {
-	ConnectionString      string        `mapstructure:"connectionString"`
+	ConnectionString      string        `mapstructure:"connectionString" mapstructurealiases:"url"`
 	ConnectionMaxIdleTime time.Duration `mapstructure:"connectionMaxIdleTime"`
 	MaxConns              int           `mapstructure:"maxConns"`
 	UseAzureAD            bool          `mapstructure:"useAzureAD"`

common/component/postgresql/interfaces/interfaces.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2021 The Dapr Authors
+Copyright 2023 The Dapr Authors
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at

common/component/postgresql/transactions/transactions.go

@@ -0,0 +1,69 @@
+/*
+Copyright 2023 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package transactions
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+
+	pginterfaces "github.com/dapr/components-contrib/common/component/postgresql/interfaces"
+	"github.com/dapr/kit/logger"
+)
+
+// ExecuteInTransaction executes a function in a transaction.
+// If the handler returns an error, the transaction is rolled back automatically.
+func ExecuteInTransaction[T any](ctx context.Context, log logger.Logger, db pginterfaces.PGXPoolConn, timeout time.Duration, fn func(ctx context.Context, tx pgx.Tx) (T, error)) (res T, err error) {
+	// Start the transaction
+	queryCtx, queryCancel := context.WithTimeout(ctx, timeout)
+	defer queryCancel()
+	tx, err := db.Begin(queryCtx)
+	if err != nil {
+		return res, fmt.Errorf("failed to begin transaction: %w", err)
+	}
+
+	// Rollback in case of failure
+	var success bool
+	defer func() {
+		if success {
+			return
+		}
+		rollbackCtx, rollbackCancel := context.WithTimeout(ctx, timeout)
+		defer rollbackCancel()
+		rollbackErr := tx.Rollback(rollbackCtx)
+		if rollbackErr != nil {
+			// Log errors only
+			log.Errorf("Error while attempting to roll back transaction: %v", rollbackErr)
+		}
+	}()
+
+	// Execute the callback
+	res, err = fn(ctx, tx)
+	if err != nil {
+		return res, err
+	}
+
+	// Commit the transaction
+	queryCtx, queryCancel = context.WithTimeout(ctx, timeout)
+	defer queryCancel()
+	err = tx.Commit(queryCtx)
+	if err != nil {
+		return res, fmt.Errorf("failed to commit transaction: %w", err)
+	}
+	success = true
+
+	return res, nil
+}

common/component/postgresql/metadata.go renamed to common/component/postgresql/v1/metadata.go

@@ -14,7 +14,7 @@ limitations under the License.
 package postgresql
 
 import (
-	"fmt"
+	"errors"
 	"time"
 
 	pgauth "github.com/dapr/components-contrib/common/authentication/postgresql"
@@ -24,31 +24,28 @@ import (
 )
 
 const (
-	cleanupIntervalKey = "cleanupIntervalInSeconds"
-	timeoutKey         = "timeoutInSeconds"
-
 	defaultTableName         = "state"
 	defaultMetadataTableName = "dapr_metadata"
-	defaultCleanupInternal   = 3600 // In seconds = 1 hour
-	defaultTimeout           = 20   // Default timeout for network requests, in seconds
+	defaultCleanupInternal   = time.Hour
+	defaultTimeout           = 20 * time.Second // Default timeout for network requests
 )
 
 type pgMetadata struct {
 	pgauth.PostgresAuthMetadata `mapstructure:",squash"`
 
 	TableName         string         `mapstructure:"tableName"`         // Could be in the format "schema.table" or just "table"
 	MetadataTableName string         `mapstructure:"metadataTableName"` // Could be in the format "schema.table" or just "table"
-	Timeout           time.Duration  `mapstructure:"timeoutInSeconds"`
-	CleanupInterval   *time.Duration `mapstructure:"cleanupIntervalInSeconds"`
+	Timeout           time.Duration  `mapstructure:"timeout" mapstructurealiases:"timeoutInSeconds"`
+	CleanupInterval   *time.Duration `mapstructure:"cleanupInterval" mapstructurealiases:"cleanupIntervalInSeconds"`
 }
 
 func (m *pgMetadata) InitWithMetadata(meta state.Metadata, azureADEnabled bool) error {
 	// Reset the object
 	m.PostgresAuthMetadata.Reset()
 	m.TableName = defaultTableName
 	m.MetadataTableName = defaultMetadataTableName
-	m.CleanupInterval = ptr.Of(defaultCleanupInternal * time.Second)
-	m.Timeout = defaultTimeout * time.Second
+	m.CleanupInterval = ptr.Of(defaultCleanupInternal)
+	m.Timeout = defaultTimeout
 
 	// Decode the metadata
 	err := metadata.DecodeMetadata(meta.Properties, &m)
@@ -64,18 +61,18 @@ func (m *pgMetadata) InitWithMetadata(meta state.Metadata, azureADEnabled bool)
 
 	// Timeout
 	if m.Timeout < 1*time.Second {
-		return fmt.Errorf("invalid value for '%s': must be greater than 0", timeoutKey)
+		return errors.New("invalid value for 'timeout': must be greater than 1s")
 	}
 
 	// Cleanup interval
 	// Non-positive value from meta means disable auto cleanup.
-	if m.CleanupInterval != nil && *m.CleanupInterval <= 0 {
-		if meta.Properties[cleanupIntervalKey] == "" {
-			// Unfortunately the mapstructure decoder decodes an empty string to 0, a missing key would be nil however
-			m.CleanupInterval = ptr.Of(defaultCleanupInternal * time.Second)
-		} else {
-			m.CleanupInterval = nil
-		}
+	// We need to do this check because an empty string and "0" are treated differently by DecodeMetadata
+	v, ok := meta.GetProperty("cleanupInterval", "cleanupIntervalInSeconds")
+	if ok && v == "" {
+		// Handle the case of an empty string, but present
+		m.CleanupInterval = ptr.Of(defaultCleanupInternal)
+	} else if (ok && v == "0") || (m.CleanupInterval != nil && *m.CleanupInterval <= 0) {
+		m.CleanupInterval = nil
 	}
 
 	return nil