Merge branch 'streaming_telemetry'

Author: kkumara3

examples/grpc_cfg.py

@@ -3,35 +3,65 @@
 This is an example to show replace and merge configs work with a get command.
 The example is using XRdocs vagrant topology for all the configurations
 '''
-
+from grpc.framework.interfaces.face.face import AbortionError
+import json
+from time import sleep
 import sys
 sys.path.insert(0, '../')
 from lib.cisco_grpc_client import CiscoGRPCClient
-import json
-from time import sleep
 
-class Example:
+
+class Example(object):
     def __init__(self):
-        self.client = CiscoGRPCClient('11.1.1.10', 57777, 10, 'vagrant', 'vagrant')
+        self.client = CiscoGRPCClient('127.0.0.1', 57777, 10, 'vagrant', 'vagrant')
     def get(self):
         path = '{"Cisco-IOS-XR-ipv4-bgp-cfg:bgp": [null]}'
         result = self.client.getconfig(path)
-        print result
+        try:
+            err, result = self.client.getconfig(path)
+            if err:
+                print err
+            print json.dumps(json.loads(result))
+        except AbortionError:
+            print(
+                'Unable to connect to local box, check your gRPC destination.'
+                )
 
     def replace(self):
         path = open('snips/bgp_start.json').read()
-        result = self.client.replaceconfig(path)
-        print result # If this is sucessful, then there should be no errors.
+        try:
+            response = self.client.replaceconfig(path)
+            if response.errors:
+                err = json.loads(response.errors)
+                print err
+        except AbortionError:
+            print(
+                'Unable to connect to local box, check your gRPC destination.'
+                )
 
     def merge(self):
         path = open('snips/bgp_merge.json').read()
-        result = self.client.mergeconfig(path)
-        print result # If this is sucessful, then there should be no errors.
+        try:
+            response = self.client.mergeconfig(path)
+            if response.errors:
+                err = json.loads(response.errors)
+                print err
+        except AbortionError:
+            print(
+                'Unable to connect to local box, check your gRPC destination.'
+                )
 
     def delete(self):
         path = open('snips/bgp_start.json').read()
-        result = self.client.deleteconfig(path)
-        print result # If this is sucessful, then there should be no errors.
+        try:
+            response = self.client.deleteconfig(path)
+            if response.errors:
+                err = json.loads(response.errors)
+                print err
+        except AbortionError:
+            print(
+                'Unable to connect to local box, check your gRPC destination.'
+                )
 
 def main():
     '''

examples/grpc_example.py

@@ -1,10 +1,11 @@
 """ Basic gRPC getcall configuration. Shows how to set up the vairables
 for tls and how to put into the class and get information from the box.
 """
+import json
+from grpc.framework.interfaces.face.face import AbortionError
 import sys
 sys.path.insert(0, '../')
 from lib.cisco_grpc_client import CiscoGRPCClient
-import json
 
 def main():
     '''
@@ -14,12 +15,19 @@ def main():
     ex: client = CiscoGRPCClient('11.1.1.10', 57777, 10, 'vagrant', 'vagrant')
     '''
     creds = open('ems.pem').read()
-    options='ems.cisco.com'
-    client = CiscoGRPCClient('11.1.1.10', 57777, 10, 'vagrant', 'vagrant', creds, options)
+    options = 'ems.cisco.com'
+    client = CiscoGRPCClient('127.0.0.1', 57777, 10, 'vagrant', 'vagrant', creds, options)
     #Test 1: Test Get config json requests
     path = '{"Cisco-IOS-XR-ip-static-cfg:router-static": [null]}'
-    result = client.getconfig(path)
-    print json.dumps(json.loads(result))
+    try:
+        err, result = client.getconfig(path)
+        if err:
+            print err
+        print json.dumps(json.loads(result))
+    except AbortionError:
+        print(
+            'Unable to connect to local box, check your gRPC destination.'
+            )
 
 if __name__ == '__main__':
     main()

examples/telemetry_collector.py

