/* Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments Copyright (C) ITsysCOM GmbH This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. You should have received a copy of the GNU Affero General Public License along with this program. If not, see */ package servmanager import ( "errors" "fmt" "sync" "github.com/cgrates/birpc/context" "github.com/cgrates/cgrates/config" "github.com/cgrates/cgrates/engine" "github.com/cgrates/cgrates/utils" ) // NewServiceManager returns a service manager func NewServiceManager(cfg *config.CGRConfig, shdChan *utils.SyncedChan, shdWg *sync.WaitGroup, connMgr *engine.ConnManager) *ServiceManager { sm := &ServiceManager{ cfg: cfg, subsystems: make(map[string]Service), shdChan: shdChan, shdWg: shdWg, connMgr: connMgr, } return sm } // ServiceManager handles service management ran by the engine type ServiceManager struct { sync.RWMutex // lock access to any shared data cfg *config.CGRConfig subsystems map[string]Service // active subsystems managed by SM shdChan *utils.SyncedChan shdWg *sync.WaitGroup connMgr *engine.ConnManager } // ArgStartService are passed to Start/StopService/Status RPC methods type ArgStartService struct { ServiceID string } // V1StartService starts a service with ID func (srvMngr *ServiceManager) V1StartService(ctx *context.Context, args ArgStartService, reply *string) (err error) { switch args.ServiceID { case utils.MetaScheduler: // stop the service using the config srvMngr.Lock() srvMngr.cfg.SchedulerCfg().Enabled = true srvMngr.Unlock() srvMngr.cfg.GetReloadChan(config.SCHEDULER_JSN) <- struct{}{} default: err = errors.New(utils.UnsupportedServiceIDCaps) } if err != nil { return err } *reply = utils.OK return } // V1StopService shuts-down a service with ID func (srvMngr *ServiceManager) V1StopService(ctx *context.Context, args ArgStartService, reply *string) (err error) { switch args.ServiceID { case utils.MetaScheduler: // stop the service using the config srvMngr.Lock() srvMngr.cfg.SchedulerCfg().Enabled = false srvMngr.Unlock() srvMngr.cfg.GetReloadChan(config.SCHEDULER_JSN) <- struct{}{} default: err = errors.New(utils.UnsupportedServiceIDCaps) } if err != nil { return err } *reply = utils.OK return } // V1ServiceStatus returns the service status func (srvMngr *ServiceManager) V1ServiceStatus(ctx *context.Context, args ArgStartService, reply *string) error { srvMngr.RLock() defer srvMngr.RUnlock() var running bool switch args.ServiceID { case utils.MetaScheduler: running = srvMngr.subsystems[utils.SchedulerS].IsRunning() default: return errors.New(utils.UnsupportedServiceIDCaps) } if running { *reply = utils.RunningCaps } else { *reply = utils.StoppedCaps } return nil } // GetConfig returns the Configuration func (srvMngr *ServiceManager) GetConfig() *config.CGRConfig { srvMngr.RLock() defer srvMngr.RUnlock() return srvMngr.cfg } // StartServices starts all enabled services func (srvMngr *ServiceManager) StartServices() (err error) { go srvMngr.handleReload() for _, service := range srvMngr.subsystems { if service.ShouldRun() && !service.IsRunning() { srvMngr.shdWg.Add(1) go func(srv Service) { if err := srv.Start(); err != nil && err != utils.ErrServiceAlreadyRunning { // in case the service was started in another gorutine utils.Logger.Err(fmt.Sprintf("<%s> failed to start %s because: %s", utils.ServiceManager, srv.ServiceName(), err)) srvMngr.shdChan.CloseOnce() } }(service) } } // startServer() return } // AddServices adds given services func (srvMngr *ServiceManager) AddServices(services ...Service) { srvMngr.Lock() for _, srv := range services { if _, has := srvMngr.subsystems[srv.ServiceName()]; !has { // do not rewrite the service srvMngr.subsystems[srv.ServiceName()] = srv } } srvMngr.Unlock() } func (srvMngr *ServiceManager) handleReload() { for { select { case <-srvMngr.shdChan.Done(): srvMngr.ShutdownServices() return case <-srvMngr.GetConfig().GetReloadChan(config.ATTRIBUTE_JSN): go srvMngr.reloadService(utils.AttributeS) case <-srvMngr.GetConfig().GetReloadChan(config.ChargerSCfgJson): go srvMngr.reloadService(utils.ChargerS) case <-srvMngr.GetConfig().GetReloadChan(config.THRESHOLDS_JSON): go srvMngr.reloadService(utils.ThresholdS) case <-srvMngr.GetConfig().GetReloadChan(config.STATS_JSON): go srvMngr.reloadService(utils.StatS) case <-srvMngr.GetConfig().GetReloadChan(config.TRENDS_JSON): go srvMngr.reloadService(utils.TrendS) case <-srvMngr.GetConfig().GetReloadChan(config.RANKINGS_JSON): go srvMngr.reloadService(utils.RankingS) case <-srvMngr.GetConfig().GetReloadChan(config.RESOURCES_JSON): go srvMngr.reloadService(utils.ResourceS) case <-srvMngr.GetConfig().GetReloadChan(config.RouteSJson): go srvMngr.reloadService(utils.RouteS) case <-srvMngr.GetConfig().GetReloadChan(config.SCHEDULER_JSN): go srvMngr.reloadService(utils.SchedulerS) case <-srvMngr.GetConfig().GetReloadChan(config.RALS_JSN): go srvMngr.reloadService(utils.RALService) case <-srvMngr.GetConfig().GetReloadChan(config.ApierS): go func() { srvMngr.reloadService(utils.APIerSv1) srvMngr.reloadService(utils.APIerSv2) }() case <-srvMngr.GetConfig().GetReloadChan(config.CDRS_JSN): go srvMngr.reloadService(utils.CDRServer) case <-srvMngr.GetConfig().GetReloadChan(config.SessionSJson): go srvMngr.reloadService(utils.SessionS) case <-srvMngr.GetConfig().GetReloadChan(config.ERsJson): go srvMngr.reloadService(utils.ERs) case <-srvMngr.GetConfig().GetReloadChan(config.DNSAgentJson): go srvMngr.reloadService(utils.DNSAgent) case <-srvMngr.GetConfig().GetReloadChan(config.FreeSWITCHAgentJSN): go srvMngr.reloadService(utils.FreeSWITCHAgent) case <-srvMngr.GetConfig().GetReloadChan(config.KamailioAgentJSN): go srvMngr.reloadService(utils.KamailioAgent) case <-srvMngr.GetConfig().GetReloadChan(config.AsteriskAgentJSN): go srvMngr.reloadService(utils.AsteriskAgent) case <-srvMngr.GetConfig().GetReloadChan(config.RA_JSN): go srvMngr.reloadService(utils.RadiusAgent) case <-srvMngr.GetConfig().GetReloadChan(config.DA_JSN): go srvMngr.reloadService(utils.DiameterAgent) case <-srvMngr.GetConfig().GetReloadChan(config.HttpAgentJson): go srvMngr.reloadService(utils.HTTPAgent) case <-srvMngr.GetConfig().GetReloadChan(config.PrometheusAgentJSON): go srvMngr.reloadService(utils.PrometheusAgent) case <-srvMngr.GetConfig().GetReloadChan(config.AnalyzerCfgJson): go srvMngr.reloadService(utils.AnalyzerS) case <-srvMngr.GetConfig().GetReloadChan(config.DispatcherSJson): go srvMngr.reloadService(utils.DispatcherS) case <-srvMngr.GetConfig().GetReloadChan(config.DATADB_JSN): go srvMngr.reloadService(utils.DataDB) case <-srvMngr.GetConfig().GetReloadChan(config.STORDB_JSN): go srvMngr.reloadService(utils.StorDB) case <-srvMngr.GetConfig().GetReloadChan(config.EEsJson): go srvMngr.reloadService(utils.EEs) case <-srvMngr.GetConfig().GetReloadChan(config.RPCConnsJsonName): go srvMngr.connMgr.Reload() case <-srvMngr.GetConfig().GetReloadChan(config.SIPAgentJson): go srvMngr.reloadService(utils.SIPAgent) case <-srvMngr.GetConfig().GetReloadChan(config.RegistrarCJson): go srvMngr.reloadService(utils.RegistrarC) case <-srvMngr.GetConfig().GetReloadChan(config.HTTP_JSN): go srvMngr.reloadService(utils.GlobalVarS) case <-srvMngr.GetConfig().GetReloadChan(config.CoreSCfgJson): go srvMngr.reloadService(utils.CoreS) case <-srvMngr.GetConfig().GetReloadChan(config.JanusAgentJson): go srvMngr.reloadService(utils.JanusAgent) case <-srvMngr.GetConfig().GetReloadChan(config.IPsJSON): // go srvMngr.reloadService(utils.IPs) } // handle RPC server } } func (srvMngr *ServiceManager) reloadService(srviceName string) (err error) { srv := srvMngr.GetService(srviceName) if srv.ShouldRun() { if srv.IsRunning() { if err = srv.Reload(); err != nil { utils.Logger.Err(fmt.Sprintf("<%s> failed to reload <%s> err <%s>", utils.ServiceManager, srv.ServiceName(), err)) srvMngr.shdChan.CloseOnce() return // stop if we encounter an error } } else { srvMngr.shdWg.Add(1) if err = srv.Start(); err != nil { utils.Logger.Err(fmt.Sprintf("<%s> failed to start <%s> err <%s>", utils.ServiceManager, srv.ServiceName(), err)) srvMngr.shdChan.CloseOnce() return // stop if we encounter an error } } } else if srv.IsRunning() { if err = srv.Shutdown(); err != nil { utils.Logger.Err(fmt.Sprintf("<%s> failed to stop service <%s> err <%s>", utils.ServiceManager, srv.ServiceName(), err)) srvMngr.shdChan.CloseOnce() } srvMngr.shdWg.Done() } return } // GetService returns the named service func (srvMngr *ServiceManager) GetService(subsystem string) (srv Service) { srvMngr.RLock() srv = srvMngr.subsystems[subsystem] srvMngr.RUnlock() return } // ShutdownServices will stop all services func (srvMngr *ServiceManager) ShutdownServices() { for _, srv := range srvMngr.subsystems { // gracefully stop all running subsystems if srv.IsRunning() { go func(srv Service) { if err := srv.Shutdown(); err != nil { utils.Logger.Err(fmt.Sprintf("<%s> Failed to shutdown subsystem <%s> because: %s", utils.ServiceManager, srv.ServiceName(), err)) } srvMngr.shdWg.Done() }(srv) } } } // Service interface that describes what functions should a service implement type Service interface { // Start should handle the service start Start() error // Reload handles the change of config Reload() error // Shutdown stops the service Shutdown() error // IsRunning returns if the service is running IsRunning() bool // ShouldRun returns if the service should be running ShouldRun() bool // ServiceName returns the service name ServiceName() string }