postgresql: change plugin id (#77444)

* postgres: change plugin id

* fixed cue file

* codeowners update

* fixed backend test

pkg/tsdb/grafana-postgresql-datasource/locker.go

@@ -0,0 +1,85 @@
+package postgres
+
+import (
+	"fmt"
+	"sync"
+)
+
+// locker is a named reader/writer mutual exclusion lock.
+// The lock for each particular key can be held by an arbitrary number of readers or a single writer.
+type locker struct {
+	locks   map[any]*sync.RWMutex
+	locksRW *sync.RWMutex
+}
+
+func newLocker() *locker {
+	return &locker{
+		locks:   make(map[any]*sync.RWMutex),
+		locksRW: new(sync.RWMutex),
+	}
+}
+
+// Lock locks named rw mutex with specified key for writing.
+// If the lock with the same key is already locked for reading or writing,
+// Lock blocks until the lock is available.
+func (lkr *locker) Lock(key any) {
+	lk, ok := lkr.getLock(key)
+	if !ok {
+		lk = lkr.newLock(key)
+	}
+	lk.Lock()
+}
+
+// Unlock unlocks named rw mutex with specified key for writing. It is a run-time error if rw is
+// not locked for writing on entry to Unlock.
+func (lkr *locker) Unlock(key any) {
+	lk, ok := lkr.getLock(key)
+	if !ok {
+		panic(fmt.Errorf("lock for key '%s' not initialized", key))
+	}
+	lk.Unlock()
+}
+
+// RLock locks named rw mutex with specified key for reading.
+//
+// It should not be used for recursive read locking for the same key; a blocked Lock
+// call excludes new readers from acquiring the lock. See the
+// documentation on the golang RWMutex type.
+func (lkr *locker) RLock(key any) {
+	lk, ok := lkr.getLock(key)
+	if !ok {
+		lk = lkr.newLock(key)
+	}
+	lk.RLock()
+}
+
+// RUnlock undoes a single RLock call for specified key;
+// it does not affect other simultaneous readers of locker for specified key.
+// It is a run-time error if locker for specified key is not locked for reading
+func (lkr *locker) RUnlock(key any) {
+	lk, ok := lkr.getLock(key)
+	if !ok {
+		panic(fmt.Errorf("lock for key '%s' not initialized", key))
+	}
+	lk.RUnlock()
+}
+
+func (lkr *locker) newLock(key any) *sync.RWMutex {
+	lkr.locksRW.Lock()
+	defer lkr.locksRW.Unlock()
+
+	if lk, ok := lkr.locks[key]; ok {
+		return lk
+	}
+	lk := new(sync.RWMutex)
+	lkr.locks[key] = lk
+	return lk
+}
+
+func (lkr *locker) getLock(key any) (*sync.RWMutex, bool) {
+	lkr.locksRW.RLock()
+	defer lkr.locksRW.RUnlock()
+
+	lock, ok := lkr.locks[key]
+	return lock, ok
+}

pkg/tsdb/grafana-postgresql-datasource/locker_test.go

@@ -0,0 +1,63 @@
+package postgres
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestIntegrationLocker(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Tests with Sleep")
+	}
+	const notUpdated = "not_updated"
+	const atThread1 = "at_thread_1"
+	const atThread2 = "at_thread_2"
+	t.Run("Should lock for same keys", func(t *testing.T) {
+		updated := notUpdated
+		locker := newLocker()
+		locker.Lock(1)
+		var wg sync.WaitGroup
+		wg.Add(1)
+		defer func() {
+			locker.Unlock(1)
+			wg.Wait()
+		}()
+
+		go func() {
+			locker.RLock(1)
+			defer func() {
+				locker.RUnlock(1)
+				wg.Done()
+			}()
+			require.Equal(t, atThread1, updated, "Value should be updated in different thread")
+			updated = atThread2
+		}()
+		time.Sleep(time.Millisecond * 10)
+		require.Equal(t, notUpdated, updated, "Value should not be updated in different thread")
+		updated = atThread1
+	})
+
+	t.Run("Should not lock for different keys", func(t *testing.T) {
+		updated := notUpdated
+		locker := newLocker()
+		locker.Lock(1)
+		defer locker.Unlock(1)
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() {
+			locker.RLock(2)
+			defer func() {
+				locker.RUnlock(2)
+				wg.Done()
+			}()
+			require.Equal(t, notUpdated, updated, "Value should not be updated in different thread")
+			updated = atThread2
+		}()
+		wg.Wait()
+		require.Equal(t, atThread2, updated, "Value should be updated in different thread")
+		updated = atThread1
+	})
+}

pkg/tsdb/grafana-postgresql-datasource/macros.go

