diff --git a/controllers/booking.go b/controllers/booking.go index 1a0ca66..d1e57f4 100644 --- a/controllers/booking.go +++ b/controllers/booking.go @@ -67,8 +67,8 @@ func (o *BookingController) Search() { */ // store and return Id or post with UUID user, peerID, groups := oclib.ExtractTokenInfo(*o.Ctx.Request) - start_date, _ := time.Parse("2006-01-02", o.Ctx.Input.Param(":start_date")) - end_date, _ := time.Parse("2006-01-02", o.Ctx.Input.Param(":end_date")) + start_date, _ := time.ParseInLocation("2006-01-02", o.Ctx.Input.Param(":start_date"), time.UTC) + end_date, _ := time.ParseInLocation("2006-01-02", o.Ctx.Input.Param(":end_date"), time.UTC) isDraft := o.Ctx.Input.Query("is_draft") sd := primitive.NewDateTimeFromTime(start_date) ed := primitive.NewDateTimeFromTime(end_date) diff --git a/controllers/workflow_execution.go b/controllers/workflow_execution.go index d42b626..e7b0f77 100644 --- a/controllers/workflow_execution.go +++ b/controllers/workflow_execution.go @@ -35,8 +35,8 @@ func (o *WorkflowExecutionController) SearchPerDate() { */ // user, peerID, groups := oclib.ExtractTokenInfo(*o.Ctx.Request) // store and return Id or post with UUID - start_date, _ := time.Parse("2006-01-02", o.Ctx.Input.Param(":start_date")) - end_date, _ := time.Parse("2006-01-02", o.Ctx.Input.Param(":end_date")) + start_date, _ := time.ParseInLocation("2006-01-02", o.Ctx.Input.Param(":start_date"), time.UTC) + end_date, _ := time.ParseInLocation("2006-01-02", o.Ctx.Input.Param(":end_date"), time.UTC) sd := primitive.NewDateTimeFromTime(start_date) ed := primitive.NewDateTimeFromTime(end_date) f := dbs.Filters{ diff --git a/docker_scheduler.json b/docker_scheduler.json index 478989d..8d407cb 100644 --- a/docker_scheduler.json +++ b/docker_scheduler.json @@ -2,5 +2,12 @@ "MONGO_URL":"mongodb://mongo:27017/", "NATS_URL":"nats://nats:4222", "MONGO_DATABASE":"DC_myDC", - "LOKI_URL": "http://loki:3100" + "LOKI_URL": "http://loki:3100", + "KUBERNETES_SERVICE_HOST": "kubernetes.default.svc.cluster.local", + "KUBERNETES_SERVICE_PORT": "6443", + "KUBERNETES_NAMESPACE": "default", + "KUBERNETES_IMAGE": "opencloudregistry/oc-monitord", + "KUBE_CA": "LS0tLS1CRUdJTiBDRVJUSUZJQ0FURS0tLS0tCk1JSUJkakNDQVIyZ0F3SUJBZ0lCQURBS0JnZ3Foa2pPUFFRREFqQWpNU0V3SHdZRFZRUUREQmhyTTNNdGMyVnkKZG1WeUxXTmhRREUzTnpNeE1qY3dPVFl3SGhjTk1qWXdNekV3TURjeE9ERTJXaGNOTXpZd016QTNNRGN4T0RFMgpXakFqTVNFd0h3WURWUVFEREJock0zTXRjMlZ5ZG1WeUxXTmhRREUzTnpNeE1qY3dPVFl3V1RBVEJnY3Foa2pPClBRSUJCZ2dxaGtqT1BRTUJCd05DQUFReG81cXQ0MGxEekczRHJKTE1wRVBrd0ZBY1FmbC8vVE1iWjZzemMreHAKbmVzVzRTSTdXK1lWdFpRYklmV2xBMTRaazQvRFlDMHc1YlgxZU94RVVuL0pvMEl3UURBT0JnTlZIUThCQWY4RQpCQU1DQXFRd0R3WURWUjBUQVFIL0JBVXdBd0VCL3pBZEJnTlZIUTRFRmdRVXBLM2pGK25IRlZSbDcwb3ZRVGZnCmZabGNQZE13Q2dZSUtvWkl6ajBFQXdJRFJ3QXdSQUlnVnkyaUx0Y0xaYm1vTnVoVHdKbU5sWlo3RVlBYjJKNW0KSjJYbG1UbVF5a2tDSUhLbzczaDBkdEtUZTlSa0NXYTJNdStkS1FzOXRFU0tBV0x1emlnYXBHYysKLS0tLS1FTkQgQ0VSVElGSUNBVEUtLS0tLQo=", + "KUBE_CERT": "LS0tLS1CRUdJTiBDRVJUSUZJQ0FURS0tLS0tCk1JSUJrakNDQVRlZ0F3SUJBZ0lJQUkvSUg2R2Rodm93Q2dZSUtvWkl6ajBFQXdJd0l6RWhNQjhHQTFVRUF3d1kKYXpOekxXTnNhV1Z1ZEMxallVQXhOemN6TVRJM01EazJNQjRYRFRJMk1ETXhNREEzTVRneE5sb1hEVEkzTURNeApNREEzTVRneE5sb3dNREVYTUJVR0ExVUVDaE1PYzNsemRHVnRPbTFoYzNSbGNuTXhGVEFUQmdOVkJBTVRESE41CmMzUmxiVHBoWkcxcGJqQlpNQk1HQnlxR1NNNDlBZ0VHQ0NxR1NNNDlBd0VIQTBJQUJQTTdBVEZQSmFMMjUrdzAKUU1vZUIxV2hBRW4vWnViM0tSRERrYnowOFhwQWJ2akVpdmdnTkdpdG4wVmVsaEZHamRmNHpBT29Nd1J3M21kbgpYSGtHVDB5alNEQkdNQTRHQTFVZER3RUIvd1FFQXdJRm9EQVRCZ05WSFNVRUREQUtCZ2dyQmdFRkJRY0RBakFmCkJnTlZIU01FR0RBV2dCUVZLOThaMEMxcFFyVFJSMGVLZHhIa2o0ejFJREFLQmdncWhrak9QUVFEQWdOSkFEQkcKQWlFQXZYWll6Zk9iSUtlWTRtclNsRmt4ZS80a0E4K01ieDc1UDFKRmNlRS8xdGNDSVFDNnM0ZXlZclhQYmNWSgpxZm5EamkrZ1RacGttN0tWSTZTYTlZN2FSRGFabUE9PQotLS0tLUVORCBDRVJUSUZJQ0FURS0tLS0tCi0tLS0tQkVHSU4gQ0VSVElGSUNBVEUtLS0tLQpNSUlCZURDQ0FSMmdBd0lCQWdJQkFEQUtCZ2dxaGtqT1BRUURBakFqTVNFd0h3WURWUVFEREJock0zTXRZMnhwClpXNTBMV05oUURFM056TXhNamN3T1RZd0hoY05Nall3TXpFd01EY3hPREUyV2hjTk16WXdNekEzTURjeE9ERTIKV2pBak1TRXdId1lEVlFRRERCaHJNM010WTJ4cFpXNTBMV05oUURFM056TXhNamN3T1RZd1dUQVRCZ2NxaGtqTwpQUUlCQmdncWhrak9QUU1CQndOQ0FBUzV1NGVJbStvVnV1SFI0aTZIOU1kVzlyUHdJbFVPNFhIMEJWaDRUTGNlCkNkMnRBbFVXUW5FakxMdlpDWlVaYTlzTlhKOUVtWWt5S0dtQWR2TE9FbUVrbzBJd1FEQU9CZ05WSFE4QkFmOEUKQkFNQ0FxUXdEd1lEVlIwVEFRSC9CQVV3QXdFQi96QWRCZ05WSFE0RUZnUVVGU3ZmR2RBdGFVSzAwVWRIaW5jUgo1SStNOVNBd0NnWUlLb1pJemowRUF3SURTUUF3UmdJaEFMY2xtQnR4TnpSVlBvV2hoVEVKSkM1Z3VNSGsvcFZpCjFvYXJ2UVJxTWRKcUFpRUEyR1dNTzlhZFFYTEQwbFZKdHZMVkc1M3I0M0lxMHpEUUQwbTExMVZyL1MwPQotLS0tLUVORCBDRVJUSUZJQ0FURS0tLS0tCg==", + "KUBE_DATA": "LS0tLS1CRUdJTiBFQyBQUklWQVRFIEtFWS0tLS0tCk1IY0NBUUVFSUVkSTRZN3lRU1ZwRGNrblhsQmJEaXBWZHRMWEVsYVBkN3VBZHdBWFFya2xvQW9HQ0NxR1NNNDkKQXdFSG9VUURRZ0FFOHpzQk1VOGxvdmJuN0RSQXloNEhWYUVBU2Y5bTV2Y3BFTU9SdlBUeGVrQnUrTVNLK0NBMAphSzJmUlY2V0VVYU4xL2pNQTZnekJIRGVaMmRjZVFaUFRBPT0KLS0tLS1FTkQgRUMgUFJJVkFURSBLRVktLS0tLQo=" } \ No newline at end of file diff --git a/go.mod b/go.mod index 2e1db49..7b1c6cb 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module oc-scheduler go 1.25.0 require ( - cloud.o-forge.io/core/oc-lib v0.0.0-20260318143822-5976795d4406 + cloud.o-forge.io/core/oc-lib v0.0.0-20260319080542-c7884f5cde5d github.com/beego/beego/v2 v2.3.8 github.com/google/uuid v1.6.0 github.com/robfig/cron v1.2.0 @@ -68,7 +68,7 @@ require ( github.com/mitchellh/mapstructure v1.5.0 // indirect github.com/montanaflynn/stats v0.7.1 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect - github.com/nats-io/nats.go v1.44.0 + github.com/nats-io/nats.go v1.44.0 // indirect github.com/nats-io/nkeys v0.4.11 // indirect github.com/nats-io/nuid v1.0.1 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect diff --git a/go.sum b/go.sum index eff8cfb..0c21d2b 100644 --- a/go.sum +++ b/go.sum @@ -1,37 +1,9 @@ -cloud.o-forge.io/core/oc-lib v0.0.0-20260112144037-c35b06e0bc3c h1:9lXrj1agE1clFfxOXRrVXi4PEvlAuWKb4z977c2uk4k= -cloud.o-forge.io/core/oc-lib v0.0.0-20260112144037-c35b06e0bc3c/go.mod h1:vHWauJsS6ryf7UDqq8hRXoYD5RsONxcFTxeZPOztEuI= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316075231-465b91fd6ecb h1:yO8KQpNHYIv4O6LrkRacFsTQrLv5qYYeO8KD1e1eunA= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316075231-465b91fd6ecb/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316082848-9e5102893f8a h1:4HHebXbTSzkD1MG/1GU5kZx45xx9IQ0sibndPuarlp0= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316082848-9e5102893f8a/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316094939-48f034316b91 h1:wm4oeR1mQE1esHAte9dYB8HC+pjY+G7zwfgQUxPO5g8= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316094939-48f034316b91/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316095931-a86e78841b34 h1:OxxfSNhdkqX165YzurzicnrU55s6n4pZjOg+HmkDzUc= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316095931-a86e78841b34/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316104105-deb819c5af95 h1:efOmy48+aw8vGGqHHUfrxVQJq0TlIux0/4aDcH7Wcpw= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316104105-deb819c5af95/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316104558-4a076ba23738 h1:L/xd9d1MCyhwQLwiuaAzID7pRUnotikGSe7jhSqtqPs= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316104558-4a076ba23738/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316104751-40a986af41b8 h1:02FkLYGjbGp/gq8Ej31KIXwF8QaQzJG/IODQt6GogT8= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316104751-40a986af41b8/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316113239-6d8efd137ac5 h1:NF+TYU0it9cWsrTGngv9KVGgrglMCO522/huR2RJNu0= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316113239-6d8efd137ac5/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316114821-9bf2c566e922 h1:B1DzkKyidaSLC7cdJ3jg+kQR9gU20DlGS+KjI8SmlDg= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316114821-9bf2c566e922/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316121650-a4d81cbb67f4 h1:k/xjsnRPIQjoaXp59x0CdwncpJa8KV7Fiyf78fgx7Ao= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316121650-a4d81cbb67f4/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316145919-b110cbc260c9 h1:+Yk9oHpChZhQYce2GY3HnFfW6AdeYAO31kczhwwpKgc= -cloud.o-forge.io/core/oc-lib v0.0.0-20260316145919-b110cbc260c9/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= -cloud.o-forge.io/core/oc-lib v0.0.0-20260317080147-9b2f9451767e h1:tKipQ9WFDJZXgex5MORwI3v0lJsEPaHHeIJqVWA3Vzk= -cloud.o-forge.io/core/oc-lib v0.0.0-20260317080147-9b2f9451767e/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= -cloud.o-forge.io/core/oc-lib v0.0.0-20260317083202-65237f0d1f3f h1:X8ytAjBzEqnFL1YQnjm9Ol/aoCiU/H6IgdzX74ZhFig= -cloud.o-forge.io/core/oc-lib v0.0.0-20260317083202-65237f0d1f3f/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= -cloud.o-forge.io/core/oc-lib v0.0.0-20260317090440-1ac735cef10e h1:e/oYMPAqD27l3Rd473Xny/2Ut/LZnBYXAzfQArNOmrs= -cloud.o-forge.io/core/oc-lib v0.0.0-20260317090440-1ac735cef10e/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= -cloud.o-forge.io/core/oc-lib v0.0.0-20260317135927-72be3118b7af h1:IySCYxJrKUpmRa2R3hXSaYxfWf/cm28NRpmwluEmzBI= -cloud.o-forge.io/core/oc-lib v0.0.0-20260317135927-72be3118b7af/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= -cloud.o-forge.io/core/oc-lib v0.0.0-20260318143822-5976795d4406 h1:FN1EtRWn228JprAbnY5K863Fzj+SzMqQtKRtwvECbLw= -cloud.o-forge.io/core/oc-lib v0.0.0-20260318143822-5976795d4406/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= +cloud.o-forge.io/core/oc-lib v0.0.0-20260319071818-28b5b7d39ffe h1:CHiWQAX7j/bMfbytCWGL2mUgSWYoDY4+bFQbCHEfypk= +cloud.o-forge.io/core/oc-lib v0.0.0-20260319071818-28b5b7d39ffe/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= +cloud.o-forge.io/core/oc-lib v0.0.0-20260319074425-5fca0480af06 h1:5nPNvh1ynFaTB6NBwjhR148iUTLZEyANbqAYQRW7dw0= +cloud.o-forge.io/core/oc-lib v0.0.0-20260319074425-5fca0480af06/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= +cloud.o-forge.io/core/oc-lib v0.0.0-20260319080542-c7884f5cde5d h1:5hM3GibJw5Uc2Z4aPSMt/3wh7RRY9zxJoeE1lGq0WY0= +cloud.o-forge.io/core/oc-lib v0.0.0-20260319080542-c7884f5cde5d/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/Masterminds/semver/v3 v3.4.0 h1:Zog+i5UMtVoCU8oKka5P7i9q9HgrJeGzI9SA1Xbatp0= github.com/Masterminds/semver/v3 v3.4.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM= diff --git a/infrastructure/check.go b/infrastructure/check.go index 446d183..a83b691 100644 --- a/infrastructure/check.go +++ b/infrastructure/check.go @@ -72,7 +72,7 @@ func (ws *WorkflowSchedule) Check(wfID string, asap bool, preemption bool, reque // 2. Resolve start start := ws.Start if asap || start.IsZero() { - start = time.Now().Add(asapBuffer) + start = time.Now().UTC().Add(asapBuffer) } // 3. Resolve end – use explicit end/duration or estimate via Planify diff --git a/infrastructure/nats_handlers.go b/infrastructure/nats_handlers.go index 74a8560..2bc3256 100644 --- a/infrastructure/nats_handlers.go +++ b/infrastructure/nats_handlers.go @@ -104,7 +104,7 @@ func handleCreateBooking(bk *booking.Booking, self *peer.Peer, adminReq *tools.A return } // Expired check only on confirmation (IsDraft→false). - if !bk.IsDraft && !prev.ExpectedStartDate.IsZero() && prev.ExpectedStartDate.Before(time.Now()) { + if !bk.IsDraft && !prev.ExpectedStartDate.IsZero() && prev.ExpectedStartDate.Before(time.Now().UTC()) { fmt.Println("ListenNATS CREATE_RESOURCE booking: expired, deleting", bk.GetID()) booking.NewAccessor(adminReq).DeleteOne(bk.GetID()) return @@ -120,7 +120,7 @@ func handleCreateBooking(bk *booking.Booking, self *peer.Peer, adminReq *tools.A return } // New booking: standard create flow. - if !bk.ExpectedStartDate.IsZero() && bk.ExpectedStartDate.Before(time.Now()) { + if !bk.ExpectedStartDate.IsZero() && bk.ExpectedStartDate.Before(time.Now().UTC()) { fmt.Println("ListenNATS: booking start date is in the past, discarding") return } diff --git a/infrastructure/planner.go b/infrastructure/planner.go index 18d5457..2c99e7a 100644 --- a/infrastructure/planner.go +++ b/infrastructure/planner.go @@ -153,7 +153,7 @@ func storePlanner(peerID string, p *planner.Planner) { if isNew { entry = &plannerEntry{} PlannerCache[peerID] = entry - plannerAddedAt[peerID] = time.Now() + plannerAddedAt[peerID] = time.Now().UTC() go evictAfter(peerID, plannerTTL) } entry.Planner = p @@ -196,7 +196,7 @@ func RequestPlannerRefresh(peerIDs []string, executionsID string) []string { if entry == nil { entry = &plannerEntry{} PlannerCache[peerID] = entry - plannerAddedAt[peerID] = time.Now() + plannerAddedAt[peerID] = time.Now().UTC() go evictAfter(peerID, plannerTTL) } shouldRequest := !entry.Refreshing diff --git a/infrastructure/scheduler.go b/infrastructure/scheduler.go index 5058d75..b16971f 100644 --- a/infrastructure/scheduler.go +++ b/infrastructure/scheduler.go @@ -55,17 +55,17 @@ type WorkflowSchedule struct { func NewScheduler(mode int, start string, end string, durationInS float64, cron string) *WorkflowSchedule { ws := &WorkflowSchedule{ UUID: uuid.New().String(), - Start: time.Now().Add(asapBuffer), + Start: time.Now().UTC().Add(asapBuffer), BookingMode: booking.BookingMode(mode), DurationS: durationInS, Cron: cron, } - s, err := time.Parse("2006-01-02T15:04:05", start) + s, err := time.ParseInLocation("2006-01-02T15:04:05", start, time.UTC) if err == nil && ws.BookingMode == booking.PLANNED { ws.Start = s // can apply a defined start other than now, if planned } - e, err := time.Parse("2006-01-02T15:04:05", end) + e, err := time.ParseInLocation("2006-01-02T15:04:05", end, time.UTC) if err == nil { ws.End = &e } @@ -153,7 +153,7 @@ func (ws *WorkflowSchedule) Schedules(wfID string, request *tools.APIRequest) (* // Obsolescence check: abort if any session execution's start date has passed. executions := loadSessionExecs(ws.UUID) for _, exec := range executions { - if !exec.ExecDate.IsZero() && exec.ExecDate.Before(time.Now()) { + if !exec.ExecDate.IsZero() && exec.ExecDate.Before(time.Now().UTC()) { return ws, nil, nil, fmt.Errorf("execution %s is obsolete (start date in the past)", exec.GetID()) } } @@ -163,7 +163,7 @@ func (ws *WorkflowSchedule) Schedules(wfID string, request *tools.APIRequest) (* } for _, exec := range executions { - go WatchExecDeadline(exec.GetID(), exec.ExecDate, selfID, request) + go WatchExecDeadline(exec.GetID(), exec.ExecutionsID, exec.ExecDate, selfID, request) } obj, _, _ := workflow.NewAccessor(request).LoadOne(wfID) diff --git a/infrastructure/session.go b/infrastructure/session.go index f1fb138..0334f68 100644 --- a/infrastructure/session.go +++ b/infrastructure/session.go @@ -1,8 +1,10 @@ package infrastructure import ( + "context" "encoding/json" "fmt" + "oc-scheduler/conf" "oc-scheduler/infrastructure/scheduling" "time" @@ -119,7 +121,8 @@ func (ws *WorkflowSchedule) UpsertSessionDrafts(wfID, executionsID string, selfI ex, _, err := utils.GenericStoreOne(exec, workflow_execution.NewAccessor(adminReq)) if err == nil { RegisterExecLock(ex.GetID()) - go WatchExecDeadline(ex.GetID(), exec.ExecDate, selfID, request) + go WatchExecDeadline( + ex.GetID(), executionsID, exec.ExecDate, selfID, request) } } @@ -269,24 +272,71 @@ func emitNATSRemove(id, schedulerPeerID, executionsID string, dt tools.DataType) // Deadline watchers // --------------------------------------------------------------------------- -// WatchExecDeadline purges all unconfirmed bookings/purchases for an execution -// one minute before its scheduled start, to avoid stale drafts blocking resources. +// WatchExecDeadline fires one minute before the execution start date. +// If the execution is still a draft it is purged; otherwise the namespace +// is created and a WatchExecEnd watcher is armed. // If the deadline has already passed (e.g. after a process restart), it fires immediately. -func WatchExecDeadline(executionID string, execDate time.Time, selfID *peer.Peer, request *tools.APIRequest) { +func WatchExecDeadline(executionID string, ns string, execDate time.Time, selfID *peer.Peer, request *tools.APIRequest) { fmt.Println("WatchExecDeadline") delay := time.Until(execDate.UTC().Add(-1 * time.Minute)) if delay <= 0 { - go purgeUnconfirmedExecution(executionID, selfID, request) + go handleExecDeadline(executionID, ns, selfID, request) return } - time.AfterFunc(delay, func() { purgeUnconfirmedExecution(executionID, selfID, request) }) + time.AfterFunc(delay, func() { handleExecDeadline(executionID, ns, selfID, request) }) } -func purgeUnconfirmedExecution(executionID string, selfID *peer.Peer, request *tools.APIRequest) { - acc := workflow_execution.NewAccessor(&tools.APIRequest{Admin: true}) - UnscheduleExecution(executionID, selfID, request) - _, _, err := acc.DeleteOne(executionID) - fmt.Printf("purgeUnconfirmedExecution: cleaned up resources for execution %s\n", err) +func handleExecDeadline(executionID string, ns string, selfID *peer.Peer, request *tools.APIRequest) { + adminReq := &tools.APIRequest{Admin: true} + res, _, err := workflow_execution.NewAccessor(adminReq).LoadOne(executionID) + if err != nil || res == nil { + fmt.Printf("handleExecDeadline: execution %s not found\n", executionID) + return + } + exec := res.(*workflow_execution.WorkflowExecution) + if exec.IsDraft { + UnscheduleExecution(executionID, selfID, request) + workflow_execution.NewAccessor(adminReq).DeleteOne(executionID) + fmt.Printf("handleExecDeadline: purged draft execution %s\n", executionID) + return + } + if serv, err := tools.NewKubernetesService( + conf.GetConfig().KubeHost+":"+conf.GetConfig().KubePort, + conf.GetConfig().KubeCA, conf.GetConfig().KubeCert, conf.GetConfig().KubeData); err != nil { + fmt.Printf("handleExecDeadline: k8s init failed for %s: %v\n", executionID, err) + } else if err := serv.ProvisionExecutionNamespace(context.Background(), ns); err != nil { + fmt.Printf("handleExecDeadline: failed to provision namespace for %s: %v\n", ns, err) + } + go WatchExecEnd(executionID, ns, exec.EndDate, exec.ExecDate) +} + +// WatchExecEnd fires at the execution end date (ExecDate+1h when EndDate is nil) +// and deletes the Kubernetes namespace associated with the execution. +func WatchExecEnd(executionID string, ns string, endDate *time.Time, execDate time.Time) { + var end time.Time + if endDate != nil { + end = *endDate + } else { + end = execDate.UTC().Add(time.Hour) + } + delay := time.Until(end.UTC()) + fire := func() { + serv, err := tools.NewKubernetesService( + conf.GetConfig().KubeHost+":"+conf.GetConfig().KubePort, + conf.GetConfig().KubeCA, conf.GetConfig().KubeCert, conf.GetConfig().KubeData) + if err != nil { + fmt.Printf("WatchExecEnd: k8s init failed for %s: %v\n", executionID, err) + return + } + if err := serv.TeardownExecutionNamespace(context.Background(), ns); err != nil { + fmt.Printf("WatchExecEnd: failed to teardown namespace %s: %v\n", ns, err) + } + } + if delay <= 0 { + go fire() + return + } + time.AfterFunc(delay, fire) } // RecoverDraftExecutions is called at startup to restore deadline watchers for @@ -308,7 +358,7 @@ func RecoverDraftExecutions() { continue } RegisterExecLock(exec.GetID()) - go WatchExecDeadline(exec.GetID(), exec.ExecDate, selfID, adminReq) + go WatchExecDeadline(exec.GetID(), exec.ExecutionsID, exec.ExecDate, selfID, adminReq) } fmt.Printf("RecoverDraftExecutions: recovered %d draft executions\n", len(results)) } diff --git a/main.go b/main.go index ac629d7..1b86153 100644 --- a/main.go +++ b/main.go @@ -1,11 +1,9 @@ package main import ( - "encoding/base64" "oc-scheduler/conf" "oc-scheduler/infrastructure" _ "oc-scheduler/routers" - "os" oclib "cloud.o-forge.io/core/oc-lib" beego "github.com/beego/beego/v2/server/web" @@ -15,21 +13,12 @@ const appname = "oc-scheduler" func main() { o := oclib.GetConfLoader(appname) - conf.GetConfig().KubeHost = o.GetStringDefault("KUBERNETES_SERVICE_HOST", os.Getenv("KUBERNETES_SERVICE_HOST")) + conf.GetConfig().KubeHost = o.GetStringDefault("KUBERNETES_SERVICE_HOST", "kubernetes.default.svc.cluster.local") conf.GetConfig().KubePort = o.GetStringDefault("KUBERNETES_SERVICE_PORT", "6443") - sDec, err := base64.StdEncoding.DecodeString(o.GetStringDefault("KUBE_CA", "")) - if err == nil { - conf.GetConfig().KubeCA = string(sDec) - } - sDec, err = base64.StdEncoding.DecodeString(o.GetStringDefault("KUBE_CERT", "")) - if err == nil { - conf.GetConfig().KubeCert = string(sDec) - } - sDec, err = base64.StdEncoding.DecodeString(o.GetStringDefault("KUBE_DATA", "")) - if err == nil { - conf.GetConfig().KubeData = string(sDec) - } + conf.GetConfig().KubeCA = o.GetStringDefault("KUBE_CA", "LS0tLS1CRUdJTiBDRVJUSUZJQ0FURS0tLS0tCk1JSUJkakNDQVIyZ0F3SUJBZ0lCQURBS0JnZ3Foa2pPUFFRREFqQWpNU0V3SHdZRFZRUUREQmhyTTNNdGMyVnkKZG1WeUxXTmhRREUzTnpNeE1qY3dPVFl3SGhjTk1qWXdNekV3TURjeE9ERTJXaGNOTXpZd016QTNNRGN4T0RFMgpXakFqTVNFd0h3WURWUVFEREJock0zTXRjMlZ5ZG1WeUxXTmhRREUzTnpNeE1qY3dPVFl3V1RBVEJnY3Foa2pPClBRSUJCZ2dxaGtqT1BRTUJCd05DQUFReG81cXQ0MGxEekczRHJKTE1wRVBrd0ZBY1FmbC8vVE1iWjZzemMreHAKbmVzVzRTSTdXK1lWdFpRYklmV2xBMTRaazQvRFlDMHc1YlgxZU94RVVuL0pvMEl3UURBT0JnTlZIUThCQWY4RQpCQU1DQXFRd0R3WURWUjBUQVFIL0JBVXdBd0VCL3pBZEJnTlZIUTRFRmdRVXBLM2pGK25IRlZSbDcwb3ZRVGZnCmZabGNQZE13Q2dZSUtvWkl6ajBFQXdJRFJ3QXdSQUlnVnkyaUx0Y0xaYm1vTnVoVHdKbU5sWlo3RVlBYjJKNW0KSjJYbG1UbVF5a2tDSUhLbzczaDBkdEtUZTlSa0NXYTJNdStkS1FzOXRFU0tBV0x1emlnYXBHYysKLS0tLS1FTkQgQ0VSVElGSUNBVEUtLS0tLQo=") + conf.GetConfig().KubeCert = o.GetStringDefault("KUBE_CERT", "LS0tLS1CRUdJTiBDRVJUSUZJQ0FURS0tLS0tCk1JSUJrakNDQVRlZ0F3SUJBZ0lJQUkvSUg2R2Rodm93Q2dZSUtvWkl6ajBFQXdJd0l6RWhNQjhHQTFVRUF3d1kKYXpOekxXTnNhV1Z1ZEMxallVQXhOemN6TVRJM01EazJNQjRYRFRJMk1ETXhNREEzTVRneE5sb1hEVEkzTURNeApNREEzTVRneE5sb3dNREVYTUJVR0ExVUVDaE1PYzNsemRHVnRPbTFoYzNSbGNuTXhGVEFUQmdOVkJBTVRESE41CmMzUmxiVHBoWkcxcGJqQlpNQk1HQnlxR1NNNDlBZ0VHQ0NxR1NNNDlBd0VIQTBJQUJQTTdBVEZQSmFMMjUrdzAKUU1vZUIxV2hBRW4vWnViM0tSRERrYnowOFhwQWJ2akVpdmdnTkdpdG4wVmVsaEZHamRmNHpBT29Nd1J3M21kbgpYSGtHVDB5alNEQkdNQTRHQTFVZER3RUIvd1FFQXdJRm9EQVRCZ05WSFNVRUREQUtCZ2dyQmdFRkJRY0RBakFmCkJnTlZIU01FR0RBV2dCUVZLOThaMEMxcFFyVFJSMGVLZHhIa2o0ejFJREFLQmdncWhrak9QUVFEQWdOSkFEQkcKQWlFQXZYWll6Zk9iSUtlWTRtclNsRmt4ZS80a0E4K01ieDc1UDFKRmNlRS8xdGNDSVFDNnM0ZXlZclhQYmNWSgpxZm5EamkrZ1RacGttN0tWSTZTYTlZN2FSRGFabUE9PQotLS0tLUVORCBDRVJUSUZJQ0FURS0tLS0tCi0tLS0tQkVHSU4gQ0VSVElGSUNBVEUtLS0tLQpNSUlCZURDQ0FSMmdBd0lCQWdJQkFEQUtCZ2dxaGtqT1BRUURBakFqTVNFd0h3WURWUVFEREJock0zTXRZMnhwClpXNTBMV05oUURFM056TXhNamN3T1RZd0hoY05Nall3TXpFd01EY3hPREUyV2hjTk16WXdNekEzTURjeE9ERTIKV2pBak1TRXdId1lEVlFRRERCaHJNM010WTJ4cFpXNTBMV05oUURFM056TXhNamN3T1RZd1dUQVRCZ2NxaGtqTwpQUUlCQmdncWhrak9QUU1CQndOQ0FBUzV1NGVJbStvVnV1SFI0aTZIOU1kVzlyUHdJbFVPNFhIMEJWaDRUTGNlCkNkMnRBbFVXUW5FakxMdlpDWlVaYTlzTlhKOUVtWWt5S0dtQWR2TE9FbUVrbzBJd1FEQU9CZ05WSFE4QkFmOEUKQkFNQ0FxUXdEd1lEVlIwVEFRSC9CQVV3QXdFQi96QWRCZ05WSFE0RUZnUVVGU3ZmR2RBdGFVSzAwVWRIaW5jUgo1SStNOVNBd0NnWUlLb1pJemowRUF3SURTUUF3UmdJaEFMY2xtQnR4TnpSVlBvV2hoVEVKSkM1Z3VNSGsvcFZpCjFvYXJ2UVJxTWRKcUFpRUEyR1dNTzlhZFFYTEQwbFZKdHZMVkc1M3I0M0lxMHpEUUQwbTExMVZyL1MwPQotLS0tLUVORCBDRVJUSUZJQ0FURS0tLS0tCg==") + conf.GetConfig().KubeData = o.GetStringDefault("KUBE_DATA", "LS0tLS1CRUdJTiBFQyBQUklWQVRFIEtFWS0tLS0tCk1IY0NBUUVFSUVkSTRZN3lRU1ZwRGNrblhsQmJEaXBWZHRMWEVsYVBkN3VBZHdBWFFya2xvQW9HQ0NxR1NNNDkKQXdFSG9VUURRZ0FFOHpzQk1VOGxvdmJuN0RSQXloNEhWYUVBU2Y5bTV2Y3BFTU9SdlBUeGVrQnUrTVNLK0NBMAphSzJmUlY2V0VVYU4xL2pNQTZnekJIRGVaMmRjZVFaUFRBPT0KLS0tLS1FTkQgRUMgUFJJVkFURSBLRVktLS0tLQo=") oclib.InitAPI(appname)