Merge pull request #2473 from MinerKasch/minerkasch/yarn-consolidate-tags

olivielpeau · olivielpeau · commit 0bda452dfc12 · 2016-05-09T19:08:55.000-04:00
[yarn] Consolidate Tags
diff --git a/checks.d/yarn.py b/checks.d/yarn.py
@@ -52,9 +52,10 @@
 
 '''
 # stdlib
-from urlparse import urljoin
+from urlparse import urljoin, urlsplit, urlunsplit
 
 # 3rd party
+from requests.exceptions import Timeout, HTTPError, InvalidURL, ConnectionError
 import requests
 
 # Project
@@ -63,9 +64,7 @@
 # Default settings
 DEFAULT_RM_URI = 'http://localhost:8088'
 DEFAULT_TIMEOUT = 5
-
-# Path to retrieve cluster info
-INFO_URI = '/ws/v1/cluster/info'
+DEFAULT_CUSTER_NAME = 'default_cluster'
 
 # Path to retrieve cluster metrics
 YARN_CLUSTER_METRICS_PATH = '/ws/v1/cluster/metrics'
@@ -78,10 +77,14 @@
 
 # Metric types
 GAUGE = 'gauge'
+INCREMENT = 'increment'
 
 # Name of the service check
 SERVICE_CHECK_NAME = 'yarn.can_connect'
 
+# Application states to collect
+YARN_APPLICATION_STATES = 'RUNNING'
+
 # Cluster metrics identifier
 YARN_CLUSTER_METRICS_ELEMENT = 'clusterMetrics'
 
@@ -114,15 +117,15 @@
 
 # Application metrics for YARN
 YARN_APP_METRICS = {
-    'progress': ('yarn.apps.progress', GAUGE),
-    'startedTime': ('yarn.apps.started_time', GAUGE),
-    'finishedTime': ('yarn.apps.finished_time', GAUGE),
-    'elapsedTime': ('yarn.apps.elapsed_time', GAUGE),
-    'allocatedMB': ('yarn.apps.allocated_mb', GAUGE),
-    'allocatedVCores': ('yarn.apps.allocated_vcores', GAUGE),
-    'runningContainers': ('yarn.apps.running_containers', GAUGE),
-    'memorySeconds': ('yarn.apps.memory_seconds', GAUGE),
-    'vcoreSeconds': ('yarn.apps.vcore_seconds', GAUGE),
+    'progress': ('yarn.apps.progress', INCREMENT),
+    'startedTime': ('yarn.apps.started_time', INCREMENT),
+    'finishedTime': ('yarn.apps.finished_time', INCREMENT),
+    'elapsedTime': ('yarn.apps.elapsed_time', INCREMENT),
+    'allocatedMB': ('yarn.apps.allocated_mb', INCREMENT),
+    'allocatedVCores': ('yarn.apps.allocated_vcores', INCREMENT),
+    'runningContainers': ('yarn.apps.running_containers', INCREMENT),
+    'memorySeconds': ('yarn.apps.memory_seconds', INCREMENT),
+    'vcoreSeconds': ('yarn.apps.vcore_seconds', INCREMENT),
 }
 
 # Node metrics for YARN
@@ -146,61 +149,61 @@ def check(self, instance):
         # Get properties from conf file
         rm_address = instance.get('resourcemanager_uri', DEFAULT_RM_URI)
 
-        # Get the cluster ID from Yarn
-        cluster_id = self._get_cluster_id(rm_address)
+        # Get additional tags from the conf file
+        tags = instance.get('tags', [])
+        if tags is None:
+            tags = []
+        else:
+            tags = list(set(tags))
 
-        # Get metrics from the Resource Manager
-        self._yarn_cluster_metrics(cluster_id, rm_address)
-        self._yarn_app_metrics(rm_address)
-        self._yarn_node_metrics(cluster_id, rm_address)
+        # Get the cluster name from the conf file
+        cluster_name = instance.get('cluster_name')
+        if cluster_name is None:
+            self.warning("The cluster_name must be specified in the instance configuration, defaulting to '%s'" % (DEFAULT_CUSTER_NAME))
+            cluster_name = DEFAULT_CUSTER_NAME
 
-    def _get_cluster_id(self, rm_address):
-        '''
-        Return the cluster ID, otherwise raise an exception
-        '''
-        info_json = self._rest_request_to_json(rm_address, INFO_URI)
+        tags.append('cluster_name:%s' % cluster_name)
 
-        cluster_id = info_json.get('clusterInfo', {}).get('id')
-        if cluster_id is not None:
-            return cluster_id
-
-        raise Exception('Unable to retrieve cluster ID from ResourceManger')
+        # Get metrics from the Resource Manager
+        self._yarn_cluster_metrics(rm_address, tags)
+        self._yarn_app_metrics(rm_address, tags)
+        self._yarn_node_metrics(rm_address, tags)
 
-    def _yarn_cluster_metrics(self, cluster_id, rm_address):
+    def _yarn_cluster_metrics(self, rm_address, addl_tags):
         '''
         Get metrics related to YARN cluster
         '''
         metrics_json = self._rest_request_to_json(rm_address, YARN_CLUSTER_METRICS_PATH)
 
         if metrics_json:
-            tags = ['cluster_id:' + str(cluster_id)]
 
             yarn_metrics = metrics_json[YARN_CLUSTER_METRICS_ELEMENT]
 
             if yarn_metrics is not None:
-                self._set_yarn_metrics_from_json(tags, yarn_metrics, YARN_CLUSTER_METRICS)
+                self._set_yarn_metrics_from_json(addl_tags, yarn_metrics, YARN_CLUSTER_METRICS)
 
-    def _yarn_app_metrics(self, rm_address):
+    def _yarn_app_metrics(self, rm_address, addl_tags):
         '''