@@ -18,7 +18,9 @@ def print_connectivity(connectivity):
 # Handle connectivity changes.
 client.connectivityhandler(print_connectivity)
 # Iterate over subscription recvs.
-for segment in client.getsubscription(subscription_id):
+for segment in client.getsubscription(subscription_id, unmarshal=True):
+    # unmarshal is an optional argument, default is unmarshal = True
+    # If unmarshal is false, out is in gpb k/v
 	recv_count += 1
 	print json.dumps(segment, indent=4, separators=(',', ': '))
 	print 'End Telemetry Segment'

lib/cisco_grpc_client.py

@@ -1,3 +1,25 @@
+# Copyright 2016 Cisco Systems All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#     * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+#
+# The contents of this file are 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.
+
+
 import grpc
 from grpc.beta import implementations
 import ems_grpc_pb2
@@ -6,9 +28,10 @@
 import telemetry_pb2
 
 class CiscoGRPCClient(object):
+    """This class creates grpc calls using python.
+    """
     def __init__(self, host, port, timeout, user, password, creds=None, options=None):
-        """This class creates grpc calls using python.
-            :param user: Username for device login
+        """:param user: Username for device login
             :param password: Password for device login
             :param host: The ip address for the device
             :param port: The port for the device
@@ -28,7 +51,7 @@ def __init__(self, host, port, timeout, user, password, creds=None, options=None
             self._creds = implementations.ssl_channel_credentials(creds)
             self._options = options
             channel = grpc.secure_channel(
-            self._target, self._creds, (('grpc.ssl_target_name_override', self._options,),))
+                self._target, self._creds, (('grpc.ssl_target_name_override', self._options,),))
             self._channel = implementations.Channel(channel)
         else:
             self._host = host
@@ -56,13 +79,14 @@ def getconfig(self, path):
             :rtype: Response stream object
         """
         message = ems_grpc_pb2.ConfigGetArgs(yangpathjson=path)
-        responses = self._stub.GetConfig(message,self._timeout, metadata = self._metadata)
-        objects = ''
+        responses = self._stub.GetConfig(message, self._timeout, metadata=self._metadata)
+        objects, err = '', ''
         for response in responses:
             objects += response.yangjson
-        return objects
+            err += response.errors
+        return err, objects
 
-    def getsubscription(self, sub_id):
+    def getsubscription(self, sub_id, unmarshal=True):
         """Telemetry subscription function
             :param sub_id: Subscription ID
             :type: string
@@ -72,59 +96,67 @@ def getsubscription(self, sub_id):
         sub_args = ems_grpc_pb2.CreateSubsArgs(ReqId=1, encode=3, subidstr=sub_id)
         stream = self._stub.CreateSubs(sub_args, timeout=self._timeout, metadata=self._metadata)
         for segment in stream:
-            # Go straight for telemetry data
-            telemetry_pb = telemetry_pb2.Telemetry()
-            telemetry_pb.ParseFromString(segment.data)
-            # Return in JSON format instead of protobuf.
-            yield protobuf_json.pb2json(telemetry_pb)
+            if not unmarshal:
+                yield segment
+            else:
+                # Go straight for telemetry data
+                telemetry_pb = telemetry_pb2.Telemetry()
+                telemetry_pb.ParseFromString(segment.data)
+                # Return in JSON format instead of protobuf.
+                yield protobuf_json.pb2json(telemetry_pb)
 
     def connectivityhandler(self, callback):
+        """Passing of a callback to monitor connectivety state updates.
+        :param callback: A callback for monitoring
+        :type: function
+        """
         self._channel.subscribe(callback, True)
 
-    def mergeconfig (self, yangjson):
+    def mergeconfig(self, yangjson):
         """Merge grpc call equivalent  of PATCH RESTconf call
             :param data: JSON
             :type data: str
             :return: Return the response object
             :rtype: Response object
         """
-        message = ems_grpc_pb2.ConfigArgs(yangjson= yangjson)
-        response = self._stub.MergeConfig(message, self._timeout, metadata = self._metadata)
+        message = ems_grpc_pb2.ConfigArgs(yangjson=yangjson)
+        response = self._stub.MergeConfig(message, self._timeout, metadata=self._metadata)
         return response
 
