zeromicro/go-zero
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: support third party orm to interact with go-zero (#1286)

Browse Source 
* fixes #987

* chore: fix test failure

* chore: add comments

* feat: support third party orm to interact with go-zero

* chore: refactor
This commit is contained in:
Kevin Wan
2021-12-01 20:22:15 +08:00
committed by GitHub
parent 543d590710
commit 9d528dddd6
24 changed files with 55 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
zrpc/resolver/internal/kube/deploy/clusterrole.yaml Normal file
View File

@@ -0,0 +1,9 @@
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
name: endpoints-reader
rules:
- apiGroups: [""]
resources: ["endpoints"]
verbs: ["get", "watch", "list"]

12
zrpc/resolver/internal/kube/deploy/clusterrolebinding.yaml Normal file
View File

@@ -0,0 +1,12 @@
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
name: endpoints-reader
subjects:
- kind: ServiceAccount
name: endpoints-reader
namespace: kevin # the namespace that the ServiceAccount resides in
roleRef:
kind: ClusterRole
name: endpoints-reader
apiGroup: rbac.authorization.k8s.io

5
zrpc/resolver/internal/kube/deploy/serviceaccount.yaml Normal file
View File

@@ -0,0 +1,5 @@
apiVersion: v1
kind: ServiceAccount
metadata:
name: endpoints-reader
namespace: kevin # the namespace to create the ServiceAccount

139
zrpc/resolver/internal/kube/eventhandler.go Normal file
View File

@@ -0,0 +1,139 @@
package kube
import (
"sync"
"github.com/tal-tech/go-zero/core/lang"
"github.com/tal-tech/go-zero/core/logx"
v1 "k8s.io/api/core/v1"
)
// EventHandler is ResourceEventHandler implementation.
type EventHandler struct {
update func([]string)
endpoints map[string]lang.PlaceholderType
lock sync.Mutex
}
// NewEventHandler returns an EventHandler.
func NewEventHandler(update func([]string)) *EventHandler {
return &EventHandler{
update: update,
endpoints: make(map[string]lang.PlaceholderType),
}
}
// OnAdd handles the endpoints add events.
func (h *EventHandler) OnAdd(obj interface{}) {
endpoints, ok := obj.(*v1.Endpoints)
if !ok {
logx.Errorf("%v is not an object with type *v1.Endpoints", obj)
return
}
h.lock.Lock()
defer h.lock.Unlock()
var changed bool
for _, sub := range endpoints.Subsets {
for _, point := range sub.Addresses {
if _, ok := h.endpoints[point.IP]; !ok {
h.endpoints[point.IP] = lang.Placeholder
changed = true
}
}
}
if changed {
h.notify()
}
}
// OnDelete handles the endpoints delete events.
func (h *EventHandler) OnDelete(obj interface{}) {
endpoints, ok := obj.(*v1.Endpoints)
if !ok {
logx.Errorf("%v is not an object with type *v1.Endpoints", obj)
return
}
h.lock.Lock()
defer h.lock.Unlock()
var changed bool
for _, sub := range endpoints.Subsets {
for _, point := range sub.Addresses {
if _, ok := h.endpoints[point.IP]; ok {
delete(h.endpoints, point.IP)
changed = true
}
}
}
if changed {
h.notify()
}
}
// OnUpdate handles the endpoints update events.
func (h *EventHandler) OnUpdate(oldObj, newObj interface{}) {
oldEndpoints, ok := oldObj.(*v1.Endpoints)
if !ok {
logx.Errorf("%v is not an object with type *v1.Endpoints", oldObj)
return
}
newEndpoints, ok := newObj.(*v1.Endpoints)
if !ok {
logx.Errorf("%v is not an object with type *v1.Endpoints", newObj)
return
}
if oldEndpoints.ResourceVersion == newEndpoints.ResourceVersion {
return
}
h.Update(newEndpoints)
}
// Update updates the endpoints.
func (h *EventHandler) Update(endpoints *v1.Endpoints) {
h.lock.Lock()
defer h.lock.Unlock()
old := h.endpoints
h.endpoints = make(map[string]lang.PlaceholderType)
for _, sub := range endpoints.Subsets {
for _, point := range sub.Addresses {
h.endpoints[point.IP] = lang.Placeholder
}
}
if diff(old, h.endpoints) {
h.notify()
}
}
func (h *EventHandler) notify() {
var targets []string
for k := range h.endpoints {
targets = append(targets, k)
}
h.update(targets)
}
func diff(o, n map[string]lang.PlaceholderType) bool {
if len(o) != len(n) {
return true
}
for k := range o {
if _, ok := n[k]; !ok {
return true
}
}
return false
}

276
zrpc/resolver/internal/kube/eventhandler_test.go Normal file
View File

@@ -0,0 +1,276 @@
package kube
import (
"testing"
"github.com/stretchr/testify/assert"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
func TestAdd(t *testing.T) {
var endpoints []string
h := NewEventHandler(func(change []string) {
endpoints = change
})
h.OnAdd("bad")
h.OnAdd(&v1.Endpoints{Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
{
IP: "0.0.0.3",
},
},
},
}})
assert.ElementsMatch(t, []string{"0.0.0.1", "0.0.0.2", "0.0.0.3"}, endpoints)
}
func TestDelete(t *testing.T) {
var endpoints []string
h := NewEventHandler(func(change []string) {
endpoints = change
})
h.OnAdd(&v1.Endpoints{Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
{
IP: "0.0.0.3",
},
},
},
}})
h.OnDelete("bad")
h.OnDelete(&v1.Endpoints{Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
},
},
}})
assert.ElementsMatch(t, []string{"0.0.0.3"}, endpoints)
}
func TestUpdate(t *testing.T) {
var endpoints []string
h := NewEventHandler(func(change []string) {
endpoints = change
})
h.OnUpdate(&v1.Endpoints{
Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
},
},
},
ObjectMeta: metav1.ObjectMeta{
ResourceVersion: "1",
},
}, &v1.Endpoints{
Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
{
IP: "0.0.0.3",
},
},
},
},
ObjectMeta: metav1.ObjectMeta{
ResourceVersion: "2",
},
})
assert.ElementsMatch(t, []string{"0.0.0.1", "0.0.0.2", "0.0.0.3"}, endpoints)
}
func TestUpdateNoChange(t *testing.T) {
h := NewEventHandler(func(change []string) {
assert.Fail(t, "should not called")
})
h.OnUpdate(&v1.Endpoints{
Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
},
},
},
ObjectMeta: metav1.ObjectMeta{
ResourceVersion: "1",
},
}, &v1.Endpoints{
Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
},
},
},
ObjectMeta: metav1.ObjectMeta{
ResourceVersion: "1",
},
})
}
func TestUpdateChangeWithDifferentVersion(t *testing.T) {
var endpoints []string
h := NewEventHandler(func(change []string) {
endpoints = change
})
h.OnAdd(&v1.Endpoints{Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.3",
},
},
},
}})
h.OnUpdate(&v1.Endpoints{
Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.3",
},
},
},
},
ObjectMeta: metav1.ObjectMeta{
ResourceVersion: "1",
},
}, &v1.Endpoints{
Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
},
},
},
ObjectMeta: metav1.ObjectMeta{
ResourceVersion: "2",
},
})
assert.ElementsMatch(t, []string{"0.0.0.1", "0.0.0.2"}, endpoints)
}
func TestUpdateNoChangeWithDifferentVersion(t *testing.T) {
var endpoints []string
h := NewEventHandler(func(change []string) {
endpoints = change
})
h.OnAdd(&v1.Endpoints{Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
},
},
}})
h.OnUpdate("bad", &v1.Endpoints{Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
},
},
}})
h.OnUpdate(&v1.Endpoints{Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
},
},
}}, "bad")
h.OnUpdate(&v1.Endpoints{
Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
},
},
},
ObjectMeta: metav1.ObjectMeta{
ResourceVersion: "1",
},
}, &v1.Endpoints{
Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
},
},
},
ObjectMeta: metav1.ObjectMeta{
ResourceVersion: "2",
},
})
assert.ElementsMatch(t, []string{"0.0.0.1", "0.0.0.2"}, endpoints)
}

