Skip to content

Commit

Permalink
add huawei NE8000 series support
Browse files Browse the repository at this point in the history
  • Loading branch information
automixer committed Apr 13, 2024
1 parent baa9efb commit d12b377
Show file tree
Hide file tree
Showing 13 changed files with 238 additions and 130 deletions.
1 change: 1 addition & 0 deletions config-keys.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ devices:
# "[a-zA-Z0-9_:\\-/]". Any character in the description that doesn't match
# the pattern will be removed
options: <plugin options> # Plugin specific options. See plugin-options.yaml
vendor: # Can be "generic" or "huawei". If not present, "generic" is used.

# gNMI related keys:
force_encoding: proto # Force the gNMI client to use a specific encoding. Acceptable values are:
Expand Down
1 change: 1 addition & 0 deletions pkg/core/confmgmt.go
Original file line number Diff line number Diff line change
Expand Up @@ -159,6 +159,7 @@ func (c *Core) buildGnmiClientCfg(yCfg *yamlConfig, index int) {
TLSCa: src.Keys["tls_ca"],
ForceEncoding: src.Keys["force_encoding"],
DevName: src.Keys["name"],
Vendor: src.Keys["vendor"],
}
// Bool values
flag, _ := strconv.ParseBool(src.Keys["tls"])
Expand Down
148 changes: 61 additions & 87 deletions pkg/gnmiclient/gnmiclient.go
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ const (
type plugin interface {
GetPlugName() string
GetPathsToSubscribe() []string
GetDataModels() []string
GetDataModel() string
OnSync(status bool)
Notification(nf *gnmi.Notification)
}
Expand All @@ -53,6 +53,7 @@ type Config struct {
GnmiSubscriptionMode gnmi.SubscriptionMode
GnmiUpdatesOnly bool
OverSampling int64
Vendor string
}

// GnmiClient The gNMI client object
Expand All @@ -61,14 +62,16 @@ type GnmiClient struct {
config Config
shutdown func()
encoding gnmi.Encoding
plugins map[string]plugin // Map key: plugin name
xPaths map[string]plugin // Map key: subscribed xPath (schema path)
xPathList []string // Full paths list to be subscribed. Include YANG keys filter
plugins map[string]plugin // Map key: plugin name
xPathList map[string][]string // Map key: plugin name. Paths to be subscribed, including YANG keys filter
xPaths map[string]plugin // Map key: subscribed xPath (schema path used for routing subResponses)

}

// New Creates a new GnmiClient instance
func New(cfg Config) (*GnmiClient, error) {
gClient := &GnmiClient{config: cfg}
gClient.xPathList = make(map[string][]string)
if err := gClient.clientMon.configure(cfg.DevName); err != nil {
return nil, err
}
Expand All @@ -95,14 +98,14 @@ func (c *GnmiClient) RegisterPlugin(name string, plug plugin) error {
return fmt.Errorf("plugin cannot be nil")
}

// Before registering, check for duplicated xpath
if c.xPaths == nil {
c.xPaths = make(map[string]plugin)
}

plugPaths := plug.GetPathsToSubscribe()
re := regexp.MustCompile(`\[.*?]`)
for _, reqPath := range plugPaths {
c.xPathList = append(c.xPathList, reqPath)
c.xPathList[name] = append(c.xPathList[name], reqPath)
// Remove keys from YANG path
reqPath = re.ReplaceAllString(reqPath, "")
c.xPaths[reqPath] = plug
Expand Down Expand Up @@ -212,14 +215,16 @@ func (c *GnmiClient) checkCapabilities(ctx context.Context, stub gnmi.GNMIClient
supportedModels[model.Name] = model
}
for _, plug := range c.plugins {
for _, reqModel := range plug.GetDataModels() {
if _, ok := supportedModels[reqModel]; !ok {
return fmt.Errorf("the yang model <%s> is not supported by %s", reqModel, c.config.DevName)
}
reqModel := plug.GetDataModel()
if _, ok := supportedModels[reqModel]; !ok {
return fmt.Errorf("the yang model <%s> is not supported by %s", reqModel, c.config.DevName)
}
}

// Pick an encoding
// Pick protocol buffer by default
c.encoding = gnmi.Encoding_PROTO

// Override if required
if c.config.ForceEncoding != "" {
// Config enforces encoding
c.config.ForceEncoding = strings.ToUpper(c.config.ForceEncoding)
Expand All @@ -237,85 +242,10 @@ func (c *GnmiClient) checkCapabilities(ctx context.Context, stub gnmi.GNMIClient
default:
return fmt.Errorf("the encoding %s is not supported by gNMI", c.config.ForceEncoding)
}
} else {
// Pick the first available
suppEnc := caps.GetSupportedEncodings()
if len(suppEnc) > 0 {
c.encoding = suppEnc[0]
} else {
return fmt.Errorf("%s: error reading supported encodings", c.config.DevName)
}
}
return nil
}

// subscribe creates a subscription client and sends SubscribeRequests to the server.
// It returns the subscription client and any error encountered during the process.
// The method subscribes to plugins' specified paths and constructs SubscriptionList and
// SubscribeRequest for each configured plugin.
func (c *GnmiClient) subscribe(ctx context.Context, stub gnmi.GNMIClient) (gnmi.GNMI_SubscribeClient, error) {
// Create client
gNMISubClt, err := stub.Subscribe(ctx)
if err != nil {
return nil, err
}
if c.config.OverSampling == 0 {
c.config.OverSampling = oversampling
}
if c.config.OverSampling < 1 || c.config.OverSampling > 10 {
log.Warningf("%s: Oversampling must fall between 1 and 10", c.config.DevName)
c.config.OverSampling = oversampling
}
sampleInterval := uint64(c.config.ScrapeInterval.Nanoseconds() / c.config.OverSampling)

// Prepare Subscription struct slice
subscriptions := make([]*gnmi.Subscription, 0)
for i := 0; i < len(c.plugins); i++ {
// One subscription for each plugin's path
for _, path := range c.xPathList {
p, err := ygot.StringToPath(path, ygot.StructuredPath, ygot.StringSlicePath)
if err != nil {
log.Error(err)
continue
}
newSub := &gnmi.Subscription{
Path: p,
Mode: c.config.GnmiSubscriptionMode,
SampleInterval: sampleInterval,
SuppressRedundant: false,
HeartbeatInterval: 0,
}
subscriptions = append(subscriptions, newSub)
}
}

// Prepare the SubscriptionList struct
subList := &gnmi.SubscriptionList{
Prefix: nil,
Subscription: subscriptions,
Qos: nil,
Mode: gnmi.SubscriptionList_STREAM,
AllowAggregation: false,
UseModels: nil,
Encoding: c.encoding,
UpdatesOnly: c.config.GnmiUpdatesOnly,
}

// Prepare the SubscribeRequest struct
request := &gnmi.SubscribeRequest{
Request: &gnmi.SubscribeRequest_Subscribe{Subscribe: subList},
Extension: nil,
}

// Send it to the device
err = gNMISubClt.Send(request)
if err != nil {
return nil, err
}

return gNMISubClt, nil
}

// receive takes care of receiving the GNMI stream from the device
func (c *GnmiClient) receive(sub gnmi.GNMI_SubscribeClient) error {
ch := make(chan *gnmi.SubscribeResponse, srBufferSize)
Expand Down Expand Up @@ -364,6 +294,10 @@ func (c *GnmiClient) routeSr(sr *gnmi.SubscribeResponse) {
nf := sr.GetUpdate() // Beware! GetUpdate() actually returns a notification, not an Update :-(
c.incNfCounters(uint64(len(nf.GetUpdate())), uint64(len(nf.GetDelete())))
if nf.GetPrefix().GetTarget() != "" {
// Huawei specific
if c.config.Vendor == "huawei" {
c.removeDmPfxFromPath(nf)
}
// Normal messages routing
if _, ok := c.plugins[nf.Prefix.Target]; !ok {
// Unknown destination
Expand All @@ -372,12 +306,18 @@ func (c *GnmiClient) routeSr(sr *gnmi.SubscribeResponse) {
}
c.plugins[nf.Prefix.Target].Notification(nf)
} else {
// Device does not support gnmi targeting, or the subscription does not include a prefix target
// Huawei specific
if c.config.Vendor == "huawei" {
c.removeDmPfxFromPath(nf)
}

// The device does not support gnmi targeting, or the subscription does not include a target
pfx, _ := ygot.PathToSchemaPath(nf.Prefix)
if len(pfx) < 2 {
// Empty prefix
pfx = ""
}

// Search for Updates
for _, upd := range nf.GetUpdate() {
path, _ := ygot.PathToSchemaPath(upd.Path)
Expand Down Expand Up @@ -406,6 +346,40 @@ func (c *GnmiClient) routeSr(sr *gnmi.SubscribeResponse) {
}
}

// removeDmPfxFromPath sanitizes the prefix, updates, and deletes paths in the given
// gnmi.Notification object. It removes any namespace prefix from the path names to
// ensure consistent handling of paths across plugins.
// NOTE: the deprecated "element" field is not supported
func (c *GnmiClient) removeDmPfxFromPath(nf *gnmi.Notification) {
// Sanitize Prefix
if nf.Prefix != nil && len(nf.Prefix.Elem) > 0 {
splitted := strings.SplitAfter(nf.Prefix.Elem[0].Name, ":")
if len(splitted) == 2 {
nf.Prefix.Elem[0].Name = splitted[1]
}
}

// Sanitize updates
for i := 0; i < len(nf.Update); i++ {
if nf.Update[i] != nil && nf.Update[i].Path != nil && len(nf.Update[i].Path.Elem) > 0 {
splitted := strings.SplitAfter(nf.Update[i].Path.Elem[0].Name, ":")
if len(splitted) == 2 {
nf.Update[i].Path.Elem[0].Name = splitted[1]
}
}
}

// Sanitize deletes
for i := 0; i < len(nf.Delete); i++ {
if len(nf.Delete[i].Elem) > 0 {
splitted := strings.SplitAfter(nf.Delete[i].Elem[0].Name, ":")
if len(splitted) == 2 {
nf.Delete[i].Elem[0].Name = splitted[1]
}
}
}
}

// run is the main loop for gNMI worker thread. It establishes a connection to the target
// device using the specified dial options, checks the device capabilities, subscribes to
// gNMI telemetry, and continuously receives the gNMI stream. It runs until the context is
Expand Down
89 changes: 89 additions & 0 deletions pkg/gnmiclient/subscribe.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
package gnmiclient

import (
"context"
log "github.com/golang/glog"
"github.com/openconfig/gnmi/proto/gnmi"
"github.com/openconfig/ygot/ygot"
)

// subscribe creates a subscription client and sends SubscribeRequests to the server.
// It returns the subscription client and any error encountered during the process.
func (c *GnmiClient) subscribe(ctx context.Context, stub gnmi.GNMIClient) (gnmi.GNMI_SubscribeClient, error) {
// Create client
gNMISubClt, err := stub.Subscribe(ctx)
if err != nil {
return nil, err
}
if c.config.OverSampling == 0 {
c.config.OverSampling = oversampling
}
if c.config.OverSampling < 1 || c.config.OverSampling > 10 {
log.Warningf("%s: Oversampling must fall between 1 and 10", c.config.DevName)
c.config.OverSampling = oversampling
}

// Prepare the subscription list
subLists := c.newSubList()

// Subscribe
for _, sl := range subLists {
// Prepare the SubscribeRequest struct
req := &gnmi.SubscribeRequest{
Request: &gnmi.SubscribeRequest_Subscribe{Subscribe: sl},
Extension: nil,
}
// Send it to the device
err = gNMISubClt.Send(req)
if err != nil {
return nil, err
}
}

return gNMISubClt, nil
}

// newSubList creates a list with a single subscriptions for all the configured plugins.
// This is the default way for subscribing telemetries.
func (c *GnmiClient) newSubList() []*gnmi.SubscriptionList {
var subs []*gnmi.Subscription
var subLists []*gnmi.SubscriptionList

for _, plug := range c.plugins {
for _, path := range c.xPathList[plug.GetPlugName()] {
// Huawei requires prepending the datamodel name to paths
if c.config.Vendor == "huawei" {
path = plug.GetDataModel() + ":" + path[1:]
}

// One subscription for each plugin's path
p, err := ygot.StringToPath(path, ygot.StructuredPath, ygot.StringSlicePath)
if err != nil {
log.Error(err)
continue
}
newSub := &gnmi.Subscription{
Path: p,
Mode: c.config.GnmiSubscriptionMode,
SampleInterval: uint64(c.config.ScrapeInterval.Nanoseconds() / c.config.OverSampling),
SuppressRedundant: false,
HeartbeatInterval: 0,
}
subs = append(subs, newSub)
}
}

// One subscription list per device
subLists = append(subLists, &gnmi.SubscriptionList{
Prefix: nil,
Subscription: subs,
Qos: nil,
Mode: gnmi.SubscriptionList_STREAM,
AllowAggregation: false,
UseModels: nil,
Encoding: c.encoding,
UpdatesOnly: c.config.GnmiUpdatesOnly,
})

return subLists
}
2 changes: 1 addition & 1 deletion pkg/plugins/metrics.go
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ import (
"github.com/automixer/gtexporter/pkg/exporter"
)

// smMetric is a struct used for self monitoring tasks. It contains common fields inherited from
// smMetric is a struct used for self-monitoring tasks. It contains common fields inherited from
// exporter.MetricCommons, as well as additional fields specific to the
// SM exporter.
type smMetric struct {
Expand Down
Loading

0 comments on commit d12b377

Please sign in to comment.