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

merge 5.1 branch code

Browse Source
This commit is contained in:
barnett 2019-06-25 22:01:03 +08:00
commit 6b0713ccc4
40 changed files with 549 additions and 186 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
api/controller/resources.go
View File

@ -600,7 +600,6 @@ func (t *TenantStruct) CreateService(w http.ResponseWriter, r *http.Request) {
tenantID := r.Context().Value(middleware.ContextKey("tenant_id")).(string)
ss.TenantID = tenantID
logrus.Debugf("begin to create service %s", ss.ServiceAlias)
if err := handler.GetServiceManager().ServiceCreate(&ss); err != nil {
httputil.ReturnError(r, w, 500, fmt.Sprintf("create service error, %v", err))
return

34
api/controller/third_party_service.go
View File

@ -20,13 +20,15 @@ package controller
import (
"encoding/json"
"github.com/Sirupsen/logrus"
"net/http"
"strings"
"github.com/Sirupsen/logrus"
"github.com/goodrain/rainbond/api/controller/validation"
"github.com/goodrain/rainbond/api/handler"
"github.com/goodrain/rainbond/api/middleware"
"github.com/goodrain/rainbond/api/model"
"github.com/goodrain/rainbond/db/errors"
httputil "github.com/goodrain/rainbond/util/http"
)
@ -52,16 +54,20 @@ func (t *ThirdPartyServiceController) addEndpoints(w http.ResponseWriter, r *htt
if !httputil.ValidatorRequestStructAndErrorResponse(r, w, &data, nil) {
return
}
values := make(map[string][]string)
if strings.Contains(data.IP, "127.0.0.1") {
values["ip"] = []string{"The ip field is can't contains '127.0.0.1'"}
}
if len(values) != 0 {
ipAddress := strings.Split(data.Address, ":")[0]
if err := validation.ValidateEndpointIP(ipAddress); len(err) > 0 {
values := make(map[string][]string, 1)
values["ip"] = err
httputil.ReturnValidationError(r, w, values)
return
}
sid := r.Context().Value(middleware.ContextKey("service_id")).(string)
if err := handler.Get3rdPartySvcHandler().AddEndpoints(sid, &data); err != nil {
if err == errors.ErrRecordAlreadyExist {
httputil.ReturnError(r, w, 400, err.Error())
return
}
httputil.ReturnError(r, w, 500, err.Error())
return
}
@ -73,14 +79,16 @@ func (t *ThirdPartyServiceController) updEndpoints(w http.ResponseWriter, r *htt
if !httputil.ValidatorRequestStructAndErrorResponse(r, w, &data, nil) {
return
}
values := make(map[string][]string)
if strings.Contains(data.IP, "127.0.0.1") {
values["ip"] = []string{"The ip field is can't contains '127.0.0.1'"}
}
if len(values) != 0 {
httputil.ReturnValidationError(r, w, values)
return
if data.Address != "" {
ipAddress := strings.Split(data.Address, ":")[0]
if err := validation.ValidateEndpointIP(ipAddress); len(err) > 0 {
values := make(map[string][]string, 1)
values["address"] = err
httputil.ReturnValidationError(r, w, values)
return
}
}
if err := handler.Get3rdPartySvcHandler().UpdEndpoints(&data); err != nil {
httputil.ReturnError(r, w, 500, err.Error())
return

59
api/controller/validation/validation.go Normal file
View File

@ -0,0 +1,59 @@
package validation
import (
"fmt"
"net"
"strings"
k8svalidation "k8s.io/apimachinery/pkg/util/validation"
)
// ValidateDomain tests that the argument is a valid domain.
func ValidateDomain(domain string) []string {
if strings.TrimSpace(domain) == "" {
return nil
}
var errs []string
if strings.Contains(domain, "*") {
errs = k8svalidation.IsWildcardDNS1123Subdomain(domain)
} else {
errs = k8svalidation.IsDNS1123Subdomain(domain)
}
return errs
}
// ValidateEndpointAddress tests that the argument is a valid endpoint address.
func ValidateEndpointAddress(address string) []string {
ip := net.ParseIP(address)
if ip == nil {
return ValidateDomain(address)
}
return ValidateEndpointIP(address)
}
// ValidateEndpointIP tests that the argument is a valid IP address.
func ValidateEndpointIP(ipAddress string) []string {
// We disallow some IPs as endpoints or external-ips. Specifically,
// unspecified and loopback addresses are nonsensical and link-local
// addresses tend to be used for node-centric purposes (e.g. metadata
// service).
err := []string{}
ip := net.ParseIP(ipAddress)
if ip == nil {
err = append(err, fmt.Sprintf("%s must be a valid IP address", ipAddress))
return err
}
if ip.IsUnspecified() {
err = append(err, fmt.Sprintf("%s may not be unspecified (0.0.0.0)", ipAddress))
}
if ip.IsLoopback() {
err = append(err, fmt.Sprintf("%s may not be in the loopback range (127.0.0.0/8)", ipAddress))
}
if ip.IsLinkLocalUnicast() {
err = append(err, fmt.Sprintf("%s may not be in the link-local range (169.254.0.0/16)", ipAddress))
}
if ip.IsLinkLocalMulticast() {
err = append(err, fmt.Sprintf("%s may not be in the link-local multicast range (224.0.0.0/24)", ipAddress))
}
return err
}

34
api/handler/app_restore_action.go
View File

@ -20,6 +20,12 @@ type AppRestoreAction struct {
func (a *AppRestoreAction) RestoreEnvs(tenantID, serviceID string, req *apimodel.RestoreEnvsReq) error {
// delete existing env
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
if err := db.GetManager().TenantServiceEnvVarDaoTransactions(tx).DelByServiceIDAndScope(serviceID, req.Scope); err != nil {
tx.Rollback()
return err
@ -55,6 +61,12 @@ func (a *AppRestoreAction) RestoreEnvs(tenantID, serviceID string, req *apimodel
func (a *AppRestoreAction) RestorePorts(tenantID, serviceID string, req *apimodel.RestorePortsReq) error {
// delete existing ports
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
if err := db.GetManager().TenantServicesPortDaoTransactions(tx).DelByServiceID(serviceID); err != nil {
tx.Rollback()
return err
@ -69,8 +81,8 @@ func (a *AppRestoreAction) RestorePorts(tenantID, serviceID string, req *apimode
port.ContainerPort = item.ContainerPort
port.Protocol = item.Protocol
port.PortAlias = item.PortAlias
port.IsInnerService = item.IsInnerService
port.IsOuterService = item.IsOuterService
port.IsInnerService = &item.IsInnerService
port.IsOuterService = &item.IsOuterService
if err := db.GetManager().TenantServicesPortDaoTransactions(tx).AddModel(port); err != nil {
if err == errors.ErrRecordAlreadyExist {
// ignore record already exist
@ -89,6 +101,12 @@ func (a *AppRestoreAction) RestorePorts(tenantID, serviceID string, req *apimode
func (a *AppRestoreAction) RestoreVolumes(tenantID, serviceID string, req *apimodel.RestoreVolumesReq) error {
// delete existing volumes
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
if err := db.GetManager().TenantServiceVolumeDaoTransactions(tx).DelShareableBySID(serviceID); err != nil {
tx.Rollback()
return err
@ -135,6 +153,12 @@ func (a *AppRestoreAction) RestoreVolumes(tenantID, serviceID string, req *apimo
// RestoreProbe restores service probe.
func (a *AppRestoreAction) RestoreProbe(serviceID string, req *apimodel.ServiceProbe) error {
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
if err := db.GetManager().ServiceProbeDaoTransactions(tx).DelByServiceID(serviceID); err != nil {
tx.Rollback()
return err
@ -199,6 +223,12 @@ func (a *AppRestoreAction) RestoreDeps(tenantID, serviceID string, req *apimodel
// RestoreDepVols restores service dependent volumes.
func (a *AppRestoreAction) RestoreDepVols(tenantID, serviceID string, req *apimodel.RestoreDepVolsReq) error {
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
if err := db.GetManager().TenantServiceMountRelationDaoTransactions(tx).DELTenantServiceMountRelationByServiceID(serviceID); err != nil {
tx.Rollback()
return err

44
api/handler/gateway_action.go
View File

@ -69,6 +69,12 @@ func (g *GatewayAction) AddHTTPRule(req *apimodel.AddHTTPRuleStruct) (string, er
// begin transaction
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
if err := db.GetManager().HTTPRuleDaoTransactions(tx).AddModel(httpRule); err != nil {
tx.Rollback()
return "", err
@ -111,6 +117,12 @@ func (g *GatewayAction) AddHTTPRule(req *apimodel.AddHTTPRuleStruct) (string, er
// UpdateHTTPRule updates http rule
func (g *GatewayAction) UpdateHTTPRule(req *apimodel.UpdateHTTPRuleStruct) (string, error) {
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
rule, err := g.dbmanager.HTTPRuleDaoTransactions(tx).GetHTTPRuleByID(req.HTTPRuleID)
if err != nil {
tx.Rollback()
@ -197,6 +209,12 @@ func (g *GatewayAction) UpdateHTTPRule(req *apimodel.UpdateHTTPRuleStruct) (stri
func (g *GatewayAction) DeleteHTTPRule(req *apimodel.DeleteHTTPRuleStruct) (string, error) {
// begin transaction
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
// delete http rule
httpRule, err := g.dbmanager.HTTPRuleDaoTransactions(tx).GetHTTPRuleByID(req.HTTPRuleID)
svcID := httpRule.ServiceID
@ -286,6 +304,12 @@ func (g *GatewayAction) UpdateCertificate(req apimodel.AddHTTPRuleStruct, httpRu
func (g *GatewayAction) AddTCPRule(req *apimodel.AddTCPRuleStruct) (string, error) {
// begin transaction
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
// add port
port := &model.TenantServiceLBMappingPort{
ServiceID: req.ServiceID,
@ -334,6 +358,12 @@ func (g *GatewayAction) AddTCPRule(req *apimodel.AddTCPRuleStruct) (string, erro
func (g *GatewayAction) UpdateTCPRule(req *apimodel.UpdateTCPRuleStruct, minPort int) (string, error) {
// begin transaction
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
// get old tcp rule
tcpRule, err := g.dbmanager.TCPRuleDaoTransactions(tx).GetTCPRuleByID(req.TCPRuleID)
if err != nil {
@ -410,6 +440,12 @@ func (g *GatewayAction) UpdateTCPRule(req *apimodel.UpdateTCPRuleStruct, minPort
func (g *GatewayAction) DeleteTCPRule(req *apimodel.DeleteTCPRuleStruct) (string, error) {
// begin transaction
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
tcpRule, err := db.GetManager().TCPRuleDaoTransactions(tx).GetTCPRuleByID(req.TCPRuleID)
if err != nil {
tx.Rollback()
@ -621,7 +657,7 @@ func (g *GatewayAction) RuleConfig(req *apimodel.RuleConfigReq) error {
})
setheaders := make(map[string]string)
for _, item := range req.Body.SetHeaders {
if strings.TrimSpace(item.Key) == ""{
if strings.TrimSpace(item.Key) == "" {
continue
}
if strings.TrimSpace(item.Value) == "" {
@ -639,6 +675,12 @@ func (g *GatewayAction) RuleConfig(req *apimodel.RuleConfigReq) error {
}
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
if err := g.dbmanager.GwRuleConfigDaoTransactions(tx).DeleteByRuleID(req.RuleID); err != nil {
tx.Rollback()
return err

28
api/handler/group/group_backup.go
View File

@ -26,26 +26,18 @@ import (
"strings"
"time"
v1 "github.com/goodrain/rainbond/worker/appm/types/v1"
"github.com/coreos/etcd/clientv3"
"github.com/goodrain/rainbond/event"
"github.com/Sirupsen/logrus"
"github.com/goodrain/rainbond/worker/client"
mqclient "github.com/goodrain/rainbond/mq/client"
"github.com/jinzhu/gorm"
"github.com/pquerna/ffjson/ffjson"
"github.com/coreos/etcd/clientv3"
"github.com/goodrain/rainbond/api/util"
"github.com/goodrain/rainbond/db"
dbmodel "github.com/goodrain/rainbond/db/model"
"github.com/goodrain/rainbond/event"
mqclient "github.com/goodrain/rainbond/mq/client"
core_util "github.com/goodrain/rainbond/util"
v1 "github.com/goodrain/rainbond/worker/appm/types/v1"
"github.com/goodrain/rainbond/worker/client"
"github.com/jinzhu/gorm"
"github.com/pquerna/ffjson/ffjson"
)
//Backup GroupBackup
@ -276,11 +268,11 @@ func (h *BackupHandle) snapshot(ids []string, sourceDir string) error {
return fmt.Errorf("Get service(%s) ports error %s", id, err)
}
data.ServicePort = servicePorts
versions, err := db.GetManager().VersionInfoDao().GetVersionByServiceID(id)
if err != nil && err.Error() != gorm.ErrRecordNotFound.Error() {
version, err := db.GetManager().VersionInfoDao().GetLatestScsVersion(id)
if err != nil && err != gorm.ErrRecordNotFound {
return fmt.Errorf("Get service(%s) build versions error %s", id, err)
}
data.Versions = versions
data.Versions = []*dbmodel.VersionInfo{version}
pluginConfigs, err := db.GetManager().TenantPluginVersionConfigDao().GetPluginConfigs(id)
if err != nil && err.Error() != gorm.ErrRecordNotFound.Error() {
return fmt.Errorf("Get service(%s) plugin configs error %s", id, err)

18
api/handler/plugin.go
View File

@ -89,6 +89,12 @@ func (p *PluginAction) UpdatePluginAct(pluginID, tenantID string, cps *api_model
//DeletePluginAct DeletePluginAct
func (p *PluginAction) DeletePluginAct(pluginID, tenantID string) *util.APIHandleError {
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
//step1: delete service plugin relation
err := db.GetManager().TenantServicePluginRelationDaoTransactions(tx).DeleteALLRelationByPluginID(pluginID)
if err != nil {
@ -126,6 +132,12 @@ func (p *PluginAction) GetPlugins(tenantID string) ([]*dbmodel.TenantPlugin, *ut
//AddDefaultEnv AddDefaultEnv
func (p *PluginAction) AddDefaultEnv(est *api_model.ENVStruct) *util.APIHandleError {
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
for _, env := range est.Body.EVNInfo {
vis := &dbmodel.TenantPluginDefaultENV{
PluginID: est.PluginID,
@ -347,6 +359,12 @@ func (p *PluginAction) GetPluginBuildVersion(pluginID, versionID string) (*dbmod
func (p *PluginAction) DeletePluginBuildVersion(pluginID, versionID string) *util.APIHandleError {
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
err := db.GetManager().TenantPluginBuildVersionDaoTransactions(tx).DeleteBuildVersionByVersionID(versionID)
if err != nil {
tx.Rollback()

126
api/handler/service.go
View File

@ -235,6 +235,12 @@ func (s *ServiceAction) isWindowsService(serviceID string) bool {
//AddLabel add labels
func (s *ServiceAction) AddLabel(l *api_model.LabelsStruct, serviceID string) error {
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
for _, label := range l.Labels {
var labelModel dbmodel.TenantServiceLable
switch label.LabelKey {
@ -262,6 +268,12 @@ func (s *ServiceAction) AddLabel(l *api_model.LabelsStruct, serviceID string) er
//UpdateLabel updates labels
func (s *ServiceAction) UpdateLabel(l *api_model.LabelsStruct, serviceID string) error {
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
for _, label := range l.Labels {
// delete old labels
err := db.GetManager().TenantServiceLabelDaoTransactions(tx).
@ -300,6 +312,12 @@ func (s *ServiceAction) UpdateLabel(l *api_model.LabelsStruct, serviceID string)
//DeleteLabel deletes label
func (s *ServiceAction) DeleteLabel(l *api_model.LabelsStruct, serviceID string) error {
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
for _, label := range l.Labels {
err := db.GetManager().TenantServiceLabelDaoTransactions(tx).
DelTenantServiceLabelsByServiceIDKeyValue(serviceID, label.LabelKey, label.LabelValue)
@ -476,6 +494,12 @@ func (s *ServiceAction) ServiceCreate(sc *api_model.ServiceStruct) error {
dependIds := sc.DependIDs
ts.DeployVersion = ""
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
//create app
if err := db.GetManager().TenantServiceDaoTransactions(tx).AddModel(&ts); err != nil {
logrus.Errorf("add service error, %v", err)
@ -607,6 +631,9 @@ func (s *ServiceAction) ServiceCreate(sc *api_model.ServiceStruct) error {
// sc.Endpoints can't be nil
// sc.Endpoints.Discovery or sc.Endpoints.Static can't be nil
if sc.Kind == dbmodel.ServiceKindThirdParty.String() { // TODO: validate request data
if sc.Endpoints == nil {
return fmt.Errorf("endpoints can not be empty for third-party service")
}
if config := strings.Replace(sc.Endpoints.Discovery, " ", "", -1); config != "" {
var cfg dCfg
err := json.Unmarshal([]byte(config), &cfg)
@ -928,9 +955,8 @@ func (s *ServiceAction) PortVar(action, tenantID, serviceID string, vps *api_mod
var vpD dbmodel.TenantServicesPort
vpD.ServiceID = serviceID
vpD.TenantID = tenantID
//默认不打开
vpD.IsInnerService = false
vpD.IsOuterService = false
vpD.IsInnerService = &vp.IsInnerService
vpD.IsOuterService = &vp.IsOuterService
vpD.ContainerPort = vp.ContainerPort
vpD.MappingPort = vp.MappingPort
vpD.Protocol = vp.Protocol
@ -942,6 +968,12 @@ func (s *ServiceAction) PortVar(action, tenantID, serviceID string, vps *api_mod
}
case "delete":
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
for _, vp := range vps.Port {
if err := db.GetManager().TenantServicesPortDaoTransactions(tx).DeleteModel(serviceID, vp.ContainerPort); err != nil {
logrus.Errorf("delete port var error, %v", err)
@ -956,6 +988,12 @@ func (s *ServiceAction) PortVar(action, tenantID, serviceID string, vps *api_mod
}
case "update":
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
for _, vp := range vps.Port {
//port更新单个请求
if oldPort == 0 {
@ -968,8 +1006,8 @@ func (s *ServiceAction) PortVar(action, tenantID, serviceID string, vps *api_mod
}
vpD.ServiceID = serviceID
vpD.TenantID = tenantID
vpD.IsInnerService = vp.IsInnerService
vpD.IsOuterService = vp.IsOuterService
vpD.IsInnerService = &vp.IsInnerService
vpD.IsOuterService = &vp.IsOuterService
vpD.ContainerPort = vp.ContainerPort
vpD.MappingPort = vp.MappingPort
vpD.Protocol = vp.Protocol
@ -1036,9 +1074,16 @@ func (s *ServiceAction) PortOuter(tenantName, serviceID string, containerPort in
vsPort := &dbmodel.TenantServiceLBMappingPort{}
switch servicePort.Body.Operation {
case "close":
if p.IsOuterService { //如果端口已经开了对外
p.IsOuterService = false
if *p.IsOuterService { //如果端口已经开了对外
falsev := false
p.IsOuterService = &falsev
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
if err = db.GetManager().TenantServicesPortDaoTransactions(tx).UpdateModel(p); err != nil {
tx.Rollback()
return nil, "", err
@ -1058,7 +1103,7 @@ func (s *ServiceAction) PortOuter(tenantName, serviceID string, containerPort in
tx.Rollback()
return nil, "", fmt.Errorf("outer, get plugin mapping port error:(%s)", err)
}
if p.IsInnerService {
if *p.IsInnerService {
//发现内网未关闭则不删除该映射
logrus.Debugf("outer, close outer, but plugin inner port (%d) is exist, do not need delete", containerPort)
goto OUTERCLOSEPASS
@ -1083,7 +1128,7 @@ func (s *ServiceAction) PortOuter(tenantName, serviceID string, containerPort in
}
case "open":
if p.IsOuterService {
if *p.IsOuterService {
if p.Protocol != "http" && p.Protocol != "https" && servicePort.Body.IfCreateExPort {
vsPort, err = s.createVSPort(serviceID, p.ContainerPort)
if vsPort == nil {
@ -1092,8 +1137,15 @@ func (s *ServiceAction) PortOuter(tenantName, serviceID string, containerPort in
return vsPort, p.Protocol, nil
}
}
p.IsOuterService = true
truev := true
p.IsOuterService = &truev
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
if err = db.GetManager().TenantServicesPortDaoTransactions(tx).UpdateModel(p); err != nil {
tx.Rollback()
return nil, "", err
@ -1169,10 +1221,17 @@ func (s *ServiceAction) PortInner(tenantName, serviceID, operation string, port
return fmt.Errorf("get plugin relations error: %s", err.Error())
}
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
switch operation {
case "close":
if p.IsInnerService { //如果端口已经开了对内
p.IsInnerService = false
if *p.IsInnerService { //如果端口已经开了对内
falsev := false
p.IsInnerService = &falsev
if err = db.GetManager().TenantServicesPortDaoTransactions(tx).UpdateModel(p); err != nil {
tx.Rollback()
return fmt.Errorf("update service port error: %s", err.Error())
@ -1191,7 +1250,7 @@ func (s *ServiceAction) PortInner(tenantName, serviceID, operation string, port
tx.Rollback()
return fmt.Errorf("inner, get plugin mapping port error:(%s)", err)
}
if p.IsOuterService {
if *p.IsOuterService {
logrus.Debugf("inner, close inner, but plugin outerport (%d) is exist, do not need delete", port)
goto INNERCLOSEPASS
}
@ -1211,11 +1270,12 @@ func (s *ServiceAction) PortInner(tenantName, serviceID, operation string, port
return fmt.Errorf("already close")
}
case "open":
if p.IsInnerService {
if *p.IsInnerService {
tx.Rollback()
return fmt.Errorf("already open")
}
p.IsInnerService = true
truv := true
p.IsInnerService = &truv
if err = db.GetManager().TenantServicesPortDaoTransactions(tx).UpdateModel(p); err != nil {
tx.Rollback()
return err
@ -1275,6 +1335,12 @@ func (s *ServiceAction) ChangeLBPort(tenantID, serviceID string, containerPort,
oldmapport.Port = mapport.Port
mapport.Port = port
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
if err := db.GetManager().TenantServiceLBMappingPortDaoTransactions(tx).DELServiceLBMappingPortByServiceIDAndPort(oldmapport.ServiceID, port); err != nil {
tx.Rollback()
return nil, util.CreateAPIHandleErrorFromDBError("change lb port", err)
@ -1330,6 +1396,12 @@ func (s *ServiceAction) VolumnVar(tsv *dbmodel.TenantServiceVolume, tenantID, fi
}
// begin transaction
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
if err := db.GetManager().TenantServiceVolumeDaoTransactions(tx).AddModel(tsv); err != nil {
tx.Rollback()
return util.CreateAPIHandleErrorFromDBError("add volume", err)
@ -1353,6 +1425,12 @@ func (s *ServiceAction) VolumnVar(tsv *dbmodel.TenantServiceVolume, tenantID, fi
case "delete":
// begin transaction
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
if tsv.VolumeName != "" {
err := db.GetManager().TenantServiceVolumeDaoTransactions(tx).DeleteModel(tsv.ServiceID, tsv.VolumeName)
if err != nil && err.Error() != gorm.ErrRecordNotFound.Error() {
@ -1381,6 +1459,12 @@ func (s *ServiceAction) VolumnVar(tsv *dbmodel.TenantServiceVolume, tenantID, fi
// UpdVolume updates service volume.
func (s *ServiceAction) UpdVolume(sid string, req *api_model.UpdVolumeReq) error {
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
switch req.VolumeType {
case "config-file":
if req.VolumePath != "" {
@ -1569,6 +1653,12 @@ func (s *ServiceAction) CreateTenant(t *dbmodel.Tenants) error {
return fmt.Errorf("tenant name %s is exist", t.Name)
}
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
if err := db.GetManager().TenantDaoTransactions(tx).AddModel(t); err != nil {
if !strings.HasSuffix(err.Error(), "is exist") {
tx.Rollback()
@ -1714,6 +1804,12 @@ func (s *ServiceAction) TransServieToDelete(serviceID string) error {
}
}
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
delService := service.ChangeDelete()
delService.ID = 0
if err := db.GetManager().TenantServiceDeleteDaoTransactions(tx).AddModel(delService); err != nil {

26
api/handler/service_plugin.go
View File

@ -46,6 +46,12 @@ func (s *ServiceAction) GetTenantServicePluginRelation(serviceID string) ([]*dbm
//TenantServiceDeletePluginRelation uninstall plugin for app
func (s *ServiceAction) TenantServiceDeletePluginRelation(tenantID, serviceID, pluginID string) *util.APIHandleError {
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
deleteFunclist := []func(serviceID, pluginID string) error{
db.GetManager().TenantServicePluginRelationDaoTransactions(tx).DeleteRelationByServiceIDAndPluginID,
db.GetManager().TenantPluginVersionENVDaoTransactions(tx).DeleteEnvByPluginID,
@ -106,12 +112,18 @@ func (s *ServiceAction) SetTenantServicePluginRelation(tenantID, serviceID strin
return nil, util.CreateAPIHandleErrorFromDBError("get ports by service id", err)
}
for _, p := range ports {
if p.IsInnerService || p.IsOuterService {
if *p.IsInnerService || *p.IsOuterService {
openPorts[p.ContainerPort] = true
}
}
}
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
if configs := pss.Body.ConfigEnvs.ComplexEnvs; configs != nil {
if configs.BasePorts != nil && checkPluginHaveInbound(plugin.PluginModel) {
for _, p := range configs.BasePorts {
@ -208,6 +220,12 @@ func (s *ServiceAction) UpdateVersionEnv(uve *api_model.SetVersionEnv) *util.API
return util.CreateAPIHandleErrorFromDBError("get plugin by plugin id", err)
}
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
if len(uve.Body.ConfigEnvs.NormalEnvs) != 0 {
if err := s.upNormalEnvs(tx, uve); err != nil {
tx.Rollback()
@ -275,6 +293,12 @@ func (s *ServiceAction) SavePluginConfig(serviceID, pluginID string, config *api
return util.CreateAPIHandleError(500, err)
}
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
if err := db.GetManager().TenantPluginVersionConfigDaoTransactions(tx).AddModel(&dbmodel.TenantPluginVersionDiscoverConfig{
PluginID: pluginID,
ServiceID: serviceID,

6
api/handler/tenant.go
View File

@ -380,6 +380,12 @@ func (t *TenantAction) TransPlugins(tenantID, tenantName, fromTenant string, plu
}
goodrainID := tenantInfo.UUID
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
for _, p := range pluginList {
pluginInfo, err := db.GetManager().TenantPluginDao().GetPluginByID(p, goodrainID)
if err != nil {

34
api/handler/third_party_service_handler.go
View File

@ -20,6 +20,9 @@ package handler
import (
"fmt"
"strconv"
"strings"
"github.com/Sirupsen/logrus"
"github.com/goodrain/rainbond/api/model"
"github.com/goodrain/rainbond/db"
@ -44,11 +47,12 @@ func Create3rdPartySvcHandler(dbmanager db.Manager, statusCli *client.AppRuntime
// AddEndpoints adds endpoints for third-party service.
func (t *ThirdPartyServiceHanlder) AddEndpoints(sid string, d *model.AddEndpiontsReq) error {
address, port := convertAddressPort(d.Address)
ep := &dbmodel.Endpoint{
UUID: util.NewUUID(),
ServiceID: sid,
IP: d.IP,
Port: 0,
IP: address,
Port: port,
IsOnline: &d.IsOnline,
}
if err := t.dbmanager.EndpointsDao().AddModel(ep); err != nil {
@ -66,8 +70,10 @@ func (t *ThirdPartyServiceHanlder) UpdEndpoints(d *model.UpdEndpiontsReq) error
logrus.Warningf("EpID: %s; error getting endpoints: %v", d.EpID, err)
return err
}
if d.IP != "" {
ep.IP = d.IP
if d.Address != "" {
address, port := convertAddressPort(d.Address)
ep.IP = address
ep.Port = port
}
ep.IsOnline = &d.IsOnline
if err := t.dbmanager.EndpointsDao().UpdateModel(ep); err != nil {
@ -79,6 +85,22 @@ func (t *ThirdPartyServiceHanlder) UpdEndpoints(d *model.UpdEndpiontsReq) error
return nil
}
func convertAddressPort(s string) (address string, port int) {
addressport := strings.Split(s, ":")
// compatible with ipv6
addressSli := addressport[:func() int {
if len(addressport) == 2 {
return len(addressport) - 1
}
return len(addressport)
}()]
address = strings.Join(addressSli, ":")
if len(addressport) == 2 {
port, _ = strconv.Atoi(addressport[1])
}
return address, port
}
// DelEndpoints deletes endpoints for third-party service.
func (t *ThirdPartyServiceHanlder) DelEndpoints(epid, sid string) error {
if err := t.dbmanager.EndpointsDao().DelByUUID(epid); err != nil {
@ -100,7 +122,7 @@ func (t *ThirdPartyServiceHanlder) ListEndpoints(sid string) ([]*model.EndpointR
for _, item := range endpoints {
m[item.UUID] = &model.EndpointResp{
EpID: item.UUID,
IP: func(ip string, p int) string {
Address: func(ip string, p int) string {
if p != 0 {
return fmt.Sprintf("%s:%d", ip, p)
}
@ -127,7 +149,7 @@ func (t *ThirdPartyServiceHanlder) ListEndpoints(sid string) ([]*model.EndpointR
}
m[item.Uuid] = &model.EndpointResp{
EpID: item.Uuid,
IP: item.Ip,
Address: item.Ip,
Status: item.Status,
IsOnline: true,
IsStatic: false,

6
api/model/third_party_srvice.go
View File

@ -20,14 +20,14 @@ package model
// AddEndpiontsReq is one of the Endpoints in the request to add the endpints.
type AddEndpiontsReq struct {
IP string `json:"ip" validate:"required|ip_v4"`
Address string `json:"address" validate:"address|required"`
IsOnline bool `json:"is_online" validate:"required"`
}
// UpdEndpiontsReq is one of the Endpoints in the request to update the endpints.
type UpdEndpiontsReq struct {
EpID string `json:"ep_id" validate:"required|len:32"`
IP string `json:"ip" validate:"ip_v4"`
Address string `json:"address"`
IsOnline bool `json:"is_online" validate:"required"`
}
@ -40,7 +40,7 @@ type DelEndpiontsReq struct {
// update or delete the endpints.
type EndpointResp struct {
EpID string `json:"ep_id"`
IP string `json:"ip"`
Address string `json:"address"`
Status string `json:"status"`
IsOnline bool `json:"is_online"`
IsStatic bool `json:"is_static"`

14
builder/exector/groupapp_restore.go
View File

@ -27,15 +27,13 @@ import (
"strings"
"time"
"github.com/Sirupsen/logrus"
"github.com/coreos/etcd/clientv3"
"github.com/docker/docker/client"
"github.com/goodrain/rainbond/builder"
"github.com/goodrain/rainbond/builder/sources"
dbmodel "github.com/goodrain/rainbond/db/model"
"github.com/Sirupsen/logrus"
"github.com/docker/docker/client"
"github.com/goodrain/rainbond/db"
dbmodel "github.com/goodrain/rainbond/db/model"
"github.com/goodrain/rainbond/event"
"github.com/goodrain/rainbond/util"
"github.com/pquerna/ffjson/ffjson"
@ -399,6 +397,12 @@ func (b *BackupAPPRestore) modify(appSnapshots []*RegionServiceSnapshot) error {
}
func (b *BackupAPPRestore) restoreMetadata(appSnapshots []*RegionServiceSnapshot) error {
tx := db.GetManager().Begin()
defer func() {
if r := recover(); r != nil {
logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r)
tx.Rollback()
}
}()
for _, app := range appSnapshots {
app.Service.ID = 0
if err := db.GetManager().TenantServiceDaoTransactions(tx).AddModel(app.Service); err != nil {

5
builder/parser/docker_compose.go
View File

@ -110,6 +110,11 @@ func (d *DockerComposeParse) Parse() ParseErrorList {
logrus.Debugf("service config is %v, container name is %s", sc, sc.ContainerName)
ports := make(map[int]*types.Port)
if sc.Image == "" {
d.errappend(ErrorAndSolve(FatalError, fmt.Sprintf("ComposeFile解析错误"), SolveAdvice(fmt.Sprintf("Service %s has no image specified", kev), fmt.Sprintf("请为%s指定一个镜像", kev))))
continue
}
for _, p := range sc.Port {
pro := string(p.Protocol)
if pro != "udp" {

2
builder/parser/source_code.go
View File

@ -345,7 +345,7 @@ func (d *SourceCodeParse) Parse() ParseErrorList {
}
}
if rbdfileConfig != nil {
if rbdfileConfig != nil && d.isMulti {
rbdfileConfig.Envs = nil
rbdfileConfig.Ports = nil
}

10
builder/sources/image.go
View File

@ -40,11 +40,17 @@ import (
//"github.com/docker/docker/client"
"github.com/docker/docker/client"
"github.com/docker/docker/pkg/archive"
"github.com/goodrain/rainbond/builder"
"github.com/goodrain/rainbond/builder/model"
"github.com/goodrain/rainbond/event"
"github.com/goodrain/rainbond/builder"
)
//ErrorNoAuth error no auth
var ErrorNoAuth = fmt.Errorf("pull image require docker login")
//ErrorNoImage error no image
var ErrorNoImage = fmt.Errorf("image not exist")
//ImagePull 拉取镜像
//timeout 分钟为单位
func ImagePull(dockerCli *client.Client, image string, username, password string, logger event.Logger, timeout int) (*types.ImageInspect, error) {
@ -79,7 +85,7 @@ func ImagePull(dockerCli *client.Client, image string, username, password string
defer cancel()
//TODO: 使用1.12版本api的bug “repository name must be canonical”，使用rf.String()完整的镜像地址
readcloser, err := dockerCli.ImagePull(ctx, rf.String(), pullipo)
logger.Info(fmt.Sprintf("成功获取镜像：%s", image), map[string]string{"step": "pullimage"})
logger.Info(fmt.Sprintf("Success Pull Image：%s", image), map[string]string{"step": "pullimage"})
if err != nil {
logrus.Debugf("image name: %s readcloser error: %v", image, err.Error())
if strings.HasSuffix(err.Error(), "does not exist or no pull access") {

10
builder/sources/image_test.go
View File

@ -22,6 +22,8 @@ import (
"fmt"
"testing"
"github.com/goodrain/rainbond/event"
"github.com/docker/docker/api/types"
"github.com/docker/docker/client"
)
@ -84,3 +86,11 @@ func TestImageImport(t *testing.T) {
t.Fatal(err)
}
}
func TestImagePull(t *testing.T) {
dc, _ := client.NewEnvClient()
_, err := ImagePull(dc, "barnett/collabora:190422", "", "", event.GetTestLogger(), 60)
if err != nil {
t.Fatal(err)
}
}

2
cmd/builder/option/option.go
View File

@ -78,7 +78,7 @@ func (a *Builder) AddFlags(fs *pflag.FlagSet) {
fs.StringVar(&a.RunMode, "run", "sync", "sync data when worker start")
fs.StringVar(&a.DockerEndpoint, "dockerd", "127.0.0.1:2376", "dockerd endpoint")
fs.StringVar(&a.HostIP, "hostIP", "", "Current node Intranet IP")
fs.BoolVar(&a.CleanUp, "clean-up", false, "Turn on build version cleanup")
fs.BoolVar(&a.CleanUp, "clean-up", true, "Turn on build version cleanup")
fs.StringVar(&a.Topic, "topic", "builder", "Topic in mq,you coule choose `builder` or `windows_builder`")
}

1
db/dao/dao.go
View File

@ -318,6 +318,7 @@ type VersionInfoDao interface {
GetVersionByEventID(eventID string) (*model.VersionInfo, error)
GetVersionByDeployVersion(version, serviceID string) (*model.VersionInfo, error)
GetVersionByServiceID(serviceID string) ([]*model.VersionInfo, error)
GetLatestScsVersion(sid string) (*model.VersionInfo, error)
GetAllVersionByServiceID(serviceID string) ([]*model.VersionInfo, error)
DeleteVersionByEventID(eventID string) error
DeleteVersionByServiceID(serviceID string) error

4
db/model/tenant.go
View File

@ -227,8 +227,8 @@ type TenantServicesPort struct {
MappingPort int `gorm:"column:mapping_port" validate:"mapping_port|required|numeric_between:1,65535" json:"mapping_port"`
Protocol string `gorm:"column:protocol" validate:"protocol|required|in:http,https,tcp,grpc,udp,mysql" json:"protocol"`
PortAlias string `gorm:"column:port_alias" validate:"port_alias|required|alpha_dash" json:"port_alias"`
IsInnerService bool `gorm:"column:is_inner_service" validate:"is_inner_service|bool" json:"is_inner_service"`
IsOuterService bool `gorm:"column:is_outer_service" validate:"is_outer_service|bool" json:"is_outer_service"`
IsInnerService *bool `gorm:"column:is_inner_service" validate:"is_inner_service|bool" json:"is_inner_service"`
IsOuterService *bool `gorm:"column:is_outer_service" validate:"is_outer_service|bool" json:"is_outer_service"`
}
//TableName 表名

4
db/model/version.go
View File

@ -19,6 +19,8 @@
package model
import (
"fmt"
"github.com/Sirupsen/logrus"
"github.com/docker/distribution/reference"
)
@ -67,6 +69,6 @@ func (t *VersionInfo) CreateShareImage(hubURL, namespace, appVersion string) (st
if namespace != "" {
image.Namespace = namespace
}
image.Name = image.Name + "_" + appVersion
image.Name = fmt.Sprintf("%s:%s", t.ServiceID, t.BuildVersion)
return image.String(), nil
}

3
db/mysql/dao/3rd_party.go
View File

@ -24,6 +24,7 @@ import (
"strings"
"github.com/goodrain/rainbond/db/model"
"github.com/goodrain/rainbond/db/errors"
"github.com/jinzhu/gorm"
)
@ -44,7 +45,7 @@ func (e *EndpointDaoImpl) AddModel(mo model.Interface) error {
return err
}
} else {
return fmt.Errorf("Endpoint exists based on servicd_id(%s) and ip(%s)", ep.ServiceID, ep.IP)
return errors.ErrRecordAlreadyExist
}
return nil
}

12
db/mysql/dao/version.go
View File

@ -105,6 +105,15 @@ func (c *VersionInfoDaoImpl) GetVersionByServiceID(serviceID string) ([]*model.V
return result, nil
}
// GetLatestScsVersion returns the latest versoin that the final_status is 'success'.
func (c *VersionInfoDaoImpl) GetLatestScsVersion(sid string) (*model.VersionInfo, error) {
var result model.VersionInfo
if err := c.DB.Where("service_id=? and final_status='success'", sid).Last(&result).Error; err != nil {
return nil, err
}
return &result, nil
}
//GetAllVersionByServiceID get all versions by service id, not only successful
func (c *VersionInfoDaoImpl) GetAllVersionByServiceID(serviceID string) ([]*model.VersionInfo, error) {
var result []*model.VersionInfo
@ -144,7 +153,8 @@ func (c *VersionInfoDaoImpl) DeleteFailureVersionInfo(timePoint time.Time, statu
//SearchVersionInfo query version count >5
func (c *VersionInfoDaoImpl) SearchVersionInfo() ([]*model.VersionInfo, error) {
var result []*model.VersionInfo
if err := c.DB.Table("version_info").Select("service_id").Group("service_id").Having("count(ID) > ?", 5).Scan(&result).Error; err != nil {
versionInfo := &model.VersionInfo{}
if err := c.DB.Table(versionInfo.TableName()).Select("service_id").Group("service_id").Having("count(ID) > ?", 5).Scan(&result).Error; err != nil {
return nil, err
}
return result, nil

23
db/tenant_services_port_test.go
View File

@ -21,12 +21,13 @@ package db
import (
"context"
"fmt"
"testing"
"time"
dbconfig "github.com/goodrain/rainbond/db/config"
"github.com/goodrain/rainbond/db/model"
"github.com/goodrain/rainbond/util"
"github.com/testcontainers/testcontainers-go"
"testing"
"time"
)
func TestTenantServicesDao_GetOpenedPort(t *testing.T) {
@ -81,12 +82,14 @@ func TestTenantServicesDao_GetOpenedPort(t *testing.T) {
}
sid := util.NewUUID()
trueVal := true
falseVal := true
err = GetManager().TenantServicesPortDao().AddModel(&model.TenantServicesPort{
ServiceID: sid,
ContainerPort: 1111,
MappingPort: 1111,
IsInnerService: false,
IsOuterService: true,
IsInnerService: &falseVal,
IsOuterService: &trueVal,
})
if err != nil {
t.Fatal(err)
@ -95,8 +98,8 @@ func TestTenantServicesDao_GetOpenedPort(t *testing.T) {
ServiceID: sid,
ContainerPort: 2222,
MappingPort: 2222,
IsInnerService: true,
IsOuterService: false,
IsInnerService: &trueVal,
IsOuterService: &falseVal,
})
if err != nil {
t.Fatal(err)
@ -105,8 +108,8 @@ func TestTenantServicesDao_GetOpenedPort(t *testing.T) {
ServiceID: sid,
ContainerPort: 3333,
MappingPort: 3333,
IsInnerService: false,
IsOuterService: false,
IsInnerService: &falseVal,
IsOuterService: &falseVal,
})
if err != nil {
t.Fatal(err)
@ -115,8 +118,8 @@ func TestTenantServicesDao_GetOpenedPort(t *testing.T) {
ServiceID: sid,
ContainerPort: 5555,
MappingPort: 5555,
IsInnerService: true,
IsOuterService: true,
IsInnerService: &trueVal,
IsOuterService: &trueVal,
})
if err != nil {
t.Fatal(err)

10
db/tenant_test.go
View File

@ -164,10 +164,12 @@ func TestTenantServicesPortDao_HasOpenPort(t *testing.T) {
t.Error("Expected false for hasOpenPort, but returned true")
}
})
trueVal := true
falseVal := true
t.Run("outer service", func(t *testing.T) {
port := &model.TenantServicesPort{
ServiceID: util.NewUUID(),
IsOuterService: true,
IsOuterService: &trueVal,
}
if err := GetManager().TenantServicesPortDao().AddModel(port); err != nil {
t.Fatalf("error creating TenantServicesPort: %v", err)
@ -180,7 +182,7 @@ func TestTenantServicesPortDao_HasOpenPort(t *testing.T) {
t.Run("inner service", func(t *testing.T) {
port := &model.TenantServicesPort{
ServiceID: util.NewUUID(),
IsInnerService: true,
IsInnerService: &trueVal,
}
if err := GetManager().TenantServicesPortDao().AddModel(port); err != nil {
t.Fatalf("error creating TenantServicesPort: %v", err)
@ -193,8 +195,8 @@ func TestTenantServicesPortDao_HasOpenPort(t *testing.T) {
t.Run("not inner or outer service", func(t *testing.T) {
port := &model.TenantServicesPort{
ServiceID: util.NewUUID(),
IsInnerService: false,
IsOuterService: false,
IsInnerService: &falseVal,
IsOuterService: &falseVal,
}
if err := GetManager().TenantServicesPortDao().AddModel(port); err != nil {
t.Fatalf("error creating TenantServicesPort: %v", err)

4
gateway/metric/collectors/socket_message.go
View File

@ -28,7 +28,6 @@ import (
"os"
"github.com/Sirupsen/logrus"
jsoniter "github.com/json-iterator/go"
"github.com/prometheus/client_golang/prometheus"
"k8s.io/apimachinery/pkg/util/sets"
@ -209,7 +208,7 @@ func (sc *SocketCollector) handleMessage(msg []byte) {
return
}
for _, stats := range statsBatch {
if !sc.hosts.Has(stats.Host) {
if !sc.hosts.HasAny(stats.Host, "tls"+stats.Host) {
logrus.Infof("skiping metric for host %v that is not being served", stats.Host)
continue
}
@ -297,7 +296,6 @@ func (sc *SocketCollector) Start() {
continue
}
go handleMessages(conn, sc.handleMessage)
logrus.Infof("Start socket collector for metric")
}
}

49
grctl/cmd/cluster.go
View File

@ -176,28 +176,6 @@ func summaryResult(list []map[string]string) (status string, errMessage string)
return
}
func handleRoleAndStatus(list []map[string]string) bool {
var computeFlag bool
var manageFlag bool
for _, v := range list {
if v["role"] == "compute" && v["status"] == "running" {
computeFlag = true
}
if v["role"] == "manage" && v["status"] == "running" {
manageFlag = true
}
if (strings.HasPrefix(v["role"], "compute,manage") || strings.HasPrefix(v["role"], "manage,compute")) && v["status"] == "running" {
computeFlag = true
manageFlag = true
}
}
if computeFlag && manageFlag {
return true
}
return false
}
func handleNodeReady(list []map[string]string) bool {
trueNum := 0
for _, v := range list {
@ -215,7 +193,6 @@ func clusterStatus(roleList []map[string]string, ReadyList []map[string]string)
var clusterStatus string
var errMessage string
readyStatus := handleNodeReady(ReadyList)
roleStatus := handleRoleAndStatus(roleList)
if readyStatus {
clusterStatus = "\033[0;32;32mhealthy\033[0m"
errMessage = ""
@ -223,9 +200,31 @@ func clusterStatus(roleList []map[string]string, ReadyList []map[string]string)
clusterStatus = "\033[0;31;31munhealthy\033[0m"
errMessage = "There is a service exception in the cluster"
}
if !roleStatus {
var computeFlag bool
var manageFlag bool
var gatewayFlag bool
for _, v := range roleList {
if strings.Contains(v["role"], "compute") && v["status"] == "running" {
computeFlag = true
}
if strings.Contains(v["role"], "manage") && v["status"] == "running" {
manageFlag = true
}
if strings.Contains(v["role"], "gateway") && v["status"] == "running" {
manageFlag = true
}
}
if !manageFlag {
clusterStatus = "\033[0;33;33munavailable\033[0m"
errMessage = "No compute nodes or management nodes are available in the cluster"
errMessage = "No management nodes are available in the cluster"
}
if !computeFlag {
clusterStatus = "\033[0;33;33munavailable\033[0m"
errMessage = "No compute nodes are available in the cluster"
}
if !gatewayFlag {
clusterStatus = "\033[0;33;33munavailable\033[0m"
errMessage = "No gateway nodes are available in the cluster"
}
return clusterStatus, errMessage
}

7
grctl/cmd/node.go
View File

@ -150,7 +150,12 @@ func getStatusShow(v *client.HostNode) (status string) {
nss.message = append(nss.message, "unhealth")
nss.color = color.FgRed
}
return nss.String()
result := nss.String()
if strings.Contains(result, "unknown") {
result = "unknown"
}
return result
}
func handleStatus(serviceTable *termtables.Table, v *client.HostNode) {
serviceTable.AddRow(v.ID, v.InternalIP, v.HostName, v.Role.String(), getStatusShow(v))

31
grctl/cmd/service.go
View File

@ -36,6 +36,7 @@ import (
"github.com/tidwall/gjson"
"github.com/urfave/cli"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
corev1 "k8s.io/api/core/v1"
//conf "github.com/goodrain/rainbond/cmd/grctl/option"
)
@ -507,16 +508,9 @@ func showServiceDeployInfo(c *cli.Context) error {
containerTable := termtables.CreateTable()
containerTable.AddHeaders("ID", "Name", "Image", "State")
for j := 0; j < len(pod.Status.ContainerStatuses); j++ {
var t string
con := pod.Status.ContainerStatuses[j]
if con.State.Running != nil {
t = con.State.Running.StartedAt.Format(time.RFC3339)
}
var conID string
if con.ContainerID != "" {
conID = con.ContainerID[9:21]
}
containerTable.AddRow(conID, con.Name, con.Image, t)
cstatus := pod.Status.ContainerStatuses[j]
cid, s := getContainerIDAndState(cstatus)
containerTable.AddRow(cid, cstatus.Name, cstatus.Image, s)
}
fmt.Println(containerTable.Render())
} else {
@ -529,6 +523,23 @@ func showServiceDeployInfo(c *cli.Context) error {
return nil
}
func getContainerIDAndState(status corev1.ContainerStatus) (cid, s string) {
state := status.State
containerID := status.ContainerID
if state.Running != nil {
s = "Running"
}
if state.Waiting != nil {
s = "Waiting"
}
if state.Terminated != nil {
s = "Terminated"
containerID = state.Terminated.ContainerID
}
cid = containerID[9:21]
return
}
func showTenantServices(ctx *cli.Context) error {
tenantAlias := ctx.String("tenantAlias")
if tenantAlias == "" {

2
node/core/envoy/v2/builder.go
View File

@ -205,6 +205,7 @@ func CreateCircuitBreaker(options RainbondPluginOptions) *cluster.CircuitBreaker
circuitBreakers := &cluster.CircuitBreakers{
Thresholds: []*cluster.CircuitBreakers_Thresholds{
&cluster.CircuitBreakers_Thresholds{
Priority: core.RoutingPriority_DEFAULT,
MaxConnections: ConversionUInt32(uint32(options.MaxConnections)),
MaxRequests: ConversionUInt32(uint32(options.MaxRequests)),
MaxRetries: ConversionUInt32(uint32(options.MaxActiveRetries)),
@ -270,6 +271,7 @@ func CreateRoute(clusterName, prefix string, headers []*route.HeaderMatcher, wei
},
},
},
Priority: core.RoutingPriority_DEFAULT,
},
},
}

2
node/nodem/envoy/conver/cds_conver.go
View File

@ -57,7 +57,7 @@ func OneNodeCluster(serviceAlias, namespace string, configs *corev1.ConfigMap, s
}
}
if len(clusters) == 0 {
logrus.Warnf("create clusters zero length for service %s", serviceAlias)
logrus.Warningf("configmap name: %s; plugin-config: %s; create clusters zero length", configs.Name, configs.Data["plugin-config"])
}
return clusters, nil
}

2
node/nodem/envoy/conver/lds_conver.go
View File

@ -74,7 +74,7 @@ func OneNodeListerner(serviceAlias, namespace string, configs *corev1.ConfigMap,
}
}
if len(listener) == 0 {
logrus.Warnf("create listener zero length for service %s", serviceAlias)
logrus.Warningf("configmap name: %s; plugin-config: %s; create listener zero length", configs.Name, configs.Data["plugin-config"])
}
return listener, nil
}

24
node/nodem/envoy/server_v2.go
View File

@ -25,22 +25,19 @@ import (
"sync"
"sync/atomic"
"k8s.io/apimachinery/pkg/labels"
"github.com/goodrain/rainbond/node/nodem/envoy/conver"
api_model "github.com/goodrain/rainbond/api/model"
corev1 "k8s.io/api/core/v1"
"github.com/Sirupsen/logrus"
v2 "github.com/envoyproxy/go-control-plane/envoy/api/v2"
"github.com/envoyproxy/go-control-plane/envoy/api/v2/core"
discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v2"
"github.com/envoyproxy/go-control-plane/pkg/cache"
"github.com/envoyproxy/go-control-plane/pkg/server"
api_model "github.com/goodrain/rainbond/api/model"
"github.com/goodrain/rainbond/cmd/node/option"
"github.com/goodrain/rainbond/node/kubecache"
"github.com/goodrain/rainbond/node/nodem/envoy/conver"
"google.golang.org/grpc"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/labels"
)
//DiscoverServerManager envoy discover server
@ -144,10 +141,11 @@ func createNodeID(namespace, pluginID, serviceAlias string) string {
//GetDependService get depend service
func (d *DiscoverServerManager) GetDependService(namespace, depServiceAlias string) ([]*corev1.Service, []*corev1.Endpoints) {
logrus.Debugf("dep service alias: %s", depServiceAlias)
labelname := fmt.Sprintf("name=%sService", depServiceAlias)
selector, err := labels.Parse(labelname)
if err != nil {
logrus.Errorf("parse label name failure %s", err.Error())
logrus.Errorf("label name: %s; parse label name failure %s", labelname, err.Error())
return nil, nil
}
services, err := d.kubecli.GetServices(namespace, selector)
@ -168,7 +166,7 @@ func (d *DiscoverServerManager) GetSelfService(namespace, serviceAlias string) (
labelname := fmt.Sprintf("name=%sServiceOUT", serviceAlias)
selector, err := labels.Parse(labelname)
if err != nil {
logrus.Errorf("parse label name failure %s", err.Error())
logrus.Errorf("label name: %s; parse label name failure %s", labelname, err.Error())
return nil, nil
}
services, err := d.kubecli.GetServices(namespace, selector)
@ -186,7 +184,8 @@ func (d *DiscoverServerManager) GetSelfService(namespace, serviceAlias string) (
//NewNodeConfig new NodeConfig
func (d *DiscoverServerManager) NewNodeConfig(config *corev1.ConfigMap) (*NodeConfig, error) {
servicaAlias := config.Labels["service_alias"]
logrus.Debugf("cm name: %s; plugin-config: %s", config.GetName(), config.Data["plugin-config"])
servicaAlias := config.Labels["service_alias"]
namespace := config.Namespace
configs, pluginID, err := conver.GetPluginConfigs(config)
if err != nil {
@ -232,6 +231,9 @@ func (d *DiscoverServerManager) UpdateNodeConfig(nc *NodeConfig) error {
nc.clusters = clusters
}
clusterLoadAssignment := conver.OneNodeClusterLoadAssignment(nc.serviceAlias, nc.namespace, endpoint, services)
if len(clusterLoadAssignment) == 0 {
logrus.Warningf("configmap name: %s; plugin-config: %s; empty clusterLoadAssignment", nc.config.Name, nc.config.Data["plugin-config"])
}
if err != nil {
logrus.Errorf("create envoy endpoints failure %s", err.Error())
} else {
@ -242,7 +244,7 @@ func (d *DiscoverServerManager) UpdateNodeConfig(nc *NodeConfig) error {
func (d *DiscoverServerManager) setSnapshot(nc *NodeConfig) error {
if len(nc.clusters) < 1 || len(nc.listeners) < 1 {
logrus.Warn("node config cluster length is zero or listener length is zero,not set snapshot")
logrus.Warningf("node id: %s; node config cluster length is zero or listener length is zero,not set snapshot", nc.GetID())
return nil
}
snapshot := cache.NewSnapshot(nc.GetVersion(), nc.endpoints, nc.clusters, nil, nc.listeners)

2
node/nodem/logger/manage.go
View File

@ -262,7 +262,7 @@ func (c *ContainerLogManage) watchContainer() error {
if !strings.Contains(err.Error(), "No such container") {
logrus.Errorf("get container detail info failure %s", err.Error())
}
container.ID = event.ID
break
}
c.cacheContainer(ContainerEvent{Action: event.Action, Container: container})
}

46
worker/appm/controller/stop.go
View File

@ -64,16 +64,18 @@ func (s *stopController) stopOne(app v1.AppService) error {
//step 1: delete services
if services := app.GetServices(); services != nil {
for _, service := range services {
err := s.manager.client.CoreV1().Services(app.TenantID).Delete(service.Name, &metav1.DeleteOptions{})
if err != nil && !errors.IsNotFound(err) {
return fmt.Errorf("delete service failure:%s", err.Error())
if service != nil && service.Name != "" {
err := s.manager.client.CoreV1().Services(app.TenantID).Delete(service.Name, &metav1.DeleteOptions{})
if err != nil && !errors.IsNotFound(err) {
return fmt.Errorf("delete service failure:%s", err.Error())
}
}
}
}
//step 2: delete secrets
if secrets := app.GetSecrets(); secrets != nil {
for _, secret := range secrets {
if secret != nil {
if secret != nil && secret.Name != "" {
err := s.manager.client.CoreV1().Secrets(app.TenantID).Delete(secret.Name, &metav1.DeleteOptions{})
if err != nil && !errors.IsNotFound(err) {
return fmt.Errorf("delete secret failure:%s", err.Error())
@ -85,21 +87,25 @@ func (s *stopController) stopOne(app v1.AppService) error {
//step 3: delete ingress
if ingresses := app.GetIngress(); ingresses != nil {
for _, ingress := range ingresses {
err := s.manager.client.ExtensionsV1beta1().Ingresses(app.TenantID).Delete(ingress.Name, &metav1.DeleteOptions{})
if err != nil && !errors.IsNotFound(err) {
return fmt.Errorf("delete ingress failure:%s", err.Error())
if ingress != nil && ingress.Name != "" {
err := s.manager.client.ExtensionsV1beta1().Ingresses(app.TenantID).Delete(ingress.Name, &metav1.DeleteOptions{})
if err != nil && !errors.IsNotFound(err) {
return fmt.Errorf("delete ingress failure:%s", err.Error())
}
s.manager.store.OnDelete(ingress)
}
s.manager.store.OnDelete(ingress)
}
}
//step 4: delete configmap
if configs := app.GetConfigMaps(); configs != nil {
for _, config := range configs {
err := s.manager.client.CoreV1().ConfigMaps(app.TenantID).Delete(config.Name, &metav1.DeleteOptions{})
if err != nil && !errors.IsNotFound(err) {
return fmt.Errorf("delete config map failure:%s", err.Error())
if config != nil && config.Name != "" {
err := s.manager.client.CoreV1().ConfigMaps(app.TenantID).Delete(config.Name, &metav1.DeleteOptions{})
if err != nil && !errors.IsNotFound(err) {
return fmt.Errorf("delete config map failure:%s", err.Error())
}
s.manager.store.OnDelete(config)
}
s.manager.store.OnDelete(config)
}
}
//step 5: delete statefulset or deployment
@ -110,7 +116,7 @@ func (s *stopController) stopOne(app v1.AppService) error {
}
s.manager.store.OnDelete(statefulset)
}
if deployment := app.GetDeployment(); deployment != nil {
if deployment := app.GetDeployment(); deployment != nil && deployment.Name != "" {
err := s.manager.client.AppsV1().Deployments(app.TenantID).Delete(deployment.Name, &metav1.DeleteOptions{})
if err != nil && !errors.IsNotFound(err) {
return fmt.Errorf("delete deployment failure:%s", err.Error())
@ -121,13 +127,15 @@ func (s *stopController) stopOne(app v1.AppService) error {
var gracePeriodSeconds int64
if pods := app.GetPods(); pods != nil {
for _, pod := range pods {
err := s.manager.client.CoreV1().Pods(app.TenantID).Delete(pod.Name, &metav1.DeleteOptions{
GracePeriodSeconds: &gracePeriodSeconds,
})
if err != nil && !errors.IsNotFound(err) {
return fmt.Errorf("delete pod failure:%s", err.Error())
if pod != nil && pod.Name != "" {
err := s.manager.client.CoreV1().Pods(app.TenantID).Delete(pod.Name, &metav1.DeleteOptions{
GracePeriodSeconds: &gracePeriodSeconds,
})
if err != nil && !errors.IsNotFound(err) {
return fmt.Errorf("delete pod failure:%s", err.Error())
}
s.manager.store.OnDelete(pod)
}
s.manager.store.OnDelete(pod)
}
}
//step 7: waiting endpoint ready

8
worker/appm/conversion/gateway.go
View File

@ -147,10 +147,10 @@ func (a *AppServiceBuild) Build() (*v1.K8sResources, error) {
if ports != nil && len(ports) > 0 {
for i := range ports {
port := ports[i]
if port.IsInnerService {
if *port.IsInnerService {
services = append(services, a.createInnerService(port))
}
if port.IsOuterService {
if *port.IsOuterService {
service := a.createOuterService(port)
services = append(services, service)
relContainerPort := pp[int32(port.ContainerPort)]
@ -427,10 +427,10 @@ func (a *AppServiceBuild) BuildOnPort(p int, isOut bool) (*corev1.Service, error
return nil, fmt.Errorf("find service port from db error %s", err.Error())
}
if port != nil {
if !isOut && port.IsInnerService {
if !isOut && *port.IsInnerService {
return a.createInnerService(port), nil
}
if isOut && port.IsOuterService {
if isOut && *port.IsOuterService {
return a.createOuterService(port), nil
}
}

24
worker/appm/conversion/gateway_test.go
View File

@ -20,23 +20,21 @@ package conversion
import (
"fmt"
"github.com/goodrain/rainbond/gateway/annotations/parser"
"strconv"
"testing"
"time"
"github.com/goodrain/rainbond/db"
"github.com/goodrain/rainbond/db/dao"
"github.com/goodrain/rainbond/db/model"
"github.com/goodrain/rainbond/gateway/annotations/parser"
v1 "github.com/goodrain/rainbond/worker/appm/types/v1"
"github.com/rafrombrc/gomock/gomock"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/util/intstr"
"k8s.io/client-go/tools/clientcmd"
"strconv"
"testing"
"time"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/tools/clientcmd"
)
const (
@ -114,8 +112,8 @@ func TestApplyTcpRule(t *testing.T) {
ContainerPort: containerPort,
Protocol: "http",
PortAlias: "GRD835895000",
IsInnerService: false,
IsOuterService: true,
IsInnerService: func() *bool { b := false; return &b }(),
IsOuterService: func() *bool { b := true; return &b }(),
}
service := &corev1.Service{
@ -337,8 +335,8 @@ func TestAppServiceBuild_ApplyHttpRule(t *testing.T) {
ContainerPort: containerPort,
MappingPort: containerPort,
Protocol: "http",
IsInnerService: false,
IsOuterService: true,
IsInnerService: func() *bool { b := false; return &b }(),
IsOuterService: func() *bool { b := true; return &b }(),
}
service := &corev1.Service{
@ -499,8 +497,8 @@ func TestAppServiceBuild_ApplyHttpRuleWithCertificate(t *testing.T) {
ContainerPort: containerPort,
MappingPort: containerPort,
Protocol: "http",
IsInnerService: false,
IsOuterService: true,
IsInnerService: func() *bool { b := false; return &b }(),
IsOuterService: func() *bool { b := true; return &b }(),
}
service := &corev1.Service{

4
worker/appm/conversion/version.go
View File

@ -249,7 +249,7 @@ func createEnv(as *v1.AppService, dbmanager db.Manager) (*[]corev1.EnvVar, error
portStr += ":"
}
portStr += fmt.Sprintf("%d", port.ContainerPort)
if port.IsOuterService && (port.Protocol == "http" || port.Protocol == "https") {
if *port.IsOuterService && (port.Protocol == "http" || port.Protocol == "https") {
envs = append(envs, corev1.EnvVar{Name: "DEFAULT_DOMAIN", Value: createDefaultDomain(as.TenantName, as.ServiceAlias, port.ContainerPort)})
}
renvs := convertRulesToEnvs(as, dbmanager, port, true)
@ -311,7 +311,7 @@ func createEnv(as *v1.AppService, dbmanager db.Manager) (*[]corev1.EnvVar, error
}
func convertRulesToEnvs(as *v1.AppService, dbmanager db.Manager, port *dbmodel.TenantServicesPort, usePort bool) []corev1.EnvVar {
if !port.IsOuterService {
if !*port.IsOuterService {
return nil
}

6
worker/appm/conversion/version_test.go
View File

@ -22,10 +22,10 @@ import (
"testing"
"github.com/goodrain/rainbond/db"
"github.com/goodrain/rainbond/db/dao"
"github.com/goodrain/rainbond/db/model"
v1 "github.com/goodrain/rainbond/worker/appm/types/v1"
"github.com/rafrombrc/gomock/gomock"
"github.com/goodrain/rainbond/db/dao"
)
func TestTenantServiceVersion(t *testing.T) {
@ -53,8 +53,8 @@ func TestConvertRulesToEnvs(t *testing.T) {
ContainerPort: 0,
Protocol: "http",
PortAlias: "GRD835895000",
IsInnerService: false,
IsOuterService: true,
IsInnerService: func() *bool { b := false; return &b }(),
IsOuterService: func() *bool { b := true; return &b }(),
}
renvs := convertRulesToEnvs(as, dbmanager, port, true)

8
worker/appm/thirdparty/thirdparty.go vendored
View File

@ -201,11 +201,11 @@ func (t *thirdparty) k8sEndpoints(as *v1.AppService, epinfo []*v1.RbdEndpoint) (
p := ports[0]
var res []*corev1.Endpoints
if p.IsInnerService {
if *p.IsInnerService {
ep := &corev1.Endpoints{}
ep.Namespace = as.TenantID
// inner or outer
if p.IsInnerService {
if *p.IsInnerService {
ep.Name = fmt.Sprintf("service-%d-%d", p.ID, p.ContainerPort)
ep.Labels = as.GetCommonLabels(map[string]string{
"name": as.ServiceAlias + "Service",
@ -214,11 +214,11 @@ func (t *thirdparty) k8sEndpoints(as *v1.AppService, epinfo []*v1.RbdEndpoint) (
}
res = append(res, ep)
}
if p.IsOuterService {
if *p.IsOuterService {
ep := &corev1.Endpoints{}
ep.Namespace = as.TenantID
// inner or outer
if p.IsOuterService {
if *p.IsOuterService {
ep.Name = fmt.Sprintf("service-%d-%dout", p.ID, p.ContainerPort)
ep.Labels = as.GetCommonLabels(map[string]string{
"name": as.ServiceAlias + "ServiceOUT",