Copilot commented on code in PR #779:
URL:
https://github.com/apache/incubator-seata-go/pull/779#discussion_r2263332144
##########
pkg/discovery/zk.go:
##########
@@ -14,17 +14,247 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package discovery
-type ZkRegistryService struct{}
+import (
+ "fmt"
+ "github.com/go-zookeeper/zk"
+ "seata.apache.org/seata-go/pkg/discovery/mock"
+ "seata.apache.org/seata-go/pkg/util/log"
+ "strconv"
+ "strings"
+ "sync"
+)
+
Review Comment:
Production code should not import mock packages. The mock import should be
conditionally imported or the interface should be defined in the main package
to avoid coupling production code with test utilities.
```suggestion
"seata.apache.org/seata-go/pkg/util/log"
"strconv"
"strings"
"sync"
)
// ZkConnInterface abstracts the methods used from zk.Conn for easier
testing and mocking.
type ZkConnInterface interface {
GetW(path string) ([]byte, *zk.Stat, <-chan zk.Event, error)
Get(path string) ([]byte, *zk.Stat, error)
Close() error
}
```
##########
pkg/discovery/zk.go:
##########
@@ -14,17 +14,247 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package discovery
-type ZkRegistryService struct{}
+import (
+ "fmt"
+ "github.com/go-zookeeper/zk"
+ "seata.apache.org/seata-go/pkg/discovery/mock"
+ "seata.apache.org/seata-go/pkg/util/log"
+ "strconv"
+ "strings"
+ "sync"
+)
+
+// zkConnAdapter wraps a real *zk.Conn to implement ZkConnInterface.
+type zkConnAdapter struct {
+ conn *zk.Conn
+}
+
+func (a *zkConnAdapter) GetW(path string) ([]byte, *zk.Stat, <-chan zk.Event,
error) {
+ return a.conn.GetW(path)
+}
+
+func (a *zkConnAdapter) Get(path string) ([]byte, *zk.Stat, error) {
+ return a.conn.Get(path)
+}
+
+func (a *zkConnAdapter) Close() error {
+ a.conn.Close()
+ return nil
+}
+
+const (
+ zookeeperClusterPrefix = "/registry-seata"
+)
+
+type ZookeeperRegistryService struct {
+ conn mock.ZkConnInterface
+ vgroupMapping map[string]string
+ grouplist map[string][]*ServiceInstance
+ rwLock sync.RWMutex
+ stopCh chan struct{}
+}
+
+func newZookeeperRegistryService(config *ServiceConfig, zkConfig
*ZookeeperConfig) RegistryService {
+ if zkConfig == nil {
+ log.Fatalf("zookeeper config is nil")
+ panic("zookeeper config is nil")
+ }
+
+ // Connect to the actual *zk.Conn
+ conn, _, err := zk.Connect([]string{zkConfig.ServerAddr},
zkConfig.SessionTimeout)
+ if err != nil {
+ log.Fatalf("failed to create zookeeper client")
+ panic("failed to create zookeeper client")
+ }
+
+ // Wrap it into an adapter
+ adapter := &zkConnAdapter{conn: conn}
+
+ vgroupMapping := config.VgroupMapping
+
+ // init groplist
+ grouplist, err := initFromServiceConfig(config)
+ if err != nil {
+ log.Errorf("Error initializing service config: %v", err)
+ return nil
+ }
+
+ zkRegistryService := &ZookeeperRegistryService{
+ conn: adapter,
+ vgroupMapping: vgroupMapping,
+ grouplist: grouplist,
+ stopCh: make(chan struct{}),
+ }
+
+ //go zkRegistryService.watch(zookeeperClusterPrefix)
+ go zkRegistryService.watch(zkConfig.NodePath)
+
+ return zkRegistryService
+}
+
+func (s *ZookeeperRegistryService) watch(path string) {
+ for {
+ // Get initial data and set the watch
+ data, _, events, err := s.conn.GetW(path)
+ if err != nil {
+ log.Infof("Failed to get server instances from
Zookeeper: %v", err)
+ return
+ }
-func (s *ZkRegistryService) Lookup(key string) ([]*ServiceInstance, error) {
- //TODO implement me
- panic("implement me")
+ // Handle initial data
+ s.handleZookeeperData(path, data)
+
+ // Listen for changes to Zookeeper nodes
+ for {
+ select {
+ case event := <-events:
+ // Re-establish the watch to continue listening
for changes
+ data, _, events, err = s.conn.GetW(path)
+ if err != nil {
+ log.Errorf("Failed to set watch on
Zookeeper node: %v", err)
+ return
+ }
+ switch event.Type {
+ case zk.EventNodeCreated,
zk.EventNodeDataChanged:
+ log.Infof("Node updated: %s",
event.Path)
+ s.handleZookeeperData(event.Path, data)
+
+ case zk.EventNodeDeleted:
+ log.Infof("Node deleted: %s",
event.Path)
+ s.removeServiceInstance(event.Path)
+ }
+
+ case <-s.stopCh:
+ log.Warn("Received stop signal, stopping
watch.")
+ return
+ }
+ }
+ }
+}
+
+func (s *ZookeeperRegistryService) handleZookeeperData(path string, data
[]byte) {
+ clusterName, serverInstance, err := parseZookeeperData(path, data)
+ if err != nil {
+ log.Errorf("Zookeeper data error: %s", err)
+ return
+ }
+
+ s.rwLock.Lock()
+ if s.grouplist[clusterName] == nil {
+ s.grouplist[clusterName] = []*ServiceInstance{serverInstance}
+ } else {
+ s.grouplist[clusterName] = append(s.grouplist[clusterName],
serverInstance)
+ }
+ s.rwLock.Unlock()
+}
+
+func (s *ZookeeperRegistryService) removeServiceInstance(path string) {
+ clusterName, ip, port, err := parseClusterAndAddress(path)
+ if err != nil {
+ log.Errorf("Zookeeper path error: %s", err)
+ return
+ }
+
+ s.rwLock.Lock()
+ serviceInstances := s.grouplist[clusterName]
+ if serviceInstances == nil {
+ log.Warnf("Zookeeper doesn't exist cluster: %s", clusterName)
+ s.rwLock.Unlock()
+ return
+ }
+ s.grouplist[clusterName] = removeValueFromList(serviceInstances, ip,
port)
+ s.rwLock.Unlock()
+}
+
+func parseZookeeperData(path string, data []byte) (string, *ServiceInstance,
error) {
+ parts := strings.Split(string(data), addressSplitChar)
+ if len(parts) != 2 {
+ return "", nil, fmt.Errorf("Zookeeper data has incorrect
format: %s", string(data))
+ }
+ ip := parts[0]
+ port, err := strconv.Atoi(parts[1])
+ if err != nil {
+ return "", nil, fmt.Errorf("Invalid port in Zookeeper data:
%w", err)
+ }
+ clusterName := strings.TrimPrefix(path, zookeeperClusterPrefix+"/")
Review Comment:
The function uses hardcoded `zookeeperClusterPrefix` but the actual path
being watched is `zkConfig.NodePath`. This will cause incorrect cluster name
parsing when a custom node path is used.
```suggestion
clusterName := strings.TrimPrefix(path, nodePathPrefix+"/")
```
##########
pkg/discovery/zk.go:
##########
@@ -14,17 +14,247 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package discovery
-type ZkRegistryService struct{}
+import (
+ "fmt"
+ "github.com/go-zookeeper/zk"
+ "seata.apache.org/seata-go/pkg/discovery/mock"
+ "seata.apache.org/seata-go/pkg/util/log"
+ "strconv"
+ "strings"
+ "sync"
+)
+
+// zkConnAdapter wraps a real *zk.Conn to implement ZkConnInterface.
+type zkConnAdapter struct {
+ conn *zk.Conn
+}
+
+func (a *zkConnAdapter) GetW(path string) ([]byte, *zk.Stat, <-chan zk.Event,
error) {
+ return a.conn.GetW(path)
+}
+
+func (a *zkConnAdapter) Get(path string) ([]byte, *zk.Stat, error) {
+ return a.conn.Get(path)
+}
+
+func (a *zkConnAdapter) Close() error {
+ a.conn.Close()
+ return nil
+}
+
+const (
+ zookeeperClusterPrefix = "/registry-seata"
+)
+
+type ZookeeperRegistryService struct {
+ conn mock.ZkConnInterface
+ vgroupMapping map[string]string
+ grouplist map[string][]*ServiceInstance
+ rwLock sync.RWMutex
+ stopCh chan struct{}
+}
+
+func newZookeeperRegistryService(config *ServiceConfig, zkConfig
*ZookeeperConfig) RegistryService {
+ if zkConfig == nil {
+ log.Fatalf("zookeeper config is nil")
+ panic("zookeeper config is nil")
+ }
+
+ // Connect to the actual *zk.Conn
+ conn, _, err := zk.Connect([]string{zkConfig.ServerAddr},
zkConfig.SessionTimeout)
+ if err != nil {
+ log.Fatalf("failed to create zookeeper client")
+ panic("failed to create zookeeper client")
+ }
+
+ // Wrap it into an adapter
+ adapter := &zkConnAdapter{conn: conn}
+
+ vgroupMapping := config.VgroupMapping
+
+ // init groplist
+ grouplist, err := initFromServiceConfig(config)
+ if err != nil {
+ log.Errorf("Error initializing service config: %v", err)
+ return nil
+ }
+
+ zkRegistryService := &ZookeeperRegistryService{
+ conn: adapter,
+ vgroupMapping: vgroupMapping,
+ grouplist: grouplist,
+ stopCh: make(chan struct{}),
+ }
+
+ //go zkRegistryService.watch(zookeeperClusterPrefix)
Review Comment:
Remove commented out code. Dead code should be removed to maintain code
cleanliness.
```suggestion
```
##########
pkg/discovery/mock/mock_zk_client.go:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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 MockGen. DO NOT EDIT.
+// Source: test_etcd_client.go
Review Comment:
The comment indicates this was generated from 'test_etcd_client.go' but this
is a ZooKeeper mock. The source comment should reference the correct file.
```suggestion
```
--
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]
