Merge pull request mosn#68 from alipay/add_crosscompile

Add crosscompile && fix  Newcluster && fix decode-error

Makefile

@@ -67,6 +67,29 @@ build-local:
 	@cd build/bundles/${MAJOR_VERSION}/binary && $(shell which md5sum) -b ${TARGET} | cut -d' ' -f1  > ${TARGET}.md5
 	cp configs/${CONFIG_FILE} build/bundles/${MAJOR_VERSION}/binary
 
+build-linux32:
+	@rm -rf build/bundles/${MAJOR_VERSION}/binary
+	CGO_ENABLED=0 env GOOS=linux GOARCH=386 go build\
+		-ldflags "-B 0x$(shell head -c20 /dev/urandom|od -An -tx1|tr -d ' \n') -X main.Version=${MAJOR_VERSION}(${GIT_VERSION})" \
+		-v -o ${TARGET} \
+		${PROJECT_NAME}/cmd/mosn/main
+	mkdir -p build/bundles/${MAJOR_VERSION}/binary
+	mv ${TARGET} build/bundles/${MAJOR_VERSION}/binary
+	@cd build/bundles/${MAJOR_VERSION}/binary && $(shell which md5sum) -b ${TARGET} | cut -d' ' -f1  > ${TARGET}.md5
+	cp configs/${CONFIG_FILE} build/bundles/${MAJOR_VERSION}/binary
+
+build-linux64:
+	@rm -rf build/bundles/${MAJOR_VERSION}/binary
+	CGO_ENABLED=0 env GOOS=linux GOARCH=amd64 go build\
+		-ldflags "-B 0x$(shell head -c20 /dev/urandom|od -An -tx1|tr -d ' \n') -X main.Version=${MAJOR_VERSION}(${GIT_VERSION})" \
+		-v -o ${TARGET} \
+		${PROJECT_NAME}/cmd/mosn/main
+	mkdir -p build/bundles/${MAJOR_VERSION}/binary
+	mv ${TARGET} build/bundles/${MAJOR_VERSION}/binary
+	@cd build/bundles/${MAJOR_VERSION}/binary && $(shell which md5sum) -b ${TARGET} | cut -d' ' -f1  > ${TARGET}.md5
+	cp configs/${CONFIG_FILE} build/bundles/${MAJOR_VERSION}/binary
+
+
 image:
 	@rm -rf IMAGEBUILD
 	cp -r build/contrib/builder/image IMAGEBUILD && cp build/bundles/${MAJOR_VERSION}/binary/${TARGET} IMAGEBUILD && cp -r resource IMAGEBUILD && cp -r etc IMAGEBUILD

docs/CONTRIBUTING.md

@@ -5,7 +5,7 @@ standard Github development process, using Github tracker for issues and
 merging pull requests into master . If you would like to contribute something, 
 or simply want to hack on the code this document should help you get started.
 
-### Sign the Contributor License Agreement
+### [Sign the Contributor License Agreement](https://www.clahub.com/agreements/alipay/sofa-mosn)
 Before we accept a non-trivial patch or pull request we will need you to 
 sign the Contributor License Agreement. Signing the contributor’s agreement 
 does not grant anyone commit rights to the main repository, but it does mean 

docs/develop/quickstart.md

@@ -40,14 +40,26 @@ cd sofa-mosn
 
 ## 编译代码
 
-在项目根目录下执行如下命令编译 MOSN 的二进制文件：
+在项目根目录下，根据自己机器的类型以及欲执行二进制的环境，选择以下命令编译 MOSN 的二进制文件：
++ 运行 `dep ensure` 保证所有依赖均OK，运行较慢，耐心等待
++ 使用 docker 镜像编译 
 
 ```bash
-dep ensure	   // dep速度较慢，耐心等待
-make build         //使用docker编译
-// or
-make build-local   // 使用本地的go编译环境
+    make build // 编译出 linux 64bit 可运行二进制文件
 ```
++ 非 docker，本地编译
+    + 编译本地可运行二进制文件
+    ```bash
+        make build-local
+    ```
+    + 非 linux 机器交叉编译 linux 64bit 可运行二进制文件
+    ```bash
+        make build-linux64
+    ```
+    + 非 linux 机器交叉编译 linux 32bit 可运行二进制文件
+    ```bash
+        make build-linux32
+    ```
 
 完成后可以在 `build/bundles/${version}/binary` 目录下找到编译好的二进制文件。
 