@@ -0,0 +1,160 @@
+package postgres
+
+import (
+	"fmt"
+	"regexp"
+	"strings"
+	"time"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana-plugin-sdk-go/backend/gtime"
+
+	"github.com/grafana/grafana/pkg/tsdb/sqleng"
+)
+
+const rsIdentifier = `([_a-zA-Z0-9]+)`
+const sExpr = `\$` + rsIdentifier + `\(([^\)]*)\)`
+
+type postgresMacroEngine struct {
+	*sqleng.SQLMacroEngineBase
+	timescaledb bool
+}
+
+func newPostgresMacroEngine(timescaledb bool) sqleng.SQLMacroEngine {
+	return &postgresMacroEngine{
+		SQLMacroEngineBase: sqleng.NewSQLMacroEngineBase(),
+		timescaledb:        timescaledb,
+	}
+}
+
+func (m *postgresMacroEngine) Interpolate(query *backend.DataQuery, timeRange backend.TimeRange, sql string) (string, error) {
+	// TODO: Handle error
+	rExp, _ := regexp.Compile(sExpr)
+	var macroError error
+
+	sql = m.ReplaceAllStringSubmatchFunc(rExp, sql, func(groups []string) string {
+		// detect if $__timeGroup is supposed to add AS time for pre 5.3 compatibility
+		// if there is a ',' directly after the macro call $__timeGroup is probably used
+		// in the old way. Inside window function ORDER BY $__timeGroup will be followed
+		// by ')'
+		if groups[1] == "__timeGroup" {
+			if index := strings.Index(sql, groups[0]); index >= 0 {
+				index += len(groups[0])
+				// check for character after macro expression
+				if len(sql) > index && sql[index] == ',' {
+					groups[1] = "__timeGroupAlias"
+				}
+			}
+		}
+
+		args := strings.Split(groups[2], ",")
+		for i, arg := range args {
+			args[i] = strings.Trim(arg, " ")
+		}
+		res, err := m.evaluateMacro(timeRange, query, groups[1], args)
+		if err != nil && macroError == nil {
+			macroError = err
+			return "macro_error()"
+		}
+		return res
+	})
+
+	if macroError != nil {
+		return "", macroError
+	}
+
+	return sql, nil
+}
+
+//nolint:gocyclo
+func (m *postgresMacroEngine) evaluateMacro(timeRange backend.TimeRange, query *backend.DataQuery, name string, args []string) (string, error) {
+	switch name {
+	case "__time":
+		if len(args) == 0 {
+			return "", fmt.Errorf("missing time column argument for macro %v", name)
+		}
+		return fmt.Sprintf("%s AS \"time\"", args[0]), nil
+	case "__timeEpoch":
+		if len(args) == 0 {
+			return "", fmt.Errorf("missing time column argument for macro %v", name)
+		}
+		return fmt.Sprintf("extract(epoch from %s) as \"time\"", args[0]), nil
+	case "__timeFilter":
+		if len(args) == 0 {
+			return "", fmt.Errorf("missing time column argument for macro %v", name)
+		}
+
+		return fmt.Sprintf("%s BETWEEN '%s' AND '%s'", args[0], timeRange.From.UTC().Format(time.RFC3339Nano), timeRange.To.UTC().Format(time.RFC3339Nano)), nil
+	case "__timeFrom":
+		return fmt.Sprintf("'%s'", timeRange.From.UTC().Format(time.RFC3339Nano)), nil
+	case "__timeTo":
+		return fmt.Sprintf("'%s'", timeRange.To.UTC().Format(time.RFC3339Nano)), nil
+	case "__timeGroup":
+		if len(args) < 2 {
+			return "", fmt.Errorf("macro %v needs time column and interval and optional fill value", name)
+		}
+		interval, err := gtime.ParseInterval(strings.Trim(args[1], `'`))
+		if err != nil {
+			return "", fmt.Errorf("error parsing interval %v", args[1])
+		}
+		if len(args) == 3 {
+			err := sqleng.SetupFillmode(query, interval, args[2])
+			if err != nil {
+				return "", err
+			}
+		}
+
+		if m.timescaledb {
+			return fmt.Sprintf("time_bucket('%.3fs',%s)", interval.Seconds(), args[0]), nil
+		}
+
+		return fmt.Sprintf(
+			"floor(extract(epoch from %s)/%v)*%v", args[0],
+			interval.Seconds(),
+			interval.Seconds(),
+		), nil
+	case "__timeGroupAlias":
+		tg, err := m.evaluateMacro(timeRange, query, "__timeGroup", args)
+		if err == nil {
+			return tg + " AS \"time\"", nil
+		}
+		return "", err
+	case "__unixEpochFilter":
+		if len(args) == 0 {
+			return "", fmt.Errorf("missing time column argument for macro %v", name)
+		}
+		return fmt.Sprintf("%s >= %d AND %s <= %d", args[0], timeRange.From.UTC().Unix(), args[0], timeRange.To.UTC().Unix()), nil
+	case "__unixEpochNanoFilter":
+		if len(args) == 0 {
+			return "", fmt.Errorf("missing time column argument for macro %v", name)
+		}
+		return fmt.Sprintf("%s >= %d AND %s <= %d", args[0], timeRange.From.UTC().UnixNano(), args[0], timeRange.To.UTC().UnixNano()), nil
+	case "__unixEpochNanoFrom":
+		return fmt.Sprintf("%d", timeRange.From.UTC().UnixNano()), nil
+	case "__unixEpochNanoTo":
+		return fmt.Sprintf("%d", timeRange.To.UTC().UnixNano()), nil
+	case "__unixEpochGroup":
+		if len(args) < 2 {
+			return "", fmt.Errorf("macro %v needs time column and interval and optional fill value", name)
+		}
+		interval, err := gtime.ParseInterval(strings.Trim(args[1], `'`))
+		if err != nil {
+			return "", fmt.Errorf("error parsing interval %v", args[1])
+		}
+		if len(args) == 3 {
+			err := sqleng.SetupFillmode(query, interval, args[2])
+			if err != nil {
+				return "", err
+			}
+		}
+		return fmt.Sprintf("floor((%s)/%v)*%v", args[0], interval.Seconds(), interval.Seconds()), nil
+	case "__unixEpochGroupAlias":
+		tg, err := m.evaluateMacro(timeRange, query, "__unixEpochGroup", args)
+		if err == nil {
+			return tg + " AS \"time\"", nil
+		}
+		return "", err
+	default:
+		return "", fmt.Errorf("unknown macro %q", name)
+	}
+}

pkg/tsdb/grafana-postgresql-datasource/macros_test.go