-    def deleteconfig (self, yangjson):
+    def deleteconfig(self, yangjson):
         """delete grpc call
             :param data: JSON
             :type data: str
             :return: Return the response object
             :rtype: Response object
         """
-        message = ems_grpc_pb2.ConfigArgs(yangjson= yangjson)
-        response = self._stub.DeleteConfig(message, self._timeout, metadata = self._metadata)
+        message = ems_grpc_pb2.ConfigArgs(yangjson=yangjson)
+        response = self._stub.DeleteConfig(message, self._timeout, metadata=self._metadata)
         return response
 
-    def replaceconfig (self, yangjson):
+    def replaceconfig(self, yangjson):
         """Replace grpc call equivalent of PUT in restconf
             :param data: JSON
             :type data: str
             :return: Return the response object
             :rtype: Response object
         """
-        message = ems_grpc_pb2.ConfigArgs(yangjson= yangjson)
-        response= self._stub.ReplaceConfig(message, self._timeout, metadata = self._metadata)
+        message = ems_grpc_pb2.ConfigArgs(yangjson=yangjson)
+        response = self._stub.ReplaceConfig(message, self._timeout, metadata=self._metadata)
         return response
-    def getoper (self, path):
+    def getoper(self, path):
         """ Get Oper call
             :param data: JSON
             :type data: str
             :return: Return the response object
             :rtype: Response stream object
         """
         message = ems_grpc_pb2.GetOperArgs(yangpathjson=path)
-        responses = self._stub.GetOper(message,self._timeout, metadata = self._metadata)
-        objects = ''
+        responses = self._stub.GetOper(message, self._timeout, metadata=self._metadata)
+        objects, err = '', ''
         for response in responses:
             objects += response.yangjson
+            err += response.errors
         return objects
 
     def cliconfig(self, cli):
@@ -134,40 +166,38 @@ def cliconfig(self, cli):
             :return: Return the response object
             :rtype: str
         """
-        message = ems_grpc_pb2.CliConfigArgs(cli = cli)
-        response = self._stub.CliConfig(message, self._timeout, metadata = self._metadata)
+        message = ems_grpc_pb2.CliConfigArgs(cli=cli)
+        response = self._stub.CliConfig(message, self._timeout, metadata=self._metadata)
         return response
 
-    def showcmdtextoutput (self, cli):
+    def showcmdtextoutput(self, cli):
         """ Get of CLI show commands in text
             :param data: cli show
             :type data: str
             :return: Return the response object
             :rtype: str
         """
         stub = ems_grpc_pb2.beta_create_gRPCExec_stub(self._channel)
-        message = ems_grpc_pb2.ShowCmdArgs(cli = cli)
-        responses = stub.ShowCmdTextOutput(message, self._timeout, metadata = self._metadata)
-        objects = ''
+        message = ems_grpc_pb2.ShowCmdArgs(cli=cli)
+        responses = stub.ShowCmdTextOutput(message, self._timeout, metadata=self._metadata)
+        objects, err = '', ''
         for response in responses:
             objects += response.output
+            err += response.errors
         return objects
 
-    def showcmdjsonoutput (self, cli):
+    def showcmdjsonoutput(self, cli):
         """ Get of CLI show commands in json
             :param data: cli show
             :type data: str
             :return: Return the response object
             :rtype: str
         """
         stub = ems_grpc_pb2.beta_create_gRPCExec_stub(self._channel)
-        message = ems_grpc_pb2.ShowCmdArgs(cli = cli)
-        responses = stub.ShowCmdJSONOutput(message, self._timeout, metadata = self._metadata)
-        objects = ''
+        message = ems_grpc_pb2.ShowCmdArgs(cli=cli)
+        responses = stub.ShowCmdJSONOutput(message, self._timeout, metadata=self._metadata)
+        objects, err = '', ''
         for response in responses:
             objects += response.jsonoutput
+            err += response.errors
         return objects
-
-
-
-