pkg/protocol/sofarpc/codec/boltv1codec.go

@@ -29,6 +29,7 @@ import (
 	"github.com/alipay/sofa-mosn/pkg/protocol/serialize"
 	"github.com/alipay/sofa-mosn/pkg/protocol/sofarpc"
 	"github.com/alipay/sofa-mosn/pkg/types"
+	"fmt"
 )
 
 // BoltV1PropertyHeaders map the cmdkey and its data type
@@ -273,7 +274,7 @@ func (c *boltV1Codec) mapToCmd(headers map[string]string) interface{} {
 	return nil
 }
 
-func (c *boltV1Codec) Decode(context context.Context, data types.IoBuffer) (int, interface{}) {
+func (c *boltV1Codec) Decode(context context.Context, data types.IoBuffer) (interface{},error) {
 	readableBytes := data.Len()
 	read := 0
 	var cmd interface{}
@@ -322,7 +323,7 @@ func (c *boltV1Codec) Decode(context context.Context, data types.IoBuffer) (int,
 				} else { // not enough data
 
 					logger.Debugf("BoltV1 DECODE Request, no enough data for fully decode")
-					return read, nil
+					return cmd, nil
 				}
 
 				request := sofarpc.BoltRequestCommand{
@@ -347,7 +348,7 @@ func (c *boltV1Codec) Decode(context context.Context, data types.IoBuffer) (int,
 					request.Protocol, request.CmdType, request.CmdCode, request.ReqID)
 				cmd = &request
 			}
-		} else {
+		} else if dataType == sofarpc.RESPONSE{
 			//2. response
 			if readableBytes >= sofarpc.RESPONSE_HEADER_LEN_V1 {
 
@@ -382,7 +383,7 @@ func (c *boltV1Codec) Decode(context context.Context, data types.IoBuffer) (int,
 					// not enough data
 					logger.Debugf("BoltV1 DECODE RESPONSE: no enough data for fully decode")
 
-					return read, nil
+					return cmd, nil
 				}
 
 				response := sofarpc.BoltResponseCommand{
@@ -411,8 +412,11 @@ func (c *boltV1Codec) Decode(context context.Context, data types.IoBuffer) (int,
 					response.ResponseStatus, response.Protocol, response.CmdType, response.CmdCode, response.ReqID)
 				cmd = &response
 			}
+		} else {
+			// 3. unknown type error
+			return nil, fmt.Errorf("Decode Error, type = %s, value = %d", sofarpc.UnKnownReqtype, dataType)
 		}
 	}
 
-	return read, cmd
+	return cmd,nil
 }

pkg/protocol/sofarpc/codec/boltv2codec.go

@@ -28,6 +28,7 @@ import (
 	"github.com/alipay/sofa-mosn/pkg/network/buffer"
 	"github.com/alipay/sofa-mosn/pkg/protocol/sofarpc"
 	"github.com/alipay/sofa-mosn/pkg/types"
+	"fmt"
 )
 
 // types.Encoder & types.Decoder
@@ -143,7 +144,7 @@ func (c *boltV2Codec) mapToCmd(headers map[string]string) interface{} {
 	return nil
 }
 
