Skip to content
Snippets Groups Projects

Develop

Closed Develop
develop into master
Closed Ghost User requested to merge develop into master
Viewing commit c483c139
Show latest version
1 file
+ 8
− 0
Compare changes
  • Side-by-side
  • Inline
nucleus/gnmi_transport.go
+ 131
− 161
package nucleus
import (
"code.fbi.h-da.de/cocsn/gosdn/forks/goarista/gnmi"
"context"
"fmt"
"code.fbi.h-da.de/danet/gosdn/interfaces/change"
"code.fbi.h-da.de/danet/gosdn/interfaces/southbound"
ppb "code.fbi.h-da.de/danet/api/go/gosdn/pnd"
"code.fbi.h-da.de/danet/forks/goarista/gnmi"
"code.fbi.h-da.de/danet/gosdn/nucleus/errors"
"code.fbi.h-da.de/danet/gosdn/nucleus/types"
gpb "github.com/openconfig/gnmi/proto/gnmi"
"github.com/openconfig/gnmi/proto/gnmi_ext"
"github.com/openconfig/goyang/pkg/yang"
"github.com/openconfig/ygot/ygot"
"github.com/openconfig/ygot/ytypes"
log "github.com/sirupsen/logrus"
"reflect"
"strings"
)
// CtxKeyType is a custom type to be used as key in a context.WithValue() or
// context.Value() call. For more information see:
// https://www.calhoun.io/pitfalls-of-context-values-and-how-to-avoid-or-mitigate-them/
type CtxKeyType string
const (
// CtxKeyOpts context key for gnmi.SubscribeOptions
CtxKeyOpts CtxKeyType = "opts"
// CtxKeyConfig is a context key for gnmi.Config
CtxKeyConfig = "config"
tpb "code.fbi.h-da.de/danet/api/go/gosdn/transport"
)
// Gnmi implements the Transport interface and provides an SBI with the
@@ -29,97 +27,100 @@ const (
type Gnmi struct {
SetNode func(schema *yang.Entry, root interface{}, path *gpb.Path, val interface{}, opts ...ytypes.SetNodeOpt) error
RespChan chan *gpb.SubscribeResponse
Unmarshal func([]byte, []string, interface{}, ...ytypes.UnmarshalOpt) error
Options *GnmiTransportOptions
Unmarshal func([]byte, *gpb.Path, ygot.ValidatedGoStruct, ...ytypes.UnmarshalOpt) error
Options *tpb.TransportOption
client gpb.GNMIClient
config *gnmi.Config
}
// NewGnmiTransport takes a struct of GnmiTransportOptions and returns a Gnmi
// newGnmiTransport takes a struct of GnmiTransportOptions and returns a Gnmi
// transport based on the values of it.
func NewGnmiTransport(opts *GnmiTransportOptions) (*Gnmi, error) {
c, err := gnmi.Dial(&opts.Config)
// Do not call directly. Use NewTransport() instead.
func newGnmiTransport(opts *tpb.TransportOption, sbi southbound.SouthboundInterface) (*Gnmi, error) {
if opts == nil || sbi == nil {
return nil, &errors.ErrInvalidParameters{
Func: newGnmiTransport,
Param: "'opts' and 'sbi' can not be nil",
}
} else if opts.TransportOption == nil {
return nil, &errors.ErrInvalidParameters{
Func: newGnmiTransport,
Param: "'opts.TransportOption' can not be nil",
}
}
gnmiConfig := &gnmi.Config{
Addr: opts.Address,
Password: opts.Password,
Username: opts.Username,
TLS: opts.Tls,
Compression: opts.GetGnmiTransportOption().GetCompression(),
}
c, err := gnmi.Dial(gnmiConfig)
if err != nil {
return nil, err
}
log.WithFields(log.Fields{
"target": opts.Addr,
"tls": opts.TLS,
"encoding": opts.Encoding,
"target": opts.Address,
"tls": opts.Tls,
}).Info("building new gNMI transport")
return &Gnmi{
SetNode: opts.SetNode,
RespChan: opts.RespChan,
Options: opts,
client: c,
SetNode: sbi.SetNode,
RespChan: make(chan *gpb.SubscribeResponse),
Unmarshal: sbi.Unmarshal,
Options: opts,
client: c,
config: gnmiConfig,
}, nil
}
//SetOptions sets Gnmi Options
func (g *Gnmi) SetOptions(to TransportOptions) {
g.Options = to.(*GnmiTransportOptions)
}
//GetOptions returns the Gnmi options
func (g *Gnmi) GetOptions() interface{} {
return g.Options
}
// Get takes a slice of gnmi paths, splits them and calls get for each one of them.
func (g *Gnmi) Get(ctx context.Context, params ...string) (interface{}, error) {
if g.client == nil {
return nil, &ErrNilClient{}
return nil, &errors.ErrNilClient{}
}
ctx = gnmi.NewContext(ctx, g.config)
paths := gnmi.SplitPaths(params)
return g.get(ctx, paths, "")
}
// Set takes a slice of params. This slice must contain at least one operation.
// It can contain an additional arbitrary amount of operations and extensions.
func (g *Gnmi) Set(ctx context.Context, args ...interface{}) (interface{}, error) {
// Set takes a change.Payload struct.
func (g *Gnmi) Set(ctx context.Context, payload change.Payload) error {
if g.client == nil {
return nil, &ErrNilClient{}
}
if len(args) == 0 {
return nil, &ErrInvalidParameters{
f: "gnmi.Set()",
r: "no parameters provided",
}
return &errors.ErrNilClient{}
}
ctx = gnmi.NewContext(ctx, g.config)
return g.applyDiff(ctx, payload)
}
// Loop over args and create ops and exts
// Invalid args cause unhealable error
ops := make([]*gnmi.Operation, 0)
exts := make([]*gnmi_ext.Extension, 0)
for _, p := range args {
switch p.(type) {
case *gnmi.Operation:
op := p.(*gnmi.Operation)
if op.Target == "" {
op.Target = g.Options.Addr
}
ops = append(ops, op)
case *gnmi_ext.Extension:
exts = append(exts, p.(*gnmi_ext.Extension))
default:
return nil, &ErrInvalidParameters{
f: "gnmi.Set()",
r: "args contain invalid type",
}
}
func (g *Gnmi) applyDiff(ctx context.Context, payload change.Payload) error {
op := ctx.Value(types.CtxKeyOperation)
diff, err := ygot.Diff(payload.Original, payload.Modified)
if err != nil {
return err
}
if len(ops) == 0 {
return nil, &ErrInvalidParameters{
f: "gnmi.Set()",
r: "no operations provided",
req := &gpb.SetRequest{}
if diff.Update != nil {
switch op {
case ppb.ApiOperation_UPDATE:
req.Update = diff.Update
case ppb.ApiOperation_REPLACE:
req.Replace = diff.Update
default:
return &errors.ErrOperationNotSupported{Op: op}
}
} else if diff.Delete != nil {
req.Delete = diff.Delete
}
return g.set(ctx, ops, exts...)
resp, err := g.client.Set(ctx, req)
log.Info(resp)
return err
}
//Subscribe subscribes to a gNMI target
func (g *Gnmi) Subscribe(ctx context.Context, params ...string) error {
if g.client == nil {
return &ErrNilClient{}
return &errors.ErrNilClient{}
}
return g.subscribe(ctx)
}
@@ -129,49 +130,65 @@ func (g *Gnmi) Type() string {
return "gnmi"
}
// ProcessResponse takes a gNMI response and serializes the contents to the root struct.
// ProcessResponse takes a gNMI response and serializes the contents to the
// root struct. It logs all errors and returns an error containing the number
// off errors encountered during the process.
func (g *Gnmi) ProcessResponse(resp interface{}, root interface{}, s *ytypes.Schema) error {
models := s.SchemaTree
r := resp.(*gpb.GetResponse)
d, ok := root.(ygot.ValidatedGoStruct)
if !ok {
return &errors.ErrInvalidTypeAssertion{
Value: root,
Type: (*ygot.ValidatedGoStruct)(nil),
}
}
r, ok := resp.(*gpb.GetResponse)
if !ok {
return &errors.ErrInvalidTypeAssertion{
Value: resp,
Type: &gpb.GetResponse{},
}
}
rn := r.Notification
errs := make([]error, 0)
for _, msg := range rn {
for _, update := range msg.Update {
path := update.Path
fullPath := path
val, ok := update.Val.Value.(*gpb.TypedValue_JsonIetfVal)
if ok {
switch val := update.Val.Value.(type) {
case *gpb.TypedValue_JsonVal:
opts := []ytypes.UnmarshalOpt{&ytypes.IgnoreExtraFields{}}
if err := g.Unmarshal(val.JsonIetfVal, extraxtPathElements(fullPath), root, opts...); err != nil {
return err
if err := g.Unmarshal(val.JsonVal, path, d, opts...); err != nil {
errs = append(errs, err)
}
case *gpb.TypedValue_JsonIetfVal:
opts := []ytypes.UnmarshalOpt{&ytypes.IgnoreExtraFields{}}
if err := g.Unmarshal(val.JsonIetfVal, path, d, opts...); err != nil {
errs = append(errs, err)
}
default:
schema := s.RootSchema()
opts := []ytypes.SetNodeOpt{&ytypes.InitMissingElements{}, &ytypes.TolerateJSONInconsistencies{}}
if err := g.SetNode(schema, root, update.Path, update.Val, opts...); err != nil {
errs = append(errs, err)
}
return nil
}
// TODO(mk): Evaluate hardcoded model key
schema := models["Device"]
opts := []ytypes.SetNodeOpt{&ytypes.InitMissingElements{}, &ytypes.TolerateJSONInconsistencies{}}
if err := g.SetNode(schema, root, update.Path, update.Val, opts...); err != nil {
return err
}
}
}
return nil
}
func extraxtPathElements(path *gpb.Path) []string {
elems := make([]string, len(path.Elem))
for i, e := range path.Elem {
elems[i] = strings.Title(e.Name)
for _, e := range errs {
log.Error(e)
}
if len(errs) != 0 {
return fmt.Errorf("encountered %v errors during response processing\n%v", len(errs), errs)
}
return elems
return nil
}
// Capabilities calls GNMI capabilities
func (g *Gnmi) Capabilities(ctx context.Context) (interface{}, error) {
log.WithFields(log.Fields{
"target": g.Options.Addr,
"target": g.Options.Address,
}).Info("sending gNMI capabilities request")
ctx = gnmi.NewContext(ctx, &g.Options.Config)
ctx = context.WithValue(ctx, CtxKeyConfig, &g.Options.Config) //nolint
ctx = gnmi.NewContext(ctx, g.config)
ctx = context.WithValue(ctx, types.CtxKeyConfig, g.config) //nolint
resp, err := g.client.Capabilities(ctx, &gpb.CapabilityRequest{})
if err != nil {
return nil, err
@@ -181,9 +198,7 @@ func (g *Gnmi) Capabilities(ctx context.Context) (interface{}, error) {
// get calls GNMI get
func (g *Gnmi) get(ctx context.Context, paths [][]string, origin string) (interface{}, error) {
ctx = gnmi.NewContext(ctx, &g.Options.Config)
ctx = context.WithValue(ctx, CtxKeyConfig, &g.Options.Config) //nolint
ctx = context.WithValue(ctx, types.CtxKeyConfig, g.config) //nolint
req, err := gnmi.NewGetRequest(ctx, paths, origin)
if err != nil {
return nil, err
@@ -195,10 +210,10 @@ func (g *Gnmi) get(ctx context.Context, paths [][]string, origin string) (interf
// and returns any response.
func (g *Gnmi) getWithRequest(ctx context.Context, req *gpb.GetRequest) (interface{}, error) {
if req == nil {
return nil, &ErrNil{}
return nil, &errors.ErrNil{}
}
log.WithFields(log.Fields{
"target": g.Options.Addr,
"target": g.Options.Address,
"path": req.Path,
}).Info("sending gNMI get request")
@@ -209,34 +224,14 @@ func (g *Gnmi) getWithRequest(ctx context.Context, req *gpb.GetRequest) (interfa
return resp, nil
}
// Set calls GNMI set
func (g *Gnmi) set(ctx context.Context, setOps []*gnmi.Operation,
exts ...*gnmi_ext.Extension) (*gpb.SetResponse, error) {
ctx = gnmi.NewContext(ctx, &g.Options.Config)
targets := make([]string, len(setOps))
paths := make([][]string, len(setOps))
values := make([]string, len(setOps))
for i, v := range setOps {
targets[i] = v.Target
paths[i] = v.Path
values[i] = v.Val
}
log.WithFields(log.Fields{
"targets": targets,
"paths": paths,
"values": values,
}).Info("sending gNMI set request")
return gnmi.Set(ctx, g.client, setOps, exts...)
}
// Subscribe calls GNMI subscribe
func (g *Gnmi) subscribe(ctx context.Context) error {
ctx = gnmi.NewContext(ctx, &g.Options.Config)
opts, ok := ctx.Value(CtxKeyOpts).(*gnmi.SubscribeOptions)
ctx = gnmi.NewContext(ctx, g.config)
opts, ok := ctx.Value(types.CtxKeyOpts).(*gnmi.SubscribeOptions)
if !ok {
return &ErrInvalidTypeAssertion{
v: reflect.TypeOf(ctx.Value(CtxKeyOpts)),
t: reflect.TypeOf(&gnmi.SubscribeOptions{}),
return &errors.ErrInvalidTypeAssertion{
Value: ctx.Value(types.CtxKeyOpts),
Type: &gnmi.SubscribeOptions{},
}
}
go func() {
@@ -263,37 +258,12 @@ func (g *Gnmi) Close() error {
return nil
}
// GnmiTransportOptions implements the TransportOptions interface.
// GnmiTransportOptions contains all needed information to setup a Gnmi
// transport and therefore inherits gnmi.Config.
type GnmiTransportOptions struct {
// all needed gnmi transport parameters
gnmi.Config
SetNode func(schema *yang.Entry, root interface{}, path *gpb.Path,
val interface{}, opts ...ytypes.SetNodeOpt) error
Unmarshal func([]byte, []string, interface{}, ...ytypes.UnmarshalOpt) error
RespChan chan *gpb.SubscribeResponse
}
// GetAddress returns the address used by the transport to connect to a
// gRPC endpoint.
func (gto *GnmiTransportOptions) GetAddress() string {
return gto.Config.Addr
// SetPassthrough allows to pass an existing SetRequest. Used for cSBI
func (g *Gnmi) SetPassthrough(ctx context.Context, req *gpb.SetRequest) (*gpb.SetResponse, error) {
return g.client.Set(ctx, req)
}
// GetUsername returns the username used by the transport to connect to a
// gRPC endpoint.
func (gto *GnmiTransportOptions) GetUsername() string {
return gto.Config.Username
// GetPassthrough allows to pass an existing GetRequest. Used for cSBI
func (g *Gnmi) GetPassthrough(ctx context.Context, req *gpb.GetRequest) (*gpb.GetResponse, error) {
return g.client.Get(ctx, req)
}
// GetPassword returns the password used by the transport to connect to a
// gRPC endpoint.
func (gto *GnmiTransportOptions) GetPassword() string {
return gto.Config.Password
}
// IsTransportOption is needed to fulfill the requirements of the
// TransportOptions interface. It does not need any further implementation.
func (gto *GnmiTransportOptions) IsTransportOption() {}