| # !/usr/bin/python |
| # |
| # Licensed to the Apache Software Foundation (ASF) under one or more |
| # contributor license agreements. See the NOTICE file distributed with |
| # this work for additional information regarding copyright ownership. |
| # The ASF licenses this file to You 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 os |
| import re |
| import time |
| import json |
| import urllib2 |
| import sys |
| import socket |
| import types |
| import httplib |
| import logging |
| import threading |
| |
| # load six |
| sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), '', 'lib/six')) |
| import six |
| |
| # load kafka-python |
| sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), '', 'lib/kafka-python')) |
| from kafka import KafkaClient, SimpleProducer, SimpleConsumer |
| |
| logging.basicConfig(level=logging.INFO, |
| format='%(asctime)s %(name)-12s %(levelname)-6s %(message)s', |
| datefmt='%m-%d %H:%M') |
| |
| |
| class Helper: |
| def __init__(self): |
| pass |
| |
| @staticmethod |
| def load_config(config_file="config.json"): |
| """ |
| |
| :param config_file: |
| :return: |
| """ |
| # read the self-defined filters |
| script_dir = os.path.dirname(__file__) |
| rel_path = "./" + config_file |
| abs_file_path = os.path.join(script_dir, rel_path) |
| if not os.path.isfile(abs_file_path): |
| logging.error(abs_file_path + " doesn't exist, please rename config-sample.json to config.json") |
| exit(1) |
| f = open(abs_file_path, 'r') |
| json_file = f.read() |
| f.close() |
| config = json.loads(json_file) |
| return config |
| |
| @staticmethod |
| def is_number(str): |
| """ |
| |
| :param str: |
| :return: |
| """ |
| try: |
| if str == None or isinstance(str, (bool)): |
| return False |
| float(str) |
| return True |
| except: |
| return False |
| |
| @staticmethod |
| def http_get(host, port, https=False, path=None): |
| """ |
| Read url by GET method |
| |
| :param path: |
| :param url: |
| :param https: |
| :return: |
| """ |
| url = ":".join([host, str(port)]) |
| result = None |
| response = None |
| try: |
| if https: |
| logging.info("Reading https://" + str(url) + path) |
| c = httplib.HTTPSConnection(url, timeout=57) |
| c.request("GET", path) |
| response = c.getresponse() |
| else: |
| logging.info("Reading http://" + str(url) + path) |
| response = urllib2.urlopen("http://" + str(url) + path, timeout=57) |
| logging.debug("Got response") |
| result = response.read() |
| finally: |
| if response is not None: |
| response.close() |
| return result |
| |
| |
| class JmxReader(object): |
| def __init__(self, host, port, https=False): |
| self.host = host |
| self.port = port |
| self.https = https |
| self.jmx_json = None |
| self.jmx_beans = None |
| self.jmx_raw = None |
| |
| def open(self): |
| """ |
| :return: JmxReader |
| """ |
| self.read_raw() |
| self.set_raw(self.jmx_raw) |
| return self |
| |
| def read_raw(self): |
| """ |
| transfer the json string into dict |
| :param host: |
| :param port: |
| :param https: |
| :return: text |
| """ |
| self.jmx_raw = Helper.http_get(self.host, self.port, self.https, "/jmx?anonymous=true") |
| if self.jmx_raw is None: |
| raise Exception("Response from " + url + " is None") |
| return self |
| |
| def set_raw(self, text): |
| self.jmx_json = json.loads(text) |
| self.jmx_beans = self.jmx_json[u'beans'] |
| self.jmx_raw = text |
| return self |
| |
| def get_jmx_beans(self): |
| return self.jmx_beans |
| |
| def get_jmx_bean_by_name(self, name): |
| for bean in self.jmx_beans: |
| if bean.has_key("name") and bean["name"] == name: |
| return bean |
| |
| |
| class YarnWSReader: |
| def __init__(self, host, port, https=False): |
| self.host = host |
| self.port = port |
| self.https = https |
| |
| def read_cluster_info(self): |
| cluster_info = Helper.http_get(self.host, self.port, self.https, "/ws/v1/cluster/info?anonymous=true") |
| logging.debug(cluster_info) |
| return json.loads(cluster_info) |
| |
| |
| class MetricSender(object): |
| def __init__(self, config): |
| pass |
| |
| def open(self): |
| pass |
| |
| def send(self, msg): |
| raise Exception("should be overrode") |
| |
| def close(self): |
| pass |
| |
| |
| class KafkaMetricSender(MetricSender): |
| def __init__(self, config): |
| super(KafkaMetricSender, self).__init__(config) |
| kafka_config = config["output"]["kafka"] |
| # default topic |
| self.topic = kafka_config["topic"].encode('utf-8') |
| # producer |
| self.broker_list = kafka_config["brokerList"] |
| self.kafka_client = None |
| self.kafka_producer = None |
| |
| def open(self): |
| self.kafka_client = KafkaClient(self.broker_list, timeout=59) |
| self.kafka_producer = SimpleProducer(self.kafka_client, batch_send=True, batch_send_every_n=500, |
| batch_send_every_t=30) |
| |
| def send(self, msg): |
| self.kafka_producer.send_messages(self.topic, json.dumps(msg)) |
| |
| def close(self): |
| if self.kafka_producer is not None: |
| self.kafka_producer.stop() |
| if self.kafka_client is not None: |
| self.kafka_client.close() |
| |
| |
| class MetricCollector(threading.Thread): |
| def __init__(self): |
| threading.Thread.__init__(self) |
| self.config = Helper.load_config() |
| self.sender = KafkaMetricSender(self.config) |
| self.fqdn = socket.getfqdn() |
| self.init(self.config) |
| |
| def init(self, config): |
| pass |
| |
| def start(self): |
| try: |
| self.sender.open() |
| self.run() |
| finally: |
| self.sender.close() |
| |
| def collect(self, msg): |
| if not msg.has_key("timestamp"): |
| msg["timestamp"] = int(round(time.time() * 1000)) |
| if msg.has_key("value"): |
| msg["value"] = float(str(msg["value"])) |
| if not msg.has_key("host") or len(msg["host"]) == 0: |
| msg["host"] = self.fqdn |
| if not msg.has_key("site"): |
| msg["site"] = self.config["env"]["site"] |
| |
| self.sender.send(msg) |
| |
| def run(self): |
| raise Exception("`run` method should be overrode by sub-class before being called") |
| |
| |
| class Runner(object): |
| @staticmethod |
| def run(*threads): |
| """ |
| Execute concurrently |
| |
| :param threads: |
| :return: |
| """ |
| for thread in threads: |
| thread.start() |
| |
| class JmxMetricCollector(MetricCollector): |
| selected_domain = None |
| component = None |
| https = False |
| port = None |
| listeners = [] |
| |
| def init(self, config): |
| if config["input"].has_key("host"): |
| self.host = config["input"]["host"] |
| else: |
| self.host = self.fqdn |
| self.port = config["input"]["port"] |
| self.https = config["input"]["https"] |
| self.component = config["input"]["component"] |
| self.selected_domain = [s.encode('utf-8') for s in config[u'filter'].get('monitoring.group.selected')] |
| self.listeners = [] |
| |
| def register(self, *listeners): |
| """ |
| :param listeners: type of HadoopJmxListener |
| :return: |
| """ |
| for listener in listeners: |
| listener.init(self) |
| self.listeners.append(listener) |
| |
| def run(self): |
| try: |
| beans = JmxReader(self.host, self.port, self.https).open().get_jmx_beans() |
| self.on_beans(beans) |
| except Exception as e: |
| logging.exception("Failed to read jmx for " + self.host) |
| |
| def filter_bean(self, bean, mbean_domain): |
| return mbean_domain in self.selected_domain |
| |
| def on_beans(self, beans): |
| for bean in beans: |
| self.on_bean(bean) |
| |
| def on_bean(self, bean): |
| # mbean is of the form "domain:key=value,...,foo=bar" |
| mbean = bean[u'name'] |
| mbean_domain, mbean_attribute = mbean.rstrip().split(":", 1) |
| mbean_domain = mbean_domain.lower() |
| |
| if not self.filter_bean(bean, mbean_domain): |
| return |
| |
| context = bean.get("tag.Context", "") |
| metric_prefix_name = self.__build_metric_prefix(mbean_attribute, context) |
| |
| # print kafka_dict |
| for key, value in bean.iteritems(): |
| self.on_bean_kv(metric_prefix_name, key, value) |
| |
| for listener in self.listeners: |
| listener.on_bean(bean.copy()) |
| |
| def on_bean_kv(self, prefix, key, value): |
| # Skip Tags |
| if re.match(r'tag.*', key): |
| return |
| metric_name = (prefix + '.' + key).lower() |
| self.on_metric({ |
| "metric": metric_name, |
| "value": value |
| }) |
| |
| def on_metric(self, metric): |
| metric["component"] = self.component |
| |
| if Helper.is_number(metric["value"]): |
| self.collect(metric) |
| |
| for listener in self.listeners: |
| listener.on_metric(metric.copy()) |
| |
| def __build_metric_prefix(self, mbean_attribute, context): |
| mbean_list = list(prop.split("=", 1) for prop in mbean_attribute.split(",")) |
| if context == "": |
| metric_prefix_name = '.'.join([i[1] for i in mbean_list]) |
| else: |
| name_index = [i[0] for i in mbean_list].index('name') |
| mbean_list[name_index][1] = context |
| metric_prefix_name = '.'.join([i[1] for i in mbean_list]) |
| return ("hadoop." + metric_prefix_name).replace(" ", "").lower() |
| |
| |
| class JmxMetricListener: |
| def init(self, collector): |
| self.collector = collector |
| |
| def on_bean(self, bean): |
| pass |
| |
| def on_metric(self, metric): |
| pass |
