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

Compare commits

base: core:feature/event
Branches Tags
core:main core:feature/event core:demo-alpr core:bugfix/fix-exec-scheduler core:feature/order core:feature/payment core:chart core:execute-monitor core:feature/schedule_executions core:feature/dockerfile core:feature/create_argo
..
compare: core:main
Branches Tags
core:main core:feature/event core:demo-alpr core:bugfix/fix-exec-scheduler core:feature/order core:feature/payment core:chart core:execute-monitor core:feature/schedule_executions core:feature/dockerfile core:feature/create_argo

7 Commits
feature/ev ... main

Author SHA1 Message Date
mr
2a76f2b9fb new lib 2026-02-03 16:13:48 +01:00
mr
07ca18e347 Conf Oclib Package Lightest 2026-02-03 16:12:57 +01:00
mr
c3f769ccd5 oclib-debug 2026-02-03 09:43:13 +01:00
mr
26e5db4572 oclib debug 2026-02-03 08:48:31 +01:00
mr
937c811e8d TODo 2026-02-02 14:42:34 +01:00
mr
a9c7bb66ee Merge branch 'feature/event' 2026-02-02 14:36:32 +01:00
mr
1fda2b1334 add cluster name 2026-01-20 11:14:22 +01:00
7 changed files with 60 additions and 71 deletions
Expand all files Collapse all files

2
Makefile
View File

@@ -17,7 +17,7 @@ docker:
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:
kind load docker-image oc/oc-schedulerd:0.0.1 --name opencloud | true kind load docker-image oc/oc-schedulerd:0.0.1 --name $(CLUSTER_NAME) | true
publish-registry: publish-registry:
@echo "TODO" @echo "TODO"

20
conf/conf.go
View File

@@ -8,18 +8,14 @@ import (
) )
type Config struct { type Config struct {
MonitorPath string MonitorPath string
MongoUrl string Logs string
DBName string Mode string
Logs string KubeHost string
LokiUrl string KubePort string
NatsUrl string KubeCA string
Mode string KubeCert string
KubeHost string KubeData string
KubePort string
KubeCA string
KubeCert string
KubeData string
} }
var instance *Config var instance *Config

61
daemons/execute_monitor_container.go
View File

