Copilot commented on code in PR #1367: URL: https://github.com/apache/dubbo-admin/pull/1367#discussion_r2616203284
########## pkg/discovery/nacos2/listerwatcher/nacos_service.go: ########## @@ -0,0 +1,273 @@ +/* + * 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 listerwatcher + +import ( + "fmt" + "time" + + "github.com/duke-git/lancet/v2/slice" + "github.com/go-co-op/gocron" + nacosnamingclient "github.com/nacos-group/nacos-sdk-go/v2/clients/naming_client" + nacosmodel "github.com/nacos-group/nacos-sdk-go/v2/model" + nacosvo "github.com/nacos-group/nacos-sdk-go/v2/vo" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + k8sruntime "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/tools/cache" + + meshproto "github.com/apache/dubbo-admin/api/mesh/v1alpha1" + "github.com/apache/dubbo-admin/pkg/common/bizerror" + discoverycfg "github.com/apache/dubbo-admin/pkg/config/discovery" + "github.com/apache/dubbo-admin/pkg/core/logger" + meshresource "github.com/apache/dubbo-admin/pkg/core/resource/apis/mesh/v1alpha1" + coremodel "github.com/apache/dubbo-admin/pkg/core/resource/model" +) + +type NacosServiceListerWatcher struct { + cfg *discoverycfg.Config + namingClient nacosnamingclient.INamingClient + // watchingServices is used to record the services that are currently being watched + // key -> nacos service name, value -> true if it is watched in current list turn + watchingServices map[string]bool + scheduler *gocron.Scheduler + resultChan chan watch.Event + stopWatch bool +} + +func NewNacosServiceListerWatcher(cfg *discoverycfg.Config, namingClient nacosnamingclient.INamingClient) *NacosServiceListerWatcher { + return &NacosServiceListerWatcher{ + cfg: cfg, + namingClient: namingClient, + resultChan: make(chan watch.Event), + watchingServices: make(map[string]bool), + scheduler: gocron.NewScheduler(time.UTC), + stopWatch: false, + } +} + +func (lw *NacosServiceListerWatcher) List(_ metav1.ListOptions) (k8sruntime.Object, error) { + serviceNames, err := lw.fetchAllServiceNames() + if err != nil { + return nil, err + } + nacosServiceList := meshresource.NewNacosServiceResourceListWithItems() + resList := make([]*meshresource.NacosServiceResource, 0) + for _, serviceName := range serviceNames { + service, err := lw.namingClient.GetService(nacosvo.GetServiceParam{ + ServiceName: serviceName, + }) + if err != nil { + logger.Errorf("get instances of service %s failed in nacos %s, cause: %v", serviceName, lw.nacosAddress(), err) + continue + } + resList = append(resList, lw.toNacosServiceResource(serviceName, service.Hosts)) + } + nacosServiceList.Items = resList + return nacosServiceList, nil +} + +func (lw *NacosServiceListerWatcher) Watch(_ metav1.ListOptions) (watch.Interface, error) { + _, err := lw.scheduler.Every(lw.cfg.Properties.ServiceWatchPeriod).Seconds().Do(func() { + if lw.stopWatch { + logger.Debugf("stop watch all services of nacos %s", lw.mesh()) + lw.scheduler.Stop() + return + } + startTime := time.Now() + err := lw.fetchAndProcessService() + if err != nil { + logger.Errorf("fetch all service failed in nacos %s, cause: %v", lw.nacosAddress(), err) + return + } + costs := time.Now().UnixMilli() - startTime.UnixMilli() + logger.Debugf("finish fetching and processing all services in nacos %s, costs %dms", lw.nacosAddress(), costs) + }) + if err != nil { + return nil, bizerror.Wrap(err, bizerror.UnknownError, + fmt.Sprintf("watch all services of nacos %s in a schedule occurs error", lw.mesh())) + } + lw.scheduler.StartAsync() + return lw, nil +} + +func (lw *NacosServiceListerWatcher) fetchAllServiceNames() ([]string, error) { + serviceNameList := make([]string, 0) + var pageNum uint32 = 1 + var pageSize uint32 = 100 + // If the number of serviceInfos is less than the page size, it means that the last page has been reached + for { + serviceNames, err := lw.listPage(pageNum, pageSize) + if err != nil { + return nil, err + } + serviceNameList = append(serviceNameList, serviceNames...) + if len(serviceNames) <= int(pageNum*pageSize) { Review Comment: The condition logic is incorrect. This condition checks if the number of service names is less than or equal to `pageNum * pageSize`, but it should check if it's less than `pageSize` to determine if we've reached the last page. The current logic will incorrectly break the loop prematurely in most cases. ########## pkg/core/discovery/subscriber/rpc_instance.go: ########## @@ -0,0 +1,208 @@ +/* + * 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 subscriber + +import ( + "reflect" + + "github.com/duke-git/lancet/v2/strutil" + "k8s.io/client-go/tools/cache" + + "github.com/apache/dubbo-admin/pkg/common/bizerror" + "github.com/apache/dubbo-admin/pkg/common/constants" + enginecfg "github.com/apache/dubbo-admin/pkg/config/engine" + "github.com/apache/dubbo-admin/pkg/core/events" + "github.com/apache/dubbo-admin/pkg/core/logger" + meshresource "github.com/apache/dubbo-admin/pkg/core/resource/apis/mesh/v1alpha1" + coremodel "github.com/apache/dubbo-admin/pkg/core/resource/model" + "github.com/apache/dubbo-admin/pkg/core/store" + "github.com/apache/dubbo-admin/pkg/core/store/index" +) + +type RPCInstanceEventSubscriber struct { + instanceStore store.ResourceStore + rtInstanceStore store.ResourceStore + eventEmitter events.Emitter + engineCfg *enginecfg.Config +} + +func NewRPCInstanceEventSubscriber( + instanceStore store.ResourceStore, + rtInstanceStore store.ResourceStore, + emitter events.Emitter, + engineCfg *enginecfg.Config) *RPCInstanceEventSubscriber { + return &RPCInstanceEventSubscriber{ + instanceStore: instanceStore, + rtInstanceStore: rtInstanceStore, + eventEmitter: emitter, + engineCfg: engineCfg, + } +} + +func (s *RPCInstanceEventSubscriber) Name() string { + return "Discovery-" + s.ResourceKind().ToString() +} + +func (s *RPCInstanceEventSubscriber) ResourceKind() coremodel.ResourceKind { + return meshresource.RPCInstanceKind +} + +func (s *RPCInstanceEventSubscriber) ProcessEvent(event events.Event) error { + newObj, ok := event.NewObj().(*meshresource.RPCInstanceResource) + if !ok && event.NewObj() != nil { + return bizerror.NewAssertionError(meshresource.RPCInstanceKind, event.NewObj()) + } + oldObj, ok := event.OldObj().(*meshresource.RPCInstanceResource) + if !ok && event.OldObj() != nil { + return bizerror.NewAssertionError(meshresource.RPCInstanceKind, event.OldObj()) + } + var processErr error + switch event.Type() { + case cache.Added, cache.Updated, cache.Replaced, cache.Sync: + if newObj == nil { + errStr := "process rpc instance upsert event, but new obj is nil, skipped processing" + logger.Errorf(errStr) + return bizerror.New(bizerror.EventError, errStr) + } + processErr = s.processUpsert(newObj) + case cache.Deleted: + if oldObj == nil { + errStr := "process rpc instance delete event, but old obj is nil, skipped processing" + logger.Errorf(errStr) + return bizerror.New(bizerror.EventError, errStr) + } + processErr = s.processDelete(oldObj) + } + if processErr != nil { + logger.Errorf("process rpc instance event failed, cause: %s, event: %s", processErr.Error(), event.String()) + return processErr + } + logger.Infof("process rpc instance event successfully, event: %s", event.String()) + return nil +} + +func (s *RPCInstanceEventSubscriber) processUpsert(rpcInstanceRes *meshresource.RPCInstanceResource) error { + // If the conditions follow are not met, we cannot identify it as a dubbo instance, so we skip it + if !s.checkAttributeEnough(rpcInstanceRes) { + logger.Warnf("cannot identify rpc instance %s as a dubbo instance, skipped updating instance", rpcInstanceRes.Name) + return nil + } + instanceRes, err := s.getRelatedInstanceRes(rpcInstanceRes) + if err != nil { + return err + } + // if instance resource exists, that is to say the runtime instance exists and has been watched by engine + // We should merge the rpc info into it + if instanceRes != nil { + meshresource.MergeRPCInstanceIntoInstance(rpcInstanceRes, instanceRes) + return s.instanceStore.Update(instanceRes) + } + // Otherwise we can create a new instance resource by rpc instance + instanceRes = meshresource.FromRPCInstance(rpcInstanceRes) + s.findRelatedRuntimeInstanceAndMerge(instanceRes) + if err := s.instanceStore.Add(instanceRes); err != nil { + logger.Errorf("add instance resource failed, instance: %s, err: %s", instanceRes.ResourceKey(), err.Error()) + return err + } + + instanceAddEvent := events.NewResourceChangedEvent(cache.Added, nil, instanceRes) + s.eventEmitter.Send(instanceAddEvent) + logger.Debugf("rpc instance upsert trigger instance add event, event: %s", instanceAddEvent.String()) + return nil +} + +func (s *RPCInstanceEventSubscriber) processDelete(rpcInstanceRes *meshresource.RPCInstanceResource) error { + if !s.checkAttributeEnough(rpcInstanceRes) { + logger.Warnf("cannot identify rpc instance %s as a dubbo instance, skipped deleting instance", rpcInstanceRes.Name) + return nil + } + instanceRes, err := s.getRelatedInstanceRes(rpcInstanceRes) + if err != nil { + return err + } + if instanceRes == nil { + logger.Warnf("cannot find instance resource for rpc instance %s, skipped deleting instance", rpcInstanceRes.Name) + return nil + } + if err := s.instanceStore.Delete(instanceRes); err != nil { + logger.Errorf("delete instance resource failed, instance: %s, err: %s", instanceRes.ResourceKey(), err.Error()) + return err + } + instanceDeleteEvent := events.NewResourceChangedEvent(cache.Deleted, instanceRes, nil) + s.eventEmitter.Send(instanceDeleteEvent) + logger.Debugf("rpc instance delete trigger instance delete event, event: %s", instanceDeleteEvent.String()) + return nil +} + +func (s *RPCInstanceEventSubscriber) checkAttributeEnough(rpcInstanceRes *meshresource.RPCInstanceResource) bool { + if rpcInstanceRes.Spec == nil || strutil.IsBlank(rpcInstanceRes.Spec.AppName) || + strutil.IsBlank(rpcInstanceRes.Spec.Ip) || rpcInstanceRes.Spec.Port <= 0 || + strutil.IsBlank(rpcInstanceRes.Mesh) || constants.DefaultMesh == rpcInstanceRes.Mesh { + return false + } + return true +} + +func (s *RPCInstanceEventSubscriber) getRelatedInstanceRes( + rpcInstanceRes *meshresource.RPCInstanceResource) (*meshresource.InstanceResource, error) { + instanceResName := meshresource.BuildInstanceResName(rpcInstanceRes.Spec.AppName, rpcInstanceRes.Spec.Ip, rpcInstanceRes.Spec.Port) + res, exists, err := s.instanceStore.GetByKey(coremodel.BuildResourceKey(rpcInstanceRes.Mesh, instanceResName)) + if err != nil { + return nil, err + } + if !exists { + return nil, nil + } + instanceRes, ok := res.(*meshresource.InstanceResource) + if !ok { + return nil, bizerror.NewAssertionError(meshresource.InstanceKind, reflect.TypeOf(res).Name()) + } + return instanceRes, nil +} + +func (s *RPCInstanceEventSubscriber) findRelatedRuntimeInstanceAndMerge(instanceRes *meshresource.InstanceResource) { + switch s.engineCfg.Type { + case enginecfg.Kubernetes: + rtInstance := s.getRuntimeInstanceByIp(instanceRes.Spec.Ip) + if rtInstance == nil { + logger.Warnf("cannot find runtime instance for isntace %s, skipping merging", instanceRes.ResourceKey()) Review Comment: Typo in the warning message: "isntace" should be "instance". ########## pkg/store/dbcommon/gorm_store.go: ########## @@ -279,7 +277,7 @@ func (gs *GormStore) GetByKey(key string) (item interface{}, exists bool, err er First(&m) if result.Error != nil { - if errors.Is(result.Error, gorm.ErrRecordNotFound) { + if result.Error.Error() == "record not found" { Review Comment: Using string comparison `result.Error.Error() == "record not found"` instead of `errors.Is(result.Error, gorm.ErrRecordNotFound)` is fragile and could break if the error message format changes in gorm. The original approach using `errors.Is` is more robust and follows Go best practices for error checking. ########## pkg/core/resource/apis/mesh/v1alpha1/instance_helper.go: ########## @@ -0,0 +1,80 @@ +/* + * 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 v1alpha1 + +import ( + "fmt" +) + +func BuildInstanceResName(appName string, ip string, rpcProt int64) string { Review Comment: Typo in the parameter name: "rpcProt" should be "rpcPort". ########## pkg/discovery/nacos2/listerwatcher/nacos_service.go: ########## @@ -0,0 +1,273 @@ +/* + * 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 listerwatcher + +import ( + "fmt" + "time" + + "github.com/duke-git/lancet/v2/slice" + "github.com/go-co-op/gocron" + nacosnamingclient "github.com/nacos-group/nacos-sdk-go/v2/clients/naming_client" + nacosmodel "github.com/nacos-group/nacos-sdk-go/v2/model" + nacosvo "github.com/nacos-group/nacos-sdk-go/v2/vo" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + k8sruntime "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/tools/cache" + + meshproto "github.com/apache/dubbo-admin/api/mesh/v1alpha1" + "github.com/apache/dubbo-admin/pkg/common/bizerror" + discoverycfg "github.com/apache/dubbo-admin/pkg/config/discovery" + "github.com/apache/dubbo-admin/pkg/core/logger" + meshresource "github.com/apache/dubbo-admin/pkg/core/resource/apis/mesh/v1alpha1" + coremodel "github.com/apache/dubbo-admin/pkg/core/resource/model" +) + +type NacosServiceListerWatcher struct { + cfg *discoverycfg.Config + namingClient nacosnamingclient.INamingClient + // watchingServices is used to record the services that are currently being watched + // key -> nacos service name, value -> true if it is watched in current list turn + watchingServices map[string]bool + scheduler *gocron.Scheduler + resultChan chan watch.Event + stopWatch bool +} + +func NewNacosServiceListerWatcher(cfg *discoverycfg.Config, namingClient nacosnamingclient.INamingClient) *NacosServiceListerWatcher { + return &NacosServiceListerWatcher{ + cfg: cfg, + namingClient: namingClient, + resultChan: make(chan watch.Event), + watchingServices: make(map[string]bool), + scheduler: gocron.NewScheduler(time.UTC), + stopWatch: false, + } +} + +func (lw *NacosServiceListerWatcher) List(_ metav1.ListOptions) (k8sruntime.Object, error) { + serviceNames, err := lw.fetchAllServiceNames() + if err != nil { + return nil, err + } + nacosServiceList := meshresource.NewNacosServiceResourceListWithItems() + resList := make([]*meshresource.NacosServiceResource, 0) + for _, serviceName := range serviceNames { + service, err := lw.namingClient.GetService(nacosvo.GetServiceParam{ + ServiceName: serviceName, + }) + if err != nil { + logger.Errorf("get instances of service %s failed in nacos %s, cause: %v", serviceName, lw.nacosAddress(), err) + continue + } + resList = append(resList, lw.toNacosServiceResource(serviceName, service.Hosts)) + } + nacosServiceList.Items = resList + return nacosServiceList, nil +} + +func (lw *NacosServiceListerWatcher) Watch(_ metav1.ListOptions) (watch.Interface, error) { + _, err := lw.scheduler.Every(lw.cfg.Properties.ServiceWatchPeriod).Seconds().Do(func() { + if lw.stopWatch { + logger.Debugf("stop watch all services of nacos %s", lw.mesh()) + lw.scheduler.Stop() + return + } + startTime := time.Now() + err := lw.fetchAndProcessService() + if err != nil { + logger.Errorf("fetch all service failed in nacos %s, cause: %v", lw.nacosAddress(), err) + return + } + costs := time.Now().UnixMilli() - startTime.UnixMilli() + logger.Debugf("finish fetching and processing all services in nacos %s, costs %dms", lw.nacosAddress(), costs) + }) + if err != nil { + return nil, bizerror.Wrap(err, bizerror.UnknownError, + fmt.Sprintf("watch all services of nacos %s in a schedule occurs error", lw.mesh())) + } + lw.scheduler.StartAsync() + return lw, nil +} + +func (lw *NacosServiceListerWatcher) fetchAllServiceNames() ([]string, error) { + serviceNameList := make([]string, 0) + var pageNum uint32 = 1 + var pageSize uint32 = 100 + // If the number of serviceInfos is less than the page size, it means that the last page has been reached + for { + serviceNames, err := lw.listPage(pageNum, pageSize) + if err != nil { + return nil, err + } + serviceNameList = append(serviceNameList, serviceNames...) + if len(serviceNames) <= int(pageNum*pageSize) { + break + } + pageNum++ + } + return serviceNameList, nil +} + +func (lw *NacosServiceListerWatcher) fetchAndProcessService() error { + var pageNum uint32 = 1 + var pageSize uint32 = 100 + // list all services by page search and subscribe it + for { + serviceNames, err := lw.listPage(pageNum, pageSize) + if err != nil { + return err + } + slice.ForEach(serviceNames, func(index int, item string) { + lw.processNacosService(item) + }) + if len(serviceNames) <= int(pageNum*pageSize) { Review Comment: The same pagination logic issue exists here. The condition should check if `len(serviceNames) < pageSize` to determine if we've reached the last page, not `len(serviceNames) <= pageNum*pageSize`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
