http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-client/src/jvm/org/apache/storm/metrics2/reporters/ConsoleStormReporter.java ---------------------------------------------------------------------- diff --git a/storm-client/src/jvm/org/apache/storm/metrics2/reporters/ConsoleStormReporter.java b/storm-client/src/jvm/org/apache/storm/metrics2/reporters/ConsoleStormReporter.java new file mode 100644 index 0000000..bef9585 --- /dev/null +++ b/storm-client/src/jvm/org/apache/storm/metrics2/reporters/ConsoleStormReporter.java @@ -0,0 +1,69 @@ +/** + * 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 + * <p/> + * http://www.apache.org/licenses/LICENSE-2.0 + * <p/> + * 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 org.apache.storm.metrics2.reporters; + +import com.codahale.metrics.ConsoleReporter; +import com.codahale.metrics.MetricRegistry; +import java.util.Locale; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.storm.daemon.metrics.ClientMetricsUtils; +import org.apache.storm.metrics2.filters.StormMetricsFilter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class ConsoleStormReporter extends ScheduledStormReporter { + private static final Logger LOG = LoggerFactory.getLogger(ConsoleStormReporter.class); + + @Override + public void prepare(MetricRegistry registry, Map stormConf, Map reporterConf) { + LOG.debug("Preparing ConsoleReporter"); + ConsoleReporter.Builder builder = ConsoleReporter.forRegistry(registry); + + builder.outputTo(System.out); + Locale locale = ClientMetricsUtils.getMetricsReporterLocale(stormConf); + if (locale != null) { + builder.formattedFor(locale); + } + + TimeUnit rateUnit = ClientMetricsUtils.getMetricsRateUnit(stormConf); + if (rateUnit != null) { + builder.convertRatesTo(rateUnit); + } + + TimeUnit durationUnit = ClientMetricsUtils.getMetricsDurationUnit(stormConf); + if (durationUnit != null) { + builder.convertDurationsTo(durationUnit); + } + + StormMetricsFilter filter = getMetricsFilter(reporterConf); + if (filter != null) { + builder.filter(filter); + } + + //defaults to 10 + reportingPeriod = getReportPeriod(reporterConf); + + //defaults to seconds + reportingPeriodUnit = getReportPeriodUnit(reporterConf); + + reporter = builder.build(); + } + +} \ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-client/src/jvm/org/apache/storm/metrics2/reporters/CsvStormReporter.java ---------------------------------------------------------------------- diff --git a/storm-client/src/jvm/org/apache/storm/metrics2/reporters/CsvStormReporter.java b/storm-client/src/jvm/org/apache/storm/metrics2/reporters/CsvStormReporter.java new file mode 100644 index 0000000..c52cd2c --- /dev/null +++ b/storm-client/src/jvm/org/apache/storm/metrics2/reporters/CsvStormReporter.java @@ -0,0 +1,97 @@ +/** + * 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 + * <p/> + * http://www.apache.org/licenses/LICENSE-2.0 + * <p/> + * 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 org.apache.storm.metrics2.reporters; + +import com.codahale.metrics.CsvReporter; +import com.codahale.metrics.MetricRegistry; +import java.io.File; +import java.util.Locale; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.storm.daemon.metrics.ClientMetricsUtils; +import org.apache.storm.metrics2.filters.StormMetricsFilter; +import org.apache.storm.utils.ConfigUtils; +import org.apache.storm.utils.ObjectReader; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class CsvStormReporter extends ScheduledStormReporter { + private static final Logger LOG = LoggerFactory.getLogger(CsvStormReporter.class); + + public static final String CSV_LOG_DIR = "csv.log.dir"; + + @Override + public void prepare(MetricRegistry metricsRegistry, Map stormConf, Map reporterConf) { + LOG.debug("Preparing..."); + CsvReporter.Builder builder = CsvReporter.forRegistry(metricsRegistry); + + Locale locale = ClientMetricsUtils.getMetricsReporterLocale(reporterConf); + if (locale != null) { + builder.formatFor(locale); + } + + TimeUnit rateUnit = ClientMetricsUtils.getMetricsRateUnit(reporterConf); + if (rateUnit != null) { + builder.convertRatesTo(rateUnit); + } + + TimeUnit durationUnit = ClientMetricsUtils.getMetricsDurationUnit(reporterConf); + if (durationUnit != null) { + builder.convertDurationsTo(durationUnit); + } + + StormMetricsFilter filter = getMetricsFilter(reporterConf); + if (filter != null) { + builder.filter(filter); + } + + //defaults to 10 + reportingPeriod = getReportPeriod(reporterConf); + + //defaults to seconds + reportingPeriodUnit = getReportPeriodUnit(reporterConf); + + File csvMetricsDir = getCsvLogDir(stormConf, reporterConf); + reporter = builder.build(csvMetricsDir); + } + + + private static File getCsvLogDir(Map stormConf, Map reporterConf) { + String csvMetricsLogDirectory = ObjectReader.getString(reporterConf.get(CSV_LOG_DIR), null); + if (csvMetricsLogDirectory == null) { + csvMetricsLogDirectory = ConfigUtils.absoluteStormLocalDir(stormConf); + csvMetricsLogDirectory = csvMetricsLogDirectory + ConfigUtils.FILE_SEPARATOR + "csvmetrics"; + } + File csvMetricsDir = new File(csvMetricsLogDirectory); + validateCreateOutputDir(csvMetricsDir); + return csvMetricsDir; + } + + private static void validateCreateOutputDir(File dir) { + if (!dir.exists()) { + dir.mkdirs(); + } + if (!dir.canWrite()) { + throw new IllegalStateException(dir.getName() + " does not have write permissions."); + } + if (!dir.isDirectory()) { + throw new IllegalStateException(dir.getName() + " is not a directory."); + } + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-client/src/jvm/org/apache/storm/metrics2/reporters/GangliaStormReporter.java ---------------------------------------------------------------------- diff --git a/storm-client/src/jvm/org/apache/storm/metrics2/reporters/GangliaStormReporter.java b/storm-client/src/jvm/org/apache/storm/metrics2/reporters/GangliaStormReporter.java new file mode 100644 index 0000000..c9f3253 --- /dev/null +++ b/storm-client/src/jvm/org/apache/storm/metrics2/reporters/GangliaStormReporter.java @@ -0,0 +1,132 @@ +/** + * 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 + * <p/> + * http://www.apache.org/licenses/LICENSE-2.0 + * <p/> + * 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 org.apache.storm.metrics2.reporters; + +import com.codahale.metrics.MetricRegistry; +import com.codahale.metrics.ganglia.GangliaReporter; +import info.ganglia.gmetric4j.gmetric.GMetric; +import java.io.IOException; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.storm.daemon.metrics.ClientMetricsUtils; +import org.apache.storm.metrics2.filters.StormMetricsFilter; +import org.apache.storm.utils.ObjectReader; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class GangliaStormReporter extends ScheduledStormReporter { + private static final Logger LOG = LoggerFactory.getLogger(GangliaStormReporter.class); + + public static final String GANGLIA_HOST = "ganglia.host"; + public static final String GANGLIA_PORT = "ganglia.port"; + public static final String GANGLIA_PREFIXED_WITH = "ganglia.prefixed.with"; + public static final String GANGLIA_DMAX = "ganglia.dmax"; + public static final String GANGLIA_TMAX = "ganglia.tmax"; + public static final String GANGLIA_UDP_ADDRESSING_MODE = "ganglia.udp.addressing.mode"; + public static final String GANGLIA_RATE_UNIT = "ganglia.rate.unit"; + public static final String GANGLIA_DURATION_UNIT = "ganglia.duration.unit"; + public static final String GANGLIA_TTL = "ganglia.ttl"; + public static final String GANGLIA_UDP_GROUP = "ganglia.udp.group"; + + @Override + public void prepare(MetricRegistry metricsRegistry, Map stormConf, Map reporterConf) { + LOG.debug("Preparing..."); + GangliaReporter.Builder builder = GangliaReporter.forRegistry(metricsRegistry); + + TimeUnit durationUnit = ClientMetricsUtils.getMetricsDurationUnit(reporterConf); + if (durationUnit != null) { + builder.convertDurationsTo(durationUnit); + } + + TimeUnit rateUnit = ClientMetricsUtils.getMetricsRateUnit(reporterConf); + if (rateUnit != null) { + builder.convertRatesTo(rateUnit); + } + + StormMetricsFilter filter = getMetricsFilter(reporterConf); + if (filter != null) { + builder.filter(filter); + } + String prefix = getMetricsPrefixedWith(reporterConf); + if (prefix != null) { + builder.prefixedWith(prefix); + } + + Integer dmax = getGangliaDMax(reporterConf); + if (prefix != null) { + builder.withDMax(dmax); + } + + Integer tmax = getGangliaTMax(reporterConf); + if (prefix != null) { + builder.withTMax(tmax); + } + + //defaults to 10 + reportingPeriod = getReportPeriod(reporterConf); + + //defaults to seconds + reportingPeriodUnit = getReportPeriodUnit(reporterConf); + + String group = getMetricsTargetUdpGroup(reporterConf); + Integer port = getMetricsTargetPort(reporterConf); + String udpAddressingMode = getMetricsTargetUdpAddressingMode(reporterConf); + Integer ttl = getMetricsTargetTtl(reporterConf); + + GMetric.UDPAddressingMode mode = udpAddressingMode.equalsIgnoreCase("multicast") + ? GMetric.UDPAddressingMode.MULTICAST : GMetric.UDPAddressingMode.UNICAST; + + try { + GMetric sender = new GMetric(group, port, mode, ttl); + reporter = builder.build(sender); + } catch (IOException ioe) { + LOG.error("Exception in GangliaReporter config", ioe); + } + } + + + public static String getMetricsTargetUdpGroup(Map reporterConf) { + return ObjectReader.getString(reporterConf.get(GANGLIA_UDP_GROUP), null); + } + + public static String getMetricsTargetUdpAddressingMode(Map reporterConf) { + return ObjectReader.getString(reporterConf.get(GANGLIA_UDP_ADDRESSING_MODE), null); + } + + public static Integer getMetricsTargetTtl(Map reporterConf) { + return ObjectReader.getInt(reporterConf.get(GANGLIA_TTL), null); + } + + public static Integer getGangliaDMax(Map reporterConf) { + return ObjectReader.getInt(reporterConf.get(GANGLIA_DMAX), null); + } + + public static Integer getGangliaTMax(Map reporterConf) { + return ObjectReader.getInt(reporterConf.get(GANGLIA_TMAX), null); + } + + private static Integer getMetricsTargetPort(Map reporterConf) { + return ObjectReader.getInt(reporterConf.get(GANGLIA_PORT), null); + } + + private static String getMetricsPrefixedWith(Map reporterConf) { + return ObjectReader.getString(reporterConf.get(GANGLIA_PREFIXED_WITH), null); + } + +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-client/src/jvm/org/apache/storm/metrics2/reporters/GraphiteStormReporter.java ---------------------------------------------------------------------- diff --git a/storm-client/src/jvm/org/apache/storm/metrics2/reporters/GraphiteStormReporter.java b/storm-client/src/jvm/org/apache/storm/metrics2/reporters/GraphiteStormReporter.java new file mode 100644 index 0000000..57eb5a4 --- /dev/null +++ b/storm-client/src/jvm/org/apache/storm/metrics2/reporters/GraphiteStormReporter.java @@ -0,0 +1,102 @@ +/** + * 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 + * <p/> + * http://www.apache.org/licenses/LICENSE-2.0 + * <p/> + * 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 org.apache.storm.metrics2.reporters; + +import com.codahale.metrics.MetricRegistry; +import com.codahale.metrics.graphite.Graphite; +import com.codahale.metrics.graphite.GraphiteReporter; +import com.codahale.metrics.graphite.GraphiteSender; +import com.codahale.metrics.graphite.GraphiteUDP; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.storm.daemon.metrics.ClientMetricsUtils; +import org.apache.storm.metrics2.filters.StormMetricsFilter; +import org.apache.storm.utils.ObjectReader; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class GraphiteStormReporter extends ScheduledStormReporter { + private static final Logger LOG = LoggerFactory.getLogger(GraphiteStormReporter.class); + + public static final String GRAPHITE_PREFIXED_WITH = "graphite.prefixed.with"; + public static final String GRAPHITE_HOST = "graphite.host"; + public static final String GRAPHITE_PORT = "graphite.port"; + public static final String GRAPHITE_TRANSPORT = "graphite.transport"; + + @Override + public void prepare(MetricRegistry metricsRegistry, Map stormConf, Map reporterConf) { + LOG.debug("Preparing..."); + GraphiteReporter.Builder builder = GraphiteReporter.forRegistry(metricsRegistry); + + TimeUnit durationUnit = ClientMetricsUtils.getMetricsDurationUnit(reporterConf); + if (durationUnit != null) { + builder.convertDurationsTo(durationUnit); + } + + TimeUnit rateUnit = ClientMetricsUtils.getMetricsRateUnit(reporterConf); + if (rateUnit != null) { + builder.convertRatesTo(rateUnit); + } + + StormMetricsFilter filter = getMetricsFilter(reporterConf); + if (filter != null) { + builder.filter(filter); + } + String prefix = getMetricsPrefixedWith(reporterConf); + if (prefix != null) { + builder.prefixedWith(prefix); + } + + //defaults to 10 + reportingPeriod = getReportPeriod(reporterConf); + + //defaults to seconds + reportingPeriodUnit = getReportPeriodUnit(reporterConf); + + // Not exposed: + // * withClock(Clock) + + String host = getMetricsTargetHost(reporterConf); + Integer port = getMetricsTargetPort(reporterConf); + String transport = getMetricsTargetTransport(reporterConf); + GraphiteSender sender = null; + if (transport.equalsIgnoreCase("udp")) { + sender = new GraphiteUDP(host, port); + } else { + sender = new Graphite(host, port); + } + reporter = builder.build(sender); + } + + private static String getMetricsPrefixedWith(Map reporterConf) { + return ObjectReader.getString(reporterConf.get(GRAPHITE_PREFIXED_WITH), null); + } + + private static String getMetricsTargetHost(Map reporterConf) { + return ObjectReader.getString(reporterConf.get(GRAPHITE_HOST), null); + } + + private static Integer getMetricsTargetPort(Map reporterConf) { + return ObjectReader.getInt(reporterConf.get(GRAPHITE_PORT), null); + } + + private static String getMetricsTargetTransport(Map reporterConf) { + return ObjectReader.getString(reporterConf.get(GRAPHITE_TRANSPORT), "tcp"); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-client/src/jvm/org/apache/storm/metrics2/reporters/JmxStormReporter.java ---------------------------------------------------------------------- diff --git a/storm-client/src/jvm/org/apache/storm/metrics2/reporters/JmxStormReporter.java b/storm-client/src/jvm/org/apache/storm/metrics2/reporters/JmxStormReporter.java new file mode 100644 index 0000000..f995f90 --- /dev/null +++ b/storm-client/src/jvm/org/apache/storm/metrics2/reporters/JmxStormReporter.java @@ -0,0 +1,92 @@ +/** + * 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 + * <p/> + * http://www.apache.org/licenses/LICENSE-2.0 + * <p/> + * 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 org.apache.storm.metrics2.reporters; + +import com.codahale.metrics.JmxReporter; +import com.codahale.metrics.MetricRegistry; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.storm.daemon.metrics.ClientMetricsUtils; +import org.apache.storm.metrics2.filters.StormMetricsFilter; +import org.apache.storm.utils.ObjectReader; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class JmxStormReporter implements StormReporter { + private static final Logger LOG = LoggerFactory.getLogger(JmxStormReporter.class); + public static final String JMX_DOMAIN = "jmx.domain"; + JmxReporter reporter = null; + + @Override + public void prepare(MetricRegistry metricsRegistry, Map<String, Object> stormConf, Map<String, Object> reporterConf) { + LOG.info("Preparing..."); + JmxReporter.Builder builder = JmxReporter.forRegistry(metricsRegistry); + + TimeUnit durationUnit = ClientMetricsUtils.getMetricsDurationUnit(reporterConf); + if (durationUnit != null) { + builder.convertDurationsTo(durationUnit); + } + + TimeUnit rateUnit = ClientMetricsUtils.getMetricsRateUnit(reporterConf); + if (rateUnit != null) { + builder.convertRatesTo(rateUnit); + } + + String domain = getMetricsJmxDomain(reporterConf); + if (domain != null) { + builder.inDomain(domain); + } + + StormMetricsFilter filter = ScheduledStormReporter.getMetricsFilter(reporterConf); + if (filter != null) { + builder.filter(filter); + } + // other builder functions not exposed: + // * createsObjectNamesWith(ObjectNameFactory onFactory) + // * registerWith (MBeanServer) + // * specificDurationUnits (Map<String,TimeUnit> specificDurationUnits) + // * specificRateUnits(Map<String,TimeUnit> specificRateUnits) + + reporter = builder.build(); + } + + public static String getMetricsJmxDomain(Map reporterConf) { + return ObjectReader.getString(reporterConf, JMX_DOMAIN); + } + + @Override + public void start() { + if (reporter != null) { + LOG.debug("Starting..."); + reporter.start(); + } else { + throw new IllegalStateException("Attempt to start without preparing " + getClass().getSimpleName()); + } + } + + @Override + public void stop() { + if (reporter != null) { + LOG.debug("Stopping..."); + reporter.stop(); + } else { + throw new IllegalStateException("Attempt to stop without preparing " + getClass().getSimpleName()); + } + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-client/src/jvm/org/apache/storm/metrics2/reporters/ScheduledStormReporter.java ---------------------------------------------------------------------- diff --git a/storm-client/src/jvm/org/apache/storm/metrics2/reporters/ScheduledStormReporter.java b/storm-client/src/jvm/org/apache/storm/metrics2/reporters/ScheduledStormReporter.java new file mode 100644 index 0000000..61af5be --- /dev/null +++ b/storm-client/src/jvm/org/apache/storm/metrics2/reporters/ScheduledStormReporter.java @@ -0,0 +1,86 @@ +/** + * 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. + */ + +package org.apache.storm.metrics2.reporters; + +import com.codahale.metrics.ScheduledReporter; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.storm.metrics2.filters.StormMetricsFilter; +import org.apache.storm.utils.ObjectReader; +import org.apache.storm.utils.ReflectionUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public abstract class ScheduledStormReporter implements StormReporter { + private static final Logger LOG = LoggerFactory.getLogger(ScheduledStormReporter.class); + protected ScheduledReporter reporter; + protected long reportingPeriod; + protected TimeUnit reportingPeriodUnit; + + @Override + public void start() { + if (reporter != null) { + LOG.debug("Starting..."); + reporter.start(reportingPeriod, reportingPeriodUnit); + } else { + throw new IllegalStateException("Attempt to start without preparing " + getClass().getSimpleName()); + } + } + + @Override + public void stop() { + if (reporter != null) { + LOG.debug("Stopping..."); + reporter.stop(); + } else { + throw new IllegalStateException("Attempt to stop without preparing " + getClass().getSimpleName()); + } + } + + + public static TimeUnit getReportPeriodUnit(Map<String, Object> reporterConf) { + TimeUnit unit = getTimeUnitForConfig(reporterConf, REPORT_PERIOD_UNITS); + return unit == null ? TimeUnit.SECONDS : unit; + } + + private static TimeUnit getTimeUnitForConfig(Map reporterConf, String configName) { + String rateUnitString = ObjectReader.getString(reporterConf.get(configName), null); + if (rateUnitString != null) { + return TimeUnit.valueOf(rateUnitString); + } + return null; + } + + public static long getReportPeriod(Map reporterConf) { + return ObjectReader.getInt(reporterConf.get(REPORT_PERIOD), 10).longValue(); + } + + public static StormMetricsFilter getMetricsFilter(Map reporterConf) { + StormMetricsFilter filter = null; + Map<String, Object> filterConf = (Map)reporterConf.get("filter"); + if (filterConf != null) { + String clazz = (String) filterConf.get("class"); + if (clazz != null) { + filter = ReflectionUtils.newInstance(clazz); + filter.prepare(filterConf); + } + } + return filter; + } +} http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-client/src/jvm/org/apache/storm/metrics2/reporters/StormReporter.java ---------------------------------------------------------------------- diff --git a/storm-client/src/jvm/org/apache/storm/metrics2/reporters/StormReporter.java b/storm-client/src/jvm/org/apache/storm/metrics2/reporters/StormReporter.java new file mode 100644 index 0000000..907965a --- /dev/null +++ b/storm-client/src/jvm/org/apache/storm/metrics2/reporters/StormReporter.java @@ -0,0 +1,35 @@ +/** + * 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. + */ + +package org.apache.storm.metrics2.reporters; + +import com.codahale.metrics.MetricRegistry; +import com.codahale.metrics.Reporter; + +import java.util.Map; + +public interface StormReporter extends Reporter { + String REPORT_PERIOD = "report.period"; + String REPORT_PERIOD_UNITS = "report.period.units"; + + void prepare(MetricRegistry metricsRegistry, Map<String, Object> conf, Map<String, Object> reporterConf); + + void start(); + + void stop(); +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-client/src/jvm/org/apache/storm/stats/BoltExecutorStats.java ---------------------------------------------------------------------- diff --git a/storm-client/src/jvm/org/apache/storm/stats/BoltExecutorStats.java b/storm-client/src/jvm/org/apache/storm/stats/BoltExecutorStats.java index 78246bb..fbf1e92 100644 --- a/storm-client/src/jvm/org/apache/storm/stats/BoltExecutorStats.java +++ b/storm-client/src/jvm/org/apache/storm/stats/BoltExecutorStats.java @@ -15,44 +15,31 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.storm.stats; +import com.codahale.metrics.Counter; import com.google.common.collect.Lists; - +import java.util.List; import org.apache.storm.generated.BoltStats; import org.apache.storm.generated.ExecutorSpecificStats; import org.apache.storm.generated.ExecutorStats; import org.apache.storm.metric.internal.MultiCountStatAndMetric; import org.apache.storm.metric.internal.MultiLatencyStatAndMetric; -import java.util.List; - @SuppressWarnings("unchecked") public class BoltExecutorStats extends CommonStats { - - MultiCountStatAndMetric ackedStats; - MultiCountStatAndMetric failedStats; MultiCountStatAndMetric executedStats; MultiLatencyStatAndMetric processLatencyStats; MultiLatencyStatAndMetric executeLatencyStats; public BoltExecutorStats(int rate,int numStatBuckets) { super(rate,numStatBuckets); - this.ackedStats = new MultiCountStatAndMetric(numStatBuckets); - this.failedStats = new MultiCountStatAndMetric(numStatBuckets); this.executedStats = new MultiCountStatAndMetric(numStatBuckets); this.processLatencyStats = new MultiLatencyStatAndMetric(numStatBuckets); this.executeLatencyStats = new MultiLatencyStatAndMetric(numStatBuckets); } - public MultiCountStatAndMetric getAcked() { - return ackedStats; - } - - public MultiCountStatAndMetric getFailed() { - return failedStats; - } - public MultiCountStatAndMetric getExecuted() { return executedStats; } @@ -67,8 +54,6 @@ public class BoltExecutorStats extends CommonStats { @Override public void cleanupStats() { - ackedStats.close(); - failedStats.close(); executedStats.close(); processLatencyStats.close(); executeLatencyStats.close(); @@ -81,16 +66,17 @@ public class BoltExecutorStats extends CommonStats { this.getExecuteLatencies().record(key, latencyMs); } - public void boltAckedTuple(String component, String stream, long latencyMs) { + public void boltAckedTuple(String component, String stream, long latencyMs, Counter ackedCounter) { List key = Lists.newArrayList(component, stream); this.getAcked().incBy(key, this.rate); + ackedCounter.inc(this.rate); this.getProcessLatencies().record(key, latencyMs); } - public void boltFailedTuple(String component, String stream, long latencyMs) { + public void boltFailedTuple(String component, String stream, long latencyMs, Counter failedCounter) { List key = Lists.newArrayList(component, stream); this.getFailed().incBy(key, this.rate); - + failedCounter.inc(this.rate); } @Override @@ -103,8 +89,8 @@ public class BoltExecutorStats extends CommonStats { // bolt stats BoltStats boltStats = new BoltStats( - StatsUtil.windowSetConverter(valueStat(ackedStats), StatsUtil.TO_GSID, StatsUtil.IDENTITY), - StatsUtil.windowSetConverter(valueStat(failedStats), StatsUtil.TO_GSID, StatsUtil.IDENTITY), + StatsUtil.windowSetConverter(valueStat(getAcked()), StatsUtil.TO_GSID, StatsUtil.IDENTITY), + StatsUtil.windowSetConverter(valueStat(getFailed()), StatsUtil.TO_GSID, StatsUtil.IDENTITY), StatsUtil.windowSetConverter(valueStat(processLatencyStats), StatsUtil.TO_GSID, StatsUtil.IDENTITY), StatsUtil.windowSetConverter(valueStat(executedStats), StatsUtil.TO_GSID, StatsUtil.IDENTITY), StatsUtil.windowSetConverter(valueStat(executeLatencyStats), StatsUtil.TO_GSID, StatsUtil.IDENTITY)); http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-client/src/jvm/org/apache/storm/stats/CommonStats.java ---------------------------------------------------------------------- diff --git a/storm-client/src/jvm/org/apache/storm/stats/CommonStats.java b/storm-client/src/jvm/org/apache/storm/stats/CommonStats.java index b6461c5..4c95da0 100644 --- a/storm-client/src/jvm/org/apache/storm/stats/CommonStats.java +++ b/storm-client/src/jvm/org/apache/storm/stats/CommonStats.java @@ -15,8 +15,10 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.storm.stats; +import com.codahale.metrics.Counter; import java.util.Map; import org.apache.storm.generated.ExecutorStats; import org.apache.storm.metric.internal.MultiCountStatAndMetric; @@ -24,9 +26,10 @@ import org.apache.storm.metric.internal.MultiLatencyStatAndMetric; @SuppressWarnings("unchecked") public abstract class CommonStats { - private final MultiCountStatAndMetric emittedStats; private final MultiCountStatAndMetric transferredStats; + private final MultiCountStatAndMetric ackedStats; + private final MultiCountStatAndMetric failedStats; protected final int rate; @@ -34,6 +37,16 @@ public abstract class CommonStats { this.rate = rate; this.emittedStats = new MultiCountStatAndMetric(numStatBuckets); this.transferredStats = new MultiCountStatAndMetric(numStatBuckets); + this.ackedStats = new MultiCountStatAndMetric(numStatBuckets); + this.failedStats = new MultiCountStatAndMetric(numStatBuckets); + } + + public MultiCountStatAndMetric getFailed() { + return failedStats; + } + + public MultiCountStatAndMetric getAcked() { + return ackedStats; } public int getRate() { @@ -48,17 +61,21 @@ public abstract class CommonStats { return transferredStats; } - public void emittedTuple(String stream) { + public void emittedTuple(String stream, Counter emittedCounter) { this.getEmitted().incBy(stream, this.rate); + emittedCounter.inc(this.rate); } - public void transferredTuples(String stream, int amount) { + public void transferredTuples(String stream, int amount, Counter transferredCounter) { this.getTransferred().incBy(stream, this.rate * amount); + transferredCounter.inc(amount); } public void cleanupStats() { emittedStats.close(); transferredStats.close(); + ackedStats.close(); + failedStats.close(); } protected Map<String,Map<String,Long>> valueStat(MultiCountStatAndMetric metric) { http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-client/src/jvm/org/apache/storm/stats/SpoutExecutorStats.java ---------------------------------------------------------------------- diff --git a/storm-client/src/jvm/org/apache/storm/stats/SpoutExecutorStats.java b/storm-client/src/jvm/org/apache/storm/stats/SpoutExecutorStats.java index 6c3d589..2fc1502 100644 --- a/storm-client/src/jvm/org/apache/storm/stats/SpoutExecutorStats.java +++ b/storm-client/src/jvm/org/apache/storm/stats/SpoutExecutorStats.java @@ -15,55 +15,43 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.storm.stats; +import com.codahale.metrics.Counter; import org.apache.storm.generated.ExecutorSpecificStats; import org.apache.storm.generated.ExecutorStats; import org.apache.storm.generated.SpoutStats; -import org.apache.storm.metric.internal.MultiCountStatAndMetric; import org.apache.storm.metric.internal.MultiLatencyStatAndMetric; @SuppressWarnings("unchecked") public class SpoutExecutorStats extends CommonStats { - - private final MultiCountStatAndMetric ackedStats; - private final MultiCountStatAndMetric failedStats; private final MultiLatencyStatAndMetric completeLatencyStats; public SpoutExecutorStats(int rate,int numStatBuckets) { super(rate,numStatBuckets); - this.ackedStats = new MultiCountStatAndMetric(numStatBuckets); - this.failedStats = new MultiCountStatAndMetric(numStatBuckets); this.completeLatencyStats = new MultiLatencyStatAndMetric(numStatBuckets); } - public MultiCountStatAndMetric getAcked() { - return ackedStats; - } - - public MultiCountStatAndMetric getFailed() { - return failedStats; - } - public MultiLatencyStatAndMetric getCompleteLatencies() { return completeLatencyStats; } @Override public void cleanupStats() { - ackedStats.close(); - failedStats.close(); completeLatencyStats.close(); super.cleanupStats(); } - public void spoutAckedTuple(String stream, long latencyMs) { + public void spoutAckedTuple(String stream, long latencyMs, Counter ackedCounter) { this.getAcked().incBy(stream, this.rate); + ackedCounter.inc(this.rate); this.getCompleteLatencies().record(stream, latencyMs); } - public void spoutFailedTuple(String stream, long latencyMs) { + public void spoutFailedTuple(String stream, long latencyMs, Counter failedCounter) { this.getFailed().incBy(stream, this.rate); + failedCounter.inc(this.rate); } @Override @@ -76,7 +64,7 @@ public class SpoutExecutorStats extends CommonStats { // spout stats SpoutStats spoutStats = new SpoutStats( - valueStat(ackedStats), valueStat(failedStats), valueStat(completeLatencyStats)); + valueStat(getAcked()), valueStat(getFailed()), valueStat(completeLatencyStats)); ret.set_specific(ExecutorSpecificStats.spout(spoutStats)); return ret; http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-client/src/jvm/org/apache/storm/task/TopologyContext.java ---------------------------------------------------------------------- diff --git a/storm-client/src/jvm/org/apache/storm/task/TopologyContext.java b/storm-client/src/jvm/org/apache/storm/task/TopologyContext.java index 0a1765b..5985684 100644 --- a/storm-client/src/jvm/org/apache/storm/task/TopologyContext.java +++ b/storm-client/src/jvm/org/apache/storm/task/TopologyContext.java @@ -15,30 +15,35 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.storm.task; +import com.codahale.metrics.Counter; +import com.codahale.metrics.Gauge; +import com.codahale.metrics.Histogram; +import com.codahale.metrics.Meter; +import com.codahale.metrics.Timer; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.commons.lang.NotImplementedException; import org.apache.storm.generated.GlobalStreamId; import org.apache.storm.generated.Grouping; import org.apache.storm.generated.StormTopology; import org.apache.storm.hooks.ITaskHook; +import org.apache.storm.metric.api.CombinedMetric; +import org.apache.storm.metric.api.ICombiner; import org.apache.storm.metric.api.IMetric; import org.apache.storm.metric.api.IReducer; -import org.apache.storm.metric.api.ICombiner; import org.apache.storm.metric.api.ReducedMetric; -import org.apache.storm.metric.api.CombinedMetric; +import org.apache.storm.metrics2.StormMetricRegistry; import org.apache.storm.state.ISubscribedState; import org.apache.storm.tuple.Fields; import org.apache.storm.utils.Utils; - -import java.util.ArrayList; -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; - -import org.apache.commons.lang.NotImplementedException; import org.json.simple.JSONValue; /** @@ -168,9 +173,9 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo } /** - * Gets the declared output fields for the specified stream id for the - * component this task is a part of. - */ + * Gets the declared output fields for the specified stream id for the + * component this task is a part of. + */ public Fields getThisOutputFields(String streamId) { return getComponentOutputFields(getThisComponentId(), streamId); } @@ -202,8 +207,8 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo public int getThisTaskIndex() { List<Integer> tasks = new ArrayList<>(getComponentTasks(getThisComponentId())); Collections.sort(tasks); - for(int i=0; i<tasks.size(); i++) { - if(tasks.get(i) == getThisTaskId()) { + for (int i=0; i<tasks.size(); i++) { + if (tasks.get(i) == getThisTaskId()) { return i; } } @@ -329,6 +334,7 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo * You must call this during `IBolt.prepare()` or `ISpout.open()`. * @return The IMetric argument unchanged. */ + @Deprecated public <T extends IMetric> T registerMetric(String name, T metric, int timeBucketSizeInSecs) { if(_openOrPrepareWasCalled.get()) { throw new RuntimeException("TopologyContext.registerMetric can only be called from within overridden " + @@ -349,18 +355,18 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo } Map<Integer, Map<Integer, Map<String, IMetric>>> m1 = _registeredMetrics; - if(!m1.containsKey(timeBucketSizeInSecs)) { + if (!m1.containsKey(timeBucketSizeInSecs)) { m1.put(timeBucketSizeInSecs, new HashMap<Integer, Map<String, IMetric>>()); } Map<Integer, Map<String, IMetric>> m2 = m1.get(timeBucketSizeInSecs); - if(!m2.containsKey(_taskId)) { + if (!m2.containsKey(_taskId)) { m2.put(_taskId, new HashMap<String, IMetric>()); } Map<String, IMetric> m3 = m2.get(_taskId); - if(m3.containsKey(name)) { - throw new RuntimeException("The same metric name `" + name + "` was registered twice." ); + if (m3.containsKey(name)) { + throw new RuntimeException("The same metric name `" + name + "` was registered twice."); } else { m3.put(name, metric); } @@ -375,6 +381,7 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo * cause the same metric name can register twice. * So we just return the first metric we meet. */ + @Deprecated public IMetric getRegisteredMetricByName(String name) { IMetric metric = null; @@ -395,13 +402,40 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo /* * Convenience method for registering ReducedMetric. */ + @Deprecated public ReducedMetric registerMetric(String name, IReducer reducer, int timeBucketSizeInSecs) { return registerMetric(name, new ReducedMetric(reducer), timeBucketSizeInSecs); } + /* * Convenience method for registering CombinedMetric. */ + @Deprecated public CombinedMetric registerMetric(String name, ICombiner combiner, int timeBucketSizeInSecs) { return registerMetric(name, new CombinedMetric(combiner), timeBucketSizeInSecs); } + + public Timer registerTimer(String name) { + return StormMetricRegistry.registry().timer(metricName(name)); + } + + public Histogram registerHistogram(String name) { + return StormMetricRegistry.registry().histogram(metricName(name)); + } + + public Meter registerMeter(String name) { + return StormMetricRegistry.registry().meter(metricName(name)); + } + + public Counter registerCounter(String name) { + return StormMetricRegistry.registry().counter(metricName(name)); + } + + public Gauge registerGauge(String name, Gauge gauge) { + return StormMetricRegistry.registry().register(metricName(name), gauge); + } + + private String metricName(String name) { + return StormMetricRegistry.metricName(name, this); + } } http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java ---------------------------------------------------------------------- diff --git a/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java b/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java index 7a64812..3212df0 100644 --- a/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java +++ b/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java @@ -18,27 +18,34 @@ package org.apache.storm.utils; -import org.apache.storm.policy.IWaitStrategy; +import com.google.common.util.concurrent.ThreadFactoryBuilder; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import org.apache.storm.metric.api.IStatefulObject; import org.apache.storm.metric.internal.RateTracker; +import org.apache.storm.metrics2.JcMetrics; +import org.apache.storm.metrics2.StormMetricRegistry; +import org.apache.storm.policy.IWaitStrategy; import org.jctools.queues.MessagePassingQueue; import org.jctools.queues.MpscArrayQueue; import org.jctools.queues.MpscUnboundedArrayQueue; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.Map; -import java.util.concurrent.atomic.AtomicLong; - - public class JCQueue implements IStatefulObject { private static final Logger LOG = LoggerFactory.getLogger(JCQueue.class); + private static final String PREFIX = "jc-"; + private static final ScheduledThreadPoolExecutor METRICS_REPORTER_EXECUTOR = new ScheduledThreadPoolExecutor(1, + new ThreadFactoryBuilder().setDaemon(true).setNameFormat(PREFIX + "metrics-reporter").build()); public static final Object INTERRUPT = new Object(); private final ExitCondition continueRunning = () -> true; + private final JcMetrics jcMetrics; private interface Inserter { // blocking call that can be interrupted using Thread.interrupt() @@ -64,7 +71,7 @@ public class JCQueue implements IStatefulObject { int idleCount = 0; while (!inserted) { q.metrics.notifyInsertFailure(); - if (idleCount==0) { // check avoids multiple log msgs when in a idle loop + if (idleCount == 0) { // check avoids multiple log msgs when in a idle loop LOG.debug("Experiencing Back Pressure on recvQueue: '{}'. Entering BackPressure Wait", q.getName()); } @@ -90,7 +97,6 @@ public class JCQueue implements IStatefulObject { @Override public void flush() throws InterruptedException { - return; } @Override @@ -111,7 +117,7 @@ public class JCQueue implements IStatefulObject { this.currentBatch = new ArrayList<>(batchSz + 1); } - /** Blocking call - retires till element is successfully added */ + /** Blocking call - retires till element is successfully added. */ @Override public void publish(Object obj) throws InterruptedException { currentBatch.add(obj); @@ -143,7 +149,7 @@ public class JCQueue implements IStatefulObject { int retryCount = 0; while (publishCount == 0) { // retry till at least 1 element is drained q.metrics.notifyInsertFailure(); - if (retryCount==0) { // check avoids multiple log msgs when in a idle loop + if (retryCount == 0) { // check avoids multiple log msgs when in a idle loop LOG.debug("Experiencing Back Pressure when flushing batch to Q: {}. Entering BackPressure Wait.", q.getName()); } retryCount = q.backPressureWaitStrategy.idle(retryCount); @@ -236,7 +242,10 @@ public class JCQueue implements IStatefulObject { } private final MpscArrayQueue<Object> recvQueue; - private final MpscUnboundedArrayQueue<Object> overflowQ; // only holds msgs from other workers (via WorkerTransfer), when recvQueue is full + + // only holds msgs from other workers (via WorkerTransfer), when recvQueue is full + private final MpscUnboundedArrayQueue<Object> overflowQ; + private final int overflowLimit; // ensures... overflowCount <= overflowLimit. if set to 0, disables overflow. @@ -250,17 +259,28 @@ public class JCQueue implements IStatefulObject { private String queueName; private final IWaitStrategy backPressureWaitStrategy; - public JCQueue(String queueName, int size, int overflowLimit, int producerBatchSz, IWaitStrategy backPressureWaitStrategy) { + public JCQueue(String queueName, int size, int overflowLimit, int producerBatchSz, IWaitStrategy backPressureWaitStrategy, + String topologyId, String componentId, Integer taskId, int port) { this.queueName = queueName; this.overflowLimit = overflowLimit; this.recvQueue = new MpscArrayQueue<>(size); this.overflowQ = new MpscUnboundedArrayQueue<>(size); this.metrics = new JCQueue.QueueMetrics(); + this.jcMetrics = StormMetricRegistry.jcMetrics(queueName, topologyId, componentId, taskId, port); //The batch size can be no larger than half the full recvQueue size, to avoid contention issues. this.producerBatchSz = Math.max(1, Math.min(producerBatchSz, size / 2)); this.backPressureWaitStrategy = backPressureWaitStrategy; + + if (!METRICS_REPORTER_EXECUTOR.isShutdown()) { + METRICS_REPORTER_EXECUTOR.scheduleAtFixedRate(new Runnable() { + @Override + public void run() { + jcMetrics.set(metrics); + } + }, 15, 15, TimeUnit.SECONDS); + } } public String getName() { @@ -271,6 +291,7 @@ public class JCQueue implements IStatefulObject { public boolean haltWithInterrupt() { boolean res = tryPublishInternal(INTERRUPT); metrics.close(); + METRICS_REPORTER_EXECUTOR.shutdown(); return res; } @@ -294,15 +315,18 @@ public class JCQueue implements IStatefulObject { } } - public int size() { return recvQueue.size() + overflowQ.size(); } + public int size() { + return recvQueue.size() + overflowQ.size(); + } /** * Non blocking. Returns immediately if Q is empty. Returns number of elements consumed from Q + * @param consumer * @param exitCond */ private int consumeImpl(Consumer consumer, ExitCondition exitCond) throws InterruptedException { int drainCount = 0; - while ( exitCond.keepRunning() ) { + while (exitCond.keepRunning()) { Object tuple = recvQueue.poll(); if (tuple == null) { break; @@ -374,7 +398,7 @@ public class JCQueue implements IStatefulObject { } /** - * Non-blocking call, returns false if full + * Non-blocking call, returns false if full. **/ public boolean tryPublish(Object obj) { Inserter inserter = getInserter(); @@ -391,7 +415,7 @@ public class JCQueue implements IStatefulObject { * returns false if overflowLimit has reached */ public boolean tryPublishToOverflow(Object obj) { - if (overflowLimit>0 && overflowQ.size() >= overflowLimit) { + if (overflowLimit > 0 && overflowQ.size() >= overflowLimit) { return false; } overflowQ.add(obj); http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java ---------------------------------------------------------------------- diff --git a/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java b/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java index 34adbc4..cb99870 100644 --- a/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java +++ b/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java @@ -46,9 +46,11 @@ import org.slf4j.LoggerFactory; public class ConfigValidation { private static final Logger LOG = LoggerFactory.getLogger(ConfigValidation.class); - public static abstract class Validator { + public abstract static class Validator { public Validator(Map<String, Object> params) {} + public Validator() {} + public abstract void validateField(String name, Object o); } @@ -57,7 +59,7 @@ public class ConfigValidation { */ /** - * Validates if an object is not null + * Validates if an object is not null. */ public static class NotNullValidator extends Validator { @@ -71,7 +73,7 @@ public class ConfigValidation { } /** - * Validates basic types + * Validates basic types. */ public static class SimpleTypeValidator extends Validator { @@ -93,7 +95,8 @@ public class ConfigValidation { if (type.isInstance(o)) { return; } - throw new IllegalArgumentException("Field " + name + " must be of type " + type + ". Object: " + o + " actual type: " + o.getClass()); + throw new IllegalArgumentException( + "Field " + name + " must be of type " + type + ". Object: " + o + " actual type: " + o.getClass()); } } @@ -137,9 +140,10 @@ public class ConfigValidation { public StringValidator(Map<String, Object> params) { - this.acceptedValues = new HashSet<String>(Arrays.asList((String[])params.get(ConfigValidationAnnotations.ValidatorParams.ACCEPTED_VALUES))); + this.acceptedValues = + new HashSet<String>(Arrays.asList((String[])params.get(ConfigValidationAnnotations.ValidatorParams.ACCEPTED_VALUES))); - if(this.acceptedValues.isEmpty() || (this.acceptedValues.size() == 1 && this.acceptedValues.contains(""))) { + if (this.acceptedValues.isEmpty() || (this.acceptedValues.size() == 1 && this.acceptedValues.contains(""))) { this.acceptedValues = null; } } @@ -147,7 +151,7 @@ public class ConfigValidation { @Override public void validateField(String name, Object o) { SimpleTypeValidator.validateField(name, String.class, o); - if(this.acceptedValues != null) { + if (this.acceptedValues != null) { if (!this.acceptedValues.contains((String) o)) { throw new IllegalArgumentException("Field " + name + " is not an accepted value. Value: " + o + " Accepted values: " + this.acceptedValues); } @@ -194,8 +198,8 @@ public class ConfigValidation { return; } final long i; - if (o instanceof Number && - (i = ((Number) o).longValue()) == ((Number) o).doubleValue()) { + if (o instanceof Number + && (i = ((Number) o).longValue()) == ((Number) o).doubleValue()) { if (i <= Integer.MAX_VALUE && i >= Integer.MIN_VALUE) { return; } @@ -205,7 +209,7 @@ public class ConfigValidation { } /** - * Validates an entry for ImpersonationAclUser + * Validates an entry for ImpersonationAclUser. */ public static class ImpersonationAclUserEntryValidator extends Validator { @@ -214,7 +218,8 @@ public class ConfigValidation { if (o == null) { return; } - ConfigValidationUtils.NestableFieldValidator validator = ConfigValidationUtils.mapFv(ConfigValidationUtils.fv(String.class, false), + ConfigValidationUtils.NestableFieldValidator validator = + ConfigValidationUtils.mapFv(ConfigValidationUtils.fv(String.class, false), ConfigValidationUtils.listFv(String.class, false), false); validator.validateField(name, o); @SuppressWarnings("unchecked") @@ -229,7 +234,7 @@ public class ConfigValidation { } /** - * validates a list of has no duplicates + * validates a list of has no duplicates. */ public static class NoDuplicateInListValidator extends Validator { @@ -251,7 +256,7 @@ public class ConfigValidation { } /** - * Validates a String or a list of Strings + * Validates a String or a list of Strings. */ public static class StringOrStringListValidator extends Validator { @@ -269,7 +274,7 @@ public class ConfigValidation { } /** - * Validates Kryo Registration + * Validates Kryo Registration. */ public static class KryoRegValidator extends Validator { @@ -283,8 +288,8 @@ public class ConfigValidation { for (Object e : (Iterable<?>) o) { if (e instanceof Map) { for (Map.Entry<Object, Object> entry : ((Map<Object, Object>) e).entrySet()) { - if (!(entry.getKey() instanceof String) || - !(entry.getValue() instanceof String)) { + if (!(entry.getKey() instanceof String) + || !(entry.getValue() instanceof String)) { throw new IllegalArgumentException( "Each element of the list " + name + " must be a String or a Map of Strings"); } @@ -302,7 +307,7 @@ public class ConfigValidation { } /** - * Validates if a number is a power of 2 + * Validates if a number is a power of 2. */ public static class PowerOf2Validator extends Validator { @@ -312,8 +317,8 @@ public class ConfigValidation { return; } final long i; - if (o instanceof Number && - (i = ((Number) o).longValue()) == ((Number) o).doubleValue()) { + if (o instanceof Number + && (i = ((Number) o).longValue()) == ((Number) o).doubleValue()) { // Test whether the integer is a power of 2. if (i > 0 && (i & (i - 1)) == 0) { return; @@ -324,7 +329,7 @@ public class ConfigValidation { } /** - * Validates each entry in a list + * Validates each entry in a list. */ public static class ListEntryTypeValidator extends Validator { @@ -346,10 +351,10 @@ public class ConfigValidation { } /** - * Validates each entry in a list against a list of custom Validators + * Validates each entry in a list against a list of custom Validators. * Each validator in the list of validators must inherit or be an instance of Validator class */ - public static class ListEntryCustomValidator extends Validator{ + public static class ListEntryCustomValidator extends Validator { private Class<?>[] entryValidators; @@ -366,7 +371,8 @@ public class ConfigValidation { } } - public static void validateField(String name, Class<?>[] validators, Object o) throws IllegalAccessException, InstantiationException, NoSuchMethodException, InvocationTargetException { + public static void validateField(String name, Class<?>[] validators, Object o) + throws IllegalAccessException, InstantiationException, NoSuchMethodException, InvocationTargetException { if (o == null) { return; } @@ -386,9 +392,9 @@ public class ConfigValidation { } /** - * validates each key and value in a map of a certain type + * validates each key and value in a map of a certain type. */ - public static class MapEntryTypeValidator extends Validator{ + public static class MapEntryTypeValidator extends Validator { private Class<?> keyType; private Class<?> valueType; @@ -410,9 +416,9 @@ public class ConfigValidation { } /** - * validates each key and each value against the respective arrays of validators + * validates each key and each value against the respective arrays of validators. */ - public static class MapEntryCustomValidator extends Validator{ + public static class MapEntryCustomValidator extends Validator { private Class<?>[] keyValidators; private Class<?>[] valueValidators; @@ -432,7 +438,8 @@ public class ConfigValidation { } @SuppressWarnings("unchecked") - public static void validateField(String name, Class<?>[] keyValidators, Class<?>[] valueValidators, Object o) throws IllegalAccessException, InstantiationException, NoSuchMethodException, InvocationTargetException { + public static void validateField(String name, Class<?>[] keyValidators, Class<?>[] valueValidators, Object o) + throws IllegalAccessException, InstantiationException, NoSuchMethodException, InvocationTargetException { if (o == null) { return; } @@ -460,9 +467,9 @@ public class ConfigValidation { } /** - * Validates a positive number + * Validates a positive number. */ - public static class PositiveNumberValidator extends Validator{ + public static class PositiveNumberValidator extends Validator { private boolean includeZero; @@ -484,7 +491,7 @@ public class ConfigValidation { return; } if (o instanceof Number) { - if(includeZero) { + if (includeZero) { if (((Number) o).doubleValue() >= 0.0) { return; } @@ -518,14 +525,14 @@ public class ConfigValidation { @Override public void validateField(String name, Object o) { - if(o == null) { + if (o == null) { return; } SimpleTypeValidator.validateField(name, Map.class, o); - if(!((Map<?, ?>) o).containsKey("class") ) { - throw new IllegalArgumentException( "Field " + name + " must have map entry with key: class"); + if (!((Map<?, ?>) o).containsKey("class")) { + throw new IllegalArgumentException("Field " + name + " must have map entry with key: class"); } - if(!((Map<?, ?>) o).containsKey("parallelism.hint") ) { + if (!((Map<?, ?>) o).containsKey("parallelism.hint")) { throw new IllegalArgumentException("Field " + name + " must have map entry with key: parallelism.hint"); } @@ -534,21 +541,65 @@ public class ConfigValidation { } } + public static class MetricReportersValidator extends Validator { + private static final String NIMBUS = "nimbus"; + private static final String SUPERVISOR = "supervisor"; + private static final String WORKER = "worker"; + private static final String CLASS = "class"; + private static final String FILTER = "filter"; + private static final String DAEMONS = "daemons"; + + @Override + public void validateField(String name, Object o) { + if (o == null) { + return; + } + SimpleTypeValidator.validateField(name, Map.class, o); + if (!((Map) o).containsKey(CLASS)) { + throw new IllegalArgumentException("Field " + name + " must have map entry with key: class"); + } + if (!((Map) o).containsKey(DAEMONS)) { + throw new IllegalArgumentException("Field " + name + " must have map entry with key: daemons"); + } else { + // daemons can only be 'nimbus', 'supervisor', or 'worker' + Object list = ((Map)o).get(DAEMONS); + if (!(list instanceof List)) { + throw new IllegalArgumentException("Field 'daemons' must be a non-null list."); + } + List daemonList = (List)list; + for (Object string : daemonList) { + if (string instanceof String + && (string.equals(NIMBUS) || string.equals(SUPERVISOR) || string.equals(WORKER))) { + continue; + } + throw new IllegalArgumentException("Field 'daemons' must contain at least one of the following:" + + " \"nimbus\", \"supervisor\", or \"worker\""); + } + + } + if (((Map)o).containsKey(FILTER)) { + Map filterMap = (Map)((Map)o).get(FILTER); + SimpleTypeValidator.validateField(CLASS, String.class, filterMap.get(CLASS)); + } + SimpleTypeValidator.validateField(name, String.class, ((Map) o).get(CLASS)); + } + } + public static class EventLoggerRegistryValidator extends Validator { @Override public void validateField(String name, Object o) { - if(o == null) { + if (o == null) { return; } SimpleTypeValidator.validateField(name, Map.class, o); - if(!((Map<?, ?>) o).containsKey("class") ) { - throw new IllegalArgumentException( "Field " + name + " must have map entry with key: class"); + if (!((Map<?, ?>) o).containsKey("class")) { + throw new IllegalArgumentException("Field " + name + " must have map entry with key: class"); } SimpleTypeValidator.validateField(name, String.class, ((Map<?, ?>) o).get("class")); - if(((Map<?, ?>) o).containsKey("arguments") ) { + if (((Map<?, ?>) o).containsKey("arguments")) { SimpleTypeValidator.validateField(name, Map.class, ((Map<?, ?>) o).get("arguments")); } } @@ -570,10 +621,9 @@ public class ConfigValidation { throw new IllegalArgumentException("Field " + name + " must be set."); } - if (o instanceof String && - (((String) o).equals("NONE") || - ((String) o).equals("DIGEST") || - ((String) o).equals("KERBEROS"))) { + if (o instanceof String + && (((String) o).equals("NONE") || ((String) o).equals("DIGEST") + || ((String) o).equals("KERBEROS"))) { return; } throw new IllegalArgumentException("Field " + name + " must be one of \"NONE\", \"DIGEST\", or \"KERBEROS\""); @@ -684,9 +734,9 @@ public class ConfigValidation { private static List<Class<?>> configClasses = null; //We follow the model of service loaders (Even though it is not a service). - private static final String CONFIG_CLASSES_NAME = "META-INF/services/"+Validated.class.getName(); + private static final String CONFIG_CLASSES_NAME = "META-INF/services/" + Validated.class.getName(); - private synchronized static List<Class<?>> getConfigClasses() { + private static synchronized List<Class<?>> getConfigClasses() { if (configClasses == null) { List<Class<?>> ret = new ArrayList<>(); Set<String> classesToScan = new HashSet<>(); @@ -695,7 +745,7 @@ public class ConfigValidation { try { try (BufferedReader in = new BufferedReader(new InputStreamReader(url.openStream()))) { String line; - while((line = in.readLine()) != null) { + while ((line = in.readLine()) != null) { line = line.replaceAll("#.*$", "").trim(); if (!line.isEmpty()) { classesToScan.add(line); @@ -720,7 +770,7 @@ public class ConfigValidation { } /** - * Validates a field given field name as string + * Validates a field given field name as string. * * @param fieldName provided as a string * @param conf map of confs @@ -783,9 +833,7 @@ public class ConfigValidation { //If validator has a constructor that takes a Map as an argument call that constructor if (hasConstructor(clazz, Map.class)) { o = clazz.getConstructor(Map.class).newInstance(params); - } - //If not call default constructor - else { + } else { //If not call default constructor o = clazz.newInstance(); } o.validateField(field.getName(), conf.get(key)); @@ -797,7 +845,7 @@ public class ConfigValidation { } /** - * Validate all confs in map + * Validate all confs in map. * * @param conf map of configs */ @@ -811,15 +859,15 @@ public class ConfigValidation { private static final int ACC_STATIC = 0x0008; private static final int ACC_FINAL = 0x0010; private static final int DESIRED_FIELD_ACC = ACC_PUBLIC | ACC_STATIC | ACC_FINAL; + public static boolean isFieldAllowed(Field field) { - return field.getAnnotation(NotConf.class) == null && - String.class.equals(field.getType()) && - ((field.getModifiers() & DESIRED_FIELD_ACC) == DESIRED_FIELD_ACC) && - !field.isSynthetic(); + return field.getAnnotation(NotConf.class) == null + && String.class.equals(field.getType()) + && ((field.getModifiers() & DESIRED_FIELD_ACC) == DESIRED_FIELD_ACC) && !field.isSynthetic(); } /** - * Validate all confs in map + * Validate all confs in map. * * @param conf map of configs * @param classes config class @@ -848,9 +896,10 @@ public class ConfigValidation { } } - private static Map<String,Object> getParamsFromAnnotation(Class<?> validatorClass, Object v) throws InvocationTargetException, IllegalAccessException { + private static Map<String,Object> getParamsFromAnnotation(Class<?> validatorClass, Object v) + throws InvocationTargetException, IllegalAccessException { Map<String, Object> params = new HashMap<String, Object>(); - for(Method method : validatorClass.getDeclaredMethods()) { + for (Method method : validatorClass.getDeclaredMethods()) { Object value = null; try { @@ -858,7 +907,7 @@ public class ConfigValidation { } catch (IllegalArgumentException ex) { value = null; } - if(value != null) { + if (value != null) { params.put(method.getName(), value); } } http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-client/test/jvm/org/apache/storm/utils/JCQueueBackpressureTest.java ---------------------------------------------------------------------- diff --git a/storm-client/test/jvm/org/apache/storm/utils/JCQueueBackpressureTest.java b/storm-client/test/jvm/org/apache/storm/utils/JCQueueBackpressureTest.java index 29ba179..e48d090 100644 --- a/storm-client/test/jvm/org/apache/storm/utils/JCQueueBackpressureTest.java +++ b/storm-client/test/jvm/org/apache/storm/utils/JCQueueBackpressureTest.java @@ -46,7 +46,7 @@ public class JCQueueBackpressureTest extends TestCase { } private static JCQueue createQueue(String name, int queueSize) { - return new JCQueue(name, queueSize, 0, 1, new WaitStrategyPark(0)); + return new JCQueue(name, queueSize, 0, 1, new WaitStrategyPark(0), "test", "test",1000, 1000); } private static class TestConsumer implements Consumer { http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-client/test/jvm/org/apache/storm/utils/JCQueueTest.java ---------------------------------------------------------------------- diff --git a/storm-client/test/jvm/org/apache/storm/utils/JCQueueTest.java b/storm-client/test/jvm/org/apache/storm/utils/JCQueueTest.java index bdc4937..3d48de2 100644 --- a/storm-client/test/jvm/org/apache/storm/utils/JCQueueTest.java +++ b/storm-client/test/jvm/org/apache/storm/utils/JCQueueTest.java @@ -200,10 +200,10 @@ public class JCQueueTest { } private JCQueue createQueue(String name, int queueSize) { - return new JCQueue(name, queueSize, 0, 1, waitStrategy); + return createQueue(name, 1, queueSize); } private JCQueue createQueue(String name, int batchSize, int queueSize) { - return new JCQueue(name, queueSize, 0, batchSize, waitStrategy); + return new JCQueue(name, queueSize, 0, batchSize, waitStrategy, "test", "test",1000, 1000); } } http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-server/src/main/java/org/apache/storm/DaemonConfig.java ---------------------------------------------------------------------- diff --git a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java index ec1815d..c6cff6b 100644 --- a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java +++ b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java @@ -18,19 +18,22 @@ package org.apache.storm; +import static org.apache.storm.validation.ConfigValidationAnnotations.isBoolean; import static org.apache.storm.validation.ConfigValidationAnnotations.isInteger; +import static org.apache.storm.validation.ConfigValidationAnnotations.isListEntryCustom; import static org.apache.storm.validation.ConfigValidationAnnotations.isPositiveNumber; import static org.apache.storm.validation.ConfigValidationAnnotations.isString; import static org.apache.storm.validation.ConfigValidationAnnotations.isStringList; import static org.apache.storm.validation.ConfigValidationAnnotations.isStringOrStringList; -import static org.apache.storm.validation.ConfigValidationAnnotations.NotNull; -import static org.apache.storm.validation.ConfigValidationAnnotations.isListEntryCustom; -import static org.apache.storm.validation.ConfigValidationAnnotations.isBoolean; import static org.apache.storm.validation.ConfigValidationAnnotations.isNumber; import static org.apache.storm.validation.ConfigValidationAnnotations.isImplementationOfClass; import static org.apache.storm.validation.ConfigValidationAnnotations.isMapEntryType; import static org.apache.storm.validation.ConfigValidationAnnotations.isNoDuplicateInList; import static org.apache.storm.validation.ConfigValidationAnnotations.isMapEntryCustom; +import static org.apache.storm.validation.ConfigValidationAnnotations.NotNull; + +import java.util.ArrayList; +import java.util.Map; import org.apache.storm.container.ResourceIsolationInterface; import org.apache.storm.nimbus.ITopologyActionNotifierPlugin; @@ -42,9 +45,6 @@ import org.apache.storm.security.auth.IAuthorizer; import org.apache.storm.validation.ConfigValidation; import org.apache.storm.validation.Validated; -import java.util.ArrayList; -import java.util.Map; - /** * Storm configs are specified as a plain old map. This class provides constants for * all the configurations possible on a Storm cluster. Each constant is paired with an annotation @@ -69,32 +69,12 @@ public class DaemonConfig implements Validated { public static final String STORM_DAEMON_METRICS_REPORTER_PLUGINS = "storm.daemon.metrics.reporter.plugins"; /** - * A specify Locale for daemon metrics reporter plugin. - * Use the specified IETF BCP 47 language tag string for a Locale. - */ - @isString - public static final String STORM_DAEMON_METRICS_REPORTER_PLUGIN_LOCALE = "storm.daemon.metrics.reporter.plugin.locale"; - - /** * A specify domain for daemon metrics reporter plugin to limit reporting to specific domain. */ @isString public static final String STORM_DAEMON_METRICS_REPORTER_PLUGIN_DOMAIN = "storm.daemon.metrics.reporter.plugin.domain"; /** - * A specify rate-unit in TimeUnit to specify reporting frequency for daemon metrics reporter plugin. - */ - @isString - public static final String STORM_DAEMON_METRICS_REPORTER_PLUGIN_RATE_UNIT = "storm.daemon.metrics.reporter.plugin.rate.unit"; - - /** - * A specify duration-unit in TimeUnit to specify reporting window for daemon metrics reporter plugin. - */ - @isString - public static final String STORM_DAEMON_METRICS_REPORTER_PLUGIN_DURATION_UNIT = "storm.daemon.metrics.reporter.plugin.duration.unit"; - - - /** * A specify csv reporter directory for CvsPreparableReporter daemon metrics reporter. */ @isString @@ -303,7 +283,7 @@ public class DaemonConfig implements Validated { public static final String NIMBUS_ASSIGNMENTS_SERVICE_THREAD_QUEUE_SIZE = "nimbus.assignments.service.thread.queue.size"; /** - * class controls heartbeats recovery strategy + * class controls heartbeats recovery strategy. */ @isString public static final String NIMBUS_WORKER_HEARTBEATS_RECOVERY_STRATEGY_CLASS = "nimbus.worker.heartbeats.recovery.strategy.class"; @@ -829,7 +809,8 @@ public class DaemonConfig implements Validated { */ @NotNull @isPositiveNumber - public static final String STORM_CLUSTER_METRICS_CONSUMER_PUBLISH_INTERVAL_SECS = "storm.cluster.metrics.consumer.publish.interval.secs"; + public static final String STORM_CLUSTER_METRICS_CONSUMER_PUBLISH_INTERVAL_SECS = + "storm.cluster.metrics.consumer.publish.interval.secs"; /** * Enables user-first classpath. See topology.classpath.beginning. @@ -863,8 +844,8 @@ public class DaemonConfig implements Validated { /** * For ArtifactoryConfigLoader, this can either be a reference to an individual file in Artifactory or to a directory. - * If it is a directory, the file with the largest lexographic name will be returned. Users need to add "artifactory+" to the beginning of - * the real URI to use ArtifactoryConfigLoader. + * If it is a directory, the file with the largest lexographic name will be returned. Users need to add "artifactory+" + * to the beginning of the real URI to use ArtifactoryConfigLoader. * For FileConfigLoader, this is the URI pointing to a file. */ @isString http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-server/src/main/java/org/apache/storm/daemon/metrics/MetricsUtils.java ---------------------------------------------------------------------- diff --git a/storm-server/src/main/java/org/apache/storm/daemon/metrics/MetricsUtils.java b/storm-server/src/main/java/org/apache/storm/daemon/metrics/MetricsUtils.java index b5bd8bc..85ee27f 100644 --- a/storm-server/src/main/java/org/apache/storm/daemon/metrics/MetricsUtils.java +++ b/storm-server/src/main/java/org/apache/storm/daemon/metrics/MetricsUtils.java @@ -15,8 +15,14 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.storm.daemon.metrics; +import java.io.File; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + import org.apache.storm.DaemonConfig; import org.apache.storm.daemon.metrics.reporters.JmxPreparableReporter; import org.apache.storm.daemon.metrics.reporters.PreparableReporter; @@ -26,15 +32,8 @@ import org.apache.storm.utils.ReflectionUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.File; -import java.util.ArrayList; -import java.util.List; -import java.util.Locale; -import java.util.Map; -import java.util.concurrent.TimeUnit; - public class MetricsUtils { - private final static Logger LOG = LoggerFactory.getLogger(MetricsUtils.class); + private static final Logger LOG = LoggerFactory.getLogger(ClientMetricsUtils.class); public static List<PreparableReporter> getPreparableReporters(Map<String, Object> topoConf) { List<String> clazzes = (List<String>) topoConf.get(DaemonConfig.STORM_DAEMON_METRICS_REPORTER_PLUGINS); @@ -60,30 +59,6 @@ public class MetricsUtils { return reporter; } - public static Locale getMetricsReporterLocale(Map<String, Object> topoConf) { - String languageTag = ObjectReader.getString(topoConf.get(DaemonConfig.STORM_DAEMON_METRICS_REPORTER_PLUGIN_LOCALE), null); - if (languageTag != null) { - return Locale.forLanguageTag(languageTag); - } - return null; - } - - public static TimeUnit getMetricsRateUnit(Map<String, Object> topoConf) { - return getTimeUnitForCofig(topoConf, DaemonConfig.STORM_DAEMON_METRICS_REPORTER_PLUGIN_RATE_UNIT); - } - - public static TimeUnit getMetricsDurationUnit(Map<String, Object> topoConf) { - return getTimeUnitForCofig(topoConf, DaemonConfig.STORM_DAEMON_METRICS_REPORTER_PLUGIN_DURATION_UNIT); - } - - private static TimeUnit getTimeUnitForCofig(Map<String, Object> topoConf, String configName) { - String rateUnitString = ObjectReader.getString(topoConf.get(configName), null); - if (rateUnitString != null) { - return TimeUnit.valueOf(rateUnitString); - } - return null; - } - public static File getCsvLogDir(Map<String, Object> topoConf) { String csvMetricsLogDirectory = ObjectReader.getString(topoConf.get(DaemonConfig.STORM_DAEMON_METRICS_REPORTER_CSV_LOG_DIR), null); if (csvMetricsLogDirectory == null) { http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/ConsolePreparableReporter.java ---------------------------------------------------------------------- diff --git a/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/ConsolePreparableReporter.java b/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/ConsolePreparableReporter.java index bc34b4b..cd6d069 100644 --- a/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/ConsolePreparableReporter.java +++ b/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/ConsolePreparableReporter.java @@ -15,20 +15,20 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.storm.daemon.metrics.reporters; import com.codahale.metrics.ConsoleReporter; import com.codahale.metrics.MetricRegistry; -import org.apache.storm.daemon.metrics.MetricsUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import java.util.Locale; import java.util.Map; import java.util.concurrent.TimeUnit; +import org.apache.storm.daemon.metrics.ClientMetricsUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public class ConsolePreparableReporter implements PreparableReporter<ConsoleReporter> { - private final static Logger LOG = LoggerFactory.getLogger(ConsolePreparableReporter.class); + private static final Logger LOG = LoggerFactory.getLogger(ConsolePreparableReporter.class); ConsoleReporter reporter = null; @Override @@ -37,17 +37,17 @@ public class ConsolePreparableReporter implements PreparableReporter<ConsoleRepo ConsoleReporter.Builder builder = ConsoleReporter.forRegistry(metricsRegistry); builder.outputTo(System.out); - Locale locale = MetricsUtils.getMetricsReporterLocale(topoConf); + Locale locale = ClientMetricsUtils.getMetricsReporterLocale(topoConf); if (locale != null) { builder.formattedFor(locale); } - TimeUnit rateUnit = MetricsUtils.getMetricsRateUnit(topoConf); + TimeUnit rateUnit = ClientMetricsUtils.getMetricsRateUnit(topoConf); if (rateUnit != null) { builder.convertRatesTo(rateUnit); } - TimeUnit durationUnit = MetricsUtils.getMetricsDurationUnit(topoConf); + TimeUnit durationUnit = ClientMetricsUtils.getMetricsDurationUnit(topoConf); if (durationUnit != null) { builder.convertDurationsTo(durationUnit); } http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/CsvPreparableReporter.java ---------------------------------------------------------------------- diff --git a/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/CsvPreparableReporter.java b/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/CsvPreparableReporter.java index b67cc65..3ab6a99 100644 --- a/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/CsvPreparableReporter.java +++ b/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/CsvPreparableReporter.java @@ -15,21 +15,22 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.storm.daemon.metrics.reporters; import com.codahale.metrics.CsvReporter; import com.codahale.metrics.MetricRegistry; -import org.apache.storm.daemon.metrics.MetricsUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import java.io.File; import java.util.Locale; import java.util.Map; import java.util.concurrent.TimeUnit; +import org.apache.storm.daemon.metrics.ClientMetricsUtils; +import org.apache.storm.daemon.metrics.MetricsUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public class CsvPreparableReporter implements PreparableReporter<CsvReporter> { - private final static Logger LOG = LoggerFactory.getLogger(CsvPreparableReporter.class); + private static final Logger LOG = LoggerFactory.getLogger(CsvPreparableReporter.class); CsvReporter reporter = null; @Override @@ -37,17 +38,17 @@ public class CsvPreparableReporter implements PreparableReporter<CsvReporter> { LOG.debug("Preparing..."); CsvReporter.Builder builder = CsvReporter.forRegistry(metricsRegistry); - Locale locale = MetricsUtils.getMetricsReporterLocale(topoConf); + Locale locale = ClientMetricsUtils.getMetricsReporterLocale(topoConf); if (locale != null) { builder.formatFor(locale); } - TimeUnit rateUnit = MetricsUtils.getMetricsRateUnit(topoConf); + TimeUnit rateUnit = ClientMetricsUtils.getMetricsRateUnit(topoConf); if (rateUnit != null) { builder.convertRatesTo(rateUnit); } - TimeUnit durationUnit = MetricsUtils.getMetricsDurationUnit(topoConf); + TimeUnit durationUnit = ClientMetricsUtils.getMetricsDurationUnit(topoConf); if (durationUnit != null) { builder.convertDurationsTo(durationUnit); } http://git-wip-us.apache.org/repos/asf/storm/blob/21da5dff/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/JmxPreparableReporter.java ---------------------------------------------------------------------- diff --git a/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/JmxPreparableReporter.java b/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/JmxPreparableReporter.java index 21aab16..c369879 100644 --- a/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/JmxPreparableReporter.java +++ b/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/JmxPreparableReporter.java @@ -15,21 +15,21 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.storm.daemon.metrics.reporters; import com.codahale.metrics.JmxReporter; import com.codahale.metrics.MetricRegistry; +import java.util.Map; +import java.util.concurrent.TimeUnit; import org.apache.storm.DaemonConfig; -import org.apache.storm.daemon.metrics.MetricsUtils; +import org.apache.storm.daemon.metrics.ClientMetricsUtils; import org.apache.storm.utils.ObjectReader; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.Map; -import java.util.concurrent.TimeUnit; - public class JmxPreparableReporter implements PreparableReporter<JmxReporter> { - private final static Logger LOG = LoggerFactory.getLogger(JmxPreparableReporter.class); + private static final Logger LOG = LoggerFactory.getLogger(JmxPreparableReporter.class); JmxReporter reporter = null; @Override @@ -40,7 +40,7 @@ public class JmxPreparableReporter implements PreparableReporter<JmxReporter> { if (domain != null) { builder.inDomain(domain); } - TimeUnit rateUnit = MetricsUtils.getMetricsRateUnit(topoConf); + TimeUnit rateUnit = ClientMetricsUtils.getMetricsRateUnit(topoConf); if (rateUnit != null) { builder.convertRatesTo(rateUnit); }
