Add PostgreSQL Messaging Pub/Sub Watermill #1680

teodor-yanev · 2023-11-16T10:30:42Z

Issue described in: #1636

Testing

2023-11-16 10:20:18 {"level":"info","Timestamp":1700122818822984756,"message":"Initializing logger in level: debug"}
2023-11-16 10:20:18 {"level":"info","Timestamp":1700122818822995839,"message":"No cloud provider credentials specified, using password"}
2023-11-16 10:20:18 {"level":"info","Timestamp":1700122818827850756,"message":"Connected to DB"}
2023-11-16 10:20:18 {"level":"debug","component":"watermill","count":"1","Timestamp":1700122818827921006,"message":"Adding publisher decorators"}
2023-11-16 10:20:18 {"level":"debug","component":"watermill","count":"1","Timestamp":1700122818827937214,"message":"Adding subscriber decorators"}
2023-11-16 10:20:18 {"level":"debug","component":"watermill","count":"1","Timestamp":1700122818827971464,"message":"Adding middleware"}
2023-11-16 10:20:18 {"level":"debug","component":"watermill","count":"3","Timestamp":1700122818827978714,"message":"Adding middleware"}
2023-11-16 10:20:18 {"level":"info","component":"watermill","topic":"internal.entity.event","handler_name":"github.com/stacklok/minder/internal/engine.(*Executor).HandleEntityEvent-fm-internal.entity.event","Timestamp":1700122818837505464,"message":"Adding handler"}
2023-11-16 10:20:18 {"level":"info","component":"watermill","handler_name":"github.com/stacklok/minder/internal/reconcilers.(*Reconciler).handleRepoReconcilerEvent-fm-internal.repo.reconciler.event","topic":"internal.repo.reconciler.event","Timestamp":1700122818837799547,"message":"Adding handler"}
2023-11-16 10:20:18 {"level":"info","component":"watermill","topic":"internal.profile.init.event","handler_name":"github.com/stacklok/minder/internal/reconcilers.(*Reconciler).handleProfileInitEvent-fm-internal.profile.init.event","Timestamp":1700122818837813172,"message":"Adding handler"}
2023-11-16 10:20:18 {"level":"debug","component":"watermill","Timestamp":1700122818837889714,"message":"Loading plugins"}
2023-11-16 10:20:18 {"level":"info","component":"watermill","count":3,"Timestamp":1700122818837920672,"message":"Running router handlers"}
2023-11-16 10:20:18 {"level":"debug","component":"watermill","subscriber_name":"github.com/stacklok/minder/internal/engine.(*Executor).HandleEntityEvent-fm-internal.entity.event","topic":"internal.entity.event","Timestamp":1700122818838012047,"message":"Subscribing to topic"}
2023-11-16 10:20:18 {"Timestamp":1700122818838127922,"message":"2023/11/16 08:20:18 Starting gRPC server on 0.0.0.0:8090"}
2023-11-16 10:20:18 {"Timestamp":1700122818838998756,"message":"2023/11/16 08:20:18 Starting HTTP server on 0.0.0.0:8080"}
2023-11-16 10:20:18 {"Timestamp":1700122818839487422,"message":"2023/11/16 08:20:18 Starting metrics server on 0.0.0.0:9090"}
2023-11-16 10:20:18 [watermill] 2023/11/16 08:20:18.838062 schema.go:28: level=INFO  msg="Initializing subscriber schema" query="[ 
2023-11-16 10:20:18 CREATE TABLE IF NOT EXISTS "watermill_internal.entity.event" (
2023-11-16 10:20:18 "offset" SERIAL,
2023-11-16 10:20:18 "uuid" VARCHAR(36) NOT NULL,
2023-11-16 10:20:18 "created_at" TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
2023-11-16 10:20:18 "payload" JSON DEFAULT NULL,
2023-11-16 10:20:18 "metadata" JSON DEFAULT NULL,
2023-11-16 10:20:18 "transaction_id" xid8 NOT NULL,
2023-11-16 10:20:18 PRIMARY KEY ("transaction_id", "offset")
2023-11-16 10:20:18 );
2023-11-16 10:20:18  
2023-11-16 10:20:18 CREATE TABLE IF NOT EXISTS "watermill_offsets_internal.entity.event" (
2023-11-16 10:20:18 consumer_group VARCHAR(255) NOT NULL,
2023-11-16 10:20:18 offset_acked BIGINT,
2023-11-16 10:20:18 last_processed_transaction_id xid8 NOT NULL,
2023-11-16 10:20:18 PRIMARY KEY(consumer_group)
2023-11-16 10:20:18 )]" subscriber_id=01HFBKEY8BZBT23P4EEANZD004 
2023-11-16 10:20:18 [watermill] 2023/11/16 08:20:18.845592 schema.go:28: level=INFO  msg="Initializing subscriber schema" query="[ 
2023-11-16 10:20:18 CREATE TABLE IF NOT EXISTS "watermill_internal.repo.reconciler.event" (
2023-11-16 10:20:18 "offset" SERIAL,
2023-11-16 10:20:18 "uuid" VARCHAR(36) NOT NULL,
2023-11-16 10:20:18 {"level":"debug","component":"watermill","subscriber_name":"github.com/stacklok/minder/internal/reconcilers.(*Reconciler).handleRepoReconcilerEvent-fm-internal.repo.reconciler.event","topic":"internal.repo.reconciler.event","Timestamp":1700122818845564006,"message":"Subscribing to topic"}
2023-11-16 10:20:18 {"level":"info","component":"watermill","subscriber_name":"github.com/stacklok/minder/internal/engine.(*Executor).HandleEntityEvent-fm-internal.entity.event","topic":"internal.entity.event","Timestamp":1700122818845722756,"message":"Starting handler"}
2023-11-16 10:20:18 {"level":"debug","component":"watermill","subscriber_name":"github.com/stacklok/minder/internal/reconcilers.(*Reconciler).handleProfileInitEvent-fm-internal.profile.init.event","topic":"internal.profile.init.event","Timestamp":1700122818849883131,"message":"Subscribing to topic"}
2023-11-16 10:20:18 {"level":"info","component":"watermill","subscriber_name":"github.com/stacklok/minder/internal/reconcilers.(*Reconciler).handleRepoReconcilerEvent-fm-internal.repo.reconciler.event","topic":"internal.repo.reconciler.event","Timestamp":1700122818849898172,"message":"Starting handler"}
2023-11-16 10:20:18 {"level":"info","component":"watermill","topic":"internal.profile.init.event","subscriber_name":"github.com/stacklok/minder/internal/reconcilers.(*Reconciler).handleProfileInitEvent-fm-internal.profile.init.event","Timestamp":1700122818860212756,"message":"Starting handler"}
2023-11-16 10:20:33 {"level":"error","Resource":{"service":"minder.v1.UserService","method":"GetUser"},"Attributes":{"http.code":"NotFound","http.content-type":["application/grpc"],"http.duration":"2.34775ms","http.user_agent":["minder-cli/v0.0.15+ref.d8c9ae80 (darwin) arm64 (go1.21.3) grpc-go/1.59.0"],"exception.message":"rpc error: code = NotFound desc = user not found"},"Timestamp":1700122833197264929}
2023-11-16 10:20:33 {"level":"info","Resource":{"service":"minder.v1.UserService","method":"CreateUser"},"Attributes":{"http.code":"OK","http.content-type":["application/grpc"],"http.duration":"6.48ms","http.user_agent":["minder-cli/v0.0.15+ref.d8c9ae80 (darwin) arm64 (go1.21.3) grpc-go/1.59.0"]},"Timestamp":1700122833204679846}
2023-11-16 10:20:18 "created_at" TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
2023-11-16 10:20:18 "payload" JSON DEFAULT NULL,
2023-11-16 10:20:18 "metadata" JSON DEFAULT NULL,
2023-11-16 10:20:18 "transaction_id" xid8 NOT NULL,
2023-11-16 10:20:18 PRIMARY KEY ("transaction_id", "offset")
2023-11-16 10:20:18 );
2023-11-16 10:20:18  
2023-11-16 10:20:18 CREATE TABLE IF NOT EXISTS "watermill_offsets_internal.repo.reconciler.event" (
2023-11-16 10:20:18 consumer_group VARCHAR(255) NOT NULL,
2023-11-16 10:20:18 offset_acked BIGINT,
2023-11-16 10:20:18 last_processed_transaction_id xid8 NOT NULL,
2023-11-16 10:20:18 PRIMARY KEY(consumer_group)
2023-11-16 10:20:18 )]" subscriber_id=01HFBKEY8BZBT23P4EEANZD004 
2023-11-16 10:20:18 [watermill] 2023/11/16 08:20:18.849949 schema.go:28: level=INFO  msg="Initializing subscriber schema" query="[ 
2023-11-16 10:20:18 CREATE TABLE IF NOT EXISTS "watermill_internal.profile.init.event" (
2023-11-16 10:20:18 "offset" SERIAL,
2023-11-16 10:20:18 "uuid" VARCHAR(36) NOT NULL,
2023-11-16 10:20:18 "created_at" TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
2023-11-16 10:20:18 "payload" JSON DEFAULT NULL,
2023-11-16 10:20:18 "metadata" JSON DEFAULT NULL,
2023-11-16 10:20:18 "transaction_id" xid8 NOT NULL,
2023-11-16 10:20:18 PRIMARY KEY ("transaction_id", "offset")
2023-11-16 10:20:18 );
2023-11-16 10:20:18  
2023-11-16 10:20:18 CREATE TABLE IF NOT EXISTS "watermill_offsets_internal.profile.init.event" (
2023-11-16 10:20:18 consumer_group VARCHAR(255) NOT NULL,
2023-11-16 10:20:18 offset_acked BIGINT,
2023-11-16 10:20:18 last_processed_transaction_id xid8 NOT NULL,
2023-11-16 10:20:18 PRIMARY KEY(consumer_group)
2023-11-16 10:20:18 )]" subscriber_id=01HFBKEY8BZBT23P4EEANZD004