-func (c *boltV2Codec) Decode(context context.Context, data types.IoBuffer) (int, interface{}) {
+func (c *boltV2Codec) Decode(context context.Context, data types.IoBuffer) (interface{},error) {
 	readableBytes := data.Len()
 	read := 0
 	var cmd interface{}
@@ -191,7 +192,7 @@ func (c *boltV2Codec) Decode(context context.Context, data types.IoBuffer) (int,
 					data.Drain(read)
 				} else { // not enough data
 					logger.Debugf("[BOLTV2 Decoder]no enough data for fully decode")
-					return read, nil
+					return cmd, nil
 				}
 
 				request := &sofarpc.BoltV2RequestCommand{
@@ -220,7 +221,7 @@ func (c *boltV2Codec) Decode(context context.Context, data types.IoBuffer) (int,
 
 				cmd = request
 			}
-		} else {
+		} else if dataType == sofarpc.RESPONSE {
 			//2. resposne
 			if readableBytes >= sofarpc.RESPONSE_HEADER_LEN_V2 {
 
@@ -253,7 +254,7 @@ func (c *boltV2Codec) Decode(context context.Context, data types.IoBuffer) (int,
 					}
 				} else { // not enough data
 					logger.Debugf("[BOLTBV2 Decoder]no enough data for fully decode")
-					return read, nil
+					return cmd, nil
 				}
 
 				response := &sofarpc.BoltV2ResponseCommand{
@@ -283,10 +284,14 @@ func (c *boltV2Codec) Decode(context context.Context, data types.IoBuffer) (int,
 				logger.Debugf("[Decoder]bolt v2 decode response:%+v\n", response)
 				cmd = response
 			}
+		} else {
+			// 3. unknown type error
+			return nil, fmt.Errorf("Decode Error, type = %s, value = %d", sofarpc.UnKnownReqtype, dataType)
+
 		}
 	}
 
-	return read, cmd
+	return cmd,nil
 }
 
 func (c *boltV2Codec) insertToBytes(slice []byte, idx int, b byte) {

pkg/protocol/sofarpc/protocols.go

@@ -101,16 +101,14 @@ func (p *protocols) Decode(context context.Context, data types.IoBuffer, filter
 		logger.Debugf("Decoderprotocol code = %x, maybeProtocolVersion = %x", protocolCode, maybeProtocolVersion)
 
 		if proto, exists := p.protocolMaps[protocolCode]; exists {
-			if read, cmd := proto.GetDecoder().Decode(context, data); cmd != nil {
+			if cmd,error := proto.GetDecoder().Decode(context, data); cmd != nil && error == nil {
 				if err := proto.GetCommandHandler().HandleCommand(context, cmd, filter); err != nil {
 					filter.OnDecodeError(err, nil)
 					break
 				}
-			} else if 0 == read {
-				// protocol type error
-				errMsg := UnKnownReqtype
-				logger.Errorf(errMsg+" "+"CmdCode = %+v", cmd)
-				filter.OnDecodeError(errors.New(errMsg), nil)
+			} else if error != nil {
+				// request type error, the second byte in protocol
+				filter.OnDecodeError(error, nil)
 				break
 			} else {
 				break

pkg/server/handler.go

@@ -32,6 +32,7 @@ import (
 	"github.com/alipay/sofa-mosn/pkg/log"
 	"github.com/alipay/sofa-mosn/pkg/network"
 	"github.com/alipay/sofa-mosn/pkg/types"
+	"fmt"
 )
 
 // ConnectionHandler
@@ -65,7 +66,9 @@ func NewHandler(clusterManagerFilter types.ClusterManagerFilter, clMng types.Clu
 func (ch *connHandler) UpdateClusterConfig(clusters []v2.Cluster) error {
 
 	for _, cluster := range clusters {
-		ch.clusterManager.AddOrUpdatePrimaryCluster(cluster)
+		if !ch.clusterManager.AddOrUpdatePrimaryCluster(cluster) {
+			return fmt.Errorf("UpdateClusterConfig: AddOrUpdatePrimaryCluster failure, cluster name = %s",cluster.Name)
+		}
 	}
 
 	// TODO: remove cluster

pkg/types/protocol.go

@@ -82,5 +82,5 @@ type Encoder interface {
 type Decoder interface {
 	// Decode decodes binary to a model
 	// return 1. bytes decoded 2. decoded cmd
-	Decode(context context.Context, data IoBuffer) (int, interface{})
+	Decode(context context.Context, data IoBuffer) (interface{}, error)
 }

pkg/upstream/cluster/cluster.go

@@ -51,6 +51,8 @@ func NewCluster(clusterConfig v2.Cluster, sourceAddr net.Addr, addedViaAPI bool)
 
 	case v2.SIMPLE_CLUSTER, v2.DYNAMIC_CLUSTER:
 		newCluster = newSimpleInMemCluster(clusterConfig, sourceAddr, addedViaAPI)
+	default:
+		return nil
 	}
 
 	// init health check for cluster's host

pkg/upstream/cluster/clusteradapter.go

@@ -18,11 +18,10 @@
 package cluster
 
 import (
-	"errors"
-
 	"github.com/alipay/sofa-mosn/internal/api/v2"
 	"github.com/alipay/sofa-mosn/pkg/log"
 	"github.com/alipay/sofa-mosn/pkg/protocol/sofarpc"
+	"fmt"
 )
 
 // Adap is the instance of cluster Adapter
@@ -50,12 +49,15 @@ func (ca *Adapter) TriggerClusterUpdate(clusterName string, hosts []v2.Host) err
 				// todo support more default health check @boqin
 				cluster.HealthCheck = sofarpc.DefaultSofaRPCHealthCheckConf
 			}
-
-			ca.clusterMng.AddOrUpdatePrimaryCluster(cluster)
+
+			if !ca.clusterMng.AddOrUpdatePrimaryCluster(cluster) {
+				return fmt.Errorf("TriggerClusterUpdate: AddOrUpdatePrimaryCluster failure, cluster name = %s",cluster.Name)
+			}
+
 		} else {
 			msg := "cluster doesn't support auto discovery "
 			log.DefaultLogger.Errorf(msg)
-			return errors.New(msg)
+			return fmt.Errorf(msg)
 		}
 	}
 
@@ -77,8 +79,10 @@ func (ca *Adapter) TriggerClusterAdded(cluster v2.Cluster) {
 		if ca.clusterMng.registryUseHealthCheck {
 			cluster.HealthCheck = sofarpc.DefaultSofaRPCHealthCheckConf
 		}
-
-		ca.clusterMng.AddOrUpdatePrimaryCluster(cluster)
+
+		if !ca.clusterMng.AddOrUpdatePrimaryCluster(cluster) {
+			log.DefaultLogger.Errorf("TriggerClusterAdded: AddOrUpdatePrimaryCluster failure, cluster name = %s",cluster.Name)
+		}
 	} else {
 		log.DefaultLogger.Debugf("Added PrimaryCluster: %s Already Exist", cluster.Name)
 	}

pkg/upstream/cluster/clustermanager.go

@@ -74,7 +74,10 @@ func NewClusterManager(sourceAddr net.Addr, clusters []v2.Cluster,
 	//Add cluster to cm
 	//Register upstream update type
 	for _, cluster := range clusters {
-		cm.AddOrUpdatePrimaryCluster(cluster)
+
+		if !cm.AddOrUpdatePrimaryCluster(cluster) {
+			log.DefaultLogger.Errorf("NewClusterManager: AddOrUpdatePrimaryCluster failure, cluster name = %s",cluster.Name)
+		}
 	}
 
 	// Add hosts to cluster
@@ -111,11 +114,8 @@ func (cm *clusterManager) AddOrUpdatePrimaryCluster(cluster v2.Cluster) bool {
 			return false
 		}
 	}
-
 	// todo for static cluster, shouldn't use this way
-	cm.loadCluster(cluster, true)
-
-	return true
+	 return cm.loadCluster(cluster, true)
 }
 
 func (cm *clusterManager) ClusterExist(clusterName string) bool {
@@ -126,10 +126,14 @@ func (cm *clusterManager) ClusterExist(clusterName string) bool {
 	return false
 }
 
-func (cm *clusterManager) loadCluster(clusterConfig v2.Cluster, addedViaAPI bool) types.Cluster {
+func (cm *clusterManager) loadCluster(clusterConfig v2.Cluster, addedViaAPI bool) bool {
 	//clusterConfig.UseHealthCheck
 	cluster := NewCluster(clusterConfig, cm.sourceAddr, addedViaAPI)
-
+
+	if nil == cluster {
+		return false
+	}
+
 	cluster.Initialize(func() {
 		cluster.PrioritySet().AddMemberUpdateCb(func(priority uint32, hostsAdded []types.Host, hostsRemoved []types.Host) {
 		})
@@ -140,7 +144,7 @@ func (cm *clusterManager) loadCluster(clusterConfig v2.Cluster, addedViaAPI bool
 		addedViaAPI: addedViaAPI,
 	})
 
-	return cluster
+	return true
 }
 
 func (cm *clusterManager) getOrCreateClusterSnapshot(clusterName string) *clusterSnapshot {
@@ -193,6 +197,7 @@ func (cm *clusterManager) UpdateClusterHosts(clusterName string, priority uint32
 		return fmt.Errorf("cluster's hostset %s can't be update", clusterName)
 
 	}
+
 	return fmt.Errorf("cluster %s not found", clusterName)
 
 }