test/Rainbond
1
0
Fork 0
mirror of https://gitee.com/rainbond/Rainbond.git synced 2024-12-02 19:57:42 +08:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Merge branch 'V3.7'

Browse Source
This commit is contained in:
barnettZQG 2018-10-24 10:20:33 +08:00
commit c221724bc6
43 changed files with 474 additions and 259 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
api/apiFunc/api_interface.go
View File

@ -39,6 +39,8 @@ type TenantInterface interface {
TransPlugins(w http.ResponseWriter, r *http.Request)
ServicesCount(w http.ResponseWriter, r *http.Request)
GetManyDeployVersion(w http.ResponseWriter, r *http.Request)
LimitTenantMemory(w http.ResponseWriter, r *http.Request)
TenantResourcesStatus(w http.ResponseWriter, r *http.Request)
}
//ServiceInterface ServiceInterface

3
api/apiRouters/version2/v2Routers.go
View File

@ -101,6 +101,9 @@ func (v2 *V2) tenantNameRouter() chi.Router {
r.Post("/groupapp/backups/{backup_id}/restore", controller.Restore)
r.Get("/groupapp/backups/{backup_id}/restore/{restore_id}", controller.RestoreResult)
r.Post("/deployversions", controller.GetManager().GetManyDeployVersion)
//团队资源限制
r.Post("/limit_memory", controller.GetManager().LimitTenantMemory)
r.Get("/limit_memory", controller.GetManager().TenantResourcesStatus)
return r
}

5
api/controller/app.go
View File

@ -13,13 +13,12 @@ import (
"github.com/Sirupsen/logrus"
"github.com/go-chi/chi"
"github.com/goodrain/rainbond/api/controller/upload"
"github.com/goodrain/rainbond/api/handler"
"github.com/goodrain/rainbond/api/model"
"github.com/goodrain/rainbond/db"
httputil "github.com/goodrain/rainbond/util/http"
"github.com/jinzhu/gorm"
"github.com/goodrain/rainbond/api/controller/upload"
"github.com/ncw/directio"
)
type AppStruct struct{}
@ -197,7 +196,7 @@ func (a *AppStruct) Upload(w http.ResponseWriter, r *http.Request) {
os.MkdirAll(dirName, 0755)
fileName := fmt.Sprintf("%s/%s", dirName, header.Filename)
file, err := directio.OpenFile(fileName, os.O_WRONLY|os.O_CREATE, 0644)
file, err := os.OpenFile(fileName, os.O_WRONLY|os.O_CREATE, 0644)
if err != nil {
logrus.Errorf("Failed to open file: ", err.Error())
httputil.ReturnError(r, w, 502, "Failed to open file: "+err.Error())

87
api/controller/serviceAction.go
View File

@ -40,6 +40,8 @@ import (
httputil "github.com/goodrain/rainbond/util/http"
"github.com/jinzhu/gorm"
"github.com/thedevsaddam/govalidator"
"io/ioutil"
"github.com/pquerna/ffjson/ffjson"
)
//TIMELAYOUT timelayout
@ -863,3 +865,88 @@ func (t *TenantStruct) RollBack(w http.ResponseWriter, r *http.Request) {
httputil.ReturnSuccess(r, w, sEvent)
return
}
type limitMemory struct {
LimitMemory int `json:"limit_memory"`
}
func (t *TenantStruct) LimitTenantMemory(w http.ResponseWriter, r *http.Request) {
var lm limitMemory
body, err := ioutil.ReadAll(r.Body)
if err != nil {
httputil.ReturnError(r, w, 500, err.Error())
return
}
err = ffjson.Unmarshal(body, &lm)
if err != nil {
httputil.ReturnError(r, w, 500, err.Error())
return
}
tenantID := r.Context().Value(middleware.ContextKey("tenant_id")).(string)
tenant, err := db.GetManager().TenantDao().GetTenantByUUID(tenantID)
if err != nil {
httputil.ReturnError(r, w, 400, err.Error())
return
}
tenant.LimitMemory = lm.LimitMemory
if err := db.GetManager().TenantDao().UpdateModel(tenant); err != nil {
httputil.ReturnError(r, w, 500, err.Error())
}
httputil.ReturnSuccess(r, w, "success!")
}
type SourcesInfo struct {
TenantId string `json:"tenant_id"`
AvailableMemory int `json:"available_memory"`
Status bool `json:"status"`
}
func (t *TenantStruct) TenantResourcesStatus(w http.ResponseWriter, r *http.Request) {
tenantID := r.Context().Value(middleware.ContextKey("tenant_id")).(string)
tenant, err := db.GetManager().TenantDao().GetTenantByUUID(tenantID)
if err != nil {
httputil.ReturnError(r, w, 400, err.Error())
return
}
//11ms
services, err := handler.GetServiceManager().GetService(tenant.UUID)
if err != nil {
msg := httputil.ResponseBody{
Msg: fmt.Sprintf("get service error, %v", err),
}
httputil.Return(r, w, 500, msg)
return
}
statsInfo, _ := handler.GetTenantManager().StatsMemCPU(services)
if tenant.LimitMemory == 0 {
sourcesInfo := SourcesInfo{
TenantId: tenantID,
AvailableMemory: 0,
Status: true,
}
httputil.ReturnSuccess(r, w, sourcesInfo)
return
}
if statsInfo.MEM >= tenant.LimitMemory {
sourcesInfo := SourcesInfo{
TenantId: tenantID,
AvailableMemory: tenant.LimitMemory - statsInfo.MEM,
Status: false,
}
httputil.ReturnSuccess(r, w, sourcesInfo)
} else {
sourcesInfo := SourcesInfo{
TenantId: tenantID,
AvailableMemory: tenant.LimitMemory - statsInfo.MEM,
Status: true,
}
httputil.ReturnSuccess(r, w, sourcesInfo)
}
}

105
api/db/db.go
View File

@ -36,6 +36,8 @@ import (
"github.com/Sirupsen/logrus"
tsdbClient "github.com/bluebreezecf/opentsdb-goclient/client"
tsdbConfig "github.com/bluebreezecf/opentsdb-goclient/config"
"github.com/jinzhu/gorm"
dbModel "github.com/goodrain/rainbond/db/model"
)
//ConDB struct
@ -66,7 +68,9 @@ func CreateDBManager(conf option.Config) error {
logrus.Errorf("get db manager failed,%s", err.Error())
return err
}
logrus.Debugf("init db manager success")
// api database initialization
go dataInitialization()
return nil
}
@ -195,3 +199,102 @@ func BuildTaskBuild(t *BuildTaskStruct) (*pb.EnqueueRequest, error) {
}
return &er, nil
}
func GetBegin() *gorm.DB {
return db.GetManager().Begin()
}
func dbInit() error {
logrus.Info("api database initialization starting...")
begin := GetBegin()
// Permissions set
var rac dbModel.RegionAPIClass
if err := begin.Where("class_level=? and prefix=?", "server_source", "/v2/show").Find(&rac).Error; err != nil {
if err == gorm.ErrRecordNotFound {
data := map[string]string{
"/v2/show": "server_source",
"/v2/opentsdb": "server_source",
"/v2/resources": "server_source",
"/v2/builder": "server_source",
"/v2/tenants": "server_source",
"/v2/app": "server_source",
"/api/v1": "server_source",
"/v2/nodes": "node_manager",
"/v2/job": "node_manager",
"/v2/tasks": "node_manager",
"/v2/taskgroups": "node_manager",
"/v2/tasktemps": "node_manager",
"/v2/configs": "node_manager",
}
tx := begin
var rollback bool
for k, v := range data {
if err := db.GetManager().RegionAPIClassDaoTransactions(tx).AddModel(&dbModel.RegionAPIClass{
ClassLevel: v,
Prefix: k,
}); err != nil {
tx.Rollback()
rollback = true
break
}
}
if !rollback {
tx.Commit()
}
} else {
return err
}
}
//Port Protocol support
var rps dbModel.RegionProcotols
if err := begin.Where("protocol_group=? and protocol_child=?", "http", "http").Find(&rps).Error; err != nil {
if err == gorm.ErrRecordNotFound {
data := map[string][]string{
"http": []string{"http"},
"stream": []string{"mysql", "tcp", "udp"},
}
tx := begin
var rollback bool
for k, v := range data {
for _, v1 := range v {
if err := db.GetManager().RegionProcotolsDaoTransactions(tx).AddModel(&dbModel.RegionProcotols{
ProtocolGroup: k,
ProtocolChild: v1,
APIVersion: "v2",
IsSupport: true,
}); err != nil {
tx.Rollback()
rollback = true
break
}
}
}
if !rollback {
tx.Commit()
}
} else {
return err
}
}
return nil
}
func dataInitialization() {
timer := time.NewTimer(time.Second * 2)
defer timer.Stop()
for {
err := dbInit()
if err != nil {
logrus.Error("Initializing database failed, ", err)
}else {
logrus.Info("api database initialization success!")
return
}
select {
case <-timer.C:
timer.Reset(time.Second * 2)
}
}
}