Update: after adding the second database "minder_queue", we have a nice separation of concerns:

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

jhrozek · 2023-11-16T12:17:58Z

Maybe a stupid question, but since there appears to be a single table, how does publishing and subscribing work with multiple minder processes? Does the subscriber implement some locking with e.g. SELECT FOR UPDATE?

JAORMX

I see that the postgres driver brings some new tables to the mix. Where are those created? The minder-server database user does not have permissions to create tables, only the migration user is able to do this. I'm guessing this would be an issue.

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

internal/config/config.go

config/config.yaml.example

internal/events/eventer.go

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

deployment/helm/templates/deployment.yaml

internal/config/db.go

jhrozek

I'll remove the block once we discuss the env variable, but on first glance it seems incorrect

deployment/helm/templates/deployment.yaml

internal/config/db.go

This removes the boilerplate on other parts of the code and instead moves the responsibility to eventer.

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

Env var will be added as an extraEnv in the mediator-app.yaml in infra

stacklokbot

Minder analyzed this PR and found no vulnerable dependencies.

teodor-yanev · 2023-11-21T18:58:23Z

Maybe a stupid question, but since there appears to be a single table, how does publishing and subscribing work with multiple minder processes? Does the subscriber implement some locking with e.g. SELECT FOR UPDATE?

