This is an automated email from the ASF dual-hosted git repository. asifdxtreme pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-service-center.git
The following commit(s) were added to refs/heads/master by this push: new 6eee554 SCB-586 Query providers api will return null when provider is removed (#351) 6eee554 is described below commit 6eee5543417cd7d9a2ba4a059fdc7a51a47f7998 Author: little-cui <sure_0...@qq.com> AuthorDate: Wed May 16 17:11:09 2018 +0800 SCB-586 Query providers api will return null when provider is removed (#351) SCB-586 Query providers api will return null when provider is removed --- .gitignore | 2 + scripts/release/make_release.sh | 7 +-- server/error/error.go | 13 +++-- server/error/error_test.go | 46 +++++++++++++++ server/service/instances.go | 50 +---------------- server/service/instances_test.go | 65 ---------------------- server/service/service_suite_test.go | 14 ----- server/service/util/dependency.go | 5 +- server/service/util/rule_util.go | 10 ++-- server/service/watch.go | 72 ++++++++++++++++++++++++ server/service/watch_test.go | 105 +++++++++++++++++++++++++++++++++++ 11 files changed, 247 insertions(+), 142 deletions(-) diff --git a/.gitignore b/.gitignore index 855c698..9923352 100644 --- a/.gitignore +++ b/.gitignore @@ -14,9 +14,11 @@ vendor/** # for local UT server/service/conf/ +frontend/conf/ etc/data/ etc/ssl/ integration/conf/ frontend/bower_components/ frontend/node_modules/ server/govern/conf/ +tmp/ \ No newline at end of file diff --git a/scripts/release/make_release.sh b/scripts/release/make_release.sh index b7a18f1..abc746a 100755 --- a/scripts/release/make_release.sh +++ b/scripts/release/make_release.sh @@ -88,13 +88,11 @@ build_linux(){ export BUILD_NUMBER=$RELEASE GO_LDFLAGS="${GO_LDFLAGS} -X 'github.com/apache/incubator-servicecomb-service-center/version.BUILD_TAG=$(date +%Y%m%d%H%M%S).$BUILD_NUMBER.$GIT_COMMIT'" GO_LDFLAGS="${GO_LDFLAGS} -X 'github.com/apache/incubator-servicecomb-service-center/version.VERSION=$BUILD_NUMBER'" - go build --ldflags "${GO_LDFLAGS}" -o service-center - cp -r service-center $PACKAGE_PREFIX-$PACKAGE-linux-amd64 + go build --ldflags "${GO_LDFLAGS}" -o $PACKAGE_PREFIX-$PACKAGE-linux-amd64/service-center ## Build Frontend Release cd frontend - go build -o frontend - cp -r frontend ../$PACKAGE_PREFIX-$PACKAGE-linux-amd64 + go build -o ../$PACKAGE_PREFIX-$PACKAGE-linux-amd64/frontend cd .. prepare_conf @@ -116,7 +114,6 @@ build_linux(){ ## Archive the release tar -czvf $PACKAGE_PREFIX-$PACKAGE-linux-amd64.tar.gz $PACKAGE_PREFIX-$PACKAGE-linux-amd64 - } # Build Windows Release diff --git a/server/error/error.go b/server/error/error.go index a12662c..aca44a0 100644 --- a/server/error/error.go +++ b/server/error/error.go @@ -18,6 +18,7 @@ package error import ( "encoding/json" + "fmt" ) var errors = map[int32]string{ @@ -94,23 +95,23 @@ type Error struct { Detail string `json:"detail,omitempty"` } -func (e Error) Error() string { +func (e *Error) Error() string { if len(e.Detail) == 0 { return e.Message } return e.Message + "(" + e.Detail + ")" } -func (e Error) Marshal() []byte { +func (e *Error) Marshal() []byte { bs, _ := json.Marshal(e) return bs } -func (e Error) StatusCode() int { +func (e *Error) StatusCode() int { return int(e.Code / 1000) } -func (e Error) InternalError() bool { +func (e *Error) InternalError() bool { if e.Code >= 500000 { return true } @@ -125,6 +126,10 @@ func NewError(code int32, detail string) *Error { } } +func NewErrorf(code int32, format string, args ...interface{}) *Error { + return NewError(code, fmt.Sprintf(format, args...)) +} + func RegisterErrors(errs map[int32]string) { for err, msg := range errs { if err < 400000 || err >= 600000 { diff --git a/server/error/error_test.go b/server/error/error_test.go index da0e6af..1b61ff8 100644 --- a/server/error/error_test.go +++ b/server/error/error_test.go @@ -26,4 +26,50 @@ func TestError_StatusCode(t *testing.T) { if e.StatusCode() != http.StatusServiceUnavailable { t.Fatalf("TestError_StatusCode %v failed", e) } + + if !e.InternalError() { + t.Fatalf("TestInternalError failed") + } +} + +func TestNewError(t *testing.T) { + var err error + err = NewError(ErrInvalidParams, "test1") + if err == nil { + t.Fatalf("TestNewError failed") + } + err = NewErrorf(ErrInvalidParams, "%s", "test2") + if err == nil { + t.Fatalf("TestNewErrorf failed") + } + + if len(err.Error()) == 0 { + t.Fatalf("TestError failed") + } + + if len(err.(*Error).Marshal()) == 0 { + t.Fatalf("TestMarshal failed") + } + + if err.(*Error).StatusCode() != http.StatusBadRequest { + t.Fatalf("TestStatusCode failed, %d", err.(*Error).StatusCode()) + } + + if err.(*Error).InternalError() { + t.Fatalf("TestInternalError failed") + } + + err = NewErrorf(ErrInvalidParams, "") + if len(err.Error()) == 0 { + t.Fatalf("TestNewErrorf with empty detial failed") + } +} + +func TestRegisterErrors(t *testing.T) { + RegisterErrors(map[int32]string{503999: "test1", 1: "none"}) + + e := NewError(503999, "test2") + if e.Message != "test1" { + t.Fatalf("TestRegisterErrors failed") + } } diff --git a/server/service/instances.go b/server/service/instances.go index e65bcb4..ffb4076 100644 --- a/server/service/instances.go +++ b/server/service/instances.go @@ -28,9 +28,7 @@ import ( "github.com/apache/incubator-servicecomb-service-center/server/infra/quota" "github.com/apache/incubator-servicecomb-service-center/server/infra/registry" "github.com/apache/incubator-servicecomb-service-center/server/plugin" - nf "github.com/apache/incubator-servicecomb-service-center/server/service/notification" serviceUtil "github.com/apache/incubator-servicecomb-service-center/server/service/util" - "github.com/gorilla/websocket" "golang.org/x/net/context" "math" "strconv" @@ -450,14 +448,14 @@ func (s *InstanceService) getInstancePreCheck(ctx context.Context, in interface{ if len(tags) > 0 { tagsFromETCD, err := serviceUtil.GetTagsUtils(ctx, targetDomainProject, providerServiceId) if err != nil { - return scerr.NewError(scerr.ErrInternal, fmt.Sprintf("An error occurred in query provider tags(%s)", err.Error())) + return scerr.NewErrorf(scerr.ErrInternal, "An error occurred in query provider tags(%s)", err.Error()) } if len(tagsFromETCD) == 0 { return scerr.NewError(scerr.ErrTagNotExists, "Provider has no tag") } for _, tag := range tags { if _, ok := tagsFromETCD[tag]; !ok { - return scerr.NewError(scerr.ErrTagNotExists, fmt.Sprintf("Provider tags do not contain '%s'", tag)) + return scerr.NewErrorf(scerr.ErrTagNotExists, "Provider tags do not contain '%s'", tag) } } } @@ -694,50 +692,6 @@ func (s *InstanceService) UpdateInstanceProperties(ctx context.Context, in *pb.U }, nil } -func (s *InstanceService) WatchPreOpera(ctx context.Context, in *pb.WatchInstanceRequest) error { - if in == nil || len(in.SelfServiceId) == 0 { - return errors.New("Request format invalid.") - } - domainProject := util.ParseDomainProject(ctx) - if !serviceUtil.ServiceExist(ctx, domainProject, in.SelfServiceId) { - return errors.New("Service does not exist.") - } - return nil -} - -func (s *InstanceService) Watch(in *pb.WatchInstanceRequest, stream pb.ServiceInstanceCtrl_WatchServer) error { - var err error - if err = s.WatchPreOpera(stream.Context(), in); err != nil { - util.Logger().Errorf(err, "establish watch failed: invalid params.") - return err - } - domainProject := util.ParseDomainProject(stream.Context()) - watcher := nf.NewInstanceListWatcher(in.SelfServiceId, apt.GetInstanceRootKey(domainProject)+"/", nil) - err = nf.GetNotifyService().AddSubscriber(watcher) - util.Logger().Infof("start watch instance status, watcher %s %s", watcher.Subject(), watcher.Id()) - return nf.HandleWatchJob(watcher, stream, nf.GetNotifyService().Config.NotifyTimeout) -} - -func (s *InstanceService) WebSocketWatch(ctx context.Context, in *pb.WatchInstanceRequest, conn *websocket.Conn) { - util.Logger().Infof("New a web socket watch with %s", in.SelfServiceId) - if err := s.WatchPreOpera(ctx, in); err != nil { - nf.EstablishWebSocketError(conn, err) - return - } - nf.DoWebSocketListAndWatch(ctx, in.SelfServiceId, nil, conn) -} - -func (s *InstanceService) WebSocketListAndWatch(ctx context.Context, in *pb.WatchInstanceRequest, conn *websocket.Conn) { - util.Logger().Infof("New a web socket list and watch with %s", in.SelfServiceId) - if err := s.WatchPreOpera(ctx, in); err != nil { - nf.EstablishWebSocketError(conn, err) - return - } - nf.DoWebSocketListAndWatch(ctx, in.SelfServiceId, func() ([]*pb.WatchInstanceResponse, int64) { - return serviceUtil.QueryAllProvidersInstances(ctx, in.SelfServiceId) - }, conn) -} - func (s *InstanceService) ClusterHealth(ctx context.Context) (*pb.GetInstancesResponse, error) { domainProject := util.StringJoin([]string{apt.REGISTRY_DOMAIN, apt.REGISTRY_PROJECT}, "/") serviceId, err := serviceUtil.GetServiceId(ctx, &pb.MicroServiceKey{ diff --git a/server/service/instances_test.go b/server/service/instances_test.go index 18e2da1..9c91841 100644 --- a/server/service/instances_test.go +++ b/server/service/instances_test.go @@ -21,26 +21,11 @@ import ( "github.com/apache/incubator-servicecomb-service-center/server/core" pb "github.com/apache/incubator-servicecomb-service-center/server/core/proto" scerr "github.com/apache/incubator-servicecomb-service-center/server/error" - "github.com/apache/incubator-servicecomb-service-center/server/service" . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" - "golang.org/x/net/context" - "google.golang.org/grpc" "os" ) -type grpcWatchServer struct { - grpc.ServerStream -} - -func (x *grpcWatchServer) Send(m *pb.WatchInstanceResponse) error { - return nil -} - -func (x *grpcWatchServer) Context() context.Context { - return getContext() -} - var _ = Describe("'Instance' service", func() { Describe("execute 'register' operartion", func() { var ( @@ -1256,54 +1241,4 @@ var _ = Describe("'Instance' service", func() { }) }) }) - - Describe("execute 'watch' operartion", func() { - var ( - serviceId string - ) - - It("should be passed", func() { - respCreate, err := serviceResource.Create(getContext(), &pb.CreateServiceRequest{ - Service: &pb.MicroService{ - ServiceName: "service_name_watch", - AppId: "service_name_watch", - Version: "1.0.0", - Level: "BACK", - Status: pb.MS_UP, - }, - }) - Expect(err).To(BeNil()) - Expect(respCreate.Response.Code).To(Equal(pb.Response_SUCCESS)) - serviceId = respCreate.ServiceId - }) - - Context("when request is invalid", func() { - It("should be failed", func() { - By("service does not exist") - IC := instanceResource.(*service.InstanceService) - err := IC.WatchPreOpera(getContext(), &pb.WatchInstanceRequest{ - SelfServiceId: "-1", - }) - Expect(err).NotTo(BeNil()) - - err = IC.Watch(&pb.WatchInstanceRequest{ - SelfServiceId: "-1", - }, &grpcWatchServer{}) - Expect(err).NotTo(BeNil()) - - By("service id is empty") - err = instanceResource.(*service.InstanceService).WatchPreOpera(getContext(), &pb.WatchInstanceRequest{ - SelfServiceId: "", - }) - Expect(err).NotTo(BeNil()) - - By("request is valid") - err = instanceResource.(*service.InstanceService).WatchPreOpera(getContext(), - &pb.WatchInstanceRequest{ - SelfServiceId: serviceId, - }) - Expect(err).To(BeNil()) - }) - }) - }) }) diff --git a/server/service/service_suite_test.go b/server/service/service_suite_test.go index e64a8c4..696ab2d 100644 --- a/server/service/service_suite_test.go +++ b/server/service/service_suite_test.go @@ -57,17 +57,3 @@ func TestRegisterGrpcServices(t *testing.T) { }() service.RegisterGrpcServices(nil) } - -func TestInstanceService_WebSocketWatch(t *testing.T) { - defer func() { - recover() - }() - instanceResource.WebSocketWatch(context.Background(), &pb.WatchInstanceRequest{}, nil) -} - -func TestInstanceService_WebSocketListAndWatch(t *testing.T) { - defer func() { - recover() - }() - instanceResource.WebSocketListAndWatch(context.Background(), &pb.WatchInstanceRequest{}, nil) -} diff --git a/server/service/util/dependency.go b/server/service/util/dependency.go index fb5741b..0803901 100644 --- a/server/service/util/dependency.go +++ b/server/service/util/dependency.go @@ -646,7 +646,7 @@ func (dep *Dependency) UpdateProvidersRuleOfConsumer(conKey string) error { _, err := backend.Registry().Do(context.TODO(), registry.DEL, registry.WithStrKey(conKey), - ) + ) if err != nil { util.Logger().Errorf(nil, "Upload dependency rule failed.") return err @@ -725,6 +725,9 @@ func (dr *DependencyRelation) GetDependencyProviders(opts ...DependencyRelationF key.AppId, key.ServiceName, key.Version) continue } + if provider == nil { + continue + } if op.NonSelf && providerId == dr.consumer.ServiceId { continue } diff --git a/server/service/util/rule_util.go b/server/service/util/rule_util.go index e158c80..c9bb01c 100644 --- a/server/service/util/rule_util.go +++ b/server/service/util/rule_util.go @@ -204,7 +204,7 @@ func parsePattern(v reflect.Value, rule *pb.ServiceRule, tagsOfConsumer map[stri if !key.IsValid() { util.Logger().Errorf(nil, "can not find service %s field '%s', rule %s", consumerId, rule.Attribute, rule.RuleId) - return "", scerr.NewError(scerr.ErrInternal, fmt.Sprintf("Can not find field '%s'", rule.Attribute)) + return "", scerr.NewErrorf(scerr.ErrInternal, "Can not find field '%s'", rule.Attribute) } return key.String(), nil @@ -239,7 +239,7 @@ func Accessible(ctx context.Context, consumerId string, providerId string) *scer consumerService, err := GetService(ctx, domainProject, consumerId) if err != nil { - return scerr.NewError(scerr.ErrInternal, fmt.Sprintf("An error occurred in query consumer(%s)", err.Error())) + return scerr.NewErrorf(scerr.ErrInternal, "An error occurred in query consumer(%s)", err.Error()) } if consumerService == nil { return scerr.NewError(scerr.ErrServiceNotExists, "consumer serviceId is invalid") @@ -248,7 +248,7 @@ func Accessible(ctx context.Context, consumerId string, providerId string) *scer // 跨应用权限 providerService, err := GetService(ctx, targetDomainProject, providerId) if err != nil { - return scerr.NewError(scerr.ErrInternal, fmt.Sprintf("An error occurred in query provider(%s)", err.Error())) + return scerr.NewErrorf(scerr.ErrInternal, "An error occurred in query provider(%s)", err.Error()) } if providerService == nil { return scerr.NewError(scerr.ErrServiceNotExists, "provider serviceId is invalid") @@ -264,7 +264,7 @@ func Accessible(ctx context.Context, consumerId string, providerId string) *scer // 黑白名单 rules, err := GetRulesUtil(ctx, targetDomainProject, providerId) if err != nil { - return scerr.NewError(scerr.ErrInternal, fmt.Sprintf("An error occurred in query provider rules(%s)", err.Error())) + return scerr.NewErrorf(scerr.ErrInternal, "An error occurred in query provider rules(%s)", err.Error()) } if len(rules) == 0 { @@ -273,7 +273,7 @@ func Accessible(ctx context.Context, consumerId string, providerId string) *scer validateTags, err := GetTagsUtils(ctx, domainProject, consumerService.ServiceId) if err != nil { - return scerr.NewError(scerr.ErrInternal, fmt.Sprintf("An error occurred in query consumer tags(%s)", err.Error())) + return scerr.NewErrorf(scerr.ErrInternal, "An error occurred in query consumer tags(%s)", err.Error()) } return MatchRules(rules, consumerService, validateTags) diff --git a/server/service/watch.go b/server/service/watch.go new file mode 100644 index 0000000..e0b5780 --- /dev/null +++ b/server/service/watch.go @@ -0,0 +1,72 @@ +/* + * 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 service + +import ( + "errors" + "github.com/apache/incubator-servicecomb-service-center/pkg/util" + apt "github.com/apache/incubator-servicecomb-service-center/server/core" + pb "github.com/apache/incubator-servicecomb-service-center/server/core/proto" + nf "github.com/apache/incubator-servicecomb-service-center/server/service/notification" + serviceUtil "github.com/apache/incubator-servicecomb-service-center/server/service/util" + "github.com/gorilla/websocket" + "golang.org/x/net/context" +) + +func (s *InstanceService) WatchPreOpera(ctx context.Context, in *pb.WatchInstanceRequest) error { + if in == nil || len(in.SelfServiceId) == 0 { + return errors.New("Request format invalid.") + } + domainProject := util.ParseDomainProject(ctx) + if !serviceUtil.ServiceExist(ctx, domainProject, in.SelfServiceId) { + return errors.New("Service does not exist.") + } + return nil +} + +func (s *InstanceService) Watch(in *pb.WatchInstanceRequest, stream pb.ServiceInstanceCtrl_WatchServer) error { + var err error + if err = s.WatchPreOpera(stream.Context(), in); err != nil { + util.Logger().Errorf(err, "establish watch failed: invalid params.") + return err + } + domainProject := util.ParseDomainProject(stream.Context()) + watcher := nf.NewInstanceListWatcher(in.SelfServiceId, apt.GetInstanceRootKey(domainProject)+"/", nil) + err = nf.GetNotifyService().AddSubscriber(watcher) + util.Logger().Infof("start watch instance status, watcher %s %s", watcher.Subject(), watcher.Id()) + return nf.HandleWatchJob(watcher, stream, nf.GetNotifyService().Config.NotifyTimeout) +} + +func (s *InstanceService) WebSocketWatch(ctx context.Context, in *pb.WatchInstanceRequest, conn *websocket.Conn) { + util.Logger().Infof("New a web socket watch with %s", in.SelfServiceId) + if err := s.WatchPreOpera(ctx, in); err != nil { + nf.EstablishWebSocketError(conn, err) + return + } + nf.DoWebSocketListAndWatch(ctx, in.SelfServiceId, nil, conn) +} + +func (s *InstanceService) WebSocketListAndWatch(ctx context.Context, in *pb.WatchInstanceRequest, conn *websocket.Conn) { + util.Logger().Infof("New a web socket list and watch with %s", in.SelfServiceId) + if err := s.WatchPreOpera(ctx, in); err != nil { + nf.EstablishWebSocketError(conn, err) + return + } + nf.DoWebSocketListAndWatch(ctx, in.SelfServiceId, func() ([]*pb.WatchInstanceResponse, int64) { + return serviceUtil.QueryAllProvidersInstances(ctx, in.SelfServiceId) + }, conn) +} diff --git a/server/service/watch_test.go b/server/service/watch_test.go new file mode 100644 index 0000000..b0ef72e --- /dev/null +++ b/server/service/watch_test.go @@ -0,0 +1,105 @@ +/* + * 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 service_test + +import ( + pb "github.com/apache/incubator-servicecomb-service-center/server/core/proto" + "github.com/apache/incubator-servicecomb-service-center/server/service" + . "github.com/onsi/ginkgo" + . "github.com/onsi/gomega" + "golang.org/x/net/context" + "google.golang.org/grpc" + "testing" +) + +type grpcWatchServer struct { + grpc.ServerStream +} + +func (x *grpcWatchServer) Send(m *pb.WatchInstanceResponse) error { + return nil +} + +func (x *grpcWatchServer) Context() context.Context { + return getContext() +} + +func TestInstanceService_WebSocketWatch(t *testing.T) { + defer func() { + recover() + }() + instanceResource.WebSocketWatch(context.Background(), &pb.WatchInstanceRequest{}, nil) +} + +func TestInstanceService_WebSocketListAndWatch(t *testing.T) { + defer func() { + recover() + }() + instanceResource.WebSocketListAndWatch(context.Background(), &pb.WatchInstanceRequest{}, nil) +} + +var _ = Describe("'Instance' service", func() { + Describe("execute 'watch' operartion", func() { + var ( + serviceId string + ) + + It("should be passed", func() { + respCreate, err := serviceResource.Create(getContext(), &pb.CreateServiceRequest{ + Service: &pb.MicroService{ + ServiceName: "service_name_watch", + AppId: "service_name_watch", + Version: "1.0.0", + Level: "BACK", + Status: pb.MS_UP, + }, + }) + Expect(err).To(BeNil()) + Expect(respCreate.Response.Code).To(Equal(pb.Response_SUCCESS)) + serviceId = respCreate.ServiceId + }) + + Context("when request is invalid", func() { + It("should be failed", func() { + By("service does not exist") + IC := instanceResource.(*service.InstanceService) + err := IC.WatchPreOpera(getContext(), &pb.WatchInstanceRequest{ + SelfServiceId: "-1", + }) + Expect(err).NotTo(BeNil()) + + err = IC.Watch(&pb.WatchInstanceRequest{ + SelfServiceId: "-1", + }, &grpcWatchServer{}) + Expect(err).NotTo(BeNil()) + + By("service id is empty") + err = instanceResource.(*service.InstanceService).WatchPreOpera(getContext(), &pb.WatchInstanceRequest{ + SelfServiceId: "", + }) + Expect(err).NotTo(BeNil()) + + By("request is valid") + err = instanceResource.(*service.InstanceService).WatchPreOpera(getContext(), + &pb.WatchInstanceRequest{ + SelfServiceId: serviceId, + }) + Expect(err).To(BeNil()) + }) + }) + }) +}) -- To stop receiving notification emails like this one, please contact asifdxtr...@apache.org.