Merge branch 'V5.2'

.gitignore

@@ -58,4 +58,5 @@ vendor
 
 .scannerwork
 
-Library
+Library
+.cache

api/handler/gateway_action.go

@@ -643,6 +643,16 @@ func (g *GatewayAction) RuleConfig(req *apimodel.RuleConfigReq) error {
 		Key:    "proxy-body-size",
 		Value:  strconv.Itoa(req.Body.ProxyBodySize),
 	})
+	configs = append(configs, &model.GwRuleConfig{
+		RuleID: req.RuleID,
+		Key:    "proxy-buffer-size",
+		Value:  strconv.Itoa(req.Body.ProxyBufferSize) + "k",
+	})
+	configs = append(configs, &model.GwRuleConfig{
+		RuleID: req.RuleID,
+		Key:    "proxy-buffer-numbers",
+		Value:  strconv.Itoa(req.Body.ProxyBufferNumbers),
+	})
 	setheaders := make(map[string]string)
 	for _, item := range req.Body.SetHeaders {
 		if strings.TrimSpace(item.Key) == "" {
@@ -661,10 +671,12 @@ func (g *GatewayAction) RuleConfig(req *apimodel.RuleConfigReq) error {
 			Value:  v,
 		})
 	}
+
 	rule, err := g.dbmanager.HTTPRuleDao().GetHTTPRuleByID(req.RuleID)
 	if err != nil {
 		return err
 	}
