+
本邮件由 Zadig 系统自动发出,请勿直接回复。
+
Made By Zadig Team ♥ Happy Coding.
+
\ No newline at end of file
diff --git a/pkg/microservice/user/core/service/user/user.go b/pkg/microservice/user/core/service/user/user.go
new file mode 100644
index 0000000000000000000000000000000000000000..0a6be6d67fdc325819ca769c2903c40c9ec454e5
--- /dev/null
+++ b/pkg/microservice/user/core/service/user/user.go
@@ -0,0 +1,604 @@
+/*
+Copyright 2021 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 user
+
+import (
+ _ "embed"
+ "errors"
+ "fmt"
+ "net/url"
+ "time"
+
+ "github.com/dexidp/dex/connector/ldap"
+ ldapv3 "github.com/go-ldap/ldap/v3"
+ "github.com/go-sql-driver/mysql"
+ "github.com/golang-jwt/jwt"
+ "github.com/google/uuid"
+ "go.uber.org/zap"
+ "golang.org/x/crypto/bcrypt"
+
+ configbase "github.com/koderover/zadig/pkg/config"
+ "github.com/koderover/zadig/pkg/microservice/user/config"
+ "github.com/koderover/zadig/pkg/microservice/user/core"
+ "github.com/koderover/zadig/pkg/microservice/user/core/repository/models"
+ "github.com/koderover/zadig/pkg/microservice/user/core/repository/orm"
+ "github.com/koderover/zadig/pkg/microservice/user/core/service/login"
+ "github.com/koderover/zadig/pkg/setting"
+ "github.com/koderover/zadig/pkg/shared/client/systemconfig"
+ e "github.com/koderover/zadig/pkg/tool/errors"
+ "github.com/koderover/zadig/pkg/tool/mail"
+ "github.com/koderover/zadig/pkg/types"
+)
+
+type User struct {
+ Name string `json:"name"`
+ Password string `json:"password"`
+ Email string `json:"email"`
+ Account string `json:"account"`
+ Phone string `json:"phone,omitempty"`
+}
+
+type UpdateUserInfo struct {
+ Name string `json:"name,omitempty"`
+ Email string `json:"email,omitempty"`
+ Phone string `json:"phone,omitempty"`
+}
+
+type QueryArgs struct {
+ Name string `json:"name,omitempty"`
+ Account string `json:"account,omitempty"`
+ IdentityType string `json:"identity_type,omitempty"`
+ UIDs []string `json:"uids,omitempty"`
+ PerPage int `json:"per_page,omitempty"`
+ Page int `json:"page,omitempty"`
+}
+
+type Password struct {
+ Uid string `json:"uid"`
+ OldPassword string `json:"oldPassword"`
+ NewPassword string `json:"newPassword"`
+}
+
+type ResetParams struct {
+ Uid string `json:"uid"`
+ Password string `json:"password"`
+}
+
+type SyncUserInfo struct {
+ Account string `json:"account"`
+ IdentityType string `json:"identityType"`
+ Name string `json:"name"`
+ Email string `json:"email"`
+}
+
+type RetrieveResp struct {
+ Email string `json:"email"`
+}
+
+func SearchAndSyncUser(ldapId string, logger *zap.SugaredLogger) error {
+ systemConfigClient := systemconfig.New()
+ si, err := systemConfigClient.GetLDAPConnector(ldapId)
+ if err != nil {
+ logger.Errorf("SearchAndSyncUser GetLDAPConnector error, error msg:%s", err)
+ return fmt.Errorf("SearchAndSyncUser GetLDAPConnector error, error msg:%s", err)
+ }
+ if si == nil || si.Config == nil {
+ logger.Error("can't find connector")
+ return fmt.Errorf("can't find connector")
+ }
+
+ config := si.Config.(*ldap.Config)
+ l, err := ldapv3.Dial("tcp", config.Host)
+ if err != nil {
+ logger.Errorf("ldap dial host:%s error, error msg:%s", config.Host, err)
+ return err
+ }
+ defer l.Close()
+
+ err = l.Bind(config.BindDN, config.BindPW)
+ if err != nil {
+ logger.Errorf("ldap bind host:%s error, error msg:%s", config.Host, err)
+ return err
+ }
+
+ searchRequest := ldapv3.NewSearchRequest(
+ config.GroupSearch.BaseDN,
+ ldapv3.ScopeWholeSubtree, ldapv3.NeverDerefAliases, 0, 0, false,
+ config.GroupSearch.Filter, // The filter to apply
+ []string{config.GroupSearch.NameAttr, config.UserSearch.NameAttr, config.UserSearch.PreferredUsernameAttrAttr,
+ config.UserSearch.EmailAttr}, // A list attributes to retrieve
+ nil,
+ )
+
+ sr, err := l.Search(searchRequest)
+ if err != nil {
+ logger.Errorf("ldap search host:%s error, error msg:%s", config.Host, err)
+ return err
+ }
+ for _, entry := range sr.Entries {
+ account := config.UserSearch.PreferredUsernameAttrAttr
+ name := account
+ if len(config.UserSearch.NameAttr) != 0 {
+ name = config.UserSearch.NameAttr
+ }
+ _, err := SyncUser(&SyncUserInfo{
+ Account: entry.GetAttributeValue(account),
+ Name: entry.GetAttributeValue(name),
+ Email: entry.GetAttributeValue(config.UserSearch.EmailAttr),
+ IdentityType: si.ID, // ldap may have not only one instance, so use id as identityType
+ }, logger)
+ if err != nil {
+ logger.Errorf("ldap host:%s sync user error, error msg:%s", config.Host, err)
+ return err
+ }
+ }
+ return nil
+}
+
+func GetUser(uid string, logger *zap.SugaredLogger) (*types.UserInfo, error) {
+ user, err := orm.GetUserByUid(uid, core.DB)
+ if err != nil {
+ logger.Errorf("GetUser getUserByUid:%s error, error msg:%s", uid, err.Error())
+ return nil, err
+ }
+ if user == nil {
+ return nil, nil
+ }
+ userLogin, err := orm.GetUserLogin(uid, user.Account, config.AccountLoginType, core.DB)
+ if err != nil {
+ logger.Errorf("GetUser GetUserLogin:%s error, error msg:%s", uid, err.Error())
+ return nil, err
+ }
+ userInfo := mergeUserLogin([]models.User{*user}, []models.UserLogin{*userLogin}, logger)
+ userInfoRes := &userInfo[0]
+ userInfoRes.APIToken = user.APIToken
+ //TODO Create a permanent OpenAPI token
+ if user.APIToken == "" {
+ token, err := login.CreateToken(&login.Claims{
+ Name: user.Name,
+ UID: user.UID,
+ Email: user.Email,
+ PreferredUsername: user.Account,
+ StandardClaims: jwt.StandardClaims{
+ Audience: setting.ProductName,
+ //24*365*100=876000
+ ExpiresAt: time.Now().Add(876000 * time.Hour).Unix(),
+ },
+ FederatedClaims: login.FederatedClaims{
+ ConnectorId: user.IdentityType,
+ UserId: user.Account,
+ },
+ })
+ if err != nil {
+ logger.Errorf("LocalLogin user:%s create token error, error msg:%s", user.Account, err.Error())
+ return nil, err
+ }
+ userInfoRes.APIToken = token
+ userWithToken := &models.User{
+ APIToken: token,
+ }
+ err = orm.UpdateUser(uid, userWithToken, core.DB)
+ if err != nil {
+ logger.Errorf("UpdateUser user:%s save token error:%s", user.Account, err.Error())
+ return nil, err
+ }
+ }
+ return userInfoRes, nil
+}
+
+func SearchUserByAccount(args *QueryArgs, logger *zap.SugaredLogger) (*types.UsersResp, error) {
+ user, err := orm.GetUser(args.Account, args.IdentityType, core.DB)
+ if err != nil {
+ logger.Errorf("SearchUserByAccount GetUser By account:%s error, error msg:%s", args.Account, err.Error())
+ return nil, err
+ }
+ if user == nil {
+ return &types.UsersResp{
+ Users: nil,
+ TotalCount: 0,
+ }, nil
+ }
+ userLogins, err := orm.ListUserLogins([]string{user.UID}, core.DB)
+ if err != nil {
+ logger.Errorf("SearchUserByAccount ListUserLogins By uid:%s error, error msg:%s", user.UID, err.Error())
+ return nil, err
+ }
+ usersInfo := mergeUserLogin([]models.User{*user}, *userLogins, logger)
+ return &types.UsersResp{
+ Users: usersInfo,
+ TotalCount: int64(len(usersInfo)),
+ }, nil
+}
+
+func SearchUsers(args *QueryArgs, logger *zap.SugaredLogger) (*types.UsersResp, error) {
+ count, err := orm.GetUsersCount(args.Name)
+ if err != nil {
+ logger.Errorf("SeachUsers GetUsersCount By name:%s error, error msg:%s", args.Name, err.Error())
+ return nil, err
+ }
+ if count == 0 {
+ return &types.UsersResp{
+ TotalCount: 0,
+ }, nil
+ }
+
+ users, err := orm.ListUsers(args.Page, args.PerPage, args.Name, core.DB)
+ if err != nil {
+ logger.Errorf("SeachUsers SeachUsers By name:%s error, error msg:%s", args.Name, err.Error())
+ return nil, err
+ }
+ var uids []string
+ for _, user := range users {
+ uids = append(uids, user.UID)
+ }
+ userLogins, err := orm.ListUserLogins(uids, core.DB)
+ if err != nil {
+ logger.Errorf("SeachUsers ListUserLogins By uids:%s error, error msg:%s", uids, err.Error())
+ return nil, err
+ }
+ usersInfo := mergeUserLogin(users, *userLogins, logger)
+ return &types.UsersResp{
+ Users: usersInfo,
+ TotalCount: count,
+ }, nil
+}
+
+func mergeUserLogin(users []models.User, userLogins []models.UserLogin, logger *zap.SugaredLogger) []types.UserInfo {
+ userLoginMap := make(map[string]models.UserLogin)
+ for _, userLogin := range userLogins {
+ userLoginMap[userLogin.UID] = userLogin
+ }
+ var usersInfo []types.UserInfo
+ for _, user := range users {
+ if userLogin, ok := userLoginMap[user.UID]; ok {
+ usersInfo = append(usersInfo, types.UserInfo{
+ LastLoginTime: userLogin.LastLoginTime,
+ Uid: user.UID,
+ Phone: user.Phone,
+ Name: user.Name,
+ Email: user.Email,
+ IdentityType: user.IdentityType,
+ Account: user.Account,
+ })
+ } else {
+ logger.Error("user:%s login info not exist")
+ }
+ }
+ return usersInfo
+}
+
+func SearchUsersByUIDs(uids []string, logger *zap.SugaredLogger) (*types.UsersResp, error) {
+ users, err := orm.ListUsersByUIDs(uids, core.DB)
+ if err != nil {
+ logger.Errorf("SearchUsersByUIDs SeachUsers By uids:%s error, error msg:%s", uids, err.Error())
+ return nil, err
+ }
+ userLogins, err := orm.ListUserLogins(uids, core.DB)
+ if err != nil {
+ logger.Errorf("SearchUsersByUIDs ListUserLogins By uids:%s error, error msg:%s", uids, err.Error())
+ return nil, err
+ }
+ usersInfo := mergeUserLogin(users, *userLogins, logger)
+ return &types.UsersResp{
+ Users: usersInfo,
+ TotalCount: int64(len(usersInfo)),
+ }, nil
+}
+
+func getLoginId(user *models.User, loginType config.LoginType) string {
+ switch loginType {
+ case config.AccountLoginType:
+ return user.Account
+ default:
+ return user.Account
+ }
+
+}
+
+func DeleteUserByUID(uid string, logger *zap.SugaredLogger) error {
+ tx := core.DB.Begin()
+ defer func() {
+ if r := recover(); r != nil {
+ tx.Rollback()
+ }
+ }()
+ err := orm.DeleteUserByUid(uid, tx)
+ if err != nil {
+ tx.Rollback()
+ logger.Errorf("DeleteUserByUID DeleteUserByUid :%s error, error msg:%s", uid, err.Error())
+ return err
+ }
+ err = orm.DeleteUserLoginByUid(uid, tx)
+ if err != nil {
+ tx.Rollback()
+ logger.Errorf("DeleteUserByUID DeleteUserLoginByUid:%s error, error msg:%s", uid, err.Error())
+ return err
+ }
+ return tx.Commit().Error
+}
+
+//go:embed retrieve.html
+var retrieveHemlTemplate []byte
+
+func Retrieve(account string, logger *zap.SugaredLogger) (*RetrieveResp, error) {
+ user, err := orm.GetUser(account, config.SystemIdentityType, core.DB)
+ if err != nil {
+ logger.Errorf("Retrieve GetUser:%s error, error msg:%s ", account, err)
+ return nil, fmt.Errorf("Retrieve GetUser:%s error, error msg:%s ", account, err)
+ }
+ if user == nil {
+ return nil, fmt.Errorf("user not exist")
+ }
+ if len(user.Email) == 0 {
+ logger.Errorf("the account:%s has not email", account)
+ return nil, fmt.Errorf("the account has not email")
+ }
+
+ token, err := login.CreateToken(&login.Claims{
+ Name: user.Name,
+ UID: user.UID,
+ Email: user.Email,
+ StandardClaims: jwt.StandardClaims{
+ Audience: setting.ProductName,
+ ExpiresAt: time.Now().Add(5 * time.Minute).Unix(),
+ },
+ FederatedClaims: login.FederatedClaims{
+ UserId: user.Account,
+ ConnectorId: user.IdentityType,
+ },
+ })
+ if err != nil {
+ logger.Errorf("Retrieve user:%s create token error, error msg:%s", user.Account, err)
+ return nil, err
+ }
+ v := url.Values{}
+ v.Add("idtoken", token)
+ retrieveURL := configbase.SystemAddress() + "/signin?" + v.Encode()
+ body, err := mail.RenderEmailTemplate(retrieveURL, string(retrieveHemlTemplate))
+ if err != nil {
+ logger.Errorf("Retrieve renderEmailTemplate error, error msg:%s ", err)
+ return nil, fmt.Errorf("Retrieve renderEmailTemplate error, error msg:%s ", err)
+ }
+ systemConfigClient := systemconfig.New()
+ email, err := systemConfigClient.GetEmailHost()
+ if err != nil {
+ logger.Errorf("Retrieve GetEmailHost error, error msg:%s", err)
+ return nil, fmt.Errorf("Retrieve GetEmailHost error, error msg:%s ", err)
+ }
+ err = mail.SendEmail(&mail.EmailParams{
+ From: email.UserName,
+ To: user.Email,
+ Subject: "重置密码",
+ Host: email.Name,
+ UserName: email.UserName,
+ Password: email.Password,
+ Port: email.Port,
+ Body: body,
+ })
+ if err != nil {
+ logger.Errorf("Retrieve SendEmail error, error msg:%s ", err)
+ return nil, err
+ }
+ return &RetrieveResp{
+ Email: user.Email,
+ }, nil
+}
+
+func CreateUser(args *User, logger *zap.SugaredLogger) (*models.User, error) {
+ uid, _ := uuid.NewUUID()
+ user := &models.User{
+ Name: args.Name,
+ Email: args.Email,
+ IdentityType: config.SystemIdentityType,
+ Phone: args.Phone,
+ Account: args.Account,
+ UID: uid.String(),
+ }
+ tx := core.DB.Begin()
+ defer func() {
+ if r := recover(); r != nil {
+ tx.Rollback()
+ }
+ }()
+ err := orm.CreateUser(user, tx)
+ if err != nil {
+ tx.Rollback()
+ logger.Errorf("CreateUser CreateUser :%v error, error msg:%s", user, err.Error())
+ var mysqlErr *mysql.MySQLError
+ if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 {
+ return nil, e.ErrCreateUser.AddErr(err).AddDesc("存在相同用户名")
+ }
+ return nil, e.ErrCreateUser.AddErr(err)
+ }
+ hashedPassword, _ := bcrypt.GenerateFromPassword([]byte(args.Password), bcrypt.DefaultCost)
+ userLogin := &models.UserLogin{
+ UID: user.UID,
+ Password: string(hashedPassword),
+ LastLoginTime: 0,
+ LoginId: getLoginId(user, config.AccountLoginType),
+ LoginType: int(config.AccountLoginType),
+ }
+ err = orm.CreateUserLogin(userLogin, tx)
+ if err != nil {
+ tx.Rollback()
+ logger.Errorf("CreateUser CreateUserLogin:%v error, error msg:%s", user, err.Error())
+ return nil, err
+ }
+ return user, tx.Commit().Error
+}
+
+func UpdateUser(uid string, args *UpdateUserInfo, _ *zap.SugaredLogger) error {
+ user := &models.User{
+ Name: args.Name,
+ Email: args.Email,
+ Phone: args.Phone,
+ }
+ return orm.UpdateUser(uid, user, core.DB)
+
+}
+
+func UpdatePassword(args *Password, logger *zap.SugaredLogger) error {
+ user, err := orm.GetUserByUid(args.Uid, core.DB)
+ if err != nil {
+ logger.Errorf("UpdatePassword GetUserByUid:%s error, error msg:%s", args.Uid, err.Error())
+ return err
+ }
+ if user == nil {
+ return fmt.Errorf("user not exist")
+ }
+ userLogin, err := orm.GetUserLogin(user.UID, user.Account, config.AccountLoginType, core.DB)
+ if err != nil {
+ logger.Errorf("UpdatePassword GetUserLogin:%s error, error msg:%s", args.Uid, err.Error())
+ return err
+ }
+ if userLogin == nil {
+ logger.Errorf("UpdatePassword GetUserLogin:%s not exist", args.Uid)
+ return fmt.Errorf("userLogin not exist")
+ }
+ password := []byte(args.OldPassword)
+ err = bcrypt.CompareHashAndPassword([]byte(userLogin.Password), password)
+ if err == bcrypt.ErrMismatchedHashAndPassword {
+ return fmt.Errorf("password is wrong")
+ }
+ if err != nil {
+ logger.Errorf("UpdatePassword CompareHashAndPassword userLogin password:%s, password:%s error,"+
+ " error msg:%s", userLogin.Password, password, err.Error())
+ return err
+ }
+ hashedPassword, _ := bcrypt.GenerateFromPassword([]byte(args.NewPassword), bcrypt.DefaultCost)
+ userLogin = &models.UserLogin{
+ UID: user.UID,
+ Password: string(hashedPassword),
+ }
+ err = orm.UpdateUserLogin(user.UID, userLogin, core.DB)
+ if err != nil {
+ logger.Errorf("UpdatePassword UpdateUserLogin:%v error, error msg:%s", userLogin, err.Error())
+ return err
+ }
+ return nil
+}
+
+func Reset(args *ResetParams, logger *zap.SugaredLogger) error {
+ user, err := orm.GetUserByUid(args.Uid, core.DB)
+ if err != nil {
+ logger.Errorf("Reset GetUserByUid:%s error, error msg:%s", args.Uid, err)
+ return err
+ }
+ if user == nil {
+ logger.Error("user not exist")
+ return fmt.Errorf("user not exist")
+ }
+
+ hashedPassword, _ := bcrypt.GenerateFromPassword([]byte(args.Password), bcrypt.DefaultCost)
+ userLogin := &models.UserLogin{
+ UID: user.UID,
+ Password: string(hashedPassword),
+ }
+ err = orm.UpdateUserLogin(user.UID, userLogin, core.DB)
+ if err != nil {
+ logger.Errorf("UpdatePassword UpdateUserLogin:%v error, error msg:%s", userLogin, err.Error())
+ return err
+ }
+ return nil
+}
+
+func SyncUser(syncUserInfo *SyncUserInfo, logger *zap.SugaredLogger) (*models.User, error) {
+ user, err := orm.GetUser(syncUserInfo.Account, syncUserInfo.IdentityType, core.DB)
+ if err != nil {
+ logger.Error("SyncUser get user:%s error, error msg:%s", syncUserInfo.Account, err.Error())
+ return nil, err
+ }
+ tx := core.DB.Begin()
+ defer func() {
+ if r := recover(); r != nil {
+ tx.Rollback()
+ }
+ }()
+ if user == nil {
+ uid, _ := uuid.NewUUID()
+ user = &models.User{
+ UID: uid.String(),
+ Name: syncUserInfo.Name,
+ Account: syncUserInfo.Account,
+ Email: syncUserInfo.Email,
+ IdentityType: syncUserInfo.IdentityType,
+ }
+ err = orm.CreateUser(user, tx)
+ if err != nil {
+ tx.Rollback()
+ logger.Error("SyncUser create user:%s error, error msg:%s", syncUserInfo.Account, err.Error())
+ return nil, err
+ }
+ } else {
+ err = orm.UpdateUser(user.UID, &models.User{
+ Name: syncUserInfo.Name,
+ Account: syncUserInfo.Account,
+ Email: syncUserInfo.Email,
+ }, tx)
+ if err != nil {
+ tx.Rollback()
+ logger.Error("SyncUser update user:%s error, error msg:%s", syncUserInfo.Account, err.Error())
+ return nil, err
+ }
+ }
+ userLogin, err := orm.GetUserLogin(user.UID, user.Account, config.AccountLoginType, tx)
+ if err != nil {
+ tx.Rollback()
+ logger.Error("UpdateLoginInfo get user:%s login error, error msg:%s", user.UID, err.Error())
+ return nil, err
+ }
+ err = login.CheckSignature(userLogin.LastLoginTime > 0, logger)
+ if err != nil {
+ return nil, err
+ }
+ if userLogin != nil {
+ userLogin.LastLoginTime = time.Now().Unix()
+ err = orm.UpdateUserLogin(user.UID, userLogin, tx)
+ if err != nil {
+ tx.Rollback()
+ logger.Error("UpdateLoginInfo update user:%s login error, error msg:%s", user.UID, err.Error())
+ return nil, err
+ }
+ } else {
+ err = orm.CreateUserLogin(&models.UserLogin{
+ UID: user.UID,
+ LastLoginTime: time.Now().Unix(),
+ LoginId: getLoginId(user, config.AccountLoginType),
+ LoginType: int(config.AccountLoginType),
+ }, tx)
+ if err != nil {
+ tx.Rollback()
+ logger.Error("UpdateLoginInfo create user:%s login error, error msg:%s", user.UID, err.Error())
+ return nil, err
+ }
+ }
+ err = tx.Commit().Error
+ if err != nil {
+ logger.Errorf("SyncUser tx commit error, error msg:%s ", err)
+ return nil, err
+ }
+ return user, nil
+}
+
+func GetUserCount(logger *zap.SugaredLogger) ([]*types.UserCountByType, error) {
+ resp, err := orm.CountUserByType(core.DB)
+ if err != nil {
+ logger.Errorf("Failed to count user by type from db, the error is: %s", err.Error())
+ return nil, err
+ }
+ return resp, nil
+}
diff --git a/pkg/microservice/user/server/rest/router.go b/pkg/microservice/user/server/rest/router.go
new file mode 100644
index 0000000000000000000000000000000000000000..51139f3a6bddf883dc16c6882df6b72498b8e540
--- /dev/null
+++ b/pkg/microservice/user/server/rest/router.go
@@ -0,0 +1,35 @@
+/*
+Copyright 2021 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 rest
+
+import (
+ "github.com/gin-gonic/gin"
+
+ "github.com/koderover/zadig/pkg/microservice/user/core/handler"
+)
+
+func (s *engine) injectRouterGroup(router *gin.RouterGroup) {
+ for _, r := range []injector{
+ new(handler.Router),
+ } {
+ r.Inject(router.Group("/api/v1"))
+ }
+}
+
+type injector interface {
+ Inject(router *gin.RouterGroup)
+}
diff --git a/pkg/microservice/user/server/rest/server.go b/pkg/microservice/user/server/rest/server.go
new file mode 100644
index 0000000000000000000000000000000000000000..18843b0a5ead8f194d217a281527d5152e939795
--- /dev/null
+++ b/pkg/microservice/user/server/rest/server.go
@@ -0,0 +1,76 @@
+/*
+Copyright 2021 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 rest
+
+import (
+ "net/http"
+
+ "github.com/gin-gonic/gin"
+
+ "github.com/koderover/zadig/pkg/config"
+ ginmiddleware "github.com/koderover/zadig/pkg/middleware/gin"
+ "github.com/koderover/zadig/pkg/tool/log"
+)
+
+type engine struct {
+ *gin.Engine
+
+ mode string
+}
+
+func NewEngine() *engine {
+ s := &engine{mode: config.Mode()}
+
+ gin.SetMode(s.mode)
+
+ s.injectMiddlewares()
+ s.injectRouters()
+
+ return s
+}
+
+func (s *engine) injectMiddlewares() {
+ g := gin.New()
+ defer func() {
+ s.Engine = g
+ }()
+
+ if s.mode == gin.TestMode {
+ return
+ }
+ g.Use(ginmiddleware.Response())
+ g.Use(ginmiddleware.RequestID())
+ g.Use(ginmiddleware.RequestLog(log.NewFileLogger(config.RequestLogFile())))
+ g.Use(gin.Recovery())
+}
+
+func (s *engine) injectRouters() {
+ g := s.Engine
+
+ g.NoRoute(func(c *gin.Context) {
+ c.String(http.StatusNotFound, "Invalid path: %s", c.Request.URL.Path)
+ })
+ g.HandleMethodNotAllowed = true
+ g.NoMethod(func(c *gin.Context) {
+ c.String(http.StatusMethodNotAllowed, "Method not allowed: %s %s", c.Request.Method, c.Request.URL.Path)
+ })
+
+ apiRouters := g.Group("")
+ s.injectRouterGroup(apiRouters)
+
+ s.Engine = g
+}
diff --git a/pkg/microservice/user/server/server.go b/pkg/microservice/user/server/server.go
new file mode 100644
index 0000000000000000000000000000000000000000..aa0f55e5dc51bc614d6fe57f1bf620e9918011b4
--- /dev/null
+++ b/pkg/microservice/user/server/server.go
@@ -0,0 +1,60 @@
+/*
+Copyright 2021 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 server
+
+import (
+ "context"
+ "net/http"
+ "time"
+
+ "github.com/koderover/zadig/pkg/microservice/user/core"
+ "github.com/koderover/zadig/pkg/microservice/user/server/rest"
+ "github.com/koderover/zadig/pkg/tool/log"
+)
+
+func Serve(ctx context.Context) error {
+ core.Start(ctx)
+ defer core.Stop(ctx)
+
+ log.Info("Start user system service")
+
+ engine := rest.NewEngine()
+ server := &http.Server{Addr: ":80", Handler: engine}
+
+ stopChan := make(chan struct{})
+ go func() {
+ defer close(stopChan)
+
+ <-ctx.Done()
+
+ ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
+ defer cancel()
+
+ if err := server.Shutdown(ctx); err != nil {
+ log.Errorf("Failed to stop server, error: %s", err)
+ }
+ }()
+
+ if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+ log.Errorf("Failed to start http server, error: %s", err)
+ return err
+ }
+
+ <-stopChan
+
+ return nil
+}
diff --git a/pkg/microservice/warpdrive/config/config.go b/pkg/microservice/warpdrive/config/config.go
index 2795735f46e0997740541ee678daa26d4f6832f0..9d74ab6d5b442d2f7930db6b2e760b9cad0c2d6f 100644
--- a/pkg/microservice/warpdrive/config/config.go
+++ b/pkg/microservice/warpdrive/config/config.go
@@ -34,10 +34,6 @@ func NSQLookupAddrs() []string {
return strings.Split(viper.GetString(setting.ENVNsqLookupAddrs), ",")
}
-func PoetryAPIRootKey() string {
- return viper.GetString(setting.ENVPoetryAPIRootKey)
-}
-
func ReleaseImageTimeout() string {
return viper.GetString(setting.ReleaseImageTimeout)
}
@@ -46,14 +42,6 @@ func Home() string {
return viper.GetString(setting.Home)
}
-func DefaultRegistryAddr() string {
- return viper.GetString(setting.DefaultRegistryAddr)
-}
-
-func DefaultRegistryAK() string {
- return viper.GetString(setting.DefaultRegistryAK)
-}
-
-func DefaultRegistrySK() string {
- return viper.GetString(setting.DefaultRegistrySK)
+func HomeV2() string {
+ return viper.GetString(setting.HomeV2)
}
diff --git a/pkg/microservice/warpdrive/config/const.go b/pkg/microservice/warpdrive/config/const.go
index 79acdadf07495d869b93c41bbabe5d120997cb9f..5ee27d05921ea48a0200745f7bff571b9dd3e644 100644
--- a/pkg/microservice/warpdrive/config/const.go
+++ b/pkg/microservice/warpdrive/config/const.go
@@ -19,24 +19,31 @@ package config
type TaskType string
const (
- TaskPipeline TaskType = "pipeline"
- TaskBuild TaskType = "buildv2"
- TaskJenkinsBuild TaskType = "jenkins_build"
- TaskArtifact TaskType = "artifact"
- TaskDeploy TaskType = "deploy"
- TaskTestingV2 TaskType = "testingv2"
- TaskDistributeToS3 TaskType = "distribute2kodo"
- TaskReleaseImage TaskType = "release_image"
- TaskJira TaskType = "jira"
- TaskDockerBuild TaskType = "docker_build"
- TaskSecurity TaskType = "security"
- TaskResetImage TaskType = "reset_image"
- TaskDistribute TaskType = "distribute"
+ TaskPipeline TaskType = "pipeline"
+ TaskBuild TaskType = "buildv2"
+ TaskBuildV3 TaskType = "buildv3"
+ TaskArtifactDeploy TaskType = "artifact_deploy"
+ TaskJenkinsBuild TaskType = "jenkins_build"
+ TaskArtifact TaskType = "artifact"
+ TaskDeploy TaskType = "deploy"
+ TaskTestingV2 TaskType = "testingv2"
+ TaskDistributeToS3 TaskType = "distribute2kodo"
+ TaskReleaseImage TaskType = "release_image"
+ TaskJira TaskType = "jira"
+ TaskDockerBuild TaskType = "docker_build"
+ TaskSecurity TaskType = "security"
+ TaskResetImage TaskType = "reset_image"
+ TaskDistribute TaskType = "distribute"
+ TaskTrigger TaskType = "trigger"
+ TaskExtension TaskType = "extension"
+ TaskArtifactPackage TaskType = "artifact_package"
+ TaskScanning TaskType = "scanning"
)
type Status string
const (
+ StatusInit Status = "init"
StatusDisabled Status = "disabled"
StatusCreated Status = "created"
StatusRunning Status = "running"
@@ -49,21 +56,28 @@ const (
StatusQueued Status = "queued"
StatusBlocked Status = "blocked"
QueueItemPending Status = "pending"
+ StatusPrepare Status = "prepare"
)
type PipelineType string
const (
- // SingleType 单服务工作流
+ // SingleType single-service workflow
SingleType PipelineType = "single"
- // WorkflowType 多服务工作流
+ // WorkflowType multi-service workflow
WorkflowType PipelineType = "workflow"
- // FreestyleType 自由编排工作流
+ // FreestyleType freeStyle workflow
FreestyleType PipelineType = "freestyle"
- // TestType 测试
+ // TestType testing
TestType PipelineType = "test"
- // ServiceType 服务
+ // ServiceType pipeline
ServiceType PipelineType = "service"
+ // workflowTypeV3 服务
+ WorkflowTypeV3 PipelineType = "workflow_v3"
+ // ArtifactType artifact build
+ ArtifactType PipelineType = "artifact"
+ // ScanningType is the type for scanning
+ ScanningType PipelineType = "scanning"
)
type NotifyType int
diff --git a/pkg/microservice/warpdrive/core/service/common/dockerhosts.go b/pkg/microservice/warpdrive/core/service/common/dockerhosts.go
new file mode 100644
index 0000000000000000000000000000000000000000..33200e3954150ae5ca30773f15eead8769635b84
--- /dev/null
+++ b/pkg/microservice/warpdrive/core/service/common/dockerhosts.go
@@ -0,0 +1,201 @@
+/*
+Copyright 2022 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 common
+
+import (
+ "context"
+ "fmt"
+ "math/rand"
+ "sync"
+ "time"
+
+ "github.com/buraksezer/consistent"
+ "github.com/cespare/xxhash"
+ "go.uber.org/zap"
+ appsv1 "k8s.io/api/apps/v1"
+ "k8s.io/apimachinery/pkg/util/wait"
+ "sigs.k8s.io/controller-runtime/pkg/client"
+
+ "github.com/koderover/zadig/pkg/config"
+ "github.com/koderover/zadig/pkg/setting"
+ kubeclient "github.com/koderover/zadig/pkg/shared/kube/client"
+)
+
+var once sync.Once
+
+type Member string
+
+func (m Member) String() string {
+ return string(m)
+}
+
+type hasher struct{}
+
+func (h hasher) Sum64(data []byte) uint64 {
+ return xxhash.Sum64(data)
+}
+
+type ClusterID string
+
+type DockerHostsI interface {
+ GetBestHost(ClusterID, string) string
+
+ Sync()
+}
+
+type dockerhosts struct {
+ rwLock *sync.RWMutex
+ store map[ClusterID]*consistent.Consistent
+ hubServerAddr string
+ syncInterval time.Duration
+
+ logger *zap.SugaredLogger
+}
+
+var dockerHosts DockerHostsI
+
+func init() {
+ rand.Seed(time.Now().UTC().UnixNano())
+}
+
+func NewDockerHosts(hubServerAddr string, logger *zap.SugaredLogger) DockerHostsI {
+ once.Do(func() {
+ dockerHosts = &dockerhosts{
+ rwLock: &sync.RWMutex{},
+ store: map[ClusterID]*consistent.Consistent{},
+ hubServerAddr: hubServerAddr,
+ logger: logger,
+ }
+
+ go dockerHosts.Sync()
+ })
+
+ return dockerHosts
+}
+
+func (d *dockerhosts) GetBestHost(clusterID ClusterID, key string) string {
+ if d.store[clusterID] == nil {
+ d.initClusterInfo(clusterID)
+ }
+
+ member := d.store[clusterID].LocateKey([]byte(key))
+
+ return member.String()
+}
+
+func (d *dockerhosts) initClusterInfo(clusterID ClusterID) {
+ d.rwLock.Lock()
+ defer d.rwLock.Unlock()
+
+ members := d.getDockerHostsSvc(clusterID)
+
+ cfg := consistent.Config{
+ PartitionCount: 271,
+ ReplicationFactor: 20,
+ Load: 1.25,
+ Hasher: hasher{},
+ }
+
+ d.store[clusterID] = consistent.New(members, cfg)
+}
+
+func (d *dockerhosts) getDockerHostsSvc(clusterID ClusterID) []consistent.Member {
+ ns := config.Namespace()
+ if string(clusterID) != setting.LocalClusterID {
+ ns = setting.AttachedClusterNamespace
+ }
+
+ kclient, err := kubeclient.GetKubeClient(d.hubServerAddr, string(clusterID))
+ if err != nil {
+ d.logger.Warnf("Failed to get kubeclient for cluster %q: %s. Try to use default dockerhosts.", clusterID, err)
+ return d.getDefaultDockerHosts()
+ }
+
+ dindSts := &appsv1.StatefulSet{}
+ err = kclient.Get(context.TODO(), client.ObjectKey{
+ Name: "dind",
+ Namespace: ns,
+ }, dindSts)
+ if err != nil {
+ d.logger.Warnf("Failed to get dind statefuleset in namespace %q of cluster %q: %s", ns, clusterID, err)
+ return d.getDefaultDockerHosts()
+ }
+
+ members := []consistent.Member{}
+ for i := 0; i < int(*dindSts.Spec.Replicas); i++ {
+ members = append(members, d.genDindAddr(i))
+ }
+
+ return members
+}
+
+func (d *dockerhosts) getDefaultDockerHosts() []consistent.Member {
+ return []consistent.Member{d.genDindAddr(0)}
+}
+
+func (d *dockerhosts) genDindAddr(idx int) Member {
+ return Member(fmt.Sprintf("tcp://dind-%d.dind:2375", idx))
+}
+
+func (d *dockerhosts) Sync() {
+ d.logger.Info("Begin to sync")
+
+ wait.Forever(func() {
+ d.rwLock.Lock()
+ defer d.rwLock.Unlock()
+
+ for clusterID, consistentHash := range d.store {
+ currentMembers := d.getDockerHostsSvc(clusterID)
+ oldMembers := consistentHash.GetMembers()
+
+ d.logger.Infof("Cluster: %q. Current Members: %d. Old Members: %d", clusterID, len(currentMembers), len(oldMembers))
+
+ addedMembers, deletedMembers := d.diffMembers(oldMembers, currentMembers)
+ for _, member := range addedMembers {
+ consistentHash.Add(member)
+ }
+ for _, member := range deletedMembers {
+ consistentHash.Remove(member.String())
+ }
+ }
+ }, 3*time.Minute)
+}
+
+func (d *dockerhosts) diffMembers(old, current []consistent.Member) (added, deleted []consistent.Member) {
+ curMap := make(map[consistent.Member]struct{}, len(current))
+ for _, member := range current {
+ curMap[member] = struct{}{}
+ }
+
+ oldMap := make(map[consistent.Member]struct{}, len(old))
+ for _, member := range old {
+ oldMap[member] = struct{}{}
+ }
+
+ for _, member := range old {
+ if _, found := curMap[member]; !found {
+ deleted = append(deleted, member)
+ }
+ }
+ for _, member := range current {
+ if _, found := oldMap[member]; !found {
+ added = append(added, member)
+ }
+ }
+
+ return added, deleted
+}
diff --git a/pkg/microservice/warpdrive/core/service/common/dockerhosts_test.go b/pkg/microservice/warpdrive/core/service/common/dockerhosts_test.go
new file mode 100644
index 0000000000000000000000000000000000000000..21cf2685c71cf5ac7aad3b090260980fd40f568c
--- /dev/null
+++ b/pkg/microservice/warpdrive/core/service/common/dockerhosts_test.go
@@ -0,0 +1,50 @@
+/*
+Copyright 2022 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 common
+
+import (
+ "math/rand"
+ "testing"
+ "time"
+
+ "go.uber.org/zap"
+)
+
+func init() {
+ rand.Seed(time.Now().UTC().UnixNano())
+}
+
+func TestNewDockerHosts(t *testing.T) {
+ hubServerAddr := "hub-server.zadig"
+ logger, _ := zap.NewProduction()
+ dockerhosts := NewDockerHosts(hubServerAddr, logger.Sugar())
+
+ clusters := []ClusterID{ClusterID("local"), ClusterID("worker0"), ClusterID("worker1")}
+ svcs := []string{"svc0", "svc1", "svc2", "svc3", "svc4"}
+
+ for i := 0; i < len(clusters); i++ {
+ t.Logf("In Cluster %q\n", clusters[i])
+
+ for j := 0; j < 10; j++ {
+ svc := svcs[rand.Intn(len(svcs))]
+ host := dockerhosts.GetBestHost(clusters[i], svc)
+ t.Logf("\tHost for svc %q: %q\n", svc, host)
+ }
+
+ t.Log("\n")
+ }
+}
diff --git a/pkg/microservice/warpdrive/core/service/common/types.go b/pkg/microservice/warpdrive/core/service/common/types.go
new file mode 100644
index 0000000000000000000000000000000000000000..953c10c0e565afde34d7a473fe467fbb52238787
--- /dev/null
+++ b/pkg/microservice/warpdrive/core/service/common/types.go
@@ -0,0 +1,29 @@
+/*
+Copyright 2022 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 common
+
+import "github.com/koderover/zadig/pkg/microservice/warpdrive/config"
+
+type Stage struct {
+ //Note: The same stage cannot temporarily run different types of tasks
+ TaskType config.TaskType `bson:"type" json:"type"`
+ Status config.Status `bson:"status" json:"status"`
+ RunParallel bool `bson:"run_parallel" json:"run_parallel"`
+ Desc string `bson:"desc,omitempty" json:"desc,omitempty"`
+ SubTasks map[string]map[string]interface{} `bson:"sub_tasks" json:"sub_tasks"`
+ AfterAll bool `bson:"after_all" json:"after_all"`
+}
diff --git a/pkg/microservice/warpdrive/core/service/taskcontroller/interface.go b/pkg/microservice/warpdrive/core/service/taskcontroller/interface.go
new file mode 100644
index 0000000000000000000000000000000000000000..bfa6caaf3a1be9fed6604945f7bc603362392de1
--- /dev/null
+++ b/pkg/microservice/warpdrive/core/service/taskcontroller/interface.go
@@ -0,0 +1,27 @@
+/*
+Copyright 2022 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 taskcontroller
+
+import "context"
+
+type ControllerI interface {
+ // Init initializes the controller.
+ Init(ctx context.Context) error
+
+ // Stop stops process logics.
+ Stop(ctx context.Context) error
+}
diff --git a/pkg/microservice/warpdrive/core/service/taskcontroller/task_controller.go b/pkg/microservice/warpdrive/core/service/taskcontroller/task_controller.go
index 67ad5c1abcaa67a6c3675e578a376b8c3db686c0..6a04745df09aa05f9d492528ea292482de616123 100644
--- a/pkg/microservice/warpdrive/core/service/taskcontroller/task_controller.go
+++ b/pkg/microservice/warpdrive/core/service/taskcontroller/task_controller.go
@@ -31,47 +31,57 @@ import (
"github.com/koderover/zadig/pkg/tool/nsqcli"
)
-func InitTaskController(ctx context.Context) error {
+type controller struct {
+ consumers []*nsq.Consumer
+ producers []*nsq.Producer
+}
+
+func NewController() ControllerI {
+ return &controller{
+ consumers: []*nsq.Consumer{},
+ producers: []*nsq.Producer{},
+ }
+}
+
+func (c *controller) Init(ctx context.Context) error {
go func() {
if err := client.Start(ctx); err != nil {
panic(err)
}
}()
- //初始化nsq
cfg := nsq.NewConfig()
- // 注意 WD_POD_NAME 必须使用 Downward API 配置环境变量
cfg.UserAgent = config.WarpDrivePodName()
- // FIXME: Min: FIX MAGIC NUMBER
cfg.MaxAttempts = 50
cfg.LookupdPollInterval = 1 * time.Second
- //nsqd 和服务起在一起 FIXME: Min: FIX MAGIC ADDR
+ cfg.MsgTimeout = 1 * time.Minute
nsqClient := nsqcli.NewNsqClient(config.NSQLookupAddrs(), "127.0.0.1:4151")
- //Process topic
processor, err := nsq.NewConsumer(setting.TopicProcess, "process", cfg)
if err != nil {
return fmt.Errorf("init nsq processor error: %v", err)
}
+
processor.SetLogger(log.New(os.Stdout, "nsq consumer:", 0), nsq.LogLevelError)
+ c.consumers = append(c.consumers, processor)
- //Cancel topic
- //监听不同的channel,确保取消消息到每一个wd
canceller, err := nsq.NewConsumer(setting.TopicCancel, cfg.UserAgent, cfg)
if err != nil {
return fmt.Errorf("init nsq canceller error: %v", err)
}
+
canceller.SetLogger(log.New(os.Stdout, "nsq consumer:", 0), nsq.LogLevelError)
+ c.consumers = append(c.consumers, canceller)
- //Sender
nsqdAddr := "127.0.0.1:4150"
sender, err := nsq.NewProducer(nsqdAddr, cfg)
if err != nil {
return fmt.Errorf("init nsq sender error: %v", err)
}
+
sender.SetLogger(log.New(os.Stdout, "nsq producer:", 0), nsq.LogLevelError)
+ c.producers = append(c.producers, sender)
- // 初始化nsq topic
err = nsqClient.EnsureNsqdTopics([]string{setting.TopicAck, setting.TopicItReport, setting.TopicNotification})
if err != nil {
return fmt.Errorf("ensure nsq topic error: %v", err)
@@ -80,22 +90,31 @@ func InitTaskController(ctx context.Context) error {
execHandler := &ExecHandler{
Sender: sender,
}
-
processor.AddHandler(execHandler)
-
- //Add task plugin initiators to exec Handler
+ // Add task plugin initiators to exec Handler.
initTaskPlugins(execHandler)
cancelHandler := &CancelHandler{}
-
canceller.AddHandler(cancelHandler)
if err := processor.ConnectToNSQLookupds(config.NSQLookupAddrs()); err != nil {
return fmt.Errorf("processor could not connect to %v", config.NSQLookupAddrs())
}
-
if err := canceller.ConnectToNSQLookupds(config.NSQLookupAddrs()); err != nil {
return fmt.Errorf("canceller could not connect to %v", config.NSQLookupAddrs())
}
+
+ return nil
+}
+
+func (c *controller) Stop(ctx context.Context) error {
+ for _, consumer := range c.consumers {
+ consumer.Stop()
+ }
+
+ for _, producer := range c.producers {
+ producer.Stop()
+ }
+
return nil
}
diff --git a/pkg/microservice/warpdrive/core/service/taskcontroller/task_handler.go b/pkg/microservice/warpdrive/core/service/taskcontroller/task_handler.go
index 301be7bf65044b4a5752bd43d188703c1fc8101f..c0ecadb8e60b6c11e37a9efbc94297d9088cb4e7 100644
--- a/pkg/microservice/warpdrive/core/service/taskcontroller/task_handler.go
+++ b/pkg/microservice/warpdrive/core/service/taskcontroller/task_handler.go
@@ -26,8 +26,10 @@ import (
"github.com/nsqio/go-nsq"
uuid "github.com/satori/go.uuid"
"go.uber.org/zap"
+ "k8s.io/apimachinery/pkg/util/sets"
"github.com/koderover/zadig/pkg/microservice/warpdrive/config"
+ "github.com/koderover/zadig/pkg/microservice/warpdrive/core/service/common"
plugins "github.com/koderover/zadig/pkg/microservice/warpdrive/core/service/taskplugin"
"github.com/koderover/zadig/pkg/microservice/warpdrive/core/service/types"
"github.com/koderover/zadig/pkg/microservice/warpdrive/core/service/types/task"
@@ -45,7 +47,12 @@ var (
xl *zap.SugaredLogger
)
-// ExecHandler ...
+// TODO: Leave the logic here until we know why it exists.
+var durationBeforeNextTask = 10 * time.Second
+
+// Note: `durationTouchMsg` is used to emit `TOUCH` cmd and it should smaller than `durationBeforeNextTask`.
+var durationTouchMsg = 5 * time.Second
+
// Sender: sender to send ack/notification
// TaskPlugins: registered task plugin initiators to initiate specific plugin to execute task
type ExecHandler struct {
@@ -53,41 +60,49 @@ type ExecHandler struct {
TaskPlugins map[config.TaskType]plugins.Initiator
}
-// CancelHandler ...
type CancelHandler struct{}
-// HandleMessage ...
// Message handler to handle task execution message
func (h *ExecHandler) HandleMessage(message *nsq.Message) error {
defer func() {
// 每次处理完消息, 等待一段时间不处理新消息
- time.Sleep(time.Second * 10)
+ time.Sleep(durationBeforeNextTask)
}()
xl = log.SugaredLogger()
- // 如果存在运行中的 PipelineTask, 则重新requeue pipeline task
- // task处理逻辑全部放在requeue之后,防止requeue影响正在运行的task
- if pipelineTask != nil {
- xl.Infof("warpdrive instance have one running pipeline task %s:%d", pipelineTask.PipelineName, pipelineTask.TaskID)
- message.Requeue(time.Millisecond * 100)
- return nil
- }
-
// 获取 PipelineTask 内容
if err := json.Unmarshal(message.Body, &pipelineTask); err != nil {
xl.Errorf("unmarshal PipelineTask error: %v", err)
return nil
}
- xl.Infof("receiving pipeline task %s:%d message", pipelineTask.PipelineName, pipelineTask.TaskID)
+ taskName := fmt.Sprintf("%s:%d", pipelineTask.PipelineName, pipelineTask.TaskID)
+ xl.Infof("Receiving pipeline task %s message", taskName)
- // xl - global logger
xl = Logger(pipelineTask)
-
- // 初始化 Context, CancelFunc, PipelineTask
ctx, cancel = context.WithCancel(context.Background())
- go h.runPipelineTask(ctx, cancel, xl)
+ go func(ctx context.Context, taskName string) {
+ for {
+ select {
+ case <-ctx.Done():
+ xl.Infof("Pipeline task %q has been canceled. Exit.", taskName)
+ return
+ case <-time.After(durationTouchMsg):
+ if pipelineTask == nil {
+ xl.Infof("Pipeline task %q has completed. Exit.", taskName)
+ return
+ }
+
+ xl.Infof("After %s, touch message %q.", durationTouchMsg.String(), taskName)
+ message.Touch()
+ }
+ }
+ }(ctx, taskName)
+
+ h.runPipelineTask(ctx, cancel, xl)
+
+ // Note: If returning `nil`, we emit `FIN` cmd to nsq indicating that the messsage has been processed succefully.
return nil
}
@@ -128,7 +143,9 @@ func (h *ExecHandler) runPipelineTask(ctx context.Context, cancel context.Cancel
return
}
- // 选取当前最空闲的dockerhost
+ // Deprecated.
+ // Note: This logic is reserved for compatibility with plugins other than build. This logic can be removed if we have understood all of the plugins.
+ // For attached clusters, this logic is wrong because it still deals with dind in the local cluster.
dockerHost, err := plugins.GetBestDockerHost(pipelineTask.ConfigPayload.Docker.HostList, string(pipelineTask.Type), pipelineTask.ConfigPayload.Build.KubeNamespace, xl)
if err != nil {
errMsg := fmt.Sprintf("[%s]Cannot find docker host: %v", pipelineTask.PipelineName, err)
@@ -167,7 +184,6 @@ func (h *ExecHandler) runPipelineTask(ctx context.Context, cancel context.Cancel
// Return 之前会执行defer内容,更新pipeline end time, 发送ACK,发送notification
}
-// HandleMessage ...
func (h *CancelHandler) HandleMessage(message *nsq.Message) error {
xl = Logger(pipelineTask)
@@ -251,6 +267,7 @@ func (h *ExecHandler) SendNotification() {
Status: config.Status(pipelineTask.Status),
TeamName: pipelineTask.TeamName,
Type: pipelineTask.Type,
+ Stages: pipelineTask.Stages,
},
CreateTime: time.Now().Unix(),
IsRead: false,
@@ -268,7 +285,7 @@ func (h *ExecHandler) SendNotification() {
}
}
-func (h *ExecHandler) runStage(stagePosition int, stage *task.Stage) {
+func (h *ExecHandler) runStage(stagePosition int, stage *common.Stage, concurrency int64) {
xl.Infof("start to execute pipeline stage: %s at position: %d", stage.TaskType, stagePosition)
pluginInitiator, ok := h.TaskPlugins[stage.TaskType]
if !ok {
@@ -285,9 +302,8 @@ func (h *ExecHandler) runStage(stagePosition int, stage *task.Stage) {
// Default worker concurrency is 1, run tasks sequentially
var workerConcurrency = 1
if runParallel {
- // MaxWorkerInParallel is 5 for now
- if len(stage.SubTasks) > maxWorkerInParallel {
- workerConcurrency = maxWorkerInParallel
+ if len(stage.SubTasks) > int(concurrency) {
+ workerConcurrency = int(concurrency)
} else {
workerConcurrency = len(stage.SubTasks)
}
@@ -297,30 +313,68 @@ func (h *ExecHandler) runStage(stagePosition int, stage *task.Stage) {
// Task is struct for worker
var tasks []*Task
+ //tasks been preprocessed, map[serviceName]=[]Tasks
+ pluginsByService := make(map[string]*plugins.HelmDeployTaskPlugin)
+ // helm deploy plugins map[fullServiceName]=>HelmDeployPlugin
+ helmDeployPlugins := make(map[string]*plugins.HelmDeployTaskPlugin)
+
+ // preprocess subTasks, make batchTask with multiple subTasks
+ // eg: multiple deploys of same helm chart
+ if stage.TaskType == config.TaskDeploy || stage.TaskType == config.TaskResetImage {
+ for fullServiceName, subTask := range stage.SubTasks {
+ deployTask, err := plugins.ToDeployTask(subTask)
+ if err != nil {
+ xl.Errorf("failed to get deplot task, err: %s", err)
+ continue
+ }
+ if deployTask.ServiceType != setting.HelmDeployType {
+ continue
+ }
+ workerConcurrency = 1
+ pluginInstance := plugins.InitializeHelmDeployTaskPlugin(config.TaskDeploy)
+ pluginInstance.Task = deployTask
+ if _, ok := pluginsByService[deployTask.ServiceName]; !ok {
+ pluginsByService[deployTask.ServiceName] = pluginInstance
+ }
+ helmDeployPlugins[fullServiceName] = pluginInstance
+ }
+ }
// 每个SubTask会initiate一个plugin instance来执行
+ preProcessedServices := sets.NewString()
for serviceName, subTask := range stage.SubTasks {
+ if deployPlugin, ok := helmDeployPlugins[serviceName]; ok {
+ svcName := deployPlugin.Task.ServiceName
+ pluginsByService[svcName].ContentPlugins = append(pluginsByService[svcName].ContentPlugins, deployPlugin)
+ if !preProcessedServices.Has(svcName) {
+ xl.Infof("new batch sub task of service name: %s, type: %s", serviceName, stage.TaskType)
+ batchTask := NewTask(ctx, h.executeTask, pluginsByService[svcName], subTask, stagePosition, serviceName, xl)
+ tasks = append(tasks, batchTask)
+ }
+ preProcessedServices.Insert(svcName)
+ continue
+ }
+
var pluginInstance plugins.TaskPlugin
xl.Infof("new sub task of service name: %s, type: %s", serviceName, stage.TaskType)
pluginInstance = pluginInitiator(stage.TaskType)
- //xl.Errorf("%v", ctx.Value(CtxKeyBuildInfos))
- tasks = append(tasks, NewTask(ctx, h.executeTask, pluginInstance, subTask, stagePosition, serviceName, xl))
- }
- // 判断subTask是否是deploy,如果是的话判断是否是helm类型的服务,
- //todo helm类型的服务的部署暂时只支持串行执行
- for _, subTask := range stage.SubTasks {
- if deploy, err := plugins.ToDeployTask(subTask); err == nil {
- if deploy.ServiceType == "helm" {
- workerConcurrency = 1
- break
- }
- }
+ taskObj := NewTask(ctx, h.executeTask, pluginInstance, subTask, stagePosition, serviceName, xl)
+ tasks = append(tasks, taskObj)
}
// 设置WorkPool来控制最大并发数和并发执行
workerPool := NewPool(tasks, workerConcurrency)
// 发起workerConcurrency个并发执行,等待所有Task执行完成并返回
workerPool.Run()
+
+ // set related task status
+ for _, helmDeployPlugin := range helmDeployPlugins {
+ if len(helmDeployPlugin.ContentPlugins) == 0 {
+ continue
+ }
+ updatePluginSubTask(helmDeployPlugin, pipelineTask, stagePosition, helmDeployPlugin.Task.ContainerName, xl)
+ }
+
// Worker is completed
xl.Info("execution completed of subtasks in stage")
stageStatus := getStageStatus(workerPool.Tasks, xl)
@@ -337,7 +391,7 @@ func (h *ExecHandler) runStage(stagePosition int, stage *task.Stage) {
func (h *ExecHandler) execute(ctx context.Context, pipelineTask *task.Task, pipelineCtx *task.PipelineCtx, xl *zap.SugaredLogger) {
xl.Info("start pipeline task executor...")
// 如果是pipeline 1.0, 先将subtasks进行transform,转化为stages结构
- if pipelineTask.Type == config.SingleType || pipelineTask.Type == "" {
+ if pipelineTask.Type == config.SingleType || pipelineTask.Type == "" || pipelineTask.Type == config.WorkflowTypeV3 {
err := transformToStages(pipelineTask, xl)
// 初始化出错时,直接返回pipeline状态错误
if err != nil {
@@ -347,20 +401,48 @@ func (h *ExecHandler) execute(ctx context.Context, pipelineTask *task.Task, pipe
}
}
- // Stage之间仅支持串行
+ // Only serial is supported between stages
+ // If the stage status is StatusFailed/StatusCancelled/StatusTimeout, other than the extension stage will not be executed
+ isSkip := false
for stagePosition, stage := range pipelineTask.Stages {
- if !stage.AfterAll {
- h.runStage(stagePosition, stage)
- // 如果一个Stage执行失败了,跳出执行循环,并且更新pipelinetask状态为失败,发送ACK,并返回
- if stage.Status == config.StatusFailed || stage.Status == config.StatusCancelled || stage.Status == config.StatusTimeout {
- break
- }
+ if stage.AfterAll {
+ continue
+ }
+
+ if !isSkip || stage.TaskType == config.TaskExtension {
+ h.runStage(stagePosition, stage, pipelineTask.ConfigPayload.BuildConcurrency)
+ }
+
+ if stage.Status == config.StatusFailed || stage.Status == config.StatusCancelled || stage.Status == config.StatusTimeout {
+ isSkip = true
+ continue
}
}
+ updatePipelineStatus(pipelineTask, xl)
+ deployStageStatus := config.StatusInit
+ testStageStatus := config.StatusInit
for stagePosition, stage := range pipelineTask.Stages {
+ if stage.TaskType == config.TaskDeploy || stage.TaskType == config.TaskArtifact {
+ deployStageStatus = stage.Status
+ } else if stage.TaskType == config.TaskTestingV2 {
+ testStageStatus = stage.Status
+ }
if stage.AfterAll {
- h.runStage(stagePosition, stage)
+ if stage.TaskType == config.TaskResetImage {
+ switch pipelineTask.ResetImagePolicy {
+ case setting.ResetImagePolicyTaskCompleted, setting.ResetImagePolicyTaskCompletedOrder:
+ case setting.ResetImagePolicyDeployFailed:
+ if deployStageStatus == config.StatusInit || (deployStageStatus != config.StatusFailed && deployStageStatus != config.StatusCancelled && deployStageStatus != config.StatusTimeout) {
+ continue
+ }
+ case setting.ResetImagePolicyTestFailed:
+ if testStageStatus == config.StatusInit || (testStageStatus != config.StatusFailed && testStageStatus != config.StatusCancelled && testStageStatus != config.StatusTimeout) {
+ continue
+ }
+ }
+ }
+ h.runStage(stagePosition, stage, pipelineTask.ConfigPayload.BuildConcurrency)
}
}
@@ -411,6 +493,15 @@ func (h *ExecHandler) executeTask(taskCtx context.Context, plugin plugins.TaskPl
} else if pipelineTask.Type == config.ServiceType {
fileName = strings.Replace(strings.ToLower(fmt.Sprintf("%s-%s-%d-%s-%s", config.ServiceType, pipelineTask.PipelineName, pipelineTask.TaskID, plugin.Type(), servicename)),
"_", "-", -1)
+ } else if pipelineTask.Type == config.WorkflowTypeV3 {
+ fileName = strings.Replace(strings.ToLower(fmt.Sprintf("%s-%s-%d-%s-%s", config.WorkflowTypeV3, pipelineTask.PipelineName, pipelineTask.TaskID, plugin.Type(), fmt.Sprintf("%s-job", pipelineTask.PipelineName))),
+ "_", "-", -1)
+ } else if pipelineTask.Type == config.ArtifactType {
+ fileName = strings.Replace(strings.ToLower(fmt.Sprintf("%s-%s-%d-%s", config.ArtifactType, pipelineTask.PipelineName, pipelineTask.TaskID, plugin.Type())),
+ "_", "-", -1)
+ } else if pipelineTask.Type == config.ScanningType {
+ fileName = strings.Replace(strings.ToLower(fmt.Sprintf("%s-%s-%d-%s", config.ScanningType, pipelineTask.PipelineName, pipelineTask.TaskID, plugin.Type())),
+ "_", "-", -1)
}
plugin.Init(jobName, fileName, xl)
@@ -435,7 +526,12 @@ func (h *ExecHandler) executeTask(taskCtx context.Context, plugin plugins.TaskPl
}
// 设置 SubTask 初始状态
- plugin.SetStatus(config.StatusRunning)
+ switch plugin.Type() {
+ case config.TaskBuild, config.TaskTestingV2:
+ plugin.SetStatus(config.StatusPrepare)
+ default:
+ plugin.SetStatus(config.StatusRunning)
+ }
// 设置 SubTask 开始时间
plugin.SetStartTime()
@@ -443,18 +539,18 @@ func (h *ExecHandler) executeTask(taskCtx context.Context, plugin plugins.TaskPl
// 清除上一次错误信息
plugin.ResetError()
- updatePipelineSubTask(plugin.GetTask(), pipelineTask, pos, servicename, xl)
+ updatePluginSubTask(plugin, pipelineTask, pos, servicename, xl)
h.SendAck()
plugin.SetAckFunc(func() {
- updatePipelineSubTask(plugin.GetTask(), pipelineTask, pos, servicename, xl)
+ updatePluginSubTask(plugin, pipelineTask, pos, servicename, xl)
h.SendAck()
})
xl.Info("start to call plugin.Run")
// 如果是并行跑,用servicename来区分不同的workspace
runCtx := *pipelineCtx
- if pipelineTask.Type == config.WorkflowType {
+ if pipelineTask.Type == config.WorkflowType || pipelineTask.Type == config.WorkflowTypeV3 {
runCtx.Workspace = fmt.Sprintf("%s/%s", pipelineCtx.Workspace, servicename)
}
// 运行 SubTask, 如果需要异步,请在方法内实现
@@ -463,15 +559,18 @@ func (h *ExecHandler) executeTask(taskCtx context.Context, plugin plugins.TaskPl
// 如果 SubTask 执行失败, 则不继续执行, 发送 Task 失败执行结果
// Failed, Timeout, Cancelled
if plugin.IsTaskFailed() {
+ plugin.Complete(ctx, pipelineTask, servicename)
+ xl.Infof("task status: %s", plugin.Status())
+
plugin.SetEndTime()
- updatePipelineSubTask(plugin.GetTask(), pipelineTask, pos, servicename, xl)
+ updatePluginSubTask(plugin, pipelineTask, pos, servicename, xl)
return plugin.Status(), fmt.Errorf("pipeline task failed: task_handler:308")
}
xl.Infof("task status: %s", plugin.Status())
// 等待完成前, 更新 SubTask 执行结果到 PipelineTask
- updatePipelineSubTask(plugin.GetTask(), pipelineTask, pos, servicename, xl)
+ updatePluginSubTask(plugin, pipelineTask, pos, servicename, xl)
h.SendAck()
// 等待 SubTask 结束
@@ -488,7 +587,7 @@ func (h *ExecHandler) executeTask(taskCtx context.Context, plugin plugins.TaskPl
}
// 更新 SubTask 执行结果到 PipelineTask
plugin.SetEndTime()
- updatePipelineSubTask(plugin.GetTask(), pipelineTask, pos, servicename, xl)
+ updatePluginSubTask(plugin, pipelineTask, pos, servicename, xl)
h.SendAck()
xl.Infof("end sub task [%s:%s]", plugin.Type(), plugin.Status())
@@ -496,7 +595,6 @@ func (h *ExecHandler) executeTask(taskCtx context.Context, plugin plugins.TaskPl
}
func Logger(pipelineTask *task.Task) *zap.SugaredLogger {
- // 初始化Logger
l := log.Logger()
if pipelineTask != nil {
l.With(zap.String(setting.RequestID, pipelineTask.ReqID))
diff --git a/pkg/microservice/warpdrive/core/service/taskcontroller/task_helper.go b/pkg/microservice/warpdrive/core/service/taskcontroller/task_helper.go
index 867b82e336686a67232365bf6609fef68dd905c5..6076af284fe35d42007dfd7ced1bb07ae7e80620 100644
--- a/pkg/microservice/warpdrive/core/service/taskcontroller/task_helper.go
+++ b/pkg/microservice/warpdrive/core/service/taskcontroller/task_helper.go
@@ -29,6 +29,7 @@ import (
configbase "github.com/koderover/zadig/pkg/config"
"github.com/koderover/zadig/pkg/microservice/warpdrive/config"
+ "github.com/koderover/zadig/pkg/microservice/warpdrive/core/service/common"
plugins "github.com/koderover/zadig/pkg/microservice/warpdrive/core/service/taskplugin"
"github.com/koderover/zadig/pkg/microservice/warpdrive/core/service/taskplugin/github"
"github.com/koderover/zadig/pkg/microservice/warpdrive/core/service/taskplugin/s3"
@@ -44,7 +45,7 @@ import (
// 1. 转换经过序列化的 SubTasks 到 Stages
// Stage Map: *Stage -> map [service->subtask]
func transformToStages(pipelineTask *task.Task, xl *zap.SugaredLogger) error {
- var pipelineStages []*task.Stage
+ var pipelineStages []*common.Stage
// 工作流1.0,单服务工作流,SubTasks一维数组
// Transform into stages and assign to stages
// Task的数据结构中如果没有赋值Type,也按照1.0处理
@@ -57,7 +58,7 @@ func transformToStages(pipelineTask *task.Task, xl *zap.SugaredLogger) error {
}
// Pipeline 1.0中的一个Subtask对应到Pipeline 2.0中的一个Stage
// Stage中subtasks仅有一个subtask, key为service_name
- stage := &task.Stage{
+ stage := &common.Stage{
TaskType: subTaskPreview.TaskType,
// Pipeline 1.0中,每个type subtask只有一个,不存在并行执行
RunParallel: false,
@@ -91,6 +92,17 @@ func initPipelineTask(pipelineTask *task.Task, xl *zap.SugaredLogger) {
}
}
+func updatePluginSubTask(plugin plugins.TaskPlugin, pipelineTask *task.Task, pos int, servicename string, xl *zap.SugaredLogger) {
+ helmDeployPlugin, ok := plugin.(*plugins.HelmDeployTaskPlugin)
+ if ok && helmDeployPlugin != nil {
+ for _, p := range helmDeployPlugin.ContentPlugins {
+ updatePipelineSubTask(p.Task, pipelineTask, pos, p.Task.ContainerName, xl)
+ }
+ } else {
+ updatePipelineSubTask(plugin.GetTask(), pipelineTask, pos, servicename, xl)
+ }
+}
+
// Notes:
// 1.0中pos代表subtasks位置
// 2.0中pos代表stages位置
@@ -121,10 +133,10 @@ func updatePipelineSubTask(t interface{}, pipelineTask *task.Task, pos int, serv
// 同时更新stages
// TODO: 完善Stage其他字段
if len(pipelineTask.Stages) == 0 {
- pipelineTask.Stages = make([]*task.Stage, len(pipelineTask.SubTasks))
+ pipelineTask.Stages = make([]*common.Stage, len(pipelineTask.SubTasks))
}
if pipelineTask.Stages[pos] == nil {
- pipelineTask.Stages[pos] = &task.Stage{}
+ pipelineTask.Stages[pos] = &common.Stage{}
}
pipelineTask.Stages[pos].SubTasks = map[string]map[string]interface{}{servicename: subTask}
} else if pipelineTask.Type == config.WorkflowType {
@@ -136,6 +148,12 @@ func updatePipelineSubTask(t interface{}, pipelineTask *task.Task, pos int, serv
} else if pipelineTask.Type == config.ServiceType {
xl.Info("pipeline type is service type: pipeline 3.0")
pipelineTask.Stages[pos].SubTasks[servicename] = subTask
+ } else if pipelineTask.Type == config.WorkflowTypeV3 {
+ xl.Info("pipeline type is workflow type: pipeline 3.0")
+ pipelineTask.Stages[pos].SubTasks[servicename] = subTask
+ } else if pipelineTask.Type == config.ArtifactType {
+ xl.Info("pipeline type is artifact-package type: pipeline 3.0")
+ pipelineTask.Stages[pos].SubTasks[servicename] = subTask
}
}
@@ -144,7 +162,7 @@ func updatePipelineSubTask(t interface{}, pipelineTask *task.Task, pos int, serv
func updatePipelineStageStatus(stageStatus config.Status, pipelineTask *task.Task, pos int, xl *zap.SugaredLogger) {
xl.Infof("updating pipeline task, stage status: %s, stage position: %d", stageStatus, pos)
if pipelineTask.Stages[pos] == nil {
- pipelineTask.Stages[pos] = &task.Stage{}
+ pipelineTask.Stages[pos] = &common.Stage{}
}
pipelineTask.Stages[pos].Status = stageStatus
}
@@ -538,16 +556,22 @@ func getSubTaskTypeAndIsRestart(subTask map[string]interface{}) bool {
func initTaskPlugins(execHandler *ExecHandler) {
pluginConf := map[config.TaskType]plugins.Initiator{
- config.TaskJira: plugins.InitializeJiraTaskPlugin,
- config.TaskBuild: plugins.InitializeBuildTaskPlugin,
- config.TaskJenkinsBuild: plugins.InitializeJenkinsBuildPlugin,
- config.TaskDockerBuild: plugins.InitializeDockerBuildTaskPlugin,
- config.TaskDeploy: plugins.InitializeDeployTaskPlugin,
- config.TaskTestingV2: plugins.InitializeTestTaskPlugin,
- config.TaskSecurity: plugins.InitializeSecurityPlugin,
- config.TaskReleaseImage: plugins.InitializeReleaseImagePlugin,
- config.TaskDistributeToS3: plugins.InitializeDistribute2S3TaskPlugin,
- config.TaskResetImage: plugins.InitializeDeployTaskPlugin,
+ config.TaskJira: plugins.InitializeJiraTaskPlugin,
+ config.TaskBuild: plugins.InitializeBuildTaskPlugin,
+ config.TaskBuildV3: plugins.InitializeBuildTaskV3Plugin,
+ config.TaskArtifactDeploy: plugins.InitializeArtifactTaskPlugin,
+ config.TaskJenkinsBuild: plugins.InitializeJenkinsBuildPlugin,
+ config.TaskDockerBuild: plugins.InitializeDockerBuildTaskPlugin,
+ config.TaskDeploy: plugins.InitializeDeployTaskPlugin,
+ config.TaskTestingV2: plugins.InitializeTestTaskPlugin,
+ config.TaskSecurity: plugins.InitializeSecurityPlugin,
+ config.TaskReleaseImage: plugins.InitializeReleaseImagePlugin,
+ config.TaskDistributeToS3: plugins.InitializeDistribute2S3TaskPlugin,
+ config.TaskResetImage: plugins.InitializeDeployTaskPlugin,
+ config.TaskTrigger: plugins.InitializeTriggerTaskPlugin,
+ config.TaskArtifactPackage: plugins.InitializeArtifactPackagePlugin,
+ config.TaskExtension: plugins.InitializeExtensionTaskPlugin,
+ config.TaskScanning: plugins.InitializeScanningTaskPlugin,
}
for name, pluginInitiator := range pluginConf {
registerTaskPlugin(execHandler, name, pluginInitiator)
diff --git a/pkg/microservice/warpdrive/core/service/taskplugin/artifact_deploy.go b/pkg/microservice/warpdrive/core/service/taskplugin/artifact_deploy.go
new file mode 100644
index 0000000000000000000000000000000000000000..581085f09785f63c11289b40698c9b2bf3b9f156
--- /dev/null
+++ b/pkg/microservice/warpdrive/core/service/taskplugin/artifact_deploy.go
@@ -0,0 +1,385 @@
+/*
+Copyright 2021 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 taskplugin
+
+import (
+ "context"
+ "fmt"
+ "math/rand"
+ "strconv"
+ "strings"
+ "time"
+
+ "go.uber.org/zap"
+ "gopkg.in/yaml.v3"
+ "k8s.io/apimachinery/pkg/util/sets"
+ "k8s.io/client-go/kubernetes"
+ "k8s.io/client-go/rest"
+ "sigs.k8s.io/controller-runtime/pkg/client"
+
+ zadigconfig "github.com/koderover/zadig/pkg/config"
+ "github.com/koderover/zadig/pkg/microservice/warpdrive/config"
+ "github.com/koderover/zadig/pkg/microservice/warpdrive/core/service/types/task"
+ "github.com/koderover/zadig/pkg/setting"
+ krkubeclient "github.com/koderover/zadig/pkg/tool/kube/client"
+ "github.com/koderover/zadig/pkg/tool/kube/updater"
+)
+
+const (
+ // ArtifactDeployTaskV2Timeout ...
+ ArtifactDeployTaskV2Timeout = 60 * 60 * 1 // 60 minutes
+)
+
+// InitializeArtifactTaskPlugin to initialize build task plugin, and return reference
+func InitializeArtifactTaskPlugin(taskType config.TaskType) TaskPlugin {
+ return &ArtifactDeployTaskPlugin{
+ Name: taskType,
+ kubeClient: krkubeclient.Client(),
+ clientset: krkubeclient.Clientset(),
+ restConfig: krkubeclient.RESTConfig(),
+ }
+}
+
+// BuildTaskPlugin is Plugin, name should be compatible with task type
+type ArtifactDeployTaskPlugin struct {
+ Name config.TaskType
+ KubeNamespace string
+ JobName string
+ FileName string
+ kubeClient client.Client
+ clientset kubernetes.Interface
+ restConfig *rest.Config
+ Task *task.Build
+ Log *zap.SugaredLogger
+
+ ack func()
+}
+
+func (p *ArtifactDeployTaskPlugin) SetAckFunc(ack func()) {
+ p.ack = ack
+}
+
+// Init ...
+func (p *ArtifactDeployTaskPlugin) Init(jobname, filename string, xl *zap.SugaredLogger) {
+ p.JobName = jobname
+ p.Log = xl
+ p.FileName = filename
+}
+
+func (p *ArtifactDeployTaskPlugin) Type() config.TaskType {
+ return p.Name
+}
+
+// Status ...
+func (p *ArtifactDeployTaskPlugin) Status() config.Status {
+ return p.Task.TaskStatus
+}
+
+// SetStatus ...
+func (p *ArtifactDeployTaskPlugin) SetStatus(status config.Status) {
+ p.Task.TaskStatus = status
+}
+
+// TaskTimeout ...
+func (p *ArtifactDeployTaskPlugin) TaskTimeout() int {
+ if p.Task.Timeout == 0 {
+ p.Task.Timeout = ArtifactDeployTaskV2Timeout
+ } else {
+ if !p.Task.IsRestart {
+ p.Task.Timeout = p.Task.Timeout * 60
+ }
+ }
+ return p.Task.Timeout
+}
+
+func (p *ArtifactDeployTaskPlugin) SetBuildStatusCompleted(status config.Status) {
+ p.Task.BuildStatus.Status = status
+ p.Task.BuildStatus.EndTime = time.Now().Unix()
+}
+
+func (p *ArtifactDeployTaskPlugin) Run(ctx context.Context, pipelineTask *task.Task, pipelineCtx *task.PipelineCtx, serviceName string) {
+ switch p.Task.ClusterID {
+ case setting.LocalClusterID:
+ p.KubeNamespace = zadigconfig.Namespace()
+ default:
+ p.KubeNamespace = setting.AttachedClusterNamespace
+
+ crClient, clientset, restConfig, err := GetK8sClients(pipelineTask.ConfigPayload.HubServerAddr, p.Task.ClusterID)
+ if err != nil {
+ p.Log.Error(err)
+ p.Task.TaskStatus = config.StatusFailed
+ p.Task.Error = err.Error()
+ p.SetBuildStatusCompleted(config.StatusFailed)
+ return
+ }
+
+ p.kubeClient = crClient
+ p.clientset = clientset
+ p.restConfig = restConfig
+ }
+
+ envName := pipelineTask.WorkflowArgs.Namespace
+ envNameVar := &task.KeyVal{Key: "ENV_NAME", Value: envName, IsCredential: false}
+ p.Task.JobCtx.EnvVars = append(p.Task.JobCtx.EnvVars, envNameVar)
+
+ taskIDVar := &task.KeyVal{Key: "TASK_ID", Value: strconv.FormatInt(pipelineTask.TaskID, 10), IsCredential: false}
+ p.Task.JobCtx.EnvVars = append(p.Task.JobCtx.EnvVars, taskIDVar)
+
+ privateKeys := sets.String{}
+ for _, privateKey := range pipelineTask.ConfigPayload.PrivateKeys {
+ privateKeys.Insert(privateKey.Name)
+ }
+
+ privateKeysVar := &task.KeyVal{Key: "AGENTS", Value: strings.Join(privateKeys.List(), ","), IsCredential: false}
+ p.Task.JobCtx.EnvVars = append(p.Task.JobCtx.EnvVars, privateKeysVar)
+
+ // env host ips
+ for envName, HostIPs := range p.Task.EnvHostInfo {
+ envHostKeysVar := &task.KeyVal{Key: envName + "_HOST_IPs", Value: strings.Join(HostIPs, ","), IsCredential: false}
+ p.Task.JobCtx.EnvVars = append(p.Task.JobCtx.EnvVars, envHostKeysVar)
+ }
+
+ // env host names
+ for envName, names := range p.Task.EnvHostNames {
+ envHostKeysVar := &task.KeyVal{Key: envName + "_HOST_NAMEs", Value: strings.Join(names, ","), IsCredential: false}
+ p.Task.JobCtx.EnvVars = append(p.Task.JobCtx.EnvVars, envHostKeysVar)
+ }
+
+ // ARTIFACT
+ if p.Task.ArtifactInfo != nil {
+ var workspace = "/workspace"
+ if pipelineTask.ConfigPayload.ClassicBuild {
+ workspace = pipelineCtx.Workspace
+ }
+ pipelineTask.ArtifactInfo = p.Task.ArtifactInfo
+ artifactKeysVar := &task.KeyVal{Key: "ARTIFACT", Value: fmt.Sprintf("%s/%s", workspace, p.Task.ArtifactInfo.FileName), IsCredential: false}
+ p.Task.JobCtx.EnvVars = append(p.Task.JobCtx.EnvVars, artifactKeysVar)
+ }
+
+ p.KubeNamespace = pipelineTask.ConfigPayload.Build.KubeNamespace
+
+ //instantiates variables like ${