@@ -0,0 +1,249 @@
+package postgres
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMacroEngine(t *testing.T) {
+	timescaledbEnabled := false
+	engine := newPostgresMacroEngine(timescaledbEnabled)
+	timescaledbEnabled = true
+	engineTS := newPostgresMacroEngine(timescaledbEnabled)
+	query := &backend.DataQuery{}
+
+	t.Run("Given a time range between 2018-04-12 00:00 and 2018-04-12 00:05", func(t *testing.T) {
+		from := time.Date(2018, 4, 12, 18, 0, 0, 0, time.UTC)
+		to := from.Add(5 * time.Minute)
+		timeRange := backend.TimeRange{From: from, To: to}
+
+		t.Run("interpolate __time function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "select $__time(time_column)")
+			require.NoError(t, err)
+			require.Equal(t, "select time_column AS \"time\"", sql)
+		})
+
+		t.Run("interpolate __time function wrapped in aggregation", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "select min($__time(time_column))")
+			require.NoError(t, err)
+
+			require.Equal(t, "select min(time_column AS \"time\")", sql)
+		})
+
+		t.Run("interpolate __timeFilter function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "WHERE $__timeFilter(time_column)")
+			require.NoError(t, err)
+
+			require.Equal(t, fmt.Sprintf("WHERE time_column BETWEEN '%s' AND '%s'", from.Format(time.RFC3339Nano), to.Format(time.RFC3339Nano)), sql)
+		})
+
+		t.Run("interpolate __timeFrom function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "select $__timeFrom()")
+			require.NoError(t, err)
+
+			require.Equal(t, "select '2018-04-12T18:00:00Z'", sql)
+		})
+
+		t.Run("interpolate __timeTo function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "select $__timeTo()")
+			require.NoError(t, err)
+
+			require.Equal(t, "select '2018-04-12T18:05:00Z'", sql)
+		})
+
+		t.Run("interpolate __timeGroup function pre 5.3 compatibility", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "SELECT $__timeGroup(time_column,'5m'), value")
+			require.NoError(t, err)
+
+			require.Equal(t, "SELECT floor(extract(epoch from time_column)/300)*300 AS \"time\", value", sql)
+
+			sql, err = engine.Interpolate(query, timeRange, "SELECT $__timeGroup(time_column,'5m') as time, value")
+			require.NoError(t, err)
+
+			require.Equal(t, "SELECT floor(extract(epoch from time_column)/300)*300 as time, value", sql)
+		})
+
+		t.Run("interpolate __timeGroup function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "SELECT $__timeGroup(time_column,'5m')")
+			require.NoError(t, err)
+			sql2, err := engine.Interpolate(query, timeRange, "SELECT $__timeGroupAlias(time_column,'5m')")
+			require.NoError(t, err)
+
+			require.Equal(t, "SELECT floor(extract(epoch from time_column)/300)*300", sql)
+			require.Equal(t, sql2, sql+" AS \"time\"")
+		})
+
+		t.Run("interpolate __timeGroup function with spaces between args", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "$__timeGroup(time_column , '5m')")
+			require.NoError(t, err)
+			sql2, err := engine.Interpolate(query, timeRange, "$__timeGroupAlias(time_column , '5m')")
+			require.NoError(t, err)
+
+			require.Equal(t, "floor(extract(epoch from time_column)/300)*300", sql)
+			require.Equal(t, sql2, sql+" AS \"time\"")
+		})
+
+		t.Run("interpolate __timeGroup function with TimescaleDB enabled", func(t *testing.T) {
+			sql, err := engineTS.Interpolate(query, timeRange, "GROUP BY $__timeGroup(time_column,'5m')")
+			require.NoError(t, err)
+			require.Equal(t, "GROUP BY time_bucket('300.000s',time_column)", sql)
+		})
+
+		t.Run("interpolate __timeGroup function with spaces between args and TimescaleDB enabled", func(t *testing.T) {
+			sql, err := engineTS.Interpolate(query, timeRange, "GROUP BY $__timeGroup(time_column , '5m')")
+			require.NoError(t, err)
+			require.Equal(t, "GROUP BY time_bucket('300.000s',time_column)", sql)
+		})
+
+		t.Run("interpolate __timeGroup function with large time range as an argument and TimescaleDB enabled", func(t *testing.T) {
+			sql, err := engineTS.Interpolate(query, timeRange, "GROUP BY $__timeGroup(time_column , '12d')")
+			require.NoError(t, err)
+			require.Equal(t, "GROUP BY time_bucket('1036800.000s',time_column)", sql)
+		})
+
+		t.Run("interpolate __timeGroup function with small time range as an argument and TimescaleDB enabled", func(t *testing.T) {
+			sql, err := engineTS.Interpolate(query, timeRange, "GROUP BY $__timeGroup(time_column , '20ms')")
+			require.NoError(t, err)
+			require.Equal(t, "GROUP BY time_bucket('0.020s',time_column)", sql)
+		})
+
+		t.Run("interpolate __unixEpochFilter function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochFilter(time)")
+			require.NoError(t, err)
+			require.Equal(t, fmt.Sprintf("select time >= %d AND time <= %d", from.Unix(), to.Unix()), sql)
+		})
+
+		t.Run("interpolate __unixEpochNanoFilter function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochNanoFilter(time)")
+			require.NoError(t, err)
+			require.Equal(t, fmt.Sprintf("select time >= %d AND time <= %d", from.UnixNano(), to.UnixNano()), sql)
+		})
+
+		t.Run("interpolate __unixEpochNanoFrom function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochNanoFrom()")
+			require.NoError(t, err)
+			require.Equal(t, fmt.Sprintf("select %d", from.UnixNano()), sql)
+		})
+
+		t.Run("interpolate __unixEpochNanoTo function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochNanoTo()")
+			require.NoError(t, err)
+			require.Equal(t, fmt.Sprintf("select %d", to.UnixNano()), sql)
+		})
+
+		t.Run("interpolate __unixEpochGroup function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "SELECT $__unixEpochGroup(time_column+time_adjustment,'5m')")
+			require.NoError(t, err)
+			sql2, err := engine.Interpolate(query, timeRange, "SELECT $__unixEpochGroupAlias(time_column+time_adjustment,'5m')")
+			require.NoError(t, err)
+			require.Equal(t, "SELECT floor((time_column+time_adjustment)/300)*300", sql)
+			require.Equal(t, sql2, sql+" AS \"time\"")
+		})
+	})
+
+	t.Run("Given a time range between 1960-02-01 07:00 and 1965-02-03 08:00", func(t *testing.T) {
+		from := time.Date(1960, 2, 1, 7, 0, 0, 0, time.UTC)
+		to := time.Date(1965, 2, 3, 8, 0, 0, 0, time.UTC)
+		timeRange := backend.TimeRange{
+			From: from,
+			To:   to,
+		}
+
+		t.Run("interpolate __timeFilter function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "WHERE $__timeFilter(time_column)")
+			require.NoError(t, err)
+			require.Equal(t, fmt.Sprintf("WHERE time_column BETWEEN '%s' AND '%s'", from.Format(time.RFC3339Nano), to.Format(time.RFC3339Nano)), sql)
+		})
+
+		t.Run("interpolate __unixEpochFilter function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochFilter(time)")
+			require.NoError(t, err)
+			require.Equal(t, fmt.Sprintf("select time >= %d AND time <= %d", from.Unix(), to.Unix()), sql)
+		})
+
+		t.Run("interpolate __unixEpochNanoFilter function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochNanoFilter(time)")
+			require.NoError(t, err)
+			require.Equal(t, fmt.Sprintf("select time >= %d AND time <= %d", from.UnixNano(), to.UnixNano()), sql)
+		})
+	})
+
+	t.Run("Given a time range between 1960-02-01 07:00 and 1980-02-03 08:00", func(t *testing.T) {
+		from := time.Date(1960, 2, 1, 7, 0, 0, 0, time.UTC)
+		to := time.Date(1980, 2, 3, 8, 0, 0, 0, time.UTC)
+		timeRange := backend.TimeRange{
+			From: from,
+			To:   to,
+		}
+
+		t.Run("interpolate __timeFilter function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "WHERE $__timeFilter(time_column)")
+			require.NoError(t, err)
+			require.Equal(t, fmt.Sprintf("WHERE time_column BETWEEN '%s' AND '%s'", from.Format(time.RFC3339Nano), to.Format(time.RFC3339Nano)), sql)
+		})
+
+		t.Run("interpolate __unixEpochFilter function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochFilter(time)")
+			require.NoError(t, err)
+			require.Equal(t, fmt.Sprintf("select time >= %d AND time <= %d", from.Unix(), to.Unix()), sql)
+		})
+
+		t.Run("interpolate __unixEpochNanoFilter function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochNanoFilter(time)")
+			require.NoError(t, err)
+			require.Equal(t, fmt.Sprintf("select time >= %d AND time <= %d", from.UnixNano(), to.UnixNano()), sql)
+		})
+	})
+
+	t.Run("Given a time range between 1960-02-01 07:00:00.5 and 1980-02-03 08:00:00.5", func(t *testing.T) {
+		from := time.Date(1960, 2, 1, 7, 0, 0, 500e6, time.UTC)
+		to := time.Date(1980, 2, 3, 8, 0, 0, 500e6, time.UTC)
+		timeRange := backend.TimeRange{
+			From: from,
+			To:   to,
+		}
+		require.Equal(t, "1960-02-01T07:00:00.5Z", from.Format(time.RFC3339Nano))
+		require.Equal(t, "1980-02-03T08:00:00.5Z", to.Format(time.RFC3339Nano))
+
+		t.Run("interpolate __timeFilter function", func(t *testing.T) {
+			sql, err := engine.Interpolate(query, timeRange, "WHERE $__timeFilter(time_column)")
+			require.NoError(t, err)
+			require.Equal(t, fmt.Sprintf("WHERE time_column BETWEEN '%s' AND '%s'", from.Format(time.RFC3339Nano), to.Format(time.RFC3339Nano)), sql)
+		})
+	})
+}
+
+func TestMacroEngineConcurrency(t *testing.T) {
+	engine := newPostgresMacroEngine(false)
+	query1 := backend.DataQuery{
+		JSON: []byte{},
+	}
+	query2 := backend.DataQuery{
+		JSON: []byte{},
+	}
+	from := time.Date(2018, 4, 12, 18, 0, 0, 0, time.UTC)
+	to := from.Add(5 * time.Minute)
+	timeRange := backend.TimeRange{From: from, To: to}
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	go func(query backend.DataQuery) {
+		defer wg.Done()
+		_, err := engine.Interpolate(&query, timeRange, "SELECT $__timeGroup(time_column,'5m')")
+		require.NoError(t, err)
+	}(query1)
+
+	go func(query backend.DataQuery) {
+		_, err := engine.Interpolate(&query, timeRange, "SELECT $__timeGroup(time_column,'5m')")
+		require.NoError(t, err)
+		defer wg.Done()
+	}(query2)
+
+	wg.Wait()
+}

pkg/tsdb/grafana-postgresql-datasource/postgres.go

