diff --git a/go.mod b/go.mod index 01dd0fe2c2..a963260ae6 100644 --- a/go.mod +++ b/go.mod @@ -32,7 +32,7 @@ require ( github.com/docker/go-connections v0.4.0 github.com/envoyproxy/go-control-plane v0.11.1-0.20230524094728-9239064ad72f github.com/gin-gonic/gin v1.9.1 - github.com/go-co-op/gocron v1.17.0 + github.com/go-co-op/gocron/v2 v2.14.0 github.com/go-ldap/ldap/v3 v3.3.0 github.com/go-redsync/redsync/v4 v4.11.0 github.com/go-resty/resty/v2 v2.7.0 @@ -82,7 +82,7 @@ require ( github.com/shirou/gopsutil/v3 v3.22.8 github.com/spf13/cobra v1.8.0 github.com/spf13/viper v1.8.1 - github.com/stretchr/testify v1.9.0 + github.com/stretchr/testify v1.10.0 github.com/swaggo/files v0.0.0-20220728132757-551d4a08d97a github.com/swaggo/gin-swagger v1.5.3 github.com/tidwall/gjson v1.14.3 @@ -234,6 +234,7 @@ require ( github.com/jinzhu/inflection v1.0.0 // indirect github.com/jmespath/go-jmespath v0.4.0 // indirect github.com/jmoiron/sqlx v1.3.5 // indirect + github.com/jonboulle/clockwork v0.4.0 // indirect github.com/josharian/intern v1.0.0 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/kennygrant/sanitize v1.2.4 // indirect diff --git a/pkg/cli/upgradeassistant/cmd/migrate/321.go b/pkg/cli/upgradeassistant/cmd/migrate/321.go new file mode 100644 index 0000000000..c711961e97 --- /dev/null +++ b/pkg/cli/upgradeassistant/cmd/migrate/321.go @@ -0,0 +1,99 @@ +/* + * Copyright 2024 The KodeRover Authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package migrate + +import ( + "context" + "fmt" + "time" + + "github.com/koderover/zadig/v2/pkg/cli/upgradeassistant/internal/upgradepath" + "github.com/koderover/zadig/v2/pkg/microservice/aslan/config" + "github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/repository/models" + commonmodels "github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/repository/models" + commonrepo "github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/repository/mongodb" + "github.com/koderover/zadig/v2/pkg/shared/handler" + "github.com/koderover/zadig/v2/pkg/util" + "go.mongodb.org/mongo-driver/bson" +) + +func init() { + upgradepath.RegisterHandler("3.2.0", "3.2.1", V320ToV321) + upgradepath.RegisterHandler("3.2.1", "3.2.0", V321ToV320) +} + +func V320ToV321() error { + ctx := handler.NewBackgroupContext() + + ctx.Logger.Infof("-------- start migrate release plan cronjob --------") + err := migrateReleasePlanCron(ctx) + if err != nil { + err = fmt.Errorf("failed to migrate release plan cronjob, error: %w", err) + ctx.Logger.Error(err) + return err + } + + return nil +} + +func V321ToV320() error { + return nil +} + +func migrateReleasePlanCron(ctx *handler.Context) error { + // delete all release plan cronjob first + _, err := commonrepo.NewCronjobColl().DeleteMany(ctx, + bson.M{"type": "release_plan"}, + ) + if err != nil { + return fmt.Errorf("failed to delete release plan cronjobs, error: %w", err) + } + + cursor, err := commonrepo.NewReleasePlanColl().ListByCursor() + if err != nil { + return fmt.Errorf("failed to list release plans, error: %w", err) + } + for cursor.Next(context.Background()) { + var releasePlan models.ReleasePlan + if err := cursor.Decode(&releasePlan); err != nil { + return err + } + + if releasePlan.ScheduleExecuteTime != 0 && releasePlan.Status == config.StatusExecuting { + if time.Unix(releasePlan.ScheduleExecuteTime, 0).After(time.Now()) { + releasePlanCronName := util.GetReleasePlanCronName(releasePlan.ID.Hex(), releasePlan.Name, releasePlan.Index) + cronjob := &commonmodels.Cronjob{ + Enabled: true, + Name: releasePlanCronName, + Type: "release_plan", + JobType: string(config.UnixstampSchedule), + UnixStamp: releasePlan.ScheduleExecuteTime, + ReleasePlanArgs: &commonmodels.ReleasePlanArgs{ + ID: releasePlan.ID.Hex(), + Name: releasePlan.Name, + Index: releasePlan.Index, + }, + } + if err := commonrepo.NewCronjobColl().Upsert(cronjob); err != nil { + return fmt.Errorf("failed to create new release plan schedule job, error: %w", err) + } + } + } + } + + return nil +} diff --git a/pkg/microservice/aslan/config/consts.go b/pkg/microservice/aslan/config/consts.go index be0570b6ff..f3f70f9263 100644 --- a/pkg/microservice/aslan/config/consts.go +++ b/pkg/microservice/aslan/config/consts.go @@ -58,6 +58,8 @@ const ( TimingSchedule ScheduleType = "timing" // GapSchedule 间隔循环 GapSchedule ScheduleType = "gap" + // UnixstampSchedule 时间戳 + UnixstampSchedule ScheduleType = "unix_stamp" ) type SlackNotifyType string diff --git a/pkg/microservice/aslan/core/build/handler/openapi.go b/pkg/microservice/aslan/core/build/handler/openapi.go index eadd7eeccd..e005a67d46 100644 --- a/pkg/microservice/aslan/core/build/handler/openapi.go +++ b/pkg/microservice/aslan/core/build/handler/openapi.go @@ -217,6 +217,9 @@ func OpenAPIGetBuildModule(c *gin.Context) { return } + serviceName := c.Query("serviceName") + serviceModule := c.Query("serviceModule") + // authorization checks if !ctx.Resources.IsSystemAdmin { if _, ok := ctx.Resources.ProjectAuthInfo[projectKey]; !ok { @@ -230,5 +233,5 @@ func OpenAPIGetBuildModule(c *gin.Context) { } } - ctx.Resp, ctx.RespErr = buildservice.OpenAPIGetBuildModule(name, projectKey, ctx.Logger) + ctx.Resp, ctx.RespErr = buildservice.OpenAPIGetBuildModule(name, serviceName, serviceModule, projectKey, ctx.Logger) } diff --git a/pkg/microservice/aslan/core/build/service/openapi.go b/pkg/microservice/aslan/core/build/service/openapi.go index a6f411154c..7912d816e5 100644 --- a/pkg/microservice/aslan/core/build/service/openapi.go +++ b/pkg/microservice/aslan/core/build/service/openapi.go @@ -316,7 +316,7 @@ func OpenAPIListBuildModules(projectName string, pageNum, pageSize int64, logger return resp, nil } -func OpenAPIGetBuildModule(name, projectName string, logger *zap.SugaredLogger) (*OpenAPIBuildDetailResp, error) { +func OpenAPIGetBuildModule(name, serviceName, serviceModule, projectName string, logger *zap.SugaredLogger) (*OpenAPIBuildDetailResp, error) { opt := &commonrepo.BuildFindOption{ Name: name, ProductName: projectName, @@ -348,18 +348,39 @@ func OpenAPIGetBuildModule(name, projectName string, logger *zap.SugaredLogger) } resp.Repos = make([]*OpenAPIRepo, 0) - for _, rp := range build.Repos { - repo := &OpenAPIRepo{ - RepoName: rp.RepoName, - Branch: rp.Branch, - Source: rp.Source, - RepoOwner: rp.RepoOwner, - RemoteName: rp.RemoteName, - CheckoutPath: rp.CheckoutPath, - Submodules: rp.SubModules, - Hidden: rp.Hidden, + if serviceName == "" || serviceModule == "" || build.TemplateID == "" { + for _, rp := range build.Repos { + repo := &OpenAPIRepo{ + RepoName: rp.RepoName, + Branch: rp.Branch, + Source: rp.Source, + RepoOwner: rp.RepoOwner, + RemoteName: rp.RemoteName, + CheckoutPath: rp.CheckoutPath, + Submodules: rp.SubModules, + Hidden: rp.Hidden, + } + resp.Repos = append(resp.Repos, repo) + } + } else { + for _, svcBuild := range build.Targets { + if svcBuild.ServiceName == serviceName && svcBuild.ServiceModule == serviceModule { + for _, rp := range svcBuild.Repos { + repo := &OpenAPIRepo{ + RepoName: rp.RepoName, + Branch: rp.Branch, + Source: rp.Source, + RepoOwner: rp.RepoOwner, + RemoteName: rp.RemoteName, + CheckoutPath: rp.CheckoutPath, + Submodules: rp.SubModules, + Hidden: rp.Hidden, + } + resp.Repos = append(resp.Repos, repo) + } + break + } } - resp.Repos = append(resp.Repos, repo) } resp.TargetServices = make([]*commonmodels.ServiceWithModule, 0) diff --git a/pkg/microservice/aslan/core/code/client/gitee/gitee.go b/pkg/microservice/aslan/core/code/client/gitee/gitee.go index 3642375421..32bec35781 100644 --- a/pkg/microservice/aslan/core/code/client/gitee/gitee.go +++ b/pkg/microservice/aslan/core/code/client/gitee/gitee.go @@ -152,7 +152,7 @@ func (c *Client) ListProjects(opt client.ListOpt) ([]*client.Project, error) { for _, project := range projects { res = append(res, &client.Project{ ID: project.ID, - Name: project.Name, + Name: project.Path, DefaultBranch: project.DefaultBranch, }) } diff --git a/pkg/microservice/aslan/core/common/repository/models/cronjob.go b/pkg/microservice/aslan/core/common/repository/models/cronjob.go index 0485632778..e756dae479 100644 --- a/pkg/microservice/aslan/core/common/repository/models/cronjob.go +++ b/pkg/microservice/aslan/core/common/repository/models/cronjob.go @@ -25,6 +25,7 @@ type Cronjob struct { Name string `bson:"name" json:"name"` Type string `bson:"type" json:"type"` Number uint64 `bson:"number" json:"number"` + UnixStamp int64 `bson:"unix_stamp" json:"unix_stamp"` Frequency string `bson:"frequency" json:"frequency"` Time string `bson:"time" json:"time"` Cron string `bson:"cron" json:"cron"` diff --git a/pkg/microservice/aslan/core/common/repository/models/workflow.go b/pkg/microservice/aslan/core/common/repository/models/workflow.go index 5e3375509c..c6d047a5de 100644 --- a/pkg/microservice/aslan/core/common/repository/models/workflow.go +++ b/pkg/microservice/aslan/core/common/repository/models/workflow.go @@ -121,6 +121,7 @@ type Schedule struct { ID primitive.ObjectID `bson:"_id,omitempty" json:"id,omitempty"` Number uint64 `bson:"number" json:"number"` Frequency string `bson:"frequency" json:"frequency"` + UnixStamp int64 `bson:"unix_stamp" json:"unix_stamp"` Time string `bson:"time" json:"time"` MaxFailures int `bson:"max_failures,omitempty" json:"max_failures,omitempty"` TaskArgs *TaskArgs `bson:"task_args,omitempty" json:"task_args,omitempty"` diff --git a/pkg/microservice/aslan/core/common/repository/mongodb/release_plan.go b/pkg/microservice/aslan/core/common/repository/mongodb/release_plan.go index c07e4094b3..52e161eed6 100644 --- a/pkg/microservice/aslan/core/common/repository/mongodb/release_plan.go +++ b/pkg/microservice/aslan/core/common/repository/mongodb/release_plan.go @@ -71,6 +71,10 @@ func (c *ReleasePlanColl) EnsureIndex(ctx context.Context) error { Keys: bson.M{"success_time": 1}, Options: options.Index().SetUnique(false), }, + { + Keys: bson.M{"update_time": 1}, + Options: options.Index().SetUnique(false), + }, } _, err := c.Indexes().CreateMany(ctx, mod) @@ -124,6 +128,13 @@ func (c *ReleasePlanColl) DeleteByID(ctx context.Context, idString string) error return err } +type SortReleasePlanBy string + +const ( + SortReleasePlanByIndex SortReleasePlanBy = "index" + SortReleasePlanByUpdateTime SortReleasePlanBy = "update_time" +) + type ListReleasePlanOption struct { PageNum int64 PageSize int64 @@ -131,7 +142,10 @@ type ListReleasePlanOption struct { Manager string SuccessTimeStart int64 SuccessTimeEnd int64 + UpdateTimeStart int64 + UpdateTimeEnd int64 IsSort bool + SortBy SortReleasePlanBy ExcludedFields []string Status config.ReleasePlanStatus } @@ -147,8 +161,15 @@ func (c *ReleasePlanColl) ListByOptions(opt *ListReleasePlanOption) ([]*models.R ctx := context.Background() opts := options.Find() if opt.IsSort { - opts.SetSort(bson.D{{"index", -1}}) + if opt.SortBy == SortReleasePlanByIndex { + opts.SetSort(bson.D{{"index", -1}}) + } else if opt.SortBy == SortReleasePlanByUpdateTime { + opts.SetSort(bson.D{{"update_time", -1}}) + } else { + opts.SetSort(bson.D{{"index", -1}}) + } } + if opt.PageNum > 0 && opt.PageSize > 0 { opts.SetSkip((opt.PageNum - 1) * opt.PageSize) opts.SetLimit(opt.PageSize) @@ -162,6 +183,9 @@ func (c *ReleasePlanColl) ListByOptions(opt *ListReleasePlanOption) ([]*models.R if opt.SuccessTimeStart > 0 && opt.SuccessTimeEnd > 0 { query["success_time"] = bson.M{"$gte": opt.SuccessTimeStart, "$lte": opt.SuccessTimeEnd} } + if opt.UpdateTimeStart > 0 && opt.UpdateTimeEnd > 0 { + query["update_time"] = bson.M{"$gte": opt.UpdateTimeStart, "$lte": opt.UpdateTimeEnd} + } if opt.Status != "" { query["status"] = opt.Status } @@ -230,3 +254,8 @@ func (c *ReleasePlanColl) ListFinishedReleasePlan(startTime, endTime int64) ([]* return resp, nil } + +func (c *ReleasePlanColl) ListByCursor() (*mongo.Cursor, error) { + query := bson.M{} + return c.Collection.Find(context.TODO(), query) +} diff --git a/pkg/microservice/aslan/core/common/service/cronjob.go b/pkg/microservice/aslan/core/common/service/cronjob.go index fd32eaf120..3bdc6adfee 100644 --- a/pkg/microservice/aslan/core/common/service/cronjob.go +++ b/pkg/microservice/aslan/core/common/service/cronjob.go @@ -21,10 +21,11 @@ import ( ) type CronjobPayload struct { - Name string `json:"name"` - ProductName string `json:"product_name"` - Action string `json:"action"` - JobType string `json:"job_type"` - DeleteList []string `json:"delete_list,omitempty"` - JobList []*models.Schedule `json:"job_list,omitempty"` + Name string `json:"name"` + ProductName string `json:"product_name"` + Action string `json:"action"` + JobType string `json:"job_type"` + ScheduleType string `json:"schedule_type"` + DeleteList []string `json:"delete_list,omitempty"` + JobList []*models.Schedule `json:"job_list,omitempty"` } diff --git a/pkg/microservice/aslan/core/common/service/instantmessage/workflow_task.go b/pkg/microservice/aslan/core/common/service/instantmessage/workflow_task.go index 6333db2150..ada45f078e 100644 --- a/pkg/microservice/aslan/core/common/service/instantmessage/workflow_task.go +++ b/pkg/microservice/aslan/core/common/service/instantmessage/workflow_task.go @@ -566,7 +566,7 @@ func (w *Service) getNotificationContent(notify *models.NotifyCtl, task *models. } } } - if image != "" { + if image != "" && !strings.HasPrefix(image, "{{.") && !strings.Contains(image, "}}") { jobTplcontent += fmt.Sprintf("{{if eq .WebHookType \"dingding\"}}##### {{end}}**镜像信息**：%s \n", image) mailJobTplcontent += fmt.Sprintf("镜像信息：%s \n", image) workflowNotifyJobTaskSpec.Image = image diff --git a/pkg/microservice/aslan/core/cron/handler/cron.go b/pkg/microservice/aslan/core/cron/handler/cron.go index 3853041c72..afb9be9607 100644 --- a/pkg/microservice/aslan/core/cron/handler/cron.go +++ b/pkg/microservice/aslan/core/cron/handler/cron.go @@ -64,6 +64,7 @@ type cronjobResp struct { Name string `json:"name"` Type string `json:"type"` Number uint64 `json:"number"` + UnixStamp int64 `json:"unix_stamp"` Frequency string `json:"frequency"` Time string `json:"time"` Cron string `json:"cron"` @@ -92,6 +93,7 @@ func ListActiveCronjobFailsafe(c *gin.Context) { Name: cronjob.Name, Type: cronjob.Type, Number: cronjob.Number, + UnixStamp: cronjob.UnixStamp, Frequency: cronjob.Frequency, Time: cronjob.Time, Cron: cronjob.Cron, @@ -124,6 +126,7 @@ func ListActiveCronjob(c *gin.Context) { Name: cronjob.Name, Type: cronjob.Type, Number: cronjob.Number, + UnixStamp: cronjob.UnixStamp, Frequency: cronjob.Frequency, Time: cronjob.Time, Cron: cronjob.Cron, diff --git a/pkg/microservice/aslan/core/environment/service/environment.go b/pkg/microservice/aslan/core/environment/service/environment.go index 542c8f474f..fb86c570ad 100644 --- a/pkg/microservice/aslan/core/environment/service/environment.go +++ b/pkg/microservice/aslan/core/environment/service/environment.go @@ -1493,7 +1493,7 @@ func GetAffectedServices(productName, envName string, arg *K8sRendersetArg, log func GeneEstimatedValues(productName, envName, serviceOrReleaseName, scene, format string, arg *EstimateValuesArg, isHelmChartDeploy bool, log *zap.SugaredLogger) (interface{}, error) { var ( productSvc *commonmodels.ProductService - tmplSvc *commonmodels.Service + tmplSvc *commonmodels.Service productInfo *commonmodels.Product err error ) @@ -3791,6 +3791,7 @@ func cronJobToSchedule(input *commonmodels.Cronjob) *commonmodels.Schedule { return &commonmodels.Schedule{ ID: input.ID, Number: input.Number, + UnixStamp: input.UnixStamp, Frequency: input.Frequency, Time: input.Time, MaxFailures: input.MaxFailure, diff --git a/pkg/microservice/aslan/core/log/handler/router.go b/pkg/microservice/aslan/core/log/handler/router.go index 8fa82633d4..5a517430a3 100644 --- a/pkg/microservice/aslan/core/log/handler/router.go +++ b/pkg/microservice/aslan/core/log/handler/router.go @@ -38,7 +38,6 @@ func (*Router) Inject(router *gin.RouterGroup) { sse := router.Group("sse") { sse.GET("/pods/:podName/containers/:containerName", GetContainerLogsSSE) - sse.GET("/production/pods/:podName/containers/:containerName", GetProductionEnvContainerLogsSSE) sse.GET("/testing/:test_name/tasks/:task_id", GetTestingContainerLogsSSE) sse.GET("/scanning/:id/task/:scan_id", GetScanningContainerLogsSSE) sse.GET("/v4/workflow/:workflowName/:taskID/:jobName/:lines", GetWorkflowJobContainerLogsSSE) diff --git a/pkg/microservice/aslan/core/log/handler/sse.go b/pkg/microservice/aslan/core/log/handler/sse.go index 5123e08af1..b4e5a6f9ed 100644 --- a/pkg/microservice/aslan/core/log/handler/sse.go +++ b/pkg/microservice/aslan/core/log/handler/sse.go @@ -55,69 +55,54 @@ func GetContainerLogsSSE(c *gin.Context) { envName := c.Query("envName") productName := c.Query("projectName") + isProduction := c.Query("production") == "true" - // authorization checks - if !ctx.Resources.IsSystemAdmin { - if _, ok := ctx.Resources.ProjectAuthInfo[productName]; !ok { - ctx.UnAuthorized = true - internalhandler.JSONResponse(c, ctx) - return - } - if !ctx.Resources.ProjectAuthInfo[productName].Env.View && - !ctx.Resources.ProjectAuthInfo[productName].IsProjectAdmin { - permitted, err := internalhandler.GetCollaborationModePermission(ctx.UserID, productName, types.ResourceTypeEnvironment, envName, types.EnvActionView) - if err != nil || !permitted { + if !isProduction { + // authorization checks + if !ctx.Resources.IsSystemAdmin { + if _, ok := ctx.Resources.ProjectAuthInfo[productName]; !ok { ctx.UnAuthorized = true internalhandler.JSONResponse(c, ctx) return } + if !ctx.Resources.ProjectAuthInfo[productName].Env.View && + !ctx.Resources.ProjectAuthInfo[productName].IsProjectAdmin { + permitted, err := internalhandler.GetCollaborationModePermission(ctx.UserID, productName, types.ResourceTypeEnvironment, envName, types.EnvActionView) + if err != nil || !permitted { + ctx.UnAuthorized = true + internalhandler.JSONResponse(c, ctx) + return + } + } } - } - - internalhandler.Stream(c, func(ctx context.Context, streamChan chan interface{}) { - logservice.ContainerLogStream(ctx, streamChan, envName, productName, c.Param("podName"), c.Param("containerName"), true, tails, logger) - }, logger) -} -func GetProductionEnvContainerLogsSSE(c *gin.Context) { - logger := ginzap.WithContext(c).Sugar() - ctx, err := internalhandler.NewContextWithAuthorization(c) - if err != nil { - ctx.RespErr = fmt.Errorf("authorization Info Generation failed: err %s", err) - ctx.UnAuthorized = true - internalhandler.JSONResponse(c, ctx) - return - } - - tails, err := strconv.ParseInt(c.Query("tails"), 10, 64) - if err != nil { - tails = int64(10) - } - - envName := c.Query("envName") - productName := c.Query("projectName") - - // authorization checks - if !ctx.Resources.IsSystemAdmin { - if _, ok := ctx.Resources.ProjectAuthInfo[productName]; !ok { - ctx.UnAuthorized = true - internalhandler.JSONResponse(c, ctx) - return - } - if !ctx.Resources.ProjectAuthInfo[productName].ProductionEnv.View && - !ctx.Resources.ProjectAuthInfo[productName].IsProjectAdmin { - permitted, err := internalhandler.GetCollaborationModePermission(ctx.UserID, productName, types.ResourceTypeEnvironment, envName, types.ProductionEnvActionView) - if err != nil || !permitted { + internalhandler.Stream(c, func(ctx context.Context, streamChan chan interface{}) { + logservice.ContainerLogStream(ctx, streamChan, envName, productName, c.Param("podName"), c.Param("containerName"), true, tails, logger) + }, logger) + } else { + // authorization checks + if !ctx.Resources.IsSystemAdmin { + if _, ok := ctx.Resources.ProjectAuthInfo[productName]; !ok { ctx.UnAuthorized = true internalhandler.JSONResponse(c, ctx) return } + if !ctx.Resources.ProjectAuthInfo[productName].ProductionEnv.View && + !ctx.Resources.ProjectAuthInfo[productName].IsProjectAdmin { + permitted, err := internalhandler.GetCollaborationModePermission(ctx.UserID, productName, types.ResourceTypeEnvironment, envName, types.ProductionEnvActionView) + if err != nil || !permitted { + ctx.UnAuthorized = true + internalhandler.JSONResponse(c, ctx) + return + } + } } + + internalhandler.Stream(c, func(ctx context.Context, streamChan chan interface{}) { + logservice.ContainerLogStream(ctx, streamChan, envName, productName, c.Param("podName"), c.Param("containerName"), true, tails, logger) + }, logger) } - internalhandler.Stream(c, func(ctx context.Context, streamChan chan interface{}) { - logservice.ContainerLogStream(ctx, streamChan, envName, productName, c.Param("podName"), c.Param("containerName"), true, tails, logger) - }, logger) } func GetWorkflowJobContainerLogsSSE(c *gin.Context) { diff --git a/pkg/microservice/aslan/core/release_plan/handler/release_plan.go b/pkg/microservice/aslan/core/release_plan/handler/release_plan.go index 7ef23436f7..c85dc2d5e7 100644 --- a/pkg/microservice/aslan/core/release_plan/handler/release_plan.go +++ b/pkg/microservice/aslan/core/release_plan/handler/release_plan.go @@ -204,7 +204,7 @@ func ScheduleExecuteReleasePlan(c *gin.Context) { return } - ctx.RespErr = service.ScheduleExecuteReleasePlan(ctx, c.Param("id")) + ctx.RespErr = service.ScheduleExecuteReleasePlan(ctx, c.Param("id"), c.Query("jobID")) } func SkipReleaseJob(c *gin.Context) { diff --git a/pkg/microservice/aslan/core/release_plan/service/release_plan.go b/pkg/microservice/aslan/core/release_plan/service/release_plan.go index e37015b22b..1f916d00eb 100644 --- a/pkg/microservice/aslan/core/release_plan/service/release_plan.go +++ b/pkg/microservice/aslan/core/release_plan/service/release_plan.go @@ -28,6 +28,7 @@ import ( "github.com/google/uuid" "github.com/pkg/errors" "github.com/samber/lo" + "go.mongodb.org/mongo-driver/bson/primitive" "go.mongodb.org/mongo-driver/mongo" "github.com/koderover/zadig/v2/pkg/microservice/aslan/config" @@ -123,7 +124,7 @@ func CreateReleasePlan(c *handler.Context, args *models.ReleasePlan) error { return nil } -func upsertReleasePlanCron(id, name string, index int64, ScheduleExecuteTime int64) error { +func upsertReleasePlanCron(id, name string, index int64, status config.ReleasePlanStatus, ScheduleExecuteTime int64) error { var ( err error payload *commonservice.CronjobPayload @@ -146,67 +147,110 @@ func upsertReleasePlanCron(id, name string, index int64, ScheduleExecuteTime int found = true } - if found { - origEnabled := releasePlanCron.Enabled - releasePlanCron.Enabled = enable - releasePlanCron.ReleasePlanArgs = &commonmodels.ReleasePlanArgs{ - ID: id, - Name: name, - Index: index, - } - releasePlanCron.Cron = util.UnixStampToCronExpr(ScheduleExecuteTime) + if status != config.StatusExecuting { + // delete cron job if status is not executing + if found { + err = commonrepo.NewCronjobColl().Delete(&commonrepo.CronjobDeleteOption{ + IDList: []string{releasePlanCron.ID.Hex()}, + }) + if err != nil { + fmtErr := fmt.Errorf("Failed to delete release plan schedule job %s, error: %w", releasePlanCron.ID.Hex(), err) + log.Error(fmtErr) + } - err = commonrepo.NewCronjobColl().Upsert(releasePlanCron) - if err != nil { - fmtErr := fmt.Errorf("Failed to upsert cron job, error: %w", err) - log.Error(fmtErr) - return err + payload = &commonservice.CronjobPayload{ + Name: releasePlanCronName, + JobType: setting.ReleasePlanCronjob, + Action: setting.TypeEnableCronjob, + ScheduleType: setting.UnixStampSchedule, + DeleteList: []string{releasePlanCron.ID.Hex()}, + } } + } else { + // upsert cron job if status is executing + if found { + origEnabled := releasePlanCron.Enabled + releasePlanCron.Enabled = enable + releasePlanCron.ReleasePlanArgs = &commonmodels.ReleasePlanArgs{ + ID: id, + Name: name, + Index: index, + } + releasePlanCron.JobType = setting.UnixStampSchedule + releasePlanCron.UnixStamp = ScheduleExecuteTime + + if origEnabled && !enable { + // need to disable cronjob + err = commonrepo.NewCronjobColl().Delete(&commonrepo.CronjobDeleteOption{ + IDList: []string{releasePlanCron.ID.Hex()}, + }) + if err != nil { + fmtErr := fmt.Errorf("Failed to delete cron job %s, error: %w", releasePlanCron.ID.Hex(), err) + log.Error(fmtErr) + } - if origEnabled && !enable { - // need to disable cronjob - payload = &commonservice.CronjobPayload{ - Name: releasePlanCronName, - JobType: setting.ReleasePlanCronjob, - Action: setting.TypeEnableCronjob, - DeleteList: []string{releasePlanCron.ID.Hex()}, + payload = &commonservice.CronjobPayload{ + Name: releasePlanCronName, + JobType: setting.ReleasePlanCronjob, + Action: setting.TypeEnableCronjob, + ScheduleType: setting.UnixStampSchedule, + DeleteList: []string{releasePlanCron.ID.Hex()}, + } + } else if !origEnabled && enable || origEnabled && enable { + err = commonrepo.NewCronjobColl().Upsert(releasePlanCron) + if err != nil { + fmtErr := fmt.Errorf("Failed to upsert cron job, error: %w", err) + log.Error(fmtErr) + return err + } + + payload = &commonservice.CronjobPayload{ + Name: releasePlanCronName, + JobType: setting.ReleasePlanCronjob, + Action: setting.TypeEnableCronjob, + ScheduleType: setting.UnixStampSchedule, + JobList: []*commonmodels.Schedule{cronJobToSchedule(releasePlanCron)}, + } + } else { + // !origEnabled && !enable + return nil + } + } else { + if !enable { + return nil + } + + input := &commonmodels.Cronjob{ + Enabled: enable, + Name: releasePlanCronName, + Type: setting.ReleasePlanCronjob, + JobType: setting.UnixStampSchedule, + UnixStamp: ScheduleExecuteTime, + ReleasePlanArgs: &commonmodels.ReleasePlanArgs{ + ID: id, + Name: name, + Index: index, + }, + } + + err = commonrepo.NewCronjobColl().Upsert(input) + if err != nil { + fmtErr := fmt.Errorf("Failed to upsert cron job, error: %w", err) + log.Error(fmtErr) + return err } - } else if !origEnabled && enable || origEnabled && enable { payload = &commonservice.CronjobPayload{ - Name: releasePlanCronName, - JobType: setting.ReleasePlanCronjob, - Action: setting.TypeEnableCronjob, - JobList: []*commonmodels.Schedule{cronJobToSchedule(releasePlanCron)}, + Name: releasePlanCronName, + JobType: setting.ReleasePlanCronjob, + Action: setting.TypeEnableCronjob, + ScheduleType: setting.UnixStampSchedule, + JobList: []*commonmodels.Schedule{cronJobToSchedule(input)}, } - } else { - // !origEnabled && !enable - return nil - } - } else { - input := &commonmodels.Cronjob{ - Name: releasePlanCronName, - Type: setting.ReleasePlanCronjob, - } - input.Enabled = true - input.Cron = util.UnixStampToCronExpr(ScheduleExecuteTime) - input.ReleasePlanArgs = &commonmodels.ReleasePlanArgs{ - ID: id, - Name: name, - Index: index, } + } - err = commonrepo.NewCronjobColl().Upsert(input) - if err != nil { - fmtErr := fmt.Errorf("Failed to upsert cron job, error: %w", err) - log.Error(fmtErr) - return err - } - payload = &commonservice.CronjobPayload{ - Name: releasePlanCronName, - JobType: setting.ReleasePlanCronjob, - Action: setting.TypeEnableCronjob, - JobList: []*commonmodels.Schedule{cronJobToSchedule(input)}, - } + if payload == nil { + return nil } pl, _ := json.Marshal(payload) @@ -296,6 +340,23 @@ func DeleteReleasePlan(c *gin.Context, username, id string) error { return errors.Wrap(err, "get plan") } internalhandler.InsertOperationLog(c, username, "", "删除", "发布计划", info.Name, "", log.SugaredLogger()) + + releasePlanCronName := util.GetReleasePlanCronName(id, info.Name, info.Index) + releasePlanCron, err := commonrepo.NewCronjobColl().GetByName(releasePlanCronName, setting.ReleasePlanCronjob) + if err != nil { + if err != mongo.ErrNoDocuments && err != mongo.ErrNilDocument { + return e.ErrUpsertCronjob.AddErr(fmt.Errorf("failed to get release plan cron job, err: %w", err)) + } + } else { + err = commonrepo.NewCronjobColl().Delete(&commonrepo.CronjobDeleteOption{ + IDList: []string{releasePlanCron.ID.Hex()}, + }) + if err != nil { + fmtErr := fmt.Errorf("Failed to delete release plan schedule job %s, error: %w", releasePlanCron.ID.Hex(), err) + log.Error(fmtErr) + } + } + return mongodb.NewReleasePlanColl().DeleteByID(context.Background(), id) } @@ -435,11 +496,37 @@ func ExecuteReleaseJob(c *handler.Context, planID string, args *ExecuteReleaseJo return nil } -func ScheduleExecuteReleasePlan(c *handler.Context, planID string) error { +func ScheduleExecuteReleasePlan(c *handler.Context, planID, jobID string) error { approveLock := getLock(planID) approveLock.Lock() defer approveLock.Unlock() + // check if the job is already executed + jobObjectID, err := primitive.ObjectIDFromHex(jobID) + if err != nil { + return errors.Wrap(err, "invalid job ID") + } + _, err = commonrepo.NewCronjobColl().GetByID(jobObjectID) + if err != nil { + if !mongodb.IsErrNoDocuments(err) { + err = fmt.Errorf("Failed to get release job schedule job %s, error: %v", jobID, err) + log.Error(err) + return err + } else { + err = fmt.Errorf("Release job schedule job %s not found", jobID) + log.Error(err) + return err + } + } + + // delete the schedule job after executed + err = commonrepo.NewCronjobColl().Delete(&commonrepo.CronjobDeleteOption{ + IDList: []string{jobID}, + }) + if err != nil { + log.Errorf("Failed to delete release job schedule job %s, error: %v", jobID, err) + } + ctx := context.Background() plan, err := mongodb.NewReleasePlanColl().GetByID(ctx, planID) if err != nil { @@ -449,7 +536,7 @@ func ScheduleExecuteReleasePlan(c *handler.Context, planID string) error { } if plan.Status != config.StatusExecuting { - err = errors.Errorf("plan ID is %s, name is %s, index is %d, status is %s, can not execute", plan.ID, plan.Name, plan.Index, plan.Status) + err = errors.Errorf("plan ID is %s, name is %s, index is %d, status is %s, can not execute", plan.ID.Hex(), plan.Name, plan.Index, plan.Status) log.Error(err) return err } @@ -457,19 +544,19 @@ func ScheduleExecuteReleasePlan(c *handler.Context, planID string) error { if !(plan.StartTime == 0 && plan.EndTime == 0) { now := time.Now().Unix() if now < plan.StartTime || now > plan.EndTime { - err = errors.Errorf("plan ID is %s, name is %s, index is %d, it's not in the release time range", plan.ID, plan.Name, plan.Index) + err = errors.Errorf("plan ID is %s, name is %s, index is %d, it's not in the release time range", plan.ID.Hex(), plan.Name, plan.Index) log.Error(err) return err } } if plan.Approval != nil && plan.Approval.Enabled == true && plan.Approval.Status != config.StatusPassed { - err = errors.Errorf("plan ID is %s, name is %s, index is %d, it's approval status is %s, can not execute", plan.ID, plan.Name, plan.Index, plan.Approval.Status) + err = errors.Errorf("plan ID is %s, name is %s, index is %d, it's approval status is %s, can not execute", plan.ID.Hex(), plan.Name, plan.Index, plan.Approval.Status) log.Error(err) return err } - log.Infof("schedule execute release plan, plan ID: %s, name: %s, index: %d", plan.ID, plan.Name, plan.Index) + log.Infof("schedule execute release plan, plan ID: %s, name: %s, index: %d", plan.ID.Hex(), plan.Name, plan.Index) for _, job := range plan.Jobs { if job.Type == config.JobWorkflow { @@ -482,6 +569,21 @@ func ScheduleExecuteReleasePlan(c *handler.Context, planID string) error { Name: job.Name, Type: string(job.Type), } + + go func() { + if err := mongodb.NewReleasePlanLogColl().Create(&models.ReleasePlanLog{ + PlanID: planID, + Username: "系统", + Account: "", + Verb: VerbExecute, + TargetName: args.Name, + TargetType: TargetTypeReleaseJob, + CreatedAt: time.Now().Unix(), + }); err != nil { + log.Errorf("create release plan log error: %v", err) + } + }() + executor, err := NewReleaseJobExecutor(&ExecuteReleaseJobContext{ AuthResources: c.Resources, UserID: c.UserID, @@ -508,27 +610,13 @@ func ScheduleExecuteReleasePlan(c *handler.Context, planID string) error { plan.Status = config.StatusSuccess } - log.Infof("schedule execute release job, plan ID: %s, name: %s, index: %d, job ID: %s, job name: %s", plan.ID, plan.Name, plan.Index, job.ID, job.Name) + log.Infof("schedule execute release job, plan ID: %s, name: %s, index: %d, job ID: %s, job name: %s", plan.ID.Hex(), plan.Name, plan.Index, job.ID, job.Name) if err = mongodb.NewReleasePlanColl().UpdateByID(ctx, planID, plan); err != nil { err = errors.Wrap(err, "update plan") log.Error(err) return err } - - go func() { - if err := mongodb.NewReleasePlanLogColl().Create(&models.ReleasePlanLog{ - PlanID: planID, - Username: "系统", - Account: "", - Verb: VerbExecute, - TargetName: args.Name, - TargetType: TargetTypeReleaseJob, - CreatedAt: time.Now().Unix(), - }); err != nil { - log.Errorf("create release plan log error: %v", err) - } - }() } } @@ -652,10 +740,6 @@ func UpdateReleasePlanStatus(c *handler.Context, planID, status string, isSystem return errors.Errorf("approval status is %s, can not execute", plan.Approval.Status) } - if err := upsertReleasePlanCron(plan.ID.Hex(), plan.Name, plan.Index, plan.ScheduleExecuteTime); err != nil { - return errors.Wrap(err, "upsert release plan cron") - } - plan.ExecutingTime = time.Now().Unix() setReleaseJobsForExecuting(plan) case config.StatusWaitForApprove: @@ -685,6 +769,10 @@ func UpdateReleasePlanStatus(c *handler.Context, planID, status string, isSystem return errors.Wrap(err, "update plan") } + if err := upsertReleasePlanCron(plan.ID.Hex(), plan.Name, plan.Index, plan.Status, plan.ScheduleExecuteTime); err != nil { + return errors.Wrap(err, "upsert release plan cron") + } + go func() { if err := mongodb.NewReleasePlanLogColl().Create(&models.ReleasePlanLog{ PlanID: planID, @@ -773,7 +861,7 @@ func ApproveReleasePlan(c *handler.Context, planID string, req *ApproveRequest) plan.Status = config.StatusExecuting plan.ApprovalTime = time.Now().Unix() - if err := upsertReleasePlanCron(plan.ID.Hex(), plan.Name, plan.Index, plan.ScheduleExecuteTime); err != nil { + if err := upsertReleasePlanCron(plan.ID.Hex(), plan.Name, plan.Index, plan.Status, plan.ScheduleExecuteTime); err != nil { err = errors.Wrap(err, "upsert release plan cron") log.Error(err) } @@ -875,6 +963,7 @@ func cronJobToSchedule(input *commonmodels.Cronjob) *commonmodels.Schedule { return &commonmodels.Schedule{ ID: input.ID, Number: input.Number, + UnixStamp: input.UnixStamp, Frequency: input.Frequency, Time: input.Time, MaxFailures: input.MaxFailure, @@ -891,6 +980,7 @@ const ( ListReleasePlanTypeName ListReleasePlanType = "name" ListReleasePlanTypeManager ListReleasePlanType = "manager" ListReleasePlanTypeSuccessTime ListReleasePlanType = "success_time" + ListReleasePlanTypeUpdateTime ListReleasePlanType = "update_time" ListReleasePlanTypeStatus ListReleasePlanType = "status" ) @@ -950,10 +1040,37 @@ func ListReleasePlans(opt *ListReleasePlanOption) (*ListReleasePlanResp, error) SuccessTimeStart: timeStart, SuccessTimeEnd: timeEnd, IsSort: true, + SortBy: mongodb.SortReleasePlanByUpdateTime, PageNum: opt.PageNum, PageSize: opt.PageSize, ExcludedFields: []string{"jobs", "logs"}, }) + case ListReleasePlanTypeUpdateTime: + timeArr := strings.Split(opt.Keyword, "-") + if len(timeArr) != 2 { + return nil, errors.New("invalid update time range") + } + + timeStart := int64(0) + timeEnd := int64(0) + timeStart, err = strconv.ParseInt(timeArr[0], 10, 64) + if err != nil { + return nil, errors.Wrap(err, "invalid update time start") + } + timeEnd, err = strconv.ParseInt(timeArr[1], 10, 64) + if err != nil { + return nil, errors.Wrap(err, "invalid update time end") + } + + list, total, err = mongodb.NewReleasePlanColl().ListByOptions(&mongodb.ListReleasePlanOption{ + UpdateTimeStart: timeStart, + UpdateTimeEnd: timeEnd, + IsSort: true, + SortBy: mongodb.SortReleasePlanByUpdateTime, + PageNum: opt.PageNum, + PageSize: opt.PageSize, + ExcludedFields: []string{"jobs", "logs"}, + }) case ListReleasePlanTypeStatus: list, total, err = mongodb.NewReleasePlanColl().ListByOptions(&mongodb.ListReleasePlanOption{ Status: config.ReleasePlanStatus(opt.Keyword), diff --git a/pkg/microservice/aslan/core/release_plan/service/watcher.go b/pkg/microservice/aslan/core/release_plan/service/watcher.go index a3dba99232..8d4109ff43 100644 --- a/pkg/microservice/aslan/core/release_plan/service/watcher.go +++ b/pkg/microservice/aslan/core/release_plan/service/watcher.go @@ -200,7 +200,7 @@ func updatePlanApproval(plan *models.ReleasePlan) error { plan.Status = config.StatusExecuting plan.ApprovalTime = time.Now().Unix() - if err := upsertReleasePlanCron(plan.ID.Hex(), plan.Name, plan.Index, plan.ScheduleExecuteTime); err != nil { + if err := upsertReleasePlanCron(plan.ID.Hex(), plan.Name, plan.Index, plan.Status, plan.ScheduleExecuteTime); err != nil { err = errors.Wrap(err, "upsert release plan cron") log.Error(err) } diff --git a/pkg/microservice/aslan/core/service.go b/pkg/microservice/aslan/core/service.go index 7f881797ee..7e83f99192 100644 --- a/pkg/microservice/aslan/core/service.go +++ b/pkg/microservice/aslan/core/service.go @@ -23,7 +23,7 @@ import ( "sync" "time" - newgoCron "github.com/go-co-op/gocron" + newgoCron "github.com/go-co-op/gocron/v2" _ "github.com/go-sql-driver/mysql" "github.com/hashicorp/go-multierror" "github.com/koderover/zadig/v2/pkg/tool/clientmanager" @@ -156,9 +156,13 @@ func Stop(ctx context.Context) { var Scheduler *newgoCron.Scheduler func initCron() { - Scheduler = newgoCron.NewScheduler(time.Local) + Scheduler, err := newgoCron.NewScheduler() + if err != nil { + log.Fatalf("failed to create scheduler: %v", err) + return + } - Scheduler.Every(5).Minutes().Do(func() { + Scheduler.NewJob(newgoCron.DurationJob(5*time.Minute), newgoCron.NewTask(func() { log.Infof("[CRONJOB] updating tokens for gitlab....") codehostList, err := mongodb2.NewCodehostColl().List(&mongodb2.ListArgs{ Source: "gitlab", @@ -175,11 +179,11 @@ func initCron() { } } log.Infof("[CRONJOB] gitlab token updated....") - }) + })) - Scheduler.Every(1).Days().At("04:00").Do(cleanCacheFiles) + Scheduler.NewJob(newgoCron.DailyJob(1, newgoCron.NewAtTimes(newgoCron.NewAtTime(4, 0, 0))), newgoCron.NewTask(cleanCacheFiles)) - Scheduler.StartAsync() + Scheduler.Start() } func initService() { diff --git a/pkg/microservice/aslan/core/workflow/service/workflow/workflow_v4.go b/pkg/microservice/aslan/core/workflow/service/workflow/workflow_v4.go index d2bae88446..84581c04ca 100644 --- a/pkg/microservice/aslan/core/workflow/service/workflow/workflow_v4.go +++ b/pkg/microservice/aslan/core/workflow/service/workflow/workflow_v4.go @@ -2083,6 +2083,7 @@ func cronJobToSchedule(input *commonmodels.Cronjob) *commonmodels.Schedule { ID: input.ID, Number: input.Number, Frequency: input.Frequency, + UnixStamp: input.UnixStamp, Time: input.Time, MaxFailures: input.MaxFailure, WorkflowV4Args: input.WorkflowV4Args, diff --git a/pkg/microservice/cron/core/service/cronjob.go b/pkg/microservice/cron/core/service/cronjob.go index 91184d97b6..fab49e1117 100644 --- a/pkg/microservice/cron/core/service/cronjob.go +++ b/pkg/microservice/cron/core/service/cronjob.go @@ -17,12 +17,13 @@ limitations under the License. package service type CronjobPayload struct { - Name string `json:"name"` - ProductName string `json:"product_name"` - Action string `json:"action"` - JobType string `json:"job_type"` - DeleteList []string `json:"delete_list,omitempty"` - JobList []*Schedule `json:"job_list,omitempty"` + Name string `json:"name"` + ProductName string `json:"product_name"` + Action string `json:"action"` + JobType string `json:"job_type"` + ScheduleType string `json:"schedule_type"` + DeleteList []string `json:"delete_list,omitempty"` + JobList []*Schedule `json:"job_list,omitempty"` } type Cronjob struct { @@ -31,6 +32,7 @@ type Cronjob struct { Type string `json:"type"` Number uint64 `json:"number"` Frequency string `json:"frequency"` + UnixStamp int64 `json:"unix_stamp"` Time string `json:"time"` Cron string `json:"cron"` ProductName string `json:"product_name,omitempty"` diff --git a/pkg/microservice/cron/core/service/scheduler/cronjob_handler.go b/pkg/microservice/cron/core/service/scheduler/cronjob_handler.go index 9fb85261f1..6d78bfefa2 100644 --- a/pkg/microservice/cron/core/service/scheduler/cronjob_handler.go +++ b/pkg/microservice/cron/core/service/scheduler/cronjob_handler.go @@ -26,6 +26,7 @@ import ( "strings" "time" + newgoCron "github.com/go-co-op/gocron/v2" "github.com/rfyiamcool/cronlib" "github.com/koderover/zadig/v2/pkg/microservice/cron/core/service" @@ -42,20 +43,22 @@ const ( ) type CronjobHandler struct { - aslanCli *client.Client - Scheduler *cronlib.CronSchduler + aslanCli *client.Client + Scheduler *cronlib.CronSchduler + NewScheduler newgoCron.Scheduler } -func NewCronjobHandler(client *client.Client, scheduler *cronlib.CronSchduler) *CronjobHandler { - InitExistedCronjob(client, scheduler) +func NewCronjobHandler(client *client.Client, scheduler *cronlib.CronSchduler, newScheduler newgoCron.Scheduler) *CronjobHandler { + InitExistedCronjob(client, scheduler, newScheduler) return &CronjobHandler{ - aslanCli: client, - Scheduler: scheduler, + aslanCli: client, + Scheduler: scheduler, + NewScheduler: newScheduler, } } -func InitExistedCronjob(client *client.Client, scheduler *cronlib.CronSchduler) { +func InitExistedCronjob(client *client.Client, scheduler *cronlib.CronSchduler, newScheduler newgoCron.Scheduler) { log.Infof("Initializing existing cronjob ....") initChan := make(chan []*service.Cronjob, 1) @@ -99,7 +102,7 @@ func InitExistedCronjob(client *client.Client, scheduler *cronlib.CronSchduler) select { case jobList = <-initChan: for _, job := range jobList { - err := registerCronjob(job, client, scheduler) + err := registerCronjob(job, client, scheduler, newScheduler) if err != nil { fmt.Printf("Failed to init job with id: %s, err: %s\n", job.ID, err) } @@ -111,7 +114,7 @@ func InitExistedCronjob(client *client.Client, scheduler *cronlib.CronSchduler) select { case jobList = <-failsafeChan: for _, job := range jobList { - err := registerCronjob(job, client, scheduler) + err := registerCronjob(job, client, scheduler, newScheduler) if err != nil { fmt.Printf("Failed to init job with id: %s, err: %s\n", job.ID, err) } @@ -128,77 +131,97 @@ func (h *CronjobHandler) HandleMessage(msgs []*service.CronjobPayload) error { for _, msg := range msgs { switch msg.Action { case setting.TypeEnableCronjob: - err := h.updateCronjob(msg.Name, msg.ProductName, msg.JobType, msg.JobList, msg.DeleteList) + err := h.updateCronjob(msg.Name, msg.ProductName, msg.ScheduleType, msg.JobType, msg.JobList, msg.DeleteList) if err != nil { log.Errorf("Failed to update cronjob, the error is: %v", err) return err } case setting.TypeDisableCronjob: - err := h.stopCronjob(msg.Name, msg.JobType) + err := h.stopCronjob(msg.Name, msg.JobType, msg.ScheduleType) if err != nil { log.Errorf("Failed to stop all cron job, the error is: %v", err) return err } default: - log.Errorf("unsupported cronjob action: NOT RECONSUMING") + log.Errorf("unsupported cronjob action: %+v", msg) } } return nil } -func (h *CronjobHandler) updateCronjob(name, productName, jobType string, jobList []*service.Schedule, deleteList []string) error { - //首先根据deleteList停止不需要的cronjob - for _, deleteID := range deleteList { - jobID := deleteID - log.Infof("stopping Job of ID: %s", jobID) - h.Scheduler.StopService(jobID) - } - // 根据job内容来在scheduler中新增cronjob - for _, job := range jobList { - var cron string - if job.Type == setting.FixedGapCronjob || job.Type == setting.FixedDayTimeCronjob { - cronString, err := convertFixedTimeToCron(job) - if err != nil { - return err - } - cron = cronString - } else { - cron = fmt.Sprintf("%s%s", "0 ", job.Cron) +func (h *CronjobHandler) updateCronjob(name, productName, scheduleType, jobType string, jobList []*service.Schedule, deleteList []string) error { + if scheduleType == setting.UnixStampSchedule { + //首先根据deleteList停止不需要的cronjob + for _, deleteID := range deleteList { + jobID := deleteID + log.Infof("stopping UnixStamp Schedule Job of ID: %s", jobID) + h.NewScheduler.RemoveByTags(jobID) } - switch jobType { - case setting.WorkflowCronjob: - err := h.registerWorkFlowJob(name, cron, job) - if err != nil { - return err - } - case setting.TestingCronjob: - err := h.registerTestJob(name, productName, cron, job) - if err != nil { - return err - } - case setting.WorkflowV4Cronjob: - err := h.registerWorkFlowV4Job(name, cron, job) - if err != nil { - return err - } - case setting.EnvAnalysisCronjob: - err := h.registerEnvAnalysisJob(name, cron, job) - if err != nil { - return err + + // 根据job内容来在scheduler中新增cronjob + for _, job := range jobList { + switch jobType { + case setting.ReleasePlanCronjob: + err := h.registerReleasePlanJob(name, job) + if err != nil { + return err + } + default: + log.Errorf("unrecognized cron job type for job id: %s", job.ID) } - case setting.EnvSleepCronjob: - err := h.registerEnvSleepJob(name, cron, job) - if err != nil { - return err + } + + } else { + //首先根据deleteList停止不需要的cronjob + for _, deleteID := range deleteList { + jobID := deleteID + log.Infof("stopping Job of ID: %s", jobID) + h.Scheduler.StopService(jobID) + } + + // 根据job内容来在scheduler中新增cronjob + for _, job := range jobList { + var cron string + if job.Type == setting.FixedGapCronjob || job.Type == setting.FixedDayTimeCronjob { + cronString, err := convertFixedTimeToCron(job) + if err != nil { + return err + } + cron = cronString + } else { + cron = fmt.Sprintf("%s%s", "0 ", job.Cron) } - case setting.ReleasePlanCronjob: - err := h.registerReleasePlanJob(name, cron, job) - if err != nil { - return err + + switch jobType { + case setting.WorkflowCronjob: + err := h.registerWorkFlowJob(name, cron, job) + if err != nil { + return err + } + case setting.TestingCronjob: + err := h.registerTestJob(name, productName, cron, job) + if err != nil { + return err + } + case setting.WorkflowV4Cronjob: + err := h.registerWorkFlowV4Job(name, cron, job) + if err != nil { + return err + } + case setting.EnvAnalysisCronjob: + err := h.registerEnvAnalysisJob(name, cron, job) + if err != nil { + return err + } + case setting.EnvSleepCronjob: + err := h.registerEnvSleepJob(name, cron, job) + if err != nil { + return err + } + default: + log.Errorf("unrecognized cron job type for job id: %s", job.ID) } - default: - log.Errorf("unrecognized cron job type for job id: %s", job.ID) } } return nil @@ -333,7 +356,7 @@ func (h *CronjobHandler) registerTestJob(name, productName, schedule string, job // FIXME // UNDER CURRENT SERVICE STRUCTURE, STOPPING CRONJOB SERVICE AND UPDATING DB RECORD // ARE NOT ATOMIC, THIS WILL CAUSE SERIOUS PROBLEM IF UPDATE FAILED -func (h *CronjobHandler) stopCronjob(name, ptype string) error { +func (h *CronjobHandler) stopCronjob(name, ptype, scheduleType string) error { var jobList []*service.Cronjob listAPI := fmt.Sprintf("%s/cron/cronjob/type/%s/name/%s", h.aslanCli.APIBase, ptype, name) header := http.Header{} @@ -348,9 +371,16 @@ func (h *CronjobHandler) stopCronjob(name, ptype string) error { return err } - for _, job := range jobList { - log.Infof("stopping cronjob of ID: %s", job.ID) - h.Scheduler.StopService(job.ID) + if scheduleType == setting.UnixStampSchedule { + for _, job := range jobList { + log.Infof("stopping unixstamp schedule job of ID: %s", job.ID) + h.NewScheduler.RemoveByTags(job.ID) + } + } else { + for _, job := range jobList { + log.Infof("stopping cronjob of ID: %s", job.ID) + h.Scheduler.StopService(job.ID) + } } disableAPI := fmt.Sprintf("%s/cron/cronjob/disable", h.aslanCli.APIBase) @@ -369,206 +399,195 @@ func (h *CronjobHandler) stopCronjob(name, ptype string) error { return nil } -func registerCronjob(job *service.Cronjob, client *client.Client, scheduler *cronlib.CronSchduler) error { - switch job.Type { - case setting.WorkflowCronjob: - args := &service.WorkflowTaskArgs{ - WorkflowName: job.Name, - WorklowTaskCreator: setting.CronTaskCreator, - } - if job.WorkflowArgs != nil { - args.Description = job.WorkflowArgs.Description - args.ProductTmplName = job.WorkflowArgs.ProductTmplName - args.Target = job.WorkflowArgs.Target - args.Namespace = job.WorkflowArgs.Namespace - args.Tests = job.WorkflowArgs.Tests - args.DistributeEnabled = job.WorkflowArgs.DistributeEnabled - } - var cron string - if job.JobType == setting.CrontabCronjob { - cron = fmt.Sprintf("%s%s", "0 ", job.Cron) - } else { - cron, _ = convertCronString(job.JobType, job.Time, job.Frequency, job.Number) - } - scheduleJob, err := cronlib.NewJobModel(cron, func() { - if err := client.ScheduleCall(path.Join("workflow/workflowtask", job.WorkflowArgs.WorkflowName), args, log.SugaredLogger()); err != nil { - log.Errorf("[%s]RunScheduledTask err: %v", job.Name, err) - } - }) - if err != nil { - log.Errorf("Failed to generate job of ID: %s to scheduler, the error is: %v", job.ID, err) - return err - } - log.Infof("registering jobID: %s with cron: %s", job.ID, cron) - err = scheduler.UpdateJobModel(job.ID, scheduleJob) - if err != nil { - log.Errorf("Failed to register job of ID: %s to scheduler, the error is: %v", job.ID, err) - return err - } - case setting.WorkflowV4Cronjob: - if job.WorkflowV4Args == nil { - return fmt.Errorf("workflow args is nil") - } - var cron string - if job.JobType == setting.CrontabCronjob { - cron = fmt.Sprintf("%s%s", "0 ", job.Cron) - } else { - cron, _ = convertCronString(job.JobType, job.Time, job.Frequency, job.Number) - } - scheduleJob, err := cronlib.NewJobModel(cron, func() { - if err := client.ScheduleCall(fmt.Sprintf("workflow/v4/workflowtask/trigger?triggerName=%s", setting.CronTaskCreator), job.WorkflowV4Args, log.SugaredLogger()); err != nil { - log.Errorf("[%s]RunScheduledTask err: %v", job.Name, err) - } - }) - if err != nil { - log.Errorf("Failed to generate job of ID: %s to scheduler, the error is: %v", job.ID, err) - return err - } - log.Infof("registering jobID: %s with cron: %s", job.ID, cron) - err = scheduler.UpdateJobModel(job.ID, scheduleJob) - if err != nil { - log.Errorf("Failed to register job of ID: %s to scheduler, the error is: %v", job.ID, err) - return err - } - case setting.TestingCronjob: - args := &service.TestTaskArgs{ - TestName: job.Name, - ProductName: job.ProductName, - TestTaskCreator: setting.CronTaskCreator, - } - var cron string - if job.JobType == setting.CrontabCronjob { - cron = fmt.Sprintf("%s%s", "0 ", job.Cron) - } else { - cron, _ = convertCronString(job.JobType, job.Time, job.Frequency, job.Number) - } - scheduleJob, err := cronlib.NewJobModel(cron, func() { - if err := client.ScheduleCall("testing/testtask", args, log.SugaredLogger()); err != nil { - log.Errorf("[%s]RunScheduledTask err: %v", job.Name, err) +func registerCronjob(job *service.Cronjob, client *client.Client, scheduler *cronlib.CronSchduler, newScheduler newgoCron.Scheduler) error { + if job.JobType == setting.UnixStampSchedule { + switch job.Type { + case setting.ReleasePlanCronjob: + if job.ReleasePlanArgs == nil { + log.Errorf("ReleasePlanArgs is nil, name: %v, jobID: %v", job.Name, job.ID) + return nil } - }) - if err != nil { - log.Errorf("Failed to generate job of ID: %s to scheduler, the error is: %v", job.ID, err) - return err - } - log.Infof("registering jobID: %s with cron: %s", job.ID, cron) - err = scheduler.UpdateJobModel(job.ID, scheduleJob) - if err != nil { - log.Errorf("Failed to register job of ID: %s to scheduler, the error is: %v", job.ID, err) - return err - } - case setting.EnvAnalysisCronjob: - if job.EnvAnalysisArgs == nil { - return nil - } - var cron string - if job.JobType == "" || job.JobType == setting.CrontabCronjob { - cron = fmt.Sprintf("%s%s", "0 ", job.Cron) - } else { - cron, _ = convertCronString(job.JobType, job.Time, job.Frequency, job.Number) - } + executeReleasePlanFunc := func() { + base := "release_plan/v1" + url := base + fmt.Sprintf("/%s/schedule_execute?jobID=%s", job.ReleasePlanArgs.ID, job.ID) + if err := client.ScheduleCall(url, nil, log.SugaredLogger()); err != nil { + log.Errorf("[%s] RunScheduledTask err: %v", job.Name, err) + } - scheduleJob, err := cronlib.NewJobModel(cron, func() { - base := "environment/environments/" - production := "false" - if job.EnvAnalysisArgs.Production { - production = "true" + log.Infof("schedule executed release plan, jobID: %v, schdule time: %v; release plan ID: %v, index: %v, name: %v", job.ID, time.Unix(job.UnixStamp, 0), job.ReleasePlanArgs.ID, job.ReleasePlanArgs.Index, job.ReleasePlanArgs.Name) } - url := base + fmt.Sprintf("%s/analysis?projectName=%s&triggerName=%s&userName=%s&production=%s", job.EnvAnalysisArgs.EnvName, job.EnvAnalysisArgs.ProductName, setting.CronTaskCreator, setting.CronTaskCreator, production) - - if err := client.ScheduleCall(url, nil, log.SugaredLogger()); err != nil { - log.Errorf("[%s]RunScheduledTask err: %v", job.Name, err) + // delete old schedule job first + tag := job.ID + newScheduler.RemoveByTags(tag) + + scheduleTime := time.Unix(job.UnixStamp, 0) + jobName := fmt.Sprintf("release_plan:%s:%d:%s", job.ReleasePlanArgs.Name, job.ReleasePlanArgs.Index, scheduleTime) + if scheduleTime.Before(time.Now()) { + if time.Now().Sub(scheduleTime) <= time.Second*30 { + // now - schedule time <= 30s + // maybe missed the schedule time because of cron service restart + // so start this immediately + log.Infof("found an release plan outdated <= 30s, start it immediately, jobID: %v, schedule time: %v; release plan ID: %v, index: %v, name: %v", job.ID, scheduleTime, job.ReleasePlanArgs.ID, job.ReleasePlanArgs.Index, job.ReleasePlanArgs.Name) + _, err := newScheduler.NewJob(newgoCron.OneTimeJob(newgoCron.OneTimeJobStartImmediately()), newgoCron.NewTask(executeReleasePlanFunc), newgoCron.WithTags(tag), newgoCron.WithName(jobName)) + if err != nil { + log.Errorf("Failed to create jobID: %s, jobName: %v, schedule time: %v; release plan ID: %v, index: %v, name: %v, error: %v", job.ID, job.Name, scheduleTime, job.ReleasePlanArgs.ID, job.ReleasePlanArgs.Index, job.ReleasePlanArgs.Name, err) + return err + } + } else { + // now - schedule time > 30s + // schedule time is too old + log.Errorf("found an release plan outdate > 30s, drop it, jobID: %v, schedule time: %v; release plan ID: %v, index: %v, name: %v", job.ID, scheduleTime, job.ReleasePlanArgs.ID, job.ReleasePlanArgs.Index, job.ReleasePlanArgs.Name) + return nil + } + } else { + // schedule time correct + _, err := newScheduler.NewJob(newgoCron.OneTimeJob(newgoCron.OneTimeJobStartDateTime(scheduleTime)), newgoCron.NewTask(executeReleasePlanFunc), newgoCron.WithTags(tag), newgoCron.WithName(jobName)) + if err != nil { + log.Errorf("Failed to create jobID: %s, jobName: %v, schedule time: %v; release plan ID: %v, index: %v, name: %v, error: %v", job.ID, job.Name, scheduleTime, job.ReleasePlanArgs.ID, job.ReleasePlanArgs.Index, job.ReleasePlanArgs.Name, err) + return err + } } - }) - if err != nil { - log.Errorf("Failed to create job of ID: %s, the error is: %v", job.ID, err) - return err - } - log.Infof("registering jobID: %s with cron: %s", job.ID, cron) - err = scheduler.UpdateJobModel(job.ID, scheduleJob) - if err != nil { - log.Errorf("Failed to register job of ID: %s to scheduler, the error is: %v", job.ID, err) - return err - } - case setting.EnvSleepCronjob: - if job.EnvArgs == nil { + log.Infof("registering jobID: %s with name: %v, schedule time: %v; release plan ID: %v, index: %v, name: %v", job.ID, job.Name, scheduleTime, job.ReleasePlanArgs.ID, job.ReleasePlanArgs.Index, job.ReleasePlanArgs.Name) return nil } - var cron string - if job.JobType == "" || job.JobType == setting.CrontabCronjob { - cron = fmt.Sprintf("%s%s", "0 ", job.Cron) - } else { - cron, _ = convertCronString(job.JobType, job.Time, job.Frequency, job.Number) - } - scheduleJob, err := cronlib.NewJobModel(cron, func() { - base := "environment/environments/" - production := "false" - if job.EnvArgs.Production { - production = "true" + } else { + switch job.Type { + case setting.WorkflowCronjob: + return nil + case setting.WorkflowV4Cronjob: + if job.WorkflowV4Args == nil { + return fmt.Errorf("workflow args is nil") } - - url := "" - if job.EnvArgs.Name == util.GetEnvSleepCronName(job.EnvArgs.ProductName, job.EnvArgs.EnvName, true) { - url = base + fmt.Sprintf("%s/sleep?projectName=%s&action=enable&production=%s", job.EnvArgs.EnvName, job.EnvArgs.ProductName, production) - } else if job.EnvArgs.Name == util.GetEnvSleepCronName(job.EnvArgs.ProductName, job.EnvArgs.EnvName, false) { - url = base + fmt.Sprintf("%s/sleep?projectName=%s&action=disable&production=%s", job.EnvArgs.EnvName, job.EnvArgs.ProductName, production) + var cron string + if job.JobType == setting.CrontabCronjob { + cron = fmt.Sprintf("%s%s", "0 ", job.Cron) + } else { + cron, _ = convertCronString(job.JobType, job.Time, job.Frequency, job.Number) + } + scheduleJob, err := cronlib.NewJobModel(cron, func() { + if err := client.ScheduleCall(fmt.Sprintf("workflow/v4/workflowtask/trigger?triggerName=%s", setting.CronTaskCreator), job.WorkflowV4Args, log.SugaredLogger()); err != nil { + log.Errorf("[%s]RunScheduledTask err: %v", job.Name, err) + } + }) + if err != nil { + log.Errorf("Failed to generate job of ID: %s to scheduler, the error is: %v", job.ID, err) + return err + } + log.Infof("registering jobID: %s with cron: %s", job.ID, cron) + err = scheduler.UpdateJobModel(job.ID, scheduleJob) + if err != nil { + log.Errorf("Failed to register job of ID: %s to scheduler, the error is: %v", job.ID, err) + return err + } + case setting.TestingCronjob: + args := &service.TestTaskArgs{ + TestName: job.Name, + ProductName: job.ProductName, + TestTaskCreator: setting.CronTaskCreator, + } + var cron string + if job.JobType == setting.CrontabCronjob { + cron = fmt.Sprintf("%s%s", "0 ", job.Cron) + } else { + cron, _ = convertCronString(job.JobType, job.Time, job.Frequency, job.Number) + } + scheduleJob, err := cronlib.NewJobModel(cron, func() { + if err := client.ScheduleCall("testing/testtask", args, log.SugaredLogger()); err != nil { + log.Errorf("[%s]RunScheduledTask err: %v", job.Name, err) + } + }) + if err != nil { + log.Errorf("Failed to generate job of ID: %s to scheduler, the error is: %v", job.ID, err) + return err + } + log.Infof("registering jobID: %s with cron: %s", job.ID, cron) + err = scheduler.UpdateJobModel(job.ID, scheduleJob) + if err != nil { + log.Errorf("Failed to register job of ID: %s to scheduler, the error is: %v", job.ID, err) + return err + } + case setting.EnvAnalysisCronjob: + if job.EnvAnalysisArgs == nil { + return nil } - if err := client.ScheduleCall(url, nil, log.SugaredLogger()); err != nil { - log.Errorf("[%s]RunScheduledTask err: %v", job.Name, err) + var cron string + if job.JobType == "" || job.JobType == setting.CrontabCronjob { + cron = fmt.Sprintf("%s%s", "0 ", job.Cron) + } else { + cron, _ = convertCronString(job.JobType, job.Time, job.Frequency, job.Number) } - }) - if err != nil { - log.Errorf("Failed to create job of ID: %s, the error is: %v", job.ID, err) - return err - } - log.Infof("registering jobID: %s with cron: %s", job.ID, cron) - err = scheduler.UpdateJobModel(job.ID, scheduleJob) - if err != nil { - log.Errorf("Failed to register job of ID: %s to scheduler, the error is: %v", job.ID, err) - return err - } - case setting.ReleasePlanCronjob: - var cron string - if job.JobType == "" || job.JobType == setting.CrontabCronjob { - cron = fmt.Sprintf("%s%s", "0 ", job.Cron) - } else { - cron, _ = convertCronString(job.JobType, job.Time, job.Frequency, job.Number) - } + scheduleJob, err := cronlib.NewJobModel(cron, func() { + base := "environment/environments/" + production := "false" + if job.EnvAnalysisArgs.Production { + production = "true" + } - if job.ReleasePlanArgs == nil { - log.Errorf("ReleasePlanArgs is nil, name: %v, schedule: %v, jobID: %v", job.Name, cron, job.ID) - return nil - } - scheduleJob, err := cronlib.NewJobModel(cron, func() { - base := "release_plan/v1" - url := base + fmt.Sprintf("/%s/schedule_execute", job.ReleasePlanArgs.ID) - if err := client.ScheduleCall(url, nil, log.SugaredLogger()); err != nil { - log.Errorf("[%s]RunScheduledTask err: %v", job.Name, err) - } + url := base + fmt.Sprintf("%s/analysis?projectName=%s&triggerName=%s&userName=%s&production=%s", job.EnvAnalysisArgs.EnvName, job.EnvAnalysisArgs.ProductName, setting.CronTaskCreator, setting.CronTaskCreator, production) - scheduler.StopService(job.ID) + if err := client.ScheduleCall(url, nil, log.SugaredLogger()); err != nil { + log.Errorf("[%s]RunScheduledTask err: %v", job.Name, err) + } + }) + if err != nil { + log.Errorf("Failed to create job of ID: %s, the error is: %v", job.ID, err) + return err + } - log.Infof("schedule executed release plan, jobID: %v, cron: %v; release plan ID: %v, index: %v, name: %v", job.ID, job.Cron, job.ReleasePlanArgs.ID, job.ReleasePlanArgs.Index, job.ReleasePlanArgs.Name) - }) - if err != nil { - log.Errorf("Failed to create jobID: %s, jobName: %v, cron: %v; release plan ID: %v, index: %v, name: %v, error: %v", job.ID, job.Name, cron, job.ReleasePlanArgs.ID, job.ReleasePlanArgs.Index, job.ReleasePlanArgs.Name, err) - return err - } + log.Infof("registering jobID: %s with cron: %s", job.ID, cron) + err = scheduler.UpdateJobModel(job.ID, scheduleJob) + if err != nil { + log.Errorf("Failed to register job of ID: %s to scheduler, the error is: %v", job.ID, err) + return err + } + case setting.EnvSleepCronjob: + if job.EnvArgs == nil { + return nil + } + var cron string + if job.JobType == "" || job.JobType == setting.CrontabCronjob { + cron = fmt.Sprintf("%s%s", "0 ", job.Cron) + } else { + cron, _ = convertCronString(job.JobType, job.Time, job.Frequency, job.Number) + } + scheduleJob, err := cronlib.NewJobModel(cron, func() { + base := "environment/environments/" + production := "false" + if job.EnvArgs.Production { + production = "true" + } + + url := "" + if job.EnvArgs.Name == util.GetEnvSleepCronName(job.EnvArgs.ProductName, job.EnvArgs.EnvName, true) { + url = base + fmt.Sprintf("%s/sleep?projectName=%s&action=enable&production=%s", job.EnvArgs.EnvName, job.EnvArgs.ProductName, production) + } else if job.EnvArgs.Name == util.GetEnvSleepCronName(job.EnvArgs.ProductName, job.EnvArgs.EnvName, false) { + url = base + fmt.Sprintf("%s/sleep?projectName=%s&action=disable&production=%s", job.EnvArgs.EnvName, job.EnvArgs.ProductName, production) + } + + if err := client.ScheduleCall(url, nil, log.SugaredLogger()); err != nil { + log.Errorf("[%s]RunScheduledTask err: %v", job.Name, err) + } + }) + if err != nil { + log.Errorf("Failed to create job of ID: %s, the error is: %v", job.ID, err) + return err + } - log.Infof("registering jobID: %s with name: %v, cron: %v; release plan ID: %v, index: %v, name: %v", job.ID, job.Name, cron, job.ReleasePlanArgs.ID, job.ReleasePlanArgs.Index, job.ReleasePlanArgs.Name) - err = scheduler.UpdateJobModel(job.ID, scheduleJob) - if err != nil { - log.Errorf("Failed to register job of ID: %s to scheduler, the error is: %v", job.ID, err) - return err + log.Infof("registering jobID: %s with cron: %s", job.ID, cron) + err = scheduler.UpdateJobModel(job.ID, scheduleJob) + if err != nil { + log.Errorf("Failed to register job of ID: %s to scheduler, the error is: %v", job.ID, err) + return err + } + default: + fmt.Printf("Not supported type of service: %s\n", job.Type) + return errors.New("not supported service type") } - default: - fmt.Printf("Not supported type of service: %s\n", job.Type) - return errors.New("not supported service type") } return nil } @@ -640,32 +659,39 @@ func (h *CronjobHandler) registerEnvSleepJob(name, schedule string, job *service return nil } -func (h *CronjobHandler) registerReleasePlanJob(name, schedule string, job *service.Schedule) error { +func (h *CronjobHandler) registerReleasePlanJob(name string, job *service.Schedule) error { if job.ReleasePlanArgs == nil { - log.Errorf("ReleasePlanArgs is nil, name: %v, schedule: %v, jobID: %v", name, schedule, job.ID.Hex()) + log.Errorf("ReleasePlanArgs is nil, name: %v, jobID: %v", name, job.ID.Hex()) return nil } - scheduleJob, err := cronlib.NewJobModel(schedule, func() { + + executeReleasePlanFunc := func() { base := "release_plan/v1" - url := base + fmt.Sprintf("/%s/schedule_execute", job.ReleasePlanArgs.ID) + url := base + fmt.Sprintf("/%s/schedule_execute?jobID=%s", job.ReleasePlanArgs.ID, job.ID.Hex()) if err := h.aslanCli.ScheduleCall(url, nil, log.SugaredLogger()); err != nil { log.Errorf("[%s]RunScheduledTask err: %v", name, err) } - h.Scheduler.StopService(job.ID.Hex()) + log.Infof("schedule executed release plan, jobID: %v, schedule time: %v; release plan ID: %v, index: %v, name: %v", job.ID.Hex(), time.Unix(job.UnixStamp, 0), job.ReleasePlanArgs.ID, job.ReleasePlanArgs.Index, job.ReleasePlanArgs.Name) + } - log.Infof("schedule executed release plan, jobID: %v, cron: %v; release plan ID: %v, index: %v, name: %v", job.ID.Hex(), job.Cron, job.ReleasePlanArgs.ID, job.ReleasePlanArgs.Index, job.ReleasePlanArgs.Name) - }) - if err != nil { - log.Errorf("Failed to create jobID: %s, jobName: %v, cron: %v; release plan ID: %v, index: %v, name: %v, error: %v", job.ID.Hex(), name, schedule, job.ReleasePlanArgs.ID, job.ReleasePlanArgs.Index, job.ReleasePlanArgs.Name, err) - return err + // delete old schedule job first + tag := job.ID.Hex() + h.NewScheduler.RemoveByTags(tag) + + scheduleTime := time.Unix(job.UnixStamp, 0) + if scheduleTime.Before(time.Now()) { + log.Errorf("release plan schedule time is before now, jobID: %v, schedule time: %v; release plan ID: %v, index: %v, name: %v", job.ID, scheduleTime, job.ReleasePlanArgs.ID, job.ReleasePlanArgs.Index, job.ReleasePlanArgs.Name) + return nil } - log.Infof("registering jobID: %s with name: %v, cron: %v; release plan ID: %v, index: %v, name: %v", job.ID.Hex(), name, schedule, job.ReleasePlanArgs.ID, job.ReleasePlanArgs.Index, job.ReleasePlanArgs.Name) - err = h.Scheduler.UpdateJobModel(job.ID.Hex(), scheduleJob) + jobName := fmt.Sprintf("release_plan:%s:%d:%s", job.ReleasePlanArgs.Name, job.ReleasePlanArgs.Index, scheduleTime) + _, err := h.NewScheduler.NewJob(newgoCron.OneTimeJob(newgoCron.OneTimeJobStartDateTime(scheduleTime)), newgoCron.NewTask(executeReleasePlanFunc), newgoCron.WithTags(tag), newgoCron.WithName(jobName)) if err != nil { - log.Errorf("Failed to register job of ID: %s to scheduler, the error is: %v", job.ID, err) + log.Errorf("Failed to create jobID: %s, jobName: %v, schedule time: %v; release plan ID: %v, index: %v, name: %v, error: %v", job.ID.Hex(), name, scheduleTime, job.ReleasePlanArgs.ID, job.ReleasePlanArgs.Index, job.ReleasePlanArgs.Name, err) return err } + + log.Infof("registering jobID: %s with name: %v, schedule time: %v; release plan ID: %v, index: %v, name: %v", job.ID.Hex(), name, scheduleTime, job.ReleasePlanArgs.ID, job.ReleasePlanArgs.Index, job.ReleasePlanArgs.Name) return nil } diff --git a/pkg/microservice/cron/core/service/scheduler/schedule_env.go b/pkg/microservice/cron/core/service/scheduler/schedule_env.go index 9075f3be9c..506cb558ee 100644 --- a/pkg/microservice/cron/core/service/scheduler/schedule_env.go +++ b/pkg/microservice/cron/core/service/scheduler/schedule_env.go @@ -91,7 +91,7 @@ func (c *CronClient) UpsertEnvServiceScheduler(log *zap.SugaredLogger) { } } c.lastSchedulersRWMutex.Unlock() - log.Infof("[vm] [%s] deleted service scheduler..", key) + // log.Infof("[vm] [%s] deleted service scheduler..", key) continue } @@ -136,7 +136,7 @@ func (c *CronClient) UpsertEnvServiceScheduler(log *zap.SugaredLogger) { c.SchedulerController[key] = c.Schedulers[key].Start() c.SchedulerControllerRWMutex.Unlock() - log.Infof("[vm] [%s] added service scheduler..", key) + // log.Infof("[vm] [%s] added service scheduler..", key) } } break diff --git a/pkg/microservice/cron/core/service/scheduler/schedule_env_update.go b/pkg/microservice/cron/core/service/scheduler/schedule_env_update.go index 3ff80f79bb..5ee5f75a4c 100644 --- a/pkg/microservice/cron/core/service/scheduler/schedule_env_update.go +++ b/pkg/microservice/cron/core/service/scheduler/schedule_env_update.go @@ -56,7 +56,7 @@ func (c *CronClient) UpsertEnvValueSyncScheduler(log *zap.SugaredLogger) { log.Infof("start init env values sync scheduler... env count: %v", len(envs)) for _, env := range envs { - log.Infof("schedule_env_update handle single helm env: %s/%s", env.ProductName, env.EnvName) + // log.Debugf("schedule_env_update handle single helm env: %s/%s", env.ProductName, env.EnvName) envObj, err := c.AslanCli.GetEnvService(env.ProductName, env.EnvName, log) if err != nil { log.Errorf("failed to get env data, productName:%s envName:%s err:%v", env.ProductName, env.EnvName, err) diff --git a/pkg/microservice/cron/core/service/scheduler/schedule_workflow.go b/pkg/microservice/cron/core/service/scheduler/schedule_workflow.go index 2dc906852a..d83c76b43d 100644 --- a/pkg/microservice/cron/core/service/scheduler/schedule_workflow.go +++ b/pkg/microservice/cron/core/service/scheduler/schedule_workflow.go @@ -93,66 +93,6 @@ func (c *CronClient) UpsertWorkflowScheduler(log *zap.SugaredLogger) { c.SchedulerControllerRWMutex.Unlock() } - pipelines, err := c.AslanCli.ListPipelines(log) - if err != nil { - log.Error(err) - return - } - - log.Info("start init pipeline scheduler..") - for _, pipeline := range pipelines { - key := "pipeline-" + pipeline.Name - taskMap[key] = true - - c.enabledMapRWMutex.Lock() - c.lastSchedulersRWMutex.Lock() - if _, ok := c.lastSchedulers[key]; ok && reflect.DeepEqual(pipeline.Schedules.Items, c.lastSchedulers[key]) { - // 增加判断：enabled的值未被更新时才能跳过 - if enabled, ok := c.enabledMap[key]; ok && enabled == pipeline.Schedules.Enabled { - c.lastSchedulersRWMutex.Unlock() - c.enabledMapRWMutex.Unlock() - continue - } - } - c.enabledMap[key] = pipeline.Schedules.Enabled - c.lastSchedulers[key] = pipeline.Schedules.Items - c.lastSchedulersRWMutex.Unlock() - c.enabledMapRWMutex.Unlock() - - newScheduler := gocron.NewScheduler() - for _, schedule := range pipeline.Schedules.Items { - if schedule != nil { - if err := schedule.Validate(); err != nil { - log.Errorf("[%s] invalid schedule: %v", key, err) - continue - } - BuildScheduledPipelineJob(newScheduler, schedule).Do(c.RunScheduledPipelineTask, pipeline, schedule.TaskArgs, log) - } - } - // 所有scheduler总开关 - if !pipeline.Schedules.Enabled { - newScheduler.Clear() - } - - c.SchedulersRWMutex.Lock() - c.Schedulers[key] = newScheduler - c.SchedulersRWMutex.Unlock() - - log.Infof("[%s] building schedulers..", key) - // 停掉旧的scheduler - c.SchedulerControllerRWMutex.Lock() - sc, ok := c.SchedulerController[key] - c.SchedulerControllerRWMutex.Unlock() - if ok { - sc <- true - } - - log.Infof("[%s]lens of scheduler: %d", key, c.Schedulers[key].Len()) - c.SchedulerControllerRWMutex.Lock() - c.SchedulerController[key] = c.Schedulers[key].Start() - c.SchedulerControllerRWMutex.Unlock() - } - ScheduleNames := sets.NewString( CleanJobScheduler, UpsertWorkflowScheduler, UpsertTestScheduler, InitStatScheduler, InitOperationStatScheduler, diff --git a/pkg/microservice/cron/core/service/scheduler/scheduler.go b/pkg/microservice/cron/core/service/scheduler/scheduler.go index 1ceff56e50..c063d912e3 100644 --- a/pkg/microservice/cron/core/service/scheduler/scheduler.go +++ b/pkg/microservice/cron/core/service/scheduler/scheduler.go @@ -23,12 +23,11 @@ import ( "sync" "time" + newgoCron "github.com/go-co-op/gocron/v2" "github.com/jasonlvhit/gocron" "github.com/rfyiamcool/cronlib" "go.uber.org/zap" - newgoCron "github.com/go-co-op/gocron" - configbase "github.com/koderover/zadig/v2/pkg/config" "github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/repository/mongodb" "github.com/koderover/zadig/v2/pkg/microservice/cron/core/service" @@ -66,13 +65,17 @@ type CronClient struct { } type CronV3Client struct { - Scheduler *newgoCron.Scheduler + Scheduler newgoCron.Scheduler AslanCli *aslan.Client } func NewCronV3() *CronV3Client { + scheduler, err := newgoCron.NewScheduler() + if err != nil { + log.Fatalf("failed to create scheduler: %v", err) + } return &CronV3Client{ - Scheduler: newgoCron.NewScheduler(time.Local), + Scheduler: scheduler, AslanCli: aslan.New(configbase.AslanServiceAddress()), } } @@ -80,7 +83,7 @@ func NewCronV3() *CronV3Client { func (c *CronV3Client) Start() { var lastConfig *aslan.CleanConfig - c.Scheduler.Every(5).Seconds().Do(func() { + c.Scheduler.NewJob(newgoCron.DurationJob(5*time.Second), newgoCron.NewTask(func() { // get the docker clean config config, err := c.AslanCli.GetDockerCleanConfig() if err != nil { @@ -92,26 +95,26 @@ func (c *CronV3Client) Start() { lastConfig = config log.Infof("config changed to %v", config) if config.CronEnabled { - c.Scheduler.RemoveByTag(string(types.CleanDockerTag)) - _, err = c.Scheduler.Cron(config.Cron).Tag(string(types.CleanDockerTag)).Do(func() { + c.Scheduler.RemoveByTags(string(types.CleanDockerTag)) + + _, err := c.Scheduler.NewJob(newgoCron.CronJob(config.Cron, false), newgoCron.NewTask(func() { log.Infof("trigger aslan docker clean,reg: %v", config.Cron) // call docker clean if err := c.AslanCli.DockerClean(); err != nil { log.Errorf("fail to clean docker cache , err:%s", err) } - }) + })) if err != nil { log.Errorf("fail to add docker_cache clean cron job:reg: %v,err:%s", config.Cron, err) } } else { log.Infof("remove docker_cache clean job , job tag: %v", types.CleanDockerTag) - c.Scheduler.RemoveByTag(string(types.CleanDockerTag)) - + c.Scheduler.RemoveByTags(string(types.CleanDockerTag)) } } - }) + })) - c.Scheduler.StartAsync() + c.Scheduler.Start() } const ( @@ -146,14 +149,19 @@ const ( ) // NewCronClient ... -// 服务初始化 +// 注意初始化失败会panic func NewCronClient() *CronClient { aslanCli := client.NewAslanClient(fmt.Sprintf("%s/api", configbase.AslanServiceAddress())) cronjobScheduler := cronlib.New() cronjobScheduler.Start() + newgoCronSchedule, err := newgoCron.NewScheduler() + if err != nil { + log.Fatalf("failed to create scheduler: %v", err) + } + newgoCronSchedule.Start() - cronjobHandler := NewCronjobHandler(aslanCli, cronjobScheduler) + cronjobHandler := NewCronjobHandler(aslanCli, cronjobScheduler, newgoCronSchedule) go func() { for { diff --git a/pkg/microservice/cron/core/service/types.go b/pkg/microservice/cron/core/service/types.go index 3795d62dfe..591ba7e37e 100644 --- a/pkg/microservice/cron/core/service/types.go +++ b/pkg/microservice/cron/core/service/types.go @@ -37,6 +37,8 @@ const ( TimingSchedule ScheduleType = "timing" // GapSchedule 间隔循环 GapSchedule ScheduleType = "gap" + // UnixstampSchedule 时间戳定时 + UnixstampSchedule ScheduleType = "unix_stamp" ) type PipelineResource struct { @@ -80,6 +82,7 @@ type PipelineSpec struct { type Schedule struct { ID primitive.ObjectID `bson:"_id,omitempty" json:"id,omitempty"` Number uint64 `bson:"number" json:"number"` + UnixStamp int64 `bson:"unix_stamp" json:"unix_stamp"` Frequency string `bson:"frequency" json:"frequency"` Time string `bson:"time" json:"time"` MaxFailures int `bson:"max_failures,omitempty" json:"max_failures,omitempty"` diff --git a/pkg/setting/consts.go b/pkg/setting/consts.go index 8c41fe8401..bc1b3ebfb8 100644 --- a/pkg/setting/consts.go +++ b/pkg/setting/consts.go @@ -560,6 +560,7 @@ const ( FixedDayTimeCronjob = "timing" FixedGapCronjob = "gap" CrontabCronjob = "crontab" + UnixStampSchedule = "unix_stamp" // 定时器的所属job类型 WorkflowCronjob = "workflow" diff --git a/pkg/tool/gitee/repositories.go b/pkg/tool/gitee/repositories.go index 07655fbe99..1f89cb185d 100644 --- a/pkg/tool/gitee/repositories.go +++ b/pkg/tool/gitee/repositories.go @@ -32,6 +32,7 @@ import ( type Project struct { ID int `json:"id"` Name string `json:"name"` + Path string `json:"path"` DefaultBranch string `json:"default_branch,omitempty"` Namespace *NamespaceInfo `json:"namespace,omitempty"` }