Good question, to answer this one, I did some digging in files like https://github.com/ThreeDotsLabs/watermill-sql/blob/master/pkg/sql/offsets_adapter_postgresql.go and https://github.com/ThreeDotsLabs/watermill/blob/master/_examples/real-world-examples/exactly-once-delivery-counter/README.md, and the scarce documentation in https://watermill.io/pubsubs/sql/ so I can present an overall idea of how it functions:

There are essentially two types of tables involved for each topic and consumer group:

Message Table:

Contains the actual messages that are published.
Each message typically has a unique identifier, the payload, and possibly other metadata.
This table is used for both publishing (inserting messages) and subscribing (reading messages).

Offsets Table:

Used for tracking the progress of each consumer or consumer group.
It records the last message offset (or ID) that has been acknowledged or processed by a consumer group.
Ensures that each message is processed exactly once by each consumer group and helps maintain the order of message processing.
Uses mechanisms like SELECT ... FOR UPDATE to lock rows and manage concurrent access by different consumers.

Relationship Between the Tables

The message table and the offsets table are related in the sense that the offsets table keeps track of which messages from the message table have been processed. The offsets table stores an identifier (like an offset or a transaction ID) that corresponds to records in the message table.

jhrozek · 2023-11-21T20:27:42Z

One more question, but I guess this can be a follow-up..how is the InitSchema attribute used? I can't seem to find the usage:

