blob: 61754b3248c608565a2a5ed201e88e70497a2b01 [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
19/* Source-Router
20
21 This router implements source routing where the caller identifies the
22 component the message should be routed to. The `RouteField` should be
23 configured with the gRPC field name to inspect to determine the
24 destination. This field is assumed to be a string. That string will
25 then be used to identify a particular connection on a particular
26 backend.
27
28 The source-router must be configured with a backend cluster, as all routers
29 must identify a backend cluster. However, that backend cluster
30 is merely a placeholder and is not used by the source-router. The
31 source-router's Route() function will return whatever backend cluster is
32 specified by the `RouteField`.
33*/
34
35import (
36 "errors"
37 "fmt"
38 "github.com/golang/protobuf/proto"
39 "github.com/opencord/voltha-go/common/log"
40 "google.golang.org/grpc"
41 "regexp"
42 "strconv"
43)
44
45type SourceRouter struct {
Scott Baker4989fe92019-10-09 17:03:06 -070046 name string
47 grpcService string
48 methodMap map[string]byte
49 cluster *cluster
Scott Bakere7144bc2019-10-01 14:16:47 -070050}
51
52func newSourceRouter(rconf *RouterConfig, config *RouteConfig) (Router, error) {
Scott Baker4989fe92019-10-09 17:03:06 -070053 var err error
Scott Bakere7144bc2019-10-01 14:16:47 -070054 var rtrn_err = false
55 var pkg_re = regexp.MustCompile(`^(\.[^.]+\.)(.+)$`)
56 // Validate the configuration
57
58 // A name must exist
59 if config.Name == "" {
60 log.Error("A router 'name' must be specified")
61 rtrn_err = true
62 }
63
64 if rconf.ProtoPackage == "" {
65 log.Error("A 'package' must be specified")
66 rtrn_err = true
67 }
68
69 if rconf.ProtoService == "" {
70 log.Error("A 'service' must be specified")
71 rtrn_err = true
72 }
73
74 if config.RouteField == "" {
75 log.Error("A 'routing_field' must be specified")
76 rtrn_err = true
77 }
78
79 // TODO The overrieds section is currently not being used
80 // so the router will route all methods based on the
81 // routing_field. This needs to be added so that methods
82 // can have different routing fields.
83 dr := SourceRouter{
84 name: config.Name,
85 grpcService: rconf.ProtoService,
86 methodMap: make(map[string]byte),
87 }
88
89 // Build the routing structure based on the loaded protobuf
90 // descriptor file and the config information.
91 type key struct {
92 method string
93 field string
94 }
95 var fieldNumberLookup = make(map[key]byte)
96 for _, f := range rconf.protoDescriptor.File {
97 // Build a temporary map of message types by name.
98 for _, m := range f.MessageType {
99 for _, fld := range m.Field {
100 log.Debugf("Processing message '%s', field '%s'", *m.Name, *fld.Name)
101 fieldNumberLookup[key{*m.Name, *fld.Name}] = byte(*fld.Number)
102 }
103 }
104 }
105 for _, f := range rconf.protoDescriptor.File {
106 if *f.Package == rconf.ProtoPackage {
107 for _, s := range f.Service {
108 if *s.Name == rconf.ProtoService {
109 log.Debugf("Loading package data '%s' for service '%s' for router '%s'", *f.Package, *s.Name, dr.name)
110 // Now create a map keyed by method name with the value being the
111 // field number of the route selector.
112 var ok bool
113 for _, m := range s.Method {
114 // Find the input type in the messages and extract the
115 // field number and save it for future reference.
116 log.Debugf("Processing method '%s'", *m.Name)
117 // Determine if this is a method we're supposed to be processing.
118 if needMethod(*m.Name, config) {
119 log.Debugf("Enabling method '%s'", *m.Name)
120 pkg_methd := pkg_re.FindStringSubmatch(*m.InputType)
121 if pkg_methd == nil {
122 log.Errorf("Regular expression didn't match input type '%s'", *m.InputType)
123 rtrn_err = true
124 }
125 // The input type has the package name prepended to it. Remove it.
126 //in := (*m.InputType)[len(rconf.ProtoPackage)+2:]
127 in := pkg_methd[PKG_MTHD_MTHD]
128 dr.methodMap[*m.Name], ok = fieldNumberLookup[key{in, config.RouteField}]
129 if !ok {
130 log.Errorf("Method '%s' has no field named '%s' in it's parameter message '%s'",
131 *m.Name, config.RouteField, in)
132 rtrn_err = true
133 }
134 }
135 }
136 }
137 }
138 }
139 }
140
141 // We need to pick a cluster, because server will call cluster.handler. The choice we make doesn't
142 // matter, as we can return a different cluster from Route().
Scott Baker4989fe92019-10-09 17:03:06 -0700143 var ok bool
Scott Bakere7144bc2019-10-01 14:16:47 -0700144 if dr.cluster, ok = clusters[config.backendCluster.Name]; !ok {
145 if dr.cluster, err = newBackendCluster(config.backendCluster); err != nil {
146 log.Errorf("Could not create a backend for router %s", config.Name)
147 rtrn_err = true
148 }
149 }
150
151 if rtrn_err {
Scott Baker4989fe92019-10-09 17:03:06 -0700152 return dr, fmt.Errorf("Failed to create a new router '%s'", dr.name)
Scott Bakere7144bc2019-10-01 14:16:47 -0700153 }
154
155 return dr, nil
156}
157
158func (ar SourceRouter) Service() string {
159 return ar.grpcService
160}
161
162func (ar SourceRouter) Name() string {
163 return ar.name
164}
165
166func (ar SourceRouter) skipField(data *[]byte, idx *int) error {
167 switch (*data)[*idx] & 3 {
168 case 0: // Varint
169 // skip the field number/type
170 *idx++
171 // if the msb is set, then more bytes to follow
172 for (*data)[*idx] >= 128 {
173 *idx++
174 }
175 // the last byte doesn't have the msb set
176 *idx++
177 case 1: // 64 bit
178 *idx += 9
179 case 2: // Length delimited
180 // skip the field number / type
181 *idx++
182 // read a varint that tells length of string
183 b := proto.NewBuffer((*data)[*idx:])
184 t, _ := b.DecodeVarint()
185 // skip the length varint and the string bytes
186 // TODO: This assumes the varint was one byte long -- max string length is 127 bytes
187 *idx += int(t) + 1
188 case 3: // Deprecated
189 case 4: // Deprecated
190 case 5: // 32 bit
191 *idx += 5
192 }
193 return nil
194}
195
196func (ar SourceRouter) decodeProtoField(payload []byte, fieldId byte) (string, error) {
197 idx := 0
198 b := proto.NewBuffer([]byte{})
199 //b.DebugPrint("The Buffer", payload)
200 for { // Find the route selector field
201 log.Debugf("Decoding source value attributeNumber: %d from %v at index %d", fieldId, payload, idx)
202 log.Debugf("Attempting match with payload: %d, methodTable: %d", payload[idx], fieldId)
203 if payload[idx]>>3 == fieldId {
204 log.Debugf("Method match with payload: %d, methodTable: %d", payload[idx], fieldId)
205 // TODO: Consider supporting other selector types.... Way, way in the future
206 // ok, the future is now, support strings as well... ugh.
207 var selector string
208 switch payload[idx] & 3 {
209 case 0: // Integer
210 b.SetBuf(payload[idx+1:])
211 v, e := b.DecodeVarint()
212 if e == nil {
213 log.Debugf("Decoded the ing field: %v", v)
214 selector = strconv.Itoa(int(v))
215 } else {
216 log.Errorf("Failed to decode varint %v", e)
217 return "", e
218 }
219 case 2: // Length delimited AKA string
220 b.SetBuf(payload[idx+1:])
221 v, e := b.DecodeStringBytes()
222 if e == nil {
223 log.Debugf("Decoded the string field: %v", v)
224 selector = v
225 } else {
226 log.Errorf("Failed to decode string %v", e)
227 return "", e
228 }
229 default:
Scott Baker4989fe92019-10-09 17:03:06 -0700230 err := errors.New("Only integer and string route selectors are permitted")
Scott Bakere7144bc2019-10-01 14:16:47 -0700231 log.Error(err)
232 return "", err
233 }
234 return selector, nil
235 } else if err := ar.skipField(&payload, &idx); err != nil {
236 log.Errorf("Parsing message failed %v", err)
237 return "", err
238 }
239 }
240}
241
242func (ar SourceRouter) Route(sel interface{}) (*backend, *connection) {
243 log.Debugf("SourceRouter sel %v", sel)
244 switch sl := sel.(type) {
245 case *requestFrame:
246 log.Debugf("Route called for nbFrame with method %s", sl.methodInfo.method)
247 // Not a south affinity binding method, proceed with north affinity binding.
248 if selector, err := ar.decodeProtoField(sl.payload, ar.methodMap[sl.methodInfo.method]); err == nil {
249 // selector is
250
251 for _, cluster := range clusters {
252 for _, backend := range cluster.backends {
253 log.Debugf("Checking backend %s", backend.name)
254 for _, connection := range backend.connections {
255 log.Debugf("Checking connection %s", connection.name)
256 // caller specified a backend and a connection
257 if backend.name+"."+connection.name == selector {
258 return backend, connection
259 }
260 }
261 // caller specified just a backend
262 if backend.name == selector {
263 return backend, nil
264 }
265 }
266 }
267 sl.err = fmt.Errorf("Backend %s not found", selector)
268 return nil, nil
269 }
270 default:
271 log.Errorf("Internal: invalid data type in Route call %v", sel)
272 return nil, nil
273 }
274 log.Errorf("Bad routing in SourceRouter:Route")
275 return nil, nil
276}
277
278func (ar SourceRouter) GetMetaKeyVal(serverStream grpc.ServerStream) (string, string, error) {
279 return "", "", nil
280}
281
282func (ar SourceRouter) IsStreaming(_ string) (bool, bool) {
283 panic("not implemented")
284}
285
286func (ar SourceRouter) BackendCluster(mthd string, metaKey string) (*cluster, error) {
287 // unsupported?
288 return ar.cluster, nil
289}
290
291func (ar SourceRouter) FindBackendCluster(beName string) *cluster {
292 // unsupported?
293 if beName == ar.cluster.name {
294 return ar.cluster
295 }
296 return nil
297}
298
299func (rr SourceRouter) ReplyHandler(sel interface{}) error { // This is a no-op
300 return nil
301}