VOL-1848 API for setting and querying loglevel of api-server;
Add source-router to support routing UpdateLogLevel to cores;
Add logging endpoints to rocore
Change-Id: I89eea3599ea3006fe92e6917221cd1fd235ec5e4
diff --git a/afrouter/afrouter/source-router.go b/afrouter/afrouter/source-router.go
new file mode 100644
index 0000000..c9f94ef
--- /dev/null
+++ b/afrouter/afrouter/source-router.go
@@ -0,0 +1,320 @@
+/*
+ * Copyright 2018-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package afrouter
+
+/* Source-Router
+
+ This router implements source routing where the caller identifies the
+ component the message should be routed to. The `RouteField` should be
+ configured with the gRPC field name to inspect to determine the
+ destination. This field is assumed to be a string. That string will
+ then be used to identify a particular connection on a particular
+ backend.
+
+ The source-router must be configured with a backend cluster, as all routers
+ must identify a backend cluster. However, that backend cluster
+ is merely a placeholder and is not used by the source-router. The
+ source-router's Route() function will return whatever backend cluster is
+ specified by the `RouteField`.
+*/
+
+import (
+ "errors"
+ "fmt"
+ "github.com/golang/protobuf/proto"
+ pb "github.com/golang/protobuf/protoc-gen-go/descriptor"
+ "github.com/opencord/voltha-go/common/log"
+ "google.golang.org/grpc"
+ "io/ioutil"
+ "regexp"
+ "strconv"
+)
+
+type SourceRouter struct {
+ name string
+ //association associationType
+ routingField string
+ grpcService string
+ protoDescriptor *pb.FileDescriptorSet
+ methodMap map[string]byte
+ cluster *cluster
+}
+
+func newSourceRouter(rconf *RouterConfig, config *RouteConfig) (Router, error) {
+ var err error = nil
+ var rtrn_err = false
+ var pkg_re = regexp.MustCompile(`^(\.[^.]+\.)(.+)$`)
+ // Validate the configuration
+
+ // A name must exist
+ if config.Name == "" {
+ log.Error("A router 'name' must be specified")
+ rtrn_err = true
+ }
+
+ if rconf.ProtoPackage == "" {
+ log.Error("A 'package' must be specified")
+ rtrn_err = true
+ }
+
+ if rconf.ProtoService == "" {
+ log.Error("A 'service' must be specified")
+ rtrn_err = true
+ }
+
+ if config.RouteField == "" {
+ log.Error("A 'routing_field' must be specified")
+ rtrn_err = true
+ }
+
+ // TODO The overrieds section is currently not being used
+ // so the router will route all methods based on the
+ // routing_field. This needs to be added so that methods
+ // can have different routing fields.
+ dr := SourceRouter{
+ name: config.Name,
+ grpcService: rconf.ProtoService,
+ methodMap: make(map[string]byte),
+ }
+
+ // Load the protobuf descriptor file
+ dr.protoDescriptor = &pb.FileDescriptorSet{}
+ fb, err := ioutil.ReadFile(rconf.ProtoFile)
+ if err != nil {
+ log.Errorf("Could not open proto file '%s'", rconf.ProtoFile)
+ rtrn_err = true
+ }
+ err = proto.Unmarshal(fb, dr.protoDescriptor)
+ if err != nil {
+ log.Errorf("Could not unmarshal %s, %v", "proto.pb", err)
+ rtrn_err = true
+ }
+
+ // Build the routing structure based on the loaded protobuf
+ // descriptor file and the config information.
+ type key struct {
+ method string
+ field string
+ }
+ var msgs = make(map[key]byte)
+ for _, f := range dr.protoDescriptor.File {
+ // Build a temporary map of message types by name.
+ for _, m := range f.MessageType {
+ for _, fld := range m.Field {
+ log.Debugf("Processing message '%s', field '%s'", *m.Name, *fld.Name)
+ msgs[key{*m.Name, *fld.Name}] = byte(*fld.Number)
+ }
+ }
+ }
+ log.Debugf("The map contains: %v", msgs)
+ for _, f := range dr.protoDescriptor.File {
+ if *f.Package == rconf.ProtoPackage {
+ for _, s := range f.Service {
+ if *s.Name == rconf.ProtoService {
+ log.Debugf("Loading package data '%s' for service '%s' for router '%s'", *f.Package, *s.Name, dr.name)
+ // Now create a map keyed by method name with the value being the
+ // field number of the route selector.
+ var ok bool
+ for _, m := range s.Method {
+ // Find the input type in the messages and extract the
+ // field number and save it for future reference.
+ log.Debugf("Processing method '%s'", *m.Name)
+ // Determine if this is a method we're supposed to be processing.
+ if needMethod(*m.Name, config) {
+ log.Debugf("Enabling method '%s'", *m.Name)
+ pkg_methd := pkg_re.FindStringSubmatch(*m.InputType)
+ if pkg_methd == nil {
+ log.Errorf("Regular expression didn't match input type '%s'", *m.InputType)
+ rtrn_err = true
+ }
+ // The input type has the package name prepended to it. Remove it.
+ //in := (*m.InputType)[len(rconf.ProtoPackage)+2:]
+ in := pkg_methd[PKG_MTHD_MTHD]
+ dr.methodMap[*m.Name], ok = msgs[key{in, config.RouteField}]
+ if !ok {
+ log.Errorf("Method '%s' has no field named '%s' in it's parameter message '%s'",
+ *m.Name, config.RouteField, in)
+ rtrn_err = true
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ // We need to pick a cluster, because server will call cluster.handler. The choice we make doesn't
+ // matter, as we can return a different cluster from Route().
+ ok := true
+ if dr.cluster, ok = clusters[config.backendCluster.Name]; !ok {
+ if dr.cluster, err = newBackendCluster(config.backendCluster); err != nil {
+ log.Errorf("Could not create a backend for router %s", config.Name)
+ rtrn_err = true
+ }
+ }
+
+ if rtrn_err {
+ return dr, errors.New(fmt.Sprintf("Failed to create a new router '%s'", dr.name))
+ }
+
+ return dr, nil
+}
+
+func (ar SourceRouter) Service() string {
+ return ar.grpcService
+}
+
+func (ar SourceRouter) Name() string {
+ return ar.name
+}
+
+func (ar SourceRouter) skipField(data *[]byte, idx *int) error {
+ switch (*data)[*idx] & 3 {
+ case 0: // Varint
+ // skip the field number/type
+ *idx++
+ // if the msb is set, then more bytes to follow
+ for (*data)[*idx] >= 128 {
+ *idx++
+ }
+ // the last byte doesn't have the msb set
+ *idx++
+ case 1: // 64 bit
+ *idx += 9
+ case 2: // Length delimited
+ // skip the field number / type
+ *idx++
+ // read a varint that tells length of string
+ b := proto.NewBuffer((*data)[*idx:])
+ t, _ := b.DecodeVarint()
+ // skip the length varint and the string bytes
+ // TODO: This assumes the varint was one byte long -- max string length is 127 bytes
+ *idx += int(t) + 1
+ case 3: // Deprecated
+ case 4: // Deprecated
+ case 5: // 32 bit
+ *idx += 5
+ }
+ return nil
+}
+
+func (ar SourceRouter) decodeProtoField(payload []byte, fieldId byte) (string, error) {
+ idx := 0
+ b := proto.NewBuffer([]byte{})
+ //b.DebugPrint("The Buffer", payload)
+ for { // Find the route selector field
+ log.Debugf("Decoding source value attributeNumber: %d from %v at index %d", fieldId, payload, idx)
+ log.Debugf("Attempting match with payload: %d, methodTable: %d", payload[idx], fieldId)
+ if payload[idx]>>3 == fieldId {
+ log.Debugf("Method match with payload: %d, methodTable: %d", payload[idx], fieldId)
+ // TODO: Consider supporting other selector types.... Way, way in the future
+ // ok, the future is now, support strings as well... ugh.
+ var selector string
+ switch payload[idx] & 3 {
+ case 0: // Integer
+ b.SetBuf(payload[idx+1:])
+ v, e := b.DecodeVarint()
+ if e == nil {
+ log.Debugf("Decoded the ing field: %v", v)
+ selector = strconv.Itoa(int(v))
+ } else {
+ log.Errorf("Failed to decode varint %v", e)
+ return "", e
+ }
+ case 2: // Length delimited AKA string
+ b.SetBuf(payload[idx+1:])
+ v, e := b.DecodeStringBytes()
+ if e == nil {
+ log.Debugf("Decoded the string field: %v", v)
+ selector = v
+ } else {
+ log.Errorf("Failed to decode string %v", e)
+ return "", e
+ }
+ default:
+ err := errors.New(fmt.Sprintf("Only integer and string route selectors are permitted"))
+ log.Error(err)
+ return "", err
+ }
+ return selector, nil
+ } else if err := ar.skipField(&payload, &idx); err != nil {
+ log.Errorf("Parsing message failed %v", err)
+ return "", err
+ }
+ }
+}
+
+func (ar SourceRouter) Route(sel interface{}) (*backend, *connection) {
+ log.Debugf("SourceRouter sel %v", sel)
+ switch sl := sel.(type) {
+ case *requestFrame:
+ log.Debugf("Route called for nbFrame with method %s", sl.methodInfo.method)
+ // Not a south affinity binding method, proceed with north affinity binding.
+ if selector, err := ar.decodeProtoField(sl.payload, ar.methodMap[sl.methodInfo.method]); err == nil {
+ // selector is
+
+ for _, cluster := range clusters {
+ for _, backend := range cluster.backends {
+ log.Debugf("Checking backend %s", backend.name)
+ for _, connection := range backend.connections {
+ log.Debugf("Checking connection %s", connection.name)
+ // caller specified a backend and a connection
+ if backend.name+"."+connection.name == selector {
+ return backend, connection
+ }
+ }
+ // caller specified just a backend
+ if backend.name == selector {
+ return backend, nil
+ }
+ }
+ }
+ sl.err = fmt.Errorf("Backend %s not found", selector)
+ return nil, nil
+ }
+ default:
+ log.Errorf("Internal: invalid data type in Route call %v", sel)
+ return nil, nil
+ }
+ log.Errorf("Bad routing in SourceRouter:Route")
+ return nil, nil
+}
+
+func (ar SourceRouter) GetMetaKeyVal(serverStream grpc.ServerStream) (string, string, error) {
+ return "", "", nil
+}
+
+func (ar SourceRouter) IsStreaming(_ string) (bool, bool) {
+ panic("not implemented")
+}
+
+func (ar SourceRouter) BackendCluster(mthd string, metaKey string) (*cluster, error) {
+ // unsupported?
+ return ar.cluster, nil
+}
+
+func (ar SourceRouter) FindBackendCluster(beName string) *cluster {
+ // unsupported?
+ if beName == ar.cluster.name {
+ return ar.cluster
+ }
+ return nil
+}
+
+func (rr SourceRouter) ReplyHandler(sel interface{}) error { // This is a no-op
+ return nil
+}