core/oc-schedulerd
6
0
Fork 0
Code Issues 7 Pull Requests Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
mr
2025-11-20 16:30:58 +01:00
parent 7717a02f7f
commit 8bbbd1dcfb
3 changed files with 22 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Makefile
View File

@@ -13,7 +13,7 @@ clean:
rm -rf oc-schedulerd rm -rf oc-schedulerd
docker: docker:
DOCKER_BUILDKIT=1 docker build -t oc-schedulerd --build-arg MONITORD_IMAGE=oc-monitord -f Dockerfile . --build-arg=HOST=$(HOST) DOCKER_BUILDKIT=1 docker build -t oc-schedulerd --build-arg MONITORD_IMAGE=oc-monitord -f Dockerfile . --build-arg=HOST=$(HOST) --build-arg=KUBERNETES_HOST=$(KUBERNETES_HOST) --build-arg=KUBERNETES_SERVICE_PORT=$(KUBERNETES_SERVICE_PORT) --build-arg=KUBE_CA=$(KUBE_CA) --build-arg=KUBE_CERT=$(KUBE_CERT) --build-arg=KUBE_DATA=$(KUBE_DATA)
docker tag oc-schedulerd:latest oc/oc-schedulerd:0.0.1 docker tag oc-schedulerd:latest oc/oc-schedulerd:0.0.1
publish-kind: publish-kind:

6
daemons/execution_manager.go
View File

@@ -26,7 +26,7 @@ func (em *ExecutionManager) RetrieveNextExecutions() {
if exec.ExecDate.Before(time.Now().UTC()) { if exec.ExecDate.Before(time.Now().UTC()) {
logger.Info().Msg("Will execute " + execId + " soon") logger.Info().Msg("Will execute " + execId + " soon")
go em.executeExecution(&exec) go em.executeExecution(&exec)
delete(executions,execId) delete(executions, execId)
} }
} }
} }
@@ -60,7 +60,7 @@ func (em *ExecutionManager) executeExecution(execution *workflow_execution.Workf
} }
args := executor.PrepareMonitorExec() args := executor.PrepareMonitorExec()
executor.LaunchMonitor(args,logger) executor.LaunchMonitor(args, logger)
// if exec_method == "k8s" { // if exec_method == "k8s" {
// logger.Error().Msg("TODO : executing oc-monitor in a k8s") // logger.Error().Msg("TODO : executing oc-monitor in a k8s")
@@ -76,7 +76,7 @@ func (em *ExecutionManager) executeExecution(execution *workflow_execution.Workf
// ExecutionID: Execution.ExecutionsID, // ExecutionID: Execution.ExecutionsID,
// PeerID: Execution.CreatorID, // PeerID: Execution.CreatorID,
// LokiUrl: conf.GetConfig().LokiUrl, // LokiUrl: conf.GetConfig().LokiUrl,
// } // }
// monitor.LaunchLocalMonitor() // monitor.LaunchLocalMonitor()
// } // }

33
daemons/schedule_manager.go
View File

@@ -24,7 +24,7 @@ type ScheduledExecution struct {
func (sb *ScheduledExecution) DeleteSchedules(uuid string) { func (sb *ScheduledExecution) DeleteSchedules(uuid string) {
Executions.Mu.Lock() Executions.Mu.Lock()
defer Executions.Mu.Unlock() defer Executions.Mu.Unlock()
delete(sb.Execs,uuid) delete(sb.Execs, uuid)
} }
func (sb *ScheduledExecution) AddSchedules(new_executions []*workflow_execution.WorkflowExecution, logger zerolog.Logger) { func (sb *ScheduledExecution) AddSchedules(new_executions []*workflow_execution.WorkflowExecution, logger zerolog.Logger) {
@@ -39,10 +39,10 @@ func (sb *ScheduledExecution) AddSchedules(new_executions []*workflow_execution.
} }
func (sb *ScheduledExecution) execIsSet(exec *workflow_execution.WorkflowExecution) bool { func (sb *ScheduledExecution) execIsSet(exec *workflow_execution.WorkflowExecution) bool {
if _, ok := sb.Execs[exec.UUID]; ok{ if _, ok := sb.Execs[exec.UUID]; ok {
return true return true
} }
return false return false
} }
@@ -58,18 +58,21 @@ type ScheduleManager struct {
// on workflows' scheduling. Messages must contain // on workflows' scheduling. Messages must contain
// workflow execution ID, to allow retrieval of execution infos // workflow execution ID, to allow retrieval of execution infos
func (s *ScheduleManager) ListenNATS() { func (s *ScheduleManager) ListenNATS() {
nc, err := nats.Connect(oclib.GetConfig().NATSUrl) for {
if err != nil { nc, err := nats.Connect(oclib.GetConfig().NATSUrl)
s.Logger.Error().Msg("Could not connect to NATS") if err != nil {
return s.Logger.Error().Msg("Could not connect to NATS")
time.Sleep(10 * time.Second)
continue
}
defer nc.Close()
var wg sync.WaitGroup
wg.Add(2)
go s.listenForChange(nc, tools.REMOVE.GenerateKey(oclib.WORKFLOW.String()), true, wg)
go s.listenForChange(nc, tools.CREATE.GenerateKey(oclib.WORKFLOW.String()), false, wg)
wg.Wait()
break
} }
defer nc.Close()
var wg sync.WaitGroup
wg.Add(2)
go s.listenForChange(nc, tools.REMOVE.GenerateKey(oclib.WORKFLOW.String()), true, wg)
go s.listenForChange(nc, tools.CREATE.GenerateKey(oclib.WORKFLOW.String()), false, wg)
wg.Wait()
} }
// Goroutine listening to a NATS server for updates // Goroutine listening to a NATS server for updates
@@ -121,7 +124,7 @@ func (s *ScheduleManager) getExecution(from time.Time, to time.Time) (exec_list
} }
res := oclib.NewRequest(oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION), "", "", []string{}, nil).Search(&f, "", false) res := oclib.NewRequest(oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION), "", "", []string{}, nil).Search(&f, "", false)
if res.Code != 200 { if res.Code != 200 {
s.Logger.Error().Msg("Error loading") s.Logger.Error().Msg("Error loading " + res.Err)
return return
} }
for _, exec := range res.Data { for _, exec := range res.Data {