[mapreduce] An agent that reports MapReduce metrics

conf.d/mapreduce.yaml.example

@@ -0,0 +1,61 @@
+instances:
+  #
+  # The MapReduce check retrieves metrics from YARN'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
+
+init_config:
+  #
+  # Optional metrics can be specified for counters. For more information on
+  # counters visit the MapReduce documentation page:
+  # https://hadoop.apache.org/docs/current/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapredAppMasterRest.html#Job_Counters_API
+  #
+
+  general_counters:
+    #
+    # general_counters are job agnostic metrics that create a metric for each
+    # specified counter
+    #
+    - counter_group_name: 'org.apache.hadoop.mapreduce.TaskCounter'
+      counters:
+        - counter_name: 'MAP_INPUT_RECORDS'
+        - counter_name: 'MAP_OUTPUT_RECORDS'
+        - counter_name: 'REDUCE_INPUT_RECORDS'
+        - counter_name: 'REDUCE_OUTPUT_RECORDS'
+    #
+    # Additional counter's can be specified as following
+    #
+
+    # - counter_group_name: 'org.apache.hadoop.mapreduce.FileSystemCounter'
+    #   counters:
+    #     - counter_name: 'HDFS_BYTES_READ'
+
+  job_specific_counters:
+    #
+    # job_specific_counters are metrics that are specific to a particular job.
+    # The following example specifies counters for the jobs 'Foo' and 'Bar'.
+    #
+
+    # - job_name: 'Foo'
+    #   metrics:
+    #     - counter_group_name: 'org.apache.hadoop.mapreduce.FileSystemCounter'
+    #       counters:
+    #         - counter_name: 'FILE_BYTES_WRITTEN'
+    #         - counter_name: 'HDFS_BYTES_WRITTEN'
+    #     - counter_group_name: 'org.apache.hadoop.mapreduce.FileSystemCounter'
+    #       counters:
+    #         - counter_name: 'HDFS_BYTES_READ'
+    # - job_name: 'Bar'
+    #   metrics:
+    #     - counter_group_name: 'org.apache.hadoop.mapreduce.FileSystemCounter'
+    #       counters:
+    #         - counter_name: 'FILE_BYTES_WRITTEN'

tests/checks/fixtures/mapreduce/apps_metrics

@@ -0,0 +1 @@
+{"apps":{"app":[{"id":"application_1453738555560_0001","trackingUrl":"http://localhost:8088/proxy/application_1453738555560_0001/","user":"vagrant","name":"WordCount"}]}}

tests/checks/fixtures/mapreduce/cluster_info

@@ -0,0 +1 @@
+{"clusterInfo":{"id":1453738555560}}

tests/checks/fixtures/mapreduce/job_counter_metrics

@@ -0,0 +1 @@
+{"jobCounters":{"id":"job_1453738555560_0001","counterGroup":[{"counterGroupName":"org.apache.hadoop.mapreduce.FileSystemCounter","counter":[{"name":"FILE_BYTES_READ","totalCounterValue":0,"mapCounterValue":1,"reduceCounterValue":2},{"name":"FILE_BYTES_WRITTEN","totalCounterValue":3,"mapCounterValue":4,"reduceCounterValue":5}]},{"counterGroupName":"org.apache.hadoop.mapreduce.TaskCounter","counter":[{"name":"MAP_INPUT_RECORDS","totalCounterValue":6,"mapCounterValue":7,"reduceCounterValue":8},{"name":"MAP_OUTPUT_RECORDS","totalCounterValue":9,"mapCounterValue":10,"reduceCounterValue":11}]}]}}

tests/checks/fixtures/mapreduce/job_metrics

