hoernschen/dendrite
1
0
Fork 0
mirror of https://github.com/hoernschen/dendrite.git synced 2025-07-31 13:22:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Configuration format v1 (#1230)

Browse source 
* Initial pass at refactoring config (not finished)

* Don't forget current state and EDU servers

* More shifting around

* Update server key API tests

* Fix roomserver test

* Fix more tests

* Further tweaks

* Fix current state server test (sort of)

* Maybe fix appservices

* Fix client API test

* Include database connection string in database options

* Fix sync API build

* Update config test

* Fix unit tests

* Fix federation sender build

* Fix gobind build

* Set Listen address for all services in HTTP monolith mode

* Validate config, reinstate appservice derived in directory, tweaks

* Tweak federation API test

* Set MaxOpenConnections/MaxIdleConnections to previous values

* Update generate-config
This commit is contained in:
Neil Alexander 2020-08-10 14:18:04 +01:00 • committed by GitHub
parent fdabba1851
commit 4b09f445c9
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
155 changed files with 1716 additions and 1503 deletions
Download patch file Download diff file Expand all files Collapse all files

4
syncapi/consumers/clientapi.go
View file

@ -37,14 +37,14 @@ type OutputClientDataConsumer struct {
// NewOutputClientDataConsumer creates a new OutputClientData consumer. Call Start() to begin consuming from room servers.
func NewOutputClientDataConsumer(
cfg *config.Dendrite,
cfg *config.SyncAPI,
kafkaConsumer sarama.Consumer,
n *sync.Notifier,
store storage.Database,
) *OutputClientDataConsumer {
consumer := internal.ContinualConsumer{
Topic: string(cfg.Kafka.Topics.OutputClientData),
Topic: string(cfg.Matrix.Kafka.Topics.OutputClientData),
Consumer: kafkaConsumer,
PartitionStore: store,
}

4
syncapi/consumers/eduserver_sendtodevice.go
View file

@ -41,14 +41,14 @@ type OutputSendToDeviceEventConsumer struct {
// NewOutputSendToDeviceEventConsumer creates a new OutputSendToDeviceEventConsumer.
// Call Start() to begin consuming from the EDU server.
func NewOutputSendToDeviceEventConsumer(
cfg *config.Dendrite,
cfg *config.SyncAPI,
kafkaConsumer sarama.Consumer,
n *sync.Notifier,
store storage.Database,
) *OutputSendToDeviceEventConsumer {
consumer := internal.ContinualConsumer{
Topic: string(cfg.Kafka.Topics.OutputSendToDeviceEvent),
Topic: string(cfg.Matrix.Kafka.Topics.OutputSendToDeviceEvent),
Consumer: kafkaConsumer,
PartitionStore: store,
}

4
syncapi/consumers/eduserver_typing.go
View file

@ -37,14 +37,14 @@ type OutputTypingEventConsumer struct {
// NewOutputTypingEventConsumer creates a new OutputTypingEventConsumer.
// Call Start() to begin consuming from the EDU server.
func NewOutputTypingEventConsumer(
cfg *config.Dendrite,
cfg *config.SyncAPI,
kafkaConsumer sarama.Consumer,
n *sync.Notifier,
store storage.Database,
) *OutputTypingEventConsumer {
consumer := internal.ContinualConsumer{
Topic: string(cfg.Kafka.Topics.OutputTypingEvent),
Topic: string(cfg.Matrix.Kafka.Topics.OutputTypingEvent),
Consumer: kafkaConsumer,
PartitionStore: store,
}

4
syncapi/consumers/roomserver.go
View file

@ -40,7 +40,7 @@ type OutputRoomEventConsumer struct {
// NewOutputRoomEventConsumer creates a new OutputRoomEventConsumer. Call Start() to begin consuming from room servers.
func NewOutputRoomEventConsumer(
cfg *config.Dendrite,
cfg *config.SyncAPI,
kafkaConsumer sarama.Consumer,
n *sync.Notifier,
store storage.Database,
@ -49,7 +49,7 @@ func NewOutputRoomEventConsumer(
) *OutputRoomEventConsumer {
consumer := internal.ContinualConsumer{
Topic: string(cfg.Kafka.Topics.OutputRoomEvent),
Topic: string(cfg.Matrix.Kafka.Topics.OutputRoomEvent),
Consumer: kafkaConsumer,
PartitionStore: store,
}

4
syncapi/routing/messages.go
View file

@ -36,7 +36,7 @@ type messagesReq struct {
db storage.Database
rsAPI api.RoomserverInternalAPI
federation *gomatrixserverlib.FederationClient
cfg *config.Dendrite
cfg *config.SyncAPI
roomID string
from *types.TopologyToken
to *types.TopologyToken
@ -61,7 +61,7 @@ func OnIncomingMessagesRequest(
req *http.Request, db storage.Database, roomID string,
federation *gomatrixserverlib.FederationClient,
rsAPI api.RoomserverInternalAPI,
cfg *config.Dendrite,
cfg *config.SyncAPI,
) util.JSONResponse {
var err error

2
syncapi/routing/routing.go
View file

@ -39,7 +39,7 @@ func Setup(
publicAPIMux *mux.Router, srp *sync.RequestPool, syncDB storage.Database,
userAPI userapi.UserInternalAPI, federation *gomatrixserverlib.FederationClient,
rsAPI api.RoomserverInternalAPI,
cfg *config.Dendrite,
cfg *config.SyncAPI,
) {
r0mux := publicAPIMux.PathPrefix(pathPrefixR0).Subrouter()

5
syncapi/storage/postgres/syncserver.go
View file

@ -21,6 +21,7 @@ import (
// Import the postgres database driver.
_ "github.com/lib/pq"
"github.com/matrix-org/dendrite/eduserver/cache"
"github.com/matrix-org/dendrite/internal/config"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/syncapi/storage/shared"
)
@ -34,10 +35,10 @@ type SyncServerDatasource struct {
}
// NewDatabase creates a new sync server database
func NewDatabase(dbDataSourceName string, dbProperties sqlutil.DbProperties) (*SyncServerDatasource, error) {
func NewDatabase(dbProperties *config.DatabaseOptions) (*SyncServerDatasource, error) {
var d SyncServerDatasource
var err error
if d.db, err = sqlutil.Open("postgres", dbDataSourceName, dbProperties); err != nil {
if d.db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err
}
if err = d.PartitionOffsetStatements.Prepare(d.db, "syncapi"); err != nil {

10
syncapi/storage/sqlite3/syncserver.go
View file

@ -22,6 +22,7 @@ import (
_ "github.com/mattn/go-sqlite3"
"github.com/matrix-org/dendrite/eduserver/cache"
"github.com/matrix-org/dendrite/internal/config"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/syncapi/storage/shared"
)
@ -37,13 +38,10 @@ type SyncServerDatasource struct {
// NewDatabase creates a new sync server database
// nolint: gocyclo
func NewDatabase(dataSourceName string) (*SyncServerDatasource, error) {
func NewDatabase(dbProperties *config.DatabaseOptions) (*SyncServerDatasource, error) {
var d SyncServerDatasource
cs, err := sqlutil.ParseFileURI(dataSourceName)
if err != nil {
return nil, err
}
if d.db, err = sqlutil.Open(sqlutil.SQLiteDriverName(), cs, nil); err != nil {
var err error
if d.db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err
}
if err = d.prepare(); err != nil {

22
syncapi/storage/storage.go
View file

@ -17,25 +17,21 @@
package storage
import (
"net/url"
"fmt"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/internal/config"
"github.com/matrix-org/dendrite/syncapi/storage/postgres"
"github.com/matrix-org/dendrite/syncapi/storage/sqlite3"
)
// NewSyncServerDatasource opens a database connection.
func NewSyncServerDatasource(dataSourceName string, dbProperties sqlutil.DbProperties) (Database, error) {
uri, err := url.Parse(dataSourceName)
if err != nil {
return postgres.NewDatabase(dataSourceName, dbProperties)
}
switch uri.Scheme {
case "postgres":
return postgres.NewDatabase(dataSourceName, dbProperties)
case "file":
return sqlite3.NewDatabase(dataSourceName)
func NewSyncServerDatasource(dbProperties *config.DatabaseOptions) (Database, error) {
switch {
case dbProperties.ConnectionString.IsSQLite():
return sqlite3.NewDatabase(dbProperties)
case dbProperties.ConnectionString.IsPostgres():
return postgres.NewDatabase(dbProperties)
default:
return postgres.NewDatabase(dataSourceName, dbProperties)
return nil, fmt.Errorf("unexpected database type")
}
}

5
syncapi/storage/storage_test.go
View file

@ -9,6 +9,7 @@ import (
"testing"
"time"
"github.com/matrix-org/dendrite/internal/config"
"github.com/matrix-org/dendrite/syncapi/storage"
"github.com/matrix-org/dendrite/syncapi/storage/sqlite3"
"github.com/matrix-org/dendrite/syncapi/types"
@ -59,7 +60,9 @@ func MustCreateDatabase(t *testing.T) storage.Database {
t.Fatalf("tried to delete stale test database but failed: %s", err)
}
}
db, err := sqlite3.NewDatabase(fmt.Sprintf("file:%s", dbname))
db, err := sqlite3.NewDatabase(&config.DatabaseOptions{
ConnectionString: config.DataSource(fmt.Sprintf("file:%s", dbname)),
})
if err != nil {
t.Fatalf("NewSyncServerDatasource returned %s", err)
}

24
syncapi/storage/storage_wasm.go
View file

@ -16,27 +16,19 @@ package storage
import (
"fmt"
"net/url"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/internal/config"
"github.com/matrix-org/dendrite/syncapi/storage/sqlite3"
)
// NewPublicRoomsServerDatabase opens a database connection.
func NewSyncServerDatasource(
dataSourceName string,
dbProperties sqlutil.DbProperties, // nolint:unparam
) (Database, error) {
uri, err := url.Parse(dataSourceName)
if err != nil {
return nil, fmt.Errorf("Cannot use postgres implementation")
}
switch uri.Scheme {
case "postgres":
return nil, fmt.Errorf("Cannot use postgres implementation")
case "file":
return sqlite3.NewDatabase(dataSourceName)
func NewSyncServerDatasource(dbProperties *config.DatabaseOptions) (Database, error) {
switch {
case dbProperties.ConnectionString.IsSQLite():
return sqlite3.NewDatabase(dbProperties)
case dbProperties.ConnectionString.IsPostgres():
return nil, fmt.Errorf("can't use Postgres implementation")
default:
return nil, fmt.Errorf("Cannot use postgres implementation")
return nil, fmt.Errorf("unexpected database type")
}
}

6
syncapi/syncapi.go
View file

@ -44,9 +44,9 @@ func AddPublicRoutes(
keyAPI keyapi.KeyInternalAPI,
currentStateAPI currentstateapi.CurrentStateInternalAPI,
federation *gomatrixserverlib.FederationClient,
cfg *config.Dendrite,
cfg *config.SyncAPI,
) {
syncDB, err := storage.NewSyncServerDatasource(string(cfg.Database.SyncAPI), cfg.DbProperties())
syncDB, err := storage.NewSyncServerDatasource(&cfg.Database)
if err != nil {
logrus.WithError(err).Panicf("failed to connect to sync db")
}
@ -65,7 +65,7 @@ func AddPublicRoutes(
requestPool := sync.NewRequestPool(syncDB, notifier, userAPI, keyAPI, currentStateAPI)
keyChangeConsumer := consumers.NewOutputKeyChangeEventConsumer(
cfg.Matrix.ServerName, string(cfg.Kafka.Topics.OutputKeyChangeEvent),
cfg.Matrix.ServerName, string(cfg.Matrix.Kafka.Topics.OutputKeyChangeEvent),
consumer, notifier, keyAPI, currentStateAPI, syncDB,
)
if err = keyChangeConsumer.Start(); err != nil {