jakub@flatline-2 ~/devel/mediator (postresql-messaging-watermill●)$ git grep init_schema
internal/config/events.go:      InitSchema bool           `mapstructure:"init_schema" default:"true"`
jakub@flatline-2 ~/devel/mediator (postresql-messaging-watermill●)$ git grep InitSchema
internal/config/events.go:      // InitSchema is whether or not to initialize the schema
internal/config/events.go:      InitSchema bool           `mapstructure:"init_schema" default:"true"`

JAORMX · 2023-11-22T05:19:26Z

It isn't as of now. I made it configurable because I wanna figure out if it's possible to move this logic out of the workload, so then we could just toggle that.

teodor-yanev · 2023-11-22T09:17:21Z

It isn't as of now. I made it configurable because I wanna figure out if it's possible to move this logic out of the workload, so then we could just toggle that.

Should we remove it for now and re-add it when it's present in the application logic?

teodor-yanev requested review from JAORMX, jhrozek and evankanderson November 16, 2023 10:30

teodor-yanev self-assigned this Nov 16, 2023

stacklokbot reviewed Nov 16, 2023

View reviewed changes

JAORMX reviewed Nov 16, 2023

View reviewed changes

teodor-yanev added 2 commits November 17, 2023 13:44

add-postgresql-messaging-pubsub-watermill

2fa4c1a

fix helm tests

ec775c0

teodor-yanev force-pushed the postresql-messaging-watermill branch from 3a4d2be to ec775c0 Compare November 17, 2023 11:44

stacklokbot reviewed Nov 17, 2023

View reviewed changes

add-second-db-for-eventing

627ec78

stacklokbot reviewed Nov 17, 2023

View reviewed changes

fix-helm-tests-2

66ceb69

stacklokbot reviewed Nov 17, 2023

View reviewed changes

check-unchecked-err-db

341d77e

stacklokbot reviewed Nov 17, 2023

View reviewed changes

JAORMX requested changes Nov 20, 2023

View reviewed changes

internal/config/config.go Outdated Show resolved Hide resolved

config/config.yaml.example Outdated Show resolved Hide resolved

config/config.yaml.example Outdated Show resolved Hide resolved

internal/events/eventer.go Outdated Show resolved Hide resolved

teodor-yanev added 2 commits November 21, 2023 13:49

Merge branch 'main' into postresql-messaging-watermill

ee970d4

refactoring

e8df075

stacklokbot reviewed Nov 21, 2023

View reviewed changes

revert testing docker changes

345af29

stacklokbot reviewed Nov 21, 2023

View reviewed changes

update env var name

063435c

stacklokbot reviewed Nov 21, 2023

View reviewed changes

JAORMX requested changes Nov 21, 2023

View reviewed changes

deployment/helm/templates/deployment.yaml Outdated Show resolved Hide resolved

internal/config/db.go Outdated Show resolved Hide resolved

jhrozek requested changes Nov 21, 2023

