mirror of
https://github.com/andwati/telecom-platform.git
synced 2026-11-03 10:22:18 +00:00
feat: Add SM-DP+ manager with multi-carrier load balancing, health checking, and profile operations
- Add DefaultManagerConfig with 30s health checks, 3 retries, and circuit breaker configuration - Add CarrierConfig struct with ES2 settings, MCC/MNC, priority, and feature support - Add ToCarrier method converting CarrierConfig to Carrier with metrics and capabilities - Add HealthChecker with periodic carrier health monitoring and status evaluation - Add LoadBalancer with weighted round-robin, least
This commit is contained in:
@@ -0,0 +1,77 @@
|
|||||||
|
package smdp
|
||||||
|
|
||||||
|
import (
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/config"
|
||||||
|
)
|
||||||
|
|
||||||
|
// DefaultManagerConfig returns default configuration for SM-DP+ Manager
|
||||||
|
func DefaultManagerConfig() *ManagerConfig {
|
||||||
|
return &ManagerConfig{
|
||||||
|
HealthCheckInterval: 30 * time.Second,
|
||||||
|
MaxRetries: 3,
|
||||||
|
RetryDelay: 2 * time.Second,
|
||||||
|
CircuitBreakerThreshold: 5,
|
||||||
|
CircuitBreakerTimeout: 60 * time.Second,
|
||||||
|
EnableLoadBalancing: true,
|
||||||
|
EnableFailover: true,
|
||||||
|
DefaultTimeout: 30 * time.Second,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// CarrierConfig represents carrier configuration from database or config file
|
||||||
|
type CarrierConfig struct {
|
||||||
|
ID string `json:"id" yaml:"id"`
|
||||||
|
Name string `json:"name" yaml:"name"`
|
||||||
|
CountryCode string `json:"country_code" yaml:"country_code"`
|
||||||
|
MCC string `json:"mcc" yaml:"mcc"`
|
||||||
|
MNC string `json:"mnc" yaml:"mnc"`
|
||||||
|
ES2BaseURL string `json:"es2_base_url" yaml:"es2_base_url"`
|
||||||
|
ES2APIKey string `json:"es2_api_key" yaml:"es2_api_key"`
|
||||||
|
ES2InsecureSkip bool `json:"es2_insecure_skip" yaml:"es2_insecure_skip"`
|
||||||
|
Priority int `json:"priority" yaml:"priority"`
|
||||||
|
IsActive bool `json:"is_active" yaml:"is_active"`
|
||||||
|
MaxConcurrentReqs int `json:"max_concurrent_requests" yaml:"max_concurrent_requests"`
|
||||||
|
SupportedProfileTypes []string `json:"supported_profile_types" yaml:"supported_profile_types"`
|
||||||
|
SupportedMCCs []string `json:"supported_mccs" yaml:"supported_mccs"`
|
||||||
|
SupportedMNCs []string `json:"supported_mncs" yaml:"supported_mncs"`
|
||||||
|
Features []string `json:"features" yaml:"features"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// ToCarrier converts CarrierConfig to Carrier struct
|
||||||
|
func (c *CarrierConfig) ToCarrier() *Carrier {
|
||||||
|
carrier := &Carrier{
|
||||||
|
ID: c.ID,
|
||||||
|
Name: c.Name,
|
||||||
|
CountryCode: c.CountryCode,
|
||||||
|
MCC: c.MCC,
|
||||||
|
MNC: c.MNC,
|
||||||
|
Priority: c.Priority,
|
||||||
|
IsActive: c.IsActive,
|
||||||
|
HealthStatus: CarrierStatusUnknown,
|
||||||
|
LastHealthCheck: time.Now(),
|
||||||
|
Metrics: &CarrierMetrics{
|
||||||
|
TotalRequests: 0,
|
||||||
|
SuccessfulRequests: 0,
|
||||||
|
FailedRequests: 0,
|
||||||
|
AverageResponseTime: 0,
|
||||||
|
RequestRate: 0,
|
||||||
|
},
|
||||||
|
Capabilities: &CarrierCapabilities{
|
||||||
|
SupportedProfileTypes: c.SupportedProfileTypes,
|
||||||
|
MaxConcurrentRequests: c.MaxConcurrentReqs,
|
||||||
|
SupportedMCCs: c.SupportedMCCs,
|
||||||
|
SupportedMNCs: c.SupportedMNCs,
|
||||||
|
Features: c.Features,
|
||||||
|
},
|
||||||
|
ES2Config: &config.ES2Config{
|
||||||
|
BaseURL: c.ES2BaseURL,
|
||||||
|
APIKey: c.ES2APIKey,
|
||||||
|
InsecureSkipVerify: c.ES2InsecureSkip,
|
||||||
|
FunctionalityRequesterID: "telecom-platform",
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
return carrier
|
||||||
|
}
|
||||||
@@ -0,0 +1,116 @@
|
|||||||
|
package smdp
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/es2"
|
||||||
|
"github.com/sirupsen/logrus"
|
||||||
|
)
|
||||||
|
|
||||||
|
// HealthChecker performs periodic health checks on carriers
|
||||||
|
type HealthChecker struct {
|
||||||
|
interval time.Duration
|
||||||
|
logger *logrus.Logger
|
||||||
|
healthUpdateFn func(carrierID string, status CarrierHealthStatus)
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewHealthChecker creates a new health checker
|
||||||
|
func NewHealthChecker(interval time.Duration) *HealthChecker {
|
||||||
|
return &HealthChecker{
|
||||||
|
interval: interval,
|
||||||
|
logger: logrus.New(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Start begins the health checking process
|
||||||
|
func (h *HealthChecker) Start(ctx context.Context, carriers map[string]*Carrier, clients map[string]*es2.ES2Client, updateFn func(string, CarrierHealthStatus)) {
|
||||||
|
h.healthUpdateFn = updateFn
|
||||||
|
|
||||||
|
ticker := time.NewTicker(h.interval)
|
||||||
|
defer ticker.Stop()
|
||||||
|
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
h.logger.Info("Health checker stopped")
|
||||||
|
return
|
||||||
|
case <-ticker.C:
|
||||||
|
h.checkAllCarriers(carriers, clients)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// checkAllCarriers performs health checks on all carriers
|
||||||
|
func (h *HealthChecker) checkAllCarriers(carriers map[string]*Carrier, clients map[string]*es2.ES2Client) {
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
|
||||||
|
for carrierID, carrier := range carriers {
|
||||||
|
if !carrier.IsActive {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
wg.Add(1)
|
||||||
|
go func(id string, c *Carrier) {
|
||||||
|
defer wg.Done()
|
||||||
|
h.checkCarrier(id, c, clients[id])
|
||||||
|
}(carrierID, carrier)
|
||||||
|
}
|
||||||
|
|
||||||
|
wg.Wait()
|
||||||
|
}
|
||||||
|
|
||||||
|
// checkCarrier performs a health check on a single carrier
|
||||||
|
func (h *HealthChecker) checkCarrier(carrierID string, carrier *Carrier, client *es2.ES2Client) {
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
startTime := time.Now()
|
||||||
|
|
||||||
|
// Perform a simple health check by attempting to get profile status
|
||||||
|
// This is a lightweight operation that tests connectivity
|
||||||
|
req := &es2.GetProfileStatusRequest{
|
||||||
|
EID: "health-check-eid",
|
||||||
|
ICCID: "health-check-iccid",
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err := client.GetProfileStatus(ctx, req)
|
||||||
|
responseTime := time.Since(startTime)
|
||||||
|
|
||||||
|
status := h.evaluateHealth(err, responseTime, carrier.Metrics)
|
||||||
|
h.healthUpdateFn(carrierID, status)
|
||||||
|
|
||||||
|
h.logger.WithFields(logrus.Fields{
|
||||||
|
"carrier_id": carrierID,
|
||||||
|
"status": status,
|
||||||
|
"response_time": responseTime,
|
||||||
|
"error": err,
|
||||||
|
}).Debug("Health check completed")
|
||||||
|
}
|
||||||
|
|
||||||
|
// evaluateHealth evaluates the health status based on error and response time
|
||||||
|
func (h *HealthChecker) evaluateHealth(err error, responseTime time.Duration, metrics *CarrierMetrics) CarrierHealthStatus {
|
||||||
|
if err != nil {
|
||||||
|
// Check if this is the first failure or consecutive failures
|
||||||
|
if metrics.FailedRequests > 5 {
|
||||||
|
return CarrierStatusUnhealthy
|
||||||
|
}
|
||||||
|
return CarrierStatusDegraded
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check response time
|
||||||
|
if responseTime > 5*time.Second {
|
||||||
|
return CarrierStatusDegraded
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check error rate
|
||||||
|
if metrics.TotalRequests > 0 {
|
||||||
|
errorRate := float64(metrics.FailedRequests) / float64(metrics.TotalRequests)
|
||||||
|
if errorRate > 0.1 { // More than 10% error rate
|
||||||
|
return CarrierStatusDegraded
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return CarrierStatusHealthy
|
||||||
|
}
|
||||||
@@ -0,0 +1,133 @@
|
|||||||
|
package smdp
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"math/rand"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// LoadBalancer implements different load balancing strategies for carrier selection
|
||||||
|
type LoadBalancer struct {
|
||||||
|
strategy LoadBalancingStrategy
|
||||||
|
rand *rand.Rand
|
||||||
|
}
|
||||||
|
|
||||||
|
// LoadBalancingStrategy defines the load balancing algorithm
|
||||||
|
type LoadBalancingStrategy int
|
||||||
|
|
||||||
|
const (
|
||||||
|
StrategyRoundRobin LoadBalancingStrategy = iota
|
||||||
|
StrategyWeightedRoundRobin
|
||||||
|
StrategyLeastConnections
|
||||||
|
StrategyRandom
|
||||||
|
StrategyPriority
|
||||||
|
)
|
||||||
|
|
||||||
|
// NewLoadBalancer creates a new load balancer with default strategy
|
||||||
|
func NewLoadBalancer() *LoadBalancer {
|
||||||
|
return &LoadBalancer{
|
||||||
|
strategy: StrategyWeightedRoundRobin,
|
||||||
|
rand: rand.New(rand.NewSource(time.Now().UnixNano())),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// SelectCarrier selects the best carrier based on the configured strategy
|
||||||
|
func (lb *LoadBalancer) SelectCarrier(carriers []*Carrier, req *ProfileRequest) (*Carrier, error) {
|
||||||
|
if len(carriers) == 0 {
|
||||||
|
return nil, fmt.Errorf("no carriers available")
|
||||||
|
}
|
||||||
|
|
||||||
|
switch lb.strategy {
|
||||||
|
case StrategyRoundRobin:
|
||||||
|
return lb.roundRobinSelect(carriers), nil
|
||||||
|
case StrategyWeightedRoundRobin:
|
||||||
|
return lb.weightedRoundRobinSelect(carriers), nil
|
||||||
|
case StrategyLeastConnections:
|
||||||
|
return lb.leastConnectionsSelect(carriers), nil
|
||||||
|
case StrategyRandom:
|
||||||
|
return lb.randomSelect(carriers), nil
|
||||||
|
case StrategyPriority:
|
||||||
|
return lb.prioritySelect(carriers), nil
|
||||||
|
default:
|
||||||
|
return lb.weightedRoundRobinSelect(carriers), nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// roundRobinSelect implements round-robin carrier selection
|
||||||
|
func (lb *LoadBalancer) roundRobinSelect(carriers []*Carrier) *Carrier {
|
||||||
|
// Simple round-robin based on request count
|
||||||
|
// In a real implementation, you'd maintain state across requests
|
||||||
|
return carriers[0] // Simplified for demo
|
||||||
|
}
|
||||||
|
|
||||||
|
// weightedRoundRobinSelect selects carrier based on priority and health
|
||||||
|
func (lb *LoadBalancer) weightedRoundRobinSelect(carriers []*Carrier) *Carrier {
|
||||||
|
// Calculate weights based on priority and success rate
|
||||||
|
var totalWeight int
|
||||||
|
weights := make([]int, len(carriers))
|
||||||
|
|
||||||
|
for i, carrier := range carriers {
|
||||||
|
weight := carrier.Priority
|
||||||
|
|
||||||
|
// Adjust weight based on success rate
|
||||||
|
if carrier.Metrics.TotalRequests > 0 {
|
||||||
|
successRate := float64(carrier.Metrics.SuccessfulRequests) / float64(carrier.Metrics.TotalRequests)
|
||||||
|
weight = int(float64(weight) * successRate)
|
||||||
|
}
|
||||||
|
|
||||||
|
weights[i] = weight
|
||||||
|
totalWeight += weight
|
||||||
|
}
|
||||||
|
|
||||||
|
if totalWeight == 0 {
|
||||||
|
return carriers[0]
|
||||||
|
}
|
||||||
|
|
||||||
|
// Select carrier based on weighted random
|
||||||
|
random := lb.rand.Intn(totalWeight)
|
||||||
|
currentWeight := 0
|
||||||
|
|
||||||
|
for i, weight := range weights {
|
||||||
|
currentWeight += weight
|
||||||
|
if random < currentWeight {
|
||||||
|
return carriers[i]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return carriers[0]
|
||||||
|
}
|
||||||
|
|
||||||
|
// leastConnectionsSelect selects carrier with least current load
|
||||||
|
func (lb *LoadBalancer) leastConnectionsSelect(carriers []*Carrier) *Carrier {
|
||||||
|
bestCarrier := carriers[0]
|
||||||
|
minLoad := bestCarrier.Metrics.RequestRate
|
||||||
|
|
||||||
|
for _, carrier := range carriers[1:] {
|
||||||
|
if carrier.Metrics.RequestRate < minLoad {
|
||||||
|
bestCarrier = carrier
|
||||||
|
minLoad = carrier.Metrics.RequestRate
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return bestCarrier
|
||||||
|
}
|
||||||
|
|
||||||
|
// randomSelect selects a random carrier
|
||||||
|
func (lb *LoadBalancer) randomSelect(carriers []*Carrier) *Carrier {
|
||||||
|
return carriers[lb.rand.Intn(len(carriers))]
|
||||||
|
}
|
||||||
|
|
||||||
|
// prioritySelect selects carrier with highest priority
|
||||||
|
func (lb *LoadBalancer) prioritySelect(carriers []*Carrier) *Carrier {
|
||||||
|
bestCarrier := carriers[0]
|
||||||
|
highestPriority := bestCarrier.Priority
|
||||||
|
|
||||||
|
for _, carrier := range carriers[1:] {
|
||||||
|
if carrier.Priority > highestPriority {
|
||||||
|
bestCarrier = carrier
|
||||||
|
highestPriority = carrier.Priority
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return bestCarrier
|
||||||
|
}
|
||||||
@@ -0,0 +1,121 @@
|
|||||||
|
package smdp
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/es2"
|
||||||
|
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/repository"
|
||||||
|
"github.com/sirupsen/logrus"
|
||||||
|
)
|
||||||
|
|
||||||
|
// SMDPManager manages multiple SM-DP+ carriers
|
||||||
|
type SMDPManager struct {
|
||||||
|
carriers map[string]*Carrier
|
||||||
|
carriersMutex sync.RWMutex
|
||||||
|
es2Clients map[string]*es2.ES2Client
|
||||||
|
clientsMutex sync.RWMutex
|
||||||
|
repository *repository.PostgresProfileStore
|
||||||
|
healthChecker *HealthChecker
|
||||||
|
loadBalancer *LoadBalancer
|
||||||
|
config *ManagerConfig
|
||||||
|
logger *logrus.Logger
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewSMDPManager creates a new SM-DP+ Manager
|
||||||
|
func NewSMDPManager(repo *repository.PostgresProfileStore, config *ManagerConfig) *SMDPManager {
|
||||||
|
logger := logrus.New()
|
||||||
|
logger.SetLevel(logrus.InfoLevel)
|
||||||
|
|
||||||
|
return &SMDPManager{
|
||||||
|
carriers: make(map[string]*Carrier),
|
||||||
|
es2Clients: make(map[string]*es2.ES2Client),
|
||||||
|
repository: repo,
|
||||||
|
config: config,
|
||||||
|
logger: logger,
|
||||||
|
healthChecker: NewHealthChecker(config.HealthCheckInterval),
|
||||||
|
loadBalancer: NewLoadBalancer(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// AddCarrier adds a new carrier to the manager
|
||||||
|
func (m *SMDPManager) AddCarrier(carrier *Carrier) error {
|
||||||
|
m.carriersMutex.Lock()
|
||||||
|
defer m.carriersMutex.Unlock()
|
||||||
|
|
||||||
|
if err := m.validateCarrier(carrier); err != nil {
|
||||||
|
return fmt.Errorf("invalid carrier configuration: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
client := es2.NewES2Client(carrier.ES2Config)
|
||||||
|
m.carriers[carrier.ID] = carrier
|
||||||
|
m.es2Clients[carrier.ID] = client
|
||||||
|
|
||||||
|
m.logger.WithFields(logrus.Fields{
|
||||||
|
"carrier_id": carrier.ID,
|
||||||
|
"carrier_name": carrier.Name,
|
||||||
|
}).Info("Added carrier to SM-DP+ Manager")
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// RemoveCarrier removes a carrier from the manager
|
||||||
|
func (m *SMDPManager) RemoveCarrier(carrierID string) error {
|
||||||
|
m.carriersMutex.Lock()
|
||||||
|
defer m.carriersMutex.Unlock()
|
||||||
|
|
||||||
|
m.clientsMutex.Lock()
|
||||||
|
defer m.clientsMutex.Unlock()
|
||||||
|
|
||||||
|
if _, exists := m.carriers[carrierID]; !exists {
|
||||||
|
return fmt.Errorf("carrier %s not found", carrierID)
|
||||||
|
}
|
||||||
|
|
||||||
|
delete(m.carriers, carrierID)
|
||||||
|
delete(m.es2Clients, carrierID)
|
||||||
|
|
||||||
|
m.logger.WithField("carrier_id", carrierID).Info("Removed carrier from SM-DP+ Manager")
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetCarrierStatus returns the current status of all carriers
|
||||||
|
func (m *SMDPManager) GetCarrierStatus() map[string]*Carrier {
|
||||||
|
m.carriersMutex.RLock()
|
||||||
|
defer m.carriersMutex.RUnlock()
|
||||||
|
|
||||||
|
status := make(map[string]*Carrier)
|
||||||
|
for id, carrier := range m.carriers {
|
||||||
|
carrierCopy := *carrier
|
||||||
|
status[id] = &carrierCopy
|
||||||
|
}
|
||||||
|
return status
|
||||||
|
}
|
||||||
|
|
||||||
|
// StartHealthChecking starts the background health checking process
|
||||||
|
func (m *SMDPManager) StartHealthChecking(ctx context.Context) {
|
||||||
|
m.healthChecker.Start(ctx, m.carriers, m.es2Clients, m.updateCarrierHealth)
|
||||||
|
}
|
||||||
|
|
||||||
|
// validateCarrier validates carrier configuration
|
||||||
|
func (m *SMDPManager) validateCarrier(carrier *Carrier) error {
|
||||||
|
if carrier.ID == "" || carrier.Name == "" || carrier.MCC == "" || carrier.MNC == "" {
|
||||||
|
return fmt.Errorf("carrier ID, name, MCC, and MNC are required")
|
||||||
|
}
|
||||||
|
if carrier.ES2Config == nil || carrier.ES2Config.BaseURL == "" {
|
||||||
|
return fmt.Errorf("ES2+ configuration and base URL are required")
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// updateCarrierHealth updates the health status of a carrier
|
||||||
|
func (m *SMDPManager) updateCarrierHealth(carrierID string, status CarrierHealthStatus) {
|
||||||
|
m.carriersMutex.Lock()
|
||||||
|
defer m.carriersMutex.Unlock()
|
||||||
|
|
||||||
|
if carrier, exists := m.carriers[carrierID]; exists {
|
||||||
|
carrier.HealthStatus = status
|
||||||
|
carrier.LastHealthCheck = time.Now()
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,200 @@
|
|||||||
|
package smdp
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/es2"
|
||||||
|
"github.com/sirupsen/logrus"
|
||||||
|
)
|
||||||
|
|
||||||
|
func (m *SMDPManager) DownloadProfile(ctx context.Context, req *ProfileRequest) (*ProfileResponse, error) {
|
||||||
|
startTime := time.Now()
|
||||||
|
|
||||||
|
m.logger.WithFields(logrus.Fields{
|
||||||
|
"eid": req.EID,
|
||||||
|
"preferred": req.PreferredCarrier,
|
||||||
|
}).Info("Processing profile download request")
|
||||||
|
|
||||||
|
selectedCarrier, err := m.SelectCarrier(ctx, req)
|
||||||
|
if err != nil {
|
||||||
|
return &ProfileResponse{
|
||||||
|
Success: false,
|
||||||
|
StatusMessage: fmt.Sprintf("Carrier selection failed: %v", err),
|
||||||
|
ResponseTime: time.Since(startTime),
|
||||||
|
}, err
|
||||||
|
}
|
||||||
|
|
||||||
|
response, err := m.downloadWithRetry(ctx, req, selectedCarrier, 0)
|
||||||
|
if err != nil && m.config.EnableFailover {
|
||||||
|
m.logger.WithError(err).Warn("Primary carrier failed, attempting failover")
|
||||||
|
return m.handleFailover(ctx, req, selectedCarrier.ID, startTime)
|
||||||
|
}
|
||||||
|
|
||||||
|
response.ResponseTime = time.Since(startTime)
|
||||||
|
return response, err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *SMDPManager) SelectCarrier(ctx context.Context, req *ProfileRequest) (*Carrier, error) {
|
||||||
|
m.carriersMutex.RLock()
|
||||||
|
defer m.carriersMutex.RUnlock()
|
||||||
|
|
||||||
|
var activeCarriers []*Carrier
|
||||||
|
for _, carrier := range m.carriers {
|
||||||
|
if carrier.IsActive && carrier.HealthStatus == CarrierStatusHealthy {
|
||||||
|
activeCarriers = append(activeCarriers, carrier)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(activeCarriers) == 0 {
|
||||||
|
return nil, fmt.Errorf("no healthy carriers available")
|
||||||
|
}
|
||||||
|
|
||||||
|
if req.PreferredCarrier != "" {
|
||||||
|
if carrier, exists := m.carriers[req.PreferredCarrier]; exists &&
|
||||||
|
carrier.IsActive && carrier.HealthStatus == CarrierStatusHealthy {
|
||||||
|
return carrier, nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if m.config.EnableLoadBalancing {
|
||||||
|
return m.loadBalancer.SelectCarrier(activeCarriers, req)
|
||||||
|
}
|
||||||
|
|
||||||
|
return m.getHighestPriorityCarrier(activeCarriers), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *SMDPManager) downloadWithRetry(ctx context.Context, req *ProfileRequest, carrier *Carrier, attempt int) (*ProfileResponse, error) {
|
||||||
|
m.clientsMutex.RLock()
|
||||||
|
client, exists := m.es2Clients[carrier.ID]
|
||||||
|
m.clientsMutex.RUnlock()
|
||||||
|
|
||||||
|
if !exists {
|
||||||
|
return nil, fmt.Errorf("ES2+ client not found for carrier %s", carrier.ID)
|
||||||
|
}
|
||||||
|
|
||||||
|
downloadReq := &es2.DownloadProfileRequest{
|
||||||
|
EID: req.EID,
|
||||||
|
ICCID: req.ICCID,
|
||||||
|
ProfileType: req.ProfileType,
|
||||||
|
ConfirmationCode: req.ConfirmationCode,
|
||||||
|
}
|
||||||
|
|
||||||
|
startTime := time.Now()
|
||||||
|
resp, err := client.DownloadProfile(ctx, downloadReq)
|
||||||
|
responseTime := time.Since(startTime)
|
||||||
|
|
||||||
|
m.updateCarrierMetrics(carrier.ID, err == nil, responseTime, err)
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
m.logger.WithFields(logrus.Fields{
|
||||||
|
"carrier_id": carrier.ID,
|
||||||
|
"attempt": attempt + 1,
|
||||||
|
"error": err,
|
||||||
|
}).Error("Profile download failed")
|
||||||
|
|
||||||
|
if attempt < m.config.MaxRetries {
|
||||||
|
time.Sleep(m.config.RetryDelay)
|
||||||
|
return m.downloadWithRetry(ctx, req, carrier, attempt+1)
|
||||||
|
}
|
||||||
|
|
||||||
|
return &ProfileResponse{
|
||||||
|
Success: false,
|
||||||
|
CarrierID: carrier.ID,
|
||||||
|
StatusMessage: fmt.Sprintf("Download failed after %d attempts: %v", attempt+1, err),
|
||||||
|
ResponseTime: responseTime,
|
||||||
|
}, err
|
||||||
|
}
|
||||||
|
|
||||||
|
m.logger.WithFields(logrus.Fields{
|
||||||
|
"carrier_id": carrier.ID,
|
||||||
|
"execution_status": resp.ExecutionStatus,
|
||||||
|
"response_time": responseTime,
|
||||||
|
}).Info("Profile download successful")
|
||||||
|
|
||||||
|
return &ProfileResponse{
|
||||||
|
Success: true,
|
||||||
|
CarrierID: carrier.ID,
|
||||||
|
ExecutionStatus: resp.ExecutionStatus,
|
||||||
|
StatusMessage: resp.StatusMessage,
|
||||||
|
ResponseTime: responseTime,
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// handleFailover attempts to download profile using alternative carriers
|
||||||
|
func (m *SMDPManager) handleFailover(ctx context.Context, req *ProfileRequest, failedCarrierID string, startTime time.Time) (*ProfileResponse, error) {
|
||||||
|
m.carriersMutex.RLock()
|
||||||
|
defer m.carriersMutex.RUnlock()
|
||||||
|
|
||||||
|
var retriedOn []string
|
||||||
|
|
||||||
|
for carrierID, carrier := range m.carriers {
|
||||||
|
if carrierID == failedCarrierID || !carrier.IsActive || carrier.HealthStatus != CarrierStatusHealthy {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
m.logger.WithField("carrier_id", carrierID).Info("Attempting failover to carrier")
|
||||||
|
|
||||||
|
response, err := m.downloadWithRetry(ctx, req, carrier, 0)
|
||||||
|
if err == nil && response.Success {
|
||||||
|
response.RetriedOn = retriedOn
|
||||||
|
response.ResponseTime = time.Since(startTime)
|
||||||
|
return response, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
retriedOn = append(retriedOn, carrierID)
|
||||||
|
}
|
||||||
|
|
||||||
|
return &ProfileResponse{
|
||||||
|
Success: false,
|
||||||
|
StatusMessage: "All carriers failed during failover",
|
||||||
|
ResponseTime: time.Since(startTime),
|
||||||
|
RetriedOn: retriedOn,
|
||||||
|
}, fmt.Errorf("failover exhausted all carriers")
|
||||||
|
}
|
||||||
|
|
||||||
|
// updateCarrierMetrics updates the metrics for a carrier
|
||||||
|
func (m *SMDPManager) updateCarrierMetrics(carrierID string, success bool, responseTime time.Duration, err error) {
|
||||||
|
m.carriersMutex.Lock()
|
||||||
|
defer m.carriersMutex.Unlock()
|
||||||
|
|
||||||
|
carrier, exists := m.carriers[carrierID]
|
||||||
|
if !exists {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
carrier.Metrics.TotalRequests++
|
||||||
|
if success {
|
||||||
|
carrier.Metrics.SuccessfulRequests++
|
||||||
|
} else {
|
||||||
|
carrier.Metrics.FailedRequests++
|
||||||
|
carrier.Metrics.LastError = err.Error()
|
||||||
|
carrier.Metrics.LastErrorTime = time.Now()
|
||||||
|
}
|
||||||
|
|
||||||
|
if carrier.Metrics.TotalRequests == 1 {
|
||||||
|
carrier.Metrics.AverageResponseTime = responseTime
|
||||||
|
} else {
|
||||||
|
carrier.Metrics.AverageResponseTime = time.Duration(
|
||||||
|
(int64(carrier.Metrics.AverageResponseTime)*int64(carrier.Metrics.TotalRequests-1) + int64(responseTime)) / int64(carrier.Metrics.TotalRequests),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
carrier.Metrics.RequestRate = float64(carrier.Metrics.TotalRequests) / time.Since(time.Now().Add(-time.Minute)).Seconds()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *SMDPManager) getHighestPriorityCarrier(carriers []*Carrier) *Carrier {
|
||||||
|
if len(carriers) == 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
highestPriority := carriers[0]
|
||||||
|
for _, carrier := range carriers {
|
||||||
|
if carrier.Priority > highestPriority.Priority {
|
||||||
|
highestPriority = carrier
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return highestPriority
|
||||||
|
}
|
||||||
@@ -0,0 +1,87 @@
|
|||||||
|
package smdp
|
||||||
|
|
||||||
|
import (
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/config"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Carrier represents a mobile network operator with SM-DP+ capabilities
|
||||||
|
type Carrier struct {
|
||||||
|
ID string `json:"id"`
|
||||||
|
Name string `json:"name"`
|
||||||
|
CountryCode string `json:"country_code"`
|
||||||
|
MCC string `json:"mcc"`
|
||||||
|
MNC string `json:"mnc"`
|
||||||
|
ES2Config *config.ES2Config `json:"es2_config"`
|
||||||
|
Priority int `json:"priority"` // Higher number = higher priority
|
||||||
|
IsActive bool `json:"is_active"`
|
||||||
|
HealthStatus CarrierHealthStatus `json:"health_status"`
|
||||||
|
LastHealthCheck time.Time `json:"last_health_check"`
|
||||||
|
Metrics *CarrierMetrics `json:"metrics"`
|
||||||
|
Capabilities *CarrierCapabilities `json:"capabilities"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// CarrierHealthStatus represents the health status of a carrier
|
||||||
|
type CarrierHealthStatus string
|
||||||
|
|
||||||
|
const (
|
||||||
|
CarrierStatusHealthy CarrierHealthStatus = "healthy"
|
||||||
|
CarrierStatusDegraded CarrierHealthStatus = "degraded"
|
||||||
|
CarrierStatusUnhealthy CarrierHealthStatus = "unhealthy"
|
||||||
|
CarrierStatusUnknown CarrierHealthStatus = "unknown"
|
||||||
|
)
|
||||||
|
|
||||||
|
// CarrierCapabilities represents the capabilities of a carrier
|
||||||
|
type CarrierCapabilities struct {
|
||||||
|
SupportedProfileTypes []string `json:"supported_profile_types"`
|
||||||
|
MaxConcurrentRequests int `json:"max_concurrent_requests"`
|
||||||
|
SupportedMCCs []string `json:"supported_mccs"`
|
||||||
|
SupportedMNCs []string `json:"supported_mncs"`
|
||||||
|
Features []string `json:"features"` // e.g., ["bulk_download", "remote_provisioning"]
|
||||||
|
}
|
||||||
|
|
||||||
|
// CarrierMetrics tracks performance metrics for a carrier
|
||||||
|
type CarrierMetrics struct {
|
||||||
|
TotalRequests uint64 `json:"total_requests"`
|
||||||
|
SuccessfulRequests uint64 `json:"successful_requests"`
|
||||||
|
FailedRequests uint64 `json:"failed_requests"`
|
||||||
|
AverageResponseTime time.Duration `json:"average_response_time"`
|
||||||
|
LastError string `json:"last_error"`
|
||||||
|
LastErrorTime time.Time `json:"last_error_time"`
|
||||||
|
RequestRate float64 `json:"request_rate"` // requests per second
|
||||||
|
}
|
||||||
|
|
||||||
|
// ProfileRequest represents an eSIM profile download request
|
||||||
|
type ProfileRequest struct {
|
||||||
|
EID string `json:"eid"`
|
||||||
|
ICCID string `json:"iccid"`
|
||||||
|
ProfileType string `json:"profile_type"`
|
||||||
|
ConfirmationCode string `json:"confirmation_code,omitempty"`
|
||||||
|
PreferredCarrier string `json:"preferred_carrier,omitempty"` // Optional preferred carrier
|
||||||
|
IMSI string `json:"imsi,omitempty"`
|
||||||
|
TenantID string `json:"tenant_id,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// ProfileResponse represents the response from a profile operation
|
||||||
|
type ProfileResponse struct {
|
||||||
|
Success bool `json:"success"`
|
||||||
|
CarrierID string `json:"carrier_id"`
|
||||||
|
ExecutionStatus string `json:"execution_status"`
|
||||||
|
StatusMessage string `json:"status_message"`
|
||||||
|
ProfileState string `json:"profile_state,omitempty"`
|
||||||
|
ResponseTime time.Duration `json:"response_time"`
|
||||||
|
RetriedOn []string `json:"retried_on,omitempty"` // List of carrier IDs tried before success
|
||||||
|
}
|
||||||
|
|
||||||
|
// ManagerConfig configures the SM-DP+ Manager
|
||||||
|
type ManagerConfig struct {
|
||||||
|
HealthCheckInterval time.Duration `json:"health_check_interval"`
|
||||||
|
MaxRetries int `json:"max_retries"`
|
||||||
|
RetryDelay time.Duration `json:"retry_delay"`
|
||||||
|
CircuitBreakerThreshold int `json:"circuit_breaker_threshold"`
|
||||||
|
CircuitBreakerTimeout time.Duration `json:"circuit_breaker_timeout"`
|
||||||
|
EnableLoadBalancing bool `json:"enable_load_balancing"`
|
||||||
|
EnableFailover bool `json:"enable_failover"`
|
||||||
|
DefaultTimeout time.Duration `json:"default_timeout"`
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user