go/go-micro
1
0
Fork 0
mirror of https://github.com/go-micro/go-micro.git synced 2025-10-30 23:27:41 +02:00
Code Issues Releases Activity

[fix] zookeeper registry delete event (#2372)

Browse Source 
* [fix] #2358 zookeeper delete event
This commit is contained in:
Johnson C
2021-12-01 16:26:31 +08:00
committed by GitHub
parent b25d744f5c
commit 1e4dd94b71
8 changed files with 45 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
plugins/registry/kubernetes/client/client.go
View File

@@ -7,9 +7,9 @@ import (
"os"
"path"
log "go-micro.dev/v4/logger"
"github.com/asim/go-micro/plugins/registry/kubernetes/v4/client/api"
"github.com/asim/go-micro/plugins/registry/kubernetes/v4/client/watch"
"go-micro.dev/v4/logger"
)
var (
@@ -90,26 +90,26 @@ func NewClientInCluster() Kubernetes {
s, err := os.Stat(serviceAccountPath)
if err != nil {
log.Fatal(err)
logger.Fatal(err)
}
if s == nil || !s.IsDir() {
log.Fatal(errors.New("no k8s service account found"))
logger.Fatal(errors.New("no k8s service account found"))
}
token, err := os.ReadFile(path.Join(serviceAccountPath, "token"))
if err != nil {
log.Fatal(err)
logger.Fatal(err)
}
t := string(token)
ns, err := detectNamespace()
if err != nil {
log.Fatal(err)
logger.Fatal(err)
}
crt, err := CertPoolFromFile(path.Join(serviceAccountPath, "ca.crt"))
if err != nil {
log.Fatal(err)
logger.Fatal(err)
}
c := &http.Client{

8
plugins/registry/kubernetes/client/kubernetes.go
View File

@@ -22,10 +22,10 @@ type Pod struct {
// Meta ...
type Meta struct {
Name string `json:"name,omitempty"`
Labels map[string]*string `json:"labels,omitempty"`
Annotations map[string]*string `json:"annotations,omitempty"`
DeletionTimestamp string `json:"deletionTimestamp,omitempty"`
Name string `json:"name,omitempty"`
Labels map[string]*string `json:"labels,omitempty"`
Annotations map[string]*string `json:"annotations,omitempty"`
DeletionTimestamp string `json:"deletionTimestamp,omitempty"`
}
// Status ...

2
plugins/registry/kubernetes/go.mod
View File

@@ -2,7 +2,7 @@ module github.com/asim/go-micro/plugins/registry/kubernetes/v4
go 1.17
require go-micro.dev/v4 v4.2.1
require go-micro.dev/v4 v4.4.0
require (
github.com/Microsoft/go-winio v0.5.0 // indirect

19
plugins/registry/kubernetes/kubernetes.go
View File

@@ -11,7 +11,6 @@ import (
"time"
"github.com/asim/go-micro/plugins/registry/kubernetes/v4/client"
"go-micro.dev/v4/cmd"
"go-micro.dev/v4/registry"
)
@@ -140,7 +139,6 @@ func (c *kregistry) Register(s *registry.Service, opts ...registry.RegisterOptio
}
return nil
}
// Deregister nils out any things set in Register
@@ -231,8 +229,8 @@ func (c *kregistry) ListServices(opts ...registry.ListOption) ([]*registry.Servi
return nil, err
}
// svcs mapped by name
svcs := make(map[string]bool)
// svcs mapped by name+version
svcs := make(map[string]*registry.Service)
for _, pod := range pods.Items {
if pod.Status.Phase != podRunning || pod.Metadata.DeletionTimestamp != "" {
@@ -249,13 +247,18 @@ func (c *kregistry) ListServices(opts ...registry.ListOption) ([]*registry.Servi
if err := json.Unmarshal([]byte(*v), &svc); err != nil {
continue
}
svcs[svc.Name] = true
s, ok := svcs[svc.Name+svc.Version]
if !ok {
svcs[svc.Name+svc.Version] = &svc
continue
}
// append to service:version nodes
s.Nodes = append(s.Nodes, svc.Nodes...)
}
}
var list []*registry.Service
for val := range svcs {
list = append(list, &registry.Service{Name: val})
for _, s := range svcs {
list = append(list, s)
}
return list, nil
}

18
plugins/registry/kubernetes/kubernetes_test.go
View File

@@ -9,10 +9,10 @@ import (
"testing"
"time"
log "go-micro.dev/v4/logger"
"go-micro.dev/v4/registry"
"github.com/asim/go-micro/plugins/registry/kubernetes/v4/client"
"github.com/asim/go-micro/plugins/registry/kubernetes/v4/client/mock"
"go-micro.dev/v4/logger"
"go-micro.dev/v4/registry"
)
var (
@@ -58,7 +58,7 @@ func register(r registry.Registry, podName string, svc *registry.Service) {
})
if err := r.Register(svc); err != nil {
log.Fatalf("did not expect Register() to fail: %v", err)
logger.Fatalf("did not expect Register() to fail: %v", err)
}
os.Setenv("HOSTNAME", "")
@@ -379,8 +379,8 @@ func TestListServices(t *testing.T) {
r := setupRegistry()
defer teardownRegistry()
svc1 := &registry.Service{Name: "foo.service"}
svc2 := &registry.Service{Name: "bar.service"}
svc1 := &registry.Service{Name: "foo.service", Version: "1"}
svc2 := &registry.Service{Name: "bar.service", Version: "2"}
register(r, "pod-1", svc1)
register(r, "pod-2", svc2)
@@ -390,8 +390,8 @@ func TestListServices(t *testing.T) {
t.Fatalf("did not expect ListServices to fail %v", err)
}
if !hasServices(services, []*registry.Service{
{Name: "foo.service"},
{Name: "bar.service"},
{Name: "foo.service", Version: "1"},
{Name: "bar.service", Version: "2"},
}) {
t.Fatal("expected services to equal")
}
@@ -403,7 +403,7 @@ func TestListServices(t *testing.T) {
t.Fatalf("did not expect ListServices to fail %v", err)
}
if !hasServices(services2, []*registry.Service{
{Name: "bar.service"},
{Name: "bar.service", Version: "2"},
}) {
t.Fatal("expected services to equal")
}
@@ -427,7 +427,7 @@ func TestWatcher(t *testing.T) {
// check that service is blank
if _, err := r.GetService("foo.service"); err != registry.ErrNotFound {
log.Fatal("expected ErrNotFound")
logger.Fatal("expected ErrNotFound")
}
// setup svc

14
plugins/registry/kubernetes/watcher.go
View File

@@ -6,10 +6,10 @@ import (
"strings"
"sync"
log "go-micro.dev/v4/logger"
"go-micro.dev/v4/registry"
"github.com/asim/go-micro/plugins/registry/kubernetes/v4/client"
"github.com/asim/go-micro/plugins/registry/kubernetes/v4/client/watch"
"go-micro.dev/v4/logger"
"go-micro.dev/v4/registry"
)
type k8sWatcher struct {
@@ -28,17 +28,11 @@ func (k *k8sWatcher) updateCache() ([]*registry.Result, error) {
return nil, err
}
k.RLock()
k.RUnlock()
var results []*registry.Result
for _, pod := range podList.Items {
rslts := k.buildPodResults(&pod, nil)
for _, r := range rslts {
results = append(results, r)
}
results = append(results, rslts...)
k.Lock()
k.pods[pod.Metadata.Name] = &pod
@@ -131,7 +125,7 @@ func (k *k8sWatcher) buildPodResults(pod *client.Pod, cache *client.Pod) []*regi
func (k *k8sWatcher) handleEvent(event watch.Event) {
var pod client.Pod
if err := json.Unmarshal([]byte(event.Object), &pod); err != nil {
log.Error("K8s Watcher: Couldnt unmarshal event object from pod")
logger.Error("K8s Watcher: Couldnt unmarshal event object from pod")
return
}

2
plugins/registry/zookeeper/go.mod
View File

@@ -5,7 +5,7 @@ go 1.17
require (
github.com/go-zookeeper/zk v1.0.2
github.com/mitchellh/hashstructure v1.1.0
go-micro.dev/v4 v4.2.1
go-micro.dev/v4 v4.4.0
)
require (

10
plugins/registry/zookeeper/watcher.go
View File

@@ -4,8 +4,8 @@ import (
"errors"
"path"
"go-micro.dev/v4/registry"
"github.com/go-zookeeper/zk"
"go-micro.dev/v4/registry"
)
type zookeeperWatcher struct {
@@ -207,6 +207,14 @@ func (zw *zookeeperWatcher) Next() (*registry.Result, error) {
}
func (zw *zookeeperWatcher) overrideChildrenInfo(newChildren []string, parentPath string) {
// nodes were removed
if len(newChildren) == 0 {
zw.writeRespChan(&watchResponse{zk.Event{
Type: zk.EventNodeCreated,
}, nil, nil})
return
}
// override resp
var overrideResp *watchResponse