@@ -0,0 +1 @@
+{"jobs":{"job":[{"startTime":1453761316277,"finishTime":0,"elapsedTime":99221829,"id":"job_1453738555560_0001","name":"WordCount","user":"vagrant","state":"RUNNING","mapsTotal":1,"mapsCompleted":0,"reducesTotal":1,"reducesCompleted":0,"mapProgress":48.335266,"reduceProgress":0.0,"mapsPending":0,"mapsRunning":1,"reducesPending":1,"reducesRunning":0,"uberized":false,"diagnostics":"","newReduceAttempts":1,"runningReduceAttempts":0,"failedReduceAttempts":0,"killedReduceAttempts":0,"successfulReduceAttempts":0,"newMapAttempts":0,"runningMapAttempts":1,"failedMapAttempts":1,"killedMapAttempts":0,"successfulMapAttempts":0}]}}

tests/checks/fixtures/mapreduce/task_metrics

@@ -0,0 +1 @@
+{"tasks":{"task":[{"startTime":1453761318527,"finishTime":0,"elapsedTime":99869037,"progress":49.11076,"id":"task_1453738555560_0001_m_000000","state":"RUNNING","type":"MAP","successfulAttempt":"","status":"map > map"},{"startTime":1453761318527,"finishTime":0,"elapsedTime":123456,"progress":32.42940,"id":"task_1453738555560_0001_r_000000","state":"RUNNING","type":"REDUCE","successfulAttempt":"","status":"map > map"}]}}

tests/checks/mock/test_mapreduce.py

