CL-Softwares
/
telegraf
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(outputs.sql): Support secret for dsn (#16678)

This commit is contained in:
Lars Stegman 2025-04-14 18:14:00 +02:00 committed by GitHub
parent be9e5bfbb3
commit 799e194700
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 54 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
plugins/outputs/sql/README.md
View File

@ -76,6 +76,14 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
[CONFIGURATION.md]: ../../../docs/CONFIGURATION.md#plugins
## Secret-store support
This plugin supports secrets from secret-stores for the `data_source_name`
option. See the [secret-store documentation][SECRETSTORE] for more details on
how to use them.
[SECRETSTORE]: ../../../docs/CONFIGURATION.md#secret-store-secrets
## Configuration
```toml @sample.conf

25
plugins/outputs/sql/sql.go
View File

@ -49,7 +49,7 @@ type ConvertStruct struct {
type SQL struct {
Driver string `toml:"driver"`
DataSourceName string `toml:"data_source_name"`
DataSourceName config.Secret `toml:"data_source_name"`
TimestampColumn string `toml:"timestamp_column"`
TableTemplate string `toml:"table_template"`
TableExistsTemplate string `toml:"table_exists_template"`
@ -105,7 +105,14 @@ func (p *SQL) Init() error {
}
func (p *SQL) Connect() error {
db, err := gosql.Open(p.Driver, p.DataSourceName)
dsnBuffer, err := p.DataSourceName.Get()
if err != nil {
return fmt.Errorf("loading data source name secret failed: %w", err)
}
dsn := dsnBuffer.String()
dsnBuffer.Destroy()
db, err := gosql.Open(p.Driver, dsn)
if err != nil {
return fmt.Errorf("creating database client failed: %w", err)
}
@ -401,7 +408,15 @@ func (p *SQL) Write(metrics []telegraf.Metric) error {
// Convert a DSN possibly using v1 parameters to clickhouse-go v2 format
func (p *SQL) convertClickHouseDsn() {
u, err := url.Parse(p.DataSourceName)
dsnBuffer, err := p.DataSourceName.Get()
if err != nil {
p.Log.Errorf("loading data source name failed: %v", err)
return
}
dsn := dsnBuffer.String()
dsnBuffer.Destroy()
u, err := url.Parse(dsn)
if err != nil {
return
}
@ -443,7 +458,9 @@ func (p *SQL) convertClickHouseDsn() {
}
u.RawQuery = query.Encode()
p.DataSourceName = u.String()
if err := p.DataSourceName.Set([]byte(u.String())); err != nil {
p.Log.Errorf("updating data source name to click house dsn failed: %v", err)
}
}
func init() {

37
plugins/outputs/sql/sql_test.go
View File

@ -14,6 +14,7 @@ import (
"github.com/testcontainers/testcontainers-go/wait"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/metric"
"github.com/influxdata/telegraf/testutil"
)
@ -181,9 +182,9 @@ func TestMysqlIntegration(t *testing.T) {
defer container.Terminate()
// use the plugin to write to the database
address := fmt.Sprintf("%v:%v@tcp(%v:%v)/%v",
address := config.NewSecret([]byte(fmt.Sprintf("%v:%v@tcp(%v:%v)/%v",
username, password, container.Address, container.Ports[servicePort], dbname,
)
)))
p := &SQL{
Driver: "mysql",
DataSourceName: address,
@ -266,9 +267,9 @@ func TestMysqlUpdateSchemeIntegration(t *testing.T) {
defer container.Terminate()
// use the plugin to write to the database
address := fmt.Sprintf("%v:%v@tcp(%v:%v)/%v",
address := config.NewSecret([]byte(fmt.Sprintf("%v:%v@tcp(%v:%v)/%v",
username, password, container.Address, container.Ports[servicePort], dbname,
)
)))
p := &SQL{
Driver: "mysql",
DataSourceName: address,
@ -350,9 +351,9 @@ func TestPostgresIntegration(t *testing.T) {
// use the plugin to write to the database
// host, port, username, password, dbname
address := fmt.Sprintf("postgres://%v:%v@%v:%v/%v",
address := config.NewSecret([]byte(fmt.Sprintf("postgres://%v:%v@%v:%v/%v",
username, password, container.Address, container.Ports[servicePort], dbname,
)
)))
p := &SQL{
Driver: "pgx",
DataSourceName: address,
@ -437,9 +438,9 @@ func TestPostgresUpdateSchemeIntegration(t *testing.T) {
// use the plugin to write to the database
// host, port, username, password, dbname
address := fmt.Sprintf("postgres://%v:%v@%v:%v/%v",
address := config.NewSecret([]byte(fmt.Sprintf("postgres://%v:%v@%v:%v/%v",
username, password, container.Address, container.Ports[servicePort], dbname,
)
)))
p := &SQL{
Driver: "pgx",
DataSourceName: address,
@ -537,8 +538,8 @@ func TestClickHouseIntegration(t *testing.T) {
// use the plugin to write to the database
// host, port, username, password, dbname
address := fmt.Sprintf("tcp://%s:%s/%s?username=%s&password=%s",
container.Address, container.Ports[servicePort], dbname, username, password)
address := config.NewSecret([]byte(fmt.Sprintf("tcp://%s:%s/%s?username=%s&password=%s",
container.Address, container.Ports[servicePort], dbname, username, password)))
p := &SQL{
Driver: "clickhouse",
DataSourceName: address,
@ -632,8 +633,8 @@ func TestClickHouseUpdateSchemeIntegration(t *testing.T) {
// use the plugin to write to the database
// host, port, username, password, dbname
address := fmt.Sprintf("tcp://%s:%s/%s?username=%s&password=%s",
container.Address, container.Ports[servicePort], dbname, username, password)
address := config.NewSecret([]byte(fmt.Sprintf("tcp://%s:%s/%s?username=%s&password=%s",
container.Address, container.Ports[servicePort], dbname, username, password)))
p := &SQL{
Driver: "clickhouse",
DataSourceName: address,
@ -723,11 +724,15 @@ func TestClickHouseDsnConvert(t *testing.T) {
for _, tt := range tests {
plugin := &SQL{
Driver: "clickhouse",
DataSourceName: tt.input,
DataSourceName: config.NewSecret([]byte(tt.input)),
Log: testutil.Logger{},
}
require.NoError(t, plugin.Init())
require.Equal(t, tt.expected, plugin.DataSourceName)
resolvedSecret, err := plugin.DataSourceName.Get()
require.NoError(t, err)
resolvedDsn := resolvedSecret.String()
resolvedSecret.Destroy()
require.Equal(t, tt.expected, resolvedDsn)
}
}
@ -769,9 +774,9 @@ func TestMysqlEmptyTimestampColumnIntegration(t *testing.T) {
defer container.Terminate()
// use the plugin to write to the database
address := fmt.Sprintf("%v:%v@tcp(%v:%v)/%v",
address := config.NewSecret([]byte(fmt.Sprintf("%v:%v@tcp(%v:%v)/%v",
username, password, container.Address, container.Ports[servicePort], dbname,
)
)))
p := &SQL{
Driver: "mysql",
DataSourceName: address,

7
plugins/outputs/sql/sqlite_test.go
View File

@ -11,6 +11,7 @@ import (
"github.com/stretchr/testify/require"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/testutil"
)
@ -23,7 +24,7 @@ func TestSqlite(t *testing.T) {
address := dbfile // accepts a path or a file: URI
p := &SQL{
Driver: "sqlite",
DataSourceName: address,
DataSourceName: config.NewSecret([]byte(address)),
Convert: defaultConvert,
TimestampColumn: "timestamp",
ConnectionMaxIdle: 2,
@ -140,7 +141,7 @@ func TestSqliteUpdateScheme(t *testing.T) {
// Use the plugin to write to the database address :=
// fmt.Sprintf("file:%v", dbfile)
address := dbfile // accepts a path or a file: URI
address := config.NewSecret([]byte(dbfile)) // accepts a path or a file: URI
p := &SQL{
Driver: "sqlite",
DataSourceName: address,
@ -157,7 +158,7 @@ func TestSqliteUpdateScheme(t *testing.T) {
require.NoError(t, p.Write(testMetrics))
// read directly from the database
db, err := gosql.Open("sqlite", address)
db, err := gosql.Open("sqlite", dbfile)
require.NoError(t, err)
defer db.Close()