Merge pull request #1304 from cloudwego/release-v0.9.1

chore: release v0.9.1

README.md

@@ -30,11 +30,11 @@ Kitex is designed to be extensible to support multiple RPC messaging protocols.
 
 - **Multi-transport Protocol**
 
-For service governance, Kitex supports **TTHeader** and **HTTP2**. TTHeader can be used in conjunction with Thrift and Kitex Protobuf; HTTP2 is currently mainly used with the gRPC protocol, and it will support Thrift in the future.
+For service governance, Kitex supports **TTHeader** and **HTTP2**. TTHeader can be used in conjunction with Thrift and Kitex Protobuf.
 
 - **Multi-message Type**
 
-Kitex supports **PingPong**, **One-way**, and **Bidirectional Streaming**. Among them, One-way currently only supports Thrift protocol, two-way Streaming only supports gRPC, and Kitex will support Thrift's two-way Streaming in the future.
+Kitex supports **PingPong**, **One-way**, and **Bidirectional Streaming**. Among them, One-way currently only supports Thrift protocol.
 
 - **Service Governance**
 
@@ -73,7 +73,10 @@ Kitex has built-in code generation tools that support generating **Thrift**, **P
 - **Reference**
 
   - For Transport Protocol, Exception Instruction and Version Specification, please refer to [doc](https://www.cloudwego.io/docs/kitex/reference/).
-
+
+- **Best Practice**
+  - Kitex best practices in production, such as graceful shutdown, error handling, integration testing. [More](https://www.cloudwego.io/docs/kitex/best-practice/)
+
 - **FAQ**
 
   - Please refer to [FAQ](https://www.cloudwego.io/docs/kitex/faq/).
@@ -92,6 +95,7 @@ We provide the [kitex-benchmark](https://github.com/cloudwego/kitex-benchmark) p
 - [biz-demo](https://github.com/cloudwego/biz-demo): Business demos using Kitex.
 
 ## Blogs
+- [Enhancing Performance in Microservice Architecture with Kitex](https://www.cloudwego.io/blog/2024/01/29/enhancing-performance-in-microservice-architecture-with-kitex/)
 - [CloudWeGo: A leading practice for building enterprise cloud native middleware!](https://www.cloudwego.io/blog/2023/06/15/cloudwego-a-leading-practice-for-building-enterprise-cloud-native-middleware/)
 - [Kitex: Unifying Open Source Practice for a High-Performance RPC Framework](https://www.cloudwego.io/blog/2022/09/30/kitex-unifying-open-source-practice-for-a-high-performance-rpc-framework/)
 - [Performance Optimization on Kitex](https://www.cloudwego.io/blog/2021/09/23/performance-optimization-on-kitex/)
@@ -100,7 +104,7 @@ We provide the [kitex-benchmark](https://github.com/cloudwego/kitex-benchmark) p
 
 ## Contributing
 
-[Contributing](https://github.com/cloudwego/kitex/blob/develop/CONTRIBUTING.md).
+Contributor guide: [Contributing](https://github.com/cloudwego/kitex/blob/develop/CONTRIBUTING.md).
 
 ## License
 
@@ -118,7 +122,7 @@ Kitex is distributed under the [Apache License, version 2.0](https://github.com/
 ## Landscapes
 
 <p align="center">
-<img src="https://landscape.cncf.io/images/left-logo.svg" width="150"/>&nbsp;&nbsp;<img src="https://landscape.cncf.io/images/right-logo.svg" width="200"/>
+<img src="https://landscape.cncf.io/images/cncf-landscape-horizontal-color.svg" width="150"/>&nbsp;&nbsp;<img src="https://www.cncf.io/wp-content/uploads/2023/04/cncf-main-site-logo.svg" width="200"/>
 <br/><br/>
 CloudWeGo enriches the <a href="https://landscape.cncf.io/">CNCF CLOUD NATIVE Landscape</a>.
 </p>

README_cn.md

@@ -30,11 +30,11 @@ Kitex[kaɪt'eks] 字节跳动内部的 Golang 微服务 RPC 框架，具有**高
 
 - **多传输协议**
 
-  传输协议封装消息协议进行 RPC 互通，传输协议可以额外透传元信息，用于服务治理，Kitex 支持的传输协议有 **TTHeader**、**HTTP2**。TTHeader 可以和 Thrift、Kitex Protobuf 结合使用；HTTP2 目前主要是结合 gRPC 协议使用，后续也会支持 Thrift。
+  传输协议封装消息协议进行 RPC 互通，传输协议可以额外透传元信息，用于服务治理，Kitex 支持的传输协议有 **TTHeader**、**HTTP2**。TTHeader 可以和 Thrift、Kitex Protobuf 结合使用。
 
 - **多种消息类型**
 
-  支持 **PingPong**、**Oneway**、**双向 Streaming**。其中 Oneway 目前只对 Thrift 协议支持，双向 Streaming 只对 gRPC 支持，后续会考虑支持 Thrift 的双向 Streaming。
+  支持 **PingPong**、**Oneway**、**双向 Streaming**。其中 Oneway 目前只对 Thrift 协议支持。
 
 - **服务治理**
 
@@ -74,6 +74,9 @@ Kitex[kaɪt'eks] 字节跳动内部的 Golang 微服务 RPC 框架，具有**高
 
     - 关于应用层传输协议 TTHeader、异常说明与版本管理，请参考[文档](https://www.cloudwego.io/zh/docs/kitex/reference/)。
 
+  - **最佳实践**
+    - Kitex 在生产环境下的最佳实践，如优雅停机、错误处理、集成测试，详见：[文档](https://www.cloudwego.io/zh/docs/kitex/best-practice/)
+
   - **FAQ**
     - 请参考 [FAQ](https://www.cloudwego.io/zh/docs/kitex/faq/)。
 ## 框架性能
@@ -91,14 +94,15 @@ Kitex[kaɪt'eks] 字节跳动内部的 Golang 微服务 RPC 框架，具有**高
 
 ## 相关文章
 
+- [Kitex 两周年回顾 — 能力升级、社区合作与未来展望](https://www.cloudwego.io/zh/blog/2023/11/30/kitex-%E4%B8%A4%E5%91%A8%E5%B9%B4%E5%9B%9E%E9%A1%BE-%E8%83%BD%E5%8A%9B%E5%8D%87%E7%BA%A7%E7%A4%BE%E5%8C%BA%E5%90%88%E4%BD%9C%E4%B8%8E%E6%9C%AA%E6%9D%A5%E5%B1%95%E6%9C%9B/)
 - [高性能 RPC 框架 CloudWeGo-Kitex 内外统一的开源实践](https://www.cloudwego.io/zh/blog/2022/09/20/%E9%AB%98%E6%80%A7%E8%83%BD-rpc-%E6%A1%86%E6%9E%B6-cloudwego-kitex-%E5%86%85%E5%A4%96%E7%BB%9F%E4%B8%80%E7%9A%84%E5%BC%80%E6%BA%90%E5%AE%9E%E8%B7%B5/)
 - [字节跳动 Go RPC 框架 Kitex 性能优化实践](https://www.cloudwego.io/zh/blog/2021/09/23/%E5%AD%97%E8%8A%82%E8%B7%B3%E5%8A%A8-go-rpc-%E6%A1%86%E6%9E%B6-kitex-%E6%80%A7%E8%83%BD%E4%BC%98%E5%8C%96%E5%AE%9E%E8%B7%B5/)
 - [字节跳动在 Go 网络库上的实践](https://www.cloudwego.io/zh/blog/2021/10/09/%E5%AD%97%E8%8A%82%E8%B7%B3%E5%8A%A8%E5%9C%A8-go-%E7%BD%91%E7%BB%9C%E5%BA%93%E4%B8%8A%E7%9A%84%E5%AE%9E%E8%B7%B5/)
 - [RPC 框架 Kitex 实践入门：性能测试指南](https://www.cloudwego.io/zh/blog/2021/11/24/rpc-%E6%A1%86%E6%9E%B6-kitex-%E5%AE%9E%E8%B7%B5%E5%85%A5%E9%97%A8%E6%80%A7%E8%83%BD%E6%B5%8B%E8%AF%95%E6%8C%87%E5%8D%97/)
 
 ## 贡献代码
 
-[Contributing](CONTRIBUTING.md)。
+贡献者指南：[Contributing](CONTRIBUTING.md)。
 
 ## 开源许可
 
@@ -109,14 +113,14 @@ Kitex 基于[Apache License 2.0](LICENSE) 许可证，其依赖的三方组件
 - 如何成为 member: [COMMUNITY MEMBERSHIP](https://github.com/cloudwego/community/blob/main/COMMUNITY_MEMBERSHIP.md)
 - Issues: [Issues](https://github.com/cloudwego/kitex/issues)
 - Slack: 加入我们的 [Slack 频道](https://join.slack.com/t/cloudwego/shared_invite/zt-tmcbzewn-UjXMF3ZQsPhl7W3tEDZboA)
-- 飞书用户群（[注册飞书](https://www.feishu.cn/)进群）
+- 飞书用户群（[注册飞书](https://www.feishu.cn/)后扫码进群）
 
   ![LarkGroup](images/lark_group_cn.png)
 
 ## Landscapes
 
 <p align="center">
-<img src="https://landscape.cncf.io/images/left-logo.svg" width="150"/>&nbsp;&nbsp;<img src="https://landscape.cncf.io/images/right-logo.svg" width="200"/>
+<img src="https://landscape.cncf.io/images/cncf-landscape-horizontal-color.svg" width="150"/>&nbsp;&nbsp;<img src="https://www.cncf.io/wp-content/uploads/2023/04/cncf-main-site-logo.svg" width="200"/>
 <br/><br/>
 CloudWeGo 丰富了 <a href="https://landscape.cncf.io/">CNCF 云原生生态</a>。
 </p>

client/service_inline.go

@@ -19,14 +19,11 @@ package client
 import (
 	"context"
 	"errors"
-	"net"
 	"runtime/debug"
-
-	"github.com/bytedance/gopkg/cloud/metainfo"
+	"unsafe"
 
 	"github.com/cloudwego/kitex/client/callopt"
 	"github.com/cloudwego/kitex/internal/client"
-	internal_server "github.com/cloudwego/kitex/internal/server"
 	"github.com/cloudwego/kitex/pkg/consts"
 	"github.com/cloudwego/kitex/pkg/endpoint"
 	"github.com/cloudwego/kitex/pkg/klog"
@@ -35,15 +32,9 @@ import (
 	"github.com/cloudwego/kitex/pkg/utils"
 )
 
-var localAddr net.Addr
-
-func init() {
-	localAddr = utils.NewNetAddr("tcp", "127.0.0.1")
-}
-
 type ContextServiceInlineHandler interface {
-	WriteMeta(cliCtx, svrCtx context.Context, req interface{}) (newSvrCtx context.Context, err error)
-	ReadMeta(cliCtx, svrCtx context.Context, resp interface{}) (newCliCtx context.Context, err error)
+	WriteMeta(cliCtx context.Context, req interface{}) (newCliCtx context.Context, err error)
+	ReadMeta(cliCtx context.Context, resp interface{}) (err error)
 }
 
 type serviceInlineClient struct {
@@ -57,15 +48,12 @@ type serviceInlineClient struct {
 
 	// server info
 	serverEps endpoint.Endpoint
-	serverOpt *internal_server.Options
 
 	contextServiceInlineHandler ContextServiceInlineHandler
 }
 
 type ServerInitialInfo interface {
-	Endpoints() endpoint.Endpoint
-	Option() *internal_server.Options
-	GetServiceInfos() map[string]*serviceinfo.ServiceInfo
+	BuildServiceInlineInvokeChain() endpoint.Endpoint
 }
 
 // NewServiceInlineClient creates a kitex.Client with the given ServiceInfo, it is from generated code.
@@ -76,10 +64,7 @@ func NewServiceInlineClient(svcInfo *serviceinfo.ServiceInfo, s ServerInitialInf
 	kc := &serviceInlineClient{}
 	kc.svcInfo = svcInfo
 	kc.opt = client.NewOptions(opts)
-	kc.serverEps = s.Endpoints()
-	kc.serverOpt = s.Option()
-	kc.serverOpt.RemoteOpt.TargetSvcInfo = svcInfo
-	kc.serverOpt.RemoteOpt.SvcSearchMap = s.GetServiceInfos()
+	kc.serverEps = s.BuildServiceInlineInvokeChain()
 	if err := kc.init(); err != nil {
 		_ = kc.Close()
 		return nil, err
@@ -176,88 +161,25 @@ func (kc *serviceInlineClient) buildInvokeChain() error {
 	return nil
 }
 
-func (kc *serviceInlineClient) constructServerCtxWithMetadata(cliCtx context.Context) (serverCtx context.Context) {
-	serverCtx = context.Background()
-	// metainfo
-	// forward transmission
-	kvs := make(map[string]string, 16)
-	metainfo.SaveMetaInfoToMap(cliCtx, kvs)
-	if len(kvs) > 0 {
-		serverCtx = metainfo.SetMetaInfoFromMap(serverCtx, kvs)
-	}
-	serverCtx = metainfo.TransferForward(serverCtx)
-	// reverse transmission, backward mark
-	serverCtx = metainfo.WithBackwardValuesToSend(serverCtx)
-	return serverCtx
-}
-
-func (kc *serviceInlineClient) constructServerRPCInfo(svrCtx, cliCtx context.Context) (newServerCtx context.Context, svrRPCInfo rpcinfo.RPCInfo) {
-	rpcStats := rpcinfo.AsMutableRPCStats(rpcinfo.NewRPCStats())
-	if kc.serverOpt.StatsLevel != nil {
-		rpcStats.SetLevel(*kc.serverOpt.StatsLevel)
-	}
-	// Export read-only views to external users and keep a mapping for internal users.
-	ri := rpcinfo.NewRPCInfo(
-		rpcinfo.EmptyEndpointInfo(),
-		rpcinfo.FromBasicInfo(kc.serverOpt.Svr),
-		rpcinfo.NewServerInvocation(),
-		rpcinfo.AsMutableRPCConfig(kc.serverOpt.Configs).Clone().ImmutableView(),
-		rpcStats.ImmutableView(),
-	)
-	rpcinfo.AsMutableEndpointInfo(ri.From()).SetAddress(localAddr)
-	svrCtx = rpcinfo.NewCtxWithRPCInfo(svrCtx, ri)
-
-	cliRpcInfo := rpcinfo.GetRPCInfo(cliCtx)
-	// handle common rpcinfo
-	method := cliRpcInfo.To().Method()
-	if ink, ok := ri.Invocation().(rpcinfo.InvocationSetter); ok {
-		ink.SetMethodName(method)
-		ink.SetServiceName(kc.svcInfo.ServiceName)
-	}
-	rpcinfo.AsMutableEndpointInfo(ri.To()).SetMethod(method)
-	svrCtx = context.WithValue(svrCtx, consts.CtxKeyMethod, method)
-	return svrCtx, ri
-}
-
 func (kc *serviceInlineClient) invokeHandleEndpoint() (endpoint.Endpoint, error) {
-	svrTraceCtl := kc.serverOpt.TracerCtl
-	if svrTraceCtl == nil {
-		svrTraceCtl = &rpcinfo.TraceController{}
-	}
-
 	return func(ctx context.Context, req, resp interface{}) (err error) {
-		serverCtx := kc.constructServerCtxWithMetadata(ctx)
-		defer func() {
-			// backward key
-			kvs := metainfo.AllBackwardValuesToSend(serverCtx)
-			if len(kvs) > 0 {
-				metainfo.SetBackwardValuesFromMap(ctx, kvs)
-			}
-		}()
-		serverCtx, svrRPCInfo := kc.constructServerRPCInfo(serverCtx, ctx)
-		defer func() {
-			rpcinfo.PutRPCInfo(svrRPCInfo)
-		}()
-
-		// server trace
-		serverCtx = svrTraceCtl.DoStart(serverCtx, svrRPCInfo)
-
+		cliRpcInfo := rpcinfo.GetRPCInfo(ctx)
+		if v, ok := cliRpcInfo.Invocation().(rpcinfo.InvocationSetter); ok {
+			v.SetExtra(consts.SERVICE_INLINE_SERVICE_NAME, kc.svcInfo.ServiceName)
+		}
+		ctx = context.WithValue(ctx, consts.SERVICE_INLINE_RPCINFO_KEY, unsafe.Pointer(&cliRpcInfo))
 		if kc.contextServiceInlineHandler != nil {
-			serverCtx, err = kc.contextServiceInlineHandler.WriteMeta(ctx, serverCtx, req)
+			ctx, err = kc.contextServiceInlineHandler.WriteMeta(ctx, req)
 			if err != nil {
 				return err
 			}
 		}
 
 		// server logic
-		err = kc.serverEps(serverCtx, req, resp)
-		// finish server trace
-		// contextServiceInlineHandler may convert nil err to non nil err, so handle trace here
-		svrTraceCtl.DoFinish(serverCtx, svrRPCInfo, err)
+		err = kc.serverEps(ctx, req, resp)
 
 		if kc.contextServiceInlineHandler != nil {
-			var err1 error
-			ctx, err1 = kc.contextServiceInlineHandler.ReadMeta(ctx, serverCtx, resp)
+			err1 := kc.contextServiceInlineHandler.ReadMeta(ctx, resp)
 			if err1 != nil {
 				return err1
 			}

client/service_inline_test.go

@@ -30,21 +30,18 @@ import (
 	"github.com/cloudwego/kitex/client/callopt"
 	"github.com/cloudwego/kitex/internal/client"
 	"github.com/cloudwego/kitex/internal/mocks"
-	internal_server "github.com/cloudwego/kitex/internal/server"
 	"github.com/cloudwego/kitex/internal/test"
-	"github.com/cloudwego/kitex/pkg/consts"
 	"github.com/cloudwego/kitex/pkg/endpoint"
 	"github.com/cloudwego/kitex/pkg/kerrors"
 	"github.com/cloudwego/kitex/pkg/rpcinfo"
 	"github.com/cloudwego/kitex/pkg/rpcinfo/remoteinfo"
-	"github.com/cloudwego/kitex/pkg/serviceinfo"
 )
 
 type serverInitialInfoImpl struct {
 	EndpointsFunc func(ctx context.Context, req, resp interface{}) (err error)
 }
 
-func (s serverInitialInfoImpl) Endpoints() endpoint.Endpoint {
+func (s serverInitialInfoImpl) BuildServiceInlineInvokeChain() endpoint.Endpoint {
 	if s.EndpointsFunc != nil {
 		return s.EndpointsFunc
 	}
@@ -53,14 +50,6 @@ func (s serverInitialInfoImpl) Endpoints() endpoint.Endpoint {
 	}
 }
 
-func (s serverInitialInfoImpl) Option() *internal_server.Options {
-	return internal_server.NewOptions(nil)
-}
-
-func (s serverInitialInfoImpl) GetServiceInfos() map[string]*serviceinfo.ServiceInfo {
-	return nil
-}
-
 func newMockServerInitialInfo() ServerInitialInfo {
 	return &serverInitialInfoImpl{}
 }
@@ -347,28 +336,3 @@ func TestServiceInlineClientFinalizer(t *testing.T) {
 	t.Logf("After second GC, allocation: %f Mb, Number of allocation: %d\n", secondGCHeapAlloc, secondGCHeapObjects)
 	test.Assert(t, secondGCHeapAlloc < firstGCHeapAlloc/2 && secondGCHeapObjects < firstGCHeapObjects/2)
 }
-
-func TestServiceInlineMethodKeyCall(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	defer ctrl.Finish()
-	mtd := mocks.MockMethod
-	opts := []Option{
-		WithTransHandlerFactory(newMockCliTransHandlerFactory(ctrl)),
-		WithResolver(resolver404(ctrl)),
-		WithDialer(newDialer(ctrl)),
-		WithDestService("destService"),
-	}
-	svcInfo := mocks.ServiceInfo()
-	s := serverInitialInfoImpl{}
-	s.EndpointsFunc = func(ctx context.Context, req, resp interface{}) (err error) {
-		test.Assert(t, ctx.Value(consts.CtxKeyMethod) == mtd)
-		return nil
-	}
-	cli, err := NewServiceInlineClient(svcInfo, s, opts...)
-	test.Assert(t, err == nil)
-	ctx := context.Background()
-	req := new(MockTStruct)
-	res := new(MockTStruct)
-	err = cli.Call(ctx, mtd, req, res)
-	test.Assert(t, err == nil, err)
-}

client/stream.go

@@ -21,6 +21,7 @@ import (
 	"io"
 	"sync/atomic"
 
+	"github.com/cloudwego/kitex/pkg/kerrors"
 	"github.com/cloudwego/kitex/pkg/remote/trans/nphttp2/metadata"
 	"github.com/cloudwego/kitex/pkg/serviceinfo"
 
@@ -164,6 +165,8 @@ func (s *stream) Context() context.Context {
 func (s *stream) RecvMsg(m interface{}) (err error) {
 	err = s.recvEndpoint(s.stream, m)
 	if err == nil {
+		// BizStatusErr is returned by the server handle, meaning the stream is ended;
+		// And it should be returned to the calling business code for error handling
 		err = s.ri.Invocation().BizStatusErr()
 	}
 	if err != nil || s.streamingMode == serviceinfo.StreamingClient {
@@ -193,10 +196,23 @@ func (s *stream) DoFinish(err error) {
 		// already called
 		return
 	}
-	if err == io.EOF {
+	if !isRPCError(err) {
+		// only rpc errors are reported
 		err = nil
 	}
 	ctx := s.Context()
 	ri := rpcinfo.GetRPCInfo(ctx)
 	s.kc.opt.TracerCtl.DoFinish(ctx, ri, err)
 }
+
+func isRPCError(err error) bool {
+	if err == nil {
+		return false
+	}
+	if err == io.EOF {
+		return false
+	}
+	_, isBizStatusError := err.(kerrors.BizStatusErrorIface)
+	// if a tracer needs to get the BizStatusError, it should read from rpcinfo.invocation.bizStatusErr
+	return !isBizStatusError
+}