diff --git a/datastore/flow_postgres.go b/datastore/flow_postgres.go index 5c31d23cd..de994691d 100644 --- a/datastore/flow_postgres.go +++ b/datastore/flow_postgres.go @@ -44,38 +44,33 @@ type FlowPostgres struct { } // NewFlowPostgres initializes a SourcePostgres. -// -// Returns the postgres instance, an init function, that has to be called before -// the first use and an error. -func NewFlowPostgres(lookup environment.Environmenter) (*FlowPostgres, func(context.Context) error, error) { +func NewFlowPostgres(lookup environment.Environmenter) (*FlowPostgres, error) { addr, err := postgresDSN(lookup) if err != nil { - return nil, nil, fmt.Errorf("reading postgres dns: %w", err) + return nil, fmt.Errorf("reading postgres dns: %w", err) } config, err := pgxpool.ParseConfig(addr) if err != nil { - return nil, nil, fmt.Errorf("parse config: %w", err) + return nil, fmt.Errorf("parse config: %w", err) } config.ConnConfig.DefaultQueryExecMode = pgx.QueryExecModeSimpleProtocol - // NewWithConfig does no IO, if MinConns == 0. So the background-context - // does nothing in this case. - config.MinConns = 0 - pool, err := pgxpool.NewWithConfig(context.Background(), config) + ctx := context.Background() + pool, err := pgxpool.NewWithConfig(ctx, config) if err != nil { - return nil, nil, fmt.Errorf("creating connection pool: %w", err) + return nil, fmt.Errorf("creating connection pool: %w", err) } notifyAddr, err := postgresDSNNotify(lookup) if err != nil { - return nil, nil, fmt.Errorf("reading postgres dns for notify: %w", err) + return nil, fmt.Errorf("reading postgres dns for notify: %w", err) } notifyConf, err := pgx.ParseConfig(notifyAddr) if err != nil { - return nil, nil, fmt.Errorf("generate config for notify: %w", err) + return nil, fmt.Errorf("generate config for notify: %w", err) } flow := FlowPostgres{ @@ -83,32 +78,24 @@ func NewFlowPostgres(lookup environment.Environmenter) (*FlowPostgres, func(cont notifyConfig: notifyConf, } - init := func(ctx context.Context) error { - if err := waitPostgresAvailable(ctx, config.ConnConfig); err != nil { - return fmt.Errorf("waiting for postgres: %w", err) - } - - if err := flow.updateEnums(ctx); err != nil { - return fmt.Errorf("init enum values: %w", err) - } - return nil + if err := flow.updateEnums(ctx); err != nil { + return nil, err } - return &flow, init, nil + return &flow, nil } -// updateEnums ready the enum values from postgres for later use. func (p *FlowPostgres) updateEnums(ctx context.Context) error { c, err := p.Pool.Acquire(ctx) if err != nil { - return fmt.Errorf("acquire connection: %w", err) + return err } defer c.Release() sql := `SELECT oid, typarray FROM pg_type WHERE typtype = 'e';` rows, err := c.Conn().Query(ctx, sql) if err != nil { - return fmt.Errorf("fetch pg_types: %w", err) + return err } defer rows.Close() @@ -118,7 +105,7 @@ func (p *FlowPostgres) updateEnums(ctx context.Context) error { var oid uint32 var typarray uint32 if err := rows.Scan(&oid, &typarray); err != nil { - return fmt.Errorf("read type: %w", err) + return err } p.enums[oid] = struct{}{} @@ -327,17 +314,13 @@ func (p *FlowPostgres) Update(ctx context.Context, updateFn func(map[dskey.Key][ } if conn == nil || conn.IsClosed() { - var err error - conn, err = getPostgresConnection(ctx, p.notifyConfig) - if err != nil { - updateFn(nil, fmt.Errorf("get postgres connection: %w", err)) - continue - } + conn = getPostgresConnection(ctx, p.notifyConfig) if lastXactID > 0 { oslog.Info("Database reconnected") } - if _, err := conn.Exec(ctx, "LISTEN os_notify"); err != nil { + _, err := conn.Exec(ctx, "LISTEN os_notify") + if err != nil { updateFn(nil, fmt.Errorf("listen on channel os_notify: %w", err)) continue } @@ -438,13 +421,27 @@ func (p *FlowPostgres) Update(ctx context.Context, updateFn func(map[dskey.Key][ } } -func waitPostgresAvailable(ctx context.Context, config *pgx.ConnConfig) error { +// WaitPostgresAvailable blocks until postgres db is availabe +func WaitPostgresAvailable(lookup environment.Environmenter) error { + if _, forDocu := lookup.(*environment.ForDocu); forDocu { + return nil + } + + addr, err := postgresDSN(lookup) + if err != nil { + return fmt.Errorf("reading postgres config: %w", err) + } + + config, err := pgx.ParseConfig(addr) + if err != nil { + return fmt.Errorf("parsing postgres config: %w", err) + } + + ctx := context.Background() + for { - conn, err := getPostgresConnection(ctx, config) - if err != nil { - return fmt.Errorf("get postgres connection: %w", err) - } - err = waitDatabaseInitialized(ctx, conn) + conn := getPostgresConnection(ctx, config) + err := waitDatabaseInitialized(ctx, conn) _ = conn.Close(ctx) if err != nil { time.Sleep(1 * time.Second) @@ -456,21 +453,13 @@ func waitPostgresAvailable(ctx context.Context, config *pgx.ConnConfig) error { } // getPostgresConnection tries to connect to a database until it is successful -// TODO: Only returny on some known errors. -func getPostgresConnection(ctx context.Context, connConfig *pgx.ConnConfig) (*pgx.Conn, error) { +// TODO: Return error on credential related errors +func getPostgresConnection(ctx context.Context, connConfig *pgx.ConnConfig) *pgx.Conn { retryDelay := 1 * time.Second for { conn, err := pgx.ConnectConfig(ctx, connConfig) if err != nil { - if errors.Is(err, context.DeadlineExceeded) { - // TODO: This is inverted. The function should only retry on - // some known error and return on any unknown error. But I don't - // know what the error is, on which the service should retry. If - // nobody knows, we should fail on any error and wait for - // production to report the correct error. - return nil, fmt.Errorf("pgx connect: %w", err) - } oslog.Error("Error connecting to db: %v", err) time.Sleep(retryDelay) continue @@ -479,14 +468,13 @@ func getPostgresConnection(ctx context.Context, connConfig *pgx.ConnConfig) (*pg pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second) err = conn.Ping(pingCtx) if err != nil { - cancel() oslog.Info("Waiting for db to become ready") time.Sleep(retryDelay) continue } cancel() - return conn, nil + return conn } } diff --git a/datastore/flow_postgres_test.go b/datastore/flow_postgres_test.go index 37eea5654..e3616734f 100644 --- a/datastore/flow_postgres_test.go +++ b/datastore/flow_postgres_test.go @@ -133,14 +133,11 @@ func TestFlowPostgres(t *testing.T) { t.Fatalf("adding example data: %v", pgtest.PrityPostgresError(err, tt.insert)) } - flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) + flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) if err != nil { t.Fatalf("NewFlowPostgres(): %v", err) } defer flow.Close() - if err := init(ctx); err != nil { - t.Fatalf("init postgres: %v", err) - } keys := make([]dskey.Key, 0, len(tt.expect)) for k := range tt.expect { @@ -183,13 +180,10 @@ func TestPostgresUpdate(t *testing.T) { t.Fatalf("create connection: %v", err) } - flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) + flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) if err != nil { t.Fatalf("NewFlowPostgres(): %v", err) } - if err := init(ctx); err != nil { - t.Fatalf("init postgres: %v", err) - } keys := []dskey.Key{ dskey.MustKey("user/300/username"), @@ -259,13 +253,10 @@ func TestPostgresUpdateCollectionWithCalculatedField(t *testing.T) { t.Fatalf("create connection: %v", err) } - flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) + flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) if err != nil { t.Fatalf("NewFlowPostgres(): %v", err) } - if err := init(ctx); err != nil { - t.Fatalf("init postgres: %v", err) - } keys := []dskey.Key{ dskey.MustKey("poll/300/title"), @@ -367,13 +358,10 @@ func TestBigQuery(t *testing.T) { t.Fatalf("starting postgres: %v", err) } - flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) + flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) if err != nil { t.Fatalf("NewSource(): %v", err) } - if err := init(ctx); err != nil { - t.Fatalf("init postgres: %v", err) - } conn, err := tp.Conn(ctx) if err != nil { diff --git a/datastore/pgtest/pgtest.go b/datastore/pgtest/pgtest.go index 9955e0ef6..1b49028b8 100644 --- a/datastore/pgtest/pgtest.go +++ b/datastore/pgtest/pgtest.go @@ -21,7 +21,7 @@ import ( // But for security reasons, it can not include files outside the directory. // Therefore it is necessary to copry the files to create the sql-schema to this // folder. The embedding is necessary, to other repositories like the -// vote-service, to not need to include the meta-repo as a sub repo. +// vote-service, do not need to include the meta-repo as a sub repo. //go:generate go run ./copy_sql @@ -41,12 +41,13 @@ func RunTests(m *testing.M) int { pool, err = dockertest.NewPool(ctx, "", dockertest.WithMaxWait(2*time.Minute), ) - defer pool.Close(ctx) if err != nil { - fmt.Printf("Error: %v", err) - return 1 + panic(err) } code := m.Run() + // Close removes all tracked containers/networks and closes the client. + // Call before os.Exit — deferred functions do not run after os.Exit. + pool.Close(ctx) return code } @@ -111,7 +112,7 @@ func NewPostgresTest(t *testing.T) (*PostgresTest, error) { return nil, fmt.Errorf("create test database: %w", err) } - addr := fmt.Sprintf(`user=postgres password='openslides' host=127.0.0.1 port=%s dbname=%s`, port, database) + addr := fmt.Sprintf(`user=postgres password='openslides' host=localhost port=%s dbname=%s`, port, database) config, err := pgx.ParseConfig(addr) if err != nil { return nil, fmt.Errorf("parse config: %w", err) @@ -121,7 +122,7 @@ func NewPostgresTest(t *testing.T) (*PostgresTest, error) { pgxConfig: config, Env: map[string]string{ - "DATABASE_HOST": "127.0.0.1", + "DATABASE_HOST": "localhost", "DATABASE_PORT": port, "DATABASE_NAME": database, "DATABASE_USER": "postgres", @@ -221,14 +222,11 @@ func (tp *PostgresTest) AddData(ctx context.Context, data string) error { } // Flow returns a flow that is using the postgres instance. -func (tp *PostgresTest) Flow(ctx context.Context) (*datastore.FlowPostgres, error) { - flow, init, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) +func (tp *PostgresTest) Flow() (*datastore.FlowPostgres, error) { + flow, err := datastore.NewFlowPostgres(environment.ForTests(tp.Env)) if err != nil { return nil, fmt.Errorf("create postgres flow: %w", err) } - if err := init(ctx); err != nil { - return nil, fmt.Errorf("init postgres: %w", err) - } return flow, nil } diff --git a/environment/README.md b/environment/README.md deleted file mode 100644 index 43b743e38..000000000 --- a/environment/README.md +++ /dev/null @@ -1,72 +0,0 @@ -# Environment - -The environment package is used, to configure a go-service. The Package contains -functions, to access environment variables in a way, so that the configuration -for a service can automaticly be generated. - -For this to work, the initalization phase is not allowed to do any IO. In other -case, the IO would also happen, then the docutmentation is build. - -Therefore, a service service has to start in different phases. - -The first phase ready the environment variables, but does nothing else. For -example, it could be build like this: - -```go -func initService(lookup environment.Environmenter) (func(context.Context) error, error) { - // No IO allowed in this function - oslog.InitLog(lookup) - someType, background := somePackage.New(lookup) - - service := func(ctx context.Context) error { - // This is a callback. IO is allowed here. - - oslog.Info("Listen on %s", listenAddr) - return http.Run(ctx, someType) - } - return service, nil -} -``` - -All functions inside initService (like `somePackage.New`) are also not allowed -to do IO. If IO is needed, it has to be done in the callback. - - -When this is done, the service can either be started: - -```go -func run(ctx context.Context) error { - lookup := new(environment.ForProduction) - - service, err := initService(lookup) - if err != nil { - return fmt.Errorf("init services: %w", err) - } - - return service(ctx) -} -``` - -Or the documentation can be build: - -```go -func buildDocu() error { - lookup := new(environment.ForDocu) - - if _, err := initService(lookup); err != nil { - return fmt.Errorf("init services: %w", err) - } - - doc, err := lookup.BuildDoc() - if err != nil { - return fmt.Errorf("build doc: %w", err) - } - - fmt.Println(doc) - return nil -} -``` - -As a maxim: If the function has a `context.Context` as an argument, you can do -IO. If the function has not a `context.Context` as an argument, do not create -one with `context.Background()`. You have to do the IO elsewhere. diff --git a/environment/environment.go b/environment/environment.go index 8c54f70a1..33272ecc0 100644 --- a/environment/environment.go +++ b/environment/environment.go @@ -68,13 +68,7 @@ func (v Variable) ValueOr(loopup Environmenter, other Variable) string { // Environmenter is an type, that can return the value for environment // variables. // -// This type is used to access environment variables, but also, to autogenerate, -// which environment variables a service needs. -// -// For that to work, a function, that takes this type is only allowed to -// initialize datastructures. It is not allowed to do any IO or other blocking -// function calls. Especially, not network request like waiting for postgres. If -// IO is needed, it has to be done in a callback. +// It also saves the used environment variables. type Environmenter interface { Getenv(key string) string UseVariable(v Variable)