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/affinity-router.go b/afrouter/afrouter/affinity-router.go
index 583d0a7..7a3168e 100644
--- a/afrouter/afrouter/affinity-router.go
+++ b/afrouter/afrouter/affinity-router.go
@@ -26,6 +26,7 @@
"strconv"
)
+// TODO: Used in multiple routers, should move to common file
const (
PKG_MTHD_PKG int = 1
PKG_MTHD_MTHD int = 2
@@ -180,6 +181,7 @@
return false
}
+// TODO: Used in multiple routers, should move to common file
func needMethod(mthd string, conf *RouteConfig) bool {
for _, m := range conf.Methods {
if mthd == m {
@@ -265,7 +267,7 @@
}
}
-func (ar AffinityRouter) Route(sel interface{}) *backend {
+func (ar AffinityRouter) Route(sel interface{}) (*backend, *connection) {
switch sl := sel.(type) {
case *requestFrame:
log.Debugf("Route called for requestFrame with method %s", sl.methodInfo.method)
@@ -276,17 +278,17 @@
log.Debugf("Method '%s' affinity binds on reply", sl.methodInfo.method)
// Just round robin route the southbound request
if *ar.currentBackend, err = ar.cluster.nextBackend(*ar.currentBackend, BackendSequenceRoundRobin); err == nil {
- return *ar.currentBackend
+ return *ar.currentBackend, nil
} else {
sl.err = err
- return nil
+ return nil, nil
}
}
// 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 {
log.Debugf("Establishing affinity for selector: %s", selector)
if rtrn, ok := ar.affinity[selector]; ok {
- return rtrn
+ return rtrn, nil
} else {
// The selector isn't in the map, create a new affinity mapping
log.Debugf("MUST CREATE A NEW AFFINITY MAP ENTRY!!")
@@ -295,19 +297,19 @@
ar.setAffinity(selector, *ar.currentBackend)
//ar.affinity[selector] = *ar.currentBackend
//log.Debugf("New affinity set to backend %s",(*ar.currentBackend).name)
- return *ar.currentBackend
+ return *ar.currentBackend, nil
} else {
sl.err = err
- return nil
+ return nil, nil
}
}
}
default:
log.Errorf("Internal: invalid data type in Route call %v", sel)
- return nil
+ return nil, nil
}
log.Errorf("Bad lookup in affinity map %v", ar.affinity)
- return nil
+ return nil, nil
}
func (ar AffinityRouter) GetMetaKeyVal(serverStream grpc.ServerStream) (string, string, error) {
diff --git a/afrouter/afrouter/api.go b/afrouter/afrouter/api.go
index 3e47ef8..8b72897 100644
--- a/afrouter/afrouter/api.go
+++ b/afrouter/afrouter/api.go
@@ -21,6 +21,7 @@
"fmt"
"github.com/opencord/voltha-go/common/log"
pb "github.com/opencord/voltha-protos/go/afrouter"
+ common_pb "github.com/opencord/voltha-protos/go/common"
"golang.org/x/net/context"
"google.golang.org/grpc"
"net"
@@ -210,6 +211,47 @@
return &pb.Count{Count: uint32(runtime.NumGoroutine())}, nil
}
+func (aa ArouterApi) UpdateLogLevel(ctx context.Context, in *common_pb.Logging) (*pb.Empty, error) {
+ intLevel := int(in.Level)
+
+ if in.PackageName == "" {
+ log.SetAllLogLevel(intLevel)
+ log.SetDefaultLogLevel(intLevel)
+ } else if in.PackageName == "default" {
+ log.SetDefaultLogLevel(intLevel)
+ } else {
+ log.SetPackageLogLevel(in.PackageName, intLevel)
+ }
+
+ return &pb.Empty{}, nil
+}
+
+func (aa ArouterApi) GetLogLevels(ctx context.Context, in *common_pb.LoggingComponent) (*common_pb.Loggings, error) {
+ logLevels := &common_pb.Loggings{}
+
+ // do the per-package log levels
+ for _, packageName := range log.GetPackageNames() {
+ level, err := log.GetPackageLogLevel(packageName)
+ if err != nil {
+ return nil, err
+ }
+ logLevel := &common_pb.Logging{
+ ComponentName: in.ComponentName,
+ PackageName: packageName,
+ Level: common_pb.LogLevel_LogLevel(level)}
+ logLevels.Items = append(logLevels.Items, logLevel)
+ }
+
+ // now do the default log level
+ logLevel := &common_pb.Logging{
+ ComponentName: in.ComponentName,
+ PackageName: "default",
+ Level: common_pb.LogLevel_LogLevel(log.GetDefaultLogLevel())}
+ logLevels.Items = append(logLevels.Items, logLevel)
+
+ return logLevels, nil
+}
+
func (aa *ArouterApi) serve() {
// Start a serving thread
go func() {
diff --git a/afrouter/afrouter/backend.go b/afrouter/afrouter/backend.go
index 3c7815d..3db98c5 100644
--- a/afrouter/afrouter/backend.go
+++ b/afrouter/afrouter/backend.go
@@ -111,8 +111,18 @@
// connections are non-existent.
var atLeastOne = false
var errStr strings.Builder
+
log.Debugf("There are %d/%d streams to open", len(be.openConns), len(be.connections))
+ if nf.connection != nil {
+ // Debug statement triggered by source router. Other routers have no connection preference.
+ log.Debugf("Looking for connection %s", nf.connection.name)
+ }
for cn, conn := range be.openConns {
+ // If source-router was used, it will indicate a specific connection to be used
+ if nf.connection != nil && nf.connection != cn {
+ continue
+ }
+
log.Debugf("Opening stream for connection '%s'", cn.name)
if stream, err := grpc.NewClientStream(r.ctx, clientStreamDescForProxying, conn, r.methodInfo.all); err != nil {
log.Debugf("Failed to create a client stream '%s', %v", cn.name, err)
diff --git a/afrouter/afrouter/binding-router.go b/afrouter/afrouter/binding-router.go
index dd73756..2e72571 100644
--- a/afrouter/afrouter/binding-router.go
+++ b/afrouter/afrouter/binding-router.go
@@ -80,37 +80,37 @@
func (br BindingRouter) ReplyHandler(v interface{}) error {
return nil
}
-func (br BindingRouter) Route(sel interface{}) *backend {
+func (br BindingRouter) Route(sel interface{}) (*backend, *connection) {
var err error
switch sl := sel.(type) {
case *requestFrame:
if b, ok := br.bindings[sl.metaVal]; ok == true { // binding exists, just return it
- return b
+ return b, nil
} else { // establish a new binding or error.
if sl.metaVal != "" {
err = errors.New(fmt.Sprintf("Attempt to route on non-existent metadata value '%s' in key '%s'",
sl.metaVal, sl.metaKey))
log.Error(err)
sl.err = err
- return nil
+ return nil, nil
}
if sl.methodInfo.method != br.bindingMethod {
err = errors.New(fmt.Sprintf("Binding must occur with method %s but attempted with method %s",
br.bindingMethod, sl.methodInfo.method))
log.Error(err)
sl.err = err
- return nil
+ return nil, nil
}
log.Debugf("MUST CREATE A NEW BINDING MAP ENTRY!!")
if len(br.bindings) < len(br.beCluster.backends) {
if *br.currentBackend, err = br.beCluster.nextBackend(*br.currentBackend, BackendSequenceRoundRobin); err == nil {
// Use the name of the backend as the metaVal for this new binding
br.bindings[(*br.currentBackend).name] = *br.currentBackend
- return *br.currentBackend
+ return *br.currentBackend, nil
} else {
log.Error(err)
sl.err = err
- return nil
+ return nil, nil
}
} else {
err = errors.New(fmt.Sprintf("Backends exhausted in attempt to bind for metakey '%s' with value '%s'",
@@ -119,9 +119,9 @@
sl.err = err
}
}
- return nil
+ return nil, nil
default:
- return nil
+ return nil, nil
}
}
diff --git a/afrouter/afrouter/cluster.go b/afrouter/afrouter/cluster.go
index e33db67..7fc581b 100644
--- a/afrouter/afrouter/cluster.go
+++ b/afrouter/afrouter/cluster.go
@@ -157,6 +157,7 @@
if err := src.RecvMsg(f); err != nil {
return nil, err
}
+ log.Debugf("Assigned backend %s", f.backend.name)
// Check that the backend was routable and actually has connections open.
// If it doesn't then return a nil backend to indicate this
if f.backend == nil {
diff --git a/afrouter/afrouter/codec.go b/afrouter/afrouter/codec.go
index 7ea6ef2..27f4619 100644
--- a/afrouter/afrouter/codec.go
+++ b/afrouter/afrouter/codec.go
@@ -50,6 +50,7 @@
payload []byte
router Router
backend *backend
+ connection *connection // optional, if the router preferred one connection over another
err error
methodInfo methodDetails
serialNo uint64
@@ -79,12 +80,13 @@
t.payload = data
// This is were the afinity value is pulled from the payload
// and the backend selected.
- t.backend = t.router.Route(v)
+ t.backend, t.connection = t.router.Route(v)
name := "<nil>"
if t.backend != nil {
name = t.backend.name
}
log.Debugf("Routing returned %s for method %s", name, t.methodInfo.method)
+
return nil
default:
return cdc.parentCodec.Unmarshal(data, v)
diff --git a/afrouter/afrouter/enums.go b/afrouter/afrouter/enums.go
index 02b1d3d..1847191 100644
--- a/afrouter/afrouter/enums.go
+++ b/afrouter/afrouter/enums.go
@@ -133,11 +133,12 @@
RouteTypeRpcAffinityHeader
RouteTypeBinding
RouteTypeRoundRobin
+ RouteTypeSource
)
// String names for display in error messages.
-var stringToRouteType = map[string]routeType{"rpc_affinity_message": RouteTypeRpcAffinityMessage, "rpc_affinity_header": RouteTypeRpcAffinityHeader, "binding": RouteTypeBinding, "round_robin": RouteTypeRoundRobin}
-var routeTypeToString = map[routeType]string{RouteTypeRpcAffinityMessage: "rpc_affinity_message", RouteTypeRpcAffinityHeader: "rpc_affinity_header", RouteTypeBinding: "binding", RouteTypeRoundRobin: "round_robin"}
+var stringToRouteType = map[string]routeType{"rpc_affinity_message": RouteTypeRpcAffinityMessage, "rpc_affinity_header": RouteTypeRpcAffinityHeader, "binding": RouteTypeBinding, "round_robin": RouteTypeRoundRobin, "source": RouteTypeSource}
+var routeTypeToString = map[routeType]string{RouteTypeRpcAffinityMessage: "rpc_affinity_message", RouteTypeRpcAffinityHeader: "rpc_affinity_header", RouteTypeBinding: "binding", RouteTypeRoundRobin: "round_robin", RouteTypeSource: "source"}
func (t routeType) String() string {
if str, have := routeTypeToString[t]; have {
diff --git a/afrouter/afrouter/method-router.go b/afrouter/afrouter/method-router.go
index fd3b974..56d025e 100644
--- a/afrouter/afrouter/method-router.go
+++ b/afrouter/afrouter/method-router.go
@@ -176,16 +176,16 @@
return nil
}
-func (mr MethodRouter) Route(sel interface{}) *backend {
+func (mr MethodRouter) Route(sel interface{}) (*backend, *connection) {
switch sl := sel.(type) {
case *requestFrame:
if r, ok := mr.methodRouter[sl.metaKey][sl.methodInfo.method]; ok {
return r.Route(sel)
}
log.Errorf("Attept to route on non-existent method '%s'", sl.methodInfo.method)
- return nil
+ return nil, nil
default:
- return nil
+ return nil, nil
}
}
diff --git a/afrouter/afrouter/round-robin-router.go b/afrouter/afrouter/round-robin-router.go
index 60ff457..b5f031e 100644
--- a/afrouter/afrouter/round-robin-router.go
+++ b/afrouter/afrouter/round-robin-router.go
@@ -92,20 +92,20 @@
return rr.name
}
-func (rr RoundRobinRouter) Route(sel interface{}) *backend {
+func (rr RoundRobinRouter) Route(sel interface{}) (*backend, *connection) {
var err error
switch sl := sel.(type) {
case *requestFrame:
// Since this is a round robin router just get the next backend
if *rr.currentBackend, err = rr.cluster.nextBackend(*rr.currentBackend, BackendSequenceRoundRobin); err == nil {
- return *rr.currentBackend
+ return *rr.currentBackend, nil
} else {
sl.err = err
- return nil
+ return nil, nil
}
default:
log.Errorf("Internal: invalid data type in Route call %v", sel)
- return nil
+ return nil, nil
}
}
diff --git a/afrouter/afrouter/router.go b/afrouter/afrouter/router.go
index 1bd2d6e..7abbceb 100644
--- a/afrouter/afrouter/router.go
+++ b/afrouter/afrouter/router.go
@@ -27,7 +27,11 @@
// The router interface
type Router interface {
Name() string
- Route(interface{}) *backend
+
+ // Route() returns a backend and a connection. The connection is optional and if unspecified, then any
+ // connection on the backend may be used.
+ Route(interface{}) (*backend, *connection)
+
Service() string
IsStreaming(string) (bool, bool)
BackendCluster(string, string) (*cluster, error)
@@ -64,6 +68,12 @@
allRouters[rconf.Name+config.Name] = r
}
return r, err
+ case RouteTypeSource:
+ r, err := newSourceRouter(rconf, config)
+ if err == nil {
+ allRouters[rconf.Name+config.Name] = r
+ }
+ return r, err
default:
return nil, errors.New(fmt.Sprintf("Internal error, undefined router type: %s", config.Type))
}
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
+}
diff --git a/afrouter/afrouter/source-router_test.go b/afrouter/afrouter/source-router_test.go
new file mode 100644
index 0000000..3d35d6e
--- /dev/null
+++ b/afrouter/afrouter/source-router_test.go
@@ -0,0 +1,140 @@
+/*
+ * Portions copyright 2019-present Open Networking Foundation
+ * Original copyright 2019-present Ciena Corporation
+ *
+ * Licensed under the Apache License, Version 2.0 (the"github.com/stretchr/testify/assert" "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
+
+import (
+ // "github.com/fullstorydev/grpcurl"
+ "github.com/golang/protobuf/proto"
+ // descpb "github.com/golang/protobuf/protoc-gen-go/descriptor"
+ // "github.com/jhump/protoreflect/dynamic"
+ "github.com/opencord/voltha-go/common/log"
+ common_pb "github.com/opencord/voltha-protos/go/common"
+ "github.com/stretchr/testify/assert"
+ //"io/ioutil"
+ "testing"
+)
+
+const (
+ PROTOFILE = "../../vendor/github.com/opencord/voltha-protos/go/voltha.pb"
+)
+
+func init() {
+ log.SetDefaultLogger(log.JSON, log.DebugLevel, nil)
+ log.AddPackage(log.JSON, log.WarnLevel, nil)
+}
+
+func MakeTestConfig() (*ConnectionConfig, *BackendConfig, *BackendClusterConfig, *RouteConfig, *RouterConfig) {
+ connectionConfig := ConnectionConfig{
+ Name: "ro_vcore01",
+ Addr: "foo",
+ Port: "123",
+ }
+
+ backendConfig := BackendConfig{
+ Name: "ro_vcore0",
+ Type: BackendSingleServer,
+ Connections: []ConnectionConfig{connectionConfig},
+ }
+
+ backendClusterConfig := BackendClusterConfig{
+ Name: "ro_vcore",
+ Backends: []BackendConfig{backendConfig},
+ }
+
+ routeConfig := RouteConfig{
+ Name: "logger",
+ Type: RouteTypeSource,
+ RouteField: "component_name",
+ BackendCluster: "ro_vcore",
+ backendCluster: &backendClusterConfig,
+ Methods: []string{"UpdateLogLevel", "GetLogLevel"},
+ }
+
+ routerConfig := RouterConfig{
+ Name: "vcore",
+ ProtoService: "VolthaService",
+ ProtoPackage: "voltha",
+ Routes: []RouteConfig{routeConfig},
+ ProtoFile: PROTOFILE,
+ }
+ return &connectionConfig, &backendConfig, &backendClusterConfig, &routeConfig, &routerConfig
+}
+
+func TestSourceRouterInit(t *testing.T) {
+ _, _, _, routeConfig, routerConfig := MakeTestConfig()
+
+ router, err := newSourceRouter(routerConfig, routeConfig)
+
+ assert.NotEqual(t, router, nil)
+ assert.Equal(t, err, nil)
+
+ assert.Equal(t, router.Service(), "VolthaService")
+ assert.Equal(t, router.Name(), "logger")
+
+ cluster, err := router.BackendCluster("foo", "bar")
+ assert.Equal(t, cluster, clusters["ro_vcore"])
+ assert.Equal(t, err, nil)
+
+ assert.Equal(t, router.FindBackendCluster("ro_vcore"), clusters["ro_vcore"])
+ assert.Equal(t, router.ReplyHandler("foo"), nil)
+}
+
+func TestDecodeProtoField(t *testing.T) {
+ _, _, _, routeConfig, routerConfig := MakeTestConfig()
+
+ router, err := newSourceRouter(routerConfig, routeConfig)
+ assert.Equal(t, err, nil)
+
+ loggingMessage := &common_pb.Logging{Level: 1,
+ PackageName: "default",
+ ComponentName: "ro_vcore0.ro_vcore01"}
+
+ loggingData, err := proto.Marshal(loggingMessage)
+ assert.Equal(t, err, nil)
+
+ s, err := router.(SourceRouter).decodeProtoField(loggingData, 2) // field 2 is package_name
+ assert.Equal(t, s, "default")
+
+ s, err = router.(SourceRouter).decodeProtoField(loggingData, 3) // field 2 is component_name
+ assert.Equal(t, s, "ro_vcore0.ro_vcore01")
+}
+
+func TestRoute(t *testing.T) {
+ _, _, _, routeConfig, routerConfig := MakeTestConfig()
+
+ router, err := newSourceRouter(routerConfig, routeConfig)
+ assert.Equal(t, err, nil)
+
+ loggingMessage := &common_pb.Logging{Level: 1,
+ PackageName: "default",
+ ComponentName: "ro_vcore0.ro_vcore01"}
+
+ loggingData, err := proto.Marshal(loggingMessage)
+ assert.Equal(t, err, nil)
+
+ sel := &requestFrame{payload: loggingData,
+ err: nil,
+ methodInfo: newMethodDetails("/volta.VolthaService/UpdateLogLevel")}
+
+ backend, connection := router.Route(sel)
+
+ assert.Equal(t, sel.err, nil)
+ assert.NotEqual(t, backend, nil)
+ assert.Equal(t, backend.name, "ro_vcore0")
+ assert.NotEqual(t, connection, nil)
+ assert.Equal(t, connection.name, "ro_vcore01")
+}