gnmi_transport.go

package nucleus

import (
	"code.fbi.h-da.de/cocsn/gosdn/forks/goarista/gnmi"
	"context"
	gpb "github.com/openconfig/gnmi/proto/gnmi"
	"github.com/openconfig/gnmi/proto/gnmi_ext"
	"github.com/openconfig/goyang/pkg/yang"
	"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"
)

// Gnmi implements the Transport interface and provides an SBI with the
// possibility to access a gNMI endpoint.
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
	client    gpb.GNMIClient
}

// 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)
	if err != nil {
		return nil, err
	}
	log.WithFields(log.Fields{
		"target":   opts.Addr,
		"tls":      opts.TLS,
		"encoding": opts.Encoding,
	}).Info("building new gNMI transport")
	return &Gnmi{
		SetNode:  opts.SetNode,
		RespChan: opts.RespChan,
		Options:  opts,
		client:   c,
	}, 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{}
	}
	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) {
	if g.client == nil {
		return nil, &ErrNilClient{}
	}
	if len(args) == 0 {
		return nil, &ErrInvalidParameters{
			f: "gnmi.Set()",
			r: "no parameters provided",
		}
	}

	// 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",
			}
		}
	}
	if len(ops) == 0 {
		return nil, &ErrInvalidParameters{
			f: "gnmi.Set()",
			r: "no operations provided",
		}
	}
	return g.set(ctx, ops, exts...)
}

//Subscribe subscribes to a gNMI target
func (g *Gnmi) Subscribe(ctx context.Context, params ...string) error {
	if g.client == nil {
		return &ErrNilClient{}
	}
	return g.subscribe(ctx)
}

// Type returns the gNMI transport type
func (g *Gnmi) Type() string {
	return "gnmi"
}

// ProcessResponse takes a gNMI response and serializes the contents to the root struct.
func (g *Gnmi) ProcessResponse(resp interface{}, root interface{}, s *ytypes.Schema) error {
	models := s.SchemaTree
	r := resp.(*gpb.GetResponse)
	rn := r.Notification
	for _, msg := range rn {
		for _, update := range msg.Update {
			path := update.Path
			fullPath := path
			val, ok := update.Val.Value.(*gpb.TypedValue_JsonIetfVal)
			if ok {
				opts := []ytypes.UnmarshalOpt{&ytypes.IgnoreExtraFields{}}
				if err := g.Unmarshal(val.JsonIetfVal, extraxtPathElements(fullPath), root, opts...); err != nil {
					return 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)
	}
	return elems
}

// Capabilities calls GNMI capabilities
func (g *Gnmi) Capabilities(ctx context.Context) (interface{}, error) {
	log.WithFields(log.Fields{
		"target": g.Options.Addr,
	}).Info("sending gNMI capabilities request")
	ctx = gnmi.NewContext(ctx, &g.Options.Config)
	ctx = context.WithValue(ctx, CtxKeyConfig, &g.Options.Config) //nolint
	resp, err := g.client.Capabilities(ctx, &gpb.CapabilityRequest{})
	if err != nil {
		return nil, err
	}
	return resp, nil
}

// 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
	req, err := gnmi.NewGetRequest(ctx, paths, origin)
	if err != nil {
		return nil, err
	}
	return g.getWithRequest(ctx, req)
}

// getWithRequest takes a fully formed GetRequest, performs the Get,
// and returns any response.
func (g *Gnmi) getWithRequest(ctx context.Context, req *gpb.GetRequest) (interface{}, error) {
	if req == nil {
		return nil, &ErrNil{}
	}
	log.WithFields(log.Fields{
		"target": g.Options.Addr,
		"path":   req.Path,
	}).Info("sending gNMI get request")

	resp, err := g.client.Get(ctx, req)
	if err != nil {
		return nil, err
	}
	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)
	if !ok {
		return &ErrInvalidTypeAssertion{
			v: reflect.TypeOf(ctx.Value(CtxKeyOpts)),
			t: reflect.TypeOf(&gnmi.SubscribeOptions{}),
		}
	}
	go func() {
		log.WithFields(log.Fields{
			"address":  opts.Target,
			"paths":    opts.Paths,
			"mode":     opts.Mode,
			"interval": opts.SampleInterval,
		}).Info("subscribed to gNMI target")
		for {
			resp := <-g.RespChan
			if resp != nil {
				if err := gnmi.LogSubscribeResponse(resp); err != nil {
					log.Fatal(err)
				}
			}
		}
	}()
	return gnmi.SubscribeErr(ctx, g.client, opts, g.RespChan)
}

// Close calls GNMI close
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
}

// GetUsername returns the username used by the transport to connect to a
// gRPC endpoint.
func (gto *GnmiTransportOptions) GetUsername() string {
	return gto.Config.Username
}

// 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() {}