Postgres: Switch the datasource plugin from lib/pq to pgx (#108443)
* Postgres: Switch the datasource plugin from lib/pq to pgx * Fix lint
This commit is contained in:
Zoltán Bedi
+7
-3
@@ -576,9 +576,13 @@ exports[`better eslint`] = {
|
||||
],
|
||||
"packages/grafana-sql/src/components/configuration/ConnectionLimits.tsx:5381": [
|
||||
[0, 0, 0, "Add noMargin prop to Field components to remove built-in margins. Use layout components like Stack or Grid with the gap prop instead for consistent spacing.", "0"],
|
||||
[0, 0, 0, "Add noMargin prop to Field components to remove built-in margins. Use layout components like Stack or Grid with the gap prop instead for consistent spacing.", "1"],
|
||||
[0, 0, 0, "Add noMargin prop to Field components to remove built-in margins. Use layout components like Stack or Grid with the gap prop instead for consistent spacing.", "2"],
|
||||
[0, 0, 0, "Add noMargin prop to Field components to remove built-in margins. Use layout components like Stack or Grid with the gap prop instead for consistent spacing.", "3"]
|
||||
[0, 0, 0, "Add noMargin prop to Field components to remove built-in margins. Use layout components like Stack or Grid with the gap prop instead for consistent spacing.", "1"]
|
||||
],
|
||||
"packages/grafana-sql/src/components/configuration/MaxLifetimeField.tsx:5381": [
|
||||
[0, 0, 0, "Add noMargin prop to Field components to remove built-in margins. Use layout components like Stack or Grid with the gap prop instead for consistent spacing.", "0"]
|
||||
],
|
||||
"packages/grafana-sql/src/components/configuration/MaxOpenConnectionsField.tsx:5381": [
|
||||
[0, 0, 0, "Add noMargin prop to Field components to remove built-in margins. Use layout components like Stack or Grid with the gap prop instead for consistent spacing.", "0"]
|
||||
],
|
||||
"packages/grafana-sql/src/components/configuration/TLSSecretsConfig.tsx:5381": [
|
||||
[0, 0, 0, "Add noMargin prop to Field components to remove built-in margins. Use layout components like Stack or Grid with the gap prop instead for consistent spacing.", "0"],
|
||||
|
||||
@@ -960,6 +960,10 @@ export interface FeatureToggles {
|
||||
*/
|
||||
metricsFromProfiles?: boolean;
|
||||
/**
|
||||
* Enables using PGX instead of libpq for PostgreSQL datasource
|
||||
*/
|
||||
postgresDSUsePGX?: boolean;
|
||||
/**
|
||||
* Enables creating alerts from Tempo data source
|
||||
*/
|
||||
tempoAlerting?: boolean;
|
||||
|
||||
@@ -6,6 +6,8 @@ import { Field, Icon, InlineLabel, Label, Stack, Switch, Tooltip } from '@grafan
|
||||
|
||||
import { SQLConnectionLimits, SQLOptions } from '../../types';
|
||||
|
||||
import { MaxLifetimeField } from './MaxLifetimeField';
|
||||
import { MaxOpenConnectionsField } from './MaxOpenConnectionsField';
|
||||
import { NumberInput } from './NumberInput';
|
||||
|
||||
interface Props<T> {
|
||||
@@ -87,40 +89,11 @@ export const ConnectionLimits = <T extends SQLConnectionLimits>(props: Props<T>)
|
||||
<ConfigSubSection
|
||||
title={t('grafana-sql.components.connection-limits.title-connection-limits', 'Connection limits')}
|
||||
>
|
||||
<Field
|
||||
label={
|
||||
<Label>
|
||||
<Stack gap={0.5}>
|
||||
<span>
|
||||
<Trans i18nKey="grafana-sql.components.connection-limits.max-open">Max open</Trans>
|
||||
</span>
|
||||
<Tooltip
|
||||
content={
|
||||
<span>
|
||||
<Trans i18nKey="grafana-sql.components.connection-limits.content-max-open">
|
||||
The maximum number of open connections to the database. If <i>Max idle connections</i> is greater
|
||||
than 0 and the <i>Max open connections</i> is less than <i>Max idle connections</i>, then
|
||||
<i>Max idle connections</i> will be reduced to match the <i>Max open connections</i> limit. If set
|
||||
to 0, there is no limit on the number of open connections.
|
||||
</Trans>
|
||||
</span>
|
||||
}
|
||||
>
|
||||
<Icon name="info-circle" size="sm" />
|
||||
</Tooltip>
|
||||
</Stack>
|
||||
</Label>
|
||||
}
|
||||
>
|
||||
<NumberInput
|
||||
value={jsonData.maxOpenConns}
|
||||
defaultValue={config.sqlConnectionLimits.maxOpenConns}
|
||||
onChange={(value) => {
|
||||
onMaxConnectionsChanged(value);
|
||||
}}
|
||||
width={labelWidth}
|
||||
/>
|
||||
</Field>
|
||||
<MaxOpenConnectionsField
|
||||
labelWidth={labelWidth}
|
||||
onMaxConnectionsChanged={onMaxConnectionsChanged}
|
||||
jsonData={jsonData}
|
||||
/>
|
||||
|
||||
<Field
|
||||
label={
|
||||
@@ -191,38 +164,11 @@ export const ConnectionLimits = <T extends SQLConnectionLimits>(props: Props<T>)
|
||||
)}
|
||||
</Field>
|
||||
|
||||
<Field
|
||||
label={
|
||||
<Label>
|
||||
<Stack gap={0.5}>
|
||||
<span>
|
||||
<Trans i18nKey="grafana-sql.components.connection-limits.max-lifetime">Max lifetime</Trans>
|
||||
</span>
|
||||
<Tooltip
|
||||
content={
|
||||
<span>
|
||||
<Trans i18nKey="grafana-sql.components.connection-limits.content-max-lifetime">
|
||||
The maximum amount of time in seconds a connection may be reused. If set to 0, connections are
|
||||
reused forever.
|
||||
</Trans>
|
||||
</span>
|
||||
}
|
||||
>
|
||||
<Icon name="info-circle" size="sm" />
|
||||
</Tooltip>
|
||||
</Stack>
|
||||
</Label>
|
||||
}
|
||||
>
|
||||
<NumberInput
|
||||
value={jsonData.connMaxLifetime}
|
||||
defaultValue={config.sqlConnectionLimits.connMaxLifetime}
|
||||
onChange={(value) => {
|
||||
onJSONDataNumberChanged('connMaxLifetime')(value);
|
||||
}}
|
||||
width={labelWidth}
|
||||
/>
|
||||
</Field>
|
||||
<MaxLifetimeField
|
||||
labelWidth={labelWidth}
|
||||
onMaxLifetimeChanged={onJSONDataNumberChanged('connMaxLifetime')}
|
||||
jsonData={jsonData}
|
||||
/>
|
||||
</ConfigSubSection>
|
||||
);
|
||||
};
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
import { Trans } from '@grafana/i18n';
|
||||
import { config } from '@grafana/runtime';
|
||||
import { Field, Icon, Label, Stack, Tooltip } from '@grafana/ui';
|
||||
|
||||
import { SQLOptions } from '../../types';
|
||||
|
||||
import { NumberInput } from './NumberInput';
|
||||
|
||||
interface Props {
|
||||
labelWidth: number;
|
||||
onMaxLifetimeChanged: (number?: number) => void;
|
||||
jsonData: SQLOptions;
|
||||
}
|
||||
export function MaxLifetimeField({ labelWidth, onMaxLifetimeChanged, jsonData }: Props) {
|
||||
return (
|
||||
<Field
|
||||
label={
|
||||
<Label>
|
||||
<Stack gap={0.5}>
|
||||
<span>
|
||||
<Trans i18nKey="grafana-sql.components.connection-limits.max-lifetime">Max lifetime</Trans>
|
||||
</span>
|
||||
<Tooltip
|
||||
content={
|
||||
<span>
|
||||
<Trans i18nKey="grafana-sql.components.connection-limits.content-max-lifetime">
|
||||
The maximum amount of time in seconds a connection may be reused. If set to 0, connections are
|
||||
reused forever.
|
||||
</Trans>
|
||||
</span>
|
||||
}
|
||||
>
|
||||
<Icon name="info-circle" size="sm" />
|
||||
</Tooltip>
|
||||
</Stack>
|
||||
</Label>
|
||||
}
|
||||
>
|
||||
<NumberInput
|
||||
value={jsonData.connMaxLifetime}
|
||||
defaultValue={config.sqlConnectionLimits.connMaxLifetime}
|
||||
onChange={onMaxLifetimeChanged}
|
||||
width={labelWidth}
|
||||
/>
|
||||
</Field>
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
import { Trans } from '@grafana/i18n';
|
||||
import { config } from '@grafana/runtime';
|
||||
import { Field, Icon, Label, Stack, Tooltip } from '@grafana/ui';
|
||||
|
||||
import { SQLOptions } from '../../types';
|
||||
|
||||
import { NumberInput } from './NumberInput';
|
||||
|
||||
interface Props {
|
||||
labelWidth: number;
|
||||
onMaxConnectionsChanged: (number?: number) => void;
|
||||
jsonData: SQLOptions;
|
||||
}
|
||||
|
||||
export function MaxOpenConnectionsField({ labelWidth, onMaxConnectionsChanged, jsonData }: Props) {
|
||||
return (
|
||||
<Field
|
||||
label={
|
||||
<Label>
|
||||
<Stack gap={0.5}>
|
||||
<span>
|
||||
<Trans i18nKey="grafana-sql.components.connection-limits.max-open">Max open</Trans>
|
||||
</span>
|
||||
<Tooltip
|
||||
content={
|
||||
<span>
|
||||
<Trans i18nKey="grafana-sql.components.connection-limits.content-max-open">
|
||||
The maximum number of open connections to the database. If <i>Max idle connections</i> is greater
|
||||
than 0 and the <i>Max open connections</i> is less than <i>Max idle connections</i>, then
|
||||
<i>Max idle connections</i> will be reduced to match the <i>Max open connections</i> limit. If set
|
||||
to 0, there is no limit on the number of open connections.
|
||||
</Trans>
|
||||
</span>
|
||||
}
|
||||
>
|
||||
<Icon name="info-circle" size="sm" />
|
||||
</Tooltip>
|
||||
</Stack>
|
||||
</Label>
|
||||
}
|
||||
>
|
||||
<NumberInput
|
||||
value={jsonData.maxOpenConns}
|
||||
defaultValue={config.sqlConnectionLimits.maxOpenConns}
|
||||
onChange={onMaxConnectionsChanged}
|
||||
width={labelWidth}
|
||||
/>
|
||||
</Field>
|
||||
);
|
||||
}
|
||||
@@ -14,6 +14,8 @@ export { COMMON_FNS, MACRO_FUNCTIONS } from './constants';
|
||||
export { SqlDatasource } from './datasource/SqlDatasource';
|
||||
export { formatSQL } from './utils/formatSQL';
|
||||
export { ConnectionLimits } from './components/configuration/ConnectionLimits';
|
||||
export { MaxLifetimeField } from './components/configuration/MaxLifetimeField';
|
||||
export { MaxOpenConnectionsField } from './components/configuration/MaxOpenConnectionsField';
|
||||
export { Divider } from './components/configuration/Divider';
|
||||
export { TLSSecretsConfig } from './components/configuration/TLSSecretsConfig';
|
||||
export { useMigrateDatabaseFields } from './components/configuration/useMigrateDatabaseFields';
|
||||
|
||||
@@ -237,7 +237,7 @@ func NewPlugin(pluginID string, cfg *setting.Cfg, httpClientProvider *httpclient
|
||||
case Tempo:
|
||||
svc = tempo.ProvideService(httpClientProvider)
|
||||
case PostgreSQL:
|
||||
svc = postgres.ProvideService(cfg)
|
||||
svc = postgres.ProvideService(cfg, features)
|
||||
case MySQL:
|
||||
svc = mysql.ProvideService()
|
||||
case MSSQL:
|
||||
|
||||
@@ -380,7 +380,7 @@ func Initialize(cfg *setting.Cfg, opts Options, apiOpts api.ServerOptions) (*Ser
|
||||
prometheusService := prometheus.ProvideService(httpclientProvider)
|
||||
tempoService := tempo.ProvideService(httpclientProvider)
|
||||
testdatasourceService := testdatasource.ProvideService()
|
||||
postgresService := postgres.ProvideService(cfg)
|
||||
postgresService := postgres.ProvideService(cfg, featureToggles)
|
||||
mysqlService := mysql.ProvideService()
|
||||
mssqlService := mssql.ProvideService(cfg)
|
||||
entityEventsService := store.ProvideEntityEventsService(cfg, sqlStore, featureToggles)
|
||||
@@ -939,7 +939,7 @@ func InitializeForTest(t sqlutil.ITestDB, testingT interface {
|
||||
prometheusService := prometheus.ProvideService(httpclientProvider)
|
||||
tempoService := tempo.ProvideService(httpclientProvider)
|
||||
testdatasourceService := testdatasource.ProvideService()
|
||||
postgresService := postgres.ProvideService(cfg)
|
||||
postgresService := postgres.ProvideService(cfg, featureToggles)
|
||||
mysqlService := mysql.ProvideService()
|
||||
mssqlService := mssql.ProvideService(cfg)
|
||||
entityEventsService := store.ProvideEntityEventsService(cfg, sqlStore, featureToggles)
|
||||
|
||||
@@ -1655,6 +1655,12 @@ var (
|
||||
Owner: grafanaObservabilityTracesAndProfilingSquad,
|
||||
FrontendOnly: true,
|
||||
},
|
||||
{
|
||||
Name: "postgresDSUsePGX",
|
||||
Description: "Enables using PGX instead of libpq for PostgreSQL datasource",
|
||||
Stage: FeatureStageExperimental,
|
||||
Owner: grafanaOSSBigTent,
|
||||
},
|
||||
{
|
||||
Name: "tempoAlerting",
|
||||
Description: "Enables creating alerts from Tempo data source",
|
||||
|
||||
@@ -215,6 +215,7 @@ localizationForPlugins,experimental,@grafana/plugins-platform-backend,false,fals
|
||||
unifiedNavbars,GA,@grafana/plugins-platform-backend,false,false,true
|
||||
logsPanelControls,preview,@grafana/observability-logs,false,false,true
|
||||
metricsFromProfiles,experimental,@grafana/observability-traces-and-profiling,false,false,true
|
||||
postgresDSUsePGX,experimental,@grafana/oss-big-tent,false,false,false
|
||||
tempoAlerting,experimental,@grafana/observability-traces-and-profiling,false,false,true
|
||||
pluginsAutoUpdate,experimental,@grafana/plugins-platform-backend,false,false,false
|
||||
multiTenantFrontend,experimental,@grafana/grafana-frontend-platform,false,false,false
|
||||
|
||||
|
@@ -871,6 +871,10 @@ const (
|
||||
// Enables creating metrics from profiles and storing them as recording rules
|
||||
FlagMetricsFromProfiles = "metricsFromProfiles"
|
||||
|
||||
// FlagPostgresDSUsePGX
|
||||
// Enables using PGX instead of libpq for PostgreSQL datasource
|
||||
FlagPostgresDSUsePGX = "postgresDSUsePGX"
|
||||
|
||||
// FlagTempoAlerting
|
||||
// Enables creating alerts from Tempo data source
|
||||
FlagTempoAlerting = "tempoAlerting"
|
||||
|
||||
@@ -2413,6 +2413,19 @@
|
||||
"expression": "false"
|
||||
}
|
||||
},
|
||||
{
|
||||
"metadata": {
|
||||
"name": "postgresDSUsePGX",
|
||||
"resourceVersion": "1753174666056",
|
||||
"creationTimestamp": "2025-05-26T06:54:18Z",
|
||||
"deletionTimestamp": "2025-06-03T12:45:07Z"
|
||||
},
|
||||
"spec": {
|
||||
"description": "Enables using PGX instead of libpq for PostgreSQL datasource",
|
||||
"stage": "experimental",
|
||||
"codeowner": "@grafana/oss-big-tent"
|
||||
}
|
||||
},
|
||||
{
|
||||
"metadata": {
|
||||
"name": "preferLibraryPanelTitle",
|
||||
|
||||
@@ -163,7 +163,7 @@ func TestIntegrationPluginManager(t *testing.T) {
|
||||
pr := prometheus.ProvideService(hcp)
|
||||
tmpo := tempo.ProvideService(hcp)
|
||||
td := testdatasource.ProvideService()
|
||||
pg := postgres.ProvideService(cfg)
|
||||
pg := postgres.ProvideService(cfg, features)
|
||||
my := mysql.ProvideService()
|
||||
ms := mssql.ProvideService(cfg)
|
||||
db := db.InitTestDB(t, sqlstore.InitTestDBOpt{Cfg: cfg})
|
||||
|
||||
@@ -0,0 +1,151 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend/log"
|
||||
"github.com/grafana/grafana/pkg/tsdb/grafana-postgresql-datasource/sqleng"
|
||||
)
|
||||
|
||||
var validateCertFuncPgx = validateCertFilePathsPgx
|
||||
|
||||
type pgxTlsManager struct {
|
||||
logger log.Logger
|
||||
}
|
||||
|
||||
func newPgxTlsManager(logger log.Logger) *pgxTlsManager {
|
||||
return &pgxTlsManager{
|
||||
logger: logger,
|
||||
}
|
||||
}
|
||||
|
||||
// getTLSSettings retrieves TLS settings and handles certificate file creation if needed.
|
||||
func (m *pgxTlsManager) getTLSSettings(dsInfo sqleng.DataSourceInfo) (tlsSettings, error) {
|
||||
tlsConfig := tlsSettings{
|
||||
Mode: dsInfo.JsonData.Mode,
|
||||
}
|
||||
|
||||
if tlsConfig.Mode == "disable" {
|
||||
m.logger.Debug("Postgres TLS/SSL is disabled")
|
||||
return tlsConfig, nil
|
||||
}
|
||||
|
||||
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.createCertFiles(dsInfo, &tlsConfig); err != nil {
|
||||
return tlsConfig, fmt.Errorf("failed to create TLS certificate files: %w", err)
|
||||
}
|
||||
} else {
|
||||
if err := validateCertFuncPgx(tlsConfig.RootCertFile, tlsConfig.CertFile, tlsConfig.CertKeyFile); err != nil {
|
||||
return tlsConfig, fmt.Errorf("invalid TLS certificate file paths: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
return tlsConfig, nil
|
||||
}
|
||||
|
||||
// createCertFiles writes certificate files to temporary locations.
|
||||
func (m *pgxTlsManager) createCertFiles(dsInfo sqleng.DataSourceInfo, tlsConfig *tlsSettings) error {
|
||||
m.logger.Debug("Writing TLS certificate files to temporary locations")
|
||||
|
||||
var err error
|
||||
if tlsConfig.RootCertFile, err = m.writeCertFile("root-*.crt", dsInfo.DecryptedSecureJSONData["tlsCACert"]); err != nil {
|
||||
return err
|
||||
}
|
||||
if tlsConfig.CertFile, err = m.writeCertFile("client-*.crt", dsInfo.DecryptedSecureJSONData["tlsClientCert"]); err != nil {
|
||||
return err
|
||||
}
|
||||
if tlsConfig.CertKeyFile, err = m.writeCertFile("client-*.key", dsInfo.DecryptedSecureJSONData["tlsClientKey"]); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// writeCertFile writes a single certificate file to a temporary location.
|
||||
func (m *pgxTlsManager) writeCertFile(pattern, content string) (string, error) {
|
||||
if content == "" {
|
||||
return "", nil
|
||||
}
|
||||
|
||||
m.logger.Debug("Writing certificate file", "pattern", pattern)
|
||||
file, err := os.CreateTemp("", pattern)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to create temporary file: %w", err)
|
||||
}
|
||||
defer func() {
|
||||
if err := file.Close(); err != nil {
|
||||
m.logger.Error("Failed to close file", "error", err)
|
||||
}
|
||||
}()
|
||||
|
||||
if _, err := file.WriteString(content); err != nil {
|
||||
return "", fmt.Errorf("failed to write to temporary file: %w", err)
|
||||
}
|
||||
|
||||
return file.Name(), nil
|
||||
}
|
||||
|
||||
// cleanupCertFiles removes temporary certificate files.
|
||||
func (m *pgxTlsManager) cleanupCertFiles(tlsConfig tlsSettings) {
|
||||
// Only clean up if the configuration method is "file-content"
|
||||
if tlsConfig.ConfigurationMethod != "file-content" {
|
||||
m.logger.Debug("Skipping cleanup of TLS certificate files")
|
||||
return
|
||||
}
|
||||
m.logger.Debug("Cleaning up TLS certificate files")
|
||||
|
||||
files := []struct {
|
||||
path string
|
||||
name string
|
||||
}{
|
||||
{tlsConfig.RootCertFile, "root certificate"},
|
||||
{tlsConfig.CertFile, "client certificate"},
|
||||
{tlsConfig.CertKeyFile, "client key"},
|
||||
}
|
||||
|
||||
for _, file := range files {
|
||||
if file.path == "" {
|
||||
continue
|
||||
}
|
||||
if err := os.Remove(file.path); err != nil {
|
||||
m.logger.Error("Failed to remove file", "type", file.name, "path", file.path, "error", err)
|
||||
} else {
|
||||
m.logger.Debug("Successfully removed file", "type", file.name, "path", file.path)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// validateCertFilePaths validates the existence of configured certificate file paths.
|
||||
func validateCertFilePathsPgx(rootCert, clientCert, clientKey string) error {
|
||||
for _, path := range []string{rootCert, clientCert, clientKey} {
|
||||
if path == "" {
|
||||
continue
|
||||
}
|
||||
exists, err := fileExistsPgx(path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("error checking file existence: %w", err)
|
||||
}
|
||||
if !exists {
|
||||
return sqleng.ErrCertFileNotExist
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// fileExists checks if a file exists at the given path.
|
||||
func fileExistsPgx(path string) (bool, error) {
|
||||
_, err := os.Stat(path)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return false, nil
|
||||
}
|
||||
return false, err
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
@@ -0,0 +1,245 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend/log"
|
||||
"github.com/grafana/grafana/pkg/tsdb/grafana-postgresql-datasource/sqleng"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// Test getTLSSettings.
|
||||
func TestPgxGetTLSSettings(t *testing.T) {
|
||||
mockValidateCertFilePathsPgx()
|
||||
t.Cleanup(resetValidateCertFilePathsPgx)
|
||||
|
||||
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",
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tt := range testCases {
|
||||
t.Run(tt.desc, func(t *testing.T) {
|
||||
var settings tlsSettings
|
||||
var err error
|
||||
mng := pgxTlsManager{
|
||||
logger: backend.NewLoggerWith("logger", "tsdb.postgres"),
|
||||
}
|
||||
|
||||
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 mockValidateCertFilePathsPgx() {
|
||||
validateCertFuncPgx = func(rootCert, clientCert, clientKey string) error {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func resetValidateCertFilePathsPgx() {
|
||||
validateCertFuncPgx = validateCertFilePathsPgx
|
||||
}
|
||||
|
||||
func TestTLSManager_GetTLSSettings(t *testing.T) {
|
||||
logger := log.New()
|
||||
tlsManager := newPgxTlsManager(logger)
|
||||
|
||||
dsInfo := sqleng.DataSourceInfo{
|
||||
JsonData: sqleng.JsonData{
|
||||
Mode: "require",
|
||||
ConfigurationMethod: "file-content",
|
||||
},
|
||||
DecryptedSecureJSONData: map[string]string{
|
||||
"tlsCACert": "root-cert-content",
|
||||
"tlsClientCert": "client-cert-content",
|
||||
"tlsClientKey": "client-key-content",
|
||||
},
|
||||
}
|
||||
|
||||
tlsConfig, err := tlsManager.getTLSSettings(dsInfo)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "require", tlsConfig.Mode)
|
||||
assert.NotEmpty(t, tlsConfig.RootCertFile)
|
||||
assert.NotEmpty(t, tlsConfig.CertFile)
|
||||
assert.NotEmpty(t, tlsConfig.CertKeyFile)
|
||||
|
||||
// Cleanup temporary files
|
||||
tlsManager.cleanupCertFiles(tlsConfig)
|
||||
assert.NoFileExists(t, tlsConfig.RootCertFile)
|
||||
assert.NoFileExists(t, tlsConfig.CertFile)
|
||||
assert.NoFileExists(t, tlsConfig.CertKeyFile)
|
||||
}
|
||||
|
||||
func TestTLSManager_CleanupCertFiles_FilePath(t *testing.T) {
|
||||
logger := log.New()
|
||||
tlsManager := newPgxTlsManager(logger)
|
||||
|
||||
// Create temporary files for testing
|
||||
rootCertFile, err := tlsManager.writeCertFile("root-*.crt", "root-cert-content")
|
||||
require.NoError(t, err)
|
||||
clientCertFile, err := tlsManager.writeCertFile("client-*.crt", "client-cert-content")
|
||||
require.NoError(t, err)
|
||||
clientKeyFile, err := tlsManager.writeCertFile("client-*.key", "client-key-content")
|
||||
require.NoError(t, err)
|
||||
|
||||
// Simulate a configuration where the method is "file-path"
|
||||
tlsConfig := tlsSettings{
|
||||
ConfigurationMethod: "file-path",
|
||||
RootCertFile: rootCertFile,
|
||||
CertFile: clientCertFile,
|
||||
CertKeyFile: clientKeyFile,
|
||||
}
|
||||
|
||||
// Call cleanupCertFiles
|
||||
tlsManager.cleanupCertFiles(tlsConfig)
|
||||
|
||||
// Verify the files are NOT deleted
|
||||
assert.FileExists(t, rootCertFile, "Root certificate file should not be deleted")
|
||||
assert.FileExists(t, clientCertFile, "Client certificate file should not be deleted")
|
||||
assert.FileExists(t, clientKeyFile, "Client key file should not be deleted")
|
||||
|
||||
// Cleanup the files manually
|
||||
err = os.Remove(rootCertFile)
|
||||
require.NoError(t, err)
|
||||
err = os.Remove(clientCertFile)
|
||||
require.NoError(t, err)
|
||||
err = os.Remove(clientKeyFile)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func TestTLSManager_CreateCertFiles(t *testing.T) {
|
||||
logger := log.New()
|
||||
tlsManager := newPgxTlsManager(logger)
|
||||
|
||||
dsInfo := sqleng.DataSourceInfo{
|
||||
DecryptedSecureJSONData: map[string]string{
|
||||
"tlsCACert": "root-cert-content",
|
||||
"tlsClientCert": "client-cert-content",
|
||||
"tlsClientKey": "client-key-content",
|
||||
},
|
||||
}
|
||||
|
||||
tlsConfig := tlsSettings{
|
||||
ConfigurationMethod: "file-content",
|
||||
}
|
||||
err := tlsManager.createCertFiles(dsInfo, &tlsConfig)
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.FileExists(t, tlsConfig.RootCertFile)
|
||||
assert.FileExists(t, tlsConfig.CertFile)
|
||||
assert.FileExists(t, tlsConfig.CertKeyFile)
|
||||
|
||||
// Cleanup temporary files
|
||||
tlsManager.cleanupCertFiles(tlsConfig)
|
||||
assert.NoFileExists(t, tlsConfig.RootCertFile)
|
||||
assert.NoFileExists(t, tlsConfig.CertFile)
|
||||
assert.NoFileExists(t, tlsConfig.CertKeyFile)
|
||||
}
|
||||
|
||||
func TestTLSManager_WriteCertFile(t *testing.T) {
|
||||
logger := log.New()
|
||||
tlsManager := newPgxTlsManager(logger)
|
||||
|
||||
// Test writing a valid certificate file
|
||||
filePath, err := tlsManager.writeCertFile("test-*.crt", "test-cert-content")
|
||||
require.NoError(t, err)
|
||||
assert.FileExists(t, filePath)
|
||||
|
||||
content, err := os.ReadFile(filepath.Clean(filePath))
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "test-cert-content", string(content))
|
||||
|
||||
// Cleanup the file
|
||||
err = os.Remove(filePath)
|
||||
require.NoError(t, err)
|
||||
assert.NoFileExists(t, filePath)
|
||||
}
|
||||
|
||||
func TestTLSManager_CleanupCertFiles(t *testing.T) {
|
||||
logger := log.New()
|
||||
tlsManager := newPgxTlsManager(logger)
|
||||
|
||||
// Create temporary files for testing
|
||||
rootCertFile, err := tlsManager.writeCertFile("root-*.crt", "root-cert-content")
|
||||
require.NoError(t, err)
|
||||
clientCertFile, err := tlsManager.writeCertFile("client-*.crt", "client-cert-content")
|
||||
require.NoError(t, err)
|
||||
clientKeyFile, err := tlsManager.writeCertFile("client-*.key", "client-key-content")
|
||||
require.NoError(t, err)
|
||||
|
||||
tlsConfig := tlsSettings{
|
||||
ConfigurationMethod: "file-content",
|
||||
RootCertFile: rootCertFile,
|
||||
CertFile: clientCertFile,
|
||||
CertKeyFile: clientKeyFile,
|
||||
}
|
||||
|
||||
// Cleanup the files
|
||||
tlsManager.cleanupCertFiles(tlsConfig)
|
||||
|
||||
// Verify the files are deleted
|
||||
assert.NoFileExists(t, rootCertFile)
|
||||
assert.NoFileExists(t, clientCertFile)
|
||||
assert.NoFileExists(t, clientKeyFile)
|
||||
}
|
||||
@@ -15,27 +15,33 @@ import (
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend/instancemgmt"
|
||||
"github.com/grafana/grafana-plugin-sdk-go/data"
|
||||
"github.com/grafana/grafana-plugin-sdk-go/data/sqlutil"
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
"github.com/lib/pq"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend/log"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/grafana/grafana/pkg/tsdb/grafana-postgresql-datasource/sqleng"
|
||||
)
|
||||
|
||||
func ProvideService(cfg *setting.Cfg) *Service {
|
||||
func ProvideService(cfg *setting.Cfg, features featuremgmt.FeatureToggles) *Service {
|
||||
logger := backend.NewLoggerWith("logger", "tsdb.postgres")
|
||||
s := &Service{
|
||||
tlsManager: newTLSManager(logger, cfg.DataPath),
|
||||
logger: logger,
|
||||
tlsManager: newTLSManager(logger, cfg.DataPath),
|
||||
pgxTlsManager: newPgxTlsManager(logger),
|
||||
logger: logger,
|
||||
features: features,
|
||||
}
|
||||
s.im = datasource.NewInstanceManager(s.newInstanceSettings())
|
||||
return s
|
||||
}
|
||||
|
||||
type Service struct {
|
||||
tlsManager tlsSettingsProvider
|
||||
im instancemgmt.InstanceManager
|
||||
logger log.Logger
|
||||
tlsManager tlsSettingsProvider
|
||||
pgxTlsManager *pgxTlsManager
|
||||
im instancemgmt.InstanceManager
|
||||
logger log.Logger
|
||||
features featuremgmt.FeatureToggles
|
||||
}
|
||||
|
||||
func (s *Service) getDSInfo(ctx context.Context, pluginCtx backend.PluginContext) (*sqleng.DataSourceHandler, error) {
|
||||
@@ -52,6 +58,11 @@ func (s *Service) QueryData(ctx context.Context, req *backend.QueryDataRequest)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if s.features.IsEnabled(ctx, featuremgmt.FlagPostgresDSUsePGX) {
|
||||
return dsInfo.QueryDataPGX(ctx, req)
|
||||
}
|
||||
|
||||
return dsInfo.QueryData(ctx, req)
|
||||
}
|
||||
|
||||
@@ -104,6 +115,62 @@ func newPostgres(ctx context.Context, userFacingDefaultError string, rowLimit in
|
||||
return db, handler, nil
|
||||
}
|
||||
|
||||
func newPostgresPGX(ctx context.Context, userFacingDefaultError string, rowLimit int64, dsInfo sqleng.DataSourceInfo, cnnstr string, logger log.Logger, settings backend.DataSourceInstanceSettings) (*pgxpool.Pool, *sqleng.DataSourceHandler, error) {
|
||||
pgxConf, err := pgxpool.ParseConfig(cnnstr)
|
||||
if err != nil {
|
||||
logger.Error("postgres config creation failed", "error", err)
|
||||
return nil, nil, fmt.Errorf("postgres config creation failed")
|
||||
}
|
||||
|
||||
proxyClient, err := settings.ProxyClient(ctx)
|
||||
if err != nil {
|
||||
logger.Error("postgres proxy creation failed", "error", err)
|
||||
return nil, nil, fmt.Errorf("postgres proxy creation failed")
|
||||
}
|
||||
|
||||
if proxyClient.SecureSocksProxyEnabled() {
|
||||
dialer, err := proxyClient.NewSecureSocksProxyContextDialer()
|
||||
if err != nil {
|
||||
logger.Error("postgres proxy creation failed", "error", err)
|
||||
return nil, nil, fmt.Errorf("postgres proxy creation failed")
|
||||
}
|
||||
|
||||
pgxConf.ConnConfig.DialFunc = newPgxDialFunc(dialer)
|
||||
}
|
||||
|
||||
// by default pgx resolves hostnames to ip addresses. we must avoid this.
|
||||
// (certain socks-proxy related functionality relies on the hostname being preserved)
|
||||
pgxConf.ConnConfig.LookupFunc = func(_ context.Context, host string) ([]string, error) {
|
||||
return []string{host}, nil
|
||||
}
|
||||
|
||||
config := sqleng.DataPluginConfiguration{
|
||||
DSInfo: dsInfo,
|
||||
MetricColumnTypes: []string{"unknown", "text", "varchar", "char", "bpchar"},
|
||||
RowLimit: rowLimit,
|
||||
}
|
||||
|
||||
queryResultTransformer := postgresQueryResultTransformer{}
|
||||
pgxConf.MaxConnLifetime = time.Duration(config.DSInfo.JsonData.ConnMaxLifetime) * time.Second
|
||||
pgxConf.MaxConns = int32(config.DSInfo.JsonData.MaxOpenConns)
|
||||
|
||||
p, err := pgxpool.NewWithConfig(ctx, pgxConf)
|
||||
if err != nil {
|
||||
logger.Error("Failed connecting to Postgres", "err", err)
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
handler, err := sqleng.NewQueryDataHandlerPGX(userFacingDefaultError, p, config, &queryResultTransformer, newPostgresMacroEngine(dsInfo.JsonData.Timescaledb),
|
||||
logger)
|
||||
if err != nil {
|
||||
logger.Error("Failed connecting to Postgres", "err", err)
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
logger.Debug("Successfully connected to Postgres")
|
||||
return p, handler, nil
|
||||
}
|
||||
|
||||
func (s *Service) newInstanceSettings() datasource.InstanceFactoryFunc {
|
||||
logger := s.logger
|
||||
return func(ctx context.Context, settings backend.DataSourceInstanceSettings) (instancemgmt.Instance, error) {
|
||||
@@ -143,21 +210,45 @@ func (s *Service) newInstanceSettings() datasource.InstanceFactoryFunc {
|
||||
DecryptedSecureJSONData: settings.DecryptedSecureJSONData,
|
||||
}
|
||||
|
||||
cnnstr, err := s.generateConnectionString(dsInfo)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
isPGX := s.features.IsEnabled(ctx, featuremgmt.FlagPostgresDSUsePGX)
|
||||
|
||||
userFacingDefaultError, err := cfg.UserFacingDefaultError()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
_, handler, err := newPostgres(ctx, userFacingDefaultError, sqlCfg.RowLimit, dsInfo, cnnstr, logger, settings)
|
||||
var handler instancemgmt.Instance
|
||||
if isPGX {
|
||||
pgxTlsSettings, err := s.pgxTlsManager.getTLSSettings(dsInfo)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
logger.Error("Failed connecting to Postgres", "err", err)
|
||||
return nil, err
|
||||
// Ensure cleanupCertFiles is called after the connection is opened
|
||||
defer s.pgxTlsManager.cleanupCertFiles(pgxTlsSettings)
|
||||
cnnstr, err := s.generateConnectionString(dsInfo, pgxTlsSettings, isPGX)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
_, handler, err = newPostgresPGX(ctx, userFacingDefaultError, sqlCfg.RowLimit, dsInfo, cnnstr, logger, settings)
|
||||
if err != nil {
|
||||
logger.Error("Failed connecting to Postgres", "err", err)
|
||||
return nil, err
|
||||
}
|
||||
} else {
|
||||
tlsSettings, err := s.tlsManager.getTLSSettings(dsInfo)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
cnnstr, err := s.generateConnectionString(dsInfo, tlsSettings, isPGX)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
_, handler, err = newPostgres(ctx, userFacingDefaultError, sqlCfg.RowLimit, dsInfo, cnnstr, logger, settings)
|
||||
if err != nil {
|
||||
logger.Error("Failed connecting to Postgres", "err", err)
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
logger.Debug("Successfully connected to Postgres")
|
||||
@@ -170,65 +261,100 @@ func escape(input string) string {
|
||||
return strings.ReplaceAll(strings.ReplaceAll(input, `\`, `\\`), "'", `\'`)
|
||||
}
|
||||
|
||||
func (s *Service) generateConnectionString(dsInfo sqleng.DataSourceInfo) (string, error) {
|
||||
logger := s.logger
|
||||
var host string
|
||||
var port int
|
||||
type connectionParams struct {
|
||||
host string
|
||||
port int
|
||||
user string
|
||||
password string
|
||||
database string
|
||||
}
|
||||
|
||||
func parseConnectionParams(dsInfo sqleng.DataSourceInfo, logger log.Logger) (connectionParams, error) {
|
||||
var params connectionParams
|
||||
var err error
|
||||
|
||||
if strings.HasPrefix(dsInfo.URL, "/") {
|
||||
host = dsInfo.URL
|
||||
params.host = dsInfo.URL
|
||||
logger.Debug("Generating connection string with Unix socket specifier", "address", dsInfo.URL)
|
||||
} 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 {
|
||||
logger.Debug("Error parsing the IPv4 address", "address", dsInfo.URL)
|
||||
return "", sqleng.ErrParsingPostgresURL
|
||||
}
|
||||
logger.Debug("Generating IPv4 connection string with network host/port pair", "host", host, "port", port, "address", dsInfo.URL)
|
||||
} else {
|
||||
logger.Debug("Generating IPv4 connection string with network host", "host", host, "address", dsInfo.URL)
|
||||
}
|
||||
} 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 {
|
||||
logger.Debug("Error parsing the IPv6 address", "address", dsInfo.URL)
|
||||
return "", sqleng.ErrParsingPostgresURL
|
||||
}
|
||||
logger.Debug("Generating IPv6 connection string with network host/port pair", "host", host, "port", port, "address", dsInfo.URL)
|
||||
} else {
|
||||
host = dsInfo.URL[1 : len(dsInfo.URL)-1]
|
||||
logger.Debug("Generating IPv6 connection string with network host", "host", host, "address", dsInfo.URL)
|
||||
}
|
||||
params.host, params.port, err = parseNetworkAddress(dsInfo.URL, logger)
|
||||
if err != nil {
|
||||
return connectionParams{}, err
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
params.user = dsInfo.User
|
||||
params.password = dsInfo.DecryptedSecureJSONData["password"]
|
||||
params.database = dsInfo.Database
|
||||
|
||||
return params, nil
|
||||
}
|
||||
|
||||
func parseNetworkAddress(url string, logger log.Logger) (string, int, error) {
|
||||
index := strings.LastIndex(url, ":")
|
||||
v6Index := strings.Index(url, "]")
|
||||
sp := strings.SplitN(url, ":", 2)
|
||||
host := sp[0]
|
||||
port := 0
|
||||
|
||||
if v6Index == -1 {
|
||||
if len(sp) > 1 {
|
||||
var err error
|
||||
port, err = strconv.Atoi(sp[1])
|
||||
if err != nil {
|
||||
logger.Debug("Error parsing the IPv4 address", "address", url)
|
||||
return "", 0, sqleng.ErrParsingPostgresURL
|
||||
}
|
||||
logger.Debug("Generating IPv4 connection string with network host/port pair", "host", host, "port", port, "address", url)
|
||||
} else {
|
||||
logger.Debug("Generating IPv4 connection string with network host", "host", host, "address", url)
|
||||
}
|
||||
} else {
|
||||
if index == v6Index+1 {
|
||||
host = url[1 : index-1]
|
||||
var err error
|
||||
port, err = strconv.Atoi(url[index+1:])
|
||||
if err != nil {
|
||||
logger.Debug("Error parsing the IPv6 address", "address", url)
|
||||
return "", 0, sqleng.ErrParsingPostgresURL
|
||||
}
|
||||
logger.Debug("Generating IPv6 connection string with network host/port pair", "host", host, "port", port, "address", url)
|
||||
} else {
|
||||
host = url[1 : len(url)-1]
|
||||
logger.Debug("Generating IPv6 connection string with network host", "host", host, "address", url)
|
||||
}
|
||||
}
|
||||
|
||||
tlsSettings, err := s.tlsManager.getTLSSettings(dsInfo)
|
||||
return host, port, nil
|
||||
}
|
||||
|
||||
func buildBaseConnectionString(params connectionParams) string {
|
||||
connStr := fmt.Sprintf("user='%s' password='%s' host='%s' dbname='%s'",
|
||||
escape(params.user), escape(params.password), escape(params.host), escape(params.database))
|
||||
if params.port > 0 {
|
||||
connStr += fmt.Sprintf(" port=%d", params.port)
|
||||
}
|
||||
return connStr
|
||||
}
|
||||
|
||||
func (s *Service) generateConnectionString(dsInfo sqleng.DataSourceInfo, tlsSettings tlsSettings, isPGX bool) (string, error) {
|
||||
logger := s.logger
|
||||
|
||||
params, err := parseConnectionParams(dsInfo, logger)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
connStr := buildBaseConnectionString(params)
|
||||
|
||||
connStr += fmt.Sprintf(" sslmode='%s'", escape(tlsSettings.Mode))
|
||||
|
||||
// there is an issue with the lib/pq module, the `verify-ca` tls mode
|
||||
// does not work correctly. ( see https://github.com/lib/pq/issues/1106 )
|
||||
// to workaround the problem, if the `verify-ca` mode is chosen,
|
||||
// we disable sslsni.
|
||||
if tlsSettings.Mode == "verify-ca" {
|
||||
if tlsSettings.Mode == "verify-ca" && !isPGX {
|
||||
logger.Debug("Disabling sslsni for verify-ca mode")
|
||||
connStr += " sslsni=0"
|
||||
}
|
||||
|
||||
@@ -262,7 +388,7 @@ func (s *Service) CheckHealth(ctx context.Context, req *backend.CheckHealthReque
|
||||
if err != nil {
|
||||
return sqleng.ErrToHealthCheckResult(err)
|
||||
}
|
||||
return dsHandler.CheckHealth(ctx, req)
|
||||
return dsHandler.CheckHealth(ctx, req, s.features)
|
||||
}
|
||||
|
||||
func (t *postgresQueryResultTransformer) GetConverterList() []sqlutil.StringConverter {
|
||||
|
||||
@@ -0,0 +1,195 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend/log"
|
||||
"github.com/grafana/grafana-plugin-sdk-go/experimental"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/grafana/grafana/pkg/tsdb/grafana-postgresql-datasource/sqleng"
|
||||
)
|
||||
|
||||
// These tests require a real postgres database:
|
||||
// - make devenv sources=postgres_tests
|
||||
// - either set the env variable GRAFANA_TEST_DB = postgres
|
||||
// - or set `forceRun := true` below
|
||||
//
|
||||
// The tests require a PostgreSQL db named grafanadstest and a user/password grafanatest/grafanatest!
|
||||
// Use the docker/blocks/postgres_tests/docker-compose.yaml to spin up a
|
||||
// preconfigured Postgres server suitable for running these tests.
|
||||
func TestIntegrationPostgresPGXSnapshots(t *testing.T) {
|
||||
// the logic in this function is copied from postgres_tests.go
|
||||
shouldRunTest := func() bool {
|
||||
if testing.Short() {
|
||||
return false
|
||||
}
|
||||
|
||||
testDbName, present := os.LookupEnv("GRAFANA_TEST_DB")
|
||||
|
||||
if present && testDbName == "postgres" {
|
||||
return true
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
if !shouldRunTest() {
|
||||
t.Skip()
|
||||
}
|
||||
|
||||
getCnnStr := func() string {
|
||||
host := os.Getenv("POSTGRES_HOST")
|
||||
if host == "" {
|
||||
host = "localhost"
|
||||
}
|
||||
port := os.Getenv("POSTGRES_PORT")
|
||||
if port == "" {
|
||||
port = "5432"
|
||||
}
|
||||
|
||||
return fmt.Sprintf("user=grafanatest password=grafanatest host=%s port=%s dbname=grafanadstest sslmode=disable",
|
||||
host, port)
|
||||
}
|
||||
|
||||
sqlQueryCommentRe := regexp.MustCompile(`^-- (.+)\n`)
|
||||
|
||||
readSqlFile := func(path string) (string, string) {
|
||||
// the file-path is not coming from the outside,
|
||||
// it is hardcoded in this file.
|
||||
//nolint:gosec
|
||||
sqlBytes, err := os.ReadFile(path)
|
||||
require.NoError(t, err)
|
||||
|
||||
sql := string(sqlBytes)
|
||||
|
||||
// first line of the file contains the sql query to run, commented out
|
||||
match := sqlQueryCommentRe.FindStringSubmatch(sql)
|
||||
require.Len(t, match, 2)
|
||||
|
||||
rawSQL := strings.TrimSpace(match[1])
|
||||
|
||||
return rawSQL, sql
|
||||
}
|
||||
|
||||
makeQuery := func(rawSQL string, format string) backend.QueryDataRequest {
|
||||
queryData := map[string]string{
|
||||
"rawSql": rawSQL,
|
||||
"format": format,
|
||||
}
|
||||
|
||||
queryBytes, err := json.Marshal(queryData)
|
||||
require.NoError(t, err)
|
||||
|
||||
return backend.QueryDataRequest{
|
||||
Queries: []backend.DataQuery{
|
||||
{
|
||||
JSON: queryBytes,
|
||||
RefID: "A",
|
||||
TimeRange: backend.TimeRange{
|
||||
From: time.Date(2023, 12, 24, 14, 15, 22, 123456, time.UTC),
|
||||
To: time.Date(2023, 12, 24, 14, 45, 13, 876543, time.UTC),
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
tt := []struct {
|
||||
name string
|
||||
format string
|
||||
}{
|
||||
{format: "time_series", name: "simple"},
|
||||
{format: "time_series", name: "no_rows_long"},
|
||||
{format: "time_series", name: "no_rows_wide"},
|
||||
{format: "time_series", name: "7x_compat_metric_label"},
|
||||
{format: "time_series", name: "convert_to_float64"},
|
||||
{format: "time_series", name: "convert_to_float64_not"},
|
||||
{format: "time_series", name: "fill_null"},
|
||||
{format: "time_series", name: "fill_previous"},
|
||||
{format: "time_series", name: "fill_value"},
|
||||
{format: "time_series", name: "fill_value_wide"},
|
||||
{format: "table", name: "simple"},
|
||||
{format: "table", name: "multi_stat1"},
|
||||
{format: "table", name: "multi_stat2"},
|
||||
{format: "table", name: "no_rows"},
|
||||
{format: "table", name: "types_numeric"},
|
||||
{format: "table", name: "types_char"},
|
||||
{format: "table", name: "types_datetime_pgx"},
|
||||
{format: "table", name: "types_other"},
|
||||
{format: "table", name: "timestamp_convert_bigint"},
|
||||
{format: "table", name: "timestamp_convert_integer"},
|
||||
{format: "table", name: "timestamp_convert_real"},
|
||||
{format: "table", name: "timestamp_convert_double"},
|
||||
{format: "table", name: "time_group_compat_case1"},
|
||||
{format: "table", name: "time_group_compat_case2"},
|
||||
}
|
||||
|
||||
for _, test := range tt {
|
||||
require.True(t, test.format == "table" || test.format == "time_series")
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
origInterpolate := sqleng.Interpolate
|
||||
t.Cleanup(func() {
|
||||
sqleng.Interpolate = origInterpolate
|
||||
})
|
||||
|
||||
sqleng.Interpolate = func(query backend.DataQuery, timeRange backend.TimeRange, timeInterval string, sql string) string {
|
||||
return sql
|
||||
}
|
||||
|
||||
jsonData := sqleng.JsonData{
|
||||
MaxOpenConns: 10,
|
||||
MaxIdleConns: 2,
|
||||
ConnMaxLifetime: 14400,
|
||||
Timescaledb: false,
|
||||
Mode: "disable",
|
||||
ConfigurationMethod: "file-path",
|
||||
}
|
||||
|
||||
dsInfo := sqleng.DataSourceInfo{
|
||||
JsonData: jsonData,
|
||||
DecryptedSecureJSONData: map[string]string{},
|
||||
}
|
||||
|
||||
logger := log.New()
|
||||
|
||||
cnnstr := getCnnStr()
|
||||
|
||||
p, handler, err := newPostgresPGX(context.Background(), "error", 10000, dsInfo, cnnstr, logger, backend.DataSourceInstanceSettings{})
|
||||
|
||||
t.Cleanup((func() {
|
||||
_, err := p.Exec(context.Background(), "DROP TABLE tbl")
|
||||
require.NoError(t, err)
|
||||
p.Close()
|
||||
}))
|
||||
|
||||
require.NoError(t, err)
|
||||
|
||||
sqlFilePath := filepath.Join("testdata", test.format, test.name+".sql")
|
||||
goldenFileName := filepath.Join(test.format, test.name+".golden")
|
||||
|
||||
rawSQL, sql := readSqlFile(sqlFilePath)
|
||||
|
||||
_, err = p.Exec(context.Background(), sql)
|
||||
require.NoError(t, err)
|
||||
|
||||
query := makeQuery(rawSQL, test.format)
|
||||
|
||||
result, err := handler.QueryDataPGX(context.Background(), &query)
|
||||
require.Len(t, result.Responses, 1)
|
||||
response, found := result.Responses["A"]
|
||||
require.True(t, found)
|
||||
require.NoError(t, err)
|
||||
experimental.CheckGoldenJSONResponse(t, "testdata", goldenFileName, &response, updateGoldenFiles)
|
||||
})
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -159,7 +159,7 @@ func TestIntegrationGenerateConnectionString(t *testing.T) {
|
||||
UID: tt.uid,
|
||||
}
|
||||
|
||||
connStr, err := svc.generateConnectionString(ds)
|
||||
connStr, err := svc.generateConnectionString(ds, tt.tlsSettings, false)
|
||||
|
||||
if tt.expErr == "" {
|
||||
require.NoError(t, err, tt.desc)
|
||||
|
||||
@@ -33,3 +33,11 @@ func (p *postgresProxyDialer) DialTimeout(network, address string, timeout time.
|
||||
|
||||
return p.d.(proxy.ContextDialer).DialContext(ctx, network, address)
|
||||
}
|
||||
|
||||
type PgxDialFunc = func(ctx context.Context, network string, address string) (net.Conn, error)
|
||||
|
||||
func newPgxDialFunc(dialer proxy.Dialer) PgxDialFunc {
|
||||
return func(ctx context.Context, network string, addr string) (net.Conn, error) {
|
||||
return dialer.Dial(network, addr)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -10,11 +10,17 @@ import (
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend/log"
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt"
|
||||
"github.com/lib/pq"
|
||||
)
|
||||
|
||||
func (e *DataSourceHandler) CheckHealth(ctx context.Context, req *backend.CheckHealthRequest) (*backend.CheckHealthResult, error) {
|
||||
err := e.Ping()
|
||||
func (e *DataSourceHandler) CheckHealth(ctx context.Context, req *backend.CheckHealthRequest, features featuremgmt.FeatureToggles) (*backend.CheckHealthResult, error) {
|
||||
var err error
|
||||
if features.IsEnabled(ctx, featuremgmt.FlagPostgresDSUsePGX) {
|
||||
err = e.PingPGX(ctx)
|
||||
} else {
|
||||
err = e.Ping()
|
||||
}
|
||||
if err != nil {
|
||||
logCheckHealthError(ctx, e.dsInfo, err)
|
||||
if strings.EqualFold(req.PluginContext.User.Role, "Admin") {
|
||||
@@ -63,6 +69,7 @@ func ErrToHealthCheckResult(err error) (*backend.CheckHealthResult, error) {
|
||||
res.Message += fmt.Sprintf(". Error message: %s", errMessage)
|
||||
}
|
||||
}
|
||||
|
||||
if errors.Is(err, pq.ErrSSLNotSupported) {
|
||||
res.Message = "SSL error: Failed to connect to the server"
|
||||
}
|
||||
@@ -125,10 +132,10 @@ func logCheckHealthError(ctx context.Context, dsInfo DataSourceInfo, err error)
|
||||
"config_tls_client_cert_length": len(dsInfo.DecryptedSecureJSONData["tlsClientCert"]),
|
||||
"config_tls_client_key_length": len(dsInfo.DecryptedSecureJSONData["tlsClientKey"]),
|
||||
}
|
||||
configSummaryJson, marshalError := json.Marshal(configSummary)
|
||||
configSummaryJSON, marshalError := json.Marshal(configSummary)
|
||||
if marshalError != nil {
|
||||
logger.Error("Check health failed", "error", err, "message_type", "ds_config_health_check_error")
|
||||
return
|
||||
}
|
||||
logger.Error("Check health failed", "error", err, "message_type", "ds_config_health_check_error_detailed", "details", string(configSummaryJson))
|
||||
logger.Error("Check health failed", "error", err, "message_type", "ds_config_health_check_error_detailed", "details", string(configSummaryJSON))
|
||||
}
|
||||
|
||||
@@ -19,6 +19,7 @@ import (
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend/log"
|
||||
"github.com/grafana/grafana-plugin-sdk-go/data"
|
||||
"github.com/grafana/grafana-plugin-sdk-go/data/sqlutil"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// MetaKeyExecutedQueryString is the key where the executed query should get stored
|
||||
@@ -88,6 +89,7 @@ type DataSourceHandler struct {
|
||||
dsInfo DataSourceInfo
|
||||
rowLimit int64
|
||||
userError string
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
type QueryJson struct {
|
||||
@@ -489,6 +491,7 @@ type dataQueryModel struct {
|
||||
Interval time.Duration
|
||||
columnNames []string
|
||||
columnTypes []*sql.ColumnType
|
||||
columnTypesPGX []string
|
||||
timeIndex int
|
||||
timeEndIndex int
|
||||
metricIndex int
|
||||
|
||||
@@ -0,0 +1,554 @@
|
||||
package sqleng
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"runtime/debug"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend/log"
|
||||
"github.com/grafana/grafana-plugin-sdk-go/data"
|
||||
"github.com/grafana/grafana-plugin-sdk-go/data/sqlutil"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func NewQueryDataHandlerPGX(userFacingDefaultError string, p *pgxpool.Pool, config DataPluginConfiguration, queryResultTransformer SqlQueryResultTransformer,
|
||||
macroEngine SQLMacroEngine, log log.Logger) (*DataSourceHandler, error) {
|
||||
queryDataHandler := DataSourceHandler{
|
||||
queryResultTransformer: queryResultTransformer,
|
||||
macroEngine: macroEngine,
|
||||
timeColumnNames: []string{"time"},
|
||||
log: log,
|
||||
dsInfo: config.DSInfo,
|
||||
rowLimit: config.RowLimit,
|
||||
userError: userFacingDefaultError,
|
||||
}
|
||||
|
||||
if len(config.TimeColumnNames) > 0 {
|
||||
queryDataHandler.timeColumnNames = config.TimeColumnNames
|
||||
}
|
||||
|
||||
if len(config.MetricColumnTypes) > 0 {
|
||||
queryDataHandler.metricColumnTypes = config.MetricColumnTypes
|
||||
}
|
||||
|
||||
queryDataHandler.pool = p
|
||||
return &queryDataHandler, nil
|
||||
}
|
||||
|
||||
func (e *DataSourceHandler) DisposePGX() {
|
||||
e.log.Debug("Disposing DB...")
|
||||
|
||||
if e.pool != nil {
|
||||
e.pool.Close()
|
||||
}
|
||||
|
||||
e.log.Debug("DB disposed")
|
||||
}
|
||||
|
||||
func (e *DataSourceHandler) PingPGX(ctx context.Context) error {
|
||||
return e.pool.Ping(ctx)
|
||||
}
|
||||
|
||||
func (e *DataSourceHandler) QueryDataPGX(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
|
||||
result := backend.NewQueryDataResponse()
|
||||
ch := make(chan DBDataResponse, len(req.Queries))
|
||||
var wg sync.WaitGroup
|
||||
// Execute each query in a goroutine and wait for them to finish afterwards
|
||||
for _, query := range req.Queries {
|
||||
queryjson := QueryJson{
|
||||
Fill: false,
|
||||
Format: "time_series",
|
||||
}
|
||||
err := json.Unmarshal(query.JSON, &queryjson)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error unmarshal query json: %w", err)
|
||||
}
|
||||
|
||||
// the fill-params are only stored inside this function, during query-interpolation. we do not support
|
||||
// sending them in "from the outside"
|
||||
if queryjson.Fill || queryjson.FillInterval != 0.0 || queryjson.FillMode != "" || queryjson.FillValue != 0.0 {
|
||||
return nil, fmt.Errorf("query fill-parameters not supported")
|
||||
}
|
||||
|
||||
if queryjson.RawSql == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
wg.Add(1)
|
||||
go e.executeQueryPGX(ctx, query, &wg, ch, queryjson)
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
|
||||
// Read results from channels
|
||||
close(ch)
|
||||
result.Responses = make(map[string]backend.DataResponse)
|
||||
for queryResult := range ch {
|
||||
result.Responses[queryResult.refID] = queryResult.dataResponse
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (e *DataSourceHandler) handleQueryError(frameErr string, err error, query string, source backend.ErrorSource, ch chan DBDataResponse, queryResult DBDataResponse) {
|
||||
var emptyFrame data.Frame
|
||||
emptyFrame.SetMeta(&data.FrameMeta{ExecutedQueryString: query})
|
||||
if backend.IsDownstreamError(err) {
|
||||
source = backend.ErrorSourceDownstream
|
||||
}
|
||||
queryResult.dataResponse.Error = fmt.Errorf("%s: %w", frameErr, err)
|
||||
queryResult.dataResponse.ErrorSource = source
|
||||
queryResult.dataResponse.Frames = data.Frames{&emptyFrame}
|
||||
ch <- queryResult
|
||||
}
|
||||
|
||||
func (e *DataSourceHandler) handlePanic(logger log.Logger, queryResult *DBDataResponse, ch chan DBDataResponse) {
|
||||
if r := recover(); r != nil {
|
||||
logger.Error("ExecuteQuery panic", "error", r, "stack", string(debug.Stack()))
|
||||
if theErr, ok := r.(error); ok {
|
||||
queryResult.dataResponse.Error = theErr
|
||||
queryResult.dataResponse.ErrorSource = backend.ErrorSourcePlugin
|
||||
} else if theErrString, ok := r.(string); ok {
|
||||
queryResult.dataResponse.Error = errors.New(theErrString)
|
||||
queryResult.dataResponse.ErrorSource = backend.ErrorSourcePlugin
|
||||
} else {
|
||||
queryResult.dataResponse.Error = fmt.Errorf("unexpected error - %s", e.userError)
|
||||
queryResult.dataResponse.ErrorSource = backend.ErrorSourceDownstream
|
||||
}
|
||||
ch <- *queryResult
|
||||
}
|
||||
}
|
||||
|
||||
func (e *DataSourceHandler) execQuery(ctx context.Context, query string, logger log.Logger) ([]*pgconn.Result, error) {
|
||||
c, err := e.pool.Acquire(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to acquire connection: %w", err)
|
||||
}
|
||||
defer c.Release()
|
||||
|
||||
mrr := c.Conn().PgConn().Exec(ctx, query)
|
||||
defer func() {
|
||||
if err := mrr.Close(); err != nil {
|
||||
logger.Warn("Failed to close multi-result reader", "error", err)
|
||||
}
|
||||
}()
|
||||
return mrr.ReadAll()
|
||||
}
|
||||
|
||||
func (e *DataSourceHandler) executeQueryPGX(queryContext context.Context, query backend.DataQuery, wg *sync.WaitGroup,
|
||||
ch chan DBDataResponse, queryJSON QueryJson) {
|
||||
defer wg.Done()
|
||||
queryResult := DBDataResponse{
|
||||
dataResponse: backend.DataResponse{},
|
||||
refID: query.RefID,
|
||||
}
|
||||
|
||||
logger := e.log.FromContext(queryContext)
|
||||
defer e.handlePanic(logger, &queryResult, ch)
|
||||
|
||||
if queryJSON.RawSql == "" {
|
||||
panic("Query model property rawSql should not be empty at this point")
|
||||
}
|
||||
|
||||
// global substitutions
|
||||
interpolatedQuery := Interpolate(query, query.TimeRange, e.dsInfo.JsonData.TimeInterval, queryJSON.RawSql)
|
||||
|
||||
// data source specific substitutions
|
||||
interpolatedQuery, err := e.macroEngine.Interpolate(&query, query.TimeRange, interpolatedQuery)
|
||||
if err != nil {
|
||||
e.handleQueryError("interpolation failed", e.TransformQueryError(logger, err), interpolatedQuery, backend.ErrorSourcePlugin, ch, queryResult)
|
||||
return
|
||||
}
|
||||
|
||||
results, err := e.execQuery(queryContext, interpolatedQuery, logger)
|
||||
if err != nil {
|
||||
e.handleQueryError("db query error", e.TransformQueryError(logger, err), interpolatedQuery, backend.ErrorSourcePlugin, ch, queryResult)
|
||||
return
|
||||
}
|
||||
|
||||
qm, err := e.newProcessCfgPGX(queryContext, query, results, interpolatedQuery)
|
||||
if err != nil {
|
||||
e.handleQueryError("failed to get configurations", err, interpolatedQuery, backend.ErrorSourcePlugin, ch, queryResult)
|
||||
return
|
||||
}
|
||||
|
||||
frame, err := convertResultsToFrame(results, e.rowLimit)
|
||||
if err != nil {
|
||||
e.handleQueryError("convert frame from rows error", err, interpolatedQuery, backend.ErrorSourcePlugin, ch, queryResult)
|
||||
return
|
||||
}
|
||||
|
||||
e.processFrame(frame, qm, queryResult, ch, logger)
|
||||
}
|
||||
|
||||
func (e *DataSourceHandler) processFrame(frame *data.Frame, qm *dataQueryModel, queryResult DBDataResponse, ch chan DBDataResponse, logger log.Logger) {
|
||||
if frame.Meta == nil {
|
||||
frame.Meta = &data.FrameMeta{}
|
||||
}
|
||||
frame.Meta.ExecutedQueryString = qm.InterpolatedQuery
|
||||
|
||||
// If no rows were returned, clear any previously set `Fields` with a single empty `data.Field` slice.
|
||||
// Then assign `queryResult.dataResponse.Frames` the current single frame with that single empty Field.
|
||||
// This assures 1) our visualization doesn't display unwanted empty fields, and also that 2)
|
||||
// additionally-needed frame data stays intact and is correctly passed to our visulization.
|
||||
if frame.Rows() == 0 {
|
||||
frame.Fields = []*data.Field{}
|
||||
queryResult.dataResponse.Frames = data.Frames{frame}
|
||||
ch <- queryResult
|
||||
return
|
||||
}
|
||||
|
||||
if err := convertSQLTimeColumnsToEpochMS(frame, qm); err != nil {
|
||||
e.handleQueryError("converting time columns failed", err, qm.InterpolatedQuery, backend.ErrorSourcePlugin, ch, queryResult)
|
||||
return
|
||||
}
|
||||
|
||||
if qm.Format == dataQueryFormatSeries {
|
||||
// time series has to have time column
|
||||
if qm.timeIndex == -1 {
|
||||
e.handleQueryError("db has no time column", errors.New("time column is missing; make sure your data includes a time column for time series format or switch to a table format that doesn't require it"), qm.InterpolatedQuery, backend.ErrorSourceDownstream, ch, queryResult)
|
||||
return
|
||||
}
|
||||
|
||||
// Make sure to name the time field 'Time' to be backward compatible with Grafana pre-v8.
|
||||
frame.Fields[qm.timeIndex].Name = data.TimeSeriesTimeFieldName
|
||||
|
||||
for i := range qm.columnNames {
|
||||
if i == qm.timeIndex || i == qm.metricIndex {
|
||||
continue
|
||||
}
|
||||
|
||||
if t := frame.Fields[i].Type(); t == data.FieldTypeString || t == data.FieldTypeNullableString {
|
||||
continue
|
||||
}
|
||||
|
||||
var err error
|
||||
if frame, err = convertSQLValueColumnToFloat(frame, i); err != nil {
|
||||
e.handleQueryError("convert value to float failed", err, qm.InterpolatedQuery, backend.ErrorSourcePlugin, ch, queryResult)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
tsSchema := frame.TimeSeriesSchema()
|
||||
if tsSchema.Type == data.TimeSeriesTypeLong {
|
||||
var err error
|
||||
originalData := frame
|
||||
frame, err = data.LongToWide(frame, qm.FillMissing)
|
||||
if err != nil {
|
||||
e.handleQueryError("failed to convert long to wide series when converting from dataframe", err, qm.InterpolatedQuery, backend.ErrorSourcePlugin, ch, queryResult)
|
||||
return
|
||||
}
|
||||
|
||||
// Before 8x, a special metric column was used to name time series. The LongToWide transforms that into a metric label on the value field.
|
||||
// But that makes series name have both the value column name AND the metric name. So here we are removing the metric label here and moving it to the
|
||||
// field name to get the same naming for the series as pre v8
|
||||
if len(originalData.Fields) == 3 {
|
||||
for _, field := range frame.Fields {
|
||||
if len(field.Labels) == 1 { // 7x only supported one label
|
||||
name, ok := field.Labels["metric"]
|
||||
if ok {
|
||||
field.Name = name
|
||||
field.Labels = nil
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if qm.FillMissing != nil {
|
||||
// we align the start-time
|
||||
startUnixTime := qm.TimeRange.From.Unix() / int64(qm.Interval.Seconds()) * int64(qm.Interval.Seconds())
|
||||
alignedTimeRange := backend.TimeRange{
|
||||
From: time.Unix(startUnixTime, 0),
|
||||
To: qm.TimeRange.To,
|
||||
}
|
||||
|
||||
var err error
|
||||
frame, err = sqlutil.ResampleWideFrame(frame, qm.FillMissing, alignedTimeRange, qm.Interval)
|
||||
if err != nil {
|
||||
logger.Error("Failed to resample dataframe", "err", err)
|
||||
frame.AppendNotices(data.Notice{Text: "Failed to resample dataframe", Severity: data.NoticeSeverityWarning})
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
queryResult.dataResponse.Frames = data.Frames{frame}
|
||||
ch <- queryResult
|
||||
}
|
||||
|
||||
func (e *DataSourceHandler) newProcessCfgPGX(queryContext context.Context, query backend.DataQuery,
|
||||
results []*pgconn.Result, interpolatedQuery string) (*dataQueryModel, error) {
|
||||
columnNames := []string{}
|
||||
columnTypesPGX := []string{}
|
||||
|
||||
// The results will contain column information in the metadata
|
||||
for _, result := range results {
|
||||
// Get column names from the result metadata
|
||||
for _, field := range result.FieldDescriptions {
|
||||
columnNames = append(columnNames, field.Name)
|
||||
pqtype, ok := pgtype.NewMap().TypeForOID(field.DataTypeOID)
|
||||
if !ok {
|
||||
// Handle special cases for field types
|
||||
switch field.DataTypeOID {
|
||||
case pgtype.TimetzOID:
|
||||
columnTypesPGX = append(columnTypesPGX, "timetz")
|
||||
case 790:
|
||||
columnTypesPGX = append(columnTypesPGX, "money")
|
||||
default:
|
||||
return nil, fmt.Errorf("unknown data type oid: %d", field.DataTypeOID)
|
||||
}
|
||||
} else {
|
||||
columnTypesPGX = append(columnTypesPGX, pqtype.Name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
qm := &dataQueryModel{
|
||||
columnTypesPGX: columnTypesPGX,
|
||||
columnNames: columnNames,
|
||||
timeIndex: -1,
|
||||
timeEndIndex: -1,
|
||||
metricIndex: -1,
|
||||
metricPrefix: false,
|
||||
queryContext: queryContext,
|
||||
}
|
||||
|
||||
queryJSON := QueryJson{}
|
||||
err := json.Unmarshal(query.JSON, &queryJSON)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if queryJSON.Fill {
|
||||
qm.FillMissing = &data.FillMissing{}
|
||||
qm.Interval = time.Duration(queryJSON.FillInterval * float64(time.Second))
|
||||
switch strings.ToLower(queryJSON.FillMode) {
|
||||
case "null":
|
||||
qm.FillMissing.Mode = data.FillModeNull
|
||||
case "previous":
|
||||
qm.FillMissing.Mode = data.FillModePrevious
|
||||
case "value":
|
||||
qm.FillMissing.Mode = data.FillModeValue
|
||||
qm.FillMissing.Value = queryJSON.FillValue
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
qm.TimeRange.From = query.TimeRange.From.UTC()
|
||||
qm.TimeRange.To = query.TimeRange.To.UTC()
|
||||
|
||||
switch queryJSON.Format {
|
||||
case "time_series":
|
||||
qm.Format = dataQueryFormatSeries
|
||||
case "table":
|
||||
qm.Format = dataQueryFormatTable
|
||||
default:
|
||||
panic(fmt.Sprintf("Unrecognized query model format: %q", queryJSON.Format))
|
||||
}
|
||||
|
||||
for i, col := range qm.columnNames {
|
||||
for _, tc := range e.timeColumnNames {
|
||||
if col == tc {
|
||||
qm.timeIndex = i
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if qm.Format == dataQueryFormatTable && strings.EqualFold(col, "timeend") {
|
||||
qm.timeEndIndex = i
|
||||
continue
|
||||
}
|
||||
|
||||
switch col {
|
||||
case "metric":
|
||||
qm.metricIndex = i
|
||||
default:
|
||||
if qm.metricIndex == -1 {
|
||||
columnType := qm.columnTypesPGX[i]
|
||||
for _, mct := range e.metricColumnTypes {
|
||||
if columnType == mct {
|
||||
qm.metricIndex = i
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
qm.InterpolatedQuery = interpolatedQuery
|
||||
return qm, nil
|
||||
}
|
||||
|
||||
func convertResultsToFrame(results []*pgconn.Result, rowLimit int64) (*data.Frame, error) {
|
||||
frame := data.Frame{}
|
||||
m := pgtype.NewMap()
|
||||
|
||||
for _, result := range results {
|
||||
// Skip non-select statements
|
||||
if !result.CommandTag.Select() {
|
||||
continue
|
||||
}
|
||||
fields := make(data.Fields, len(result.FieldDescriptions))
|
||||
|
||||
fieldTypes, err := getFieldTypesFromDescriptions(result.FieldDescriptions, m)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for i, v := range result.FieldDescriptions {
|
||||
fields[i] = data.NewFieldFromFieldType(fieldTypes[i], 0)
|
||||
fields[i].Name = v.Name
|
||||
}
|
||||
// Create a new frame
|
||||
frame = *data.NewFrame("", fields...)
|
||||
}
|
||||
|
||||
// Add rows to the frame
|
||||
for _, result := range results {
|
||||
// Skip non-select statements
|
||||
if !result.CommandTag.Select() {
|
||||
continue
|
||||
}
|
||||
fieldDescriptions := result.FieldDescriptions
|
||||
for rowIdx := range result.Rows {
|
||||
if rowIdx == int(rowLimit) {
|
||||
frame.AppendNotices(data.Notice{
|
||||
Severity: data.NoticeSeverityWarning,
|
||||
Text: fmt.Sprintf("Results have been limited to %v because the SQL row limit was reached", rowLimit),
|
||||
})
|
||||
break
|
||||
}
|
||||
row := make([]interface{}, len(fieldDescriptions))
|
||||
for colIdx, fd := range fieldDescriptions {
|
||||
rawValue := result.Rows[rowIdx][colIdx]
|
||||
dataTypeOID := fd.DataTypeOID
|
||||
format := fd.Format
|
||||
|
||||
if rawValue == nil {
|
||||
row[colIdx] = nil
|
||||
continue
|
||||
}
|
||||
|
||||
// Convert based on type
|
||||
switch fd.DataTypeOID {
|
||||
case pgtype.Int2OID:
|
||||
var d *int16
|
||||
scanPlan := m.PlanScan(dataTypeOID, format, &d)
|
||||
err := scanPlan.Scan(rawValue, &d)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
row[colIdx] = d
|
||||
case pgtype.Int4OID:
|
||||
var d *int32
|
||||
scanPlan := m.PlanScan(dataTypeOID, format, &d)
|
||||
err := scanPlan.Scan(rawValue, &d)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
row[colIdx] = d
|
||||
case pgtype.Int8OID:
|
||||
var d *int64
|
||||
scanPlan := m.PlanScan(dataTypeOID, format, &d)
|
||||
err := scanPlan.Scan(rawValue, &d)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
row[colIdx] = d
|
||||
case pgtype.NumericOID, pgtype.Float8OID, pgtype.Float4OID:
|
||||
var d *float64
|
||||
scanPlan := m.PlanScan(dataTypeOID, format, &d)
|
||||
err := scanPlan.Scan(rawValue, &d)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
row[colIdx] = d
|
||||
case pgtype.BoolOID:
|
||||
var d *bool
|
||||
scanPlan := m.PlanScan(dataTypeOID, format, &d)
|
||||
err := scanPlan.Scan(rawValue, &d)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
row[colIdx] = d
|
||||
case pgtype.ByteaOID:
|
||||
d, err := pgtype.ByteaCodec.DecodeValue(pgtype.ByteaCodec{}, m, dataTypeOID, format, rawValue)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
str := string(d.([]byte))
|
||||
row[colIdx] = &str
|
||||
case pgtype.TimestampOID, pgtype.TimestamptzOID, pgtype.DateOID:
|
||||
var d *time.Time
|
||||
scanPlan := m.PlanScan(dataTypeOID, format, &d)
|
||||
err := scanPlan.Scan(rawValue, &d)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
row[colIdx] = d
|
||||
case pgtype.TimeOID, pgtype.TimetzOID:
|
||||
var d *string
|
||||
scanPlan := m.PlanScan(dataTypeOID, format, &d)
|
||||
err := scanPlan.Scan(rawValue, &d)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
row[colIdx] = d
|
||||
default:
|
||||
var d *string
|
||||
scanPlan := m.PlanScan(dataTypeOID, format, &d)
|
||||
err := scanPlan.Scan(rawValue, &d)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
row[colIdx] = d
|
||||
}
|
||||
}
|
||||
frame.AppendRow(row...)
|
||||
}
|
||||
}
|
||||
|
||||
return &frame, nil
|
||||
}
|
||||
|
||||
func getFieldTypesFromDescriptions(fieldDescriptions []pgconn.FieldDescription, m *pgtype.Map) ([]data.FieldType, error) {
|
||||
fieldTypes := make([]data.FieldType, len(fieldDescriptions))
|
||||
for i, v := range fieldDescriptions {
|
||||
typeName, ok := m.TypeForOID(v.DataTypeOID)
|
||||
if !ok {
|
||||
// Handle special cases for field types
|
||||
if v.DataTypeOID == pgtype.TimetzOID || v.DataTypeOID == 790 {
|
||||
fieldTypes[i] = data.FieldTypeNullableString
|
||||
} else {
|
||||
return nil, fmt.Errorf("unknown data type oid: %d", v.DataTypeOID)
|
||||
}
|
||||
} else {
|
||||
switch typeName.Name {
|
||||
case "int2":
|
||||
fieldTypes[i] = data.FieldTypeNullableInt16
|
||||
case "int4":
|
||||
fieldTypes[i] = data.FieldTypeNullableInt32
|
||||
case "int8":
|
||||
fieldTypes[i] = data.FieldTypeNullableInt64
|
||||
case "float4", "float8", "numeric":
|
||||
fieldTypes[i] = data.FieldTypeNullableFloat64
|
||||
case "bool":
|
||||
fieldTypes[i] = data.FieldTypeNullableBool
|
||||
case "timestamptz", "timestamp", "date":
|
||||
fieldTypes[i] = data.FieldTypeNullableTime
|
||||
case "json", "jsonb":
|
||||
fieldTypes[i] = data.FieldTypeNullableJSON
|
||||
default:
|
||||
fieldTypes[i] = data.FieldTypeNullableString
|
||||
}
|
||||
}
|
||||
}
|
||||
return fieldTypes, nil
|
||||
}
|
||||
+214
@@ -0,0 +1,214 @@
|
||||
// 🌟 This was machine generated. Do not edit. 🌟
|
||||
//
|
||||
// Frame[0] {
|
||||
// "typeVersion": [
|
||||
// 0,
|
||||
// 0
|
||||
// ],
|
||||
// "executedQueryString": "SELECT * FROM tbl"
|
||||
// }
|
||||
// Name:
|
||||
// Dimensions: 12 Fields by 2 Rows
|
||||
// +--------------------------------------+--------------------------------------+----------------------------------------+----------------------------------------+-------------------------------+-------------------------------+-----------------+-----------------+--------------------+--------------------+-----------------+-----------------+
|
||||
// | Name: ts | Name: tsnn | Name: tsz | Name: tsznn | Name: d | Name: dnn | Name: t | Name: tnn | Name: tz | Name: tznn | Name: i | Name: inn |
|
||||
// | Labels: | Labels: | Labels: | Labels: | Labels: | Labels: | Labels: | Labels: | Labels: | Labels: | Labels: | Labels: |
|
||||
// | Type: []*time.Time | Type: []*time.Time | Type: []*time.Time | Type: []*time.Time | Type: []*time.Time | Type: []*time.Time | Type: []*string | Type: []*string | Type: []*string | Type: []*string | Type: []*string | Type: []*string |
|
||||
// +--------------------------------------+--------------------------------------+----------------------------------------+----------------------------------------+-------------------------------+-------------------------------+-----------------+-----------------+--------------------+--------------------+-----------------+-----------------+
|
||||
// | 2023-11-15 05:06:07.123456 +0000 UTC | 2023-11-15 05:06:08.123456 +0000 UTC | 2021-07-22 11:22:33.654321 +0000 +0000 | 2021-07-22 11:22:34.654321 +0000 +0000 | 2023-12-20 00:00:00 +0000 UTC | 2023-12-21 00:00:00 +0000 UTC | 12:34:56.234567 | 12:34:57.234567 | 23:12:36.765432+01 | 23:12:37.765432+01 | 00:00:00.987654 | 00:00:00.887654 |
|
||||
// | null | 2023-11-15 05:06:09.123456 +0000 UTC | null | 2021-07-22 11:22:35.654321 +0000 +0000 | null | 2023-12-22 00:00:00 +0000 UTC | null | 12:34:58.234567 | null | 23:12:38.765432+01 | null | 00:00:00.787654 |
|
||||
// +--------------------------------------+--------------------------------------+----------------------------------------+----------------------------------------+-------------------------------+-------------------------------+-----------------+-----------------+--------------------+--------------------+-----------------+-----------------+
|
||||
//
|
||||
//
|
||||
// 🌟 This was machine generated. Do not edit. 🌟
|
||||
{
|
||||
"status": 200,
|
||||
"frames": [
|
||||
{
|
||||
"schema": {
|
||||
"meta": {
|
||||
"typeVersion": [
|
||||
0,
|
||||
0
|
||||
],
|
||||
"executedQueryString": "SELECT * FROM tbl"
|
||||
},
|
||||
"fields": [
|
||||
{
|
||||
"name": "ts",
|
||||
"type": "time",
|
||||
"typeInfo": {
|
||||
"frame": "time.Time",
|
||||
"nullable": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "tsnn",
|
||||
"type": "time",
|
||||
"typeInfo": {
|
||||
"frame": "time.Time",
|
||||
"nullable": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "tsz",
|
||||
"type": "time",
|
||||
"typeInfo": {
|
||||
"frame": "time.Time",
|
||||
"nullable": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "tsznn",
|
||||
"type": "time",
|
||||
"typeInfo": {
|
||||
"frame": "time.Time",
|
||||
"nullable": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "d",
|
||||
"type": "time",
|
||||
"typeInfo": {
|
||||
"frame": "time.Time",
|
||||
"nullable": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "dnn",
|
||||
"type": "time",
|
||||
"typeInfo": {
|
||||
"frame": "time.Time",
|
||||
"nullable": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "t",
|
||||
"type": "string",
|
||||
"typeInfo": {
|
||||
"frame": "string",
|
||||
"nullable": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "tnn",
|
||||
"type": "string",
|
||||
"typeInfo": {
|
||||
"frame": "string",
|
||||
"nullable": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "tz",
|
||||
"type": "string",
|
||||
"typeInfo": {
|
||||
"frame": "string",
|
||||
"nullable": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "tznn",
|
||||
"type": "string",
|
||||
"typeInfo": {
|
||||
"frame": "string",
|
||||
"nullable": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "i",
|
||||
"type": "string",
|
||||
"typeInfo": {
|
||||
"frame": "string",
|
||||
"nullable": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "inn",
|
||||
"type": "string",
|
||||
"typeInfo": {
|
||||
"frame": "string",
|
||||
"nullable": true
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
"data": {
|
||||
"values": [
|
||||
[
|
||||
1700024767123,
|
||||
null
|
||||
],
|
||||
[
|
||||
1700024768123,
|
||||
1700024769123
|
||||
],
|
||||
[
|
||||
1626952953654,
|
||||
null
|
||||
],
|
||||
[
|
||||
1626952954654,
|
||||
1626952955654
|
||||
],
|
||||
[
|
||||
1703030400000,
|
||||
null
|
||||
],
|
||||
[
|
||||
1703116800000,
|
||||
1703203200000
|
||||
],
|
||||
[
|
||||
"12:34:56.234567",
|
||||
null
|
||||
],
|
||||
[
|
||||
"12:34:57.234567",
|
||||
"12:34:58.234567"
|
||||
],
|
||||
[
|
||||
"23:12:36.765432+01",
|
||||
null
|
||||
],
|
||||
[
|
||||
"23:12:37.765432+01",
|
||||
"23:12:38.765432+01"
|
||||
],
|
||||
[
|
||||
"00:00:00.987654",
|
||||
null
|
||||
],
|
||||
[
|
||||
"00:00:00.887654",
|
||||
"00:00:00.787654"
|
||||
]
|
||||
],
|
||||
"nanos": [
|
||||
[
|
||||
456000,
|
||||
0
|
||||
],
|
||||
[
|
||||
456000,
|
||||
456000
|
||||
],
|
||||
[
|
||||
321000,
|
||||
0
|
||||
],
|
||||
[
|
||||
321000,
|
||||
321000
|
||||
],
|
||||
null,
|
||||
null,
|
||||
null,
|
||||
null,
|
||||
null,
|
||||
null,
|
||||
null,
|
||||
null
|
||||
]
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
-- SELECT * FROM tbl
|
||||
-- test all date/time-based postgres data types
|
||||
CREATE TEMPORARY TABLE tbl (
|
||||
ts timestamp,
|
||||
tsnn timestamp NOT NULL,
|
||||
tsz timestamp with time zone,
|
||||
tsznn timestamp with time zone NOT NULL,
|
||||
d date,
|
||||
dnn date NOT NULL,
|
||||
t time,
|
||||
tnn time NOT NULL,
|
||||
tz time with time zone,
|
||||
tznn time with time zone NOT NULL,
|
||||
i interval,
|
||||
inn interval NOT NULL
|
||||
);
|
||||
|
||||
INSERT INTO tbl (ts, tsnn, tsz, tsznn, d, dnn, t, tnn, tz, tznn, i, inn) VALUES (
|
||||
'2023-11-15 05:06:07.123456',
|
||||
'2023-11-15 05:06:08.123456',
|
||||
'2021-07-22 13:22:33.654321 Europe/Berlin',
|
||||
'2021-07-22 13:22:34.654321 Europe/Berlin',
|
||||
'2023-12-20',
|
||||
'2023-12-21',
|
||||
'12:34:56.234567',
|
||||
'12:34:57.234567',
|
||||
'23:12:36.765432+1',
|
||||
'23:12:37.765432+1',
|
||||
'987654 microsecond',
|
||||
'887654 microsecond'
|
||||
), (
|
||||
NULL,
|
||||
'2023-11-15 05:06:09.123456',
|
||||
NULL,
|
||||
'2021-07-22 13:22:35.654321 Europe/Berlin',
|
||||
NULL,
|
||||
'2023-12-22',
|
||||
NULL,
|
||||
'12:34:58.234567',
|
||||
NULL,
|
||||
'23:12:38.765432+1',
|
||||
NULL,
|
||||
'787654 microsecond'
|
||||
);
|
||||
+38
-12
@@ -10,7 +10,14 @@ import {
|
||||
} from '@grafana/data';
|
||||
import { ConfigSection, ConfigSubSection, DataSourceDescription, EditorStack } from '@grafana/plugin-ui';
|
||||
import { config } from '@grafana/runtime';
|
||||
import { ConnectionLimits, Divider, TLSSecretsConfig, useMigrateDatabaseFields } from '@grafana/sql';
|
||||
import {
|
||||
ConnectionLimits,
|
||||
Divider,
|
||||
MaxLifetimeField,
|
||||
MaxOpenConnectionsField,
|
||||
TLSSecretsConfig,
|
||||
useMigrateDatabaseFields,
|
||||
} from '@grafana/sql';
|
||||
import {
|
||||
Input,
|
||||
Select,
|
||||
@@ -76,6 +83,14 @@ export const PostgresConfigEditor = (props: DataSourcePluginOptionsEditorProps<P
|
||||
};
|
||||
};
|
||||
|
||||
const onMaxConnectionsChanged = (number?: number) => {
|
||||
updateDatasourcePluginJsonDataOption(props, 'maxOpenConns', number);
|
||||
};
|
||||
|
||||
const onMaxLifetimeChanged = (number?: number) => {
|
||||
updateDatasourcePluginJsonDataOption(props, 'connMaxLifetime', number);
|
||||
};
|
||||
|
||||
const onTimeScaleDBChanged = (event: SyntheticEvent<HTMLInputElement>) => {
|
||||
updateDatasourcePluginJsonDataOption(props, 'timescaledb', event.currentTarget.checked);
|
||||
};
|
||||
@@ -153,11 +168,7 @@ export const PostgresConfigEditor = (props: DataSourcePluginOptionsEditorProps<P
|
||||
onBlur={onUpdateDatasourceSecureJsonDataOption(props, 'password')}
|
||||
/>
|
||||
</Field>
|
||||
</ConfigSection>
|
||||
|
||||
<Divider />
|
||||
|
||||
<ConfigSection title="TLS/SSL Auth Details" isCollapsible>
|
||||
<Field
|
||||
label={
|
||||
<Label>
|
||||
@@ -184,6 +195,7 @@ export const PostgresConfigEditor = (props: DataSourcePluginOptionsEditorProps<P
|
||||
width={WIDTH_LONG}
|
||||
/>
|
||||
</Field>
|
||||
|
||||
{options.jsonData.sslmode !== PostgresTLSModes.disable ? (
|
||||
<Field
|
||||
label={
|
||||
@@ -220,8 +232,12 @@ export const PostgresConfigEditor = (props: DataSourcePluginOptionsEditorProps<P
|
||||
/>
|
||||
</Field>
|
||||
) : null}
|
||||
{jsonData.sslmode !== PostgresTLSModes.disable ? (
|
||||
<>
|
||||
</ConfigSection>
|
||||
|
||||
{jsonData.sslmode !== PostgresTLSModes.disable ? (
|
||||
<>
|
||||
<Divider />
|
||||
<ConfigSection title="TLS/SSL Auth Details">
|
||||
{jsonData.tlsConfigurationMethod === PostgresTLSMethods.fileContent ? (
|
||||
<TLSSecretsConfig
|
||||
showCACert={
|
||||
@@ -313,9 +329,9 @@ export const PostgresConfigEditor = (props: DataSourcePluginOptionsEditorProps<P
|
||||
</Field>
|
||||
</>
|
||||
)}
|
||||
</>
|
||||
) : null}
|
||||
</ConfigSection>
|
||||
</ConfigSection>
|
||||
</>
|
||||
) : null}
|
||||
|
||||
<Divider />
|
||||
|
||||
@@ -396,8 +412,18 @@ export const PostgresConfigEditor = (props: DataSourcePluginOptionsEditorProps<P
|
||||
</Field>
|
||||
</ConfigSubSection>
|
||||
|
||||
<ConnectionLimits options={options} onOptionsChange={onOptionsChange} />
|
||||
|
||||
{config.featureToggles.postgresDSUsePGX ? (
|
||||
<ConfigSubSection title="Connection limits">
|
||||
<MaxOpenConnectionsField
|
||||
labelWidth={WIDTH_LONG}
|
||||
jsonData={jsonData}
|
||||
onMaxConnectionsChanged={onMaxConnectionsChanged}
|
||||
/>
|
||||
<MaxLifetimeField labelWidth={WIDTH_LONG} jsonData={jsonData} onMaxLifetimeChanged={onMaxLifetimeChanged} />
|
||||
</ConfigSubSection>
|
||||
) : (
|
||||
<ConnectionLimits options={options} onOptionsChange={onOptionsChange} />
|
||||
)}
|
||||
{config.secureSocksDSProxyEnabled && (
|
||||
<SecureSocksProxySettings options={options} onOptionsChange={onOptionsChange} />
|
||||
)}
|
||||
|
||||
Reference in New Issue
Block a user