Setup collector and source

Author: priyansh-sumo

EventHubs/requirements.txt

@@ -7,3 +7,4 @@ azure-mgmt-eventhub==11.0.0
 azure-eventhub==5.11.5
 azure-mgmt-storage==21.1.0
 zure-mgmt-loganalytics==12.0.0
+sumologic-sdk==0.1.15

EventHubs/tests/baseeventhubtest.py

@@ -19,11 +19,6 @@ def setUp(self):
         self.create_credentials()
         self.resource_client = ResourceManagementClient(self.azure_credential, 
                                                         self.subscription_id)
-        try:
-            self.sumo_endpoint_url = os.environ["SumoEndpointURL"]
-        except KeyError:
-            raise Exception("SumoEndpointURL environment variables are not set")
-        
         self.repo_name, self.branch_name = self.get_git_info()
 
     def tearDown(self):

EventHubs/tests/test_eventhub_metrics.py

@@ -10,7 +10,10 @@ class TestEventHubMetrics(BaseEventHubTest):
 
     def setUp(self):
         super(TestEventHubMetrics, self).setUp()
-        self.RESOURCE_GROUP_NAME = "EventHubMetrics-%s" % (datetime.now().strftime("%d-%m-%y-%H-%M-%S"))
+        datetime_value = datetime.now().strftime("%d-%m-%y-%H-%M-%S")
+        self.collector_name = "azure_metric_unittest-%s" % (datetime_value)
+        self.source_name = "metric_data-%s" % (datetime_value)
+        self.RESOURCE_GROUP_NAME = "EventHubMetrics-%s" % (datetime_value)
         self.template_name = "azuredeploy_metrics.json"
         self.event_hub_namespace_prefix = "SMNamespace"
         self.eventhub_name = "insights-metrics-pt1m"
@@ -19,6 +22,8 @@ def setUp(self):
         self.expected_resource_count = 7
 
     def test_pipeline(self):
+        collector_id = self.create_collector(self.collector_name)
+        self.sumo_source_id, self.sumo_endpoint_url = self.create_source(collector_id, self.source_name)
         self.create_resource_group()
         self.deploy_template()
         self.assertTrue(self.resource_group_exists(self.RESOURCE_GROUP_NAME)) 
@@ -30,6 +35,8 @@ def test_pipeline(self):
         self.check_success_log(captured_output)
         self.check_error_log(captured_output)
         self.check_warning_log(captured_output)
+        self.delete_source(collector_id,self.sumo_source_id)
+        self.delete_collector(collector_id)
 
     def insert_mock_metrics_in_EventHub(self, filename):
         print("Inserting fake metrics in EventHub")

test_utils/basetest.py

@@ -3,6 +3,7 @@
 import json
 import datetime
 import subprocess
+from sumologic import SumoLogic
 from azure.identity import DefaultAzureCredential
 from azure.mgmt.resource.resources.models import Deployment,DeploymentMode
 
@@ -12,6 +13,9 @@ def create_credentials(self):
         self.azure_credential = DefaultAzureCredential()
         self.subscription_id = os.environ["AZURE_SUBSCRIPTION_ID"]
         self.resourcegroup_location = os.environ["AZURE_DEFAULT_REGION"]
+        self.sumo_access_id = os.environ["SUMO_ACCESS_ID"]
+        self.sumo_access_key = os.environ["SUMO_ACCESS_KEY"]
+        self.sumo_deployment = os.environ["SUMO_DEPLOYMENT"]
 
     def resource_group_exists(self, group_name):
         # grp: name,id,properties
@@ -77,3 +81,64 @@ def get_git_info(self):
             branch_name = branch_name.decode()
 
         return repo_name, branch_name
+
+    def api_endpoint(self):
+        if self.sumo_deployment == "us1":
+            return "https://api.sumologic.com/api"
+        elif self.sumo_deployment in ["ca", "au", "de", "eu", "jp", "us2", "fed", "in"]:
+            return "https://api.%s.sumologic.com/api" % self.sumo_deployment
+        else:
+            return 'https://%s-api.sumologic.net/api' % self.sumo_deployment
+        
+    def create_collector(self, collector_name):
+        print("create_collector start")
+        collector_id = None
+        self.sumologic_cli = SumoLogic(self.sumo_access_id, self.sumo_access_key, self.api_endpoint())
+        
+        collector = {
+                    'collector': {
+                        'collectorType': 'Hosted',
+                        'name': collector_name,
+                        'description': "",
+                        'category': None
+                    }
+                }
+        try:
+            resp = self.sumologic_cli.create_collector(collector, headers=None)
+            collector_id = json.loads(resp.text)['collector']['id']
+            print("created collector %s" % collector_id)
+        except Exception as e:
+            raise Exception(e)
+
+        return collector_id
+    
+    def delete_collector(self, collector_id):
+        sources = self.sumologic_cli.sources(collector_id, limit=10)
+        if len(sources) == 0:
+            response = self.sumologic_cli.delete_collector({"collector": {"id": collector_id}})
+            print("deleted collector %s : %s" % (collector_id, response.text))
+    
+    def create_source(self, collector_id, source_name):
+        print("create_source start")
+        endpoint = source_id = None
+        params = {
+            "sourceType": "HTTP",
+            "name": source_name,
+            "messagePerRequest": False,
+            "multilineProcessingEnabled": True,
+            "category": "AZURE/UnitTest/logs"
+        }
+
+        try:
+            resp = self.sumologic_cli.create_source(collector_id, {"source": params})
+            data = resp.json()['source']
+            source_id = data["id"]
+            endpoint = data["url"]
+            print(f"created source '{source_id}' endpoint '{endpoint}'")
+        except Exception as e:
+            raise Exception(e)
+        return source_id, endpoint
+    
+    def delete_source(self, collector_id, source_id):
+        response = self.sumologic_cli.delete_source(collector_id, {"source": {"id": source_id}})
+        print("deleted source %s : %s" % (source_id, response.text))