@@ -0,0 +1,309 @@
+package postgres
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"reflect"
+	"strconv"
+	"strings"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana-plugin-sdk-go/backend/datasource"
+	"github.com/grafana/grafana-plugin-sdk-go/backend/instancemgmt"
+	sdkproxy "github.com/grafana/grafana-plugin-sdk-go/backend/proxy"
+	"github.com/grafana/grafana-plugin-sdk-go/data"
+	"github.com/grafana/grafana-plugin-sdk-go/data/sqlutil"
+
+	"github.com/grafana/grafana/pkg/infra/log"
+	"github.com/grafana/grafana/pkg/setting"
+	"github.com/grafana/grafana/pkg/tsdb/sqleng"
+	"github.com/grafana/grafana/pkg/tsdb/sqleng/proxyutil"
+)
+
+var logger = log.New("tsdb.postgres")
+
+func ProvideService(cfg *setting.Cfg) *Service {
+	s := &Service{
+		tlsManager: newTLSManager(logger, cfg.DataPath),
+	}
+	s.im = datasource.NewInstanceManager(s.newInstanceSettings(cfg))
+	return s
+}
+
+type Service struct {
+	tlsManager tlsSettingsProvider
+	im         instancemgmt.InstanceManager
+}
+
+func (s *Service) getDSInfo(ctx context.Context, pluginCtx backend.PluginContext) (*sqleng.DataSourceHandler, error) {
+	i, err := s.im.Get(ctx, pluginCtx)
+	if err != nil {
+		return nil, err
+	}
+	instance := i.(*sqleng.DataSourceHandler)
+	return instance, nil
+}
+
+func (s *Service) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
+	dsInfo, err := s.getDSInfo(ctx, req.PluginContext)
+	if err != nil {
+		return nil, err
+	}
+	return dsInfo.QueryData(ctx, req)
+}
+
+func (s *Service) newInstanceSettings(cfg *setting.Cfg) datasource.InstanceFactoryFunc {
+	return func(_ context.Context, settings backend.DataSourceInstanceSettings) (instancemgmt.Instance, error) {
+		logger.Debug("Creating Postgres query endpoint")
+		jsonData := sqleng.JsonData{
+			MaxOpenConns:        cfg.SqlDatasourceMaxOpenConnsDefault,
+			MaxIdleConns:        cfg.SqlDatasourceMaxIdleConnsDefault,
+			ConnMaxLifetime:     cfg.SqlDatasourceMaxConnLifetimeDefault,
+			Timescaledb:         false,
+			ConfigurationMethod: "file-path",
+			SecureDSProxy:       false,
+		}
+
+		err := json.Unmarshal(settings.JSONData, &jsonData)
+		if err != nil {
+			return nil, fmt.Errorf("error reading settings: %w", err)
+		}
+
+		database := jsonData.Database
+		if database == "" {
+			database = settings.Database
+		}
+
+		dsInfo := sqleng.DataSourceInfo{
+			JsonData:                jsonData,
+			URL:                     settings.URL,
+			User:                    settings.User,
+			Database:                database,
+			ID:                      settings.ID,
+			Updated:                 settings.Updated,
+			UID:                     settings.UID,
+			DecryptedSecureJSONData: settings.DecryptedSecureJSONData,
+		}
+
+		cnnstr, err := s.generateConnectionString(dsInfo)
+		if err != nil {
+			return nil, err
+		}
+
+		if cfg.Env == setting.Dev {
+			logger.Debug("GetEngine", "connection", cnnstr)
+		}
+
+		driverName := "postgres"
+		// register a proxy driver if the secure socks proxy is enabled
+		proxyOpts := proxyutil.GetSQLProxyOptions(cfg.SecureSocksDSProxy, dsInfo)
+		if sdkproxy.New(proxyOpts).SecureSocksProxyEnabled() {
+			driverName, err = createPostgresProxyDriver(cnnstr, proxyOpts)
+			if err != nil {
+				return "", nil
+			}
+		}
+
+		config := sqleng.DataPluginConfiguration{
+			DriverName:        driverName,
+			ConnectionString:  cnnstr,
+			DSInfo:            dsInfo,
+			MetricColumnTypes: []string{"UNKNOWN", "TEXT", "VARCHAR", "CHAR"},
+			RowLimit:          cfg.DataProxyRowLimit,
+		}
+
+		queryResultTransformer := postgresQueryResultTransformer{}
+
+		handler, err := sqleng.NewQueryDataHandler(cfg, config, &queryResultTransformer, newPostgresMacroEngine(dsInfo.JsonData.Timescaledb),
+			logger)
+		if err != nil {
+			logger.Error("Failed connecting to Postgres", "err", err)
+			return nil, err
+		}
+
+		logger.Debug("Successfully connected to Postgres")
+		return handler, nil
+	}
+}
+
+// escape single quotes and backslashes in Postgres connection string parameters.
+func escape(input string) string {
+	return strings.ReplaceAll(strings.ReplaceAll(input, `\`, `\\`), "'", `\'`)
+}
+
+func (s *Service) generateConnectionString(dsInfo sqleng.DataSourceInfo) (string, error) {
+	var host string
+	var port int
+	if strings.HasPrefix(dsInfo.URL, "/") {
+		host = dsInfo.URL
+		logger.Debug("Generating connection string with Unix socket specifier", "socket", host)
+	} else {
+		index := strings.LastIndex(dsInfo.URL, ":")
+		v6Index := strings.Index(dsInfo.URL, "]")
+		sp := strings.SplitN(dsInfo.URL, ":", 2)
+		host = sp[0]
+		if v6Index == -1 {
+			if len(sp) > 1 {
+				var err error
+				port, err = strconv.Atoi(sp[1])
+				if err != nil {
+					return "", fmt.Errorf("invalid port in host specifier %q: %w", sp[1], err)
+				}
+
+				logger.Debug("Generating connection string with network host/port pair", "host", host, "port", port)
+			} else {
+				logger.Debug("Generating connection string with network host", "host", host)
+			}
+		} else {
+			if index == v6Index+1 {
+				host = dsInfo.URL[1 : index-1]
+				var err error
+				port, err = strconv.Atoi(dsInfo.URL[index+1:])
+				if err != nil {
+					return "", fmt.Errorf("invalid port in host specifier %q: %w", dsInfo.URL[index+1:], err)
+				}
+
+				logger.Debug("Generating ipv6 connection string with network host/port pair", "host", host, "port", port)
+			} else {
+				host = dsInfo.URL[1 : len(dsInfo.URL)-1]
+				logger.Debug("Generating ipv6 connection string with network host", "host", host)
+			}
+		}
+	}
+
+	connStr := fmt.Sprintf("user='%s' password='%s' host='%s' dbname='%s'",
+		escape(dsInfo.User), escape(dsInfo.DecryptedSecureJSONData["password"]), escape(host), escape(dsInfo.Database))
+	if port > 0 {
+		connStr += fmt.Sprintf(" port=%d", port)
+	}
+
+	tlsSettings, err := s.tlsManager.getTLSSettings(dsInfo)
+	if err != nil {
+		return "", err
+	}
+
+	connStr += fmt.Sprintf(" sslmode='%s'", escape(tlsSettings.Mode))
+
+	// Attach root certificate if provided
+	if tlsSettings.RootCertFile != "" {
+		logger.Debug("Setting server root certificate", "tlsRootCert", tlsSettings.RootCertFile)
+		connStr += fmt.Sprintf(" sslrootcert='%s'", escape(tlsSettings.RootCertFile))
+	}
+
+	// Attach client certificate and key if both are provided
+	if tlsSettings.CertFile != "" && tlsSettings.CertKeyFile != "" {
+		logger.Debug("Setting TLS/SSL client auth", "tlsCert", tlsSettings.CertFile, "tlsKey", tlsSettings.CertKeyFile)
+		connStr += fmt.Sprintf(" sslcert='%s' sslkey='%s'", escape(tlsSettings.CertFile), escape(tlsSettings.CertKeyFile))
+	} else if tlsSettings.CertFile != "" || tlsSettings.CertKeyFile != "" {
+		return "", fmt.Errorf("TLS/SSL client certificate and key must both be specified")
+	}
+
+	logger.Debug("Generated Postgres connection string successfully")
+	return connStr, nil
+}
+
+type postgresQueryResultTransformer struct{}
+
+func (t *postgresQueryResultTransformer) TransformQueryError(_ log.Logger, err error) error {
+	return err
+}
+
+// CheckHealth pings the connected SQL database
+func (s *Service) CheckHealth(ctx context.Context, req *backend.CheckHealthRequest) (*backend.CheckHealthResult, error) {
+	dsHandler, err := s.getDSInfo(ctx, req.PluginContext)
+	if err != nil {
+		return nil, err
+	}
+
+	err = dsHandler.Ping()
+
+	if err != nil {
+		logger.Error("Check health failed", "error", err)
+		return &backend.CheckHealthResult{Status: backend.HealthStatusError, Message: dsHandler.TransformQueryError(logger, err).Error()}, nil
+	}
+
+	return &backend.CheckHealthResult{Status: backend.HealthStatusOk, Message: "Database Connection OK"}, nil
+}
+
+func (t *postgresQueryResultTransformer) GetConverterList() []sqlutil.StringConverter {
+	return []sqlutil.StringConverter{
+		{
+			Name:           "handle FLOAT4",
+			InputScanKind:  reflect.Interface,
+			InputTypeName:  "FLOAT4",
+			ConversionFunc: func(in *string) (*string, error) { return in, nil },
+			Replacer: &sqlutil.StringFieldReplacer{
+				OutputFieldType: data.FieldTypeNullableFloat64,
+				ReplaceFunc: func(in *string) (any, error) {
+					if in == nil {
+						return nil, nil
+					}
+					v, err := strconv.ParseFloat(*in, 64)
+					if err != nil {
+						return nil, err
+					}
+					return &v, nil
+				},
+			},
+		},
+		{
+			Name:           "handle FLOAT8",
+			InputScanKind:  reflect.Interface,
+			InputTypeName:  "FLOAT8",
+			ConversionFunc: func(in *string) (*string, error) { return in, nil },
+			Replacer: &sqlutil.StringFieldReplacer{
+				OutputFieldType: data.FieldTypeNullableFloat64,
+				ReplaceFunc: func(in *string) (any, error) {
+					if in == nil {
+						return nil, nil
+					}
+					v, err := strconv.ParseFloat(*in, 64)
+					if err != nil {
+						return nil, err
+					}
+					return &v, nil
+				},
+			},
+		},
+		{
+			Name:           "handle NUMERIC",
+			InputScanKind:  reflect.Interface,
+			InputTypeName:  "NUMERIC",
+			ConversionFunc: func(in *string) (*string, error) { return in, nil },
+			Replacer: &sqlutil.StringFieldReplacer{
+				OutputFieldType: data.FieldTypeNullableFloat64,
+				ReplaceFunc: func(in *string) (any, error) {
+					if in == nil {
+						return nil, nil
+					}
+					v, err := strconv.ParseFloat(*in, 64)
+					if err != nil {
+						return nil, err
+					}
+					return &v, nil
+				},
+			},
+		},
+		{
+			Name:           "handle INT2",
+			InputScanKind:  reflect.Interface,
+			InputTypeName:  "INT2",
+			ConversionFunc: func(in *string) (*string, error) { return in, nil },
+			Replacer: &sqlutil.StringFieldReplacer{
+				OutputFieldType: data.FieldTypeNullableInt16,
+				ReplaceFunc: func(in *string) (any, error) {
+					if in == nil {
+						return nil, nil
+					}
+					i64, err := strconv.ParseInt(*in, 10, 16)
+					if err != nil {
+						return nil, err
+					}
+					v := int16(i64)
+					return &v, nil
+				},
+			},
+		},
+	}
+}

pkg/tsdb/grafana-postgresql-datasource/proxy.go

@@ -0,0 +1,106 @@
+package postgres
+
+import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
+	"net"
+	"time"
+
+	sdkproxy "github.com/grafana/grafana-plugin-sdk-go/backend/proxy"
+	"github.com/grafana/grafana/pkg/tsdb/sqleng"
+	"github.com/grafana/grafana/pkg/util"
+	"github.com/lib/pq"
+	"golang.org/x/net/proxy"
+	"xorm.io/core"
+)
+
+// createPostgresProxyDriver creates and registers a new sql driver that uses a postgres connector and updates the dialer to
+// route connections through the secure socks proxy
+func createPostgresProxyDriver(cnnstr string, opts *sdkproxy.Options) (string, error) {
+	sqleng.XormDriverMu.Lock()
+	defer sqleng.XormDriverMu.Unlock()
+
+	// create a unique driver per connection string
+	hash, err := util.Md5SumString(cnnstr)
+	if err != nil {
+		return "", err
+	}
+	driverName := "postgres-proxy-" + hash
+
+	// only register the driver once
+	if core.QueryDriver(driverName) == nil {
+		connector, err := pq.NewConnector(cnnstr)
+		if err != nil {
+			return "", err
+		}
+
+		driver, err := newPostgresProxyDriver(connector, opts)
+		if err != nil {
+			return "", err
+		}
+
+		sql.Register(driverName, driver)
+		core.RegisterDriver(driverName, driver)
+	}
+	return driverName, nil
+}
+
+// postgresProxyDriver is a regular postgres driver with an updated dialer.
+// This is done because there is no way to save a dialer to the postgres driver in xorm
+type postgresProxyDriver struct {
+	c *pq.Connector
+}
+
+var _ driver.DriverContext = (*postgresProxyDriver)(nil)
+var _ core.Driver = (*postgresProxyDriver)(nil)
+
+// newPostgresProxyDriver updates the dialer for a postgres connector with a dialer that proxies connections through the secure socks proxy
+// and returns a new postgres driver to register
+func newPostgresProxyDriver(connector *pq.Connector, opts *sdkproxy.Options) (*postgresProxyDriver, error) {
+	dialer, err := sdkproxy.New(opts).NewSecureSocksProxyContextDialer()
+	if err != nil {
+		return nil, err
+	}
+
+	// update the postgres dialer with the proxy dialer
+	connector.Dialer(&postgresProxyDialer{d: dialer})
+
+	return &postgresProxyDriver{connector}, nil
+}
+
+// postgresProxyDialer implements the postgres dialer using a proxy dialer, as their functions differ slightly
+type postgresProxyDialer struct {
+	d proxy.Dialer
+}
+
+// Dial uses the normal proxy dial function with the updated dialer
+func (p *postgresProxyDialer) Dial(network, addr string) (c net.Conn, err error) {
+	return p.d.Dial(network, addr)
+}
+
+// DialTimeout uses the normal postgres dial timeout function with the updated dialer
+func (p *postgresProxyDialer) DialTimeout(network, address string, timeout time.Duration) (net.Conn, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	return p.d.(proxy.ContextDialer).DialContext(ctx, network, address)
+}
+
+// Parse uses the xorm postgres dialect for the driver (this has to be implemented to register the driver with xorm)
+func (d *postgresProxyDriver) Parse(a string, b string) (*core.Uri, error) {
+	sqleng.XormDriverMu.RLock()
+	defer sqleng.XormDriverMu.RUnlock()
+
+	return core.QueryDriver("postgres").Parse(a, b)
+}
+
+// OpenConnector returns the normal postgres connector that has the updated dialer context
+func (d *postgresProxyDriver) OpenConnector(name string) (driver.Connector, error) {
+	return d.c, nil
+}
+
+// Open uses the connector with the updated dialer to open a new connection
+func (d *postgresProxyDriver) Open(dsn string) (driver.Conn, error) {
+	return d.c.Connect(context.Background())
+}

pkg/tsdb/grafana-postgresql-datasource/proxy_test.go

@@ -0,0 +1,81 @@
+package postgres
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/grafana/grafana/pkg/setting"
+	"github.com/grafana/grafana/pkg/tsdb/sqleng"
+	"github.com/grafana/grafana/pkg/tsdb/sqleng/proxyutil"
+	"github.com/lib/pq"
+	"github.com/stretchr/testify/require"
+	"xorm.io/core"
+)
+
+func TestPostgresProxyDriver(t *testing.T) {
+	dialect := "postgres"
+	settings := proxyutil.SetupTestSecureSocksProxySettings(t)
+	proxySettings := setting.SecureSocksDSProxySettings{
+		Enabled:      true,
+		ClientCert:   settings.ClientCert,
+		ClientKey:    settings.ClientKey,
+		RootCA:       settings.RootCA,
+		ProxyAddress: settings.ProxyAddress,
+		ServerName:   settings.ServerName,
+	}
+	opts := proxyutil.GetSQLProxyOptions(proxySettings, sqleng.DataSourceInfo{UID: "1", JsonData: sqleng.JsonData{SecureDSProxy: true}})
+	dbURL := "localhost:5432"
+	cnnstr := fmt.Sprintf("postgres://auser:password@%s/db?sslmode=disable", dbURL)
+	driverName, err := createPostgresProxyDriver(cnnstr, opts)
+	require.NoError(t, err)
+
+	t.Run("Driver should not be registered more than once", func(t *testing.T) {
+		testDriver, err := createPostgresProxyDriver(cnnstr, opts)
+		require.NoError(t, err)
+		require.Equal(t, driverName, testDriver)
+	})
+
+	t.Run("A new driver should be created for a new connection string", func(t *testing.T) {
+		testDriver, err := createPostgresProxyDriver("server=localhost;user id=sa;password=yourStrong(!)Password;database=db2", opts)
+		require.NoError(t, err)
+		require.NotEqual(t, driverName, testDriver)
+	})
+
+	t.Run("Parse should have the same result as xorm mssql parse", func(t *testing.T) {
+		xormDriver := core.QueryDriver(dialect)
+		xormResult, err := xormDriver.Parse(dialect, cnnstr)
+		require.NoError(t, err)
+
+		xormNewDriver := core.QueryDriver(driverName)
+		xormNewResult, err := xormNewDriver.Parse(dialect, cnnstr)
+		require.NoError(t, err)
+		require.Equal(t, xormResult, xormNewResult)
+	})
+
+	t.Run("Connector should use dialer context that routes through the socks proxy to db", func(t *testing.T) {
+		connector, err := pq.NewConnector(cnnstr)
+		require.NoError(t, err)
+		driver, err := newPostgresProxyDriver(connector, opts)
+		require.NoError(t, err)
+
+		conn, err := driver.OpenConnector(cnnstr)
+		require.NoError(t, err)
+
+		_, err = conn.Connect(context.Background())
+		require.Contains(t, err.Error(), fmt.Sprintf("socks connect %s %s->%s", "tcp", settings.ProxyAddress, dbURL))
+	})
+
+	t.Run("Connector should use dialer context that routes through the socks proxy to db", func(t *testing.T) {
+		connector, err := pq.NewConnector(cnnstr)
+		require.NoError(t, err)
+		driver, err := newPostgresProxyDriver(connector, opts)
+		require.NoError(t, err)
+
+		conn, err := driver.OpenConnector(cnnstr)
+		require.NoError(t, err)
+
+		_, err = conn.Connect(context.Background())
+		require.Contains(t, err.Error(), fmt.Sprintf("socks connect %s %s->%s", "tcp", settings.ProxyAddress, dbURL))
+	})
+}

