feat(postgres sink): Add postgres sink #21248

jorgehermo9 · 2024-09-09T22:33:26Z

This PR is not 100% ready by my side and there will likely be a few things wrong, but had a few questions and wanted to know if the direction seems right... So I would like an initial round of review if possible.

I tested the sink and it seems to be working, but I lack a lot of knowledge about Vector's internals and I'm not sure if the implementation is okay.

I inspired a lot from the databend and clickhouse sinks, but left a few questions as TODOs in the source. I found this sink a bit different from the others, as the others had the request_builder thing and encoding the payload in bytes (as most of the sinks are http based).. But I didn't think that fitted well in this case, as in the sqlx API I should wrap the events with the sqlx::types::Json type and that will do all the encoding with serde internally.

If someone want to manually test it, I used this Vector config:

[sources.demo_logs]
type = "demo_logs"
format = "apache_common"

[transforms.payload]
type = "remap"
inputs = ["demo_logs"]
source = """
.payload = .
"""

[sinks.postgres]
type = "postgres"
inputs = ["payload"]
endpoint = "postgres://postgres:postgres@localhost/test"
table = "test"

Run postgres server with podman run -e POSTGRES_PASSWORD=postgres -p 5432:5432 docker.io/postgres

and execute the following with psql -h localhost -U postgres:

CREATE DATABASE test;

then execute \c test
and last:

CREATE TABLE test (message TEXT, timestamp TIMESTAMP WITH TIME ZONE, payload JSONB);

And then, you will see logs in that table:

scripts/integration/postgres/test.yaml

jorgehermo9 · 2024-09-09T22:34:34Z

src/sinks/postgres/config.rs

+}
+
+/// Configuration for the `postgres` sink.
+#[configurable_component(sink("postgres", "Deliver log data to a PostgreSQL database."))]


should I call this sink postgres or postgres_logs?

Hm good question, could this evolve to handle both logs and metrics in the future?

I'm thinking if this can evolve to integrate with other postgres flavours such as timescaledb, which is oriented to time series

My thoughts on this: #21308 (comment)

Timescaledb tracking issue: #939

I think it would be interesting to change the input

vector/src/sinks/postgres/config.rs

Line 110 in 5b1ca45

Input::log()

of this sink to allow for metrics and traces too. I'll give it a try.

jorgehermo9 · 2024-09-09T23:04:16Z

src/sinks/postgres/service.rs