47
zrpc/resolver/internal/kube/targetparser.go Normal file
View File

@@ -0,0 +1,47 @@
package kube
import (
"fmt"
"strconv"
"strings"
"google.golang.org/grpc/resolver"
)
const (
colon = ":"
defaultNamespace = "default"
)
var emptyService Service
// Service represents a service with namespace, name and port.
type Service struct {
Namespace string
Name string
Port int
}
// ParseTarget parses the resolver.Target.
func ParseTarget(target resolver.Target) (Service, error) {
var service Service
service.Namespace = target.Authority
if len(service.Namespace) == 0 {
service.Namespace = defaultNamespace
}
segs := strings.SplitN(target.Endpoint, colon, 2)
if len(segs) < 2 {
return emptyService, fmt.Errorf("bad endpoint: %s", target.Endpoint)
}
service.Name = segs[0]
port, err := strconv.Atoi(segs[1])
if err != nil {
return emptyService, err
}
service.Port = port
return service, nil
}

83
zrpc/resolver/internal/kube/targetparser_test.go Normal file
View File

@@ -0,0 +1,83 @@
package kube
import (
"testing"
"github.com/stretchr/testify/assert"
"google.golang.org/grpc/resolver"
)
func TestParseTarget(t *testing.T) {
tests := []struct {
name string
input resolver.Target
expect Service
hasErr bool
}{
{
name: "normal case",
input: resolver.Target{
Scheme: "k8s",
Authority: "ns1",
Endpoint: "my-svc:8080",
},
expect: Service{
Namespace: "ns1",
Name: "my-svc",
Port: 8080,
},
},
{
name: "normal case",
input: resolver.Target{
Scheme: "k8s",
Authority: "",
Endpoint: "my-svc:8080",
},
expect: Service{
Namespace: defaultNamespace,
Name: "my-svc",
Port: 8080,
},
},
{
name: "no port",
input: resolver.Target{
Scheme: "k8s",
Authority: "ns1",
Endpoint: "my-svc:",
},
hasErr: true,
},
{
name: "no port, no colon",
input: resolver.Target{
Scheme: "k8s",
Authority: "ns1",
Endpoint: "my-svc",
},
hasErr: true,
},
{
name: "bad port",
input: resolver.Target{
Scheme: "k8s",
Authority: "ns1",
Endpoint: "my-svc:800a",
},
hasErr: true,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
svc, err := ParseTarget(test.input)
if test.hasErr {
assert.NotNil(t, err)
} else {
assert.Nil(t, err)
assert.Equal(t, test.expect, svc)
}
})
}
}