mirror of
https://github.com/zitadel/zitadel.git
synced 2024-12-04 23:45:07 +00:00
fix(eventstore): set application name during push to instance id (#8914)
# Which Problems Are Solved Noisy neighbours can introduce projection latencies because the projections only query events older than the start timestamp of the oldest push transaction. # How the Problems Are Solved During push we set the application name to `zitadel_es_pusher_<instance_id>` instead of `zitadel_es_pusher` which is used to query events by projections.
This commit is contained in:
Silvan
GitHub
parent
c2f88f34ae
commit
22d9e9f812
@ -15,7 +15,6 @@ import (
|
||||
|
||||
"github.com/zitadel/zitadel/internal/api/authz"
|
||||
"github.com/zitadel/zitadel/internal/database"
|
||||
"github.com/zitadel/zitadel/internal/database/dialect"
|
||||
"github.com/zitadel/zitadel/internal/eventstore"
|
||||
"github.com/zitadel/zitadel/internal/eventstore/repository"
|
||||
"github.com/zitadel/zitadel/internal/telemetry/tracing"
|
||||
@ -125,11 +124,11 @@ type CRDB struct {
|
||||
func NewCRDB(client *database.DB) *CRDB {
|
||||
switch client.Type() {
|
||||
case "cockroach":
|
||||
awaitOpenTransactionsV1 = " AND creation_date::TIMESTAMP < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = '" + dialect.EventstorePusherAppName + "')"
|
||||
awaitOpenTransactionsV2 = ` AND hlc_to_timestamp("position") < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = '` + dialect.EventstorePusherAppName + `')`
|
||||
awaitOpenTransactionsV1 = " AND creation_date::TIMESTAMP < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = ANY(?))"
|
||||
awaitOpenTransactionsV2 = ` AND hlc_to_timestamp("position") < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = ANY(?))`
|
||||
case "postgres":
|
||||
awaitOpenTransactionsV1 = ` AND EXTRACT(EPOCH FROM created_at) < (SELECT COALESCE(EXTRACT(EPOCH FROM min(xact_start)), EXTRACT(EPOCH FROM now())) FROM pg_stat_activity WHERE datname = current_database() AND application_name = '` + dialect.EventstorePusherAppName + `' AND state <> 'idle')`
|
||||
awaitOpenTransactionsV2 = ` AND "position" < (SELECT COALESCE(EXTRACT(EPOCH FROM min(xact_start)), EXTRACT(EPOCH FROM now())) FROM pg_stat_activity WHERE datname = current_database() AND application_name = '` + dialect.EventstorePusherAppName + `' AND state <> 'idle')`
|
||||
awaitOpenTransactionsV1 = ` AND EXTRACT(EPOCH FROM created_at) < (SELECT COALESCE(EXTRACT(EPOCH FROM min(xact_start)), EXTRACT(EPOCH FROM now())) FROM pg_stat_activity WHERE datname = current_database() AND application_name = ANY(?) AND state <> 'idle')`
|
||||
awaitOpenTransactionsV2 = ` AND "position" < (SELECT COALESCE(EXTRACT(EPOCH FROM min(xact_start)), EXTRACT(EPOCH FROM now())) FROM pg_stat_activity WHERE datname = current_database() AND application_name = ANY(?) AND state <> 'idle')`
|
||||
}
|
||||
|
||||
return &CRDB{client}
|
||||
|
||||
@ -272,7 +272,19 @@ func prepareConditions(criteria querier, query *repository.SearchQuery, useV1 bo
|
||||
}
|
||||
|
||||
if query.AwaitOpenTransactions {
|
||||
instanceIDs := make(database.TextArray[string], 0, 3)
|
||||
if query.InstanceID != nil {
|
||||
instanceIDs = append(instanceIDs, query.InstanceID.Value.(string))
|
||||
} else if query.InstanceIDs != nil {
|
||||
instanceIDs = append(instanceIDs, query.InstanceIDs.Value.(database.TextArray[string])...)
|
||||
}
|
||||
|
||||
for i := range instanceIDs {
|
||||
instanceIDs[i] = dialect.DBPurposeEventPusher.AppName() + "_" + instanceIDs[i]
|
||||
}
|
||||
|
||||
clauses += awaitOpenTransactions(useV1)
|
||||
args = append(args, instanceIDs)
|
||||
}
|
||||
|
||||
if clauses == "" {
|
||||
|
||||
@ -402,8 +402,8 @@ func Test_prepareCondition(t *testing.T) {
|
||||
useV1: true,
|
||||
},
|
||||
res: res{
|
||||
clause: " WHERE aggregate_type = ANY(?) AND creation_date::TIMESTAMP < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = 'zitadel_es_pusher')",
|
||||
values: []interface{}{[]eventstore.AggregateType{"user", "org"}},
|
||||
clause: " WHERE aggregate_type = ANY(?) AND creation_date::TIMESTAMP < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = ANY(?))",
|
||||
values: []interface{}{[]eventstore.AggregateType{"user", "org"}, database.TextArray[string]{}},
|
||||
},
|
||||
},
|
||||
{
|
||||
@ -419,8 +419,8 @@ func Test_prepareCondition(t *testing.T) {
|
||||
},
|
||||
},
|
||||
res: res{
|
||||
clause: ` WHERE aggregate_type = ANY(?) AND hlc_to_timestamp("position") < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = 'zitadel_es_pusher')`,
|
||||
values: []interface{}{[]eventstore.AggregateType{"user", "org"}},
|
||||
clause: ` WHERE aggregate_type = ANY(?) AND hlc_to_timestamp("position") < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = ANY(?))`,
|
||||
values: []interface{}{[]eventstore.AggregateType{"user", "org"}, database.TextArray[string]{}},
|
||||
},
|
||||
},
|
||||
{
|
||||
@ -439,8 +439,8 @@ func Test_prepareCondition(t *testing.T) {
|
||||
useV1: true,
|
||||
},
|
||||
res: res{
|
||||
clause: " WHERE aggregate_type = ANY(?) AND aggregate_id = ? AND event_type = ANY(?) AND creation_date::TIMESTAMP < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = 'zitadel_es_pusher')",
|
||||
values: []interface{}{[]eventstore.AggregateType{"user", "org"}, "1234", []eventstore.EventType{"user.created", "org.created"}},
|
||||
clause: " WHERE aggregate_type = ANY(?) AND aggregate_id = ? AND event_type = ANY(?) AND creation_date::TIMESTAMP < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = ANY(?))",
|
||||
values: []interface{}{[]eventstore.AggregateType{"user", "org"}, "1234", []eventstore.EventType{"user.created", "org.created"}, database.TextArray[string]{}},
|
||||
},
|
||||
},
|
||||
{
|
||||
@ -458,8 +458,8 @@ func Test_prepareCondition(t *testing.T) {
|
||||
},
|
||||
},
|
||||
res: res{
|
||||
clause: ` WHERE aggregate_type = ANY(?) AND aggregate_id = ? AND event_type = ANY(?) AND hlc_to_timestamp("position") < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = 'zitadel_es_pusher')`,
|
||||
values: []interface{}{[]eventstore.AggregateType{"user", "org"}, "1234", []eventstore.EventType{"user.created", "org.created"}},
|
||||
clause: ` WHERE aggregate_type = ANY(?) AND aggregate_id = ? AND event_type = ANY(?) AND hlc_to_timestamp("position") < (SELECT COALESCE(MIN(start), NOW())::TIMESTAMP FROM crdb_internal.cluster_transactions where application_name = ANY(?))`,
|
||||
values: []interface{}{[]eventstore.AggregateType{"user", "org"}, "1234", []eventstore.EventType{"user.created", "org.created"}, database.TextArray[string]{}},
|
||||
},
|
||||
},
|
||||
}
|
||||
@ -687,8 +687,8 @@ func Test_query_events_mocked(t *testing.T) {
|
||||
},
|
||||
fields: fields{
|
||||
mock: newMockClient(t).expectQuery(t,
|
||||
`SELECT creation_date, event_type, event_sequence, event_data, editor_user, resource_owner, instance_id, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE aggregate_type = \$1 AND creation_date::TIMESTAMP < \(SELECT COALESCE\(MIN\(start\), NOW\(\)\)::TIMESTAMP FROM crdb_internal\.cluster_transactions where application_name = 'zitadel_es_pusher'\) ORDER BY event_sequence DESC`,
|
||||
[]driver.Value{eventstore.AggregateType("user")},
|
||||
`SELECT creation_date, event_type, event_sequence, event_data, editor_user, resource_owner, instance_id, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE aggregate_type = \$1 AND creation_date::TIMESTAMP < \(SELECT COALESCE\(MIN\(start\), NOW\(\)\)::TIMESTAMP FROM crdb_internal\.cluster_transactions where application_name = ANY\(\$2\)\) ORDER BY event_sequence DESC`,
|
||||
[]driver.Value{eventstore.AggregateType("user"), database.TextArray[string]{}},
|
||||
),
|
||||
},
|
||||
res: res{
|
||||
@ -709,8 +709,8 @@ func Test_query_events_mocked(t *testing.T) {
|
||||
},
|
||||
fields: fields{
|
||||
mock: newMockClient(t).expectQuery(t,
|
||||
`SELECT creation_date, event_type, event_sequence, event_data, editor_user, resource_owner, instance_id, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE aggregate_type = \$1 AND creation_date::TIMESTAMP < \(SELECT COALESCE\(MIN\(start\), NOW\(\)\)::TIMESTAMP FROM crdb_internal\.cluster_transactions where application_name = 'zitadel_es_pusher'\) ORDER BY event_sequence LIMIT \$2`,
|
||||
[]driver.Value{eventstore.AggregateType("user"), uint64(5)},
|
||||
`SELECT creation_date, event_type, event_sequence, event_data, editor_user, resource_owner, instance_id, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE aggregate_type = \$1 AND creation_date::TIMESTAMP < \(SELECT COALESCE\(MIN\(start\), NOW\(\)\)::TIMESTAMP FROM crdb_internal\.cluster_transactions where application_name = ANY\(\$2\)\) ORDER BY event_sequence LIMIT \$3`,
|
||||
[]driver.Value{eventstore.AggregateType("user"), database.TextArray[string]{}, uint64(5)},
|
||||
),
|
||||
},
|
||||
res: res{
|
||||
@ -731,8 +731,8 @@ func Test_query_events_mocked(t *testing.T) {
|
||||
},
|
||||
fields: fields{
|
||||
mock: newMockClient(t).expectQuery(t,
|
||||
`SELECT creation_date, event_type, event_sequence, event_data, editor_user, resource_owner, instance_id, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE aggregate_type = \$1 AND creation_date::TIMESTAMP < \(SELECT COALESCE\(MIN\(start\), NOW\(\)\)::TIMESTAMP FROM crdb_internal\.cluster_transactions where application_name = 'zitadel_es_pusher'\) ORDER BY event_sequence DESC LIMIT \$2`,
|
||||
[]driver.Value{eventstore.AggregateType("user"), uint64(5)},
|
||||
`SELECT creation_date, event_type, event_sequence, event_data, editor_user, resource_owner, instance_id, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE aggregate_type = \$1 AND creation_date::TIMESTAMP < \(SELECT COALESCE\(MIN\(start\), NOW\(\)\)::TIMESTAMP FROM crdb_internal\.cluster_transactions where application_name = ANY\(\$2\)\) ORDER BY event_sequence DESC LIMIT \$3`,
|
||||
[]driver.Value{eventstore.AggregateType("user"), database.TextArray[string]{}, uint64(5)},
|
||||
),
|
||||
},
|
||||
res: res{
|
||||
@ -754,8 +754,8 @@ func Test_query_events_mocked(t *testing.T) {
|
||||
},
|
||||
fields: fields{
|
||||
mock: newMockClient(t).expectQuery(t,
|
||||
`SELECT creation_date, event_type, event_sequence, event_data, editor_user, resource_owner, instance_id, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events AS OF SYSTEM TIME '-1 ms' WHERE aggregate_type = \$1 AND creation_date::TIMESTAMP < \(SELECT COALESCE\(MIN\(start\), NOW\(\)\)::TIMESTAMP FROM crdb_internal\.cluster_transactions where application_name = 'zitadel_es_pusher'\) ORDER BY event_sequence DESC LIMIT \$2`,
|
||||
[]driver.Value{eventstore.AggregateType("user"), uint64(5)},
|
||||
`SELECT creation_date, event_type, event_sequence, event_data, editor_user, resource_owner, instance_id, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events AS OF SYSTEM TIME '-1 ms' WHERE aggregate_type = \$1 AND creation_date::TIMESTAMP < \(SELECT COALESCE\(MIN\(start\), NOW\(\)\)::TIMESTAMP FROM crdb_internal\.cluster_transactions where application_name = ANY\(\$2\)\) ORDER BY event_sequence DESC LIMIT \$3`,
|
||||
[]driver.Value{eventstore.AggregateType("user"), database.TextArray[string]{}, uint64(5)},
|
||||
),
|
||||
},
|
||||
res: res{
|
||||
@ -776,8 +776,8 @@ func Test_query_events_mocked(t *testing.T) {
|
||||
},
|
||||
fields: fields{
|
||||
mock: newMockClient(t).expectQueryErr(t,
|
||||
`SELECT creation_date, event_type, event_sequence, event_data, editor_user, resource_owner, instance_id, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE aggregate_type = \$1 AND creation_date::TIMESTAMP < \(SELECT COALESCE\(MIN\(start\), NOW\(\)\)::TIMESTAMP FROM crdb_internal\.cluster_transactions where application_name = 'zitadel_es_pusher'\) ORDER BY event_sequence DESC`,
|
||||
[]driver.Value{eventstore.AggregateType("user")},
|
||||
`SELECT creation_date, event_type, event_sequence, event_data, editor_user, resource_owner, instance_id, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE aggregate_type = \$1 AND creation_date::TIMESTAMP < \(SELECT COALESCE\(MIN\(start\), NOW\(\)\)::TIMESTAMP FROM crdb_internal\.cluster_transactions where application_name = ANY\(\$2\)\) ORDER BY event_sequence DESC`,
|
||||
[]driver.Value{eventstore.AggregateType("user"), database.TextArray[string]{}},
|
||||
sql.ErrConnDone),
|
||||
},
|
||||
res: res{
|
||||
@ -798,8 +798,8 @@ func Test_query_events_mocked(t *testing.T) {
|
||||
},
|
||||
fields: fields{
|
||||
mock: newMockClient(t).expectQueryScanErr(t,
|
||||
`SELECT creation_date, event_type, event_sequence, event_data, editor_user, resource_owner, instance_id, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE aggregate_type = \$1 AND creation_date::TIMESTAMP < \(SELECT COALESCE\(MIN\(start\), NOW\(\)\)::TIMESTAMP FROM crdb_internal\.cluster_transactions where application_name = 'zitadel_es_pusher'\) ORDER BY event_sequence DESC`,
|
||||
[]driver.Value{eventstore.AggregateType("user")},
|
||||
`SELECT creation_date, event_type, event_sequence, event_data, editor_user, resource_owner, instance_id, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE aggregate_type = \$1 AND creation_date::TIMESTAMP < \(SELECT COALESCE\(MIN\(start\), NOW\(\)\)::TIMESTAMP FROM crdb_internal\.cluster_transactions where application_name = ANY\(\$2\)\) ORDER BY event_sequence DESC`,
|
||||
[]driver.Value{eventstore.AggregateType("user"), database.TextArray[string]{}},
|
||||
&repository.Event{Seq: 100}),
|
||||
},
|
||||
res: res{
|
||||
@ -832,8 +832,8 @@ func Test_query_events_mocked(t *testing.T) {
|
||||
},
|
||||
fields: fields{
|
||||
mock: newMockClient(t).expectQuery(t,
|
||||
`SELECT creation_date, event_type, event_sequence, event_data, editor_user, resource_owner, instance_id, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE \(aggregate_type = \$1 OR \(aggregate_type = \$2 AND aggregate_id = \$3\)\) AND creation_date::TIMESTAMP < \(SELECT COALESCE\(MIN\(start\), NOW\(\)\)::TIMESTAMP FROM crdb_internal\.cluster_transactions where application_name = 'zitadel_es_pusher'\) ORDER BY event_sequence DESC LIMIT \$4`,
|
||||
[]driver.Value{eventstore.AggregateType("user"), eventstore.AggregateType("org"), "asdf42", uint64(5)},
|
||||
`SELECT creation_date, event_type, event_sequence, event_data, editor_user, resource_owner, instance_id, aggregate_type, aggregate_id, aggregate_version FROM eventstore.events WHERE \(aggregate_type = \$1 OR \(aggregate_type = \$2 AND aggregate_id = \$3\)\) AND creation_date::TIMESTAMP < \(SELECT COALESCE\(MIN\(start\), NOW\(\)\)::TIMESTAMP FROM crdb_internal\.cluster_transactions where application_name = ANY\(\$4\)\) ORDER BY event_sequence DESC LIMIT \$5`,
|
||||
[]driver.Value{eventstore.AggregateType("user"), eventstore.AggregateType("org"), "asdf42", database.TextArray[string]{}, uint64(5)},
|
||||
),
|
||||
},
|
||||
res: res{
|
||||
|
||||
@ -13,10 +13,14 @@ import (
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"github.com/zitadel/logging"
|
||||
|
||||
"github.com/zitadel/zitadel/internal/api/authz"
|
||||
"github.com/zitadel/zitadel/internal/database/dialect"
|
||||
"github.com/zitadel/zitadel/internal/eventstore"
|
||||
"github.com/zitadel/zitadel/internal/zerrors"
|
||||
)
|
||||
|
||||
var appNamePrefix = dialect.DBPurposeEventPusher.AppName() + "_"
|
||||
|
||||
func (es *Eventstore) Push(ctx context.Context, commands ...eventstore.Command) (events []eventstore.Event, err error) {
|
||||
tx, err := es.client.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
@ -27,6 +31,13 @@ func (es *Eventstore) Push(ctx context.Context, commands ...eventstore.Command)
|
||||
sequences []*latestSequence
|
||||
)
|
||||
|
||||
// needs to be set like this because psql complains about parameters in the SET statement
|
||||
_, err = tx.ExecContext(ctx, "SET application_name = '"+appNamePrefix+authz.GetInstance(ctx).InstanceID()+"'")
|
||||
if err != nil {
|
||||
logging.WithError(err).Warn("failed to set application name")
|
||||
return nil, err
|
||||
}
|
||||
|
||||
err = crdb.ExecuteInTx(ctx, &transaction{tx}, func() error {
|
||||
sequences, err = latestSequences(ctx, tx, commands)
|
||||
if err != nil {
|
||||
|
||||
Loading…
Reference in New Issue
Block a user