357 lines
10 KiB
Go
357 lines
10 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"crypto/tls"
|
|
"encoding/json"
|
|
"errors"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/segmentio/kafka-go"
|
|
"github.com/segmentio/kafka-go/sasl"
|
|
"github.com/segmentio/kafka-go/sasl/scram"
|
|
"github.com/tinode/chat/server/logs"
|
|
"github.com/tinode/chat/server/store/types"
|
|
)
|
|
|
|
const (
|
|
kafkaTopicAdminCommand = "im.admin.command"
|
|
kafkaTopicAdminCommandResult = "im.admin.command.result"
|
|
kafkaTopicConfigEvent = "im.config.event"
|
|
kafkaTopicBusinessEvent = "im.business.event"
|
|
kafkaTopicMessagePersist = "im.message.persist"
|
|
kafkaTopicGroupMessagePersist = "im.group.message.persist"
|
|
kafkaTopicMessagePush = "im.message.push"
|
|
kafkaTopicWorkbenchExternal = "im.workbench.event"
|
|
kafkaTopicFileSecurity = "im.file.security"
|
|
kafkaTopicCacheInvalidate = "im.cache.invalidate"
|
|
kafkaTopicAuditEvent = "im.audit.event"
|
|
kafkaTopicDeadLetter = "im.dead.letter"
|
|
)
|
|
|
|
type kafkaConfig struct {
|
|
Enabled bool `json:"enabled"`
|
|
Brokers []string `json:"brokers"`
|
|
ClientID string `json:"client_id"`
|
|
ConsumerGroup string `json:"consumer_group"`
|
|
Topics map[string]string `json:"topics"`
|
|
DialTimeoutMS int `json:"dial_timeout_ms"`
|
|
WriteTimeoutMS int `json:"write_timeout_ms"`
|
|
ReadTimeoutMS int `json:"read_timeout_ms"`
|
|
BatchTimeoutMS int `json:"batch_timeout_ms"`
|
|
RequiredAcks int `json:"required_acks"`
|
|
AllowAutoTopic bool `json:"allow_auto_topic_creation"`
|
|
TLSEnabled bool `json:"tls_enabled"`
|
|
SASLEnabled bool `json:"sasl_enabled"`
|
|
SASLMechanism string `json:"sasl_mechanism"`
|
|
SASLUsername string `json:"sasl_username"`
|
|
SASLPassword string `json:"sasl_password"`
|
|
MaxRetryCount int `json:"max_retry_count"`
|
|
RetryBackoffMS []int `json:"retry_backoff_ms"`
|
|
DeadRetentionSec int `json:"dead_letter_retention_seconds"`
|
|
}
|
|
|
|
type kafkaEnvelope struct {
|
|
EventID string `json:"event_id"`
|
|
TraceID string `json:"trace_id,omitempty"`
|
|
TenantCode string `json:"tenant_code,omitempty"`
|
|
TenantID types.TenantID `json:"tenant_id,omitempty"`
|
|
SourceService string `json:"source_service,omitempty"`
|
|
TargetService string `json:"target_service,omitempty"`
|
|
EventType string `json:"event_type"`
|
|
BizType string `json:"biz_type,omitempty"`
|
|
BizID string `json:"biz_id,omitempty"`
|
|
IdempotencyKey string `json:"idempotency_key,omitempty"`
|
|
Version int64 `json:"version,omitempty"`
|
|
OccurredAt time.Time `json:"occurred_at"`
|
|
Actor *kafkaActor `json:"actor,omitempty"`
|
|
Payload json.RawMessage `json:"payload,omitempty"`
|
|
ParentEventID string `json:"parent_event_id,omitempty"`
|
|
}
|
|
|
|
type kafkaActor struct {
|
|
Type string `json:"type,omitempty"`
|
|
ID string `json:"id,omitempty"`
|
|
DisplayName string `json:"display_name,omitempty"`
|
|
}
|
|
|
|
type kafkaRuntime struct {
|
|
config kafkaConfig
|
|
topics map[string]string
|
|
|
|
writer *kafka.Writer
|
|
ctx context.Context
|
|
cancel context.CancelFunc
|
|
wg sync.WaitGroup
|
|
}
|
|
|
|
var kafkaBus *kafkaRuntime
|
|
|
|
func initKafkaRuntime(raw json.RawMessage) error {
|
|
if len(raw) == 0 {
|
|
kafkaBus = nil
|
|
logs.Info.Println("Kafka disabled: no config")
|
|
return nil
|
|
}
|
|
var cfg kafkaConfig
|
|
if err := json.Unmarshal(raw, &cfg); err != nil {
|
|
return err
|
|
}
|
|
if !cfg.Enabled {
|
|
kafkaBus = nil
|
|
logs.Info.Println("Kafka disabled")
|
|
return nil
|
|
}
|
|
normalizeKafkaConfig(&cfg)
|
|
if len(cfg.Brokers) == 0 {
|
|
return errors.New("kafka: enabled but brokers are not configured")
|
|
}
|
|
saslMechanism, err := kafkaSASLMechanism(cfg)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
rt := &kafkaRuntime{
|
|
config: cfg,
|
|
topics: kafkaTopics(cfg.Topics),
|
|
ctx: ctx,
|
|
cancel: cancel,
|
|
}
|
|
transport := &kafka.Transport{
|
|
ClientID: cfg.ClientID,
|
|
}
|
|
if cfg.TLSEnabled {
|
|
transport.TLS = &tls.Config{MinVersion: tls.VersionTLS12}
|
|
}
|
|
transport.SASL = saslMechanism
|
|
rt.writer = &kafka.Writer{
|
|
Addr: kafka.TCP(cfg.Brokers...),
|
|
Balancer: &kafka.Hash{},
|
|
Transport: transport,
|
|
RequiredAcks: kafka.RequiredAcks(cfg.RequiredAcks),
|
|
AllowAutoTopicCreation: cfg.AllowAutoTopic,
|
|
BatchTimeout: time.Duration(cfg.BatchTimeoutMS) * time.Millisecond,
|
|
WriteTimeout: time.Duration(cfg.WriteTimeoutMS) * time.Millisecond,
|
|
ReadTimeout: time.Duration(cfg.ReadTimeoutMS) * time.Millisecond,
|
|
MaxAttempts: cfg.MaxRetryCount,
|
|
}
|
|
|
|
if err := rt.ping(context.Background()); err != nil {
|
|
cancel()
|
|
_ = rt.writer.Close()
|
|
return err
|
|
}
|
|
kafkaBus = rt
|
|
logs.Info.Printf("Kafka enabled: brokers=%s client_id=%s group=%s", strings.Join(cfg.Brokers, ","), cfg.ClientID, cfg.ConsumerGroup)
|
|
return nil
|
|
}
|
|
|
|
func normalizeKafkaConfig(cfg *kafkaConfig) {
|
|
if cfg.ClientID == "" {
|
|
cfg.ClientID = "im-server"
|
|
}
|
|
if cfg.ConsumerGroup == "" {
|
|
cfg.ConsumerGroup = "im-server"
|
|
}
|
|
if cfg.DialTimeoutMS <= 0 {
|
|
cfg.DialTimeoutMS = 3000
|
|
}
|
|
if cfg.WriteTimeoutMS <= 0 {
|
|
cfg.WriteTimeoutMS = 3000
|
|
}
|
|
if cfg.ReadTimeoutMS <= 0 {
|
|
cfg.ReadTimeoutMS = 3000
|
|
}
|
|
if cfg.BatchTimeoutMS <= 0 {
|
|
cfg.BatchTimeoutMS = 10
|
|
}
|
|
if cfg.RequiredAcks == 0 {
|
|
cfg.RequiredAcks = int(kafka.RequireOne)
|
|
}
|
|
if cfg.MaxRetryCount <= 0 {
|
|
cfg.MaxRetryCount = 3
|
|
}
|
|
if len(cfg.RetryBackoffMS) == 0 {
|
|
cfg.RetryBackoffMS = []int{10000, 30000, 60000, 300000, 900000}
|
|
}
|
|
if cfg.DeadRetentionSec <= 0 {
|
|
cfg.DeadRetentionSec = 90 * 24 * 3600
|
|
}
|
|
if cfg.SASLMechanism == "" {
|
|
cfg.SASLMechanism = "SCRAM-SHA-512"
|
|
}
|
|
}
|
|
|
|
func kafkaSASLMechanism(cfg kafkaConfig) (sasl.Mechanism, error) {
|
|
if !cfg.SASLEnabled && cfg.SASLUsername == "" {
|
|
return nil, nil
|
|
}
|
|
if cfg.SASLUsername == "" || cfg.SASLPassword == "" {
|
|
return nil, errors.New("kafka: sasl enabled but username or password is empty")
|
|
}
|
|
switch strings.ToUpper(strings.ReplaceAll(cfg.SASLMechanism, "_", "-")) {
|
|
case "", "SCRAM-SHA-512":
|
|
return scram.Mechanism(scram.SHA512, cfg.SASLUsername, cfg.SASLPassword)
|
|
case "SCRAM-SHA-256":
|
|
return scram.Mechanism(scram.SHA256, cfg.SASLUsername, cfg.SASLPassword)
|
|
default:
|
|
return nil, errors.New("kafka: unsupported sasl mechanism " + cfg.SASLMechanism)
|
|
}
|
|
}
|
|
|
|
func kafkaTopics(overrides map[string]string) map[string]string {
|
|
topics := map[string]string{
|
|
"admin_command": kafkaTopicAdminCommand,
|
|
"admin_command_result": kafkaTopicAdminCommandResult,
|
|
"config_event": kafkaTopicConfigEvent,
|
|
"business_event": kafkaTopicBusinessEvent,
|
|
"message_persist": kafkaTopicMessagePersist,
|
|
"group_message_persist": kafkaTopicGroupMessagePersist,
|
|
"message_push": kafkaTopicMessagePush,
|
|
"workbench_external": kafkaTopicWorkbenchExternal,
|
|
"file_security": kafkaTopicFileSecurity,
|
|
"cache_invalidate": kafkaTopicCacheInvalidate,
|
|
"audit_event": kafkaTopicAuditEvent,
|
|
"dead_letter": kafkaTopicDeadLetter,
|
|
}
|
|
for key, value := range overrides {
|
|
if value != "" {
|
|
topics[key] = value
|
|
}
|
|
}
|
|
return topics
|
|
}
|
|
|
|
func (rt *kafkaRuntime) ping(ctx context.Context) error {
|
|
timeout := time.Duration(rt.config.DialTimeoutMS) * time.Millisecond
|
|
dialer := rt.dialer()
|
|
pingCtx, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
conn, err := dialer.DialContext(pingCtx, "tcp", rt.config.Brokers[0])
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return conn.Close()
|
|
}
|
|
|
|
func kafkaEnabled() bool {
|
|
return kafkaBus != nil
|
|
}
|
|
|
|
func kafkaTopic(name string) string {
|
|
if kafkaBus == nil {
|
|
return ""
|
|
}
|
|
return kafkaBus.topics[name]
|
|
}
|
|
|
|
func publishKafkaEnvelope(ctx context.Context, topicName string, key string, envelope kafkaEnvelope) error {
|
|
if kafkaBus == nil {
|
|
return nil
|
|
}
|
|
if envelope.OccurredAt.IsZero() {
|
|
envelope.OccurredAt = time.Now().UTC()
|
|
}
|
|
topic := kafkaBus.topics[topicName]
|
|
if topic == "" {
|
|
topic = topicName
|
|
}
|
|
value, err := json.Marshal(envelope)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return kafkaBus.writer.WriteMessages(ctx, kafka.Message{
|
|
Topic: topic,
|
|
Key: []byte(key),
|
|
Value: value,
|
|
Time: envelope.OccurredAt,
|
|
})
|
|
}
|
|
|
|
type kafkaMessageHandler func(context.Context, kafkaEnvelope, kafka.Message) error
|
|
|
|
func startKafkaConsumer(topicName string, handler kafkaMessageHandler) {
|
|
if kafkaBus == nil || handler == nil {
|
|
return
|
|
}
|
|
topic := kafkaBus.topics[topicName]
|
|
if topic == "" {
|
|
topic = topicName
|
|
}
|
|
groupID := kafkaBus.config.ConsumerGroup
|
|
if topicName != "" {
|
|
groupID += "." + topicName
|
|
}
|
|
reader := kafka.NewReader(kafka.ReaderConfig{
|
|
Brokers: kafkaBus.config.Brokers,
|
|
GroupID: groupID,
|
|
Topic: topic,
|
|
StartOffset: kafka.FirstOffset,
|
|
MinBytes: 1,
|
|
MaxBytes: 10e6,
|
|
MaxWait: time.Second,
|
|
Dialer: kafkaBus.dialer(),
|
|
})
|
|
kafkaBus.wg.Add(1)
|
|
go func() {
|
|
defer kafkaBus.wg.Done()
|
|
defer reader.Close()
|
|
for {
|
|
msg, err := reader.FetchMessage(kafkaBus.ctx)
|
|
if err != nil {
|
|
if errors.Is(err, context.Canceled) || errors.Is(kafkaBus.ctx.Err(), context.Canceled) {
|
|
return
|
|
}
|
|
logs.Err.Println("Kafka fetch failed:", err)
|
|
continue
|
|
}
|
|
var envelope kafkaEnvelope
|
|
if err = json.Unmarshal(msg.Value, &envelope); err != nil {
|
|
logs.Err.Println("Kafka envelope decode failed:", err)
|
|
_ = reader.CommitMessages(context.Background(), msg)
|
|
continue
|
|
}
|
|
if err = handler(kafkaBus.ctx, envelope, msg); err != nil {
|
|
logs.Err.Println("Kafka handler failed:", envelope.EventID, envelope.EventType, err)
|
|
continue
|
|
}
|
|
if err = reader.CommitMessages(context.Background(), msg); err != nil {
|
|
logs.Err.Println("Kafka commit failed:", err)
|
|
}
|
|
}
|
|
}()
|
|
logs.Info.Printf("Kafka consumer started: topic=%s group=%s", topic, groupID)
|
|
}
|
|
|
|
func (rt *kafkaRuntime) dialer() *kafka.Dialer {
|
|
dialer := &kafka.Dialer{
|
|
ClientID: rt.config.ClientID,
|
|
Timeout: time.Duration(rt.config.DialTimeoutMS) * time.Millisecond,
|
|
}
|
|
if rt.config.TLSEnabled {
|
|
dialer.TLS = &tls.Config{MinVersion: tls.VersionTLS12}
|
|
}
|
|
if mechanism, err := kafkaSASLMechanism(rt.config); err != nil {
|
|
logs.Err.Println("Kafka SASL config failed:", err)
|
|
} else {
|
|
dialer.SASLMechanism = mechanism
|
|
}
|
|
return dialer
|
|
}
|
|
|
|
func stopKafkaRuntime() {
|
|
if kafkaBus == nil {
|
|
return
|
|
}
|
|
kafkaBus.cancel()
|
|
kafkaBus.wg.Wait()
|
|
if kafkaBus.writer != nil {
|
|
_ = kafkaBus.writer.Close()
|
|
}
|
|
logs.Info.Println("Stopped Kafka runtime")
|
|
kafkaBus = nil
|
|
}
|