-        Get metrics related to YARN applications
+        Get metrics for running applications
         '''
-        metrics_json = self._rest_request_to_json(rm_address, YARN_APPS_PATH)
+        metrics_json = self._rest_request_to_json(rm_address,
+            YARN_APPS_PATH,
+            states=YARN_APPLICATION_STATES)
 
         if metrics_json:
             if metrics_json['apps'] is not None:
                 if metrics_json['apps']['app'] is not None:
 
                     for app_json in metrics_json['apps']['app']:
 
-                        cluster_id = app_json['clusterId']
-                        app_id = app_json['id']
+                        app_name = app_json['name']
 
-                        tags = ['cluster_id:' + str(cluster_id), 'app_id:' + str(app_id)]
+                        tags = ['app_name:%s' % str(app_name)]
+                        tags.extend(addl_tags)
 
                         self._set_yarn_metrics_from_json(tags, app_json, YARN_APP_METRICS)
 
-
-    def _yarn_node_metrics(self, cluster_id, rm_address):
+    def _yarn_node_metrics(self, rm_address, addl_tags):
         '''
         Get metrics related to YARN nodes
         '''
@@ -210,11 +213,11 @@ def _yarn_node_metrics(self, cluster_id, rm_address):
             if metrics_json['nodes'] is not None:
                 if metrics_json['nodes']['node'] is not None:
 
-                    tags = ['cluster_id:' + str(cluster_id)]
-
                     for node_json in metrics_json['nodes']['node']:
                         node_id = node_json['id']
-                        tags.append('node_id:' + str(node_id))
+
+                        tags = ['node_id:%s' % str(node_id)]
+                        tags.extend(addl_tags)
 
                         self._set_yarn_metrics_from_json(tags, node_json, YARN_NODE_METRICS)
 
@@ -237,34 +240,51 @@ def _set_metric(self, metric_name, metric_type, value, tags=None, device_name=No
         '''
         if metric_type == GAUGE:
             self.gauge(metric_name, value, tags=tags, device_name=device_name)
+        elif metric_type == INCREMENT:
+            self.increment(metric_name, value, tags=tags, device_name=device_name)
         else:
             self.log.error('Metric type "%s" unknown' % (metric_type))
 
-    def _rest_request_to_json(self, address, object_path):
+    def _rest_request_to_json(self, address, object_path, *args, **kwargs):
         '''
         Query the given URL and return the JSON response
         '''
         response_json = None
 
