blob: c86530ffec3b87d2cce79e448cf9fcf612c01072 [file] [log] [blame]
Scott Bakere7144bc2019-10-01 14:16:47 -07001/*
2 * Copyright 2018-present Open Networking Foundation
3
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7
8 * http://www.apache.org/licenses/LICENSE-2.0
9
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17package afrouter
18
19import (
Scott Bakere7144bc2019-10-01 14:16:47 -070020 "fmt"
21 "github.com/golang/protobuf/proto"
22 "github.com/opencord/voltha-go/common/log"
23 "google.golang.org/grpc"
24 "regexp"
25 "strconv"
26)
27
28// TODO: Used in multiple routers, should move to common file
29const (
30 PKG_MTHD_PKG int = 1
31 PKG_MTHD_MTHD int = 2
32)
33
34type AffinityRouter struct {
35 name string
36 association associationType
Scott Bakere7144bc2019-10-01 14:16:47 -070037 grpcService string
38 methodMap map[string]byte
39 nbBindingMethodMap map[string]byte
40 cluster *cluster
41 affinity map[string]*backend
42 currentBackend **backend
43}
44
45func newAffinityRouter(rconf *RouterConfig, config *RouteConfig) (Router, error) {
Scott Baker4989fe92019-10-09 17:03:06 -070046 var err error
Scott Bakere7144bc2019-10-01 14:16:47 -070047 var rtrn_err = false
48 var pkg_re = regexp.MustCompile(`^(\.[^.]+\.)(.+)$`)
49 // Validate the configuration
50
51 // A name must exist
52 if config.Name == "" {
53 log.Error("A router 'name' must be specified")
54 rtrn_err = true
55 }
56
57 if rconf.ProtoPackage == "" {
58 log.Error("A 'package' must be specified")
59 rtrn_err = true
60 }
61
62 if rconf.ProtoService == "" {
63 log.Error("A 'service' must be specified")
64 rtrn_err = true
65 }
66
67 //if config.RouteField == "" {
68 // log.Error("A 'routing_field' must be specified")
69 // rtrn_err = true
70 //}
71
72 // TODO The overrieds section is currently not being used
73 // so the router will route all methods based on the
74 // routing_field. This needs to be added so that methods
75 // can have different routing fields.
76 var bptr *backend
Scott Bakere7144bc2019-10-01 14:16:47 -070077 dr := AffinityRouter{
78 name: config.Name,
79 grpcService: rconf.ProtoService,
80 affinity: make(map[string]*backend),
81 methodMap: make(map[string]byte),
82 nbBindingMethodMap: make(map[string]byte),
83 currentBackend: &bptr,
84 }
85 // An association must exist
86 dr.association = config.Association
87 if dr.association == AssociationUndefined {
88 log.Error("An association must be specified")
89 rtrn_err = true
90 }
91
92 // Build the routing structure based on the loaded protobuf
93 // descriptor file and the config information.
94 type key struct {
95 method string
96 field string
97 }
98 var fieldNumberLookup = make(map[key]byte)
99 for _, f := range rconf.protoDescriptor.File {
100 // Build a temporary map of message types by name.
101 for _, m := range f.MessageType {
102 for _, fld := range m.Field {
103 log.Debugf("Processing message '%s', field '%s'", *m.Name, *fld.Name)
104 fieldNumberLookup[key{*m.Name, *fld.Name}] = byte(*fld.Number)
105 }
106 }
107 }
108 for _, f := range rconf.protoDescriptor.File {
109 if *f.Package == rconf.ProtoPackage {
110 for _, s := range f.Service {
111 if *s.Name == rconf.ProtoService {
112 log.Debugf("Loading package data '%s' for service '%s' for router '%s'", *f.Package, *s.Name, dr.name)
113 // Now create a map keyed by method name with the value being the
114 // field number of the route selector.
115 var ok bool
116 for _, m := range s.Method {
117 // Find the input type in the messages and extract the
118 // field number and save it for future reference.
119 log.Debugf("Processing method '%s'", *m.Name)
120 // Determine if this is a method we're supposed to be processing.
121 if needMethod(*m.Name, config) {
122 log.Debugf("Enabling method '%s'", *m.Name)
123 pkg_methd := pkg_re.FindStringSubmatch(*m.InputType)
124 if pkg_methd == nil {
125 log.Errorf("Regular expression didn't match input type '%s'", *m.InputType)
126 rtrn_err = true
127 }
128 // The input type has the package name prepended to it. Remove it.
129 //in := (*m.InputType)[len(rconf.ProtoPackage)+2:]
130 in := pkg_methd[PKG_MTHD_MTHD]
131 dr.methodMap[*m.Name], ok = fieldNumberLookup[key{in, config.RouteField}]
132 if !ok {
133 log.Errorf("Method '%s' has no field named '%s' in it's parameter message '%s'",
134 *m.Name, config.RouteField, in)
135 rtrn_err = true
136 }
137 }
138 // The sb method is always included in the methods so we can check it here too.
139 if needNbBindingMethod(*m.Name, config) {
140 log.Debugf("Enabling southbound method '%s'", *m.Name)
141 // The output type has the package name prepended to it. Remove it.
142 out := (*m.OutputType)[len(rconf.ProtoPackage)+2:]
143 dr.nbBindingMethodMap[*m.Name], ok = fieldNumberLookup[key{out, config.RouteField}]
144 if !ok {
145 log.Errorf("Method '%s' has no field named '%s' in it's parameter message '%s'",
146 *m.Name, config.RouteField, out)
147 rtrn_err = true
148 }
149 }
150 }
151 }
152 }
153 }
154 }
155
156 // Create the backend cluster or link to an existing one
Scott Baker4989fe92019-10-09 17:03:06 -0700157 var ok bool
Scott Bakere7144bc2019-10-01 14:16:47 -0700158 if dr.cluster, ok = clusters[config.backendCluster.Name]; !ok {
159 if dr.cluster, err = newBackendCluster(config.backendCluster); err != nil {
160 log.Errorf("Could not create a backend for router %s", config.Name)
161 rtrn_err = true
162 }
163 }
164
165 if rtrn_err {
Scott Baker4989fe92019-10-09 17:03:06 -0700166 return dr, fmt.Errorf("Failed to create a new router '%s'", dr.name)
Scott Bakere7144bc2019-10-01 14:16:47 -0700167 }
168
169 return dr, nil
170}
171
172func needNbBindingMethod(mthd string, conf *RouteConfig) bool {
173 for _, m := range conf.NbBindingMethods {
174 if mthd == m {
175 return true
176 }
177 }
178 return false
179}
180
181// TODO: Used in multiple routers, should move to common file
182func needMethod(mthd string, conf *RouteConfig) bool {
183 for _, m := range conf.Methods {
184 if mthd == m {
185 return true
186 }
187 }
188 return false
189}
190
191func (ar AffinityRouter) Service() string {
192 return ar.grpcService
193}
194
195func (ar AffinityRouter) Name() string {
196 return ar.name
197}
198
199func (ar AffinityRouter) skipField(data *[]byte, idx *int) error {
200 switch (*data)[*idx] & 3 {
201 case 0: // Varint
202 // skip the field number/type
203 *idx++
204 // if the msb is set, then more bytes to follow
205 for (*data)[*idx] >= 128 {
206 *idx++
207 }
208 // the last byte doesn't have the msb set
209 *idx++
210 case 1: // 64 bit
211 *idx += 9
212 case 2: // Length delimited
213 *idx++
214 b := proto.NewBuffer((*data)[*idx:])
215 t, _ := b.DecodeVarint()
216 *idx += int(t) + 1
217 case 3: // Deprecated
218 case 4: // Deprecated
219 case 5: // 32 bit
220 *idx += 5
221 }
222 return nil
223}
224
225func (ar AffinityRouter) decodeProtoField(payload []byte, fieldId byte) (string, error) {
226 idx := 0
227 b := proto.NewBuffer([]byte{})
228 //b.DebugPrint("The Buffer", payload)
229 for { // Find the route selector field
Scott Baker8f26db32019-10-22 15:08:21 -0700230 if idx >= len(payload) {
231 err := fmt.Errorf("At end of message, attribute %d not found", fieldId)
232 return "", err
233 }
234
Scott Bakere7144bc2019-10-01 14:16:47 -0700235 log.Debugf("Decoding afinity value attributeNumber: %d from %v at index %d", fieldId, payload, idx)
236 log.Debugf("Attempting match with payload: %d, methodTable: %d", payload[idx], fieldId)
237 if payload[idx]>>3 == fieldId {
238 log.Debugf("Method match with payload: %d, methodTable: %d", payload[idx], fieldId)
239 // TODO: Consider supporting other selector types.... Way, way in the future
240 // ok, the future is now, support strings as well... ugh.
241 var selector string
242 switch payload[idx] & 3 {
243 case 0: // Integer
244 b.SetBuf(payload[idx+1:])
245 v, e := b.DecodeVarint()
246 if e == nil {
247 log.Debugf("Decoded the ing field: %v", v)
248 selector = strconv.Itoa(int(v))
249 } else {
250 log.Errorf("Failed to decode varint %v", e)
251 return "", e
252 }
253 case 2: // Length delimited AKA string
254 b.SetBuf(payload[idx+1:])
255 v, e := b.DecodeStringBytes()
256 if e == nil {
257 log.Debugf("Decoded the string field: %v", v)
258 selector = v
259 } else {
260 log.Errorf("Failed to decode string %v", e)
261 return "", e
262 }
263 default:
Scott Baker4989fe92019-10-09 17:03:06 -0700264 err := fmt.Errorf("Only integer and string route selectors are permitted")
Scott Bakere7144bc2019-10-01 14:16:47 -0700265 log.Error(err)
266 return "", err
267 }
268 return selector, nil
269 } else if err := ar.skipField(&payload, &idx); err != nil {
270 log.Errorf("Parsing message failed %v", err)
271 return "", err
272 }
273 }
274}
275
276func (ar AffinityRouter) Route(sel interface{}) (*backend, *connection) {
277 switch sl := sel.(type) {
278 case *requestFrame:
279 log.Debugf("Route called for requestFrame with method %s", sl.methodInfo.method)
280 // Check if this method should be affinity bound from the
281 // reply rather than the request.
282 if _, ok := ar.nbBindingMethodMap[sl.methodInfo.method]; ok {
283 var err error
284 log.Debugf("Method '%s' affinity binds on reply", sl.methodInfo.method)
285 // Just round robin route the southbound request
286 if *ar.currentBackend, err = ar.cluster.nextBackend(*ar.currentBackend, BackendSequenceRoundRobin); err == nil {
287 return *ar.currentBackend, nil
288 } else {
289 sl.err = err
290 return nil, nil
291 }
292 }
293 // Not a south affinity binding method, proceed with north affinity binding.
294 if selector, err := ar.decodeProtoField(sl.payload, ar.methodMap[sl.methodInfo.method]); err == nil {
295 log.Debugf("Establishing affinity for selector: %s", selector)
296 if rtrn, ok := ar.affinity[selector]; ok {
297 return rtrn, nil
298 } else {
299 // The selector isn't in the map, create a new affinity mapping
300 log.Debugf("MUST CREATE A NEW AFFINITY MAP ENTRY!!")
301 var err error
302 if *ar.currentBackend, err = ar.cluster.nextBackend(*ar.currentBackend, BackendSequenceRoundRobin); err == nil {
Scott Baker4989fe92019-10-09 17:03:06 -0700303 err := ar.setAffinity(selector, *ar.currentBackend)
304 if err != nil {
305 log.Errorf("Failed to set affinity during Route: %v", err)
306 // TODO: Should we return nil here? We do have a backend, so we can return it, but we did fail
307 // to set affinity...
308 }
Scott Bakere7144bc2019-10-01 14:16:47 -0700309 return *ar.currentBackend, nil
310 } else {
311 sl.err = err
312 return nil, nil
313 }
314 }
315 }
316 default:
317 log.Errorf("Internal: invalid data type in Route call %v", sel)
318 return nil, nil
319 }
320 log.Errorf("Bad lookup in affinity map %v", ar.affinity)
321 return nil, nil
322}
323
324func (ar AffinityRouter) GetMetaKeyVal(serverStream grpc.ServerStream) (string, string, error) {
325 return "", "", nil
326}
327
328func (ar AffinityRouter) IsStreaming(_ string) (bool, bool) {
329 panic("not implemented")
330}
331
332func (ar AffinityRouter) BackendCluster(mthd string, metaKey string) (*cluster, error) {
333 return ar.cluster, nil
334}
335
336func (ar AffinityRouter) FindBackendCluster(beName string) *cluster {
337 if beName == ar.cluster.name {
338 return ar.cluster
339 }
340 return nil
341}
342
343func (ar AffinityRouter) ReplyHandler(sel interface{}) error {
344 switch sl := sel.(type) {
345 case *responseFrame:
346 log.Debugf("Reply handler called for responseFrame with method %s", sl.method)
347 // Determine if reply action is required.
348 if fld, ok := ar.nbBindingMethodMap[sl.method]; ok && len(sl.payload) > 0 {
349 // Extract the field value from the frame and
350 // and set affinity accordingly
351 if selector, err := ar.decodeProtoField(sl.payload, fld); err == nil {
352 log.Debug("Settign affinity on reply")
353 if ar.setAffinity(selector, sl.backend) != nil {
354 log.Error("Setting affinity on reply failed")
355 }
356 return nil
357 } else {
Scott Baker4989fe92019-10-09 17:03:06 -0700358 err := fmt.Errorf("Failed to decode reply field %d for method %s", fld, sl.method)
Scott Bakere7144bc2019-10-01 14:16:47 -0700359 log.Error(err)
360 return err
361 }
362 }
363 return nil
364 default:
Scott Baker4989fe92019-10-09 17:03:06 -0700365 err := fmt.Errorf("Internal: invalid data type in ReplyHander call %v", sl)
Scott Bakere7144bc2019-10-01 14:16:47 -0700366 log.Error(err)
367 return err
368 }
369}
370
371func (ar AffinityRouter) setAffinity(key string, be *backend) error {
372 if be2, ok := ar.affinity[key]; !ok {
373 ar.affinity[key] = be
374 log.Debugf("New affinity set to backend %s for key %s", be.name, key)
375 } else if be2 != be {
Scott Baker4989fe92019-10-09 17:03:06 -0700376 err := fmt.Errorf("Attempting multiple sets of affinity for key %s to backend %s from %s on router %s",
377 key, be.name, ar.affinity[key].name, ar.name)
Scott Bakere7144bc2019-10-01 14:16:47 -0700378 log.Error(err)
379 return err
380 }
381 return nil
382}