mirror of
https://github.com/hoernschen/dendrite.git
synced 2025-08-01 05:42:46 +00:00
Jetstream package
This commit is contained in:
Neil Alexander
parent
a92936ffa4
commit
4c45de81b3
27 changed files with 220 additions and 288 deletions
|
|
@ -329,7 +329,7 @@ func (c *Dendrite) Verify(configErrs *ConfigErrors, isMonolith bool) {
|
|||
}
|
||||
|
||||
func (c *Dendrite) Wiring() {
|
||||
c.Global.Kafka.Matrix = &c.Global
|
||||
c.Global.JetStream.Matrix = &c.Global
|
||||
c.ClientAPI.Matrix = &c.Global
|
||||
c.EDUServer.Matrix = &c.Global
|
||||
c.FederationAPI.Matrix = &c.Global
|
||||
|
|
|
|||
|
|
@ -43,8 +43,8 @@ type Global struct {
|
|||
// Defaults to an empty array.
|
||||
TrustedIDServers []string `yaml:"trusted_third_party_id_servers"`
|
||||
|
||||
// Kafka/Naffka configuration
|
||||
Kafka Kafka `yaml:"kafka"`
|
||||
// JetStream configuration
|
||||
JetStream JetStream `yaml:"jetstream"`
|
||||
|
||||
// Metrics configuration
|
||||
Metrics Metrics `yaml:"metrics"`
|
||||
|
|
@ -63,7 +63,7 @@ func (c *Global) Defaults() {
|
|||
c.KeyID = "ed25519:auto"
|
||||
c.KeyValidityPeriod = time.Hour * 24 * 7
|
||||
|
||||
c.Kafka.Defaults()
|
||||
c.JetStream.Defaults()
|
||||
c.Metrics.Defaults()
|
||||
c.DNSCache.Defaults()
|
||||
c.Sentry.Defaults()
|
||||
|
|
@ -73,7 +73,7 @@ func (c *Global) Verify(configErrs *ConfigErrors, isMonolith bool) {
|
|||
checkNotEmpty(configErrs, "global.server_name", string(c.ServerName))
|
||||
checkNotEmpty(configErrs, "global.private_key", string(c.PrivateKeyPath))
|
||||
|
||||
c.Kafka.Verify(configErrs, isMonolith)
|
||||
c.JetStream.Verify(configErrs, isMonolith)
|
||||
c.Metrics.Verify(configErrs, isMonolith)
|
||||
c.Sentry.Verify(configErrs, isMonolith)
|
||||
c.DNSCache.Verify(configErrs, isMonolith)
|
||||
|
|
|
|||
31
setup/config/config_jetstream.go
Normal file
31
setup/config/config_jetstream.go
Normal file
|
|
@ -0,0 +1,31 @@
|
|||
package config
|
||||
|
||||
import "fmt"
|
||||
|
||||
type JetStream struct {
|
||||
Matrix *Global `yaml:"-"`
|
||||
|
||||
// A list of NATS addresses to connect to. If none are specified, an
|
||||
// internal NATS server will be used when running in monolith mode only.
|
||||
Addresses []string `yaml:"addresses"`
|
||||
// The prefix to use for stream names for this homeserver - really only
|
||||
// useful if running more than one Dendrite on the same NATS deployment.
|
||||
TopicPrefix string `yaml:"topic_prefix"`
|
||||
}
|
||||
|
||||
func (k *JetStream) TopicFor(name string) string {
|
||||
return fmt.Sprintf("%s%s", k.TopicPrefix, name)
|
||||
}
|
||||
|
||||
func (c *JetStream) Defaults() {
|
||||
c.Addresses = []string{}
|
||||
c.TopicPrefix = "Dendrite"
|
||||
}
|
||||
|
||||
func (c *JetStream) Verify(configErrs *ConfigErrors, isMonolith bool) {
|
||||
// If we are running in a polylith deployment then we need at least
|
||||
// one NATS JetStream server to talk to.
|
||||
if !isMonolith {
|
||||
checkNotZero(configErrs, "global.jetstream.addresses", int64(len(c.Addresses)))
|
||||
}
|
||||
}
|
||||
|
|
@ -1,75 +0,0 @@
|
|||
package config
|
||||
|
||||
import "fmt"
|
||||
|
||||
// Defined Kafka topics.
|
||||
const (
|
||||
TopicOutputTypingEvent = "OutputTypingEvent"
|
||||
TopicOutputSendToDeviceEvent = "OutputSendToDeviceEvent"
|
||||
TopicOutputKeyChangeEvent = "OutputKeyChangeEvent"
|
||||
TopicOutputRoomEvent = "OutputRoomEvent"
|
||||
TopicOutputClientData = "OutputClientData"
|
||||
TopicOutputReceiptEvent = "OutputReceiptEvent"
|
||||
)
|
||||
|
||||
// KafkaTopics is a convenience slice to access all defined Kafka topics.
|
||||
var KafkaTopics = []string{
|
||||
TopicOutputTypingEvent,
|
||||
TopicOutputSendToDeviceEvent,
|
||||
TopicOutputKeyChangeEvent,
|
||||
TopicOutputRoomEvent,
|
||||
TopicOutputClientData,
|
||||
TopicOutputReceiptEvent,
|
||||
}
|
||||
|
||||
type Kafka struct {
|
||||
Matrix *Global `yaml:"-"`
|
||||
|
||||
// A list of kafka/NATS addresses to connect to.
|
||||
Addresses []string `yaml:"addresses"`
|
||||
// The prefix to use for Kafka topic names for this homeserver - really only
|
||||
// useful if running more than one Dendrite on the same Kafka deployment.
|
||||
TopicPrefix string `yaml:"topic_prefix"`
|
||||
// Whether to use naffka instead of kafka.
|
||||
// Naffka can only be used when running dendrite as a single monolithic server.
|
||||
// Kafka can be used both with a monolithic server and when running the
|
||||
// components as separate servers.
|
||||
UseNaffka bool `yaml:"use_naffka"`
|
||||
// The Naffka database is used internally by the naffka library, if used.
|
||||
Database DatabaseOptions `yaml:"naffka_database"`
|
||||
// The max size a Kafka message passed between consumer/producer can have
|
||||
// Equals roughly max.message.bytes / fetch.message.max.bytes in Kafka
|
||||
MaxMessageBytes *int `yaml:"max_message_bytes"`
|
||||
// Whether to use NATS instead of kafka/naffka
|
||||
UseNATS bool `yaml:"use_nats"`
|
||||
}
|
||||
|
||||
func (k *Kafka) TopicFor(name string) string {
|
||||
return fmt.Sprintf("%s%s", k.TopicPrefix, name)
|
||||
}
|
||||
|
||||
func (c *Kafka) Defaults() {
|
||||
c.UseNaffka = true
|
||||
c.Database.Defaults(10)
|
||||
c.Addresses = []string{"localhost:2181"}
|
||||
c.Database.ConnectionString = DataSource("file:naffka.db")
|
||||
c.TopicPrefix = "Dendrite"
|
||||
|
||||
maxBytes := 1024 * 1024 * 8 // about 8MB
|
||||
c.MaxMessageBytes = &maxBytes
|
||||
}
|
||||
|
||||
func (c *Kafka) Verify(configErrs *ConfigErrors, isMonolith bool) {
|
||||
if c.UseNaffka {
|
||||
if !isMonolith {
|
||||
configErrs.Add("naffka can only be used in a monolithic server")
|
||||
}
|
||||
checkNotEmpty(configErrs, "global.kafka.database.connection_string", string(c.Database.ConnectionString))
|
||||
} else {
|
||||
// If we aren't using naffka then we need to have at least one kafka/nats
|
||||
// server to talk to.
|
||||
checkNotZero(configErrs, "global.kafka.addresses", int64(len(c.Addresses)))
|
||||
}
|
||||
checkNotEmpty(configErrs, "global.kafka.topic_prefix", string(c.TopicPrefix))
|
||||
checkPositive(configErrs, "global.kafka.max_message_bytes", int64(*c.MaxMessageBytes))
|
||||
}
|
||||
88
setup/jetstream/nats.go
Normal file
88
setup/jetstream/nats.go
Normal file
|
|
@ -0,0 +1,88 @@
|
|||
package jetstream
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/Shopify/sarama"
|
||||
"github.com/matrix-org/dendrite/setup/config"
|
||||
"github.com/sirupsen/logrus"
|
||||
|
||||
saramajs "github.com/S7evinK/saramajetstream"
|
||||
natsserver "github.com/nats-io/nats-server/v2/server"
|
||||
"github.com/nats-io/nats.go"
|
||||
natsclient "github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
var natsServer *natsserver.Server
|
||||
var natsServerMutex sync.Mutex
|
||||
|
||||
func SetupConsumerProducer(cfg *config.JetStream) (sarama.Consumer, sarama.SyncProducer) {
|
||||
natsServerMutex.Lock()
|
||||
s := natsServer
|
||||
if s == nil {
|
||||
var err error
|
||||
natsServer, err = natsserver.NewServer(&natsserver.Options{
|
||||
ServerName: "monolith",
|
||||
DontListen: true,
|
||||
JetStream: true,
|
||||
StoreDir: string(cfg.Matrix.ServerName),
|
||||
})
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
natsServer.ConfigureLogger()
|
||||
go natsServer.Start()
|
||||
s = natsServer
|
||||
}
|
||||
natsServerMutex.Unlock()
|
||||
if !natsServer.ReadyForConnections(time.Second * 10) {
|
||||
logrus.Fatalln("NATS did not start in time")
|
||||
}
|
||||
conn, err := s.InProcessConn()
|
||||
if err != nil {
|
||||
logrus.Fatalln("Failed to get a NATS in-process conn")
|
||||
}
|
||||
nc, err := natsclient.Connect("", natsclient.InProcessConn(conn))
|
||||
if err != nil {
|
||||
logrus.Fatalln("Failed to create NATS client")
|
||||
}
|
||||
return setupNATS(cfg, nc)
|
||||
}
|
||||
|
||||
func setupNATS(cfg *config.JetStream, nc *natsclient.Conn) (sarama.Consumer, sarama.SyncProducer) {
|
||||
if nc == nil {
|
||||
var err error
|
||||
nc, err = nats.Connect(strings.Join(cfg.Addresses, ","))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("Unable to connect to NATS")
|
||||
return nil, nil
|
||||
}
|
||||
}
|
||||
|
||||
s, err := nc.JetStream()
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("Unable to get JetStream context")
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
for _, stream := range streams {
|
||||
info, err := s.StreamInfo(stream.Name)
|
||||
if err != nil && err != natsclient.ErrStreamNotFound {
|
||||
logrus.WithError(err).Fatal("Unable to get stream info")
|
||||
}
|
||||
if info == nil {
|
||||
stream.Name = cfg.TopicFor(stream.Name)
|
||||
stream.Subjects = []string{stream.Name}
|
||||
|
||||
if _, err = s.AddStream(stream); err != nil {
|
||||
logrus.WithError(err).WithField("stream", stream.Name).Fatal("Unable to add stream")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
consumer := saramajs.NewJetStreamConsumer(nc, s, "")
|
||||
producer := saramajs.NewJetStreamProducer(nc, s, "")
|
||||
return consumer, producer
|
||||
}
|
||||
50
setup/jetstream/streams.go
Normal file
50
setup/jetstream/streams.go
Normal file
|
|
@ -0,0 +1,50 @@
|
|||
package jetstream
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
var (
|
||||
OutputRoomEvent = "OutputRoomEvent"
|
||||
OutputSendToDeviceEvent = "OutputSendToDeviceEvent"
|
||||
OutputKeyChangeEvent = "OutputKeyChangeEvent"
|
||||
OutputTypingEvent = "OutputTypingEvent"
|
||||
OutputClientData = "OutputClientData"
|
||||
OutputReceiptEvent = "OutputReceiptEvent"
|
||||
)
|
||||
|
||||
var streams = []*nats.StreamConfig{
|
||||
{
|
||||
Name: OutputRoomEvent,
|
||||
Retention: nats.InterestPolicy,
|
||||
Storage: nats.FileStorage,
|
||||
},
|
||||
{
|
||||
Name: OutputSendToDeviceEvent,
|
||||
Retention: nats.InterestPolicy,
|
||||
Storage: nats.FileStorage,
|
||||
},
|
||||
{
|
||||
Name: OutputKeyChangeEvent,
|
||||
Retention: nats.LimitsPolicy,
|
||||
Storage: nats.FileStorage,
|
||||
},
|
||||
{
|
||||
Name: OutputTypingEvent,
|
||||
Retention: nats.InterestPolicy,
|
||||
Storage: nats.MemoryStorage,
|
||||
MaxAge: time.Second * 60,
|
||||
},
|
||||
{
|
||||
Name: OutputClientData,
|
||||
Retention: nats.InterestPolicy,
|
||||
Storage: nats.FileStorage,
|
||||
},
|
||||
{
|
||||
Name: OutputReceiptEvent,
|
||||
Retention: nats.InterestPolicy,
|
||||
Storage: nats.FileStorage,
|
||||
},
|
||||
}
|
||||
|
|
@ -1,163 +0,0 @@
|
|||
package kafka
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
js "github.com/S7evinK/saramajetstream"
|
||||
"github.com/Shopify/sarama"
|
||||
"github.com/matrix-org/dendrite/setup/config"
|
||||
"github.com/matrix-org/naffka"
|
||||
naffkaStorage "github.com/matrix-org/naffka/storage"
|
||||
"github.com/nats-io/nats-server/v2/server"
|
||||
natsserver "github.com/nats-io/nats-server/v2/server"
|
||||
natsclient "github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
var natsServer *natsserver.Server
|
||||
var natsServerMutex sync.Mutex
|
||||
|
||||
func SetupConsumerProducer(cfg *config.Kafka) (sarama.Consumer, sarama.SyncProducer) {
|
||||
/*
|
||||
if cfg.UseNaffka {
|
||||
return setupNaffka(cfg)
|
||||
}
|
||||
*/
|
||||
if true || cfg.UseNATS {
|
||||
if true {
|
||||
natsServerMutex.Lock()
|
||||
s := natsServer
|
||||
if s == nil {
|
||||
logrus.Infof("Starting NATS")
|
||||
var err error
|
||||
natsServer, err = natsserver.NewServer(&server.Options{
|
||||
ServerName: "monolith",
|
||||
DontListen: true,
|
||||
JetStream: true,
|
||||
StoreDir: string(cfg.Matrix.ServerName),
|
||||
LogFile: "nats.log",
|
||||
Debug: true,
|
||||
})
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
natsServer.ConfigureLogger()
|
||||
go natsServer.Start()
|
||||
s = natsServer
|
||||
}
|
||||
natsServerMutex.Unlock()
|
||||
if !natsServer.ReadyForConnections(time.Second * 10) {
|
||||
logrus.Fatalln("NATS did not start in time")
|
||||
}
|
||||
conn, err := s.InProcessConn()
|
||||
if err != nil {
|
||||
logrus.Fatalln("Failed to get a NATS in-process conn")
|
||||
}
|
||||
nc, err := natsclient.Connect("", natsclient.InProcessConn(conn))
|
||||
if err != nil {
|
||||
logrus.Fatalln("Failed to create NATS client")
|
||||
}
|
||||
return setupNATS(cfg, nc)
|
||||
} else {
|
||||
return setupNATS(cfg, nil)
|
||||
}
|
||||
}
|
||||
return setupKafka(cfg)
|
||||
}
|
||||
|
||||
// setupKafka creates kafka consumer/producer pair from the config.
|
||||
func setupKafka(cfg *config.Kafka) (sarama.Consumer, sarama.SyncProducer) {
|
||||
sCfg := sarama.NewConfig()
|
||||
sCfg.Producer.MaxMessageBytes = *cfg.MaxMessageBytes
|
||||
sCfg.Producer.Return.Successes = true
|
||||
sCfg.Consumer.Fetch.Default = int32(*cfg.MaxMessageBytes)
|
||||
|
||||
consumer, err := sarama.NewConsumer(cfg.Addresses, sCfg)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("failed to start kafka consumer")
|
||||
}
|
||||
|
||||
producer, err := sarama.NewSyncProducer(cfg.Addresses, sCfg)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("failed to setup kafka producers")
|
||||
}
|
||||
|
||||
return consumer, producer
|
||||
}
|
||||
|
||||
// In monolith mode with Naffka, we don't have the same constraints about
|
||||
// consuming the same topic from more than one place like we do with Kafka.
|
||||
// Therefore, we will only open one Naffka connection in case Naffka is
|
||||
// running on SQLite.
|
||||
var naffkaInstance *naffka.Naffka
|
||||
|
||||
// setupNaffka creates kafka consumer/producer pair from the config.
|
||||
func setupNaffka(cfg *config.Kafka) (sarama.Consumer, sarama.SyncProducer) {
|
||||
if naffkaInstance != nil {
|
||||
return naffkaInstance, naffkaInstance
|
||||
}
|
||||
naffkaDB, err := naffkaStorage.NewDatabase(string(cfg.Database.ConnectionString))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("Failed to setup naffka database")
|
||||
}
|
||||
naffkaInstance, err = naffka.New(naffkaDB)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("Failed to setup naffka")
|
||||
}
|
||||
return naffkaInstance, naffkaInstance
|
||||
}
|
||||
|
||||
func setupNATS(cfg *config.Kafka, nc *natsclient.Conn) (sarama.Consumer, sarama.SyncProducer) {
|
||||
logrus.WithField("servers", cfg.Addresses).Debug("connecting to nats")
|
||||
|
||||
if nc == nil {
|
||||
var err error
|
||||
nc, err = nats.Connect(strings.Join(cfg.Addresses, ","))
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("unable to connect to NATS")
|
||||
return nil, nil
|
||||
}
|
||||
}
|
||||
|
||||
s, err := nc.JetStream()
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("unable to get jetstream context")
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// create a stream for every topic
|
||||
for _, topic := range config.KafkaTopics {
|
||||
sn := cfg.TopicFor(topic)
|
||||
stream, err := s.StreamInfo(sn)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Warn("unable to get stream info")
|
||||
}
|
||||
|
||||
if stream == nil {
|
||||
maxLifeTime := time.Second * 0
|
||||
|
||||
// Typing events can be removed from the stream, as they are only relevant for a short time
|
||||
if topic == config.TopicOutputTypingEvent {
|
||||
maxLifeTime = time.Second * 60
|
||||
}
|
||||
_, err = s.AddStream(&nats.StreamConfig{
|
||||
Name: sn,
|
||||
Subjects: []string{topic},
|
||||
MaxBytes: int64(*cfg.MaxMessageBytes),
|
||||
MaxMsgSize: int32(*cfg.MaxMessageBytes),
|
||||
MaxAge: maxLifeTime,
|
||||
Duplicates: maxLifeTime / 2,
|
||||
})
|
||||
if err != nil {
|
||||
logrus.WithError(err).WithField("stream", sn).Fatal("unable to add nats stream")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
consumer := js.NewJetStreamConsumer(nc, s, cfg.TopicPrefix)
|
||||
producer := js.NewJetStreamProducer(nc, s, cfg.TopicPrefix)
|
||||
return consumer, producer
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue