MiFaZhan
771 lines
40 KiB
Go
771 lines
40 KiB
Go
package worker
|
|
|
|
import (
|
|
"context"
|
|
"os"
|
|
"os/signal"
|
|
"strings"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/gogf/gf/v2/database/gredis"
|
|
"github.com/gogf/gf/v2/errors/gerror"
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
"github.com/gogf/gf/v2/os/gtime"
|
|
"github.com/gogf/gf/v2/util/gmode"
|
|
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/batch"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/client/upstream"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consts"
|
|
rabbitmqconsumer "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consumer/rabbitmq"
|
|
zsetconsumer "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consumer/zset"
|
|
fullsyncarchive "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/archive"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/httpapi"
|
|
fullsyncruntime "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/runtime"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/supply"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/library/tdengine"
|
|
dailyusageproducer "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/producer/dailyusage"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/producer/imei"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/producer/monthlyusage"
|
|
monthlyhistoryproducer "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/producer/monthlyusagehistory"
|
|
simstatusproducer "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/producer/simstatus"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
|
|
batchrepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/batch"
|
|
cancelrepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/cancel"
|
|
datarepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/data"
|
|
imeirepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/imei"
|
|
inboundrepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/inbound"
|
|
lifecyclerepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/lifecycle"
|
|
monthlyusagerepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/monthlyusage"
|
|
simstatusrepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/simstatus"
|
|
trafficusagerepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/trafficusage"
|
|
unsubrepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/unsub"
|
|
addpackageservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/addpackage"
|
|
allocateservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/allocate"
|
|
balanceservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/balance"
|
|
cancelservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/cancel"
|
|
dailyusageservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/dailyusage"
|
|
dataservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/data"
|
|
imeiservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/imei"
|
|
inboundservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/inbound"
|
|
lifecycleservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/lifecycle"
|
|
monthlyusageservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/monthlyusage"
|
|
monthlyhistoryservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/monthlyusagehistory"
|
|
packagecustomerservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/packagecustomer"
|
|
renewchangeservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/renewchange"
|
|
rolesyncservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/rolesync"
|
|
simstatusservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/simstatus"
|
|
trafficcutoffconfigservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/trafficcutoffconfig"
|
|
trafficusageservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/trafficusage"
|
|
unsubservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/unsub"
|
|
warningconfigservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/warningconfig"
|
|
workorderservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/workorder"
|
|
)
|
|
|
|
const (
|
|
shutdownTimeout = 10 * time.Second
|
|
consumerRestartInitialDelay = 2 * time.Second
|
|
consumerRestartMaxDelay = 30 * time.Second
|
|
consumerRestartBackoffResetAfter = time.Minute
|
|
)
|
|
|
|
const monthlyUsageTemplateQPSConfigKey = "trafficUsageFullSync.consumer.templateQPS"
|
|
|
|
func loadMonthlyUsageTemplateQPS(ctx context.Context) map[string]float64 {
|
|
value, err := g.Cfg().Get(ctx, monthlyUsageTemplateQPSConfigKey)
|
|
if err != nil {
|
|
g.Log().Warningf(ctx, "load monthly usage template QPS config failed: %+v", err)
|
|
return nil
|
|
}
|
|
if value == nil || value.IsNil() {
|
|
return nil
|
|
}
|
|
var configured map[string]float64
|
|
if err = value.Scan(&configured); err != nil {
|
|
g.Log().Warningf(ctx, "parse monthly usage template QPS config failed: %+v", err)
|
|
return nil
|
|
}
|
|
return configured
|
|
}
|
|
|
|
func loadSimStatusTemplateQPS(ctx context.Context) map[string]float64 {
|
|
value, err := g.Cfg().Get(ctx, "simStatusQueueSync.consumer.templateQPS")
|
|
if err != nil || value == nil || value.IsNil() {
|
|
return nil
|
|
}
|
|
var configured map[string]float64
|
|
if err = value.Scan(&configured); err != nil {
|
|
g.Log().Warningf(ctx, "parse sim status template QPS config failed: %+v", err)
|
|
return nil
|
|
}
|
|
return configured
|
|
}
|
|
|
|
func loadImeiTemplateQPS(ctx context.Context) map[string]float64 {
|
|
value, err := g.Cfg().Get(ctx, "imeiFullSync.consumer.templateQPS")
|
|
if err != nil || value == nil || value.IsNil() {
|
|
return nil
|
|
}
|
|
var configured map[string]float64
|
|
if err = value.Scan(&configured); err != nil {
|
|
g.Log().Warningf(ctx, "parse imei template QPS config failed: %+v", err)
|
|
return nil
|
|
}
|
|
return configured
|
|
}
|
|
|
|
func loadSimStatusRunPrefix(ctx context.Context) (string, error) {
|
|
prefix := strings.TrimSpace(g.Cfg().MustGet(ctx, "simStatusQueueSync.redisKeyPrefix", "sim-status:").String())
|
|
if prefix == "" {
|
|
return "", gerror.New("sim status redisKeyPrefix must be nonempty")
|
|
}
|
|
return prefix, nil
|
|
}
|
|
|
|
func loadDailyUsageTemplateQPS(ctx context.Context) map[string]float64 {
|
|
value, err := g.Cfg().Get(ctx, "trafficUsageDailySync.consumer.templateQPS")
|
|
if err != nil || value == nil || value.IsNil() {
|
|
return nil
|
|
}
|
|
var configured map[string]float64
|
|
if err = value.Scan(&configured); err != nil {
|
|
g.Log().Warningf(ctx, "parse daily usage template QPS config failed: %+v", err)
|
|
return nil
|
|
}
|
|
return configured
|
|
}
|
|
|
|
func loadDailyUsageRunPrefix(ctx context.Context) (string, error) {
|
|
prefix := strings.TrimSpace(g.Cfg().MustGet(ctx, "trafficUsageDailySync.redisKeyPrefix", "daily:").String())
|
|
if prefix == "" {
|
|
return "", gerror.New("daily usage redisKeyPrefix must be nonempty")
|
|
}
|
|
return prefix, nil
|
|
}
|
|
|
|
func loadImeiRunPrefix(ctx context.Context) (string, error) {
|
|
prefix := strings.TrimSpace(g.Cfg().MustGet(ctx, "imeiFullSync.redisKeyPrefix", "imei:").String())
|
|
if prefix == "" {
|
|
return "", gerror.New("imei redisKeyPrefix must be nonempty")
|
|
}
|
|
return prefix, nil
|
|
}
|
|
|
|
func loadMonthlyHistoryRunPrefix(ctx context.Context) (string, error) {
|
|
prefix := strings.TrimSpace(g.Cfg().MustGet(ctx, "trafficUsageMonthlyHistorySync.redisKeyPrefix", "monthly-history:").String())
|
|
if prefix == "" {
|
|
return "", gerror.New("monthly usage history redisKeyPrefix must be nonempty")
|
|
}
|
|
return prefix, nil
|
|
}
|
|
|
|
func loadMonthlyHistoryTemplateQPS(ctx context.Context) map[string]float64 {
|
|
value, err := g.Cfg().Get(ctx, "trafficUsageMonthlyHistorySync.consumer.templateQPS")
|
|
if err != nil || value == nil || value.IsNil() {
|
|
return nil
|
|
}
|
|
var configured map[string]float64
|
|
if err = value.Scan(&configured); err != nil {
|
|
g.Log().Warningf(ctx, "parse monthly usage history template QPS config failed: %+v", err)
|
|
return nil
|
|
}
|
|
return configured
|
|
}
|
|
|
|
func superviseZSetConsumer(ctx context.Context, eventPrefix string, run func(context.Context) error) {
|
|
retryDelay := consumerRestartInitialDelay
|
|
for {
|
|
startedAt := time.Now()
|
|
runErr := run(ctx)
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
if time.Since(startedAt) >= consumerRestartBackoffResetAfter {
|
|
retryDelay = consumerRestartInitialDelay
|
|
}
|
|
|
|
errorText := ""
|
|
if runErr != nil {
|
|
errorText = runErr.Error()
|
|
}
|
|
g.Log().Warningf(ctx, "event=%s_consumer status=restart_scheduled retryDelayMs=%d error=%q", eventPrefix, retryDelay.Milliseconds(), errorText)
|
|
|
|
timer := time.NewTimer(retryDelay)
|
|
select {
|
|
case <-ctx.Done():
|
|
timer.Stop()
|
|
return
|
|
case <-timer.C:
|
|
}
|
|
|
|
retryDelay *= 2
|
|
if retryDelay > consumerRestartMaxDelay {
|
|
retryDelay = consumerRestartMaxDelay
|
|
}
|
|
}
|
|
}
|
|
|
|
func loadFullSyncArchivePrefixes(ctx context.Context) ([]string, error) {
|
|
trafficPrefix := strings.TrimSpace(g.Cfg().MustGet(ctx, "trafficUsageFullSync.redisKeyPrefix", "flow:").String())
|
|
if trafficPrefix == "" {
|
|
trafficPrefix = "flow:"
|
|
}
|
|
statusPrefix, err := loadSimStatusRunPrefix(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
dailyPrefix, err := loadDailyUsageRunPrefix(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
monthlyHistoryPrefix, err := loadMonthlyHistoryRunPrefix(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
imeiPrefix, err := loadImeiRunPrefix(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
// Duplicate prefixes would make Catalog and SimpleReplay read the same run
|
|
// hash twice. Deduplicate while preserving the traffic-first order.
|
|
prefixes := make([]string, 0, 5)
|
|
seen := make(map[string]struct{}, 5)
|
|
for _, prefix := range []string{trafficPrefix, statusPrefix, dailyPrefix, monthlyHistoryPrefix, imeiPrefix} {
|
|
if _, exists := seen[prefix]; exists {
|
|
continue
|
|
}
|
|
seen[prefix] = struct{}{}
|
|
prefixes = append(prefixes, prefix)
|
|
}
|
|
return prefixes, nil
|
|
}
|
|
|
|
func Run(ctx context.Context) error {
|
|
workerCtx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM)
|
|
defer stop()
|
|
|
|
if err := initialize(workerCtx); err != nil {
|
|
return err
|
|
}
|
|
|
|
var (
|
|
upstreamClient = upstream.New()
|
|
usageRepository = trafficusagerepository.New()
|
|
trafficUsage = trafficusageservice.New(upstreamClient, usageRepository)
|
|
workOrder = workorderservice.New()
|
|
balance = balanceservice.New()
|
|
inboundRepo = inboundrepository.New()
|
|
inboundExecutor = inboundservice.NewExecutorWithUpstream(inboundRepo, upstreamClient)
|
|
lifecycleRepo = lifecyclerepository.New()
|
|
lifecycleExecutor = lifecycleservice.NewExecutor(lifecycleRepo, upstreamClient)
|
|
cancelRepo = cancelrepository.New()
|
|
cancelExecutor = cancelservice.NewExecutor(cancelRepo)
|
|
renewChangeRepo = renewchangeservice.NewRepository()
|
|
addPackageCardExecutor = addpackageservice.NewCardExecutor(renewChangeRepo)
|
|
addPackagePoolExecutor = addpackageservice.NewPoolExecutor(renewChangeRepo)
|
|
renewPrepareExecutor = renewchangeservice.NewPrepareExecutor(renewChangeRepo)
|
|
prepareExecutor = addpackageservice.NewPrepareMux(renewChangeRepo, renewPrepareExecutor)
|
|
renewExecutor = renewchangeservice.NewRenewExecutor(renewChangeRepo)
|
|
changeExecutor = renewchangeservice.NewChangeExecutor(renewChangeRepo)
|
|
dataRepo = datarepository.New()
|
|
dataExecutor = dataservice.NewExecutor(dataRepo, upstreamClient)
|
|
unsubRepo = unsubrepository.New()
|
|
unsubExecutor = unsubservice.NewExecutor(unsubRepo)
|
|
restartExecutor = lifecycleservice.NewRestartExecutor(lifecycleRepo, upstreamClient)
|
|
batchRepository = batchrepository.New()
|
|
batchDispatcher = batch.NewDispatcher(lifecycleExecutor)
|
|
batchRuntime = batch.NewRuntime(
|
|
batchRepository,
|
|
batchDispatcher,
|
|
batch.WithExecutionTimeout(g.Cfg().MustGet(workerCtx, "rabbitmq.batchExecute.executionTimeout", "0s").Duration()),
|
|
batch.WithFinalizationTimeout(g.Cfg().MustGet(workerCtx, "rabbitmq.batchExecute.finalizationTimeout", "10s").Duration()),
|
|
)
|
|
allocateExecutor = allocateservice.NewExecutor()
|
|
warningConfigExecutor = warningconfigservice.NewExecutor()
|
|
trafficCutoffConfigExecutor = trafficcutoffconfigservice.NewExecutor()
|
|
packageCustomerSync = packagecustomerservice.New()
|
|
)
|
|
var (
|
|
monthlyProducer *monthlyusage.Producer
|
|
monthlyConsumer *zsetconsumer.Consumer
|
|
statusProducer *simstatusproducer.Producer
|
|
statusConsumer *zsetconsumer.Consumer
|
|
dailyProducer *dailyusageproducer.Producer
|
|
dailyConsumer *zsetconsumer.Consumer
|
|
monthlyHistoryProducer *monthlyhistoryproducer.Producer
|
|
monthlyHistoryConsumer *zsetconsumer.Consumer
|
|
imeiProducer *imei.Producer
|
|
imeiConsumer *zsetconsumer.Consumer
|
|
)
|
|
if g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.enabled", false).Bool() {
|
|
prefix := g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.redisKeyPrefix", "flow:").String()
|
|
store, err := zset.New(g.Redis(), prefix)
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize monthly usage Redis store failed")
|
|
}
|
|
store.EnableArchiveIndex(g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.enabled", false).Bool())
|
|
producerConfig, configErr := loadMonthlyUsageProducerConfig(workerCtx)
|
|
if configErr != nil {
|
|
return configErr
|
|
}
|
|
var supplyTracker *supply.Tracker
|
|
if g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.supplyObservation.enabled", false).Bool() {
|
|
supplyTracker = supply.NewTracker(g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.enabled", false).Bool())
|
|
producerConfig.SupplyObservation = supplyTracker
|
|
}
|
|
monthlyProducer, err = monthlyusage.New(&monthlyusage.DatabaseScanner{}, store, producerConfig)
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize monthly usage producer failed")
|
|
}
|
|
monthlyUsageClient := upstream.NewWithCurrentUsageClient(
|
|
g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.chanClientName", "").String(),
|
|
g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.chanClientTimeout", "90s").Duration(),
|
|
)
|
|
usageHandler, handlerErr := monthlyusageservice.NewHandler(monthlyUsageClient, monthlyusagerepository.NewRepository())
|
|
if handlerErr != nil {
|
|
return gerror.Wrap(handlerErr, "initialize monthly usage handler failed")
|
|
}
|
|
monthlyConsumer, err = zsetconsumer.New(store, usageHandler, zsetconsumer.Config{
|
|
SupplyObservation: supplyTracker,
|
|
LockKey: g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.lockKey", "flow:consumer:active").String(),
|
|
LockTTL: g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.lockTTL", "1m").Duration(),
|
|
PollInterval: g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.pollInterval", "1s").Duration(),
|
|
LeaseTTL: g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.leaseTTL", "3m").Duration(),
|
|
TaskTimeout: g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.taskTimeout", "2m").Duration(),
|
|
ClaimBatch: g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.claimBatch", 100).Int(),
|
|
MobileConcurrency: g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.mobileConcurrency", 30).Int(),
|
|
TelecomConcurrency: g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.telecomConcurrency", 20).Int(),
|
|
GenericConcurrency: g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.genericConcurrency", 1).Int(),
|
|
GenericQPS: g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.genericQPS", 1.5).Float64(),
|
|
TemplateQPS: loadMonthlyUsageTemplateQPS(workerCtx),
|
|
RoundRetryEnabled: g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.retry.enabled", true).Bool(),
|
|
MaxRounds: g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.retry.maxRounds", 5).Int(),
|
|
FailureRetention: g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.retry.failureRetention", "720h").Duration(),
|
|
ShouldRetryNextRound: func(result zsetconsumer.Result) bool {
|
|
return !result.Success && result.Retryable
|
|
},
|
|
})
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize monthly usage consumer failed")
|
|
}
|
|
if g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.enabled", false).Bool() {
|
|
go superviseZSetConsumer(workerCtx, "monthly_usage", monthlyConsumer.Run)
|
|
}
|
|
}
|
|
if g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.enabled", false).Bool() {
|
|
prefix, prefixErr := loadDailyUsageRunPrefix(workerCtx)
|
|
if prefixErr != nil {
|
|
return prefixErr
|
|
}
|
|
store, err := zset.New(g.Redis(), prefix)
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize daily usage Redis store failed")
|
|
}
|
|
store.EnableArchiveIndex(g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.enabled", false).Bool())
|
|
dailyScanMode, dailyScanConcurrency, scanErr := loadHistoryUsageScanSettings(workerCtx, "trafficUsageDailySync")
|
|
if scanErr != nil {
|
|
return scanErr
|
|
}
|
|
dailyProducer, err = dailyusageproducer.New(&dailyusageproducer.DatabaseScanner{}, store, dailyusageproducer.Config{
|
|
ScanMode: dailyScanMode,
|
|
ScanConcurrency: dailyScanConcurrency,
|
|
PageSize: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.producer.pageSize", 100).Int(),
|
|
GlobalHighWaterMark: int64(g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.producer.globalHighWaterMark", 0).Int64()),
|
|
ChannelHighWaterMark: int64(g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.producer.channelHighWaterMark", 0).Int64()),
|
|
PollInterval: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.producer.pollInterval", "2s").Duration(),
|
|
RunLockKey: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.producer.runLockKey", "daily:producer:active").String(),
|
|
RunLockTTL: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.producer.runLockTTL", "5m").Duration(),
|
|
RoundRetryEnabled: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.retry.enabled", true).Bool(),
|
|
MaxRounds: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.retry.maxRounds", 3).Int(),
|
|
FailureRetention: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.retry.failureRetention", "720h").Duration(),
|
|
})
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize daily usage producer failed")
|
|
}
|
|
dailyHandler, handlerErr := dailyusageservice.NewHandler(upstream.New(), trafficusagerepository.New())
|
|
if handlerErr != nil {
|
|
return gerror.Wrap(handlerErr, "initialize daily usage handler failed")
|
|
}
|
|
dailyConsumer, err = zsetconsumer.New(store, dailyHandler, zsetconsumer.Config{
|
|
RunQueueEnabled: true,
|
|
RoundRetryEnabled: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.retry.enabled", true).Bool(),
|
|
MaxRounds: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.retry.maxRounds", 3).Int(),
|
|
FailureRetention: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.retry.failureRetention", "720h").Duration(),
|
|
ShouldRetryNextRound: dailyusageservice.ShouldRetryNextRound,
|
|
BlockedTasksAsFailures: true,
|
|
EventPrefix: "daily_usage",
|
|
LockKey: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.lockKey", "daily:consumer:active").String(),
|
|
LockTTL: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.lockTTL", "1m").Duration(),
|
|
PollInterval: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.pollInterval", "1s").Duration(),
|
|
LeaseTTL: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.leaseTTL", "3m").Duration(),
|
|
TaskTimeout: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.taskTimeout", "2m").Duration(),
|
|
ClaimBatch: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.claimBatch", 100).Int(),
|
|
MobileConcurrency: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.mobileConcurrency", 30).Int(),
|
|
TelecomConcurrency: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.telecomConcurrency", 8).Int(),
|
|
GenericConcurrency: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.genericConcurrency", 1).Int(),
|
|
GenericQPS: g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.genericQPS", 1.5).Float64(),
|
|
TemplateQPS: loadDailyUsageTemplateQPS(workerCtx),
|
|
})
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize daily usage consumer failed")
|
|
}
|
|
if g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.enabled", false).Bool() {
|
|
go superviseZSetConsumer(workerCtx, "daily_usage", dailyConsumer.Run)
|
|
}
|
|
}
|
|
if g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.enabled", false).Bool() {
|
|
prefix, prefixErr := loadMonthlyHistoryRunPrefix(workerCtx)
|
|
if prefixErr != nil {
|
|
return prefixErr
|
|
}
|
|
store, err := zset.New(g.Redis(), prefix)
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize monthly usage history Redis store failed")
|
|
}
|
|
store.EnableArchiveIndex(g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.enabled", false).Bool())
|
|
monthlyHistoryScanMode, monthlyHistoryScanConcurrency, scanErr := loadHistoryUsageScanSettings(workerCtx, "trafficUsageMonthlyHistorySync")
|
|
if scanErr != nil {
|
|
return scanErr
|
|
}
|
|
monthlyHistoryProducer, err = monthlyhistoryproducer.New(&monthlyhistoryproducer.DatabaseScanner{}, store, monthlyhistoryproducer.Config{
|
|
ScanMode: monthlyHistoryScanMode,
|
|
ScanConcurrency: monthlyHistoryScanConcurrency,
|
|
PageSize: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.producer.pageSize", 100).Int(),
|
|
GlobalHighWaterMark: int64(g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.producer.globalHighWaterMark", 0).Int64()),
|
|
ChannelHighWaterMark: int64(g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.producer.channelHighWaterMark", 0).Int64()),
|
|
PollInterval: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.producer.pollInterval", "2s").Duration(),
|
|
RunLockKey: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.producer.runLockKey", "monthly-history:producer:active").String(),
|
|
RunLockTTL: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.producer.runLockTTL", "5m").Duration(),
|
|
RoundRetryEnabled: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.retry.enabled", true).Bool(),
|
|
MaxRounds: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.retry.maxRounds", 3).Int(),
|
|
FailureRetention: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.retry.failureRetention", "720h").Duration(),
|
|
})
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize monthly usage history producer failed")
|
|
}
|
|
monthlyHistoryHandler, handlerErr := monthlyhistoryservice.NewHandler(upstream.New(), trafficusagerepository.New())
|
|
if handlerErr != nil {
|
|
return gerror.Wrap(handlerErr, "initialize monthly usage history handler failed")
|
|
}
|
|
monthlyHistoryConsumer, err = zsetconsumer.New(store, monthlyHistoryHandler, zsetconsumer.Config{
|
|
RunQueueEnabled: true,
|
|
RoundRetryEnabled: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.retry.enabled", true).Bool(),
|
|
MaxRounds: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.retry.maxRounds", 3).Int(),
|
|
FailureRetention: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.retry.failureRetention", "720h").Duration(),
|
|
ShouldRetryNextRound: monthlyhistoryservice.ShouldRetryNextRound,
|
|
BlockedTasksAsFailures: true,
|
|
EventPrefix: "monthly_usage_history",
|
|
LockKey: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.lockKey", "monthly-history:consumer:active").String(),
|
|
LockTTL: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.lockTTL", "1m").Duration(),
|
|
PollInterval: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.pollInterval", "1s").Duration(),
|
|
LeaseTTL: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.leaseTTL", "3m").Duration(),
|
|
TaskTimeout: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.taskTimeout", "2m").Duration(),
|
|
ClaimBatch: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.claimBatch", 100).Int(),
|
|
MobileConcurrency: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.mobileConcurrency", 30).Int(),
|
|
TelecomConcurrency: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.telecomConcurrency", 8).Int(),
|
|
GenericConcurrency: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.genericConcurrency", 1).Int(),
|
|
GenericQPS: g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.genericQPS", 1.5).Float64(),
|
|
TemplateQPS: loadMonthlyHistoryTemplateQPS(workerCtx),
|
|
})
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize monthly usage history consumer failed")
|
|
}
|
|
if g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.enabled", false).Bool() {
|
|
go superviseZSetConsumer(workerCtx, "monthly_usage_history", monthlyHistoryConsumer.Run)
|
|
}
|
|
}
|
|
if g.Cfg().MustGet(workerCtx, "simStatusQueueSync.enabled", false).Bool() {
|
|
prefix, prefixErr := loadSimStatusRunPrefix(workerCtx)
|
|
if prefixErr != nil {
|
|
return prefixErr
|
|
}
|
|
store, err := zset.New(g.Redis(), prefix)
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize sim status Redis store failed")
|
|
}
|
|
store.EnableArchiveIndex(g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.enabled", false).Bool())
|
|
statusProducer, err = simstatusproducer.New(&simstatusproducer.DatabaseScanner{}, store, simstatusproducer.Config{
|
|
PageSize: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.producer.pageSize", 1000).Int(),
|
|
GlobalHighWaterMark: int64(g.Cfg().MustGet(workerCtx, "simStatusQueueSync.producer.globalHighWaterMark", 0).Int64()),
|
|
PollInterval: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.producer.pollInterval", "1s").Duration(),
|
|
RunLockKey: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.producer.runLockKey", "sim-status:producer:active").String(),
|
|
RunLockTTL: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.producer.runLockTTL", "5m").Duration(),
|
|
})
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize sim status producer failed")
|
|
}
|
|
statusHandler, handlerErr := simstatusservice.NewHandler(
|
|
upstream.New(),
|
|
simstatusrepository.NewRepository(),
|
|
g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.networkRetries", 2).Int(),
|
|
)
|
|
if handlerErr != nil {
|
|
return gerror.Wrap(handlerErr, "initialize sim status handler failed")
|
|
}
|
|
statusConsumer, err = zsetconsumer.New(store, statusHandler, zsetconsumer.Config{
|
|
RunQueueEnabled: true,
|
|
RoundRetryEnabled: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.retry.enabled", true).Bool(),
|
|
MaxRounds: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.retry.maxRounds", 3).Int(),
|
|
FailureRetention: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.retry.failureRetention", "720h").Duration(),
|
|
ShouldRetryNextRound: simstatusservice.ShouldRetryNextRound,
|
|
BlockedTasksAsFailures: true,
|
|
EventPrefix: "sim_status_queue",
|
|
LockKey: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.lockKey", "sim-status:consumer:active").String(),
|
|
LockTTL: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.lockTTL", "1m").Duration(),
|
|
PollInterval: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.pollInterval", "1s").Duration(),
|
|
LeaseTTL: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.leaseTTL", "3m").Duration(),
|
|
TaskTimeout: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.taskTimeout", "2m").Duration(),
|
|
ClaimBatch: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.claimBatch", 100).Int(),
|
|
MobileConcurrency: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.mobileConcurrency", 100).Int(),
|
|
TelecomConcurrency: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.telecomConcurrency", 8).Int(),
|
|
GenericConcurrency: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.genericConcurrency", 1).Int(),
|
|
GenericQPS: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.genericQPS", 1.5).Float64(),
|
|
TemplateQPS: loadSimStatusTemplateQPS(workerCtx),
|
|
FailurePoolAlertThreshold: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.failurePoolAlertThreshold", 50000).Int64(),
|
|
})
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize sim status consumer failed")
|
|
}
|
|
if g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.enabled", false).Bool() {
|
|
go superviseZSetConsumer(workerCtx, "sim_status_queue", statusConsumer.Run)
|
|
}
|
|
}
|
|
if g.Cfg().MustGet(workerCtx, "imeiFullSync.enabled", false).Bool() {
|
|
prefix, prefixErr := loadImeiRunPrefix(workerCtx)
|
|
if prefixErr != nil {
|
|
return prefixErr
|
|
}
|
|
store, err := zset.New(g.Redis(), prefix)
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize imei Redis store failed")
|
|
}
|
|
store.EnableArchiveIndex(g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.enabled", false).Bool())
|
|
// g.Redis may be initialized from g.Cfg without registering GetConfig.
|
|
leaseConfig, configured := gredis.GetConfig()
|
|
if !configured {
|
|
value, configErr := g.Cfg().Get(workerCtx, "redis.default")
|
|
if configErr != nil {
|
|
return gerror.Wrap(configErr, "load IMEI lease Redis configuration failed")
|
|
}
|
|
leaseConfig, err = gredis.ConfigFromMap(value.Map())
|
|
if err != nil {
|
|
return gerror.Wrap(err, "parse IMEI lease Redis configuration failed")
|
|
}
|
|
}
|
|
imeiProducer, err = imei.New(&imei.DatabaseScanner{}, store, imei.Config{
|
|
PageSize: g.Cfg().MustGet(workerCtx, "imeiFullSync.producer.pageSize", 1000).Int(),
|
|
GlobalHighWaterMark: int64(g.Cfg().MustGet(workerCtx, "imeiFullSync.producer.globalHighWaterMark", 0).Int64()),
|
|
PollInterval: g.Cfg().MustGet(workerCtx, "imeiFullSync.producer.pollInterval", "1s").Duration(),
|
|
RunLockKey: g.Cfg().MustGet(workerCtx, "imeiFullSync.producer.runLockKey", "imei:producer:active").String(),
|
|
RunLockTTL: g.Cfg().MustGet(workerCtx, "imeiFullSync.producer.runLockTTL", "5m").Duration(),
|
|
LeaseRedisConfig: leaseConfig,
|
|
})
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize imei producer failed")
|
|
}
|
|
imeiHandler, handlerErr := imeiservice.NewHandler(
|
|
upstream.NewWithImeiClient(
|
|
g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.chanClientName", "").String(),
|
|
g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.chanClientTimeout", "30s").Duration(),
|
|
),
|
|
imeirepository.NewRepository(),
|
|
g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.networkRetries", 2).Int(),
|
|
)
|
|
if handlerErr != nil {
|
|
return gerror.Wrap(handlerErr, "initialize imei handler failed")
|
|
}
|
|
imeiConsumer, err = zsetconsumer.New(store, imeiHandler, zsetconsumer.Config{
|
|
RunQueueEnabled: true,
|
|
RoundRetryEnabled: g.Cfg().MustGet(workerCtx, "imeiFullSync.retry.enabled", true).Bool(),
|
|
MaxRounds: g.Cfg().MustGet(workerCtx, "imeiFullSync.retry.maxRounds", 5).Int(),
|
|
FailureRetention: g.Cfg().MustGet(workerCtx, "imeiFullSync.retry.failureRetention", "720h").Duration(),
|
|
ShouldRetryNextRound: imeiservice.ShouldRetryNextRound,
|
|
BlockedTasksAsFailures: true,
|
|
EventPrefix: "imei_queue",
|
|
LockKey: g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.lockKey", "imei:consumer:active").String(),
|
|
LockTTL: g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.lockTTL", "1m").Duration(),
|
|
PollInterval: g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.pollInterval", "1s").Duration(),
|
|
LeaseTTL: g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.leaseTTL", "3m").Duration(),
|
|
TaskTimeout: g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.taskTimeout", "2m").Duration(),
|
|
ClaimBatch: g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.claimBatch", 100).Int(),
|
|
MobileConcurrency: g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.mobileConcurrency", 100).Int(),
|
|
TelecomConcurrency: g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.telecomConcurrency", 8).Int(),
|
|
GenericConcurrency: g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.genericConcurrency", 1).Int(),
|
|
GenericQPS: g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.genericQPS", 1.5).Float64(),
|
|
TemplateQPS: loadImeiTemplateQPS(workerCtx),
|
|
// 失败池是唯一随卡数线性增长的每卡结构（单条约 1KB），
|
|
// 且会保留整个 failureRetention。超阀值只告警不截断。
|
|
FailurePoolAlertThreshold: g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.failurePoolAlertThreshold", 50000).Int64(),
|
|
})
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize imei consumer failed")
|
|
}
|
|
if g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.enabled", false).Bool() {
|
|
go superviseZSetConsumer(workerCtx, "imei_queue", imeiConsumer.Run)
|
|
}
|
|
}
|
|
var rabbitMQConsumers *rabbitmqconsumer.Consumers
|
|
if monthlyProducer != nil {
|
|
rabbitMQConsumers = rabbitmqconsumer.NewWithPackageCustomerSyncAndMonthlyUsage(trafficUsage, workOrder, balance, packageCustomerSync, monthlyProducer, batchRuntime)
|
|
} else {
|
|
rabbitMQConsumers = rabbitmqconsumer.NewWithPackageCustomerSync(trafficUsage, workOrder, balance, packageCustomerSync, batchRuntime)
|
|
}
|
|
if statusProducer != nil {
|
|
rabbitMQConsumers.WithSimStatusQueue(statusProducer, statusProducer)
|
|
}
|
|
if dailyProducer != nil {
|
|
rabbitMQConsumers.WithDailyUsageStart(dailyProducer)
|
|
}
|
|
if monthlyHistoryProducer != nil {
|
|
rabbitMQConsumers.WithMonthlyHistoryStart(monthlyHistoryProducer)
|
|
}
|
|
if imeiProducer != nil {
|
|
rabbitMQConsumers.WithImeiQueue(imeiProducer)
|
|
}
|
|
rabbitMQConsumers.WithDataReviewQuery(dataExecutor)
|
|
var (
|
|
fullSyncHTTP *httpapi.Server
|
|
fullSyncArchives []*fullsyncarchive.Service
|
|
)
|
|
{ // The authenticated control plane is always available.
|
|
prefixes, err := loadFullSyncArchivePrefixes(workerCtx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
stores := make([]*zset.Store, 0, len(prefixes))
|
|
for _, prefix := range prefixes {
|
|
store, e := zset.New(g.Redis(), prefix)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
stores = append(stores, store)
|
|
}
|
|
archiveRoot := g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.rootDir", "/app/logs/full-sync").String()
|
|
repositories := make([]*fullsyncruntime.Repository, 0, len(stores))
|
|
for _, store := range stores {
|
|
repo, e := fullsyncruntime.New(store, archiveRoot)
|
|
if e != nil {
|
|
return gerror.Wrap(e, "initialize full-sync runtime repository failed")
|
|
}
|
|
repositories = append(repositories, repo)
|
|
}
|
|
catalog, err := fullsyncruntime.NewCatalog(repositories...)
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize full-sync run catalog failed")
|
|
}
|
|
simple, err := fullsyncruntime.NewSimpleReplayWithDefault(g.Cfg().MustGet(workerCtx, "fullSyncOps.replay.maxRounds", fullsyncruntime.DefaultReplayMaxRounds).Int(), stores...)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// Only Admin exposes replay to users. This listener stays inside the
|
|
// Docker network; the Worker service must not publish port 18081.
|
|
fullSyncHTTP, err = httpapi.NewServer(httpapi.Config{Addr: ":18081", ShutdownTimeout: g.Cfg().MustGet(workerCtx, "fullSyncOps.control.shutdownTimeout", "10s").Duration()}, httpapi.Dependencies{Runs: catalog, List: catalog, Failures: catalog, Replays: repositories[0], Actions: repositories[0], Simple: simple})
|
|
if err != nil {
|
|
return gerror.Wrap(err, "initialize full-sync control server failed")
|
|
}
|
|
if err := fullSyncHTTP.Start(workerCtx); err != nil {
|
|
return gerror.Wrap(err, "start full-sync control server failed")
|
|
}
|
|
simple.Start(workerCtx)
|
|
}
|
|
if g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.enabled", false).Bool() {
|
|
prefixes, err := loadFullSyncArchivePrefixes(workerCtx)
|
|
if err != nil {
|
|
return gerror.Wrap(err, "load full-sync archive Redis prefixes failed")
|
|
}
|
|
for _, prefix := range prefixes {
|
|
store, err := zset.New(g.Redis(), prefix)
|
|
if err != nil {
|
|
return gerror.Wrapf(err, "initialize full-sync archive Redis store failed for prefix %q", prefix)
|
|
}
|
|
store.EnableArchiveIndex(true)
|
|
repo, err := fullsyncruntime.New(store, g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.rootDir", "/app/logs/full-sync").String())
|
|
if err != nil {
|
|
return gerror.Wrapf(err, "initialize full-sync archive repository failed for prefix %q", prefix)
|
|
}
|
|
service := fullsyncarchive.NewService(fullsyncarchive.Config{Root: g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.rootDir", "/app/logs/full-sync").String(), Interval: g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.pollInterval", "10s").Duration(), InitialBackoff: g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.retryMin", "5s").Duration(), MaxBackoff: g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.retryMax", "5m").Duration()}, repo, repo, repo)
|
|
if err := service.Start(workerCtx); err != nil {
|
|
return gerror.Wrapf(err, "start full-sync archive service failed for prefix %q", prefix)
|
|
}
|
|
fullSyncArchives = append(fullSyncArchives, service)
|
|
}
|
|
}
|
|
batchDispatcher.Register(consts.BatchTypeInbound, inboundExecutor)
|
|
batchDispatcher.Register(consts.BatchTypeAllocate, allocateExecutor)
|
|
batchDispatcher.Register(consts.BatchTypeCardAddPackage, addPackageCardExecutor)
|
|
batchDispatcher.Register(consts.BatchTypePoolAddPackage, addPackagePoolExecutor)
|
|
batchDispatcher.Register(consts.BatchTypePrepare, prepareExecutor)
|
|
batchDispatcher.Register(consts.BatchTypeRenew, renewExecutor)
|
|
batchDispatcher.Register(consts.BatchTypeChange, changeExecutor)
|
|
batchDispatcher.Register(consts.BatchTypeDataCut, dataExecutor)
|
|
batchDispatcher.Register(consts.BatchTypeDataStart, dataExecutor)
|
|
batchDispatcher.Register(consts.BatchTypeDataStop, dataExecutor)
|
|
// Resume durable APN sub-batches independently of RabbitMQ delivery and process restarts.
|
|
if reconciler, ok := dataExecutor.(interface{ RunReconciler(context.Context) error }); ok {
|
|
go func() {
|
|
if err := reconciler.RunReconciler(workerCtx); err != nil && workerCtx.Err() == nil {
|
|
g.Log().Errorf(workerCtx, "APN operation reconciler stopped: %v", err)
|
|
}
|
|
}()
|
|
}
|
|
batchDispatcher.Register(consts.BatchTypeSimCardUnsub, unsubExecutor)
|
|
batchDispatcher.Register(consts.BatchTypeSimCardRestart, restartExecutor)
|
|
batchDispatcher.Register(consts.BatchTypeSimCardCancel, cancelExecutor)
|
|
batchDispatcher.Register(consts.BatchTypeSimWarningConfig, warningConfigExecutor)
|
|
batchDispatcher.Register(consts.BatchTypeTrafficCutoffConfig, trafficCutoffConfigExecutor)
|
|
if err := rabbitMQConsumers.Start(workerCtx); err != nil {
|
|
_ = rabbitMQConsumers.Stop(context.Background())
|
|
return gerror.Wrap(err, "start RabbitMQ consumers failed")
|
|
}
|
|
|
|
stopRoleSync, err := rolesyncservice.Start(workerCtx)
|
|
if err != nil {
|
|
_ = rabbitMQConsumers.Stop(context.Background())
|
|
return gerror.Wrap(err, "start role sync Worker failed")
|
|
}
|
|
defer stopRoleSync()
|
|
|
|
g.Log().Info(workerCtx, "iotlcm worker started")
|
|
<-workerCtx.Done()
|
|
g.Log().Info(ctx, "iotlcm worker stopping")
|
|
stopRoleSync()
|
|
|
|
shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
|
|
defer cancel()
|
|
if err := rabbitMQConsumers.Stop(shutdownCtx); err != nil {
|
|
return gerror.Wrap(err, "stop RabbitMQ consumers failed")
|
|
}
|
|
for _, service := range fullSyncArchives {
|
|
if err := service.Stop(); err != nil {
|
|
return gerror.Wrap(err, "stop full-sync archive service failed")
|
|
}
|
|
}
|
|
if fullSyncHTTP != nil {
|
|
if err := fullSyncHTTP.Wait(); err != nil {
|
|
return gerror.Wrap(err, "stop full-sync control server failed")
|
|
}
|
|
}
|
|
if err := tdengine.Close(); err != nil {
|
|
return gerror.Wrap(err, "close TDengine failed")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func initialize(ctx context.Context) error {
|
|
if err := g.DB().PingMaster(); err != nil {
|
|
return gerror.Wrap(err, "connect database failed")
|
|
}
|
|
if err := gtime.SetTimeZone("Asia/Shanghai"); err != nil {
|
|
return gerror.Wrap(err, "set timezone failed")
|
|
}
|
|
|
|
mode := g.Cfg().MustGet(ctx, "system.gfmode", gmode.NOT_SET).String()
|
|
for _, supported := range []string{gmode.DEVELOP, gmode.TESTING, gmode.STAGING, gmode.PRODUCT} {
|
|
if mode == supported {
|
|
gmode.Set(mode)
|
|
break
|
|
}
|
|
}
|
|
return nil
|
|
}
|