zitadel/internal/query/prepare_test.go
Silvan 9c3e5e467b
perf(query): remove transactions for queries (#8614)
# Which Problems Are Solved

Queries currently execute 3 statements, begin, query, commit

# How the Problems Are Solved

remove transaction handling from query methods in database package

# Additional Changes

- Bump versions of `core_grpc_dependencies`-receipt in Makefile

# Additional info

During load tests we saw a lot of idle transactions of `zitadel_queries`
application name which is the connection pool used to query data in
zitadel. Executed query:

`select query_start - xact_start, pid, application_name, backend_start,
xact_start, query_start, state_change, wait_event_type,
wait_event,substring(query, 1, 200) query from pg_stat_activity where
datname = 'zitadel' and state <> 'idle';`

Mostly the last query executed was `begin isolation level read committed
read only`.

example: 

```
    ?column?     |  pid  |      application_name      |         backend_start         |          xact_start           |          query_start          |         state_change          | wait_event_type |  wait_event  |                                                                                                  query                                                                                                   
-----------------+-------+----------------------------+-------------------------------+-------------------------------+-------------------------------+-------------------------------+-----------------+--------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 00:00:00        | 33030 | zitadel_queries            | 2024-10-16 16:25:53.906036+00 | 2024-10-16 16:30:19.191661+00 | 2024-10-16 16:30:19.191661+00 | 2024-10-16 16:30:19.19169+00  | Client          | ClientRead   | begin isolation level read committed read only
 00:00:00        | 33035 | zitadel_queries            | 2024-10-16 16:25:53.909629+00 | 2024-10-16 16:30:19.19179+00  | 2024-10-16 16:30:19.19179+00  | 2024-10-16 16:30:19.191805+00 | Client          | ClientRead   | begin isolation level read committed read only
 00:00:00.00412  | 33028 | zitadel_queries            | 2024-10-16 16:25:53.904247+00 | 2024-10-16 16:30:19.187734+00 | 2024-10-16 16:30:19.191854+00 | 2024-10-16 16:30:19.191964+00 | Client          | ClientRead   | SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type 
 00:00:00.084662 | 33134 | zitadel_es_pusher          | 2024-10-16 16:29:54.979692+00 | 2024-10-16 16:30:19.178578+00 | 2024-10-16 16:30:19.26324+00  | 2024-10-16 16:30:19.263267+00 | Client          | ClientRead   | RELEASE SAVEPOINT cockroach_restart
 00:00:00.084768 | 33139 | zitadel_es_pusher          | 2024-10-16 16:29:54.979585+00 | 2024-10-16 16:30:19.180762+00 | 2024-10-16 16:30:19.26553+00  | 2024-10-16 16:30:19.265531+00 | LWLock          | WALWriteLock | commit
 00:00:00.077377 | 33136 | zitadel_es_pusher          | 2024-10-16 16:29:54.978582+00 | 2024-10-16 16:30:19.187883+00 | 2024-10-16 16:30:19.26526+00  | 2024-10-16 16:30:19.265431+00 | Client          | ClientRead   | WITH existing AS (                                                                                                                                                                                      +
                 |       |                            |                               |                               |                               |                               |                 |              |     (SELECT instance_id, aggregate_type, aggregate_id, "sequence" FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type = $2 AND aggregate_id = $3 ORDER BY "sequence" DE
 00:00:00.012309 | 33123 | zitadel_es_pusher          | 2024-10-16 16:29:54.963484+00 | 2024-10-16 16:30:19.175066+00 | 2024-10-16 16:30:19.187375+00 | 2024-10-16 16:30:19.187376+00 | IO              | WalSync      | commit
 00:00:00        | 33034 | zitadel_queries            | 2024-10-16 16:25:53.90791+00  | 2024-10-16 16:30:19.262921+00 | 2024-10-16 16:30:19.262921+00 | 2024-10-16 16:30:19.263133+00 | Client          | ClientRead   | begin isolation level read committed read only
 00:00:00        | 33039 | zitadel_queries            | 2024-10-16 16:25:53.914106+00 | 2024-10-16 16:30:19.191676+00 | 2024-10-16 16:30:19.191676+00 | 2024-10-16 16:30:19.191687+00 | Client          | ClientRead   | begin isolation level read committed read only
 00:00:00.24539  | 33083 | zitadel_projection_spooler | 2024-10-16 16:27:49.895548+00 | 2024-10-16 16:30:19.020058+00 | 2024-10-16 16:30:19.265448+00 | 2024-10-16 16:30:19.26546+00  | Client          | ClientRead   | SAVEPOINT exec_stmt
 00:00:00        | 33125 | zitadel_es_pusher          | 2024-10-16 16:29:54.963859+00 | 2024-10-16 16:30:19.191715+00 | 2024-10-16 16:30:19.191715+00 | 2024-10-16 16:30:19.191729+00 | Client          | ClientRead   | begin
 00:00:00.004292 | 33032 | zitadel_queries            | 2024-10-16 16:25:53.906624+00 | 2024-10-16 16:30:19.187713+00 | 2024-10-16 16:30:19.192005+00 | 2024-10-16 16:30:19.192062+00 | Client          | ClientRead   | SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type 
 00:00:00        | 33031 | zitadel_queries            | 2024-10-16 16:25:53.906422+00 | 2024-10-16 16:30:19.191625+00 | 2024-10-16 16:30:19.191625+00 | 2024-10-16 16:30:19.191645+00 | Client          | ClientRead   | begin isolation level read committed read only

```

The amount of idle transactions is significantly less if the query
transactions are removed:

example: 

```
    ?column?     |  pid  |      application_name      |         backend_start         |          xact_start           |          query_start          |         state_change          | wait_event_type | wait_event |                                                                                                  query                                                                                                   
-----------------+-------+----------------------------+-------------------------------+-------------------------------+-------------------------------+-------------------------------+-----------------+------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 00:00:00.000094 | 32741 | zitadel_queries            | 2024-10-16 16:23:49.73935+00  | 2024-10-16 16:24:59.785589+00 | 2024-10-16 16:24:59.785683+00 | 2024-10-16 16:24:59.785684+00 |                 |            | SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type 
 00:00:00        | 32762 | zitadel_es_pusher          | 2024-10-16 16:24:02.275136+00 | 2024-10-16 16:24:59.784586+00 | 2024-10-16 16:24:59.784586+00 | 2024-10-16 16:24:59.784607+00 | Client          | ClientRead | begin
 00:00:00.000167 | 32742 | zitadel_queries            | 2024-10-16 16:23:49.740489+00 | 2024-10-16 16:24:59.784274+00 | 2024-10-16 16:24:59.784441+00 | 2024-10-16 16:24:59.784442+00 |                 |            | with usr as (                                                                                                                                                                                           +
                 |       |                            |                               |                               |                               |                               |                 |            |         select u.id, u.creation_date, u.change_date, u.sequence, u.state, u.resource_owner, u.username, n.login_name as preferred_login_name                                                            +
                 |       |                            |                               |                               |                               |                               |                 |            |         from projections.users13 u                                                                                                                                                                      +
                 |       |                            |                               |                               |                               |                               |                 |            |         left join projections.l
 00:00:00.256014 | 32759 | zitadel_projection_spooler | 2024-10-16 16:24:01.418429+00 | 2024-10-16 16:24:59.52959+00  | 2024-10-16 16:24:59.785604+00 | 2024-10-16 16:24:59.785649+00 | Client          | ClientRead | UPDATE projections.milestones SET reached_date = $1 WHERE (instance_id = $2) AND (type = $3) AND (reached_date IS NULL)
 00:00:00.014199 | 32773 | zitadel_es_pusher          | 2024-10-16 16:24:02.320404+00 | 2024-10-16 16:24:59.769509+00 | 2024-10-16 16:24:59.783708+00 | 2024-10-16 16:24:59.783709+00 | IO              | WalSync    | commit
 00:00:00        | 32765 | zitadel_es_pusher          | 2024-10-16 16:24:02.28173+00  | 2024-10-16 16:24:59.780413+00 | 2024-10-16 16:24:59.780413+00 | 2024-10-16 16:24:59.780426+00 | Client          | ClientRead | begin
 00:00:00.012729 | 32777 | zitadel_es_pusher          | 2024-10-16 16:24:02.339737+00 | 2024-10-16 16:24:59.767432+00 | 2024-10-16 16:24:59.780161+00 | 2024-10-16 16:24:59.780195+00 | Client          | ClientRead | RELEASE SAVEPOINT cockroach_restart
```

---------

Co-authored-by: Tim Möhlmann <tim+github@zitadel.com>
Co-authored-by: Livio Spring <livio.a@gmail.com>
Co-authored-by: Max Peintner <max@caos.ch>
Co-authored-by: Elio Bischof <elio@zitadel.com>
Co-authored-by: Stefan Benz <46600784+stebenz@users.noreply.github.com>
Co-authored-by: Miguel Cabrerizo <30386061+doncicuto@users.noreply.github.com>
Co-authored-by: Joakim Lodén <Loddan@users.noreply.github.com>
Co-authored-by: Yxnt <Yxnt@users.noreply.github.com>
Co-authored-by: Stefan Benz <stefan@caos.ch>
Co-authored-by: Harsha Reddy <harsha.reddy@klaviyo.com>
Co-authored-by: Zach H <zhirschtritt@gmail.com>
2024-11-04 10:06:14 +01:00

396 lines
10 KiB
Go

package query
import (
"context"
"database/sql"
"database/sql/driver"
"errors"
"fmt"
"log"
"reflect"
"testing"
"time"
"github.com/DATA-DOG/go-sqlmock"
sq "github.com/Masterminds/squirrel"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/zitadel/zitadel/internal/database"
db_mock "github.com/zitadel/zitadel/internal/database/mock"
)
var (
testNow = time.Now()
dayNow = testNow.Truncate(24 * time.Hour)
)
// assertPrepare checks if the prepare func executes the correct sql query and returns the correct object
// prepareFunc must be of type
// func() (sq.SelectBuilder, func(*sql.Rows) (*struct, error))
// or
// func() (sq.SelectBuilder, func(*sql.Row) (*struct, error))
// expectedObject represents the return value of scan
// sqlExpectation represents the query executed on the database
func assertPrepare(t *testing.T, prepareFunc, expectedObject interface{}, sqlExpectation sqlExpectation, isErr checkErr, prepareArgs ...reflect.Value) bool {
t.Helper()
client, mock, err := sqlmock.New(sqlmock.ValueConverterOption(new(db_mock.TypeConverter)))
if err != nil {
t.Fatalf("failed to build mock client: %v", err)
}
mock = sqlExpectation(mock)
builder, scan, err := execPrepare(prepareFunc, prepareArgs)
if err != nil {
t.Error(err)
return false
}
errCheck := func(err error) (error, bool) {
if isErr == nil {
if err == nil {
return nil, true
} else {
return fmt.Errorf("no error expected got: %w", err), false
}
}
return isErr(err)
}
object, ok, didScan := execScan(t, &database.DB{DB: client}, builder, scan, errCheck)
if !ok {
t.Error(object)
return false
}
if didScan {
if !assert.Equal(t, expectedObject, object) {
return false
}
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("sql expectations not met: %v", err)
return false
}
return true
}
type checkErr func(error) (err error, ok bool)
type sqlExpectation func(sqlmock.Sqlmock) sqlmock.Sqlmock
func mockQuery(stmt string, cols []string, row []driver.Value, args ...driver.Value) func(m sqlmock.Sqlmock) sqlmock.Sqlmock {
return func(m sqlmock.Sqlmock) sqlmock.Sqlmock {
q := m.ExpectQuery(stmt).WithArgs(args...)
result := m.NewRows(cols)
if len(row) > 0 {
result.AddRow(row...)
}
q.WillReturnRows(result)
return m
}
}
func mockQueryScanErr(stmt string, cols []string, row []driver.Value, args ...driver.Value) func(m sqlmock.Sqlmock) sqlmock.Sqlmock {
return func(m sqlmock.Sqlmock) sqlmock.Sqlmock {
q := m.ExpectQuery(stmt).WithArgs(args...)
result := m.NewRows(cols)
if len(row) > 0 {
result.AddRow(row...)
}
q.WillReturnRows(result)
return m
}
}
func mockQueries(stmt string, cols []string, rows [][]driver.Value, args ...driver.Value) func(m sqlmock.Sqlmock) sqlmock.Sqlmock {
return func(m sqlmock.Sqlmock) sqlmock.Sqlmock {
q := m.ExpectQuery(stmt).WithArgs(args...)
result := m.NewRows(cols)
count := uint64(len(rows))
for _, row := range rows {
if cols[len(cols)-1] == "count" {
row = append(row, count)
}
result.AddRow(row...)
}
q.WillReturnRows(result)
q.RowsWillBeClosed()
return m
}
}
func mockQueriesScanErr(stmt string, cols []string, rows [][]driver.Value, args ...driver.Value) func(m sqlmock.Sqlmock) sqlmock.Sqlmock {
return func(m sqlmock.Sqlmock) sqlmock.Sqlmock {
q := m.ExpectQuery(stmt).WithArgs(args...)
result := m.NewRows(cols)
count := uint64(len(rows))
for _, row := range rows {
if cols[len(cols)-1] == "count" {
row = append(row, count)
}
result.AddRow(row...)
}
q.WillReturnRows(result)
q.RowsWillBeClosed()
return m
}
}
func mockQueryErr(stmt string, err error, args ...driver.Value) func(m sqlmock.Sqlmock) sqlmock.Sqlmock {
return func(m sqlmock.Sqlmock) sqlmock.Sqlmock {
q := m.ExpectQuery(stmt).WithArgs(args...)
q.WillReturnError(err)
return m
}
}
func execMock(t testing.TB, exp sqlExpectation, run func(db *sql.DB)) {
db, mock, err := sqlmock.New(sqlmock.ValueConverterOption(new(db_mock.TypeConverter)))
require.NoError(t, err)
defer db.Close()
mock = exp(mock)
run(db)
assert.NoError(t, mock.ExpectationsWereMet())
}
var (
rowType = reflect.TypeOf(&sql.Row{})
rowsType = reflect.TypeOf(&sql.Rows{})
selectBuilderType = reflect.TypeOf(sq.SelectBuilder{})
)
func execScan(t testing.TB, client *database.DB, builder sq.SelectBuilder, scan interface{}, errCheck checkErr) (object interface{}, ok bool, didScan bool) {
t.Helper()
scanType := reflect.TypeOf(scan)
err := validateScan(scanType)
if err != nil {
return err, false, false
}
stmt, args, err := builder.ToSql()
if err != nil {
return fmt.Errorf("unexpected error from sql builder: %w", err), false, false
}
//resultSet represents *sql.Row or *sql.Rows,
// depending on whats assignable to the scan function
var res []reflect.Value
//execute sql stmt
// if scan(*sql.Rows)...
if scanType.In(0).AssignableTo(rowsType) {
err = client.Query(func(rows *sql.Rows) error {
didScan = true
res = reflect.ValueOf(scan).Call([]reflect.Value{reflect.ValueOf(rows)})
if err, ok := res[1].Interface().(error); ok {
return err
}
return nil
}, stmt, args...)
// if scan(*sql.Row)...
} else if scanType.In(0).AssignableTo(rowType) {
err = client.QueryRow(func(r *sql.Row) error {
if r.Err() != nil {
return r.Err()
}
didScan = true
res = reflect.ValueOf(scan).Call([]reflect.Value{reflect.ValueOf(r)})
if err, ok := res[1].Interface().(error); ok {
return err
}
return nil
}, stmt, args...)
} else {
return errors.New("scan: parameter must be *sql.Row or *sql.Rows"), false, false
}
if err != nil {
err, ok := errCheck(err)
if !ok {
t.Fatal(err)
}
if didScan {
return res[0].Interface(), ok, didScan
}
return err, ok, didScan
}
//check for error
if res[1].Interface() != nil {
if err, ok := errCheck(res[1].Interface().(error)); !ok {
return fmt.Errorf("scan failed: %w", err), false, didScan
}
}
return res[0].Interface(), true, didScan
}
func validateScan(scanType reflect.Type) error {
if scanType.Kind() != reflect.Func {
return errors.New("scan is not a function")
}
if scanType.NumIn() != 1 {
return fmt.Errorf("scan: invalid number of inputs: want: 1 got %d", scanType.NumIn())
}
if scanType.NumOut() != 2 {
return fmt.Errorf("scan: invalid number of outputs: want: 2 got %d", scanType.NumOut())
}
return nil
}
func execPrepare(prepare interface{}, args []reflect.Value) (builder sq.SelectBuilder, scan interface{}, err error) {
prepareVal := reflect.ValueOf(prepare)
if err := validatePrepare(prepareVal.Type()); err != nil {
return sq.SelectBuilder{}, nil, err
}
res := prepareVal.Call(args)
return res[0].Interface().(sq.SelectBuilder), res[1].Interface(), nil
}
func validatePrepare(prepareType reflect.Type) error {
if prepareType.Kind() != reflect.Func {
return errors.New("prepare is not a function")
}
if prepareType.NumIn() != 0 && prepareType.NumIn() != 2 {
return fmt.Errorf("prepare: invalid number of inputs: want: 0 or 2 got %d", prepareType.NumIn())
}
if prepareType.NumOut() != 2 {
return fmt.Errorf("prepare: invalid number of outputs: want: 2 got %d", prepareType.NumOut())
}
if prepareType.Out(0) != selectBuilderType {
return fmt.Errorf("prepare: first return value must be: %s got %s", selectBuilderType, prepareType.Out(0))
}
if prepareType.Out(1).Kind() != reflect.Func {
return fmt.Errorf("prepare: second return value must be: %s got %s", reflect.Func, prepareType.Out(1))
}
return nil
}
func TestValidateScan(t *testing.T) {
tests := []struct {
name string
t reflect.Type
expectErr bool
}{
{
name: "not a func",
t: reflect.TypeOf(&struct{}{}),
expectErr: true,
},
{
name: "wong input count",
t: reflect.TypeOf(func() (*struct{}, error) {
log.Fatal("should not be executed")
return nil, nil
}),
expectErr: true,
},
{
name: "wrong output count",
t: reflect.TypeOf(func(interface{}) error {
log.Fatal("should not be executed")
return nil
}),
expectErr: true,
},
{
name: "correct",
t: reflect.TypeOf(func(interface{}) (*struct{}, error) {
log.Fatal("should not be executed")
return nil, nil
}),
expectErr: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := validateScan(tt.t)
if (err != nil) != tt.expectErr {
t.Errorf("unexpected err: %v", err)
}
})
}
}
func TestValidatePrepare(t *testing.T) {
tests := []struct {
name string
t reflect.Type
expectErr bool
}{
{
name: "not a func",
t: reflect.TypeOf(&struct{}{}),
expectErr: true,
},
{
name: "wong input count",
t: reflect.TypeOf(func(int) (sq.SelectBuilder, func(*sql.Rows) (interface{}, error)) {
log.Fatal("should not be executed")
return sq.SelectBuilder{}, nil
}),
expectErr: true,
},
{
name: "wrong output count",
t: reflect.TypeOf(func() sq.SelectBuilder {
log.Fatal("should not be executed")
return sq.SelectBuilder{}
}),
expectErr: true,
},
{
name: "first output type wrong",
t: reflect.TypeOf(func() (*struct{}, func(*sql.Rows) (interface{}, error)) {
log.Fatal("should not be executed")
return nil, nil
}),
expectErr: true,
},
{
name: "second output type wrong",
t: reflect.TypeOf(func() (sq.SelectBuilder, *struct{}) {
log.Fatal("should not be executed")
return sq.SelectBuilder{}, nil
}),
expectErr: true,
},
{
name: "correct",
t: reflect.TypeOf(func(context.Context, prepareDatabase) (sq.SelectBuilder, func(*sql.Rows) (interface{}, error)) {
log.Fatal("should not be executed")
return sq.SelectBuilder{}, nil
}),
expectErr: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := validatePrepare(tt.t)
if (err != nil) != tt.expectErr {
t.Errorf("unexpected err: %v", err)
}
})
}
}
type prepareDB struct{}
const asOfSystemTime = " AS OF SYSTEM TIME '-1 ms' "
func (*prepareDB) Timetravel(time.Duration) string { return asOfSystemTime }
var defaultPrepareArgs = []reflect.Value{reflect.ValueOf(context.Background()), reflect.ValueOf(new(prepareDB))}
func (*prepareDB) DatabaseName() string { return "db" }
func (*prepareDB) Username() string { return "user" }
func (*prepareDB) Type() string { return "type" }