pkg/tsdb/grafana-postgresql-datasource/tlsmanager.go

@@ -0,0 +1,224 @@
+package postgres
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/grafana/grafana/pkg/infra/fs"
+	"github.com/grafana/grafana/pkg/infra/log"
+	"github.com/grafana/grafana/pkg/tsdb/sqleng"
+)
+
+var validateCertFunc = validateCertFilePaths
+var writeCertFileFunc = writeCertFile
+
+type certFileType int
+
+const (
+	rootCert = iota
+	clientCert
+	clientKey
+)
+
+type tlsSettingsProvider interface {
+	getTLSSettings(dsInfo sqleng.DataSourceInfo) (tlsSettings, error)
+}
+
+type datasourceCacheManager struct {
+	locker *locker
+	cache  sync.Map
+}
+
+type tlsManager struct {
+	logger          log.Logger
+	dsCacheInstance datasourceCacheManager
+	dataPath        string
+}
+
+func newTLSManager(logger log.Logger, dataPath string) tlsSettingsProvider {
+	return &tlsManager{
+		logger:          logger,
+		dataPath:        dataPath,
+		dsCacheInstance: datasourceCacheManager{locker: newLocker()},
+	}
+}
+
+type tlsSettings struct {
+	Mode                string
+	ConfigurationMethod string
+	RootCertFile        string
+	CertFile            string
+	CertKeyFile         string
+}
+
+func (m *tlsManager) getTLSSettings(dsInfo sqleng.DataSourceInfo) (tlsSettings, error) {
+	tlsconfig := tlsSettings{
+		Mode: dsInfo.JsonData.Mode,
+	}
+
+	isTLSDisabled := (tlsconfig.Mode == "disable")
+
+	if isTLSDisabled {
+		m.logger.Debug("Postgres TLS/SSL is disabled")
+		return tlsconfig, nil
+	}
+
+	m.logger.Debug("Postgres TLS/SSL is enabled", "tlsMode", tlsconfig.Mode)
+
+	tlsconfig.ConfigurationMethod = dsInfo.JsonData.ConfigurationMethod
+	tlsconfig.RootCertFile = dsInfo.JsonData.RootCertFile
+	tlsconfig.CertFile = dsInfo.JsonData.CertFile
+	tlsconfig.CertKeyFile = dsInfo.JsonData.CertKeyFile
+
+	if tlsconfig.ConfigurationMethod == "file-content" {
+		if err := m.writeCertFiles(dsInfo, &tlsconfig); err != nil {
+			return tlsconfig, err
+		}
+	} else {
+		if err := validateCertFunc(tlsconfig.RootCertFile, tlsconfig.CertFile, tlsconfig.CertKeyFile); err != nil {
+			return tlsconfig, err
+		}
+	}
+	return tlsconfig, nil
+}
+
+func (t certFileType) String() string {
+	switch t {
+	case rootCert:
+		return "root certificate"
+	case clientCert:
+		return "client certificate"
+	case clientKey:
+		return "client key"
+	default:
+		panic(fmt.Sprintf("Unrecognized certFileType %d", t))
+	}
+}
+
+func getFileName(dataDir string, fileType certFileType) string {
+	var filename string
+	switch fileType {
+	case rootCert:
+		filename = "root.crt"
+	case clientCert:
+		filename = "client.crt"
+	case clientKey:
+		filename = "client.key"
+	default:
+		panic(fmt.Sprintf("unrecognized certFileType %s", fileType.String()))
+	}
+	generatedFilePath := filepath.Join(dataDir, filename)
+	return generatedFilePath
+}
+
+// writeCertFile writes a certificate file.
+func writeCertFile(logger log.Logger, fileContent string, generatedFilePath string) error {
+	fileContent = strings.TrimSpace(fileContent)
+	if fileContent != "" {
+		logger.Debug("Writing cert file", "path", generatedFilePath)
+		if err := os.WriteFile(generatedFilePath, []byte(fileContent), 0600); err != nil {
+			return err
+		}
+		// Make sure the file has the permissions expected by the Postgresql driver, otherwise it will bail
+		if err := os.Chmod(generatedFilePath, 0600); err != nil {
+			return err
+		}
+		return nil
+	}
+
+	logger.Debug("Deleting cert file since no content is provided", "path", generatedFilePath)
+	exists, err := fs.Exists(generatedFilePath)
+	if err != nil {
+		return err
+	}
+	if exists {
+		if err := os.Remove(generatedFilePath); err != nil {
+			return fmt.Errorf("failed to remove %q: %w", generatedFilePath, err)
+		}
+	}
+	return nil
+}
+
+func (m *tlsManager) writeCertFiles(dsInfo sqleng.DataSourceInfo, tlsconfig *tlsSettings) error {
+	m.logger.Debug("Writing TLS certificate files to disk")
+	tlsRootCert := dsInfo.DecryptedSecureJSONData["tlsCACert"]
+	tlsClientCert := dsInfo.DecryptedSecureJSONData["tlsClientCert"]
+	tlsClientKey := dsInfo.DecryptedSecureJSONData["tlsClientKey"]
+	if tlsRootCert == "" && tlsClientCert == "" && tlsClientKey == "" {
+		m.logger.Debug("No TLS/SSL certificates provided")
+	}
+
+	// Calculate all files path
+	workDir := filepath.Join(m.dataPath, "tls", dsInfo.UID+"generatedTLSCerts")
+	tlsconfig.RootCertFile = getFileName(workDir, rootCert)
+	tlsconfig.CertFile = getFileName(workDir, clientCert)
+	tlsconfig.CertKeyFile = getFileName(workDir, clientKey)
+
+	// Find datasource in the cache, if found, skip writing files
+	cacheKey := strconv.Itoa(int(dsInfo.ID))
+	m.dsCacheInstance.locker.RLock(cacheKey)
+	item, ok := m.dsCacheInstance.cache.Load(cacheKey)
+	m.dsCacheInstance.locker.RUnlock(cacheKey)
+	if ok {
+		if !item.(time.Time).Before(dsInfo.Updated) {
+			return nil
+		}
+	}
+
+	m.dsCacheInstance.locker.Lock(cacheKey)
+	defer m.dsCacheInstance.locker.Unlock(cacheKey)
+
+	item, ok = m.dsCacheInstance.cache.Load(cacheKey)
+	if ok {
+		if !item.(time.Time).Before(dsInfo.Updated) {
+			return nil
+		}
+	}
+
+	// Write certification directory and files
+	exists, err := fs.Exists(workDir)
+	if err != nil {
+		return err
+	}
+	if !exists {
+		if err := os.MkdirAll(workDir, 0700); err != nil {
+			return err
+		}
+	}
+
+	if err = writeCertFileFunc(m.logger, tlsRootCert, tlsconfig.RootCertFile); err != nil {
+		return err
+	}
+	if err = writeCertFileFunc(m.logger, tlsClientCert, tlsconfig.CertFile); err != nil {
+		return err
+	}
+	if err = writeCertFileFunc(m.logger, tlsClientKey, tlsconfig.CertKeyFile); err != nil {
+		return err
+	}
+
+	// Update datasource cache
+	m.dsCacheInstance.cache.Store(cacheKey, dsInfo.Updated)
+	return nil
+}
+
+// validateCertFilePaths validates configured certificate file paths.
+func validateCertFilePaths(rootCert, clientCert, clientKey string) error {
+	for _, fpath := range []string{rootCert, clientCert, clientKey} {
+		if fpath == "" {
+			continue
+		}
+		exists, err := fs.Exists(fpath)
+		if err != nil {
+			return err
+		}
+		if !exists {
+			return fmt.Errorf("certificate file %q doesn't exist", fpath)
+		}
+	}
+	return nil
+}