+
 	tx := db.GetManager().Begin()
 	defer func() {
 		if r := recover(); r != nil {
@@ -686,6 +698,7 @@ func (g *GatewayAction) RuleConfig(req *apimodel.RuleConfigReq) error {
 		tx.Rollback()
 		return err
 	}
+
 	if err := g.SendTask(map[string]interface{}{
 		"service_id": req.ServiceID,
 		"action":     "update-rule-config",

api/handler/service_batch_operation.go

@@ -312,18 +312,19 @@ func (s *ServiceDependency) buildLinkListByHead(l *list.List) []*list.List {
 	if !ok {
 		copy := list.New()
 		copy.PushBackList(l)
-		return []*list.List{l}
+		return []*list.List{copy}
 	}
 
 	var result []*list.List
 	for _, depsid := range depsids {
 		// child node is already in the linked list
-		if alreadyInLinkedList(l, depsid) {
+		if alreadyInLinkedList(l, depsid) || s.childInLinkedList(l, depsid) {
 			copy := list.New()
 			copy.PushBackList(l)
 			result = append(result, copy)
 			continue
 		}
+
 		newl := list.New()
 		newl.PushBackList(l)
 		newl.PushBack(depsid)
@@ -341,6 +342,21 @@ func (s *ServiceDependency) buildLinkListByHead(l *list.List) []*list.List {
 	return result
 }
 
+func (s *ServiceDependency) childInLinkedList(l *list.List, sid string) bool {
+	depsids, ok := s.sid2depsids[sid]
+	if !ok {
+		return false
+	}
+
+	for _, depsid := range depsids {
+		if alreadyInLinkedList(l, depsid) {
+			return true
+		}
+	}
+
+	return false
+}
+
 func alreadyInLinkedList(l *list.List, depsid string) bool {
 	pre := l.Back()
 	for pre != nil {

api/handler/service_batch_operation_test.go

@@ -167,6 +167,32 @@ func TestBuildLinkListByHead(t *testing.T) {
 				return []*list.List{l}
 			}(),
 		},
+		{
+			name: "spring cloud pig",
+			l: func() *list.List {
+				l := list.New()
+				l.PushBack("banana")
+				return l
+			}(),
+			sid2depsids: map[string][]string{
+				"apple":  []string{"banana"},
+				"banana": []string{"apple", "cat", "dog"},
+			},
+			want: func() []*list.List {
+				l1 := list.New()
+				l1.PushBack("banana")
+
+				l2 := list.New()
+				l2.PushBack("banana")
+				l2.PushBack("cat")
+
+				l3 := list.New()
+				l3.PushBack("banana")
+				l3.PushBack("dog")
+
+				return []*list.List{l1, l2, l3}
+			}(),
+		},
 	}
 
 	for idx := range tests {

api/handler/tenantsResource.go

@@ -1,9 +1,9 @@
 package handler
 
 import (
-	"github.com/sirupsen/logrus"
 	dbmodel "github.com/goodrain/rainbond/db/model"
 	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
 )
 
 // CheckTenantResource check tenant's resource is support action or not
@@ -22,7 +22,7 @@ func CheckTenantResource(tenant *dbmodel.Tenants, needMemory int) error {
 	}
 	clusterInfo, err := GetTenantManager().GetAllocatableResources()
 	if err != nil {
-		logrus.Errorf("get cluster resources failure for check tenant resource : %v.", err.Error())
+		logrus.Errorf("get cluster resources failure for check tenant resource: %v", err.Error())
 	}
 	if clusterInfo != nil {
 		clusterAvailMemory := clusterInfo.AllMemory - clusterInfo.RequestMemory

api/model/gateway_model.go

@@ -129,6 +129,8 @@ type Body struct {
 	ProxyBodySize       int          `json:"proxy_body_size,omitempty" validate:"proxy_body_size|required"`
 	SetHeaders          []*SetHeader `json:"set_headers,omitempty" `
 	Rewrites            []*Rewrite   `json:"rewrite,omitempty"`
+	ProxyBufferSize     int          `json:"proxy_buffer_size,omitempty"`
+	ProxyBufferNumbers  int          `json:"proxy_buffer_numbers,omitempty"`
 }
 
 //SetHeader set header

builder/parser/docker_compose.go

@@ -195,7 +195,7 @@ func (d *DockerComposeParse) Parse() ParseErrorList {
 		d.logger.Debug(fmt.Sprintf("start check service %s ", service.name), map[string]string{"step": "service_check", "status": "running"})
 		exist, err := sources.ImageExist(service.image.String(), hubUser, hubPass)
 		if err != nil {
-			logrus.Errorf("check image exist failure %s", err.Error())
+			logrus.Errorf("check image(%s) exist failure %s", service.image.String(), err.Error())
 		}
 		if !exist {
 			d.errappend(ErrorAndSolve(FatalError, fmt.Sprintf("服务%s镜像%s检测失败", serviceName, service.image.String()), SolveAdvice("modify_compose", fmt.Sprintf("请确认%s服务镜像名称是否正确或镜像仓库访问是否正常", serviceName))))

builder/sources/registry.go

@@ -19,10 +19,12 @@
 package sources
 
 import (
+	"fmt"
+
 	"github.com/goodrain/rainbond/builder/sources/registry"
 
-	"github.com/sirupsen/logrus"
 	"github.com/docker/distribution/reference"
+	"github.com/sirupsen/logrus"
 )
 
 //GetTagFromNamedRef get image tag by name
@@ -67,9 +69,8 @@ func ImageExist(imageName, user, password string) (bool, error) {
 			}
 		}
 		tag := GetTagFromNamedRef(name)
-		_, err = reg.ManifestV2(reference.Path(name), tag)
+		if err := reg.CheckManifest(reference.Path(name), tag); err != nil {
-		if err != nil {
+			rerr = fmt.Errorf("[ImageExist] check manifest v2: %v", err)
-			rerr = err
 			continue
 		}
 		return true, nil

builder/sources/registry/manifest.go

@@ -20,14 +20,17 @@ package registry
 
 import (
 	"bytes"
+	"fmt"
 	"io/ioutil"
 	"net/http"
 
+	"github.com/Sirupsen/logrus"
 	manifestV1 "github.com/docker/distribution/manifest/schema1"
 	manifestV2 "github.com/docker/distribution/manifest/schema2"
 	digest "github.com/opencontainers/go-digest"
 )
 
+// Manifest -
 func (registry *Registry) Manifest(repository, reference string) (*manifestV1.SignedManifest, error) {
 	url := registry.url("/v2/%s/manifests/%s", repository, reference)
 	registry.Logf("registry.manifest.get url=%s repository=%s reference=%s", url, repository, reference)
@@ -58,9 +61,10 @@ func (registry *Registry) Manifest(repository, reference string) (*manifestV1.Si
 	return signedManifest, nil
 }
 
+// ManifestV2 -
 func (registry *Registry) ManifestV2(repository, reference string) (*manifestV2.DeserializedManifest, error) {
 	url := registry.url("/v2/%s/manifests/%s", repository, reference)
-	registry.Logf("registry.manifest.get url=%s repository=%s reference=%s", url, repository, reference)
+	logrus.Debugf("registry.manifest.get url=%s repository=%s reference=%s", url, repository, reference)
 
 	req, err := http.NewRequest("GET", url, nil)
 	if err != nil {
@@ -70,7 +74,7 @@ func (registry *Registry) ManifestV2(repository, reference string) (*manifestV2.
 	req.Header.Set("Accept", manifestV2.MediaTypeManifest)
 	resp, err := registry.Client.Do(req)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("do request: %v", err)
 	}
 
 	defer resp.Body.Close()
@@ -82,11 +86,25 @@ func (registry *Registry) ManifestV2(repository, reference string) (*manifestV2.
 	deserialized := &manifestV2.DeserializedManifest{}
 	err = deserialized.UnmarshalJSON(body)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("unmarshal JSON: %v", err)
 	}
 	return deserialized, nil
 }
 
+// CheckManifest checks if the manifest of the given image is exist.
+func (registry *Registry) CheckManifest(repository, reference string) error {
+	url := registry.url("/v2/%s/manifests/%s", repository, reference)
+
+	req, err := http.NewRequest("GET", url, nil)
+	if err != nil {
+		return err
+	}
+
+	_, err = registry.Client.Do(req)
+	return err
+}
+
+// ManifestDigest -
 func (registry *Registry) ManifestDigest(repository, reference string) (digest.Digest, error) {
 	url := registry.url("/v2/%s/manifests/%s", repository, reference)
 	registry.Logf("registry.manifest.head url=%s repository=%s reference=%s", url, repository, reference)
@@ -101,6 +119,7 @@ func (registry *Registry) ManifestDigest(repository, reference string) (digest.D
 	return digest.Parse(resp.Header.Get("Docker-Content-Digest"))
 }
 
+// DeleteManifest -
 func (registry *Registry) DeleteManifest(repository string, digest digest.Digest) error {
 	url := registry.url("/v2/%s/manifests/%s", repository, digest)
 	registry.Logf("registry.manifest.delete url=%s repository=%s reference=%s", url, repository, digest)
@@ -119,6 +138,7 @@ func (registry *Registry) DeleteManifest(repository string, digest digest.Digest
 	return nil
 }
 
+// PutManifest -
 func (registry *Registry) PutManifest(repository, reference string, signedManifest *manifestV1.SignedManifest) error {
 	url := registry.url("/v2/%s/manifests/%s", repository, reference)
 	registry.Logf("registry.manifest.put url=%s repository=%s reference=%s", url, repository, reference)

cmd/init-probe/healthy/depends_health.go

@@ -78,12 +78,7 @@ func NewDependServiceHealthController() (*DependServiceHealthController, error)
 	}
 	dsc.endpointClient = v2.NewEndpointDiscoveryServiceClient(cli)
 	dsc.clusterClient = v2.NewClusterDiscoveryServiceClient(cli)
-	nameIDs := strings.Split(os.Getenv("DEPEND_SERVICE"), ",")
+	dsc.dependServiceNames = strings.Split(os.Getenv("STARTUP_SEQUENCE_DEPENDENCIES"), ",")
-	for _, nameID := range nameIDs {
-		if len(strings.Split(nameID, ":")) > 0 {
-			dsc.dependServiceNames = append(dsc.dependServiceNames, strings.Split(nameID, ":")[0])
-		}
-	}
 	return &dsc, nil
 }
 

discover.v2/k8s_discover.go

@@ -3,6 +3,7 @@ package discover
 import (
 	"context"
 	"errors"
+	"reflect"
 	"sync"
 	"time"
 
@@ -90,18 +91,24 @@ func (k *k8sDiscover) discover(name string, callback CallbackUpdate) {
 		DeleteFunc: func(obj interface{}) {
 			pod := obj.(*corev1.Pod)
 			ep := endpointForPod(pod)
+			ep.Mode = 2
 			callback.UpdateEndpoints(config.DELETE, ep)
 		},
 		UpdateFunc: func(old, cur interface{}) {
 			oldPod := old.(*corev1.Pod)
 			curPod := cur.(*corev1.Pod)
-			if oldPod.Status.Phase == curPod.Status.Phase {
+
-				return
+			if reflect.DeepEqual(oldPod, curPod) {
-			}
-			if !isPodReady(curPod) {
 				return
 			}
+
 			ep := endpointForPod(curPod)
+			if !isPodReady(curPod) {
+				logrus.Infof("unready pod(%s%s) received, delete endpoint based on the pod", curPod.Name, curPod.Namespace)
+				ep.Mode = 2
+				callback.UpdateEndpoints(config.DELETE, ep)
+				return
+			}
 			callback.UpdateEndpoints(config.SYNC, ep)
 		},
 	}

gateway/store/store.go

@@ -690,7 +690,6 @@ func (s *k8sStore) ListVirtualService() (l7vs []*v1.VirtualService, l4vs []*v1.V
 
 // ingressIsValid checks if the specified ingress is valid
 func (s *k8sStore) ingressIsValid(ing *extensions.Ingress) bool {
-
 	var endpointKey string
 	if ing.Spec.Backend != nil { // stream
 		endpointKey = fmt.Sprintf("%s/%s", ing.Namespace, ing.Spec.Backend.ServiceName)
@@ -719,20 +718,27 @@ func (s *k8sStore) ingressIsValid(ing *extensions.Ingress) bool {
 		logrus.Errorf("Cant not convert %v to %v", reflect.TypeOf(item), reflect.TypeOf(endpoint))
 		return false
 	}
-	if endpoint.Subsets == nil || len(endpoint.Subsets) == 0 {
+	if len(endpoint.Subsets) == 0 {
 		logrus.Debugf("Endpoints(%s) is empty, ignore it", endpointKey)
 		return false
 	}
-	for _, ep := range endpoint.Subsets {
+	if !hasReadyAddresses(endpoint) {
-		if (ep.Addresses == nil || len(ep.Addresses) == 0) && (ep.NotReadyAddresses == nil || len(ep.NotReadyAddresses) == 0) {
+		logrus.Debugf("Endpoints(%s) is empty, ignore it", endpointKey)
-			logrus.Debugf("Endpoints(%s) is empty, ignore it", endpointKey)
+		return false
-			return false
-		}
 	}
 
 	return true
 }
 
+func hasReadyAddresses(endpoints *corev1.Endpoints) bool {
+	for _, ep := range endpoints.Subsets {
+		if len(ep.Addresses) > 0 {
+			return true
+		}
+	}
+	return false
+}
+
 // GetIngress returns the Ingress matching key.
 func (s *k8sStore) GetIngress(key string) (*extensions.Ingress, error) {
 	return s.listers.Ingress.ByKey(key)

release.sh

@@ -4,12 +4,8 @@ set -o errexit
 # define package name
 WORK_DIR=/go/src/github.com/goodrain/rainbond
 BASE_NAME=rainbond
-IMAGE_BASE_NAME=rainbond
+IMAGE_BASE_NAME=${BUILD_IMAGE_BASE_NAME:-'rainbond'}
-if [ "$BUILD_IMAGE_BASE_NAME" ]; 
+
-then 
-IMAGE_BASE_NAME=${BUILD_IMAGE_BASE_NAME}
-fi
-CACHE=${CACHE:true}
 GO_VERSION=1.13
 
 GOPROXY=${GOPROXY:-'https://goproxy.io'}

worker/appm/conversion/version.go

@@ -192,6 +192,9 @@ func createEnv(as *v1.AppService, dbmanager db.Manager) (*[]corev1.EnvVar, error
 		Value: "streamlog",
 	})
 
+	bootSeqDepServiceIDs := as.ExtensionSet["boot_seq_dep_service_ids"]
+	logrus.Infof("boot sequence dep service ids: %s", bootSeqDepServiceIDs)
+
 	//set relation app outer env
 	var relationIDs []string
 	relations, err := dbmanager.TenantServiceRelationDao().GetTenantServiceRelations(as.ServiceID)
@@ -216,15 +219,21 @@ func createEnv(as *v1.AppService, dbmanager db.Manager) (*[]corev1.EnvVar, error
 		if err != nil {
 			return nil, err
 		}
-		var depend string
+		var Depend string
+		var startupSequenceDependencies []string
 		for _, sa := range serviceAliases {
-			if depend != "" {
+			if Depend != "" {
-				depend += ","
+				Depend += ","
+			}
+			Depend += fmt.Sprintf("%s:%s", sa.ServiceAlias, sa.ServiceID)
+
+			if bootSeqDepServiceIDs != "" && strings.Contains(bootSeqDepServiceIDs, sa.ServiceID) {
+				startupSequenceDependencies = append(startupSequenceDependencies, sa.ServiceAlias)
 			}
-			depend += fmt.Sprintf("%s:%s", sa.ServiceAlias, sa.ServiceID)
 		}
-		envs = append(envs, corev1.EnvVar{Name: "DEPEND_SERVICE", Value: depend})
+		envs = append(envs, corev1.EnvVar{Name: "DEPEND_SERVICE", Value: Depend})
 		envs = append(envs, corev1.EnvVar{Name: "DEPEND_SERVICE_COUNT", Value: strconv.Itoa(len(serviceAliases))})
+		envs = append(envs, corev1.EnvVar{Name: "STARTUP_SEQUENCE_DEPENDENCIES", Value: strings.Join(startupSequenceDependencies, ",")})
 
 		if as.GovernanceMode == model.GovernanceModeBuildInServiceMesh {
 			as.NeedProxy = true