/*
 *  Copyright 2014 Damien Claveau
 *
 *  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.
 *
 */


package com.criteo.kafka;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;

import com.yammer.metrics.Metrics;
import com.yammer.metrics.core.MetricPredicate;
import com.yammer.metrics.reporting.GraphiteReporter;

import kafka.metrics.KafkaMetricsConfig;
import kafka.metrics.KafkaMetricsReporter;
import kafka.utils.VerifiableProperties;

public class KafkaGraphiteMetricsReporter implements KafkaMetricsReporter,
	KafkaGraphiteMetricsReporterMBean {

	private static final Logger LOG = Logger.getLogger(KafkaGraphiteMetricsReporter.class);
	private static final String GRAPHITE_DEFAULT_HOST = "localhost";
	private static final int GRAPHITE_DEFAULT_PORT = 2003;
	private static final String GRAPHITE_DEFAULT_PREFIX = "kafka";
	
	private boolean initialized = false;
	private boolean running = false;
	private GraphiteReporter reporter = null;
    private String graphiteHost = GRAPHITE_DEFAULT_HOST;
    private int graphitePort = GRAPHITE_DEFAULT_PORT;
    private String graphiteGroupPrefix = GRAPHITE_DEFAULT_PREFIX;
    private MetricPredicate predicate = MetricPredicate.ALL;

	@Override
	public String getMBeanName() {
		return "kafka:type=com.criteo.kafka.KafkaGraphiteMetricsReporter";
	}

	@Override
	public synchronized void startReporter(long pollingPeriodSecs) {
		if (initialized && !running) {
			reporter.start(pollingPeriodSecs, TimeUnit.SECONDS);
			running = true;
			LOG.info(String.format("Started Kafka Graphite metrics reporter with polling period %d seconds", pollingPeriodSecs));
		}
	}

	@Override
	public synchronized void stopReporter() {
		if (initialized && running) {
			reporter.shutdown();
			running = false;
			LOG.info("Stopped Kafka Graphite metrics reporter");
            try {
            	reporter = new GraphiteReporter(
            			Metrics.defaultRegistry(),
            			graphiteHost,
            			graphitePort,
            			graphiteGroupPrefix/*,
            			predicate*/
            			);
            } catch (IOException e) {
            	LOG.error("Unable to initialize GraphiteReporter", e);
            }
		}
	}

	@Override
	public synchronized void init(VerifiableProperties props) {
		if (!initialized) {
			KafkaMetricsConfig metricsConfig = new KafkaMetricsConfig(props);
            graphiteHost = props.getString("kafka.graphite.metrics.host", GRAPHITE_DEFAULT_HOST);
            graphitePort = props.getInt("kafka.graphite.metrics.port", GRAPHITE_DEFAULT_PORT);
            graphiteGroupPrefix = props.getString("kafka.graphite.metrics.group", GRAPHITE_DEFAULT_PREFIX);
            String regex = props.getString("kafka.graphite.metrics.exclude.regex", null);

            LOG.debug("Initialize GraphiteReporter ["+graphiteHost+","+graphitePort+","+graphiteGroupPrefix+"]");

            if (regex != null) {
            	predicate = new RegexMetricPredicate(regex);
            }
            try {
            	reporter = new GraphiteReporter(
            			Metrics.defaultRegistry(),
            			graphiteHost,
            			graphitePort,
            			graphiteGroupPrefix/*,
            			predicate*/
            			);
            } catch (IOException e) {
            	LOG.error("Unable to initialize GraphiteReporter", e);
            }
            if (props.getBoolean("kafka.graphite.metrics.reporter.enabled", false)) {
            	initialized = true;
            	startReporter(metricsConfig.pollingIntervalSecs());
                LOG.debug("GraphiteReporter started.");
            }
        }
	}
}

@@ -711,18 +711,36 @@
 org.
-apache.log
+slf4j.Logger;%0Aimport org.slf
 4j.L
@@ -744,16 +744,23 @@
 j.Logger
+Factory
 ;%0A%0Aimpor
@@ -1092,18 +1092,17 @@
 eporter,
-%0A%09
+ 
 KafkaGra
@@ -1171,16 +1171,23 @@
 = Logger
+Factory
 .getLogg
@@ -1226,16 +1226,17 @@
 class);%0A
+%0A
 %09private
@@ -2086,22 +2086,8 @@
 nfo(
-String.format(
 %22Sta
@@ -2147,10 +2147,10 @@
 iod 
-%25d
+%7B%7D
  sec
@@ -2174,17 +2174,16 @@
 iodSecs)
-)
 ;%0A%09%09%7D%0A%09%7D
@@ -3352,18 +3352,28 @@
 porter %5B
-%22+
+%7B%7D,%7B%7D,%7B%7D%5D%22, 
 graphite
@@ -3376,21 +3376,18 @@
 hiteHost
-+%22,%22+
+, 
 graphite
@@ -3394,13 +3394,10 @@
 Port
-+%22,%22+
+, 
 grap
@@ -3415,12 +3415,8 @@
 efix
-+%22%5D%22
 );%0A%0A

use slf4j