+            // TODO: If a single item of the batch fails, the whole batch will fail its insert.
+            // Is this intended behaviour?
+            sqlx::query(&format!(
+                "INSERT INTO {table} SELECT * FROM jsonb_populate_recordset(NULL::{table}, $1)"


the table configuration can be a victim of sql injection, but in my opinion, we shouldn't avoid that kind of attacks at this level and the user should be responsible of ensuring that there is not sql injection in the config... The databend sink works like this

Hm, I suppose sqlx does not support parameterized table names? Does the query builder help here? If none of the above works, then we can leave as is.

I don't think that could help in this case. See this statement about sqlx's query builder.

And we cannot use a variable bind ($ syntax) in postgres for table names, as the prepared statements are bounded to a query plan and it cannot change if the target table changes.

I think this is the better way to do it... sqlx does not check for sql injection

jorgehermo9 · 2024-09-09T23:05:19Z

src/sinks/postgres/config.rs

+    pub endpoint: String,
+
+    /// The table that data is inserted into.
+    pub table: String,


Should we make the table templatable? Like the clickhouse sink. That would complicate the code a little bit (with KeyPartitioner and so. If yes, I would like some guidance about it if possible

It is a nice feature but not a must-have, we can do this incrementally. Once we finalized the rest of the comments we can come back to this if you are motivated to add this feature.

jorgehermo9 · 2024-09-09T23:06:24Z

src/sinks/postgres/integration_tests.rs

+}
+
+#[tokio::test]
+async fn test_postgres_sink() {


I think just a single test is too little.. But I couldn't figure out anything else to test. This test is very similar to the integration tests from databend sink

There are some more interesting things we can do here, at the very least send more than one events. Also, we could test failures such as sending a badly formatted payload.

Okay, going to include more tests!

aliciascott

good for docs

jorgehermo9 · 2024-09-24T23:04:30Z

src/sinks/postgres/integration_tests.rs

+use std::future::ready;
+use vector_lib::event::{BatchNotifier, BatchStatus, BatchStatusReceiver, Event, LogEvent};
+
+fn pg_host() -> String {


Copied those utility functions from the postgres_metrics source integration tests

Let's do some small refactoring here.

The folder src/test_util seems like a good location to put these utils.

You can use #[cfg(any(feature = "postgresql_metrics-integration-tests", feature = "postgres_sink-integration-tests"))] in src/test_util/mod.rs.

pront

Hi @jorgehermo9, thank you for this sizable contribution! On a high level, it looks great. I did a first review and left some comments. Don't hesitate to follow up, happy to discuss details.

scripts/integration/postgres/test.yaml

pront · 2024-10-15T16:35:17Z

src/sinks/postgres/config.rs

+}
+
+/// Configuration for the `postgres` sink.
+#[configurable_component(sink("postgres", "Deliver log data to a PostgreSQL database."))]


Hm good question, could this evolve to handle both logs and metrics in the future?

pront · 2024-10-15T20:15:45Z

src/sinks/postgres/config.rs

+    pub endpoint: String,
+
+    /// The table that data is inserted into.
+    pub table: String,


It is a nice feature but not a must-have, we can do this incrementally. Once we finalized the rest of the comments we can come back to this if you are motivated to add this feature.

pront · 2024-10-15T20:18:41Z

src/sinks/postgres/service.rs

+            // TODO: If a single item of the batch fails, the whole batch will fail its insert.
+            // Is this intended behaviour?
+            sqlx::query(&format!(
+                "INSERT INTO {table} SELECT * FROM jsonb_populate_recordset(NULL::{table}, $1)"


Hm, I suppose sqlx does not support parameterized table names? Does the query builder help here? If none of the above works, then we can leave as is.

pront · 2024-10-15T20:20:53Z

src/sinks/postgres/config.rs

+    /// The table that data is inserted into.
+    pub table: String,
+
+    /// The postgres connection pool size.


Here it would be useful to explain what this pool is used for. Maybe a link to relevant docs would suffice.

done in 21f3ad3. Do you think it is enough?

I also have doubts about using a connection pool. Can the event batches be executed in parallel for this sink? I don't know the specifics of vector's internals...

vector/src/sinks/postgres/sink.rs

Line 24 in 21f3ad3

.batched(self.batch_settings.as_byte_size_config())

If the batches of events can be processed in parallel, then a connection pool is beneficial. If the batches are processed sequentially, then we should use a single postgres connection as a pooled connection does not have sense

pront · 2024-10-15T20:25:09Z

src/sinks/postgres/integration_tests.rs

+use std::future::ready;
+use vector_lib::event::{BatchNotifier, BatchStatus, BatchStatusReceiver, Event, LogEvent};
+
+fn pg_host() -> String {


Let's do some small refactoring here.

The folder src/test_util seems like a good location to put these utils.

You can use #[cfg(any(feature = "postgresql_metrics-integration-tests", feature = "postgres_sink-integration-tests"))] in src/test_util/mod.rs.

pront · 2024-10-15T20:26:35Z

src/sinks/postgres/integration_tests.rs

+}
+
+#[tokio::test]
+async fn test_postgres_sink() {


There are some more interesting things we can do here, at the very least send more than one events. Also, we could test failures such as sending a badly formatted payload.

jorgehermo9 · 2024-10-15T20:52:25Z

Thank you very much for the review @pront! I'm kinda busy these days but I will revisit this as soon as I can :)

pront · 2024-11-25T17:16:24Z

There are a few failing checks. Also, let's add a new postgres semantic scope in https://github.com/vectordotdev/vector/blob/master/.github/workflows/semantic.yml. I will review once these are addressed. Thank you!

jorgehermo9 · 2024-11-25T18:00:22Z

I will work on this PR these days, I'll ping you whenever it is ready for another round. Thank you so much @pront!

jorgehermo9 added 5 commits September 9, 2024 01:37

feat(sinks): Initial postgres log sink implementation

f47611a

feat(sinks): Initial postgres log sink implementation

0a0576b

test(sinks): Add integration test for postgres sink

bf2b7af

feat(sinks): postgres sink config unit tests

2ecbd56

Merge branch 'master' into feature/postgres-sink

a3be3ea

jorgehermo9 requested a review from a team as a code owner September 9, 2024 22:33

github-actions bot added domain: sinks Anything related to the Vector's sinks domain: ci Anything related to Vector's CI environment labels Sep 9, 2024

jorgehermo9 commented Sep 9, 2024

View reviewed changes

scripts/integration/postgres/test.yaml Outdated Show resolved Hide resolved

jorgehermo9 commented Sep 9, 2024

View reviewed changes

docs(website): update components cue

cee4b39

jorgehermo9 requested review from a team as code owners September 9, 2024 22:38

github-actions bot added the domain: external docs Anything related to Vector's external, public documentation label Sep 9, 2024

jorgehermo9 added 6 commits September 10, 2024 00:41

chore: update spelling checklist

548e996

style: cargo fmt

ea82ad4

fix: clippy lint

ae68fc4

fix: clippy lint

89e63e3

docs: add changelog

f68f04b

docs: add changelog

f4cd4c7

jorgehermo9 commented Sep 9, 2024

View reviewed changes

aliciascott approved these changes Sep 10, 2024

View reviewed changes

jorgehermo9 mentioned this pull request Sep 18, 2024

feat: sink from Vector to Risingwave #21308

Open

jorgehermo9 commented Sep 24, 2024

View reviewed changes

pront self-assigned this Oct 15, 2024

pront reviewed Oct 15, 2024

View reviewed changes

Merge with master

9d78d67

jorgehermo9 added 3 commits November 22, 2024 17:45

chore: rename postgres integration tests feature flag

984750d

chore: rename postgres integration tests feature flag

5b1ca45

docs: include connection pool reference

21f3ad3

pront changed the title ~~feat(sink): Add postgres sink~~ feat(postgres sink): Add postgres sink Nov 25, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(postgres sink): Add postgres sink #21248

feat(postgres sink): Add postgres sink #21248

jorgehermo9 commented Sep 9, 2024 •

edited

Loading

jorgehermo9 Sep 9, 2024

pront Oct 15, 2024

jorgehermo9 Oct 15, 2024 •

edited

Loading

jorgehermo9 Nov 24, 2024

jorgehermo9 Sep 9, 2024

pront Oct 15, 2024

jorgehermo9 Nov 24, 2024

jorgehermo9 Sep 9, 2024

pront Oct 15, 2024

jorgehermo9 Nov 24, 2024

jorgehermo9 Sep 9, 2024 •

edited

Loading

pront Oct 15, 2024

jorgehermo9 Nov 24, 2024

aliciascott left a comment

jorgehermo9 Sep 24, 2024 •

edited

Loading

pront Oct 15, 2024

pront left a comment

pront Oct 15, 2024

pront Oct 15, 2024

pront Oct 15, 2024

pront Oct 15, 2024

jorgehermo9 Nov 24, 2024

pront Oct 15, 2024

pront Oct 15, 2024

jorgehermo9 commented Oct 15, 2024

pront commented Nov 25, 2024

jorgehermo9 commented Nov 25, 2024 •

edited

Loading

feat(postgres sink): Add postgres sink #21248

Are you sure you want to change the base?

feat(postgres sink): Add postgres sink #21248

Conversation

jorgehermo9 commented Sep 9, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jorgehermo9 Oct 15, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jorgehermo9 Sep 9, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aliciascott left a comment

Choose a reason for hiding this comment

jorgehermo9 Sep 24, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pront left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jorgehermo9 commented Oct 15, 2024

pront commented Nov 25, 2024

jorgehermo9 commented Nov 25, 2024 • edited Loading

jorgehermo9 commented Sep 9, 2024 •

edited

Loading

jorgehermo9 Oct 15, 2024 •

edited

Loading

jorgehermo9 Sep 9, 2024 •

edited

Loading

jorgehermo9 Sep 24, 2024 •

edited

Loading

jorgehermo9 commented Nov 25, 2024 •

edited

Loading