forked from st3v/go-plugins
-
Notifications
You must be signed in to change notification settings - Fork 1
/
kubernetes.go
279 lines (228 loc) · 6.05 KB
/
kubernetes.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
// Package kubernetes provides a kubernetes registry
package kubernetes
import (
"encoding/json"
"errors"
"fmt"
"os"
"regexp"
"strings"
"time"
"github.com/micro/go-plugins/registry/kubernetes/client"
"github.com/micro/go-micro/cmd"
"github.com/micro/go-micro/registry"
)
type kregistry struct {
client client.Kubernetes
timeout time.Duration
options registry.Options
}
var (
// used on pods as labels & services to select
// eg: svcSelectorPrefix+"svc.name"
svcSelectorPrefix = "micro.mu/selector-"
svcSelectorValue = "service"
labelTypeKey = "micro.mu/type"
labelTypeValueService = "service"
// used on k8s services to scope a serialised
// micro service by pod name
annotationServiceKeyPrefix = "micro.mu/service-"
// Pod status
podRunning = "Running"
// label name regex
labelRe = regexp.MustCompilePOSIX("[-A-Za-z0-9_.]")
)
// podSelector
var podSelector = map[string]string{
labelTypeKey: labelTypeValueService,
}
func init() {
cmd.DefaultRegistries["kubernetes"] = NewRegistry
}
func configure(k *kregistry, opts ...registry.Option) error {
for _, o := range opts {
o(&k.options)
}
// get first host
var host string
if len(k.options.Addrs) > 0 && len(k.options.Addrs[0]) > 0 {
host = k.options.Addrs[0]
}
if k.options.Timeout == 0 {
k.options.Timeout = time.Second * 1
}
// if no hosts setup, assume InCluster
var c client.Kubernetes
if len(host) == 0 {
c = client.NewClientInCluster()
} else {
c = client.NewClientByHost(host)
}
k.client = c
k.timeout = k.options.Timeout
return nil
}
// serviceName generates a valid service name for k8s labels
func serviceName(name string) string {
aname := make([]byte, len(name))
for i, r := range []byte(name) {
if !labelRe.Match([]byte{r}) {
aname[i] = '_'
continue
}
aname[i] = r
}
return string(aname)
}
// Init allows reconfig of options
func (c *kregistry) Init(opts ...registry.Option) error {
return configure(c, opts...)
}
// Options returns the registry Options
func (c *kregistry) Options() registry.Options {
return c.options
}
// Register sets a service selector label and an annotation with a
// serialised version of the service passed in.
func (c *kregistry) Register(s *registry.Service, opts ...registry.RegisterOption) error {
if len(s.Nodes) == 0 {
return errors.New("you must register at least one node")
}
// TODO: grab podname from somewhere better than this.
podName := os.Getenv("HOSTNAME")
svcName := s.Name
// encode micro service
b, err := json.Marshal(s)
if err != nil {
return err
}
svc := string(b)
pod := &client.Pod{
Metadata: &client.Meta{
Labels: map[string]*string{
labelTypeKey: &labelTypeValueService,
svcSelectorPrefix + serviceName(svcName): &svcSelectorValue,
},
Annotations: map[string]*string{
annotationServiceKeyPrefix + serviceName(svcName): &svc,
},
},
}
if _, err := c.client.UpdatePod(podName, pod); err != nil {
return err
}
return nil
}
// Deregister nils out any things set in Register
func (c *kregistry) Deregister(s *registry.Service) error {
if len(s.Nodes) == 0 {
return errors.New("you must deregister at least one node")
}
// TODO: grab podname from somewhere better than this.
podName := os.Getenv("HOSTNAME")
svcName := s.Name
pod := &client.Pod{
Metadata: &client.Meta{
Labels: map[string]*string{
svcSelectorPrefix + serviceName(svcName): nil,
},
Annotations: map[string]*string{
annotationServiceKeyPrefix + serviceName(svcName): nil,
},
},
}
if _, err := c.client.UpdatePod(podName, pod); err != nil {
return err
}
return nil
}
// GetService will get all the pods with the given service selector,
// and build services from the annotations.
func (c *kregistry) GetService(name string) ([]*registry.Service, error) {
pods, err := c.client.ListPods(map[string]string{
svcSelectorPrefix + serviceName(name): svcSelectorValue,
})
if err != nil {
return nil, err
}
if len(pods.Items) == 0 {
return nil, registry.ErrNotFound
}
// svcs mapped by version
svcs := make(map[string]*registry.Service)
// loop through items
for _, pod := range pods.Items {
if pod.Status.Phase != podRunning {
continue
}
// get serialised service from annotation
svcStr, ok := pod.Metadata.Annotations[annotationServiceKeyPrefix+serviceName(name)]
if !ok {
continue
}
// unmarshal service string
var svc registry.Service
err := json.Unmarshal([]byte(*svcStr), &svc)
if err != nil {
return nil, fmt.Errorf("could not unmarshal service '%s' from pod annotation", name)
}
// merge up pod service & ip with versioned service.
vs, ok := svcs[svc.Version]
if !ok {
svcs[svc.Version] = &svc
continue
}
vs.Nodes = append(vs.Nodes, svc.Nodes...)
}
var list []*registry.Service
for _, val := range svcs {
list = append(list, val)
}
return list, nil
}
// ListServices will list all the service names
func (c *kregistry) ListServices() ([]*registry.Service, error) {
pods, err := c.client.ListPods(podSelector)
if err != nil {
return nil, err
}
// svcs mapped by name
svcs := make(map[string]bool)
for _, pod := range pods.Items {
if pod.Status.Phase != podRunning {
continue
}
for k, v := range pod.Metadata.Annotations {
if !strings.HasPrefix(k, annotationServiceKeyPrefix) {
continue
}
// we have to unmarshal the annotation itself since the
// key is encoded to match the regex restriction.
var svc registry.Service
if err := json.Unmarshal([]byte(*v), &svc); err != nil {
continue
}
svcs[svc.Name] = true
}
}
var list []*registry.Service
for val := range svcs {
list = append(list, ®istry.Service{Name: val})
}
return list, nil
}
// Watch returns a kubernetes watcher
func (c *kregistry) Watch(opts ...registry.WatchOption) (registry.Watcher, error) {
return newWatcher(c, opts...)
}
func (c *kregistry) String() string {
return "kubernetes"
}
// NewRegistry creates a kubernetes registry
func NewRegistry(opts ...registry.Option) registry.Registry {
k := &kregistry{
options: registry.Options{},
}
configure(k, opts...)
return k
}