pkg/tsdb/grafana-postgresql-datasource/tlsmanager_test.go

@@ -0,0 +1,292 @@
+package postgres
+
+import (
+	"fmt"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/grafana/grafana/pkg/infra/log"
+	"github.com/grafana/grafana/pkg/setting"
+	"github.com/grafana/grafana/pkg/tsdb/sqleng"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	_ "github.com/lib/pq"
+)
+
+var writeCertFileCallNum int
+
+// TestDataSourceCacheManager is to test the Cache manager
+func TestDataSourceCacheManager(t *testing.T) {
+	cfg := setting.NewCfg()
+	cfg.DataPath = t.TempDir()
+	mng := tlsManager{
+		logger:          log.New("tsdb.postgres"),
+		dsCacheInstance: datasourceCacheManager{locker: newLocker()},
+		dataPath:        cfg.DataPath,
+	}
+	jsonData := sqleng.JsonData{
+		Mode:                "verify-full",
+		ConfigurationMethod: "file-content",
+	}
+	secureJSONData := map[string]string{
+		"tlsClientCert": "I am client certification",
+		"tlsClientKey":  "I am client key",
+		"tlsCACert":     "I am CA certification",
+	}
+
+	updateTime := time.Now().Add(-5 * time.Minute)
+
+	mockValidateCertFilePaths()
+	t.Cleanup(resetValidateCertFilePaths)
+
+	t.Run("Check datasource cache creation", func(t *testing.T) {
+		var wg sync.WaitGroup
+		wg.Add(10)
+		for id := int64(1); id <= 10; id++ {
+			go func(id int64) {
+				ds := sqleng.DataSourceInfo{
+					ID:                      id,
+					Updated:                 updateTime,
+					Database:                "database",
+					JsonData:                jsonData,
+					DecryptedSecureJSONData: secureJSONData,
+					UID:                     "testData",
+				}
+				s := tlsSettings{}
+				err := mng.writeCertFiles(ds, &s)
+				require.NoError(t, err)
+				wg.Done()
+			}(id)
+		}
+		wg.Wait()
+
+		t.Run("check cache creation is succeed", func(t *testing.T) {
+			for id := int64(1); id <= 10; id++ {
+				updated, ok := mng.dsCacheInstance.cache.Load(strconv.Itoa(int(id)))
+				require.True(t, ok)
+				require.Equal(t, updateTime, updated)
+			}
+		})
+	})
+
+	t.Run("Check datasource cache modification", func(t *testing.T) {
+		t.Run("check when version not changed, cache and files are not updated", func(t *testing.T) {
+			mockWriteCertFile()
+			t.Cleanup(resetWriteCertFile)
+			var wg1 sync.WaitGroup
+			wg1.Add(5)
+			for id := int64(1); id <= 5; id++ {
+				go func(id int64) {
+					ds := sqleng.DataSourceInfo{
+						ID:                      1,
+						Updated:                 updateTime,
+						Database:                "database",
+						JsonData:                jsonData,
+						DecryptedSecureJSONData: secureJSONData,
+						UID:                     "testData",
+					}
+					s := tlsSettings{}
+					err := mng.writeCertFiles(ds, &s)
+					require.NoError(t, err)
+					wg1.Done()
+				}(id)
+			}
+			wg1.Wait()
+			assert.Equal(t, writeCertFileCallNum, 0)
+		})
+
+		t.Run("cache is updated with the last datasource version", func(t *testing.T) {
+			dsV2 := sqleng.DataSourceInfo{
+				ID:                      1,
+				Updated:                 updateTime.Add(time.Minute),
+				Database:                "database",
+				JsonData:                jsonData,
+				DecryptedSecureJSONData: secureJSONData,
+				UID:                     "testData",
+			}
+			dsV3 := sqleng.DataSourceInfo{
+				ID:                      1,
+				Updated:                 updateTime.Add(2 * time.Minute),
+				Database:                "database",
+				JsonData:                jsonData,
+				DecryptedSecureJSONData: secureJSONData,
+				UID:                     "testData",
+			}
+			s := tlsSettings{}
+			err := mng.writeCertFiles(dsV2, &s)
+			require.NoError(t, err)
+			err = mng.writeCertFiles(dsV3, &s)
+			require.NoError(t, err)
+			version, ok := mng.dsCacheInstance.cache.Load("1")
+			require.True(t, ok)
+			require.Equal(t, updateTime.Add(2*time.Minute), version)
+		})
+	})
+}
+
+// Test getFileName
+
+func TestGetFileName(t *testing.T) {
+	testCases := []struct {
+		desc                  string
+		datadir               string
+		fileType              certFileType
+		expErr                string
+		expectedGeneratedPath string
+	}{
+		{
+			desc:                  "Get File Name for root certification",
+			datadir:               ".",
+			fileType:              rootCert,
+			expectedGeneratedPath: "root.crt",
+		},
+		{
+			desc:                  "Get File Name for client certification",
+			datadir:               ".",
+			fileType:              clientCert,
+			expectedGeneratedPath: "client.crt",
+		},
+		{
+			desc:                  "Get File Name for client certification",
+			datadir:               ".",
+			fileType:              clientKey,
+			expectedGeneratedPath: "client.key",
+		},
+	}
+	for _, tt := range testCases {
+		t.Run(tt.desc, func(t *testing.T) {
+			generatedPath := getFileName(tt.datadir, tt.fileType)
+			assert.Equal(t, tt.expectedGeneratedPath, generatedPath)
+		})
+	}
+}
+
+// Test getTLSSettings.
+func TestGetTLSSettings(t *testing.T) {
+	cfg := setting.NewCfg()
+	cfg.DataPath = t.TempDir()
+
+	mockValidateCertFilePaths()
+	t.Cleanup(resetValidateCertFilePaths)
+
+	updatedTime := time.Now()
+
+	testCases := []struct {
+		desc           string
+		expErr         string
+		jsonData       sqleng.JsonData
+		secureJSONData map[string]string
+		uid            string
+		tlsSettings    tlsSettings
+		updated        time.Time
+	}{
+		{
+			desc:    "Custom TLS authentication disabled",
+			updated: updatedTime,
+			jsonData: sqleng.JsonData{
+				Mode:                "disable",
+				RootCertFile:        "i/am/coding/ca.crt",
+				CertFile:            "i/am/coding/client.crt",
+				CertKeyFile:         "i/am/coding/client.key",
+				ConfigurationMethod: "file-path",
+			},
+			tlsSettings: tlsSettings{Mode: "disable"},
+		},
+		{
+			desc:    "Custom TLS authentication with file path",
+			updated: updatedTime.Add(time.Minute),
+			jsonData: sqleng.JsonData{
+				Mode:                "verify-full",
+				ConfigurationMethod: "file-path",
+				RootCertFile:        "i/am/coding/ca.crt",
+				CertFile:            "i/am/coding/client.crt",
+				CertKeyFile:         "i/am/coding/client.key",
+			},
+			tlsSettings: tlsSettings{
+				Mode:                "verify-full",
+				ConfigurationMethod: "file-path",
+				RootCertFile:        "i/am/coding/ca.crt",
+				CertFile:            "i/am/coding/client.crt",
+				CertKeyFile:         "i/am/coding/client.key",
+			},
+		},
+		{
+			desc:    "Custom TLS mode verify-full with certificate files content",
+			updated: updatedTime.Add(2 * time.Minute),
+			uid:     "xxx",
+			jsonData: sqleng.JsonData{
+				Mode:                "verify-full",
+				ConfigurationMethod: "file-content",
+			},
+			secureJSONData: map[string]string{
+				"tlsCACert":     "I am CA certification",
+				"tlsClientCert": "I am client certification",
+				"tlsClientKey":  "I am client key",
+			},
+			tlsSettings: tlsSettings{
+				Mode:                "verify-full",
+				ConfigurationMethod: "file-content",
+				RootCertFile:        filepath.Join(cfg.DataPath, "tls", "xxxgeneratedTLSCerts", "root.crt"),
+				CertFile:            filepath.Join(cfg.DataPath, "tls", "xxxgeneratedTLSCerts", "client.crt"),
+				CertKeyFile:         filepath.Join(cfg.DataPath, "tls", "xxxgeneratedTLSCerts", "client.key"),
+			},
+		},
+	}
+	for _, tt := range testCases {
+		t.Run(tt.desc, func(t *testing.T) {
+			var settings tlsSettings
+			var err error
+			mng := tlsManager{
+				logger:          log.New("tsdb.postgres"),
+				dsCacheInstance: datasourceCacheManager{locker: newLocker()},
+				dataPath:        cfg.DataPath,
+			}
+
+			ds := sqleng.DataSourceInfo{
+				JsonData:                tt.jsonData,
+				DecryptedSecureJSONData: tt.secureJSONData,
+				UID:                     tt.uid,
+				Updated:                 tt.updated,
+			}
+
+			settings, err = mng.getTLSSettings(ds)
+
+			if tt.expErr == "" {
+				require.NoError(t, err, tt.desc)
+				assert.Equal(t, tt.tlsSettings, settings)
+			} else {
+				require.Error(t, err, tt.desc)
+				assert.True(t, strings.HasPrefix(err.Error(), tt.expErr),
+					fmt.Sprintf("%s: %q doesn't start with %q", tt.desc, err, tt.expErr))
+			}
+		})
+	}
+}
+
+func mockValidateCertFilePaths() {
+	validateCertFunc = func(rootCert, clientCert, clientKey string) error {
+		return nil
+	}
+}
+
+func resetValidateCertFilePaths() {
+	validateCertFunc = validateCertFilePaths
+}
+
+func mockWriteCertFile() {
+	writeCertFileCallNum = 0
+	writeCertFileFunc = func(logger log.Logger, fileContent string, generatedFilePath string) error {
+		writeCertFileCallNum++
+		return nil
+	}
+}
+
+func resetWriteCertFile() {
+	writeCertFileCallNum = 0
+	writeCertFileFunc = writeCertFile
+}