@@ -0,0 +1,240 @@
+# stdlib
+from urlparse import urljoin
+
+# 3rd party
+import mock
+import json
+
+from tests.checks.common import AgentCheckTest, Fixtures
+
+# ID
+CLUSTER_ID = '1453738555560'
+APP_ID = 'application_1453738555560_0001'
+APP_NAME = 'WordCount'
+JOB_ID = 'job_1453738555560_0001'
+JOB_NAME = 'WordCount'
+USER_NAME = 'vagrant'
+TASK_ID = 'task_1453738555560_0001_m_000000'
+
+# Resource manager URI
+RM_URI = 'http://localhost:8088'
+
+# URL Paths
+INFO_PATH = 'ws/v1/cluster/info'
+YARN_APPS_PATH = 'ws/v1/cluster/apps'
+MAPREDUCE_JOBS_PATH = 'ws/v1/mapreduce/jobs'
+
+# Service Check Names
+YARN_SERVICE_CHECK = 'mapreduce.resource_manager.can_connect'
+MAPREDUCE_SERVICE_CHECK = 'mapreduce.application_master.can_connect'
+
+def join_url_dir(url, *args):
+    '''
+    Join a URL with multiple directories
+    '''
+    for path in args:
+        url = url.rstrip('/') + '/'
+        url = urljoin(url, path.lstrip('/'))
+
+    return url
+
+
+# Service URLs
+CLUSTER_INFO_URL = urljoin(RM_URI, INFO_PATH)
+YARN_APPS_URL = urljoin(RM_URI, YARN_APPS_PATH) + '?states=RUNNING&applicationTypes=MAPREDUCE'
+MR_JOBS_URL = join_url_dir(RM_URI, 'proxy', APP_ID, MAPREDUCE_JOBS_PATH)
+MR_JOB_COUNTERS_URL = join_url_dir(MR_JOBS_URL, JOB_ID, 'counters')
+MR_TASKS_URL = join_url_dir(MR_JOBS_URL, JOB_ID, 'tasks')
+
+
+def requests_get_mock(*args, **kwargs):
+
+    class MockResponse:
+        def __init__(self, json_data, status_code):
+            self.json_data = json_data
+            self.status_code = status_code
+
+        def json(self):
+            return json.loads(self.json_data)
+
+        def raise_for_status(self):
+            return True
+
+
+    if args[0] == CLUSTER_INFO_URL:
+        with open(Fixtures.file('cluster_info'), 'r') as f:
+            body = f.read()
+            return MockResponse(body, 200)
+
+    elif args[0] == YARN_APPS_URL:
+        with open(Fixtures.file('apps_metrics'), 'r') as f:
+            body = f.read()
+            return MockResponse(body, 200)
+
+    elif args[0] == MR_JOBS_URL:
+        with open(Fixtures.file('job_metrics'), 'r') as f:
+            body = f.read()
+            return MockResponse(body, 200)
+
+    elif args[0] == MR_JOB_COUNTERS_URL:
+        with open(Fixtures.file('job_counter_metrics'), 'r') as f:
+            body = f.read()
+            return MockResponse(body, 200)
+
+    elif args[0] == MR_TASKS_URL:
+        with open(Fixtures.file('task_metrics'), 'r') as f:
+            body = f.read()
+            return MockResponse(body, 200)
+
+
+class MapReduceCheck(AgentCheckTest):
+    CHECK_NAME = 'mapreduce'
+
+    MR_CONFIG = {
+        'resourcemanager_uri': 'http://localhost:8088'
+    }
+
+    INIT_CONFIG = {
+        'general_counters': [
+            {
+                'counter_group_name': 'org.apache.hadoop.mapreduce.FileSystemCounter',
+                'counters': [
+                    {'counter_name': 'FILE_BYTES_READ'},
+                    {'counter_name': 'FILE_BYTES_WRITTEN'}
+                ]
+            }
+        ],
+        'job_specific_counters': [
+            {
+                'job_name': 'WordCount',
+                'metrics': [
+                    {
+                        'counter_group_name': 'org.apache.hadoop.mapreduce.FileSystemCounter',
+                        'counters': [
+                            {'counter_name': 'FILE_BYTES_WRITTEN'}
+                        ]
+                    }, {
+                        'counter_group_name': 'org.apache.hadoop.mapreduce.TaskCounter',
+                        'counters': [
+                            {'counter_name': 'MAP_OUTPUT_RECORDS'}
+                        ]
+                    }
+                ]
+            }
+        ]
+    }
+
+    MAPREDUCE_JOB_METRIC_VALUES = {
+        'mapreduce.job.elapsed_time.max': 99221829,
+        'mapreduce.job.maps_total.max': 1,
+        'mapreduce.job.maps_completed.max': 0,
+        'mapreduce.job.reduces_total.max': 1,
+        'mapreduce.job.reduces_completed.max': 0,
+        'mapreduce.job.maps_pending.max': 0,
+        'mapreduce.job.maps_running.max': 1,
+        'mapreduce.job.reduces_pending.max': 1,
+        'mapreduce.job.reduces_running.max': 0,
+        'mapreduce.job.new_reduce_attempts.max': 1,
+        'mapreduce.job.running_reduce_attempts.max': 0,
+        'mapreduce.job.failed_reduce_attempts.max': 0,
+        'mapreduce.job.killed_reduce_attempts.max': 0,
+        'mapreduce.job.successful_reduce_attempts.max': 0,
+        'mapreduce.job.new_map_attempts.max': 0,
+        'mapreduce.job.running_map_attempts.max': 1,
+        'mapreduce.job.failed_map_attempts.max': 1,
+        'mapreduce.job.killed_map_attempts.max': 0,
+        'mapreduce.job.successful_map_attempts.max': 0,
+    }
+
+    MAPREDUCE_JOB_METRIC_TAGS = [
+        'cluster_id:' + CLUSTER_ID,
+        'app_name:' + APP_NAME,
+        'job_name:' + JOB_NAME,
+        'user_name:' + USER_NAME
+    ]
+
+    MAPREDUCE_MAP_TASK_METRIC_VALUES = {
+        'mapreduce.job.map.task.progress.max': 49.11076,
+        'mapreduce.job.map.task.elapsed_time.max': 99869037
+    }
+
+    MAPREDUCE_MAP_TASK_METRIC_TAGS = [
+        'cluster_id:' + CLUSTER_ID,
+        'app_name:' + APP_NAME,
+        'job_name:' + JOB_NAME,
+        'user_name:' + USER_NAME,
+        'task_type:map'
+    ]
+
+    MAPREDUCE_REDUCE_TASK_METRIC_VALUES = {
+        'mapreduce.job.reduce.task.progress.max': 32.42940,
+        'mapreduce.job.reduce.task.elapsed_time.max': 123456
+    }
+
+    MAPREDUCE_REDUCE_TASK_METRIC_TAGS = [
+        'cluster_id:' + CLUSTER_ID,
+        'app_name:' + APP_NAME,
+        'job_name:' + JOB_NAME,
+        'user_name:' + USER_NAME,
+        'task_type:reduce'
+    ]
+
+    MAPREDUCE_JOB_COUNTER_METRIC_VALUES = {
+        'mapreduce.job.counter.total_counter_value.max': {'value': 0, 'tags': ['counter_name:file_bytes_read']},
+        'mapreduce.job.counter.map_counter_value.max': {'value': 1, 'tags': ['counter_name:file_bytes_read']},
+        'mapreduce.job.counter.reduce_counter_value.max': {'value': 2, 'tags': ['counter_name:file_bytes_read']},
+        'mapreduce.job.counter.total_counter_value.max': {'value': 3, 'tags': ['counter_name:file_bytes_written']},
+        'mapreduce.job.counter.map_counter_value.max': {'value': 4, 'tags': ['counter_name:file_bytes_written']},
+        'mapreduce.job.counter.reduce_counter_value.max': {'value': 5, 'tags': ['counter_name:file_bytes_written']},
+        'mapreduce.job.counter.total_counter_value.max': {'value': 9, 'tags': ['counter_name:map_output_records']},
+        'mapreduce.job.counter.map_counter_value.max': {'value': 10, 'tags': ['counter_name:map_output_records']},
+        'mapreduce.job.counter.reduce_counter_value.max': {'value': 11, 'tags': ['counter_name:map_output_records']},
+    }
+
+    MAPREDUCE_JOB_COUNTER_METRIC_TAGS = [
+        'cluster_id:' + CLUSTER_ID,
+        'app_name:' + APP_NAME,
+        'job_name:' + JOB_NAME,
+        'user_name:' + USER_NAME
+    ]
+
+    @mock.patch('requests.get', side_effect=requests_get_mock)
+    def test_check(self, mock_requests):
+        config = {
+            'instances': [self.MR_CONFIG],
+            'init_config': self.INIT_CONFIG
+        }
+
+        self.run_check(config)
+
+        # Check the MapReduce job metrics
+        for metric, value in self.MAPREDUCE_JOB_METRIC_VALUES.iteritems():
+            self.assertMetric(metric,
+                value=value,
+                tags=self.MAPREDUCE_JOB_METRIC_TAGS)
+
+        # Check the map task metrics
+        for metric, value in self.MAPREDUCE_MAP_TASK_METRIC_VALUES.iteritems():
+            self.assertMetric(metric,
+                value=value,
+                tags=self.MAPREDUCE_MAP_TASK_METRIC_TAGS)
+
+        # Check the reduce task metrics
+        for metric, value in self.MAPREDUCE_REDUCE_TASK_METRIC_VALUES.iteritems():
+            self.assertMetric(metric,
+                value=value,
+                tags=self.MAPREDUCE_REDUCE_TASK_METRIC_TAGS)
+
+        # Check the MapReduce job counter metrics
+        for metric, attributes in self.MAPREDUCE_JOB_COUNTER_METRIC_VALUES.iteritems():
+            tags = attributes['tags']
+            tags.extend(self.MAPREDUCE_JOB_COUNTER_METRIC_TAGS)
+            self.assertMetric(metric,
+                value=attributes['value'],
+                tags=tags)
+
+        # Check the service tests
+        self.assertServiceCheckOK(YARN_SERVICE_CHECK,
+            tags=['url:http://localhost:8088'])
+        self.assertServiceCheckOK(MAPREDUCE_SERVICE_CHECK,
+            tags=['url:http://localhost:8088'])