redpanda-data / redpanda-data/connect

sql_insert doesn't support bloblang interpolation in dsn field

Open
#3,765 0 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

ux
Dominant language
Go
Stars
8.8k
Forks
969
Avg merge
1d 13h
Merged PRs (30d)
64

Description

Hi,

Is there a way to support interpolation in dsn field in sql_insert?
Our usecase: we read messages as inputs, we want to use AWS IAM role based authentication to postgres on RDS. So dsn will have password as token generated by aws.

There can be other usecases like based on metadata of message we want to route message to different DB server host.

I can see code in
https://github.com/redpanda-data/connect/blob/main/internal/impl/sql/conn_fields.go#L33
https://github.com/redpanda-data/connect/blob/main/internal/impl/sql/output_sql_insert.go#L38C9-L38C17
dsnField is not defined as NewBloblangField like arg_mapping or topics for input here.

Purposed solutions (either of below):

  1. we can change dsnField to NewBloblangField
  2. add support for IAM based authentication which will use golang aws sdk to get token and use that instead of password in sql_insert output.

output:
  sql_insert:
    driver: postgres
    dsn: 'postgres://postgres:${! meta("pg_pass") }@localhost:5432/postgres?sslmode=disable'
    table: test
    iam_based_auth:
          role: 'somerole_aws_role' 
          ....

My test config:


inputs:
  label: ""
  redpanda:
    seed_brokers: [] # No default (optional)
    topics: [] # No default (required)
    regexp_topics: false
    transaction_isolation_level: read_uncommitted
    consumer_group: "" # No default (optional)
    auto_replay_nacks: true
pipeline:
  processors:
    - mapping: |
#        we need to get this from message metadata or our dynamic plugin which will generate password for given message
#        metadata which will replace pg_pass here.
        meta pg_pass = "postgres" 
output:
  sql_insert:
    init_statement: | # No default (optional)
      CREATE TABLE IF NOT EXISTS test (
           alias VARCHAR,
           query VARCHAR not null,
           parsing_time BIGINT);
    driver: postgres
    dsn: 'postgres://postgres:${! meta("pg_pass") }@localhost:5432/postgres?sslmode=disable'
    table: test
    columns:
      [
        "alias",
        "query",
        "parsing_time"
      ]
    args_mapping: root = [
      this.alias,
      this.query,
      this.parsing_time ]
    max_in_flight: 64

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Read internal/impl/sql/conn_fields.go and internal/impl/sql/output_sql_insert.go, focusing on dsnField and the existing Bloblang fields. Reproduce the provided sql_insert configuration with a metadata-generated password, then review the relevant SQL output tests. Done means the selected approach supports the requested dynamic DSN use case and has coverage for it.

Written by the indexing model from the issue text.

Assessment

Tech stack
aws, go, postgresql
Domain
data-engineering, database
Issue type
Feature
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
42/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.