View reviewed changes

deployment/helm/templates/deployment.yaml Outdated Show resolved Hide resolved

internal/config/db.go Outdated Show resolved Hide resolved

Do database queue connection initialization on eventer (#1708)

9241ff4

This removes the boilerplate on other parts of the code and instead moves the responsibility to eventer.

stacklokbot reviewed Nov 21, 2023

View reviewed changes

remove redundant bindings and env var

96bce22

Env var will be added as an extraEnv in the mediator-app.yaml in infra

stacklokbot reviewed Nov 21, 2023

View reviewed changes

JAORMX approved these changes Nov 21, 2023

View reviewed changes

teodor-yanev requested a review from jhrozek November 21, 2023 19:31

JAORMX approved these changes Nov 22, 2023

View reviewed changes

JAORMX requested a review from stacklokbot November 22, 2023 09:23

jhrozek approved these changes Nov 22, 2023

View reviewed changes

JAORMX merged commit c063e24 into main Nov 22, 2023
12 checks passed

JAORMX deleted the postresql-messaging-watermill branch November 22, 2023 09:25

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add PostgreSQL Messaging Pub/Sub Watermill #1680

Add PostgreSQL Messaging Pub/Sub Watermill #1680

teodor-yanev commented Nov 16, 2023 •

edited

Loading

stacklokbot left a comment

stacklokbot left a comment

stacklokbot left a comment

stacklokbot left a comment

jhrozek commented Nov 16, 2023

JAORMX left a comment

stacklokbot left a comment

stacklokbot left a comment

stacklokbot left a comment

stacklokbot left a comment

stacklokbot left a comment

stacklokbot left a comment

stacklokbot left a comment

stacklokbot left a comment

stacklokbot left a comment

stacklokbot left a comment

stacklokbot left a comment

jhrozek left a comment

stacklokbot left a comment

stacklokbot left a comment

teodor-yanev commented Nov 21, 2023 •

edited

Loading

jhrozek commented Nov 21, 2023

JAORMX commented Nov 22, 2023

teodor-yanev commented Nov 22, 2023

Add PostgreSQL Messaging Pub/Sub Watermill #1680

Add PostgreSQL Messaging Pub/Sub Watermill #1680

Conversation

teodor-yanev commented Nov 16, 2023 • edited Loading

Testing

stacklokbot left a comment

Choose a reason for hiding this comment

stacklokbot left a comment

Choose a reason for hiding this comment

stacklokbot left a comment

Choose a reason for hiding this comment

stacklokbot left a comment

Choose a reason for hiding this comment

jhrozek commented Nov 16, 2023

JAORMX left a comment

Choose a reason for hiding this comment

stacklokbot left a comment

Choose a reason for hiding this comment

stacklokbot left a comment

Choose a reason for hiding this comment

stacklokbot left a comment

Choose a reason for hiding this comment

stacklokbot left a comment

Choose a reason for hiding this comment

stacklokbot left a comment

Choose a reason for hiding this comment

stacklokbot left a comment

Choose a reason for hiding this comment

stacklokbot left a comment

Choose a reason for hiding this comment

stacklokbot left a comment

Choose a reason for hiding this comment

stacklokbot left a comment

Choose a reason for hiding this comment

stacklokbot left a comment

Choose a reason for hiding this comment

stacklokbot left a comment

Choose a reason for hiding this comment

jhrozek left a comment

Choose a reason for hiding this comment

stacklokbot left a comment

Choose a reason for hiding this comment

stacklokbot left a comment

Choose a reason for hiding this comment

teodor-yanev commented Nov 21, 2023 • edited Loading

Message Table:

Offsets Table:

Relationship Between the Tables

jhrozek commented Nov 21, 2023

JAORMX commented Nov 22, 2023

teodor-yanev commented Nov 22, 2023

teodor-yanev commented Nov 16, 2023 •

edited

Loading

teodor-yanev commented Nov 21, 2023 •

edited

Loading