-        service_check_tags = ['instance:%s' % self.hostname]
+        service_check_tags = ['url:%s' % self._get_url_base(address)]
 
-        url = urljoin(address, object_path)
+        url = address
+
+        if object_path:
+            url = self._join_url_dir(url, object_path)
+
+        # Add args to the url
+        if args:
+            for directory in args:
+                url = self._join_url_dir(url, directory)
+
+        self.log.debug('Attempting to connect to "%s"' % url)
+
+        # Add kwargs as arguments
+        if kwargs:
+            query = '&'.join(['{0}={1}'.format(key, value) for key, value in kwargs.iteritems()])
+            url = urljoin(url, '?' + query)
 
         try:
             response = requests.get(url)
             response.raise_for_status()
             response_json = response.json()
 
-        except requests.exceptions.Timeout as e:
+        except Timeout as e:
             self.service_check(SERVICE_CHECK_NAME,
                 AgentCheck.CRITICAL,
                 tags=service_check_tags,
                 message="Request timeout: {0}, {1}".format(url, e))
             raise
 
-        except (requests.exceptions.HTTPError,
-                requests.exceptions.InvalidURL,
-                requests.exceptions.ConnectionError) as e:
+        except (HTTPError,
+                InvalidURL,
+                ConnectionError) as e:
             self.service_check(SERVICE_CHECK_NAME,
                 AgentCheck.CRITICAL,
                 tags=service_check_tags,
@@ -285,3 +305,20 @@ def _rest_request_to_json(self, address, object_path):
                 message='Connection to %s was successful' % url)
 
         return response_json
+
+    def _join_url_dir(self, url, *args):
+        '''
+        Join a URL with multiple directories
+        '''
+        for path in args:
+            url = url.rstrip('/') + '/'
+            url = urljoin(url, path.lstrip('/'))
+
+        return url
+
+    def _get_url_base(self, url):
+        '''
+        Return the base of a URL
+        '''
+        s = urlsplit(url)
+        return urlunsplit([s.scheme, s.netloc, '', '', ''])
diff --git a/conf.d/yarn.yaml.example b/conf.d/yarn.yaml.example
@@ -1,4 +1,23 @@
 init_config:
 
 instances:
-    -   resourcemanager_uri: http://localhost:8088
+  # The YARN check retrieves metrics from YARNS's ResourceManager. This
+  # check must be run from the Master Node and the ResourceManager URI must
+  # be specified below. The ResourceManager URI is composed of the 
+  # ResourceManager's hostname and port.
+  #
+  # The ResourceManager hostname can be found in the yarn-site.xml conf file
+  # under the property yarn.resourcemanager.address
+  #
+  # The ResourceManager port can be found in the yarn-site.xml conf file under
+  # the property yarn.resourcemanager.webapp.address
+  #
+  -  resourcemanager_uri: http://localhost:8088
+
+    # A Required friendly name for the cluster.
+    # cluster_name: MyYarnCluster
+
+    # Optional tags to be applied to every emitted metric.
+    # tags:
+    #   - key:value
+    #   - instance:production
diff --git a/tests/checks/fixtures/yarn/apps_metrics b/tests/checks/fixtures/yarn/apps_metrics
@@ -1 +1,29 @@
-{"apps":{"app":[{"finishedTime":1326815598530,"amContainerLogs":"http://host.domain.com:8042/node/containerlogs/container_1326815542473_0001_01_000001","trackingUI":"History","state":"FINISHED","user":"user1","id":"application_1326815542473_0001","clusterId":1326815542473,"finalStatus":"SUCCEEDED","amHostHttpAddress":"host.domain.com:8042","progress":100,"name":"word count","startedTime":1326815573334,"elapsedTime":25196,"diagnostics":"","trackingUrl":"http://host.domain.com:8088/proxy/application_1326815542473_0001/jobhistory/job/job_1326815542473_1_1","queue":"default","allocatedMB":0,"allocatedVCores":0,"runningContainers":0,"memorySeconds":151730,"vcoreSeconds":103}]}}
+{
+  "apps": {
+    "app": [
+      {
+        "finishedTime": 1326815598530,
+        "amContainerLogs": "http://host.domain.com:8042/node/containerlogs/container_1326815542473_0001_01_000001",
+        "trackingUI": "History",
+        "state": "RUNNING",
+        "user": "user1",
+        "id": "application_1326815542473_0001",
+        "clusterId": 1326815542473,
+        "finalStatus": "SUCCEEDED",
+        "amHostHttpAddress": "host.domain.com:8042",
+        "progress": 100,
+        "name": "word count",
+        "startedTime": 1326815573334,
+        "elapsedTime": 25196,
+        "diagnostics": "",
+        "trackingUrl": "http://host.domain.com:8088/proxy/application_1326815542473_0001/jobhistory/job/job_1326815542473_1_1",
+        "queue": "default",
+        "allocatedMB": 0,
+        "allocatedVCores": 0,
+        "runningContainers": 0,
+        "memorySeconds": 151730,
+        "vcoreSeconds": 103
+      }
+    ]
+  }
+}
diff --git a/tests/checks/fixtures/yarn/cluster_info b/tests/checks/fixtures/yarn/cluster_info
diff --git a/tests/checks/fixtures/yarn/cluster_metrics b/tests/checks/fixtures/yarn/cluster_metrics
@@ -1 +1,27 @@
-{"clusterMetrics":{"appsSubmitted":0,"appsCompleted":0,"appsPending":0,"appsRunning":0,"appsFailed":0,"appsKilled":0,"reservedMB":0,"availableMB":17408,"allocatedMB":0,"reservedVirtualCores":0,"availableVirtualCores":7,"allocatedVirtualCores":1,"containersAllocated":0,"containersReserved":0,"containersPending":0,"totalMB":17408,"totalVirtualCores":8,"totalNodes":1,"lostNodes":0,"unhealthyNodes":0,"decommissionedNodes":0,"rebootedNodes":0,"activeNodes":1}}
+{
+  "clusterMetrics": {
+    "appsSubmitted": 0,
+    "appsCompleted": 0,
+    "appsPending": 0,
+    "appsRunning": 0,
+    "appsFailed": 0,
+    "appsKilled": 0,
+    "reservedMB": 0,
+    "availableMB": 17408,
+    "allocatedMB": 0,
+    "reservedVirtualCores": 0,
+    "availableVirtualCores": 7,
+    "allocatedVirtualCores": 1,
+    "containersAllocated": 0,
+    "containersReserved": 0,
+    "containersPending": 0,
+    "totalMB": 17408,
+    "totalVirtualCores": 8,
+    "totalNodes": 1,
+    "lostNodes": 0,
+    "unhealthyNodes": 0,
+    "decommissionedNodes": 0,
+    "rebootedNodes": 0,
+    "activeNodes": 1
+  }
+}
diff --git a/tests/checks/fixtures/yarn/nodes_metrics b/tests/checks/fixtures/yarn/nodes_metrics
@@ -1 +1,21 @@
-{"nodes":{"node":[{"rack":"\/default-rack","state":"NEW","id":"h2:1235","nodeHostName":"h2","nodeHTTPAddress":"h2:2","healthStatus":"Healthy","lastHealthUpdate":1324056895432,"healthReport":"Healthy","numContainers":0,"usedMemoryMB":0,"availMemoryMB":8192,"usedVirtualCores":0,"availableVirtualCores":8}]}}
+{
+  "nodes": {
+    "node": [
+      {
+        "rack": "/default-rack",
+        "state": "NEW",
+        "id": "h2:1235",
+        "nodeHostName": "h2",
+        "nodeHTTPAddress": "h2:2",
+        "healthStatus": "Healthy",
+        "lastHealthUpdate": 1324056895432,
+        "healthReport": "Healthy",
+        "numContainers": 0,
+        "usedMemoryMB": 0,
+        "availMemoryMB": 8192,
+        "usedVirtualCores": 0,
+        "availableVirtualCores": 8
+      }
+    ]
+  }
+}
diff --git a/tests/checks/mock/test_yarn.py b/tests/checks/mock/test_yarn.py