@@ -8,29 +8,30 @@ import (
"net/http" "net/http"
"oc-schedulerd/conf" "oc-schedulerd/conf"
oclib "cloud.o-forge.io/core/oc-lib"
"github.com/rs/zerolog" "github.com/rs/zerolog"
) )
type ContainerMonitor struct { type ContainerMonitor struct {
Monitor LocalMonitor Monitor LocalMonitor
KubeCA string KubeCA string
KubeCert string KubeCert string
KubeData string KubeData string
KubeHost string KubeHost string
KubePort string KubePort string
} }
func NewContainerMonitor(UUID string, peerId string, duration int) (Executor){ func NewContainerMonitor(UUID string, peerId string, duration int) Executor {
return &ContainerMonitor{ return &ContainerMonitor{
Monitor: LocalMonitor{ Monitor: LocalMonitor{
ExecutionID: UUID, ExecutionID: UUID,
PeerID: peerId, PeerID: peerId,
Duration: duration, Duration: duration,
LokiUrl: conf.GetConfig().LokiUrl, LokiUrl: oclib.GetConfig().LokiUrl,
MongoUrl: conf.GetConfig().MongoUrl, MongoUrl: oclib.GetConfig().MongoUrl,
DBName: conf.GetConfig().DBName, DBName: oclib.GetConfig().MongoDatabase,
}, },
KubeCA: conf.GetConfig().KubeCA, KubeCA: conf.GetConfig().KubeCA,
KubeCert: conf.GetConfig().KubeCert, KubeCert: conf.GetConfig().KubeCert,
KubeData: conf.GetConfig().KubeData, KubeData: conf.GetConfig().KubeData,
KubeHost: conf.GetConfig().KubeHost, KubeHost: conf.GetConfig().KubeHost,
@@ -38,7 +39,7 @@ func NewContainerMonitor(UUID string, peerId string, duration int) (Executor){
} }
} }
func (cm *ContainerMonitor) PrepareMonitorExec() []string { func (cm *ContainerMonitor) PrepareMonitorExec() []string {
args := []string{ args := []string{
"-e", cm.Monitor.ExecutionID, "-e", cm.Monitor.ExecutionID,
@@ -49,7 +50,7 @@ func (cm *ContainerMonitor) PrepareMonitorExec() []string {
"-M", "kubernetes", "-M", "kubernetes",
"-H", cm.KubeHost, "-H", cm.KubeHost,
"-P", cm.KubePort, "-P", cm.KubePort,
"-C", cm.KubeCert, "-C", cm.KubeCert,
"-D", cm.KubeData, "-D", cm.KubeData,
"-c", cm.KubeCA, "-c", cm.KubeCA,
} }
@@ -58,8 +59,8 @@ func (cm *ContainerMonitor) PrepareMonitorExec() []string {
args = append(args, "-t", fmt.Sprintf("%d", cm.Monitor.Duration)) args = append(args, "-t", fmt.Sprintf("%d", cm.Monitor.Duration))
} }
return args return args
} }
// Contact the docker's API at the KubeHost's URL to : // Contact the docker's API at the KubeHost's URL to :
@@ -67,8 +68,8 @@ func (cm *ContainerMonitor) PrepareMonitorExec() []string {
// - Create the container // - Create the container
// - Start the container // - Start the container
func (cm *ContainerMonitor) LaunchMonitor(args []string, l zerolog.Logger) { func (cm *ContainerMonitor) LaunchMonitor(args []string, l zerolog.Logger) {
var containerID string var containerID string
imageName := "oc-monitord" imageName := "oc-monitord"
url := "http://" + cm.KubeHost + ":2375" url := "http://" + cm.KubeHost + ":2375"
@@ -81,15 +82,15 @@ func (cm *ContainerMonitor) LaunchMonitor(args []string, l zerolog.Logger) {
d, _ := io.ReadAll(resp.Body) d, _ := io.ReadAll(resp.Body)
l.Fatal().Msg("Couldn't find the oc-monitord image : " + string(d)) l.Fatal().Msg("Couldn't find the oc-monitord image : " + string(d))
} }
dataCreation := map[string]interface{}{"Image": imageName, "Cmd" : args} dataCreation := map[string]interface{}{"Image": imageName, "Cmd": args}
byteData, err := json.Marshal(dataCreation) byteData, err := json.Marshal(dataCreation)
if err != nil { if err != nil {
l.Fatal().Msg("Error when contacting the creating request body : " + err.Error()) l.Fatal().Msg("Error when contacting the creating request body : " + err.Error())
} }
r, _ := http.NewRequest("POST",url + "/containers/create", bytes.NewBuffer(byteData)) r, _ := http.NewRequest("POST", url+"/containers/create", bytes.NewBuffer(byteData))
r.Header.Add("Content-Type","application/json") r.Header.Add("Content-Type", "application/json")
resp, err = http.DefaultClient.Do(r) resp, err = http.DefaultClient.Do(r)
if err != nil { if err != nil {
l.Fatal().Msg("Error when contacting the docker API on " + url + ": " + err.Error()) l.Fatal().Msg("Error when contacting the docker API on " + url + ": " + err.Error())
@@ -116,9 +117,9 @@ func (cm *ContainerMonitor) LaunchMonitor(args []string, l zerolog.Logger) {
networkName := "oc" networkName := "oc"
dataNetwork, _ := json.Marshal(map[string]string{"Container" : containerID}) dataNetwork, _ := json.Marshal(map[string]string{"Container": containerID})
r, _ = http.NewRequest("POST",url + "/networks/" + networkName + "/connect", bytes.NewBuffer(dataNetwork)) r, _ = http.NewRequest("POST", url+"/networks/"+networkName+"/connect", bytes.NewBuffer(dataNetwork))
r.Header.Add("Content-Type","application/json") r.Header.Add("Content-Type", "application/json")
resp, err = http.DefaultClient.Do(r) resp, err = http.DefaultClient.Do(r)
if err != nil { if err != nil {
l.Fatal().Msg("Error when contacting the docker API on " + url + ": " + err.Error()) l.Fatal().Msg("Error when contacting the docker API on " + url + ": " + err.Error())
@@ -129,7 +130,7 @@ func (cm *ContainerMonitor) LaunchMonitor(args []string, l zerolog.Logger) {
return return
} }
resp, err = http.Post( url + "/containers/" + containerID + "/start", "", nil) resp, err = http.Post(url+"/containers/"+containerID+"/start", "", nil)
if err != nil { if err != nil {
l.Fatal().Msg("Error when contacting the docker API on " + url + ": " + err.Error()) l.Fatal().Msg("Error when contacting the docker API on " + url + ": " + err.Error())
} }
@@ -143,4 +144,4 @@ func (cm *ContainerMonitor) LaunchMonitor(args []string, l zerolog.Logger) {
// we can add logging with GET /containers/id/logs?stdout=true&follow=true // we can add logging with GET /containers/id/logs?stdout=true&follow=true
// logExecution(stdoutMonitord, l) // logExecution(stdoutMonitord, l)
} }

24
daemons/execute_monitor_local.go
View File

@@ -5,6 +5,7 @@ import (
"oc-schedulerd/conf" "oc-schedulerd/conf"
"os/exec" "os/exec"
oclib "cloud.o-forge.io/core/oc-lib"
"github.com/rs/zerolog" "github.com/rs/zerolog"
) )
@@ -12,20 +13,19 @@ type LocalMonitor struct {
ExecutionID string ExecutionID string
PeerID string PeerID string
Duration int Duration int
LokiUrl string LokiUrl string
MongoUrl string MongoUrl string
DBName string DBName string
} }
func NewLocalMonitor(UUID string, peerId string, duration int) (Executor){ func NewLocalMonitor(UUID string, peerId string, duration int) Executor {
return &LocalMonitor{ return &LocalMonitor{
ExecutionID: UUID, ExecutionID: UUID,
PeerID: peerId, PeerID: peerId,
Duration: duration, Duration: duration,
LokiUrl: conf.GetConfig().LokiUrl, LokiUrl: oclib.GetConfig().LokiUrl,
MongoUrl: conf.GetConfig().MongoUrl, MongoUrl: oclib.GetConfig().MongoUrl,
DBName: conf.GetConfig().DBName, DBName: oclib.GetConfig().MongoDatabase,
} }
} }
@@ -46,7 +46,6 @@ func (lm *LocalMonitor) PrepareMonitorExec() []string {
"-d", lm.DBName, "-d", lm.DBName,
} }
if lm.Duration > 0 { if lm.Duration > 0 {
args = append(args, "-t", fmt.Sprintf("%d", lm.Duration)) args = append(args, "-t", fmt.Sprintf("%d", lm.Duration))
} }
@@ -54,7 +53,7 @@ func (lm *LocalMonitor) PrepareMonitorExec() []string {
return args return args
} }
func (lm *LocalMonitor) LaunchMonitor(args []string, l zerolog.Logger) { func (lm *LocalMonitor) LaunchMonitor(args []string, l zerolog.Logger) {
cmd := exec.Command(conf.GetConfig().MonitorPath, args...) cmd := exec.Command(conf.GetConfig().MonitorPath, args...)
fmt.Printf("Command : %v\n", cmd) fmt.Printf("Command : %v\n", cmd)
@@ -70,4 +69,3 @@ func (lm *LocalMonitor) LaunchMonitor(args []string, l zerolog.Logger) {
logExecution(stdoutMonitord, l) logExecution(stdoutMonitord, l)
} }

2
go.mod
View File

@@ -5,7 +5,7 @@ go 1.23.0
toolchain go1.24.0 toolchain go1.24.0
require ( require (
cloud.o-forge.io/core/oc-lib v0.0.0-20260129122033-186ba3e689c7 cloud.o-forge.io/core/oc-lib v0.0.0-20260203150531-ef916fe2d995
github.com/beego/beego v1.12.12 github.com/beego/beego v1.12.12
github.com/google/uuid v1.6.0 github.com/google/uuid v1.6.0
github.com/goraz/onion v0.1.3 github.com/goraz/onion v0.1.3

6
go.sum
View File

@@ -14,6 +14,12 @@ cloud.o-forge.io/core/oc-lib v0.0.0-20260127143728-3c052bf16572 h1:jrUHgs4DqNWLn
cloud.o-forge.io/core/oc-lib v0.0.0-20260127143728-3c052bf16572/go.mod h1:vHWauJsS6ryf7UDqq8hRXoYD5RsONxcFTxeZPOztEuI= cloud.o-forge.io/core/oc-lib v0.0.0-20260127143728-3c052bf16572/go.mod h1:vHWauJsS6ryf7UDqq8hRXoYD5RsONxcFTxeZPOztEuI=
cloud.o-forge.io/core/oc-lib v0.0.0-20260129122033-186ba3e689c7 h1:NRFGRqN+j5g3DrtXMYN5T5XSYICG+OU2DisjBdID3j8= cloud.o-forge.io/core/oc-lib v0.0.0-20260129122033-186ba3e689c7 h1:NRFGRqN+j5g3DrtXMYN5T5XSYICG+OU2DisjBdID3j8=
cloud.o-forge.io/core/oc-lib v0.0.0-20260129122033-186ba3e689c7/go.mod h1:vHWauJsS6ryf7UDqq8hRXoYD5RsONxcFTxeZPOztEuI= cloud.o-forge.io/core/oc-lib v0.0.0-20260129122033-186ba3e689c7/go.mod h1:vHWauJsS6ryf7UDqq8hRXoYD5RsONxcFTxeZPOztEuI=
cloud.o-forge.io/core/oc-lib v0.0.0-20260203074447-30e6c9a6183c h1:c19lIseiUk5Hp+06EowfEbMWH1pK8AC/hvQ4ryWgJtY=
cloud.o-forge.io/core/oc-lib v0.0.0-20260203074447-30e6c9a6183c/go.mod h1:vHWauJsS6ryf7UDqq8hRXoYD5RsONxcFTxeZPOztEuI=
cloud.o-forge.io/core/oc-lib v0.0.0-20260203083753-4f28b9b589d6 h1:N+0xkioACl3PNo+MquCIIOL/kSICevg340IYOFGQeOw=
cloud.o-forge.io/core/oc-lib v0.0.0-20260203083753-4f28b9b589d6/go.mod h1:vHWauJsS6ryf7UDqq8hRXoYD5RsONxcFTxeZPOztEuI=
cloud.o-forge.io/core/oc-lib v0.0.0-20260203150531-ef916fe2d995 h1:ZDRvnzTTNHgMm5hYmseHdEPqQ6rn/4v+P9f/JIxPaNw=
cloud.o-forge.io/core/oc-lib v0.0.0-20260203150531-ef916fe2d995/go.mod h1:T0UCxRd8w+qCVVC0NEyDiWIGC5ADwEbQ7hFcvftd4Ks=
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
github.com/Knetic/govaluate v3.0.0+incompatible/go.mod h1:r7JcOSlj0wfOMncg0iLm8Leh48TZaKVeNIfJntJ2wa0= github.com/Knetic/govaluate v3.0.0+incompatible/go.mod h1:r7JcOSlj0wfOMncg0iLm8Leh48TZaKVeNIfJntJ2wa0=
github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=

16
main.go
View File

@@ -14,18 +14,6 @@ func main() {
l := oclib.GetLogger() l := oclib.GetLogger()
o := oclib.GetConfLoader() o := oclib.GetConfLoader()
c := oclib.SetConfig(
o.GetStringDefault("MONGO_URL", "mongodb://127.0.0.1:27017"),
o.GetStringDefault("MONGO_DATABASE", "DC_myDC"),
o.GetStringDefault("NATS_URL", "nats://localhost:4222"),
o.GetStringDefault("LOKI_URL", ""),
o.GetStringDefault("LOG_LEVEL", "info"),
)
conf.GetConfig().DBName = c.MongoDatabase
conf.GetConfig().MongoUrl = c.MongoUrl
conf.GetConfig().NatsUrl = c.NATSUrl
conf.GetConfig().LokiUrl = c.LokiUrl
conf.GetConfig().Mode = o.GetStringDefault("MODE", "") conf.GetConfig().Mode = o.GetStringDefault("MODE", "")
if conf.GetConfig().Mode == "container" { if conf.GetConfig().Mode == "container" {
@@ -57,9 +45,9 @@ func main() {
exe_mngr := daemons.ExecutionManager{} exe_mngr := daemons.ExecutionManager{}
go tools.NewNATSCaller().ListenNats(map[tools.NATSMethod]func(tools.NATSResponse){ go tools.NewNATSCaller().ListenNats(map[tools.NATSMethod]func(tools.NATSResponse){
tools.CREATE_EXECTUTION: sch_mngr.GetNextScheduledWorkflows, tools.CREATE_EXECTUTION: sch_mngr.GetNextScheduledWorkflows, // TODO: unused for now...
tools.WORKFLOW_EVENT: sch_mngr.ExecuteWorkflow, tools.WORKFLOW_EVENT: sch_mngr.ExecuteWorkflow,
tools.REMOVE_EXECUTION: daemons.Executions.DeleteSchedules, tools.REMOVE_EXECUTION: daemons.Executions.DeleteSchedules, // TODO: unused for now...
}) })
go sch_mngr.SchedulePolling() go sch_mngr.SchedulePolling()
exe_mngr.RetrieveNextExecutions() exe_mngr.RetrieveNextExecutions()