288 lines
9.5 KiB
Go
288 lines
9.5 KiB
Go
package upstream
|
|
|
|
import (
|
|
"cmpp-platform/gateway/internal/protocollog"
|
|
"cmpp-platform/gateway/internal/queue"
|
|
"context"
|
|
"fmt"
|
|
"net/http"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
const (
|
|
defaultConnectTimeout = 5 * time.Second
|
|
defaultSubmitTimeout = 10 * time.Second
|
|
defaultHTTPTimeout = 10 * time.Second
|
|
defaultWindowSize = 16
|
|
defaultHeartbeatInterval = 30 * time.Second
|
|
defaultHeartbeatMissThreshold = 3
|
|
defaultReconnectInitialDelay = 5 * time.Second
|
|
defaultReconnectMaximumDelay = 5 * time.Minute
|
|
defaultAuthReconnectDelay = 5 * time.Minute
|
|
maximumConnections = 8
|
|
maximumWindowSize = 64
|
|
)
|
|
|
|
type Manager struct {
|
|
APIBaseURL string
|
|
EventAPIBaseURL string
|
|
HTTPClient *http.Client
|
|
SubmitSegmentPublisher SubmitSegmentPublisher
|
|
EventPublisher interface {
|
|
PublishReceipt(context.Context, queue.ReceiptEvent) error
|
|
PublishUplink(context.Context, queue.UplinkEvent) error
|
|
}
|
|
ProtocolLogPublisher interface {
|
|
Publish(context.Context, protocollog.Event) error
|
|
}
|
|
GatewayInstanceID string
|
|
|
|
mu sync.Mutex
|
|
conns map[string]*connectionPool
|
|
}
|
|
|
|
// SubmitSegmentPublisher persists each supplier response before the next long-message
|
|
// segment is sent. The boundary is intentionally storage-only: HTTP callbacks belong to
|
|
// the result Outbox worker and must not consume a supplier Submit window slot.
|
|
type SubmitSegmentPublisher interface {
|
|
PublishSubmitSegment(context.Context, queue.SubmitCommand, queue.SubmitSegmentResult) error
|
|
}
|
|
|
|
type ConnectionState struct {
|
|
ChannelID string `json:"channelId"`
|
|
ConnectionID string `json:"connectionId"`
|
|
Status string `json:"status"`
|
|
DesiredConnections int `json:"desiredConnections"`
|
|
CurrentConnections int `json:"currentConnections"`
|
|
LastConnectedAt string `json:"lastConnectedAt,omitempty"`
|
|
LastDisconnectedAt string `json:"lastDisconnectedAt,omitempty"`
|
|
LastHeartbeatAt string `json:"lastHeartbeatAt,omitempty"`
|
|
ReconnectCount int `json:"reconnectCount,omitempty"`
|
|
LastReconnectAttemptAt string `json:"lastReconnectAttemptAt,omitempty"`
|
|
NextReconnectAt string `json:"nextReconnectAt,omitempty"`
|
|
LastErrorCategory string `json:"lastErrorCategory,omitempty"`
|
|
LastError string `json:"lastError,omitempty"`
|
|
}
|
|
|
|
func (m *Manager) ConnectChannel(ctx context.Context, command queue.ConnectChannelCommand) (ConnectionState, error) {
|
|
if err := validateConnectChannelCommand(command); err != nil {
|
|
return ConnectionState{}, err
|
|
}
|
|
|
|
m.mu.Lock()
|
|
m.ensureDefaultsLocked()
|
|
pool := m.conns[command.ChannelID]
|
|
config := normalizeUpstreamConfig(queue.UpstreamConfig{
|
|
GatewayHost: command.Channel.GatewayHost,
|
|
GatewayPort: command.Channel.GatewayPort,
|
|
Account: command.Channel.Account,
|
|
PasswordCipher: command.Channel.PasswordCipher,
|
|
CMPPVersion: command.Channel.CMPPVersion,
|
|
DesiredConnections: command.DesiredConnections,
|
|
WindowSize: command.Channel.WindowSize,
|
|
HeartbeatIntervalSeconds: command.Channel.HeartbeatIntervalSeconds,
|
|
HeartbeatMissThreshold: command.Channel.HeartbeatMissThreshold,
|
|
ConnectionWarmupSeconds: command.Channel.ConnectionWarmupSeconds,
|
|
ConnectionDrainSeconds: command.Channel.ConnectionDrainSeconds,
|
|
SubmitTimeoutSeconds: command.Channel.SubmitTimeoutSeconds,
|
|
FailureCooldownSeconds: command.Channel.FailureCooldownSeconds,
|
|
})
|
|
if pool != nil && !pool.matches(config) && pool.sameEndpoint(config) {
|
|
pool.reconfigure(config)
|
|
} else if pool == nil || !pool.matches(config) {
|
|
if pool != nil {
|
|
pool.close()
|
|
}
|
|
pool = m.newConnectionPool(command.ChannelID, command.ConnectionID, config)
|
|
m.conns[command.ChannelID] = pool
|
|
}
|
|
m.mu.Unlock()
|
|
|
|
pool.startSupervisor()
|
|
if err := pool.ensureConnected(); err != nil {
|
|
if pool.stopped() {
|
|
return pool.snapshotState("disconnected", nil), nil
|
|
}
|
|
pool.scheduleReconnect(err)
|
|
_ = pool.reportState(ctx, "failed", err)
|
|
return pool.snapshotState("failed", err), nil
|
|
}
|
|
if pool.stopped() {
|
|
return pool.snapshotState("disconnected", nil), nil
|
|
}
|
|
pool.resetReconnectState()
|
|
return pool.snapshotState("connected", nil), nil
|
|
}
|
|
|
|
func (m *Manager) DisconnectChannel(ctx context.Context, command queue.DisconnectChannelCommand) (ConnectionState, error) {
|
|
if command.ChannelID == "" || command.ConnectionID == "" {
|
|
return ConnectionState{}, fmt.Errorf("channelId and connectionId are required")
|
|
}
|
|
m.mu.Lock()
|
|
m.ensureDefaultsLocked()
|
|
pool := m.conns[command.ChannelID]
|
|
delete(m.conns, command.ChannelID)
|
|
m.mu.Unlock()
|
|
if pool != nil {
|
|
pool.close()
|
|
state := pool.snapshotState("disconnected", nil)
|
|
_ = pool.reportState(ctx, "disconnected", nil)
|
|
return state, nil
|
|
}
|
|
return ConnectionState{
|
|
ChannelID: command.ChannelID,
|
|
ConnectionID: command.ConnectionID,
|
|
Status: "disconnected",
|
|
DesiredConnections: 0,
|
|
CurrentConnections: 0,
|
|
LastDisconnectedAt: time.Now().UTC().Format(time.RFC3339Nano),
|
|
}, nil
|
|
}
|
|
|
|
func (m *Manager) connectionFor(cmd queue.SubmitCommand) (*connectionPool, error) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
|
|
m.ensureDefaultsLocked()
|
|
|
|
pool := m.conns[cmd.ChannelID]
|
|
if pool != nil && !pool.matches(cmd.Upstream) && pool.sameEndpoint(cmd.Upstream) {
|
|
pool.reconfigure(cmd.Upstream)
|
|
} else if pool == nil || !pool.matches(cmd.Upstream) {
|
|
if pool != nil {
|
|
pool.close()
|
|
}
|
|
pool = m.newConnectionPool(cmd.ChannelID, defaultChannelConnectionID(cmd.ChannelID), normalizeUpstreamConfig(cmd.Upstream))
|
|
m.conns[cmd.ChannelID] = pool
|
|
}
|
|
pool.startSupervisor()
|
|
if err := pool.ensureConnected(); err != nil {
|
|
pool.scheduleReconnect(err)
|
|
return nil, err
|
|
}
|
|
pool.resetReconnectState()
|
|
return pool, nil
|
|
}
|
|
|
|
func (m *Manager) ensureDefaultsLocked() {
|
|
if m.HTTPClient == nil {
|
|
m.HTTPClient = &http.Client{Timeout: defaultHTTPTimeout}
|
|
}
|
|
if m.conns == nil {
|
|
m.conns = make(map[string]*connectionPool)
|
|
}
|
|
}
|
|
|
|
// ConnectionCounts returns bounded platform totals without channel identifiers to prevent time-series cardinality growth.
|
|
func (m *Manager) ConnectionCounts() (desired int, connected int) {
|
|
m.mu.Lock()
|
|
pools := make([]*connectionPool, 0, len(m.conns))
|
|
for _, pool := range m.conns {
|
|
pools = append(pools, pool)
|
|
}
|
|
m.mu.Unlock()
|
|
for _, pool := range pools {
|
|
desired += max(pool.config.DesiredConnections, 1)
|
|
connected += pool.countActiveConnections()
|
|
}
|
|
return desired, connected
|
|
}
|
|
|
|
func (m *Manager) WindowCounts() (configured int, inFlight int) {
|
|
m.mu.Lock()
|
|
pools := make([]*connectionPool, 0, len(m.conns))
|
|
for _, pool := range m.conns {
|
|
pools = append(pools, pool)
|
|
}
|
|
m.mu.Unlock()
|
|
for _, pool := range pools {
|
|
pool.mu.Lock()
|
|
conns := append([]*connection(nil), pool.conns...)
|
|
pool.mu.Unlock()
|
|
for _, conn := range conns {
|
|
conn.mu.Lock()
|
|
configured += max(1, conn.windowLimit)
|
|
inFlight += len(conn.window)
|
|
conn.mu.Unlock()
|
|
}
|
|
}
|
|
return configured, inFlight
|
|
}
|
|
|
|
func (m *Manager) newConnectionPool(channelID string, connectionID string, config queue.UpstreamConfig) *connectionPool {
|
|
eventAPIBaseURL := m.EventAPIBaseURL
|
|
if eventAPIBaseURL == "" {
|
|
eventAPIBaseURL = m.APIBaseURL
|
|
}
|
|
return &connectionPool{
|
|
channelID: channelID,
|
|
connectionID: connectionID,
|
|
config: config,
|
|
apiBaseURL: eventAPIBaseURL,
|
|
httpClient: m.HTTPClient,
|
|
reporter: func(ctx context.Context, state ConnectionState) error {
|
|
return m.post(ctx, "/admin/gateway/connections", state)
|
|
},
|
|
protocolLogPublisher: m.ProtocolLogPublisher,
|
|
gatewayInstanceID: m.GatewayInstanceID,
|
|
eventPublisher: m.EventPublisher,
|
|
reconnectSignal: make(chan struct{}, 1),
|
|
stopCh: make(chan struct{}),
|
|
}
|
|
}
|
|
|
|
func validateConnectChannelCommand(command queue.ConnectChannelCommand) error {
|
|
if command.MessageType != queue.MessageTypeConnectChannel {
|
|
return fmt.Errorf("unsupported messageType %q", command.MessageType)
|
|
}
|
|
if command.ChannelID == "" || command.ConnectionID == "" {
|
|
return fmt.Errorf("channelId and connectionId are required")
|
|
}
|
|
if command.Channel.GatewayHost == "" || command.Channel.GatewayPort <= 0 {
|
|
return fmt.Errorf("gatewayHost and gatewayPort are required")
|
|
}
|
|
if command.Channel.Account == "" || command.Channel.PasswordCipher == "" {
|
|
return fmt.Errorf("account and passwordCipher are required")
|
|
}
|
|
if command.DesiredConnections < 1 || command.DesiredConnections > maximumConnections {
|
|
return fmt.Errorf("desiredConnections must be between 1 and %d", maximumConnections)
|
|
}
|
|
if command.Channel.WindowSize != 0 && (command.Channel.WindowSize < 1 || command.Channel.WindowSize > maximumWindowSize) {
|
|
return fmt.Errorf("windowSize must be between 1 and %d", maximumWindowSize)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func defaultChannelConnectionID(channelID string) string {
|
|
return fmt.Sprintf("%s:primary", channelID)
|
|
}
|
|
|
|
func normalizeUpstreamConfig(config queue.UpstreamConfig) queue.UpstreamConfig {
|
|
if config.DesiredConnections <= 0 {
|
|
config.DesiredConnections = 1
|
|
}
|
|
if config.WindowSize <= 0 {
|
|
config.WindowSize = defaultWindowSize
|
|
}
|
|
if config.HeartbeatIntervalSeconds <= 0 {
|
|
config.HeartbeatIntervalSeconds = int(defaultHeartbeatInterval / time.Second)
|
|
}
|
|
if config.HeartbeatMissThreshold <= 0 {
|
|
config.HeartbeatMissThreshold = defaultHeartbeatMissThreshold
|
|
}
|
|
if config.ConnectionWarmupSeconds < 0 {
|
|
config.ConnectionWarmupSeconds = 30
|
|
}
|
|
if config.ConnectionDrainSeconds <= 0 {
|
|
config.ConnectionDrainSeconds = 60
|
|
}
|
|
if config.SubmitTimeoutSeconds <= 0 {
|
|
config.SubmitTimeoutSeconds = 60
|
|
}
|
|
if config.FailureCooldownSeconds <= 0 {
|
|
config.FailureCooldownSeconds = 30
|
|
}
|
|
return config
|
|
}
|