4
api/handler/service.go
View File

@ -226,7 +226,7 @@ func (s *ServiceAction) buildFromSourceCode(r *api_model.BuildServiceStruct, ser
body["envs"] = r.Body.ENVS
body["tenant_name"] = r.Body.TenantName
body["branch"] = r.Body.Branch
body["server_type"] = r.Body.ServiceType
body["server_type"] = r.Body.ServerType
body["service_alias"] = r.Body.ServiceAlias
if r.Body.User != "" && r.Body.Password != "" {
body["user"] = r.Body.User
@ -238,7 +238,6 @@ func (s *ServiceAction) buildFromSourceCode(r *api_model.BuildServiceStruct, ser
}
func (s *ServiceAction) sourceBuild(r *api_model.BuildServiceStruct, service *dbmodel.TenantServices) error {
logrus.Debugf("build from source")
if r.Body.RepoURL == "" || r.Body.DeployVersion == "" || r.Body.EventID == "" {
return fmt.Errorf("args error")
}
@ -258,6 +257,7 @@ func (s *ServiceAction) sourceBuild(r *api_model.BuildServiceStruct, service *db
body["tenant_name"] = r.Body.TenantName
body["service_alias"] = r.Body.ServiceAlias
body["expire"] = 180
body["server_type"] = r.Body.ServerType
logrus.Debugf("app_build body is %v", body)
return s.sendTask(body, "app_build")
}

8
api/model/model.go
View File

@ -701,7 +701,11 @@ type BuildServiceStruct struct {
// 操作人员
// in: body
// required: false
Lang string `json:"lang" validate:"lang"`
Lang string `json:"lang" validate:"lang"`
// 代码服务器类型
// in: body
// required: false
ServerType string `json:"server_type" validate:"server_type"`
Runtime string `json:"runtime" validate:"runtime"`
ServiceType string `json:"service_type" validate:"service_type"`
User string `json:"user" validate:"user"`
@ -709,7 +713,7 @@ type BuildServiceStruct struct {
Operator string `json:"operator" validate:"operator"`
TenantName string `json:"tenant_name"`
ServiceAlias string `json:"service_alias"`
Cmd string `json:"cmd"`
Cmd string `json:"cmd"`
//用于云市代码包创建
SlugInfo struct {
SlugPath string `json:"slug_path"`

4
builder/exector/build_from_sourcecode_run.go
View File

@ -22,6 +22,7 @@ import (
"fmt"
"os"
"path"
"strings"
"time"
"github.com/goodrain/rainbond/util"
@ -82,7 +83,7 @@ func NewSouceCodeBuildItem(in []byte) *SourceCodeBuildItem {
eventID := gjson.GetBytes(in, "event_id").String()
logger := event.GetManager().GetLogger(eventID)
csi := sources.CodeSourceInfo{
ServerType: gjson.GetBytes(in, "server_type").String(),
ServerType: strings.Replace(gjson.GetBytes(in, "server_type").String(), " ", "", -1),
RepositoryURL: gjson.GetBytes(in, "repo_url").String(),
Branch: gjson.GetBytes(in, "branch").String(),
User: gjson.GetBytes(in, "user").String(),
@ -113,7 +114,6 @@ func NewSouceCodeBuildItem(in []byte) *SourceCodeBuildItem {
scb.CacheDir = fmt.Sprintf("/cache/build/%s/cache/%s", scb.TenantID, scb.ServiceID)
//scb.SourceDir = scb.CodeSouceInfo.GetCodeSourceDir()
scb.TGZDir = fmt.Sprintf("/grdata/build/tenant/%s/slug/%s", scb.TenantID, scb.ServiceID)
scb.CodeSouceInfo.InitServerType()
return scb
}

93
builder/exector/exector.go
View File

@ -24,8 +24,6 @@ import (
"time"
"github.com/Sirupsen/logrus"
//"github.com/docker/docker/client"
"sync"
"github.com/coreos/etcd/clientv3"
"github.com/docker/engine-api/client"
@ -36,6 +34,7 @@ import (
"github.com/goodrain/rainbond/mq/api/grpc/pb"
"github.com/goodrain/rainbond/util"
"github.com/tidwall/gjson"
"sync"
)
var TaskNum float64 = 0
@ -73,13 +72,15 @@ func NewManager(conf config.Config) (Manager, error) {
return &exectorManager{
DockerClient: dockerClient,
EtcdCli: etcdCli,
tasks: make(map[*pb.TaskMessage][]byte),
}, nil
}
type exectorManager struct {
DockerClient *client.Client
EtcdCli *clientv3.Client
wg sync.WaitGroup
tasks map[*pb.TaskMessage][]byte
taskLock sync.RWMutex
}
//TaskWorker worker interface
@ -109,44 +110,44 @@ func RegisterWorker(name string, fun func([]byte, *exectorManager) (TaskWorker,
//share-slug share app with slug
//share-image share app with image
func (e *exectorManager) AddTask(task *pb.TaskMessage) error {
e.wg.Add(1)
e.tasks[task] = task.TaskBody
TaskNum++
switch task.TaskType {
case "build_from_image":
e.buildFromImage(task.TaskBody)
e.buildFromImage(task)
case "build_from_source_code":
e.buildFromSourceCode(task.TaskBody)
e.buildFromSourceCode(task)
case "build_from_market_slug":
e.buildFromMarketSlug(task.TaskBody)
e.buildFromMarketSlug(task)
case "service_check":
go e.serviceCheck(task.TaskBody)
go e.serviceCheck(task)
case "plugin_image_build":
e.pluginImageBuild(task.TaskBody)
e.pluginImageBuild(task)
case "plugin_dockerfile_build":
e.pluginDockerfileBuild(task.TaskBody)
e.pluginDockerfileBuild(task)
case "share-slug":
e.slugShare(task.TaskBody)
e.slugShare(task)
case "share-image":
e.imageShare(task.TaskBody)
e.imageShare(task)
default:
return e.exec(task.TaskType, task.TaskBody)
return e.exec(task)
}
return nil
}
func (e *exectorManager) exec(workerName string, in []byte) error {
creater, ok := workerCreaterList[workerName]
func (e *exectorManager) exec(task *pb.TaskMessage) error {
creater, ok := workerCreaterList[task.TaskType]
if !ok {
return fmt.Errorf("`%s` tasktype can't support", workerName)
return fmt.Errorf("`%s` tasktype can't support", task.TaskType)
}
worker, err := creater(in, e)
worker, err := creater(task.TaskBody, e)
if err != nil {
logrus.Errorf("create worker for builder error.%s", err)
return err
}
go func() {
defer e.wg.Done()
defer e.removeTask(task)
defer event.GetManager().ReleaseLogger(worker.GetLogger())
defer func() {
if r := recover(); r != nil {
@ -165,14 +166,14 @@ func (e *exectorManager) exec(workerName string, in []byte) error {
}
//buildFromImage build app from docker image
func (e *exectorManager) buildFromImage(in []byte) {
i := NewImageBuildItem(in)
func (e *exectorManager) buildFromImage(task *pb.TaskMessage) {
i := NewImageBuildItem(task.TaskBody)
i.DockerClient = e.DockerClient
i.Logger.Info("从镜像构建应用任务开始执行", map[string]string{"step": "builder-exector", "status": "starting"})
status := "success"
go func() {
start := time.Now()
defer e.wg.Done()
defer e.removeTask(task)
logrus.Debugf("start build from image worker")
defer event.GetManager().ReleaseLogger(i.Logger)
defer func() {
@ -208,14 +209,14 @@ func (e *exectorManager) buildFromImage(in []byte) {
//buildFromSourceCode build app from source code
//support git repository
func (e *exectorManager) buildFromSourceCode(in []byte) {
i := NewSouceCodeBuildItem(in)
func (e *exectorManager) buildFromSourceCode(task *pb.TaskMessage) {
i := NewSouceCodeBuildItem(task.TaskBody)
i.DockerClient = e.DockerClient
i.Logger.Info("Build app version from source code start", map[string]string{"step": "builder-exector", "status": "starting"})
status := "success"
go func() {
start := time.Now()
defer e.wg.Done()
e.removeTask(task)
logrus.Debugf("start build from source code")
defer event.GetManager().ReleaseLogger(i.Logger)
defer func() {
@ -250,11 +251,11 @@ func (e *exectorManager) buildFromSourceCode(in []byte) {
}
//buildFromMarketSlug build app from market slug
func (e *exectorManager) buildFromMarketSlug(in []byte) {
eventID := gjson.GetBytes(in, "event_id").String()
func (e *exectorManager) buildFromMarketSlug(task *pb.TaskMessage) {
eventID := gjson.GetBytes(task.TaskBody, "event_id").String()
logger := event.GetManager().GetLogger(eventID)
logger.Info("云市应用代码包构建任务开始执行", map[string]string{"step": "builder-exector", "status": "starting"})
i, err := NewMarketSlugItem(in)
i, err := NewMarketSlugItem(task.TaskBody)
if err != nil {
logrus.Error("create build from market slug task error.", err.Error())
return
@ -262,7 +263,7 @@ func (e *exectorManager) buildFromMarketSlug(in []byte) {
i.Logger.Info("开始构建应用", map[string]string{"step": "builder-exector", "status": "starting"})
go func() {
start := time.Now()
defer e.wg.Done()
e.removeTask(task)
defer event.GetManager().ReleaseLogger(i.Logger)
defer func() {
if r := recover(); r != nil {
@ -293,8 +294,8 @@ func (e *exectorManager) buildFromMarketSlug(in []byte) {
}
//slugShare share app of slug
func (e *exectorManager) slugShare(in []byte) {
i, err := NewSlugShareItem(in, e.EtcdCli)
func (e *exectorManager) slugShare(task *pb.TaskMessage) {
i, err := NewSlugShareItem(task.TaskBody, e.EtcdCli)
if err != nil {
logrus.Error("create share image task error.", err.Error())
return
@ -302,7 +303,7 @@ func (e *exectorManager) slugShare(in []byte) {
i.Logger.Info("开始分享应用", map[string]string{"step": "builder-exector", "status": "starting"})
status := "success"
go func() {
defer e.wg.Done()
defer e.removeTask(task)
defer event.GetManager().ReleaseLogger(i.Logger)
defer func() {
if r := recover(); r != nil {
@ -334,8 +335,8 @@ func (e *exectorManager) slugShare(in []byte) {
}
//imageShare share app of docker image
func (e *exectorManager) imageShare(in []byte) {
i, err := NewImageShareItem(in, e.DockerClient, e.EtcdCli)
func (e *exectorManager) imageShare(task *pb.TaskMessage) {
i, err := NewImageShareItem(task.TaskBody, e.DockerClient, e.EtcdCli)
if err != nil {
logrus.Error("create share image task error.", err.Error())
return
@ -343,7 +344,7 @@ func (e *exectorManager) imageShare(in []byte) {
i.Logger.Info("开始分享应用", map[string]string{"step": "builder-exector", "status": "starting"})
status := "success"
go func() {
defer e.wg.Done()
e.removeTask(task)
defer event.GetManager().ReleaseLogger(i.Logger)
defer func() {
if r := recover(); r != nil {
@ -379,7 +380,29 @@ func (e *exectorManager) Start() error {
}
func (e *exectorManager) Stop() error {
logrus.Info("Waiting for all threads to exit.")
e.wg.Wait()
i := 0
timer := time.NewTimer(time.Second * 2)
defer timer.Stop()
for {
if i >= 15 {
logrus.Errorf("There are %d tasks not completed", len(e.tasks))
return fmt.Errorf("There are %d tasks not completed ", len(e.tasks))
}
if len(e.tasks) == 0 {
break
}
select {
case <-timer.C:
i ++
timer.Reset(time.Second * 2)
}
}
logrus.Info("All threads is exited.")
return nil
}
func (e *exectorManager) removeTask(task *pb.TaskMessage) {
e.taskLock.Lock()
defer e.taskLock.Unlock()
delete(e.tasks, task)
}

7
builder/exector/import_app.go
View File

@ -207,7 +207,7 @@ func (i *ImportApp) importApp() error {
info := strings.Split(shareSlugPath, "/")
shareSlugPath = fmt.Sprintf("%s/%s", i.ServiceSlug.NameSpace, strings.Join(info[1:], "/"))
}
return shareSlugPath
return strings.Replace(shareSlugPath, "//", "/", -1)
}
if oldimage, ok := app.CheckGet("share_image"); ok {
@ -467,7 +467,10 @@ func (i *ImportApp) loadApps() error {
return err
}
} else {
os.MkdirAll(filepath.Base(shareSlugPath), 0755)
if err := util.CheckAndCreateDir(filepath.Dir(shareSlugPath)); err != nil {
logrus.Error("Failed create slug file directory %s error %s ", filepath.Dir(shareSlugPath), err.Error())
return err
}
err := util.CopyFile(fileName, shareSlugPath)
if err != nil {
logrus.Error("Failed to copy slug file to local directory: ", shareSlugPath)

6
builder/exector/plugin_dockerfile.go
View File

@ -36,6 +36,7 @@ import (
"github.com/goodrain/rainbond/builder/model"
"github.com/Sirupsen/logrus"
"github.com/goodrain/rainbond/mq/api/grpc/pb"
)
const (
@ -44,9 +45,9 @@ const (
formatSourceDir = "/cache/build/%s/source/%s"
)
func (e *exectorManager) pluginDockerfileBuild(in []byte) {
func (e *exectorManager) pluginDockerfileBuild(task *pb.TaskMessage) {
var tb model.BuildPluginTaskBody
if err := ffjson.Unmarshal(in, &tb); err != nil {
if err := ffjson.Unmarshal(task.TaskBody, &tb); err != nil {
logrus.Errorf("unmarshal taskbody error, %v", err)
return
}
@ -55,6 +56,7 @@ func (e *exectorManager) pluginDockerfileBuild(in []byte) {
logger.Info("从dockerfile构建插件任务开始执行", map[string]string{"step": "builder-exector", "status": "starting"})
go func() {
logrus.Info("start exec build plugin from image worker")
defer e.removeTask(task)
defer event.GetManager().ReleaseLogger(logger)
for retry := 0; retry < 2; retry++ {
err := e.runD(&tb, logger)

6
builder/exector/plugin_image.go
View File

@ -33,11 +33,12 @@ import (
"github.com/goodrain/rainbond/builder/model"
"github.com/Sirupsen/logrus"
"github.com/goodrain/rainbond/mq/api/grpc/pb"
)
func (e *exectorManager) pluginImageBuild(in []byte) {
func (e *exectorManager) pluginImageBuild(task *pb.TaskMessage) {
var tb model.BuildPluginTaskBody
if err := ffjson.Unmarshal(in, &tb); err != nil {
if err := ffjson.Unmarshal(task.TaskBody, &tb); err != nil {
logrus.Errorf("unmarshal taskbody error, %v", err)
return
}
@ -46,6 +47,7 @@ func (e *exectorManager) pluginImageBuild(in []byte) {
logger.Info("从镜像构建插件任务开始执行", map[string]string{"step": "builder-exector", "status": "starting"})
go func() {
logrus.Info("start exec build plugin from image worker")
defer e.removeTask(task)
defer event.GetManager().ReleaseLogger(logger)
for retry := 0; retry < 2; retry++ {
err := e.run(&tb, logger)

6
builder/exector/service_check.go
View File

@ -28,6 +28,7 @@ import (
"github.com/goodrain/rainbond/builder/parser"
"github.com/goodrain/rainbond/event"
"github.com/pquerna/ffjson/ffjson"
"github.com/goodrain/rainbond/mq/api/grpc/pb"
)
//ServiceCheckInput 任务输入数据
@ -76,17 +77,18 @@ func CreateResult(ErrorInfos parser.ParseErrorList, ServiceInfo []parser.Service
}
//serviceCheck 应用创建源检测
func (e *exectorManager) serviceCheck(in []byte) {
func (e *exectorManager) serviceCheck(task *pb.TaskMessage) {
//step1 判断应用源类型
//step2 获取应用源介质，镜像Or源码
//step3 解析判断应用源规范
//完成
var input ServiceCheckInput
if err := ffjson.Unmarshal(in, &input); err != nil {
if err := ffjson.Unmarshal(task.TaskBody, &input); err != nil {
logrus.Error("Unmarshal service check input data error.", err.Error())
return
}
logger := event.GetManager().GetLogger(input.EventID)
defer e.removeTask(task)
defer event.GetManager().ReleaseLogger(logger)
defer func() {
if r := recover(); r != nil {

4
builder/parser/source_code.go
View File

@ -89,7 +89,6 @@ func (d *SourceCodeParse) Parse() ParseErrorList {
if csi.Branch == "" {
csi.Branch = "master"
}
csi.InitServerType()
if csi.RepositoryURL == "" {
d.logger.Error("Git项目仓库地址不能为空", map[string]string{"step": "parse"})
d.errappend(ErrorAndSolve(FatalError, "Git项目仓库地址格式错误", SolveAdvice("modify_url", "请确认并修改仓库地址")))
@ -185,7 +184,7 @@ func (d *SourceCodeParse) Parse() ParseErrorList {
d.branchs = rs.Branchs
return nil
}
logrus.Debugf("start get service code by %s server type", csi.ServerType)
//获取代码仓库
switch csi.ServerType {
case "git":
@ -198,6 +197,7 @@ func (d *SourceCodeParse) Parse() ParseErrorList {
}
default:
//default git
logrus.Warningf("do not get void server type,default use git")
if err := gitFunc(); err != nil && err.IsFatalError() {
return err
}

9
builder/sources/git.go
View File

@ -63,15 +63,6 @@ type CodeSourceInfo struct {
ServiceID string `json:"service_id"`
}
//InitServerType init server type
func (c *CodeSourceInfo) InitServerType() {
if strings.HasPrefix(c.RepositoryURL, "svn://") || strings.HasSuffix(c.RepositoryURL, ".svn") {
c.ServerType = "svn"
} else {
c.ServerType = "git"
}
}
//GetCodeSourceDir get source storage directory
func (c CodeSourceInfo) GetCodeSourceDir() string {
return GetCodeSourceDir(c.RepositoryURL, c.Branch, c.TenantID, c.ServiceID)

4
builder/sources/image.go
View File

@ -30,8 +30,6 @@ import (
"strings"
"time"
"github.com/ncw/directio"
"github.com/Sirupsen/logrus"
"github.com/docker/distribution/reference"
@ -494,7 +492,7 @@ func ImageImport(dockerCli *client.Client, image, source string, logger event.Lo
func CopyToFile(outfile string, r io.Reader) error {
// We use sequential file access here to avoid depleting the standby list
// on Windows. On Linux, this is a call directly to ioutil.TempFile
tmpFile, err := directio.OpenFile(path.Join(filepath.Dir(outfile), ".docker_temp_"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0644)
tmpFile, err := os.OpenFile(path.Join(filepath.Dir(outfile), ".docker_temp_"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0644)
if err != nil {
return err
}

1
db/db.go
View File

@ -98,6 +98,7 @@ type Manager interface {
RegionAPIClassDaoTransactions(db *gorm.DB) dao.RegionAPIClassDao
RegionProcotolsDao() dao.RegionProcotolsDao
RegionProcotolsDaoTransactions(db *gorm.DB) dao.RegionProcotolsDao
NotificationEventDao() dao.NotificationEventDao
AppBackupDao() dao.AppBackupDao

1
db/model/tenant.go
View File

@ -51,6 +51,7 @@ type Tenants struct {
Name string `gorm:"column:name;size:40;unique_index"`
UUID string `gorm:"column:uuid;size:33;unique_index"`
EID string `gorm:"column:eid"`
LimitMemory int `gorm:"column:limit_memory"`
}
//TableName 返回租户表名称

66
db/mysql/mysql.go
View File

@ -151,72 +151,6 @@ func (m *Manager) CheckTable() {
}
func (m *Manager) patchTable() {
// Permissions set
var rac model.RegionAPIClass
if err := m.db.Where("class_level=? and prefix=?", "server_source", "/v2/show").Find(&rac).Error; err != nil {
if err == gorm.ErrRecordNotFound {
data := map[string]string{
"/v2/show": "server_source",
"/v2/opentsdb": "server_source",
"/v2/resources": "server_source",
"/v2/builder": "server_source",
"/v2/tenants": "server_source",
"/v2/app": "server_source",
"/api/v1": "server_source",
"/v2/nodes": "node_manager",
"/v2/job": "node_manager",
"/v2/tasks": "node_manager",
"/v2/taskgroups": "node_manager",
"/v2/tasktemps": "node_manager",
"/v2/configs": "node_manager",
}
tx := m.Begin()
var rollback bool
for k, v := range data {
if err := m.RegionAPIClassDaoTransactions(tx).AddModel(&model.RegionAPIClass{
ClassLevel: v,
Prefix: k,
}); err != nil {
tx.Rollback()
rollback = true
break
}
}
if !rollback {
tx.Commit()
}
}
}
//Port Protocol support
var rps model.RegionProcotols
if err := m.db.Where("protocol_group=? and protocol_child=?", "http", "http").Find(&rps).Error; err != nil {
if err == gorm.ErrRecordNotFound {
data := map[string][]string{
"http": []string{"http"},
"stream": []string{"mysql", "tcp", "udp"},
}
tx := m.Begin()
var rollback bool
for k, v := range data {
for _, v1 := range v {
if err := m.RegionProcotolsDaoTransactions(tx).AddModel(&model.RegionProcotols{
ProtocolGroup: k,
ProtocolChild: v1,
APIVersion: "v2",
IsSupport: true,
}); err != nil {
tx.Rollback()
rollback = true
break
}
}
}
if !rollback {
tx.Commit()
}
}
}
//set plugin version image name length
if err := m.db.Exec("alter table tenant_plugin_build_version modify column base_image varchar(200);").Error; err != nil {
logrus.Errorf("alter table tenant_plugin_build_version error %s", err.Error())

21
entrance/core/manager.go
View File

@ -185,6 +185,27 @@ func (m *manager) add(source object.Object) {
}
func (m *manager) delete(source object.Object) {
switch source.(type) {
case *object.PoolObject:
pool := source.(*object.PoolObject)
plugin, err := m.getPlugin(pool.PluginName, pool.PluginOpts)
if err != nil {
logrus.Errorf("get default plugin error.%s pool don't add to lb", err.Error())
return
}
if plugin.GetName() == "zeus" {
//zeus do not handle pool delete
return
}
nodes, err := m.storeManager.GetNodeByPool(pool.GetName())
if err != nil {
logrus.Errorf("get node list by pool name(%s) error.%s,do not delete pool", pool.GetName(), err.Error())
return
}
if len(nodes) != 0 {
return
}
}
ok, err := m.storeManager.DeleteSource(source)
if err != nil {
logrus.Errorf("Update %s to store error.%s", source.GetName(), err.Error())

35
entrance/core/object/object.go
View File

@ -18,11 +18,19 @@
package object
import (
"fmt"
"strings"
"github.com/goodrain/rainbond/util"
)
//Object source object
type Object interface {
GetName() string
GetIndex() int64
GetEventID() string
GetHash() (string, error)
}
type PoolObject struct {
@ -31,7 +39,6 @@ type PoolObject struct {
Index int64
Name string
Note string //说明
NodeNumber int
PluginName string
PluginOpts map[string]string
Namespace string
@ -47,6 +54,9 @@ func (p *PoolObject) GetIndex() int64 {
func (p *PoolObject) GetEventID() string {
return p.EventID
}
func (p *PoolObject) GetHash() (string, error) {
return util.CreateHashString(p.ServiceVersion + p.ServiceID + p.Name + p.Note + p.PluginName + p.Namespace)
}
type NodeObject struct {
Index int64
@ -73,6 +83,9 @@ func (p *NodeObject) GetIndex() int64 {
func (p *NodeObject) GetEventID() string {
return p.EventID
}
func (p *NodeObject) GetHash() (string, error) {
return util.CreateHashString(p.Host + fmt.Sprintf("%d%t%d", p.Port, p.Ready, p.Weight) + p.Protocol + p.State + p.PoolName + p.NodeName + p.Namespace + p.PluginName)
}
type RuleObject struct {
Name string //不重复的命名规则
@ -94,9 +107,13 @@ func (p *RuleObject) GetName() string {
func (p *RuleObject) GetIndex() int64 {
return p.Index
}
//GetEventID get eventid
func (p *RuleObject) GetEventID() string {
return p.EventID
}
//Copy copy
func (p *RuleObject) Copy() *RuleObject {
var r RuleObject
r.CertificateName = p.CertificateName
@ -111,6 +128,11 @@ func (p *RuleObject) Copy() *RuleObject {
return &r
}
//GetHash get hash
func (p *RuleObject) GetHash() (string, error) {
return util.CreateHashString(fmt.Sprintf("%t%t", p.HTTPS, p.TransferHTTP) + p.Name + p.DomainName + p.PoolName + p.CertificateName + p.Namespace + p.PluginName)
}
type Certificate struct {
Name string
Index int64
@ -133,6 +155,10 @@ func (p *Certificate) GetEventID() string {
return p.EventID
}
func (p *Certificate) GetHash() (string, error) {
return util.CreateHashString(p.Name + p.Certificate + p.PrivateKey + p.PluginName)
}
type DomainObject struct {
Name string //不重复的命名 不同资源之间可以一样
Domain string
@ -153,6 +179,10 @@ func (p *DomainObject) GetEventID() string {
return p.EventID
}
func (p *DomainObject) GetHash() (string, error) {
return util.CreateHashString(p.Name + p.Domain + p.Protocol + p.PluginName)
}
type VirtualServiceObject struct {
Index int64
Name string //不重复的命名
@ -178,3 +208,6 @@ func (p *VirtualServiceObject) GetIndex() int64 {
func (p *VirtualServiceObject) GetEventID() string {
return p.EventID
}
func (p *VirtualServiceObject) GetHash() (string, error) {
return util.CreateHashString(p.Name + p.Protocol + p.PluginName + p.Note + p.DefaultPoolName + strings.Join(p.Rules, "") + fmt.Sprintf("%t%d", p.Enabled, p.Port) + strings.Join(p.Listening, "") + p.PluginName + p.Namespace)
}

42
entrance/plugin/openresty/openresty_plugin.go
View File

@ -62,7 +62,7 @@ type openresty struct {
var defaultNodeList = []NginxNode{
{
"Active",
"127.0.0.1:404",
"127.0.0.1:10004",
1,
}}
@ -204,14 +204,12 @@ func (o *openresty) AddPool(originalPools ...*object.PoolObject) error {
// nginx默认会试图将所有请求根据请求头中的host字段转发到名字与该host字段值相同的upstream
func (o *openresty) UpdatePool(pools ...*object.PoolObject) error {
var errs []error
for _, originalPool := range pools {
upstreamName, err := getUpstreamNameByPool(originalPool.Name)
if err != nil {
logrus.Error(fmt.Sprintf("Failed to update pool %s: %s", originalPool.Name, err))
continue
}
// get nodes from store, for example etcd.
originalNodes, err := o.ctx.Store.GetNodeByPool(originalPool.Name)
if err != nil {
@ -219,19 +217,11 @@ func (o *openresty) UpdatePool(pools ...*object.PoolObject) error {
errs = append(errs, err)
continue
}
if len(originalNodes) < 1 {
logrus.Info("Delete the pool, because no servers are inside the pool ", originalPool.Name)
o.deleteUpstream(originalPool.Name)
continue
}
protocol := "tcp"
_, err = o.ctx.Store.GetVSByPoolName(originalPool.Name)
if err != nil {
protocol = "http"
}
// build pool for openresty by original nodes
pool := NginxUpstream{upstreamName, []NginxNode{}, protocol}
for _, originalNode := range originalNodes {
@ -239,13 +229,11 @@ func (o *openresty) UpdatePool(pools ...*object.PoolObject) error {
if state == "" {
state = "Active"
}
addr := fmt.Sprintf("%s:%d", originalNode.Host, originalNode.Port)
if len(originalNode.Host) < 7 || originalNode.Port < 1 {
logrus.Info(fmt.Sprintf("Ignore node in pool %s, illegal address [%s]", pool.Name, addr))
continue
}
weight := originalNode.Weight
if weight < 1 {
weight = 1
@ -259,13 +247,11 @@ func (o *openresty) UpdatePool(pools ...*object.PoolObject) error {
}
if len(pool.Servers) < 1 {
logrus.Info("Ignore update the pool, because no servers are inside the pool ", pool.Name)
continue
logrus.Warningf("nginx upstream %s not have server,will use default server", pool.Name)
pool.Servers = defaultNodeList
}
// push data to all openresty instance by rest api
err = o.doEach(UPDATE, o.urlPool(pool.Name), pool)
if err != nil {
errs = append(errs, err)
continue
@ -278,16 +264,14 @@ func (o *openresty) UpdatePool(pools ...*object.PoolObject) error {
func (o *openresty) DeletePool(pools ...*object.PoolObject) error {
var errs []error
for _, pool := range pools {
err := o.deleteUpstream(pool.Name)
if err != nil {
errs = append(errs, err)
logrus.Error(err)
logrus.Errorf("delete nginx upstream %s error %s", pool.Name, err.Error())
continue
}
}
return reduceErr(errs)
}
@ -302,17 +286,14 @@ func (o *openresty) AddNode(nodes ...*object.NodeObject) error {
// 将node根据所属pool分类，根据每个pool名字取出该pool下所有node，然后全量更新
func (o *openresty) UpdateNode(nodes ...*object.NodeObject) error {
poolNames := make(map[string]string, 0)
for _, node := range nodes {
poolNames[node.PoolName] = node.NodeName
}
pools, err := o.ctx.Store.GetPools(poolNames)
if err != nil {
logrus.Error(err)
return err
}
return o.UpdatePool(pools...)
}
@ -330,17 +311,14 @@ func (o *openresty) deleteUpstream(poolName string) error {
logrus.Error(fmt.Sprintf("Failed to get upstream name %s: %s", poolName, err))
return err
}
protocol := "tcp"
_, err = o.ctx.Store.GetVSByPoolName(poolName)
if err != nil {
protocol = "http"
}
if err := o.doEach(DELETE, o.urlPool(upstreamName), Options{protocol}); err != nil {
return err
}
return nil
}
@ -450,7 +428,6 @@ func (o *openresty) UpdateRule(rules ...*object.RuleObject) error {
// get cert key pair if https
if protocol == "https" {
port = o.defaultHttpsPort
pair, err := o.ctx.Store.GetCertificate(rule.CertificateName)
if err != nil {
logrus.Error("Failed to updata the rule: ", err.Error())
@ -494,9 +471,7 @@ func (o *openresty) DeleteRule(rules ...*object.RuleObject) error {
if rule.HTTPS {
protocol = "https"
}
err := o.doEach(DELETE, o.urlServer(rule.Name), Options{protocol})
if err != nil {
errs = append(errs, err)
logrus.Error(err)
@ -536,7 +511,10 @@ func (o *openresty) UpdateVirtualService(services ...*object.VirtualServiceObjec
if service.Protocol == "" {
service.Protocol = "tcp"
}
if service.Port == 0 {
logrus.Errorf("nginx server port can not be 0")
continue
}
openrestyRule := NginxServer{
Name: service.Name,
Port: service.Port,
@ -562,20 +540,16 @@ func (o *openresty) UpdateVirtualService(services ...*object.VirtualServiceObjec
func (o *openresty) DeleteVirtualService(services ...*object.VirtualServiceObject) error {
var errs []error
for _, service := range services {
if service.Protocol == "" {
service.Protocol = "tcp"
}
err := o.doEach(DELETE, o.urlServer(service.Name), Options{service.Protocol})
if err != nil {
errs = append(errs, err)
logrus.Error(err)
continue
}
}
return reduceErr(errs)
}

1
entrance/plugin/zeus/zeus_plugin.go
View File

@ -390,7 +390,6 @@ func (z *zeus) DeletePool(pools ...*object.PoolObject) error {
if err != nil {
errs = append(errs, err)
}
}
return handleErr(errs)
}

7
entrance/source/config/config.go
View File

@ -25,8 +25,6 @@ import (
"github.com/goodrain/rainbond/entrance/core"
"github.com/Sirupsen/logrus"
"k8s.io/client-go/pkg/api/v1"
)
@ -84,7 +82,6 @@ const (
)
func (s *SourceBranch) RePoolName() string {
logrus.Debugf("tenant name is %s, in RePoolName", s.Tenant)
return fmt.Sprintf("%s@%s_%d.Pool",
s.Tenant,
s.Service,
@ -93,7 +90,6 @@ func (s *SourceBranch) RePoolName() string {
}
func (s *SourceBranch) ReVSName() string {
logrus.Debugf("tenant name is %s, in ReVSName", s.Tenant)
return fmt.Sprintf("%s_%s_%d.VS",
s.Tenant,
s.Service,
@ -102,7 +98,6 @@ func (s *SourceBranch) ReVSName() string {
}
func (s *SourceBranch) ReNodeName() string {
logrus.Debugf("tenant name is %s, in ReNodeName", s.Tenant)
return fmt.Sprintf("%s_%d.Node",
s.PodName,
s.ContainerPort,
@ -117,7 +112,6 @@ func sha8(s string) string {
}
func (s *SourceBranch) ReRuleName(domain string) string {
logrus.Debugf("tenant name is %s, in ReRuleName", s.Tenant)
return fmt.Sprintf("%s_%s_%d_%s.Rule",
s.Tenant,
s.Service,
@ -127,7 +121,6 @@ func (s *SourceBranch) ReRuleName(domain string) string {
}
func (s *SourceBranch) ReServiceId() string {
logrus.Debugf("tenant name is %s, in ReServiceId", s.Tenant)
return fmt.Sprintf("%s.%s_%d",
s.Tenant,
s.Service,

66
entrance/source/handle.go
View File

@ -69,7 +69,7 @@ func (m *Manager) addPodSource(s *config.SourceBranch) {
} else {
s.NodePort = s.ContainerPort
}
// event pool
// event pool first
m.RcPool(s)
// event node
m.RcNode(s)
@ -101,10 +101,17 @@ func (m *Manager) deletePodSource(s *config.SourceBranch) {
if !s.IsMidonet {
s.NodePort = s.ContainerPort
}
// event node first
m.RcNode(s)
// event pool
m.RcPool(s)
}
func (m *Manager) podSource(pods *v1.Pod, method core.EventMethod) {
//if pod do not have ip and method is update,ignore it
if pods.Status.PodIP == "" && method == core.UPDATEEventMethod {
return
}
index, _ := strconv.ParseInt(pods.ResourceVersion, 10, 64)
var flagHost bool
for _, envs := range pods.Spec.Containers[0].Env {
@ -116,19 +123,32 @@ func (m *Manager) podSource(pods *v1.Pod, method core.EventMethod) {
continue
}
}
ppInfo := pods.Labels["protocols"]
if ppInfo == "" {
ppInfo = "1234_._ptth"
}
logrus.Debugf("pod port protocols %v", ppInfo)
mapPP := make(map[string]string)
infoList := strings.Split(ppInfo, "-.-")
if len(infoList) > 0 {
for _, pps := range infoList {
portInfo := strings.Split(pps, "_._")
ppInfo, ok := pods.Labels["protocols"]
if ok {
if ppInfo == "" {
ppInfo = "1234_._ptth"
}
infoList := strings.Split(ppInfo, "-.-")
if len(infoList) > 0 {
for _, pps := range infoList {
portInfo := strings.Split(pps, "_._")
mapPP[portInfo[0]] = portInfo[1]
}
}
} else {
protocolsNumber := pods.Labels["protocols_number"]
number, err := strconv.Atoi(protocolsNumber);
if err != nil {
logrus.Errorf("ports number converted to int failed：", err.Error())
}
for i := 0; i < number; i++ {
protocol := pods.Labels[fmt.Sprintf("%s_%d", "protocol", i)]
portInfo := strings.Split(protocol,"_._")
mapPP[portInfo[0]] = portInfo[1]
}
}
//protocols: 5000_._http-.-8080_._stream
s := &config.SourceBranch{
Tenant: pods.Labels["tenant_name"],
@ -140,7 +160,6 @@ func (m *Manager) podSource(pods *v1.Pod, method core.EventMethod) {
Version: pods.Labels["version"],
Namespace: pods.Namespace,
}
logrus.Debugf("In podSource Tenant name is %s", s.Tenant)
for _, statusInfo := range pods.Status.Conditions {
if statusInfo.Type == "Ready" && statusInfo.Status == "True" {
s.PodStatus = true
@ -148,10 +167,8 @@ func (m *Manager) podSource(pods *v1.Pod, method core.EventMethod) {
}
if flagHost {
s.Host = pods.Status.HostIP
logrus.Debug("Net model is midonet.")
} else {
s.Host = pods.Status.PodIP
logrus.Debug("Net model is calico.")
}
for _, containersInfo := range pods.Spec.Containers {
@ -163,7 +180,6 @@ func (m *Manager) podSource(pods *v1.Pod, method core.EventMethod) {
s.ContainerPort = portInfo.ContainerPort
s.Port = portInfo.ContainerPort
s.Note = mapPP[fmt.Sprintf("%d", s.Port)]
logrus.Debugf("note for poolname %s_%d is %s", s.Tenant, s.Port, s.Note)
switch method {
case core.ADDEventMethod:
m.addPodSource(s)
@ -188,16 +204,15 @@ func (m *Manager) RcPool(s *config.SourceBranch) {
Name: s.RePoolName(),
EventID: s.EventID,
}
logrus.Debugf("Pool %s note is %s", poolobj.Name, poolobj.Note)
etPool := core.Event{
Method: s.Method,
Source: poolobj,
}
logrus.Debugf("%s a pool source %s", s.Method, poolobj.GetName())
m.CoreManager.EventChan() <- etPool
}
func (m *Manager) RcNode(s *config.SourceBranch) {
logrus.Debugf("%s readyCode is %v", s.ReNodeName(), s.PodStatus)
nodeobj := &object.NodeObject{
Namespace: s.Namespace,
Index: s.Index,
@ -210,11 +225,11 @@ func (m *Manager) RcNode(s *config.SourceBranch) {
Ready: s.PodStatus,
EventID: s.EventID,
}
logrus.Debugf("%s Node : %v ", s.Method, nodeobj)
etNode := core.Event{
Method: s.Method,
Source: nodeobj,
}
logrus.Debugf("%s a node source %s", s.Method, nodeobj.GetName())
m.CoreManager.EventChan() <- etNode
}
@ -233,13 +248,11 @@ func (m *Manager) RcVS(s *config.SourceBranch) {
DefaultPoolName: s.RePoolName(),
EventID: s.EventID,
}
logrus.Debugf("in RcVS tenant name is %s", s.Tenant)
et := core.Event{
Method: s.Method,
Source: vsobj,
}
logrus.Debug("vsName is ", s.ReVSName())
logrus.Debug(s.Method, " vs source ", vsobj)
logrus.Debugf("%s a vs source %s", s.Method, vsobj.Name)
m.CoreManager.EventChan() <- et
}
@ -262,9 +275,7 @@ type ResponseType struct {
func (m *Manager) getDomainInfo(s *config.SourceBranch) ([]model.Domain, error) {
domainURL := fmt.Sprintf(config.DomainAPIURI, m.LBAPIPort, s.Tenant, s.Service)
logrus.Debugf("In getDomainInfo tenant name is %s", s.Tenant)
var ldomain []model.Domain
logrus.Debug("get domain infos from lb url: ", domainURL)
client := &http.Client{}
request, err := http.NewRequest("GET", domainURL, nil)
if err != nil {
@ -352,13 +363,11 @@ func (m *Manager) RcRule(s *config.SourceBranch) {
CertificateName: s.CertificateName,
EventID: s.EventID,
}
logrus.Debugf("In RcRule tenant name is ", s.Tenant)
et := core.Event{
Method: s.Method,
Source: ruleobj,
}
logrus.Debug("ruleName is ", s.ReRuleName(domain))
logrus.Debug(s.Method, " rule source ", ruleobj)
logrus.Debugf("%s a rule source %s", s.Method, ruleobj.GetName())
m.CoreManager.EventChan() <- et
}
}
@ -392,14 +401,12 @@ func (m *Manager) RcDomain(s *config.SourceBranch) {
Method: s.Method,
Source: domainobj,
}
logrus.Debug("domainName is ", domain)
logrus.Debug(s.Method, " domain source ", domainobj)
m.CoreManager.EventChan() <- etDomain
}
//处理扩展域名
domainList, err := m.getDomainInfo(s)
if err != nil {
logrus.Debug("get domainlist err is ", err)
logrus.Debugf("get domainlist err is %s", err)
}
if domainList != nil && len(domainList) > 0 {
for _, domain := range domainList {
@ -448,7 +455,6 @@ func (m *Manager) RcDomain(s *config.SourceBranch) {
//TODO:
// domain支持多个，即 domain字段可能传入多个域名
func (m *Manager) serviceSource(services *v1.Service, method core.EventMethod) {
logrus.Debugf("In serviceSource service_type is %s", services.Labels["service_type"])
index, _ := strconv.ParseInt(services.ResourceVersion, 10, 64)
s := &config.SourceBranch{
Tenant: services.Labels["tenant_name"],
@ -461,11 +467,9 @@ func (m *Manager) serviceSource(services *v1.Service, method core.EventMethod) {
Method: method,
OriginPort: services.Labels["origin_port"],
}
logrus.Debug("poolName is ", s.RePoolName())
// event domain
s.Domain = m.replaceDomain(s.Domain, s)
m.RcDomain(s)
logrus.Debugf("Fprotocol is %s", services.Labels["protocol"])
//TODO: "stream" to !http
if services.Labels["protocol"] != "http" && services.Labels["protocol"] != "https" {
// event vs

45
entrance/store/manager.go
View File

@ -431,28 +431,33 @@ func (m *Manager) update(key string, source object.Object, info *SourceInfo) (bo
return false, false, err
}
if info.Index < source.GetIndex() {
i.IsOnline = info.IsOnline
data, err := json.Marshal(i)
if err != nil {
return false, false, err
}
_, err = m.keysAPI.Update(m.ctx, key, string(data))
if err != nil {
return info.IsOnline, false, err
}
//判断是否是NODE资源，如果是当host:port没有变化且ready无变化时，操作权返回false
switch info.Data.(type) {
case *object.NodeObject:
node := info.Data.(*object.NodeObject)
newNode := source.(*object.NodeObject)
oldKey := fmt.Sprintf("%s:%d %v", node.Host, node.Port, node.Ready)
newKey := fmt.Sprintf("%s:%d %v", newNode.Host, newNode.Port, newNode.Ready)
logrus.Debugf("oldkey is %s, newkey is %s", oldKey, newKey)
if oldKey == newKey {
return info.IsOnline, false, nil
//add hash inspection,Remove duplicate update operations
infohash, errI := info.Data.GetHash()
sourcehash, errS := source.GetHash()
if infohash != sourcehash || errI != nil || errS != nil {
i.IsOnline = info.IsOnline
data, err := json.Marshal(i)
if err != nil {
return false, false, err
}
_, err = m.keysAPI.Update(m.ctx, key, string(data))
if err != nil {
return info.IsOnline, false, err
}
//判断是否是NODE资源，如果是当host:port没有变化且ready无变化时，操作权返回false
switch info.Data.(type) {
case *object.NodeObject:
node := info.Data.(*object.NodeObject)
newNode := source.(*object.NodeObject)
oldKey := fmt.Sprintf("%s:%d %v", node.Host, node.Port, node.Ready)
newKey := fmt.Sprintf("%s:%d %v", newNode.Host, newNode.Port, newNode.Ready)
logrus.Debugf("oldkey is %s, newkey is %s", oldKey, newKey)
if oldKey == newKey {
return info.IsOnline, false, nil
}
}
return info.IsOnline, true, nil
}
return info.IsOnline, true, nil
}
return info.IsOnline, false, nil
}

2
grctl/cmd/init.go
View File

@ -44,7 +44,7 @@ func NewCmdInit() cli.Command {
cli.StringFlag{
Name: "role",
Usage: "node role: master,worker",
Value: "master",
Value: "master,worker",
},
cli.StringFlag{
Name: "work_dir",

2
grctl/cmd/service.go
View File

@ -198,7 +198,7 @@ func getEventLog(c *cli.Context) error {
}
} else {
logdb := &eventdb.EventFilePlugin{
HomePath: "/grdata/logs/",
HomePath: "/grdata/downloads/log/",
}
list, err := logdb.GetMessages(eventID, "debug")
if err != nil {

1
node/nodem/client/node.go
View File

@ -153,6 +153,7 @@ type NodeSystemInfo struct {
Architecture string `json:"architecture"`
MemorySize uint64 `json:"memorySize"`
NumCPU int64 `json:"cpu_num"`
}
//Decode decode node info

3
node/nodem/info/info_linux.go
View File

@ -45,6 +45,7 @@ func GetSystemInfo() (info client.NodeSystemInfo) {
}
info.OperatingSystem = runtime.GOOS
info.MemorySize, _ = getMemory()
info.NumCPU = int64(runtime.NumCPU())
return info
}
@ -72,7 +73,7 @@ func getMemory() (total uint64, free uint64) {
sysInfo := new(syscall.Sysinfo_t)
err := syscall.Sysinfo(sysInfo)
if err == nil {
return sysInfo.Totalram * uint64(syscall.Getpagesize()), sysInfo.Freeram * uint64(syscall.Getpagesize())
return uint64(sysInfo.Totalram) * uint64(sysInfo.Unit), sysInfo.Freeram * uint64(syscall.Getpagesize())
}
return 0, 0
}

2
node/nodem/info/info_test.go
View File

@ -21,5 +21,5 @@ package info
import "testing"
func TestGetSysteminfo(t *testing.T) {
t.Log(getSystemInfo())
t.Log(GetSystemInfo())
}

14
util/comman.go
View File

@ -36,8 +36,6 @@ import (
"syscall"
"time"
"github.com/ncw/directio"
"github.com/Sirupsen/logrus"
"github.com/goodrain/rainbond/util/zip"
)
@ -443,7 +441,7 @@ func Zip(source, target string) error {
if err := CheckAndCreateDir(filepath.Dir(target)); err != nil {
return err
}
zipfile, err := directio.OpenFile(target, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0666)
zipfile, err := os.OpenFile(target, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0666)
if err != nil {
return err
}
@ -491,7 +489,7 @@ func Zip(source, target string) error {
if info.IsDir() {
return nil
}
file, err := directio.OpenFile(path, os.O_RDONLY, 0)
file, err := os.OpenFile(path, os.O_RDONLY, 0)
if err != nil {
return err
}
@ -535,7 +533,7 @@ func Unzip(archive, target string) error {
return err
}
defer fileReader.Close()
targetFile, err := directio.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, file.Mode())
targetFile, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, file.Mode())
if err != nil {
return err
}
@ -575,12 +573,12 @@ func CopyFile(source, target string) error {
elem := reflect.ValueOf(sfi.Sys()).Elem()
uid := elem.FieldByName("Uid").Uint()
gid := elem.FieldByName("Gid").Uint()
sf, err := directio.OpenFile(source, os.O_RDONLY, 0)
sf, err := os.OpenFile(source, os.O_RDONLY, 0)
if err != nil {
return err
}
defer sf.Close()
tf, err := directio.OpenFile(target, os.O_RDONLY|os.O_CREATE|os.O_WRONLY, sfi.Mode())
tf, err := os.OpenFile(target, os.O_CREATE|os.O_WRONLY, sfi.Mode())
if err != nil {
return err
}
@ -589,9 +587,11 @@ func CopyFile(source, target string) error {
if err != nil {
return err
}
if err := os.Chown(target, int(uid), int(gid)); err != nil {
return err
}
return nil
}

10
util/hash.go
View File

@ -88,3 +88,13 @@ func createSmallFileHash(sourceFile, hashfile *os.File) error {
}
return nil
}
//CreateHashString create hash string
func CreateHashString(source string) (hashstr string, err error) {
md5h := md5.New()
_, err = md5h.Write([]byte(source))
if err != nil {
return "", err
}
return fmt.Sprintf("%x", md5h.Sum(nil)), nil
}

4
util/zip/reader.go
View File

@ -28,8 +28,6 @@ import (
"io"
"os"
"time"
"github.com/ncw/directio"
)
var (
@ -84,7 +82,7 @@ func OpenReader(name string) (*ReadCloser, error) {
// OpenDirectReader will open the Zip file specified by name and return a ReadCloser.
func OpenDirectReader(name string) (*ReadCloser, error) {
f, err := directio.OpenFile(name, os.O_RDONLY, 0)
f, err := os.OpenFile(name, os.O_RDONLY, 0)
if err != nil {
return nil, err
}

2
webcli/app/app.go
View File

@ -209,7 +209,7 @@ func (app *App) handleWS(w http.ResponseWriter, r *http.Request) {
ptyIo, err := pty.Start(cmd)
if err != nil {
logrus.Printf("Failed to execute command:%s", err.Error())
ExecuteCommandTotal++
ExecuteCommandFailed++
return
}
logrus.Printf("Command is running for client %s with PID %d ", r.RemoteAddr, cmd.Process.Pid)

11
worker/appm/pod.go
View File

@ -206,14 +206,11 @@ func (p *PodTemplateSpecBuild) Build(creatorID string) (*v1.PodTemplateSpec, err
outPorts, err = p.CreateUpstreamPluginMappingPort(outPorts, pluginPorts)
}
labels["service_type"] = "outer"
var pStr string
for _, p := range outPorts {
if pStr != "" {
pStr += "-.-"
}
pStr += fmt.Sprintf("%d_._%s", p.ContainerPort, p.Protocol)
labels["protocols_number"] = fmt.Sprintf("%d", len(outPorts))
for i, p := range outPorts {
key := fmt.Sprintf("%s_%d", "protocol", i)
labels[key] = fmt.Sprintf("%d_._%s", p.ContainerPort, p.Protocol)
}
labels["protocols"] = pStr
}
//step7: set hostname
if p.hostName != "" {

23
worker/executor/executor_manager.go
View File

@ -28,6 +28,7 @@ import (
"github.com/goodrain/rainbond/worker/executor/task"
"github.com/Sirupsen/logrus"
"time"
)
//Manager Manager
@ -49,7 +50,6 @@ type manager struct {
workerLock sync.RWMutex
taskManager *task.TaskManager
statusManager *status.AppRuntimeSyncClient
wg sync.WaitGroup
}
//NewManager newManager
@ -68,11 +68,27 @@ func (m *manager) Start() {
func (m *manager) Stop() {
m.workerLock.Lock()
defer m.workerLock.Unlock()
i := 0
for _, w := range m.workers {
w.Cancel()
}
timer := time.NewTimer(time.Second * 2)
defer timer.Stop()
for {
if i >= 15{
logrus.Errorf("There are %d tasks not completed", len(m.workers))
return
}
if len(m.workers) == 0{
break
}
select {
case <-timer.C:
i ++
timer.Reset(time.Second * 2)
}
}
logrus.Info("Waiting for all threads to complete.")
m.wg.Wait()
logrus.Info("All threads is exited.")
}
@ -96,8 +112,7 @@ func (m *manager) AddTask(t task.Task) error {
return fmt.Errorf("worker %s:%s is exist ", t.TaskID(), t.Logger().Event())
}
worker := newWorker(m, t)
m.wg.Add(1)
go worker.Start(m.wg)
go worker.Start()
m.workers[workerKey{t.TaskID(), t.Logger().Event()}] = worker
return nil
}

1
worker/executor/task/horizontal_scaling_task.go
View File

@ -41,6 +41,7 @@ type horizontalScalingTask struct {
func (s *horizontalScalingTask) RunSuccess() {
//设置应用状态为运行中
s.taskManager.statusManager.SetStatus(s.modelTask.ServiceID, "running")
s.taskManager.statusManager.CheckStatus(s.modelTask.ServiceID)
s.logger.Info("应用水平伸缩任务完成", map[string]string{"step": "last", "status": "success"})
}

3
worker/executor/task/restart_task.go
View File

@ -42,6 +42,7 @@ type restartTask struct {
func (s *restartTask) RunSuccess() {
//设置应用状态为运行中
s.taskManager.statusManager.SetStatus(s.modelTask.ServiceID, "running")
s.taskManager.statusManager.CheckStatus(s.modelTask.ServiceID)
s.logger.Info("应用重新启动任务完成", map[string]string{"step": "last", "status": "success"})
}
@ -52,6 +53,7 @@ func (s *restartTask) RunError(e error) {
//应用启动超时，怎么处理？
//是否关闭应用？
//暂时不自动关闭
s.taskManager.statusManager.CheckStatus(s.modelTask.ServiceID)
s.logger.Error("应用重启超时，请稍等并注意应用状态", map[string]string{"step": "callback", "status": "timeout"})
return
}
@ -93,6 +95,7 @@ func (s *restartTask) RunError(e error) {
}
//设置应用状态为已关闭
s.taskManager.statusManager.SetStatus(s.modelTask.ServiceID, "closed")
s.taskManager.statusManager.CheckStatus(s.modelTask.ServiceID)
s.logger.Error("重启失败，请重试", map[string]string{"step": "callback", "status": "failure"})
}

1
worker/executor/task/rolling_upgrade_task.go
View File

@ -47,6 +47,7 @@ type rollingUpgradeTask struct {
func (s *rollingUpgradeTask) RunSuccess() {
//设置应用状态为运行中
s.taskManager.statusManager.SetStatus(s.modelTask.ServiceID, status.RUNNING)
s.taskManager.statusManager.CheckStatus(s.modelTask.ServiceID)
s.logger.Info("应用滚动升级任务完成", map[string]string{"step": "last", "status": "success"})
}

2
worker/executor/task/start_task.go
View File

@ -45,6 +45,7 @@ type startTask struct {
func (s *startTask) RunSuccess() {
//设置应用状态为运行中
s.taskManager.statusManager.SetStatus(s.modelTask.ServiceID, "running")
s.taskManager.statusManager.CheckStatus(s.modelTask.ServiceID)
s.logger.Info("应用启动任务完成", map[string]string{"step": "last", "status": "success"})
}
@ -101,6 +102,7 @@ func (s *startTask) RunError(e error) {
}
//设置应用状态为已关闭
s.taskManager.statusManager.SetStatus(s.modelTask.ServiceID, "closed")
s.taskManager.statusManager.CheckStatus(s.modelTask.ServiceID)
s.logger.Error("启动错误，请重试", map[string]string{"step": "callback", "status": "failure"})
}

5
worker/executor/task/stop_task.go
View File

@ -41,6 +41,7 @@ type stopTask struct {
func (s *stopTask) RunSuccess() {
//设置应用状态为已关闭
s.taskManager.statusManager.SetStatus(s.modelTask.ServiceID, status.CLOSED)
s.taskManager.statusManager.CheckStatus(s.modelTask.ServiceID)
s.logger.Info("应用关闭任务完成", map[string]string{"step": "last", "status": "success"})
}
@ -50,8 +51,8 @@ func (s *stopTask) RunError(e error) {
//TODO:
//应用关闭超时，怎么处理？
//超时说明部署信息已删除，设置应用状态为已关闭
s.taskManager.statusManager.SetStatus(s.modelTask.ServiceID, status.CLOSED)
return
// s.taskManager.statusManager.SetStatus(s.modelTask.ServiceID, status.CLOSED)
// return
}
// if e == appm.ErrNotDeploy {
// return

9
worker/executor/worker.go
View File

@ -22,12 +22,11 @@ import (
"github.com/goodrain/rainbond/worker/executor/task"
"fmt"
"runtime/debug"
"sync"
)
//Worker 工作器
type Worker interface {
Start(wg sync.WaitGroup)
Start()
Cancel() error
RollBack() error
Status() string
@ -49,8 +48,10 @@ func newWorker(manager *manager, task task.Task) Worker {
return w
}
func (w *woker) Start(wg sync.WaitGroup) {
defer wg.Done()
func (w *woker) Start() {
defer func() {
w.manager.removeWorker(w.task.TaskID(), w.task.Logger().Event())
}()
defer func() {
if err := recover(); err != nil {
w.task.RunError(fmt.Errorf("worker recover %v", err))