Add Cisco model-driven telemetry & GNMI inputs (#5852)

plugins/inputs/cisco_telemetry_gnmi/README.md

@@ -0,0 +1,75 @@
+# Cisco GNMI telemetry
+
+Cisco GNMI telemetry is an input plugin that consumes telemetry data similar to the [GNMI specification](https://github.com/openconfig/reference/blob/master/rpc/gnmi/gnmi-specification.md).
+This GRPC-based protocol can utilize TLS for authentication and encryption.
+
+This plugin has been developed to support GNMI telemetry as produced by Cisco IOS XR (64-bit) version 6.5.1 and later.
+
+
+### Configuration:
+
+This is a sample configuration for the plugin.
+
+```toml
+[[inputs.cisco_telemetry_gnmi]]
+  ## Address and port of the GNMI GRPC server
+  addresses = ["10.49.234.114:57777"]
+
+  ## define credentials
+  username = "cisco"
+  password = "cisco"
+
+  ## GNMI encoding requested (one of: "proto", "json", "json_ietf")
+  # encoding = "proto"
+
+  ## redial in case of failures after
+  redial = "10s"
+
+  ## enable client-side TLS and define CA to authenticate the device
+  # enable_tls = true
+  # tls_ca = "/etc/telegraf/ca.pem"
+  # insecure_skip_verify = true
+
+  ## define client-side TLS certificate & key to authenticate to the device
+  # tls_cert = "/etc/telegraf/cert.pem"
+  # tls_key = "/etc/telegraf/key.pem"
+
+  ## GNMI subscription prefix (optional, can usually be left empty)
+  ## See: https://github.com/openconfig/reference/blob/master/rpc/gnmi/gnmi-specification.md#222-paths
+  # origin = ""
+  # prefix = ""
+  # target = ""
+
+  ## Define additional aliases to map telemetry encoding paths to simple measurement names
+  # [inputs.cisco_telemetry_gnmi.aliases]
+  #   ifcounters = "openconfig:/interfaces/interface/state/counters"
+
+  [[inputs.cisco_telemetry_gnmi.subscription]]
+    ## Name of the measurement that will be emitted
+    name = "ifcounters"
+
+    ## Origin and path of the subscription
+    ## See: https://github.com/openconfig/reference/blob/master/rpc/gnmi/gnmi-specification.md#222-paths
+    ##
+    ## origin usually refers to a (YANG) data model implemented by the device
+    ## and path to a specific substructe inside it that should be subscribed to (similar to an XPath)
+    ## YANG models can be found e.g. here: https://github.com/YangModels/yang/tree/master/vendor/cisco/xr
+    origin = "openconfig-interfaces"
+    path = "/interfaces/interface/state/counters"
+
+    # Subscription mode (one of: "target_defined", "sample", "on_change") and interval
+    subscription_mode = "sample"
+    sample_interval = "10s"
+
+    ## Suppress redundant transmissions when measured values are unchanged
+    # suppress_redundant = false
+
+    ## If suppression is enabled, send updates at least every X seconds anyway
+    # heartbeat_interval = "60s"
+```
+
+### Example Output
+```
+ifcounters,path=openconfig-interfaces:/interfaces/interface/state/counters,host=linux,name=MgmtEth0/RP0/CPU0/0,source=10.49.234.115 in-multicast-pkts=0i,out-multicast-pkts=0i,out-errors=0i,out-discards=0i,in-broadcast-pkts=0i,out-broadcast-pkts=0i,in-discards=0i,in-unknown-protos=0i,in-errors=0i,out-unicast-pkts=0i,in-octets=0i,out-octets=0i,last-clear="2019-05-22T16:53:21Z",in-unicast-pkts=0i 1559145777425000000
+ifcounters,path=openconfig-interfaces:/interfaces/interface/state/counters,host=linux,name=GigabitEthernet0/0/0/0,source=10.49.234.115 out-multicast-pkts=0i,out-broadcast-pkts=0i,in-errors=0i,out-errors=0i,in-discards=0i,out-octets=0i,in-unknown-protos=0i,in-unicast-pkts=0i,in-octets=0i,in-multicast-pkts=0i,in-broadcast-pkts=0i,last-clear="2019-05-22T16:54:50Z",out-unicast-pkts=0i,out-discards=0i 1559145777425000000
+```

plugins/inputs/cisco_telemetry_gnmi/cisco_telemetry_gnmi.go

@@ -0,0 +1,516 @@
+package cisco_telemetry_gnmi
+
+import (
+	"bytes"
+	"context"
+	"crypto/tls"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"net"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/influxdata/telegraf/metric"
+
+	"google.golang.org/grpc/credentials"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/internal"
+	internaltls "github.com/influxdata/telegraf/internal/tls"
+	"github.com/influxdata/telegraf/plugins/inputs"
+	jsonparser "github.com/influxdata/telegraf/plugins/parsers/json"
+	"github.com/openconfig/gnmi/proto/gnmi"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
+)
+
+// CiscoTelemetryGNMI plugin instance
+type CiscoTelemetryGNMI struct {
+	Addresses     []string          `toml:"addresses"`
+	Subscriptions []Subscription    `toml:"subscription"`
+	Aliases       map[string]string `toml:"aliases"`
+
+	// Optional subscription configuration
+	Encoding    string
+	Origin      string
+	Prefix      string
+	Target      string
+	UpdatesOnly bool `toml:"updates_only"`
+
+	// Cisco IOS XR credentials
+	Username string
+	Password string
+
+	// Redial
+	Redial internal.Duration
+
+	// GRPC TLS settings
+	EnableTLS bool `toml:"enable_tls"`
+	internaltls.ClientConfig
+
+	// Internal state
+	aliases map[string]string
+	acc     telegraf.Accumulator
+	cancel  context.CancelFunc
+	wg      sync.WaitGroup
+}
+
+// Subscription for a GNMI client
+type Subscription struct {
+	Name   string
+	Origin string
+	Path   string
+
+	// Subscription mode and interval
+	SubscriptionMode string            `toml:"subscription_mode"`
+	SampleInterval   internal.Duration `toml:"sample_interval"`
+
+	// Duplicate suppression
+	SuppressRedundant bool              `toml:"suppress_redundant"`
+	HeartbeatInterval internal.Duration `toml:"heartbeat_interval"`
+}
+
+// Start the http listener service
+func (c *CiscoTelemetryGNMI) Start(acc telegraf.Accumulator) error {
+	var err error
+	var ctx context.Context
+	var tlscfg *tls.Config
+	var request *gnmi.SubscribeRequest
+	c.acc = acc
+	ctx, c.cancel = context.WithCancel(context.Background())
+
+	// Validate configuration
+	if request, err = c.newSubscribeRequest(); err != nil {
+		return err
+	} else if c.Redial.Duration.Nanoseconds() <= 0 {
+		return fmt.Errorf("redial duration must be positive")
+	}
+
+	// Parse TLS config
+	if c.EnableTLS {
+		if tlscfg, err = c.ClientConfig.TLSConfig(); err != nil {
+			return err
+		}
+	}
+
+	if len(c.Username) > 0 {
+		ctx = metadata.AppendToOutgoingContext(ctx, "username", c.Username, "password", c.Password)
+	}
+
+	// Invert explicit alias list and prefill subscription names
+	c.aliases = make(map[string]string, len(c.Subscriptions)+len(c.Aliases))
+	for _, subscription := range c.Subscriptions {
+		path := subscription.Path
+		if len(subscription.Origin) > 0 {
+			path = subscription.Origin + ":" + path
+		}
+
+		name := subscription.Name
+		if len(name) == 0 {
+			name = path[strings.LastIndexByte(path, '/')+1:]
+		}
+		if len(name) > 0 {
+			c.aliases[path] = name
+		}
+	}
+	for alias, path := range c.Aliases {
+		c.aliases[path] = alias
+	}
+
+	// Create a goroutine for each device, dial and subscribe
+	c.wg.Add(len(c.Addresses))
+	for _, addr := range c.Addresses {
+		go func(address string) {
+			defer c.wg.Done()
+			for ctx.Err() == nil {
+				if err := c.subscribeGNMI(ctx, address, tlscfg, request); err != nil && ctx.Err() == nil {
+					acc.AddError(err)
+				}
+
+				select {
+				case <-ctx.Done():
+				case <-time.After(c.Redial.Duration):
+				}
+			}
+		}(addr)
+	}
+	return nil
+}
+
+// Create a new GNMI SubscribeRequest
+func (c *CiscoTelemetryGNMI) newSubscribeRequest() (*gnmi.SubscribeRequest, error) {
+	// Create subscription objects
+	subscriptions := make([]*gnmi.Subscription, len(c.Subscriptions))
+	for i, subscription := range c.Subscriptions {
+		gnmiPath, err := parsePath(subscription.Origin, subscription.Path, "")
+		if err != nil {
+			return nil, err
+		}
+		mode, ok := gnmi.SubscriptionMode_value[strings.ToUpper(subscription.SubscriptionMode)]
+		if !ok {
+			return nil, fmt.Errorf("invalid subscription mode %s", subscription.SubscriptionMode)
+		}
+		subscriptions[i] = &gnmi.Subscription{
+			Path:              gnmiPath,
+			Mode:              gnmi.SubscriptionMode(mode),
+			SampleInterval:    uint64(subscription.SampleInterval.Duration.Nanoseconds()),
+			SuppressRedundant: subscription.SuppressRedundant,
+			HeartbeatInterval: uint64(subscription.HeartbeatInterval.Duration.Nanoseconds()),
+		}
+	}
+
+	// Construct subscribe request
+	gnmiPath, err := parsePath(c.Origin, c.Prefix, c.Target)
+	if err != nil {
+		return nil, err
+	}
+
+	if c.Encoding != "proto" && c.Encoding != "json" && c.Encoding != "json_ietf" {
+		return nil, fmt.Errorf("unsupported encoding %s", c.Encoding)
+	}
+
+	return &gnmi.SubscribeRequest{
+		Request: &gnmi.SubscribeRequest_Subscribe{
+			Subscribe: &gnmi.SubscriptionList{
+				Prefix:       gnmiPath,
+				Mode:         gnmi.SubscriptionList_STREAM,
+				Encoding:     gnmi.Encoding(gnmi.Encoding_value[strings.ToUpper(c.Encoding)]),
+				Subscription: subscriptions,
+				UpdatesOnly:  c.UpdatesOnly,
+			},
+		},
+	}, nil
+}
+
+// SubscribeGNMI and extract telemetry data
+func (c *CiscoTelemetryGNMI) subscribeGNMI(ctx context.Context, address string, tlscfg *tls.Config, request *gnmi.SubscribeRequest) error {
+	var opt grpc.DialOption
+	if tlscfg != nil {
+		opt = grpc.WithTransportCredentials(credentials.NewTLS(tlscfg))
+	} else {
+		opt = grpc.WithInsecure()
+	}
+
+	client, err := grpc.DialContext(ctx, address, opt)
+	if err != nil {
+		return fmt.Errorf("failed to dial: %v", err)
+	}
+	defer client.Close()
+
+	subscribeClient, err := gnmi.NewGNMIClient(client).Subscribe(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to setup subscription: %v", err)
+	}
+
+	if err = subscribeClient.Send(request); err != nil {
+		return fmt.Errorf("failed to send subscription request: %v", err)
+	}
+
+	log.Printf("D! [inputs.cisco_telemetry_gnmi]: Connection to GNMI device %s established", address)
+	defer log.Printf("D! [inputs.cisco_telemetry_gnmi]: Connection to GNMI device %s closed", address)
+	for ctx.Err() == nil {
+		var reply *gnmi.SubscribeResponse
+		if reply, err = subscribeClient.Recv(); err != nil {
+			if err != io.EOF && ctx.Err() == nil {
+				return fmt.Errorf("aborted GNMI subscription: %v", err)
+			}
+			break
+		}
+
+		c.handleSubscribeResponse(address, reply)
+	}
+	return nil
+}
+
+// HandleSubscribeResponse message from GNMI and parse contained telemetry data
+func (c *CiscoTelemetryGNMI) handleSubscribeResponse(address string, reply *gnmi.SubscribeResponse) {
+	// Check if response is a GNMI Update and if we have a prefix to derive the measurement name
+	response, ok := reply.Response.(*gnmi.SubscribeResponse_Update)
+	if !ok {
+		return
+	}
+
+	var prefix, prefixAliasPath string
+	grouper := metric.NewSeriesGrouper()
+	timestamp := time.Unix(0, response.Update.Timestamp)
+	prefixTags := make(map[string]string)
+
+	if response.Update.Prefix != nil {
+		prefix, prefixAliasPath = c.handlePath(response.Update.Prefix, prefixTags, "")
+	}
+	prefixTags["source"], _, _ = net.SplitHostPort(address)
+	prefixTags["path"] = prefix
+
+	// Parse individual Update message and create measurements
+	var name, lastAliasPath string
+	for _, update := range response.Update.Update {
+		// Prepare tags from prefix
+		tags := make(map[string]string, len(prefixTags))
+		for key, val := range prefixTags {
+			tags[key] = val
+		}
+		aliasPath, fields := c.handleTelemetryField(update, tags, prefix)
+
+		// Inherent valid alias from prefix parsing
+		if len(prefixAliasPath) > 0 && len(aliasPath) == 0 {
+			aliasPath = prefixAliasPath
+		}
+
+		// Lookup alias if alias-path has changed
+		if aliasPath != lastAliasPath {
+			name = prefix
+			if alias, ok := c.aliases[aliasPath]; ok {
+				name = alias
+			} else {
+				log.Printf("D! [inputs.cisco_telemetry_gnmi]: No measurement alias for GNMI path: %s", name)
+			}
+		}
+
+		// Group metrics
+		for key, val := range fields {
+			grouper.Add(name, tags, timestamp, key[len(aliasPath)+1:], val)
+		}
+
+		lastAliasPath = aliasPath
+	}
+
+	// Add grouped measurements
+	for _, metric := range grouper.Metrics() {
+		c.acc.AddMetric(metric)
+	}
+}
+
+// HandleTelemetryField and add it to a measurement
+func (c *CiscoTelemetryGNMI) handleTelemetryField(update *gnmi.Update, tags map[string]string, prefix string) (string, map[string]interface{}) {
+	path, aliasPath := c.handlePath(update.Path, tags, prefix)
+
+	var value interface{}
+	var jsondata []byte
+
+	switch val := update.Val.Value.(type) {
+	case *gnmi.TypedValue_AsciiVal:
+		value = val.AsciiVal
+	case *gnmi.TypedValue_BoolVal:
+		value = val.BoolVal
+	case *gnmi.TypedValue_BytesVal:
+		value = val.BytesVal
+	case *gnmi.TypedValue_DecimalVal:
+		value = val.DecimalVal
+	case *gnmi.TypedValue_FloatVal:
+		value = val.FloatVal
+	case *gnmi.TypedValue_IntVal:
+		value = val.IntVal
+	case *gnmi.TypedValue_StringVal:
+		value = val.StringVal
+	case *gnmi.TypedValue_UintVal:
+		value = val.UintVal
+	case *gnmi.TypedValue_JsonIetfVal:
+		jsondata = val.JsonIetfVal
+	case *gnmi.TypedValue_JsonVal:
+		jsondata = val.JsonVal
+	}
+
+	name := strings.Replace(path, "-", "_", -1)
+	fields := make(map[string]interface{})
+	if value != nil {
+		fields[name] = value
+	} else if jsondata != nil {
+		if err := json.Unmarshal(jsondata, &value); err != nil {
+			c.acc.AddError(fmt.Errorf("failed to parse JSON value: %v", err))
+		} else {
+			flattener := jsonparser.JSONFlattener{Fields: fields}
+			flattener.FullFlattenJSON(name, value, true, true)
+		}
+	}
+	return aliasPath, fields
+}
+
+// Parse path to path-buffer and tag-field
+func (c *CiscoTelemetryGNMI) handlePath(path *gnmi.Path, tags map[string]string, prefix string) (string, string) {
+	var aliasPath string
+	builder := bytes.NewBufferString(prefix)
+
+	// Prefix with origin
+	if len(path.Origin) > 0 {
+		builder.WriteString(path.Origin)
+		builder.WriteRune(':')
+	}
+
+	// Parse generic keys from prefix
+	for _, elem := range path.Elem {
+		builder.WriteRune('/')
+		builder.WriteString(elem.Name)
+		name := builder.String()
+
+		if _, exists := c.aliases[name]; exists {
+			aliasPath = name
+		}
+
+		for key, val := range elem.Key {
+			key = strings.Replace(key, "-", "_", -1)
+
+			// Use short-form of key if possible
+			if _, exists := tags[key]; exists {
+				tags[name+"/"+key] = val
+			} else {
+				tags[key] = val
+			}
+		}
+	}
+
+	return builder.String(), aliasPath
+}
+
+//ParsePath from XPath-like string to GNMI path structure
+func parsePath(origin string, path string, target string) (*gnmi.Path, error) {
+	var err error
+	gnmiPath := gnmi.Path{Origin: origin, Target: target}
+
+	if len(path) > 0 && path[0] != '/' {
+		return nil, fmt.Errorf("path does not start with a '/': %s", path)
+	}
+
+	elem := &gnmi.PathElem{}
+	start, name, value, end := 0, -1, -1, -1
+
+	path = path + "/"
+
+	for i := 0; i < len(path); i++ {
+		if path[i] == '[' {
+			if name >= 0 {
+				break
+			}
+			if end < 0 {
+				end = i
+				elem.Key = make(map[string]string)
+			}
+			name = i + 1
+		} else if path[i] == '=' {
+			if name <= 0 || value >= 0 {
+				break
+			}
+			value = i + 1
+		} else if path[i] == ']' {
+			if name <= 0 || value <= name {
+				break
+			}
+			elem.Key[path[name:value-1]] = strings.Trim(path[value:i], "'\"")
+			name, value = -1, -1
+		} else if path[i] == '/' {
+			if name < 0 {
+				if end < 0 {
+					end = i
+				}
+
+				if end > start {
+					elem.Name = path[start:end]
+					gnmiPath.Elem = append(gnmiPath.Elem, elem)
+					gnmiPath.Element = append(gnmiPath.Element, path[start:i])
+				}
+
+				start, name, value, end = i+1, -1, -1, -1
+				elem = &gnmi.PathElem{}
+			}
+		}
+	}
+
+	if name >= 0 || value >= 0 {
+		err = fmt.Errorf("Invalid GNMI path: %s", path)
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	return &gnmiPath, nil
+}
+
+// Stop listener and cleanup
+func (c *CiscoTelemetryGNMI) Stop() {
+	c.cancel()
+	c.wg.Wait()
+}
+
+const sampleConfig = `
+ ## Address and port of the GNMI GRPC server
+ addresses = ["10.49.234.114:57777"]
+
+ ## define credentials
+ username = "cisco"
+ password = "cisco"
+
+ ## GNMI encoding requested (one of: "proto", "json", "json_ietf")
+ # encoding = "proto"
+
+ ## redial in case of failures after
+ redial = "10s"
+
+ ## enable client-side TLS and define CA to authenticate the device
+ # enable_tls = true
+ # tls_ca = "/etc/telegraf/ca.pem"
+ # insecure_skip_verify = true
+
+ ## define client-side TLS certificate & key to authenticate to the device
+ # tls_cert = "/etc/telegraf/cert.pem"
+ # tls_key = "/etc/telegraf/key.pem"
+
+ ## GNMI subscription prefix (optional, can usually be left empty)
+ ## See: https://github.com/openconfig/reference/blob/master/rpc/gnmi/gnmi-specification.md#222-paths
+ # origin = ""
+ # prefix = ""
+ # target = ""
+
+ ## Define additional aliases to map telemetry encoding paths to simple measurement names
+ #[inputs.cisco_telemetry_gnmi.aliases]
+ #  ifcounters = "openconfig:/interfaces/interface/state/counters"
+
+ [[inputs.cisco_telemetry_gnmi.subscription]]
+  ## Name of the measurement that will be emitted
+  name = "ifcounters"
+
+  ## Origin and path of the subscription
+  ## See: https://github.com/openconfig/reference/blob/master/rpc/gnmi/gnmi-specification.md#222-paths
+  ##
+  ## origin usually refers to a (YANG) data model implemented by the device
+  ## and path to a specific substructe inside it that should be subscribed to (similar to an XPath)
+  ## YANG models can be found e.g. here: https://github.com/YangModels/yang/tree/master/vendor/cisco/xr
+  origin = "openconfig-interfaces"
+  path = "/interfaces/interface/state/counters"
+
+  # Subscription mode (one of: "target_defined", "sample", "on_change") and interval
+  subscription_mode = "sample"
+  sample_interval = "10s"
+
+  ## Suppress redundant transmissions when measured values are unchanged
+  # suppress_redundant = false
+
+  ## If suppression is enabled, send updates at least every X seconds anyway
+  # heartbeat_interval = "60s"
+`
+
+// SampleConfig of plugin
+func (c *CiscoTelemetryGNMI) SampleConfig() string {
+	return sampleConfig
+}
+
+// Description of plugin
+func (c *CiscoTelemetryGNMI) Description() string {
+	return "Cisco GNMI telemetry input plugin based on GNMI telemetry data produced in IOS XR"
+}
+
+// Gather plugin measurements (unused)
+func (c *CiscoTelemetryGNMI) Gather(_ telegraf.Accumulator) error {
+	return nil
+}
+
+func init() {
+	inputs.Add("cisco_telemetry_gnmi", func() telegraf.Input {
+		return &CiscoTelemetryGNMI{
+			Encoding: "proto",
+			Redial:   internal.Duration{Duration: 10 * time.Second},
+		}
+	})
+}

plugins/inputs/cisco_telemetry_gnmi/cisco_telemetry_gnmi_test.go

@@ -0,0 +1,247 @@
+package cisco_telemetry_gnmi
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc/metadata"
+
+	"github.com/influxdata/telegraf/internal"
+	"github.com/influxdata/telegraf/testutil"
+	"google.golang.org/grpc"
+
+	"github.com/openconfig/gnmi/proto/gnmi"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestParsePath(t *testing.T) {
+	path := "/foo/bar/bla[shoo=woo][shoop=/woop/]/z"
+	parsed, err := parsePath("theorigin", path, "thetarget")
+
+	assert.Nil(t, err)
+	assert.Equal(t, parsed.Origin, "theorigin")
+	assert.Equal(t, parsed.Target, "thetarget")
+	assert.Equal(t, parsed.Element, []string{"foo", "bar", "bla[shoo=woo][shoop=/woop/]", "z"})
+	assert.Equal(t, parsed.Elem, []*gnmi.PathElem{{Name: "foo"}, {Name: "bar"},
+		{Name: "bla", Key: map[string]string{"shoo": "woo", "shoop": "/woop/"}}, {Name: "z"}})
+
+	parsed, err = parsePath("", "", "")
+	assert.Nil(t, err)
+	assert.Equal(t, *parsed, gnmi.Path{})
+
+	parsed, err = parsePath("", "/foo[[", "")
+	assert.Nil(t, parsed)
+	assert.Equal(t, errors.New("Invalid GNMI path: /foo[[/"), err)
+}
+
+type mockGNMIServer struct {
+	t        *testing.T
+	acc      *testutil.Accumulator
+	server   *grpc.Server
+	scenario int
+}
+
+func (m *mockGNMIServer) Capabilities(context.Context, *gnmi.CapabilityRequest) (*gnmi.CapabilityResponse, error) {
+	return nil, nil
+}
+
+func (m *mockGNMIServer) Get(context.Context, *gnmi.GetRequest) (*gnmi.GetResponse, error) {
+	return nil, nil
+}
+
+func (m *mockGNMIServer) Set(context.Context, *gnmi.SetRequest) (*gnmi.SetResponse, error) {
+	return nil, nil
+}
+
+func (m *mockGNMIServer) Subscribe(server gnmi.GNMI_SubscribeServer) error {
+	// Avoid race conditions
+	go func() {
+		if m.scenario == 0 {
+			m.acc.WaitError(1)
+		} else if m.scenario == 1 || m.scenario == 3 {
+			m.acc.Wait(4)
+		} else if m.scenario == 2 {
+			m.acc.Wait(2)
+		}
+		if m.scenario >= 0 {
+			m.server.Stop()
+		}
+	}()
+
+	metadata, ok := metadata.FromIncomingContext(server.Context())
+	assert.Equal(m.t, ok, true)
+	assert.Equal(m.t, metadata.Get("username"), []string{"theuser"})
+	assert.Equal(m.t, metadata.Get("password"), []string{"thepassword"})
+
+	switch m.scenario {
+	case 0:
+		return fmt.Errorf("testerror")
+	case 1:
+		notification := mockGNMINotification()
+		server.Send(&gnmi.SubscribeResponse{Response: &gnmi.SubscribeResponse_Update{Update: notification}})
+		server.Send(&gnmi.SubscribeResponse{Response: &gnmi.SubscribeResponse_SyncResponse{SyncResponse: true}})
+		notification.Prefix.Elem[0].Key["foo"] = "bar2"
+		notification.Update[0].Path.Elem[1].Key["name"] = "str2"
+		notification.Update[0].Val = &gnmi.TypedValue{Value: &gnmi.TypedValue_JsonVal{JsonVal: []byte{'"', '1', '2', '3', '"'}}}
+		server.Send(&gnmi.SubscribeResponse{Response: &gnmi.SubscribeResponse_Update{Update: notification}})
+		return nil
+	case 2:
+		notification := mockGNMINotification()
+		server.Send(&gnmi.SubscribeResponse{Response: &gnmi.SubscribeResponse_Update{Update: notification}})
+		return nil
+	case 3:
+		notification := mockGNMINotification()
+		notification.Prefix.Elem[0].Key["foo"] = "bar2"
+		notification.Update[0].Path.Elem[1].Key["name"] = "str2"
+		notification.Update[0].Val = &gnmi.TypedValue{Value: &gnmi.TypedValue_BoolVal{BoolVal: false}}
+		server.Send(&gnmi.SubscribeResponse{Response: &gnmi.SubscribeResponse_Update{Update: notification}})
+		return nil
+	default:
+		return fmt.Errorf("test not implemented ;)")
+	}
+}
+
+func TestGNMIError(t *testing.T) {
+	listener, _ := net.Listen("tcp", "127.0.0.1:57003")
+	server := grpc.NewServer()
+	acc := &testutil.Accumulator{}
+	gnmi.RegisterGNMIServer(server, &mockGNMIServer{t: t, scenario: 0, server: server, acc: acc})
+
+	c := &CiscoTelemetryGNMI{Addresses: []string{"127.0.0.1:57003"},
+		Username: "theuser", Password: "thepassword", Encoding: "proto",
+		Redial: internal.Duration{Duration: 1 * time.Second}}
+
+	assert.Nil(t, c.Start(acc))
+	server.Serve(listener)
+	c.Stop()
+
+	assert.Contains(t, acc.Errors, errors.New("aborted GNMI subscription: rpc error: code = Unknown desc = testerror"))
+}
+
+func mockGNMINotification() *gnmi.Notification {
+	return &gnmi.Notification{
+		Timestamp: 1543236572000000000,
+		Prefix: &gnmi.Path{
+			Origin: "type",
+			Elem: []*gnmi.PathElem{
+				{
+					Name: "model",
+					Key:  map[string]string{"foo": "bar"},
+				},
+			},
+			Target: "subscription",
+		},
+		Update: []*gnmi.Update{
+			{
+				Path: &gnmi.Path{
+					Elem: []*gnmi.PathElem{
+						{Name: "some"},
+						{
+							Name: "path",
+							Key:  map[string]string{"name": "str", "uint64": "1234"}},
+					},
+				},
+				Val: &gnmi.TypedValue{Value: &gnmi.TypedValue_IntVal{IntVal: 5678}},
+			},
+			{
+				Path: &gnmi.Path{
+					Elem: []*gnmi.PathElem{
+						{Name: "other"},
+						{Name: "path"},
+					},
+				},
+				Val: &gnmi.TypedValue{Value: &gnmi.TypedValue_StringVal{StringVal: "foobar"}},
+			},
+			{
+				Path: &gnmi.Path{
+					Elem: []*gnmi.PathElem{
+						{Name: "other"},
+						{Name: "this"},
+					},
+				},
+				Val: &gnmi.TypedValue{Value: &gnmi.TypedValue_StringVal{StringVal: "that"}},
+			},
+		},
+	}
+}
+
+func TestGNMIMultiple(t *testing.T) {
+	listener, _ := net.Listen("tcp", "127.0.0.1:57004")
+	server := grpc.NewServer()
+	acc := &testutil.Accumulator{}
+	gnmi.RegisterGNMIServer(server, &mockGNMIServer{t: t, scenario: 1, server: server, acc: acc})
+
+	c := &CiscoTelemetryGNMI{Addresses: []string{"127.0.0.1:57004"},
+		Username: "theuser", Password: "thepassword", Encoding: "proto",
+		Redial:        internal.Duration{Duration: 1 * time.Second},
+		Subscriptions: []Subscription{{Name: "alias", Origin: "type", Path: "/model", SubscriptionMode: "sample"}},
+	}
+
+	assert.Nil(t, c.Start(acc))
+
+	server.Serve(listener)
+	c.Stop()
+
+	assert.Empty(t, acc.Errors)
+
+	tags := map[string]string{"path": "type:/model", "source": "127.0.0.1", "foo": "bar", "name": "str", "uint64": "1234"}
+	fields := map[string]interface{}{"some/path": int64(5678)}
+	acc.AssertContainsTaggedFields(t, "alias", fields, tags)
+
+	tags = map[string]string{"path": "type:/model", "source": "127.0.0.1", "foo": "bar"}
+	fields = map[string]interface{}{"other/path": "foobar", "other/this": "that"}
+	acc.AssertContainsTaggedFields(t, "alias", fields, tags)
+
+	tags = map[string]string{"path": "type:/model", "foo": "bar2", "source": "127.0.0.1", "name": "str2", "uint64": "1234"}
+	fields = map[string]interface{}{"some/path": "123"}
+	acc.AssertContainsTaggedFields(t, "alias", fields, tags)
+
+	tags = map[string]string{"path": "type:/model", "source": "127.0.0.1", "foo": "bar2"}
+	fields = map[string]interface{}{"other/path": "foobar", "other/this": "that"}
+	acc.AssertContainsTaggedFields(t, "alias", fields, tags)
+}
+
+func TestGNMIMultipleRedial(t *testing.T) {
+	listener, _ := net.Listen("tcp", "127.0.0.1:57004")
+	server := grpc.NewServer()
+	acc := &testutil.Accumulator{}
+	gnmi.RegisterGNMIServer(server, &mockGNMIServer{t: t, scenario: 2, server: server, acc: acc})
+
+	c := &CiscoTelemetryGNMI{Addresses: []string{"127.0.0.1:57004"},
+		Username: "theuser", Password: "thepassword", Encoding: "proto",
+		Redial:        internal.Duration{Duration: 500 * time.Millisecond},
+		Subscriptions: []Subscription{{Name: "alias", Origin: "type", Path: "/model", SubscriptionMode: "sample"}},
+	}
+
+	assert.Nil(t, c.Start(acc))
+	server.Serve(listener)
+
+	listener, _ = net.Listen("tcp", "127.0.0.1:57004")
+	server = grpc.NewServer()
+	gnmi.RegisterGNMIServer(server, &mockGNMIServer{t: t, scenario: 3, server: server, acc: acc})
+
+	server.Serve(listener)
+	c.Stop()
+
+	assert.Empty(t, acc.Errors)
+
+	tags := map[string]string{"path": "type:/model", "source": "127.0.0.1", "foo": "bar", "name": "str", "uint64": "1234"}
+	fields := map[string]interface{}{"some/path": int64(5678)}
+	acc.AssertContainsTaggedFields(t, "alias", fields, tags)
+
+	tags = map[string]string{"path": "type:/model", "source": "127.0.0.1", "foo": "bar"}
+	fields = map[string]interface{}{"other/path": "foobar", "other/this": "that"}
+	acc.AssertContainsTaggedFields(t, "alias", fields, tags)
+
+	tags = map[string]string{"path": "type:/model", "foo": "bar2", "source": "127.0.0.1", "name": "str2", "uint64": "1234"}
+	fields = map[string]interface{}{"some/path": false}
+	acc.AssertContainsTaggedFields(t, "alias", fields, tags)
+
+	tags = map[string]string{"path": "type:/model", "source": "127.0.0.1", "foo": "bar2"}
+	fields = map[string]interface{}{"other/path": "foobar", "other/this": "that"}
+	acc.AssertContainsTaggedFields(t, "alias", fields, tags)
+}