You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by li...@apache.org on 2023/11/08 08:34:01 UTC
(dubbo-go) branch feature-triple updated: feat：add triple healthcheck server (#2473)
This is an automated email from the ASF dual-hosted git repository.
liujun pushed a commit to branch feature-triple
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/feature-triple by this push:
new 1444a5289 feat：add triple healthcheck server (#2473)
1444a5289 is described below
commit 1444a52893ee034961e706c3113dc9a5802bd391
Author: finalt <fi...@163.com>
AuthorDate: Wed Nov 8 16:33:55 2023 +0800
feat：add triple healthcheck server (#2473)
---
config/provider_config.go | 18 +-
dubbo.go | 2 +-
imports/imports.go | 2 +-
protocol/triple/health/healthServer.go | 180 ++++++++++++
protocol/triple/health/triple_health/health.pb.go | 312 +++++++++++++++++++++
protocol/triple/health/triple_health/health.proto | 62 ++++
.../triple/health/triple_health/health.triple.go | 210 ++++++++++++++
.../triple/internal/client/health_client/main.go | 75 +++++
.../triple/triple-tool/gen/generator/tripleTpl.go | 8 +-
protocol/triple/triple.go | 3 +-
registry/protocol/protocol.go | 3 +-
server/server.go | 26 +-
12 files changed, 883 insertions(+), 18 deletions(-)
diff --git a/config/provider_config.go b/config/provider_config.go
index 5ca41f33b..7bbb90a58 100644
--- a/config/provider_config.go
+++ b/config/provider_config.go
@@ -126,15 +126,15 @@ func (c *ProviderConfig) Init(rc *RootConfig) error {
for k, v := range rc.Protocols {
if v.Name == tripleConstant.TRIPLE {
// Auto create grpc based health check service.
- healthService := NewServiceConfigBuilder().
- SetProtocolIDs(k).
- SetNotRegister(true).
- SetInterface(constant.HealthCheckServiceInterface).
- Build()
- if err := healthService.Init(rc); err != nil {
- return err
- }
- c.Services[constant.HealthCheckServiceTypeName] = healthService
+ //healthService := NewServiceConfigBuilder().
+ // SetProtocolIDs(k).
+ // SetNotRegister(true).
+ // SetInterface(constant.HealthCheckServiceInterface).
+ // Build()
+ //if err := healthService.Init(rc); err != nil {
+ // return err
+ //}
+ //c.Services[constant.HealthCheckServiceTypeName] = healthService
// Auto create reflection service configure only when provider with triple service is configured.
tripleReflectionService := NewServiceConfigBuilder().
diff --git a/dubbo.go b/dubbo.go
index 46cdfe563..ad24597ff 100644
--- a/dubbo.go
+++ b/dubbo.go
@@ -131,9 +131,9 @@ func (ins *Instance) NewServer(opts ...server.ServerOption) (*server.Server, err
srvOpts = append(srvOpts, opts...)
srv, err := server.NewServer(srvOpts...)
+
if err != nil {
return nil, err
}
-
return srv, nil
}
diff --git a/imports/imports.go b/imports/imports.go
index 595b37c35..54c6ca565 100644
--- a/imports/imports.go
+++ b/imports/imports.go
@@ -71,12 +71,12 @@ import (
_ "dubbo.apache.org/dubbo-go/v3/otel/trace/zipkin"
_ "dubbo.apache.org/dubbo-go/v3/protocol/dubbo"
_ "dubbo.apache.org/dubbo-go/v3/protocol/dubbo3"
- _ "dubbo.apache.org/dubbo-go/v3/protocol/dubbo3/health"
_ "dubbo.apache.org/dubbo-go/v3/protocol/dubbo3/reflection"
_ "dubbo.apache.org/dubbo-go/v3/protocol/grpc"
_ "dubbo.apache.org/dubbo-go/v3/protocol/jsonrpc"
_ "dubbo.apache.org/dubbo-go/v3/protocol/rest"
_ "dubbo.apache.org/dubbo-go/v3/protocol/triple"
+ _ "dubbo.apache.org/dubbo-go/v3/protocol/triple/health"
_ "dubbo.apache.org/dubbo-go/v3/proxy/proxy_factory"
_ "dubbo.apache.org/dubbo-go/v3/registry/etcdv3"
_ "dubbo.apache.org/dubbo-go/v3/registry/nacos"
diff --git a/protocol/triple/health/healthServer.go b/protocol/triple/health/healthServer.go
new file mode 100644
index 000000000..43b515af3
--- /dev/null
+++ b/protocol/triple/health/healthServer.go
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package health
+
+import (
+ "context"
+ "sync"
+)
+
+import (
+ "github.com/dubbogo/gost/log/logger"
+
+ "github.com/dubbogo/grpc-go/codes"
+ "github.com/dubbogo/grpc-go/status"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/protocol/triple/health/triple_health"
+ "dubbo.apache.org/dubbo-go/v3/server"
+)
+
+type HealthTripleServer struct {
+ mu sync.RWMutex
+ // If shutdown is true, it's expected all serving status is NOT_SERVING, and
+ // will stay in NOT_SERVING.
+ shutdown bool
+ // statusMap stores the serving status of the services this Server monitors.
+ statusMap map[string]triple_health.HealthCheckResponse_ServingStatus
+ updates map[string]map[triple_health.Health_WatchServer]chan triple_health.HealthCheckResponse_ServingStatus
+}
+
+var healthServer *HealthTripleServer
+
+func NewServer() *HealthTripleServer {
+ return &HealthTripleServer{
+ statusMap: map[string]triple_health.HealthCheckResponse_ServingStatus{"": triple_health.HealthCheckResponse_NOT_SERVING},
+ updates: make(map[string]map[triple_health.Health_WatchServer]chan triple_health.HealthCheckResponse_ServingStatus),
+ }
+}
+
+func (srv *HealthTripleServer) Check(ctx context.Context, req *triple_health.HealthCheckRequest) (*triple_health.HealthCheckResponse, error) {
+ srv.mu.RLock()
+ defer srv.mu.RUnlock()
+ if servingStatus, ok := srv.statusMap[req.Service]; ok {
+ return &triple_health.HealthCheckResponse{
+ Status: servingStatus,
+ }, nil
+ }
+ return nil, status.Error(codes.NotFound, "unknown service")
+}
+
+func (srv *HealthTripleServer) Watch(ctx context.Context, request *triple_health.HealthCheckRequest, server triple_health.Health_WatchServer) error {
+ service := request.Service
+ // update channel is used for getting service status updates.
+ update := make(chan triple_health.HealthCheckResponse_ServingStatus, 1)
+ srv.mu.Lock()
+ // Puts the initial status to the channel.
+ if servingStatus, ok := srv.statusMap[service]; ok {
+ update <- servingStatus
+ } else {
+ update <- triple_health.HealthCheckResponse_SERVICE_UNKNOWN
+ }
+
+ // Registers the update channel to the correct place in the updates map.
+ if _, ok := srv.updates[service]; !ok {
+ srv.updates[service] = make(map[triple_health.Health_WatchServer]chan triple_health.HealthCheckResponse_ServingStatus)
+ }
+ srv.updates[service][server] = update
+ defer func() {
+ srv.mu.Lock()
+ delete(srv.updates[service], server)
+ srv.mu.Unlock()
+ }()
+ srv.mu.Unlock()
+
+ var lastSentStatus triple_health.HealthCheckResponse_ServingStatus = -1
+ for {
+ select {
+ // Status updated. Sends the up-to-date status to the client.
+ case servingStatus := <-update:
+ if lastSentStatus == servingStatus {
+ continue
+ }
+ lastSentStatus = servingStatus
+ err := server.Send(&triple_health.HealthCheckResponse{Status: servingStatus})
+ if err != nil {
+ return status.Error(codes.Canceled, "Stream has ended.")
+ }
+ // Context done. Removes the update channel from the updates map.
+ case <-ctx.Done():
+ return status.Error(codes.Canceled, "Stream has ended.")
+ }
+ }
+}
+
+// SetServingStatus is called when need to reset the serving status of a service
+// or insert a new service entry into the statusMap.
+func (srv *HealthTripleServer) SetServingStatus(service string, servingStatus triple_health.HealthCheckResponse_ServingStatus) {
+ srv.mu.Lock()
+ defer srv.mu.Unlock()
+ if srv.shutdown {
+ logger.Infof("health: status changing for %s to %v is ignored because health service is shutdown", service, servingStatus)
+ return
+ }
+
+ srv.setServingStatusLocked(service, servingStatus)
+}
+
+func (srv *HealthTripleServer) setServingStatusLocked(service string, servingStatus triple_health.HealthCheckResponse_ServingStatus) {
+ srv.statusMap[service] = servingStatus
+ for _, update := range srv.updates[service] {
+ // Clears previous updates, that are not sent to the client, from the channel.
+ // This can happen if the client is not reading and the server gets flow control limited.
+ select {
+ case <-update:
+ default:
+ }
+ // Puts the most recent update to the channel.
+ update <- servingStatus
+ }
+}
+
+// Shutdown sets all serving status to NOT_SERVING, and configures the server to
+// ignore all future status changes.
+//
+// This changes serving status for all services. To set status for a particular
+// services, call SetServingStatus().
+func (srv *HealthTripleServer) Shutdown() {
+ srv.mu.Lock()
+ defer srv.mu.Unlock()
+ srv.shutdown = true
+ for service := range srv.statusMap {
+ srv.setServingStatusLocked(service, triple_health.HealthCheckResponse_NOT_SERVING)
+ }
+}
+
+// Resume sets all serving status to SERVING, and configures the server to
+// accept all future status changes.
+//
+// This changes serving status for all services. To set status for a particular
+// services, call SetServingStatus().
+func (srv *HealthTripleServer) Resume() {
+ srv.mu.Lock()
+ defer srv.mu.Unlock()
+ srv.shutdown = false
+ for service := range srv.statusMap {
+ srv.setServingStatusLocked(service, triple_health.HealthCheckResponse_SERVING)
+ }
+}
+
+func init() {
+ healthServer = NewServer()
+ server.SetProServices(&server.ServiceDefinition{
+ Handler: healthServer,
+ Info: &triple_health.Health_ServiceInfo,
+ })
+}
+
+func SetServingStatusServing(service string) {
+ healthServer.SetServingStatus(service, triple_health.HealthCheckResponse_SERVING)
+}
+
+func SetServingStatusNotServing(service string) {
+ healthServer.SetServingStatus(service, triple_health.HealthCheckResponse_NOT_SERVING)
+}
diff --git a/protocol/triple/health/triple_health/health.pb.go b/protocol/triple/health/triple_health/health.pb.go
new file mode 100644
index 000000000..a80d66793
--- /dev/null
+++ b/protocol/triple/health/triple_health/health.pb.go
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// The canonical version of this proto can be found at
+// https://github.com/grpc/grpc-proto/blob/master/grpc/health/v1/health.proto
+
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// protoc-gen-go v1.29.0
+// protoc v3.15.5
+// source: health.proto
+
+package triple_health
+
+import (
+ reflect "reflect"
+ sync "sync"
+)
+
+import (
+ protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+
+ protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+)
+
+const (
+ // Verify that this generated code is sufficiently up-to-date.
+ _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+ // Verify that runtime/protoimpl is sufficiently up-to-date.
+ _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+type HealthCheckResponse_ServingStatus int32
+
+const (
+ HealthCheckResponse_UNKNOWN HealthCheckResponse_ServingStatus = 0
+ HealthCheckResponse_SERVING HealthCheckResponse_ServingStatus = 1
+ HealthCheckResponse_NOT_SERVING HealthCheckResponse_ServingStatus = 2
+ HealthCheckResponse_SERVICE_UNKNOWN HealthCheckResponse_ServingStatus = 3 // Used only by the Watch method.
+)
+
+// Enum value maps for HealthCheckResponse_ServingStatus.
+var (
+ HealthCheckResponse_ServingStatus_name = map[int32]string{
+ 0: "UNKNOWN",
+ 1: "SERVING",
+ 2: "NOT_SERVING",
+ 3: "SERVICE_UNKNOWN",
+ }
+ HealthCheckResponse_ServingStatus_value = map[string]int32{
+ "UNKNOWN": 0,
+ "SERVING": 1,
+ "NOT_SERVING": 2,
+ "SERVICE_UNKNOWN": 3,
+ }
+)
+
+func (x HealthCheckResponse_ServingStatus) Enum() *HealthCheckResponse_ServingStatus {
+ p := new(HealthCheckResponse_ServingStatus)
+ *p = x
+ return p
+}
+
+func (x HealthCheckResponse_ServingStatus) String() string {
+ return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
+}
+
+func (HealthCheckResponse_ServingStatus) Descriptor() protoreflect.EnumDescriptor {
+ return file_health_proto_enumTypes[0].Descriptor()
+}
+
+func (HealthCheckResponse_ServingStatus) Type() protoreflect.EnumType {
+ return &file_health_proto_enumTypes[0]
+}
+
+func (x HealthCheckResponse_ServingStatus) Number() protoreflect.EnumNumber {
+ return protoreflect.EnumNumber(x)
+}
+
+// Deprecated: Use HealthCheckResponse_ServingStatus.Descriptor instead.
+func (HealthCheckResponse_ServingStatus) EnumDescriptor() ([]byte, []int) {
+ return file_health_proto_rawDescGZIP(), []int{1, 0}
+}
+
+type HealthCheckRequest struct {
+ state protoimpl.MessageState
+ sizeCache protoimpl.SizeCache
+ unknownFields protoimpl.UnknownFields
+
+ Service string `protobuf:"bytes,1,opt,name=service,proto3" json:"service,omitempty"`
+}
+
+func (x *HealthCheckRequest) Reset() {
+ *x = HealthCheckRequest{}
+ if protoimpl.UnsafeEnabled {
+ mi := &file_health_proto_msgTypes[0]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+ }
+}
+
+func (x *HealthCheckRequest) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*HealthCheckRequest) ProtoMessage() {}
+
+func (x *HealthCheckRequest) ProtoReflect() protoreflect.Message {
+ mi := &file_health_proto_msgTypes[0]
+ if protoimpl.UnsafeEnabled && x != nil {
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ if ms.LoadMessageInfo() == nil {
+ ms.StoreMessageInfo(mi)
+ }
+ return ms
+ }
+ return mi.MessageOf(x)
+}
+
+// Deprecated: Use HealthCheckRequest.ProtoReflect.Descriptor instead.
+func (*HealthCheckRequest) Descriptor() ([]byte, []int) {
+ return file_health_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *HealthCheckRequest) GetService() string {
+ if x != nil {
+ return x.Service
+ }
+ return ""
+}
+
+type HealthCheckResponse struct {
+ state protoimpl.MessageState
+ sizeCache protoimpl.SizeCache
+ unknownFields protoimpl.UnknownFields
+
+ Status HealthCheckResponse_ServingStatus `protobuf:"varint,1,opt,name=status,proto3,enum=grpc.health.v1.HealthCheckResponse_ServingStatus" json:"status,omitempty"`
+}
+
+func (x *HealthCheckResponse) Reset() {
+ *x = HealthCheckResponse{}
+ if protoimpl.UnsafeEnabled {
+ mi := &file_health_proto_msgTypes[1]
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ ms.StoreMessageInfo(mi)
+ }
+}
+
+func (x *HealthCheckResponse) String() string {
+ return protoimpl.X.MessageStringOf(x)
+}
+
+func (*HealthCheckResponse) ProtoMessage() {}
+
+func (x *HealthCheckResponse) ProtoReflect() protoreflect.Message {
+ mi := &file_health_proto_msgTypes[1]
+ if protoimpl.UnsafeEnabled && x != nil {
+ ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+ if ms.LoadMessageInfo() == nil {
+ ms.StoreMessageInfo(mi)
+ }
+ return ms
+ }
+ return mi.MessageOf(x)
+}
+
+// Deprecated: Use HealthCheckResponse.ProtoReflect.Descriptor instead.
+func (*HealthCheckResponse) Descriptor() ([]byte, []int) {
+ return file_health_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *HealthCheckResponse) GetStatus() HealthCheckResponse_ServingStatus {
+ if x != nil {
+ return x.Status
+ }
+ return HealthCheckResponse_UNKNOWN
+}
+
+var File_health_proto protoreflect.FileDescriptor
+
+var file_health_proto_rawDesc = []byte{
+ 0x0a, 0x0c, 0x68, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x0e,
+ 0x67, 0x72, 0x70, 0x63, 0x2e, 0x68, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x2e, 0x76, 0x31, 0x22, 0x2e,
+ 0x0a, 0x12, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x52, 0x65, 0x71,
+ 0x75, 0x65, 0x73, 0x74, 0x12, 0x18, 0x0a, 0x07, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x18,
+ 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x22, 0xb1,
+ 0x01, 0x0a, 0x13, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x52, 0x65,
+ 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x49, 0x0a, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73,
+ 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x31, 0x2e, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x68, 0x65,
+ 0x61, 0x6c, 0x74, 0x68, 0x2e, 0x76, 0x31, 0x2e, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x43, 0x68,
+ 0x65, 0x63, 0x6b, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x2e, 0x53, 0x65, 0x72, 0x76,
+ 0x69, 0x6e, 0x67, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75,
+ 0x73, 0x22, 0x4f, 0x0a, 0x0d, 0x53, 0x65, 0x72, 0x76, 0x69, 0x6e, 0x67, 0x53, 0x74, 0x61, 0x74,
+ 0x75, 0x73, 0x12, 0x0b, 0x0a, 0x07, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12,
+ 0x0b, 0x0a, 0x07, 0x53, 0x45, 0x52, 0x56, 0x49, 0x4e, 0x47, 0x10, 0x01, 0x12, 0x0f, 0x0a, 0x0b,
+ 0x4e, 0x4f, 0x54, 0x5f, 0x53, 0x45, 0x52, 0x56, 0x49, 0x4e, 0x47, 0x10, 0x02, 0x12, 0x13, 0x0a,
+ 0x0f, 0x53, 0x45, 0x52, 0x56, 0x49, 0x43, 0x45, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e,
+ 0x10, 0x03, 0x32, 0xae, 0x01, 0x0a, 0x06, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x12, 0x50, 0x0a,
+ 0x05, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x12, 0x22, 0x2e, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x68, 0x65,
+ 0x61, 0x6c, 0x74, 0x68, 0x2e, 0x76, 0x31, 0x2e, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x43, 0x68,
+ 0x65, 0x63, 0x6b, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x23, 0x2e, 0x67, 0x72, 0x70,
+ 0x63, 0x2e, 0x68, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x2e, 0x76, 0x31, 0x2e, 0x48, 0x65, 0x61, 0x6c,
+ 0x74, 0x68, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12,
+ 0x52, 0x0a, 0x05, 0x57, 0x61, 0x74, 0x63, 0x68, 0x12, 0x22, 0x2e, 0x67, 0x72, 0x70, 0x63, 0x2e,
+ 0x68, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x2e, 0x76, 0x31, 0x2e, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68,
+ 0x43, 0x68, 0x65, 0x63, 0x6b, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x23, 0x2e, 0x67,
+ 0x72, 0x70, 0x63, 0x2e, 0x68, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x2e, 0x76, 0x31, 0x2e, 0x48, 0x65,
+ 0x61, 0x6c, 0x74, 0x68, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
+ 0x65, 0x30, 0x01, 0x42, 0x43, 0x5a, 0x41, 0x64, 0x75, 0x62, 0x62, 0x6f, 0x2e, 0x61, 0x70, 0x61,
+ 0x63, 0x68, 0x65, 0x2e, 0x6f, 0x72, 0x67, 0x2f, 0x64, 0x75, 0x62, 0x62, 0x6f, 0x2d, 0x67, 0x6f,
+ 0x2f, 0x76, 0x33, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x2f, 0x74, 0x72, 0x69,
+ 0x70, 0x6c, 0x65, 0x2f, 0x68, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x2f, 0x74, 0x72, 0x69, 0x70, 0x6c,
+ 0x65, 0x5f, 0x68, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+ file_health_proto_rawDescOnce sync.Once
+ file_health_proto_rawDescData = file_health_proto_rawDesc
+)
+
+func file_health_proto_rawDescGZIP() []byte {
+ file_health_proto_rawDescOnce.Do(func() {
+ file_health_proto_rawDescData = protoimpl.X.CompressGZIP(file_health_proto_rawDescData)
+ })
+ return file_health_proto_rawDescData
+}
+
+var file_health_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
+var file_health_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
+var file_health_proto_goTypes = []interface{}{
+ (HealthCheckResponse_ServingStatus)(0), // 0: grpc.health.v1.HealthCheckResponse.ServingStatus
+ (*HealthCheckRequest)(nil), // 1: grpc.health.v1.HealthCheckRequest
+ (*HealthCheckResponse)(nil), // 2: grpc.health.v1.HealthCheckResponse
+}
+var file_health_proto_depIdxs = []int32{
+ 0, // 0: grpc.health.v1.HealthCheckResponse.status:type_name -> grpc.health.v1.HealthCheckResponse.ServingStatus
+ 1, // 1: grpc.health.v1.Health.Check:input_type -> grpc.health.v1.HealthCheckRequest
+ 1, // 2: grpc.health.v1.Health.Watch:input_type -> grpc.health.v1.HealthCheckRequest
+ 2, // 3: grpc.health.v1.Health.Check:output_type -> grpc.health.v1.HealthCheckResponse
+ 2, // 4: grpc.health.v1.Health.Watch:output_type -> grpc.health.v1.HealthCheckResponse
+ 3, // [3:5] is the sub-list for method output_type
+ 1, // [1:3] is the sub-list for method input_type
+ 1, // [1:1] is the sub-list for extension type_name
+ 1, // [1:1] is the sub-list for extension extendee
+ 0, // [0:1] is the sub-list for field type_name
+}
+
+func init() { file_health_proto_init() }
+func file_health_proto_init() {
+ if File_health_proto != nil {
+ return
+ }
+ if !protoimpl.UnsafeEnabled {
+ file_health_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+ switch v := v.(*HealthCheckRequest); i {
+ case 0:
+ return &v.state
+ case 1:
+ return &v.sizeCache
+ case 2:
+ return &v.unknownFields
+ default:
+ return nil
+ }
+ }
+ file_health_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+ switch v := v.(*HealthCheckResponse); i {
+ case 0:
+ return &v.state
+ case 1:
+ return &v.sizeCache
+ case 2:
+ return &v.unknownFields
+ default:
+ return nil
+ }
+ }
+ }
+ type x struct{}
+ out := protoimpl.TypeBuilder{
+ File: protoimpl.DescBuilder{
+ GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+ RawDescriptor: file_health_proto_rawDesc,
+ NumEnums: 1,
+ NumMessages: 2,
+ NumExtensions: 0,
+ NumServices: 1,
+ },
+ GoTypes: file_health_proto_goTypes,
+ DependencyIndexes: file_health_proto_depIdxs,
+ EnumInfos: file_health_proto_enumTypes,
+ MessageInfos: file_health_proto_msgTypes,
+ }.Build()
+ File_health_proto = out.File
+ file_health_proto_rawDesc = nil
+ file_health_proto_goTypes = nil
+ file_health_proto_depIdxs = nil
+}
diff --git a/protocol/triple/health/triple_health/health.proto b/protocol/triple/health/triple_health/health.proto
new file mode 100644
index 000000000..e3ef667fc
--- /dev/null
+++ b/protocol/triple/health/triple_health/health.proto
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// The canonical version of this proto can be found at
+// https://github.com/grpc/grpc-proto/blob/master/grpc/health/v1/health.proto
+
+syntax = "proto3";
+
+package grpc.health.v1;
+
+option go_package = "dubbo.apache.org/dubbo-go/v3/protocol/triple/health/triple_health";
+
+message HealthCheckRequest {
+ string service = 1;
+}
+
+message HealthCheckResponse {
+ enum ServingStatus {
+ UNKNOWN = 0;
+ SERVING = 1;
+ NOT_SERVING = 2;
+ SERVICE_UNKNOWN = 3; // Used only by the Watch method.
+ }
+ ServingStatus status = 1;
+}
+
+service Health {
+ // If the requested service is unknown, the call will fail with status
+ // NOT_FOUND.
+ rpc Check(HealthCheckRequest) returns (HealthCheckResponse);
+
+ // Performs a watch for the serving status of the requested service.
+ // The server will immediately send back a message indicating the current
+ // serving status. It will then subsequently send a new message whenever
+ // the service's serving status changes.
+ //
+ // If the requested service is unknown when the call is received, the
+ // server will send a message setting the serving status to
+ // SERVICE_UNKNOWN but will *not* terminate the call. If at some
+ // future point, the serving status of the service becomes known, the
+ // server will send a new message with the service's serving status.
+ //
+ // If the call terminates with status UNIMPLEMENTED, then clients
+ // should assume this method is not supported and should not retry the
+ // call. If the call terminates with any other status (including OK),
+ // clients should retry the call with appropriate exponential backoff.
+ rpc Watch(HealthCheckRequest) returns (stream HealthCheckResponse);
+}
\ No newline at end of file
diff --git a/protocol/triple/health/triple_health/health.triple.go b/protocol/triple/health/triple_health/health.triple.go
new file mode 100644
index 000000000..8d6e44781
--- /dev/null
+++ b/protocol/triple/health/triple_health/health.triple.go
@@ -0,0 +1,210 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by protoc-gen-triple. DO NOT EDIT.
+//
+// Source: protocol\triple\health\triple_health\health.proto
+package triple_health
+
+import (
+ "context"
+ "net/http"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/client"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol"
+ "dubbo.apache.org/dubbo-go/v3/server"
+)
+
+// This is a compile-time assertion to ensure that this generated file and the Triple package
+// are compatible. If you get a compiler error that this constant is not defined, this code was
+// generated with a version of Triple newer than the one compiled into your binary. You can fix the
+// problem by either regenerating this code with an older version of Triple or updating the Triple
+// version compiled into your binary.
+const _ = triple_protocol.IsAtLeastVersion0_1_0
+
+const (
+ // HealthName is the fully-qualified name of the Health service.
+ HealthName = "grpc.health.v1.Health"
+)
+
+// These constants are the fully-qualified names of the RPCs defined in this package. They're
+// exposed at runtime as procedure and as the final two segments of the HTTP route.
+//
+// Note that these are different from the fully-qualified method names used by
+// google.golang.org/protobuf/reflect/protoreflect. To convert from these constants to
+// reflection-formatted method names, remove the leading slash and convert the remaining slash to a
+// period.
+const (
+ // HealthCheckProcedure is the fully-qualified name of the Health's Check RPC.
+ HealthCheckProcedure = "/grpc.health.v1.Health/Check"
+ // HealthWatchProcedure is the fully-qualified name of the Health's Watch RPC.
+ HealthWatchProcedure = "/grpc.health.v1.Health/Watch"
+)
+
+var (
+ _ Health = (*HealthImpl)(nil)
+
+ _ Health_WatchClient = (*HealthWatchClient)(nil)
+
+ _ Health_WatchServer = (*HealthWatchServer)(nil)
+)
+
+// Health is a client for the grpc.health.v1.Health service.
+type Health interface {
+ Check(ctx context.Context, req *HealthCheckRequest, opts ...client.CallOption) (*HealthCheckResponse, error)
+ Watch(ctx context.Context, req *HealthCheckRequest, opts ...client.CallOption) (Health_WatchClient, error)
+}
+
+// NewHealth constructs a client for the grpc.health.v1.Health service.
+func NewHealth(cli *client.Client) (Health, error) {
+ if err := cli.Init(&Health_ClientInfo); err != nil {
+ return nil, err
+ }
+ return &HealthImpl{
+ cli: cli,
+ }, nil
+}
+
+// HealthImpl implements Health.
+type HealthImpl struct {
+ cli *client.Client
+}
+
+func (c *HealthImpl) Check(ctx context.Context, req *HealthCheckRequest, opts ...client.CallOption) (*HealthCheckResponse, error) {
+ resp := new(HealthCheckResponse)
+ if err := c.cli.CallUnary(ctx, req, resp, "grpc.health.v1.Health", "Check", opts...); err != nil {
+ return nil, err
+ }
+ return resp, nil
+}
+
+func (c *HealthImpl) Watch(ctx context.Context, req *HealthCheckRequest, opts ...client.CallOption) (Health_WatchClient, error) {
+ stream, err := c.cli.CallServerStream(ctx, req, "grpc.health.v1.Health", "Watch", opts...)
+ if err != nil {
+ return nil, err
+ }
+ rawStream := stream.(*triple_protocol.ServerStreamForClient)
+ return &HealthWatchClient{rawStream}, nil
+}
+
+type Health_WatchClient interface {
+ Recv() bool
+ ResponseHeader() http.Header
+ ResponseTrailer() http.Header
+ Msg() *HealthCheckResponse
+ Err() error
+ Conn() (triple_protocol.StreamingClientConn, error)
+ Close() error
+}
+
+type HealthWatchClient struct {
+ *triple_protocol.ServerStreamForClient
+}
+
+func (cli *HealthWatchClient) Recv() bool {
+ msg := new(HealthCheckResponse)
+ return cli.ServerStreamForClient.Receive(msg)
+}
+
+func (cli *HealthWatchClient) Msg() *HealthCheckResponse {
+ msg := cli.ServerStreamForClient.Msg()
+ if msg == nil {
+ return new(HealthCheckResponse)
+ }
+ return msg.(*HealthCheckResponse)
+}
+
+func (cli *HealthWatchClient) Conn() (triple_protocol.StreamingClientConn, error) {
+ return cli.ServerStreamForClient.Conn()
+}
+
+var Health_ClientInfo = client.ClientInfo{
+ InterfaceName: "grpc.health.v1.Health",
+ MethodNames: []string{"Check", "Watch"},
+ ClientInjectFunc: func(dubboCliRaw interface{}, cli *client.Client) {
+ dubboCli := dubboCliRaw.(HealthImpl)
+ dubboCli.cli = cli
+ },
+}
+
+// HealthHandler is an implementation of the grpc.health.v1.Health service.
+type HealthHandler interface {
+ Check(context.Context, *HealthCheckRequest) (*HealthCheckResponse, error)
+ Watch(context.Context, *HealthCheckRequest, Health_WatchServer) error
+}
+
+func RegisterHealthHandler(srv *server.Server, hdlr HealthHandler, opts ...server.ServiceOption) error {
+ return srv.Register(hdlr, &Health_ServiceInfo, opts...)
+}
+
+type Health_WatchServer interface {
+ Send(*HealthCheckResponse) error
+ ResponseHeader() http.Header
+ ResponseTrailer() http.Header
+ Conn() triple_protocol.StreamingHandlerConn
+}
+
+type HealthWatchServer struct {
+ *triple_protocol.ServerStream
+}
+
+func (g *HealthWatchServer) Send(msg *HealthCheckResponse) error {
+ return g.ServerStream.Send(msg)
+}
+
+var Health_ServiceInfo = server.ServiceInfo{
+ InterfaceName: "grpc.health.v1.Health",
+ ServiceType: (*HealthHandler)(nil),
+ Methods: []server.MethodInfo{
+ {
+ Name: "Check",
+ Type: constant.CallUnary,
+ ReqInitFunc: func() interface{} {
+ return new(HealthCheckRequest)
+ },
+ MethodFunc: func(ctx context.Context, args []interface{}, handler interface{}) (interface{}, error) {
+ req := args[0].(*HealthCheckRequest)
+ res, err := handler.(HealthHandler).Check(ctx, req)
+ if err != nil {
+ return nil, err
+ }
+ return triple_protocol.NewResponse(res), nil
+ },
+ },
+ {
+ Name: "Watch",
+ Type: constant.CallServerStream,
+ ReqInitFunc: func() interface{} {
+ return new(HealthCheckRequest)
+ },
+ StreamInitFunc: func(baseStream interface{}) interface{} {
+ return &HealthWatchServer{baseStream.(*triple_protocol.ServerStream)}
+ },
+ MethodFunc: func(ctx context.Context, args []interface{}, handler interface{}) (interface{}, error) {
+ req := args[0].(*HealthCheckRequest)
+ stream := args[1].(Health_WatchServer)
+ if err := handler.(HealthHandler).Watch(ctx, req, stream); err != nil {
+ return nil, err
+ }
+ return nil, nil
+ },
+ },
+ },
+}
diff --git a/protocol/triple/internal/client/health_client/main.go b/protocol/triple/internal/client/health_client/main.go
new file mode 100644
index 000000000..9d1363941
--- /dev/null
+++ b/protocol/triple/internal/client/health_client/main.go
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+ "context"
+)
+
+import (
+ "github.com/dubbogo/gost/log/logger"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/client"
+ _ "dubbo.apache.org/dubbo-go/v3/imports"
+ health "dubbo.apache.org/dubbo-go/v3/protocol/triple/health/triple_health"
+)
+
+func main() {
+ cli, err := client.NewClient(
+ client.WithURL("tri://127.0.0.1:20000"),
+ )
+ if err != nil {
+ panic(err)
+ }
+ svc, err := health.NewHealth(cli)
+ if err != nil {
+ panic(err)
+ }
+
+ check, err := svc.Check(context.Background(), &health.HealthCheckRequest{Service: "grpc.health.v1.Health"})
+ if err != nil {
+ logger.Error(err)
+ } else {
+ logger.Info("grpc.health.v1.Health's health", check.String())
+ }
+ check, err = svc.Check(context.Background(), &health.HealthCheckRequest{Service: "greet.GreetService"})
+ if err != nil {
+ logger.Error(err)
+ } else {
+ logger.Info("greet.GreetService's health", check.String())
+ }
+
+ watch, err := svc.Watch(context.Background(), &health.HealthCheckRequest{Service: "grpc.health.v1.Health"})
+ if err != nil {
+ logger.Error(err)
+ } else {
+ if watch.Recv() {
+ logger.Info("grpc.health.v1.Health's health", watch.Msg().String())
+ }
+ }
+ watch, err = svc.Watch(context.Background(), &health.HealthCheckRequest{Service: "greet.GreetService"})
+ if err != nil {
+ logger.Error(err)
+ } else {
+ if watch.Recv() {
+ logger.Info("greet.GreetService's health", watch.Msg().String())
+ }
+ }
+}
diff --git a/protocol/triple/triple-tool/gen/generator/tripleTpl.go b/protocol/triple/triple-tool/gen/generator/tripleTpl.go
index aee6b3300..fafc6c47f 100644
--- a/protocol/triple/triple-tool/gen/generator/tripleTpl.go
+++ b/protocol/triple/triple-tool/gen/generator/tripleTpl.go
@@ -46,7 +46,11 @@ func init() {
if err != nil {
log.Fatal(err)
}
- TplPackage, err = template.New("package").Parse(PackageTpl)
+ TplPackage, err = template.New("package").Funcs(template.FuncMap{
+ "pkg": func(s string) string {
+ return strings.ReplaceAll(s, ".", "_")
+ },
+ }).Parse(PackageTpl)
if err != nil {
log.Fatal(err)
}
@@ -147,7 +151,7 @@ const PreambleTpl = `// Code generated by protoc-gen-triple. DO NOT EDIT.
// Source: {{.Source}}
`
-const PackageTpl = `package {{.Package}}triple`
+const PackageTpl = `package {{pkg .Package}}triple`
const ImportTpl = `
diff --git a/protocol/triple/triple.go b/protocol/triple/triple.go
index f5d523a1b..c68ebd156 100644
--- a/protocol/triple/triple.go
+++ b/protocol/triple/triple.go
@@ -30,6 +30,7 @@ import (
"dubbo.apache.org/dubbo-go/v3/common/constant"
"dubbo.apache.org/dubbo-go/v3/common/extension"
"dubbo.apache.org/dubbo-go/v3/protocol"
+ "dubbo.apache.org/dubbo-go/v3/protocol/triple/health"
"dubbo.apache.org/dubbo-go/v3/server"
)
@@ -64,6 +65,7 @@ func (tp *TripleProtocol) Export(invoker protocol.Invoker) protocol.Exporter {
}
exporter := NewTripleExporter(serviceKey, invoker, tp.ExporterMap())
tp.SetExporterMap(serviceKey, exporter)
+ health.SetServingStatusServing(url.Service())
logger.Infof("[TRIPLE Protocol] Export service: %s", url.String())
tp.openServer(invoker, info)
return exporter
@@ -91,7 +93,6 @@ func (tp *TripleProtocol) openServer(invoker protocol.Invoker, info *server.Serv
tp.serverMap[url.Location].RefreshService(invoker, info)
return
}
-
if _, ok := tp.ExporterMap().Load(url.ServiceKey()); !ok {
panic("[TRIPLE Protocol]" + url.Key() + "is not existing")
}
diff --git a/registry/protocol/protocol.go b/registry/protocol/protocol.go
index 822f62207..bc5a23905 100644
--- a/registry/protocol/protocol.go
+++ b/registry/protocol/protocol.go
@@ -39,7 +39,6 @@ import (
"dubbo.apache.org/dubbo-go/v3/config_center"
_ "dubbo.apache.org/dubbo-go/v3/config_center/configurator"
"dubbo.apache.org/dubbo-go/v3/protocol"
- "dubbo.apache.org/dubbo-go/v3/protocol/dubbo3/health"
"dubbo.apache.org/dubbo-go/v3/protocol/protocolwrapper"
"dubbo.apache.org/dubbo-go/v3/registry"
"dubbo.apache.org/dubbo-go/v3/registry/directory"
@@ -200,7 +199,7 @@ func (proto *registryProtocol) Export(originInvoker protocol.Invoker) protocol.E
exporter := proto.doLocalExport(originInvoker, providerUrl)
// update health status
- health.SetServingStatusServing(registryUrl.Service())
+ //health.SetServingStatusServing(registryUrl.Service())
if len(registryUrl.Protocol) > 0 {
// url to registry
diff --git a/server/server.go b/server/server.go
index e8add11f6..b472e6810 100644
--- a/server/server.go
+++ b/server/server.go
@@ -34,6 +34,8 @@ import (
registry_exposed "dubbo.apache.org/dubbo-go/v3/registry/exposed_tmp"
)
+var proServices = map[string]*ServiceDefinition{}
+
type Server struct {
invoker protocol.Invoker
info *ServiceInfo
@@ -59,6 +61,12 @@ type infoInvoker struct {
methodMap map[string]*MethodInfo
}
+type ServiceDefinition struct {
+ Handler interface{}
+ Info *ServiceInfo
+ Opts []ServiceOption
+}
+
func (ii *infoInvoker) init() {
url := ii.base.GetURL()
if url.SubURL != nil {
@@ -187,7 +195,21 @@ func NewServer(opts ...ServerOption) (*Server, error) {
if err := newSrvOpts.init(opts...); err != nil {
return nil, err
}
- return &Server{
+
+ srv := &Server{
cfg: newSrvOpts,
- }, nil
+ }
+
+ for _, service := range proServices {
+ err := srv.Register(service.Handler, service.Info, service.Opts...)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ return srv, nil
+}
+
+func SetProServices(sd *ServiceDefinition) {
+ proServices[sd.Info.InterfaceName] = sd
}