248 lines
9.5 KiB
Go
248 lines
9.5 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"log"
|
|
"net/http"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"cmpp-platform/gateway/internal/control"
|
|
"cmpp-platform/gateway/internal/health"
|
|
"cmpp-platform/gateway/internal/inbound"
|
|
platformmetrics "cmpp-platform/gateway/internal/metrics"
|
|
"cmpp-platform/gateway/internal/protocollog"
|
|
"cmpp-platform/gateway/internal/queue"
|
|
"cmpp-platform/gateway/internal/ratelimit"
|
|
"cmpp-platform/gateway/internal/resultoutbox"
|
|
"cmpp-platform/gateway/internal/submitworker"
|
|
"cmpp-platform/gateway/internal/upstream"
|
|
|
|
"github.com/redis/go-redis/v9"
|
|
)
|
|
|
|
func main() {
|
|
addr := os.Getenv("GATEWAY_HEALTH_ADDR")
|
|
if addr == "" {
|
|
addr = ":8090"
|
|
}
|
|
cmppAddr := os.Getenv("GATEWAY_CMPP_ADDR")
|
|
if cmppAddr == "" {
|
|
cmppAddr = ":17890"
|
|
}
|
|
apiBaseURL := os.Getenv("API_BASE_URL")
|
|
callbackBaseURL := getenv("GATEWAY_CALLBACK_API_BASE_URL", apiBaseURL)
|
|
gatewayInstanceID := getenv("GATEWAY_INSTANCE_ID", hostname())
|
|
protocolLogPublisher, protocolLogErr := protocollog.New(os.Getenv("REDIS_URL"))
|
|
if protocolLogErr != nil {
|
|
log.Printf("gateway protocol log Redis publisher init failed: %v", protocolLogErr)
|
|
} else {
|
|
protocolLogPublisher.Stream = getenv("GATEWAY_PROTOCOL_LOG_STREAM", "gateway.protocol.logs")
|
|
protocolLogPublisher.GatewayInstanceID = gatewayInstanceID
|
|
protocolLogPublisher.SuccessSampleRate = positiveEnvInt("GATEWAY_PROTOCOL_LOG_SUCCESS_SAMPLE_PERCENT", 10)
|
|
protocolLogPublisher.MaxLen = int64(positiveEnvInt("GATEWAY_PROTOCOL_LOG_STREAM_MAX_LEN", 200000))
|
|
}
|
|
upstreamManager := &upstream.Manager{DrainageGuardEnabled: true, APIBaseURL: apiBaseURL, EventAPIBaseURL: callbackBaseURL, ProtocolLogPublisher: protocolLogPublisher, GatewayInstanceID: gatewayInstanceID}
|
|
var worker *submitworker.Worker
|
|
var resultOutbox *resultoutbox.Outbox
|
|
channelLimiter, err := ratelimit.New(os.Getenv("REDIS_URL"))
|
|
if err != nil {
|
|
log.Fatalf("gateway channel rate limiter init failed: %v", err)
|
|
}
|
|
presenceStore, err := inbound.NewRedisPresenceStore(os.Getenv("REDIS_URL"))
|
|
if err != nil {
|
|
log.Printf("gateway downstream presence store init failed: %v", err)
|
|
}
|
|
recoveryStore, err := inbound.NewRedisRecoveryStore(os.Getenv("REDIS_URL"))
|
|
if err != nil {
|
|
log.Printf("gateway downstream recovery store init failed: %v", err)
|
|
}
|
|
|
|
go func() {
|
|
log.Printf("cmpp gateway inbound server listening on %s", cmppAddr)
|
|
inboundConcurrency := positiveEnvInt("GATEWAY_CMPP_INBOUND_MAX_CONCURRENCY", 64)
|
|
if err := (inbound.Server{
|
|
Addr: cmppAddr,
|
|
APIBaseURL: apiBaseURL,
|
|
HTTPClient: inbound.NewAPIHTTPClient(16),
|
|
SubmitHTTPClient: inbound.NewAPIHTTPClient(inboundConcurrency),
|
|
PresenceStore: presenceStore,
|
|
RecoveryStore: recoveryStore,
|
|
GatewayInstanceID: gatewayInstanceID,
|
|
MaxSubmitConcurrency: inboundConcurrency,
|
|
SecurityEventToken: os.Getenv("SECURITY_EVENT_TOKEN"),
|
|
ProtocolLogPublisher: protocolLogPublisher,
|
|
}).ListenAndServe(); err != nil {
|
|
log.Fatalf("gateway inbound server stopped: %v", err)
|
|
}
|
|
}()
|
|
|
|
if os.Getenv("GATEWAY_SUBMIT_WORKER_DISABLED") != "true" {
|
|
worker, err = submitworker.New(os.Getenv("REDIS_URL"), upstreamManager)
|
|
if err != nil {
|
|
log.Printf("gateway submit worker init failed: %v", err)
|
|
} else {
|
|
worker.Stream = getenv("GATEWAY_SUBMIT_STREAM", "gateway.submit.commands")
|
|
worker.Group = getenv("GATEWAY_SUBMIT_GROUP", "cmpp-gateway")
|
|
worker.Consumer = getenv("GATEWAY_SUBMIT_CONSUMER", "gateway-1")
|
|
worker.Concurrency = positiveEnvInt("GATEWAY_SUBMIT_WORKER_CONCURRENCY", 64)
|
|
worker.APIBaseURL = callbackBaseURL
|
|
resultOutbox = resultoutbox.New(worker.Redis)
|
|
resultOutbox.Stream = getenv("GATEWAY_SUBMIT_RESULT_STREAM", "gateway.submit.results")
|
|
resultOutbox.Group = getenv("GATEWAY_SUBMIT_RESULT_GROUP", "cmpp-api-callback")
|
|
resultOutbox.Consumer = getenv("GATEWAY_SUBMIT_RESULT_CONSUMER", "gateway-1")
|
|
resultOutbox.APIBaseURL = callbackBaseURL
|
|
resultOutbox.Concurrency = positiveEnvInt("GATEWAY_SUBMIT_RESULT_WORKER_CONCURRENCY", 8)
|
|
resultOutbox.BatchEnabled = enabledEnv("GATEWAY_CALLBACK_BATCH_ENABLED")
|
|
resultOutbox.BatchSize = positiveEnvInt("GATEWAY_CALLBACK_BATCH_SIZE", 50)
|
|
resultOutbox.BatchWait = time.Duration(positiveEnvInt("GATEWAY_CALLBACK_BATCH_WAIT_MS", 10)) * time.Millisecond
|
|
resultOutbox.GatewayInstanceID = gatewayInstanceID
|
|
resultOutbox.DeadLetterStream = getenv("GATEWAY_CALLBACK_DEAD_LETTER_STREAM", "gateway.submit.results.dead")
|
|
worker.ResultOutbox = resultOutbox
|
|
upstreamManager.SubmitSegmentPublisher = resultOutbox
|
|
upstreamManager.EventPublisher = resultOutbox
|
|
go func() {
|
|
log.Printf("cmpp gateway submit worker consuming stream=%s group=%s consumer=%s", worker.Stream, worker.Group, worker.Consumer)
|
|
if err := worker.Run(context.Background()); err != nil {
|
|
log.Printf("gateway submit worker stopped: %v", err)
|
|
}
|
|
}()
|
|
go func() {
|
|
log.Printf("cmpp gateway result Outbox consuming stream=%s group=%s consumer=%s", resultOutbox.StreamName(), resultOutbox.GroupName(), resultOutbox.Consumer)
|
|
if err := resultOutbox.Run(context.Background()); err != nil {
|
|
log.Printf("gateway result Outbox worker stopped: %v", err)
|
|
}
|
|
}()
|
|
}
|
|
}
|
|
|
|
mux := http.NewServeMux()
|
|
mux.Handle("/health", health.Handler())
|
|
mux.Handle("/metrics", platformmetrics.Handler(func(ctx context.Context) platformmetrics.Snapshot {
|
|
desired, connected := upstreamManager.ConnectionCounts()
|
|
snapshot := platformmetrics.Snapshot{
|
|
UpstreamDesired: desired, UpstreamConnected: connected,
|
|
DownstreamConnected: inbound.ActiveConnectionCount(), SubmitWorkerUp: worker != nil,
|
|
}
|
|
snapshot.UpstreamWindowConfigured, snapshot.UpstreamWindowInFlight = upstreamManager.WindowCounts()
|
|
if protocolLogPublisher != nil {
|
|
snapshot.ProtocolLogPublished, snapshot.ProtocolLogSampled, snapshot.ProtocolLogErrors = protocolLogPublisher.Counts()
|
|
if length, err := protocolLogPublisher.Redis.XLen(ctx, protocolLogPublisher.StreamName()).Result(); err == nil {
|
|
snapshot.ProtocolLogQueueLength = length
|
|
}
|
|
}
|
|
if worker != nil {
|
|
snapshot.SubmitWorkerConcurrency = worker.ConfiguredConcurrency()
|
|
snapshot.SubmitWorkerInFlight = worker.InFlight()
|
|
}
|
|
if resultOutbox != nil {
|
|
snapshot.ResultWorkerUp = true
|
|
snapshot.ResultWorkerConcurrency = resultOutbox.ConfiguredConcurrency()
|
|
snapshot.ResultWorkerInFlight = resultOutbox.InFlight()
|
|
snapshot.CallbackBatchRequests, snapshot.CallbackBatchEvents, snapshot.CallbackBatchRetries, snapshot.CallbackDeadLetters = resultOutbox.BatchCounts()
|
|
}
|
|
snapshot.InboundSubmitConcurrency, snapshot.InboundSubmitInFlight = inbound.SubmitSlotSnapshot()
|
|
if worker == nil || worker.Redis == nil {
|
|
return snapshot
|
|
}
|
|
pending, err := worker.Redis.XPending(ctx, worker.Stream, worker.Group).Result()
|
|
if err != nil {
|
|
return snapshot
|
|
}
|
|
snapshot.QueueAvailable = true
|
|
snapshot.QueuePending = pending.Count
|
|
groups, err := worker.Redis.XInfoGroups(ctx, worker.Stream).Result()
|
|
if err == nil {
|
|
for _, group := range groups {
|
|
if group.Name == worker.Group {
|
|
snapshot.QueueLag = group.Lag
|
|
break
|
|
}
|
|
}
|
|
}
|
|
entries, err := worker.Redis.XPendingExt(ctx, &redis.XPendingExtArgs{Stream: worker.Stream, Group: worker.Group, Start: "-", End: "+", Count: 1}).Result()
|
|
if err == nil && len(entries) > 0 {
|
|
milliseconds, parseErr := strconv.ParseInt(strings.SplitN(entries[0].ID, "-", 2)[0], 10, 64)
|
|
if parseErr == nil {
|
|
snapshot.QueueOldestAgeSeconds = max(0, time.Since(time.UnixMilli(milliseconds)).Seconds())
|
|
}
|
|
}
|
|
if resultOutbox != nil {
|
|
resultPending, resultErr := worker.Redis.XPending(ctx, resultOutbox.StreamName(), resultOutbox.GroupName()).Result()
|
|
if resultErr == nil {
|
|
snapshot.ResultQueueAvailable = true
|
|
snapshot.ResultQueuePending = resultPending.Count
|
|
}
|
|
resultGroups, resultErr := worker.Redis.XInfoGroups(ctx, resultOutbox.StreamName()).Result()
|
|
if resultErr == nil {
|
|
for _, group := range resultGroups {
|
|
if group.Name == resultOutbox.GroupName() {
|
|
snapshot.ResultQueueLag = group.Lag
|
|
break
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return snapshot
|
|
}))
|
|
control.Register(mux, control.Server{
|
|
APIBaseURL: callbackBaseURL,
|
|
Upstream: upstreamManager,
|
|
Limiter: channelLimiter,
|
|
Submit: func(ctx context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) {
|
|
result, submitErr := upstreamManager.Submit(ctx, command)
|
|
if resultOutbox == nil {
|
|
return result, submitErr
|
|
}
|
|
if publishErr := resultOutbox.PublishSubmitResult(ctx, command, result); publishErr != nil {
|
|
return result, publishErr
|
|
}
|
|
return result, submitErr
|
|
},
|
|
RecoveryCandidates: func(ctx context.Context) ([]inbound.DownstreamPresence, error) {
|
|
return inbound.ListRecoveryCandidates(ctx, presenceStore)
|
|
},
|
|
RecoveryStatuses: func(ctx context.Context) ([]inbound.DownstreamRecoveryStatus, error) {
|
|
if recoveryStore == nil {
|
|
return nil, nil
|
|
}
|
|
return recoveryStore.ListRecoveryStatuses(ctx)
|
|
},
|
|
})
|
|
|
|
log.Printf("cmpp gateway control server listening on %s", addr)
|
|
if err := http.ListenAndServe(addr, mux); err != nil {
|
|
log.Fatalf("gateway control server stopped: %v", err)
|
|
}
|
|
}
|
|
|
|
func getenv(key string, fallback string) string {
|
|
value := os.Getenv(key)
|
|
if value == "" {
|
|
return fallback
|
|
}
|
|
return value
|
|
}
|
|
|
|
func positiveEnvInt(key string, fallback int) int {
|
|
value, err := strconv.Atoi(os.Getenv(key))
|
|
if err != nil || value <= 0 {
|
|
return fallback
|
|
}
|
|
return value
|
|
}
|
|
|
|
func enabledEnv(key string) bool {
|
|
return os.Getenv(key) == "true"
|
|
}
|
|
|
|
func hostname() string {
|
|
name, err := os.Hostname()
|
|
if err != nil || name == "" {
|
|
return "gateway-1"
|
|
}
|
|
return name
|
|
}
|