[rhq] Branch 'feature/hadoop-plugin' - 2 commits - modules/plugins
lkrejci
lkrejci at fedoraproject.org
Mon Aug 6 08:53:18 UTC 2012
modules/plugins/hadoop/pom.xml | 4
modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/HadoopServerComponent.java | 83 ++-
modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/HadoopServerConfigurationDelegate.java | 2
modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/JobTrackerServerComponent.java | 246 ++++++++++
modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/calltime/HadoopEventAndCalltimeDelegate.java | 79 +++
modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/calltime/JobSummary.java | 177 +++++++
modules/plugins/hadoop/src/main/resources/META-INF/rhq-plugin.xml | 8
7 files changed, 565 insertions(+), 34 deletions(-)
New commits:
commit 70163595c63322617af2c74cde592a5c3bb40c1a
Author: Lukas Krejci <lkrejci at redhat.com>
Date: Mon Aug 6 10:52:34 2012 +0200
Implementing Hadoop job duration as a calltime metric.
diff --git a/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/HadoopServerComponent.java b/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/HadoopServerComponent.java
index a246b13..8b4a73b 100644
--- a/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/HadoopServerComponent.java
+++ b/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/HadoopServerComponent.java
@@ -33,6 +33,7 @@ import org.mc4j.ems.connection.bean.attribute.EmsAttribute;
import org.rhq.core.domain.configuration.Configuration;
import org.rhq.core.domain.configuration.ConfigurationUpdateStatus;
import org.rhq.core.domain.measurement.AvailabilityType;
+import org.rhq.core.domain.measurement.DataType;
import org.rhq.core.domain.measurement.MeasurementDataNumeric;
import org.rhq.core.domain.measurement.MeasurementDataTrait;
import org.rhq.core.domain.measurement.MeasurementDefinition;
@@ -42,6 +43,7 @@ import org.rhq.core.domain.measurement.MeasurementUnits;
import org.rhq.core.pluginapi.configuration.ConfigurationFacet;
import org.rhq.core.pluginapi.configuration.ConfigurationUpdateReport;
import org.rhq.core.pluginapi.event.EventContext;
+import org.rhq.core.pluginapi.event.EventPoller;
import org.rhq.core.pluginapi.event.log.Log4JLogEntryProcessor;
import org.rhq.core.pluginapi.event.log.LogFileEventPoller;
import org.rhq.core.pluginapi.inventory.ResourceComponent;
@@ -50,6 +52,8 @@ import org.rhq.core.pluginapi.measurement.MeasurementFacet;
import org.rhq.core.pluginapi.operation.OperationFacet;
import org.rhq.core.pluginapi.operation.OperationResult;
import org.rhq.core.system.ProcessInfo;
+import org.rhq.plugins.hadoop.calltime.HadoopEventAndCalltimeDelegate;
+import org.rhq.plugins.hadoop.calltime.JobSummary;
import org.rhq.plugins.jmx.JMXComponent;
import org.rhq.plugins.jmx.JMXServerComponent;
@@ -57,8 +61,9 @@ public class HadoopServerComponent extends JMXServerComponent<ResourceComponent<
JMXComponent<ResourceComponent<?>>, MeasurementFacet, OperationFacet, ConfigurationFacet {
private static final Log LOG = LogFactory.getLog(HadoopServerComponent.class);
- private static final String LOG_EVENT_TYPE = "logEntry";
- private static final String LOG_POLLING_INTERVAL_PROPERTY = "logPollingInterval";
+
+ public static final String LOG_EVENT_TYPE = "logEntry";
+ public static final String LOG_POLLING_INTERVAL_PROPERTY = "logPollingInterval";
private Map<String, Boolean> percentageMeasurements;
@@ -89,6 +94,7 @@ public class HadoopServerComponent extends JMXServerComponent<ResourceComponent<
if (events != null && eventsRegistered) {
eventsRegistered = false;
events.unregisterEventPoller(LOG_EVENT_TYPE);
+ discardPoller();
}
super.stop();
}
@@ -109,12 +115,13 @@ public class HadoopServerComponent extends JMXServerComponent<ResourceComponent<
if (!eventsRegistered) {
File logFile = determineLogFile();
int interval = Integer.parseInt(getResourceContext().getPluginConfiguration().getSimpleValue(LOG_POLLING_INTERVAL_PROPERTY, "60"));
- events.registerEventPoller(new LogFileEventPoller(events, LOG_EVENT_TYPE, logFile, new Log4JLogEntryProcessor(LOG_EVENT_TYPE, logFile)), interval);
+ events.registerEventPoller(createNewEventPoller(events, logFile), interval);
eventsRegistered = true;
}
} else if (eventsRegistered) {
eventsRegistered = false;
events.unregisterEventPoller(LOG_EVENT_TYPE);
+ discardPoller();
}
}
@@ -135,47 +142,54 @@ public class HadoopServerComponent extends JMXServerComponent<ResourceComponent<
* Gather measurement data
* @see org.rhq.core.pluginapi.measurement.MeasurementFacet#getValues(org.rhq.core.domain.measurement.MeasurementReport, java.util.Set)
*/
+ @Override
public void getValues(MeasurementReport report, Set<MeasurementScheduleRequest> metrics) throws Exception {
for (MeasurementScheduleRequest request : metrics) {
- String name = request.getName();
- int delimIndex = name.lastIndexOf(':');
- String beanName = name.substring(0, delimIndex);
- String attributeName = name.substring(delimIndex + 1);
- try {
- EmsConnection emsConnection = getEmsConnection();
- EmsBean bean = emsConnection.getBean(beanName);
- if (bean != null) {
- bean.refreshAttributes();
- EmsAttribute attribute = bean.getAttribute(attributeName);
- if (attribute != null) {
- Object valueObject = attribute.refresh();
- if (valueObject instanceof Number) {
- Number value = (Number) valueObject;
- if (percentageMeasurements.get(name)) {
- report.addData(new MeasurementDataNumeric(request, value.doubleValue() / 100));
- } else {
- report.addData(new MeasurementDataNumeric(request, value.doubleValue()));
- }
-
+ handleMetric(report, request);
+ }
+ }
+
+ protected void handleMetric(MeasurementReport report, MeasurementScheduleRequest request) throws Exception {
+ String name = request.getName();
+ int delimIndex = name.lastIndexOf(':');
+ String beanName = name.substring(0, delimIndex);
+ String attributeName = name.substring(delimIndex + 1);
+ try {
+ EmsConnection emsConnection = getEmsConnection();
+ EmsBean bean = emsConnection.getBean(beanName);
+ if (bean != null) {
+ bean.refreshAttributes();
+ EmsAttribute attribute = bean.getAttribute(attributeName);
+ if (attribute != null) {
+ Object valueObject = attribute.refresh();
+ if (valueObject instanceof Number) {
+ Number value = (Number) valueObject;
+ if (percentageMeasurements.get(name)) {
+ report.addData(new MeasurementDataNumeric(request, value.doubleValue() / 100));
} else {
- report.addData(new MeasurementDataTrait(request, valueObject.toString()));
+ report.addData(new MeasurementDataNumeric(request, value.doubleValue()));
}
+
} else {
- LOG.warn("Attribute " + attributeName + " not found");
+ report.addData(new MeasurementDataTrait(request, valueObject.toString()));
}
} else {
- LOG.warn("MBean " + beanName + " not found");
+ LOG.warn("Attribute " + attributeName + " not found");
}
- } catch (Exception e) {
- LOG.error("Failed to obtain measurement [" + name + "]", e);
+ } else {
+ LOG.warn("MBean " + beanName + " not found");
}
+ } catch (Exception e) {
+ LOG.error("Failed to obtain measurement [" + name + "]", e);
}
}
-
+
+ @Override
public Configuration loadResourceConfiguration() throws Exception {
return configurationDelegate.loadConfiguration();
}
+ @Override
public void updateResourceConfiguration(ConfigurationUpdateReport report) {
try {
Configuration updatedConfiguration = report.getConfiguration();
@@ -194,6 +208,7 @@ public class HadoopServerComponent extends JMXServerComponent<ResourceComponent<
* @return An operation result
* @see org.rhq.core.pluginapi.operation.OperationFacet
*/
+ @Override
public OperationResult invokeOperation(String name, Configuration params) throws Exception {
HadoopSupportedOperations operation = HadoopSupportedOperations.valueOf(name.toUpperCase());
String serverType = getServerType();
@@ -204,6 +219,14 @@ public class HadoopServerComponent extends JMXServerComponent<ResourceComponent<
return result;
}
+ protected EventPoller createNewEventPoller(EventContext eventContext, File logFile) {
+ return new LogFileEventPoller(eventContext, LOG_EVENT_TYPE, logFile, new Log4JLogEntryProcessor(LOG_EVENT_TYPE, logFile));
+ }
+
+ protected void discardPoller() {
+
+ }
+
private File determineLogFile() {
String username = getResourceContext().getNativeProcess().getCredentialsName().getUser();
String hostname = getResourceContext().getSystemInformation().getHostname();
@@ -243,5 +266,5 @@ public class HadoopServerComponent extends JMXServerComponent<ResourceComponent<
}
return homeDir;
- }
+ }
}
diff --git a/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/HadoopServerConfigurationDelegate.java b/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/HadoopServerConfigurationDelegate.java
index 4a42b3d..b471823 100644
--- a/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/HadoopServerConfigurationDelegate.java
+++ b/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/HadoopServerConfigurationDelegate.java
@@ -216,7 +216,7 @@ public class HadoopServerConfigurationDelegate {
}
}
- private static void parseAndAssignProps(File configFile, Map<String, PropertySimple> props)
+ public static void parseAndAssignProps(File configFile, Map<String, PropertySimple> props)
throws XMLStreamException, IOException {
FileInputStream in = new FileInputStream(configFile);
XMLStreamReader rdr = XML_INPUT_FACTORY.createXMLStreamReader(in);
diff --git a/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/JobTrackerServerComponent.java b/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/JobTrackerServerComponent.java
new file mode 100644
index 0000000..539f8f6
--- /dev/null
+++ b/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/JobTrackerServerComponent.java
@@ -0,0 +1,246 @@
+/*
+ * RHQ Management Platform
+ * Copyright (C) 2005-2012 Red Hat, Inc.
+ * All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+package org.rhq.plugins.hadoop;
+
+import java.io.File;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.rhq.core.domain.measurement.DataType;
+import org.rhq.core.domain.measurement.MeasurementDefinition;
+import org.rhq.core.domain.measurement.MeasurementReport;
+import org.rhq.core.domain.measurement.MeasurementScheduleRequest;
+import org.rhq.core.domain.measurement.calltime.CallTimeData;
+import org.rhq.core.pluginapi.event.EventContext;
+import org.rhq.core.pluginapi.event.EventPoller;
+import org.rhq.core.pluginapi.event.log.LogFileEventPoller;
+import org.rhq.core.pluginapi.inventory.InvalidPluginConfigurationException;
+import org.rhq.core.pluginapi.inventory.ResourceContext;
+import org.rhq.plugins.hadoop.calltime.HadoopEventAndCalltimeDelegate;
+import org.rhq.plugins.hadoop.calltime.JobSummary;
+
+/**
+ * Resource component for Hadoop JobTracker.
+ *
+ * @author Lukas Krejci
+ */
+public class JobTrackerServerComponent extends HadoopServerComponent {
+
+ private static final String SYNTHETIC_METRICS_PREFIX = "_synthetic_";
+ private static final String JOB_DURATION_METRIC_NAME = "_synthetic_jobDuration";
+ private static final String JOB_PRE_START_DELAY_METRIC_NAME = "_synthetic_jobPreStartDelay";
+ private static final String JOB_SUCCESS_RATE_METRIC_NAME = "_synthetic_jobSuccessRate";
+
+ private Map<String, Map<String, Set<JobSummary>>> unprocessedCalltimeMeasurements = new HashMap<String, Map<String,Set<JobSummary>>>();
+
+ private HadoopEventAndCalltimeDelegate logProcessor;
+
+ @Override
+ @SuppressWarnings({ "rawtypes" })
+ public void start(ResourceContext context) throws InvalidPluginConfigurationException, Exception {
+ super.start(context);
+
+ Set<MeasurementDefinition> measDefinitions = context.getResourceType().getMetricDefinitions();
+ for (MeasurementDefinition measDefinition : measDefinitions) {
+ if (measDefinition.getDataType() == DataType.CALLTIME) {
+ unprocessedCalltimeMeasurements.put(measDefinition.getName(), new HashMap<String, Set<JobSummary>>());
+ }
+ }
+
+ //TODO restore unprocessed calltime data from persistent storage in data dir.
+ }
+
+ @Override
+ public void stop() {
+ //TODO save unprocessed calltime data to persistent storage in data dir.
+ super.stop();
+ }
+
+ @Override
+ protected void handleMetric(MeasurementReport report, MeasurementScheduleRequest request) throws Exception {
+ if (request.getName().startsWith(SYNTHETIC_METRICS_PREFIX)) {
+ if (logProcessor != null) {
+ updateUnprocessedMeasurements();
+ Map<String, Set<JobSummary>> pendingJobs = unprocessedCalltimeMeasurements.get(request.getName());
+ report.addData(createCalltimeData(request, pendingJobs));
+ }
+ } else {
+ super.handleMetric(report, request);
+ }
+ }
+
+ @Override
+ protected EventPoller createNewEventPoller(EventContext eventContext, File logFile) {
+ logProcessor = new HadoopEventAndCalltimeDelegate(LOG_EVENT_TYPE, logFile);
+ return new LogFileEventPoller(eventContext, LOG_EVENT_TYPE, logFile, logProcessor);
+ }
+
+ @Override
+ protected void discardPoller() {
+ logProcessor = null;
+ }
+
+ private void updateUnprocessedMeasurements() {
+ if (logProcessor == null) {
+ return;
+ }
+
+ Set<JobSummary> newJobs = logProcessor.drainAccumulatedJobs();
+
+ //the job summaries are the base for all (currently both) types of calltime metrics
+ //we therefore add the newJobs to all members of the unprocessed map..
+ for (Map.Entry<String, Map<String, Set<JobSummary>>> e : unprocessedCalltimeMeasurements.entrySet()) {
+ Map<String, Set<JobSummary>> jobsByName = e.getValue();
+
+ for (JobSummary newJob : newJobs) {
+ Set<JobSummary> unprocessed = jobsByName.get(newJob.getJobName());
+ if (unprocessed == null) {
+ unprocessed = new HashSet<JobSummary>();
+ jobsByName.put(newJob.getJobName(), unprocessed);
+ }
+ unprocessed.add(newJob);
+ }
+ }
+ }
+
+ private CallTimeData
+ createCalltimeData(MeasurementScheduleRequest request, Map<String, Set<JobSummary>> pendingJobs) {
+ CallTimeData ret = new CallTimeData(request);
+
+ String metricName = request.getName();
+
+ for (Map.Entry<String, Set<JobSummary>> e : pendingJobs.entrySet()) {
+ String jobName = e.getKey();
+ Set<JobSummary> jobs = e.getValue();
+
+ if (JOB_DURATION_METRIC_NAME.equals(metricName)) {
+ initJobDurationData(ret, jobName, jobs);
+ } else if (JOB_PRE_START_DELAY_METRIC_NAME.equals(metricName)) {
+ initPreStartDelayData(ret, jobName, jobs);
+ } else if (JOB_SUCCESS_RATE_METRIC_NAME.equals(metricName)) {
+ initSuccessRateData(ret, jobName, jobs);
+ }
+ }
+
+ return ret;
+ }
+
+ private void initJobDurationData(CallTimeData data, String jobName, Set<JobSummary> pendingJobs) {
+ long beginTime = Long.MAX_VALUE;
+ long endTime = 0;
+ long totalTime = 0;
+ long minTime = Long.MAX_VALUE;
+ long maxTime = 0;
+ long count = pendingJobs.size();
+
+ for (JobSummary job : pendingJobs) {
+ if (job.getStartTime() < beginTime) {
+ beginTime = job.getStartTime();
+ }
+
+ if (job.getEndTime() > endTime) {
+ endTime = job.getEndTime();
+ }
+
+ long duration = job.getEndTime() - job.getStartTime();
+
+ totalTime += duration;
+
+ if (duration < minTime) {
+ minTime = duration;
+ }
+
+ if (duration > maxTime) {
+ maxTime = duration;
+ }
+ }
+
+ data.addAggregatedCallData(jobName, new Date(beginTime), new Date(endTime), minTime, maxTime, totalTime, count);
+ }
+
+ private void initPreStartDelayData(CallTimeData data, String jobName, Set<JobSummary> pendingJobs) {
+ long beginTime = Long.MAX_VALUE;
+ long endTime = 0;
+ long totalTime = 0;
+ long minTime = Long.MAX_VALUE;
+ long maxTime = 0;
+ long count = pendingJobs.size();
+
+ for (JobSummary job : pendingJobs) {
+ if (job.getSubmitTime() < beginTime) {
+ beginTime = job.getSubmitTime();
+ }
+
+ if (job.getStartTime() > endTime) {
+ endTime = job.getStartTime();
+ }
+
+ long duration = job.getStartTime() - job.getSubmitTime();
+
+ totalTime += duration;
+
+ if (duration < minTime) {
+ minTime = duration;
+ }
+
+ if (duration > maxTime) {
+ maxTime = duration;
+ }
+ }
+
+ data.addAggregatedCallData(jobName, new Date(beginTime), new Date(endTime), minTime, maxTime, totalTime, count);
+ }
+
+ private void initSuccessRateData(CallTimeData data, String jobName, Set<JobSummary> pendingJobs) {
+ long beginTime = Long.MAX_VALUE;
+ long endTime = 0;
+ long totalTime = 0;
+ long minTime = Long.MAX_VALUE;
+ long maxTime = 0;
+ long count = pendingJobs.size();
+
+ for (JobSummary job : pendingJobs) {
+ if (job.getStartTime() < beginTime) {
+ beginTime = job.getStartTime();
+ }
+
+ if (job.getEndTime() > endTime) {
+ endTime = job.getEndTime();
+ }
+
+ long duration = job.isSucceeded() ? 1 : 0;
+
+ totalTime += duration;
+
+ if (duration < minTime) {
+ minTime = duration;
+ }
+
+ if (duration > maxTime) {
+ maxTime = duration;
+ }
+ }
+
+ data.addAggregatedCallData(jobName, new Date(beginTime), new Date(endTime), minTime, maxTime, totalTime, count);
+ }
+}
diff --git a/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/calltime/HadoopEventAndCalltimeDelegate.java b/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/calltime/HadoopEventAndCalltimeDelegate.java
new file mode 100644
index 0000000..35bde94
--- /dev/null
+++ b/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/calltime/HadoopEventAndCalltimeDelegate.java
@@ -0,0 +1,79 @@
+/*
+ * RHQ Management Platform
+ * Copyright (C) 2005-2012 Red Hat, Inc.
+ * All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+package org.rhq.plugins.hadoop.calltime;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.regex.Matcher;
+
+import org.rhq.core.domain.event.Event;
+import org.rhq.core.pluginapi.event.log.Log4JLogEntryProcessor;
+
+/**
+ * This is an extension of the log entry processor that can also gather data for calltime metrics
+ * we gather on the hadoop jobs.
+ *
+ * @author Lukas Krejci
+ */
+public class HadoopEventAndCalltimeDelegate extends Log4JLogEntryProcessor {
+
+ private Set<JobSummary> accumulatedJobEntries = new HashSet<JobSummary>();
+
+ public HadoopEventAndCalltimeDelegate(String eventType, File logFile) {
+ super(eventType, logFile);
+ }
+
+ public synchronized Set<JobSummary> drainAccumulatedJobs() {
+ HashSet<JobSummary> ret = new HashSet<JobSummary>(accumulatedJobEntries);
+ accumulatedJobEntries.clear();
+
+ return ret;
+ }
+
+ /**
+ * This does the very same thing as the super class method but the method is synchronized
+ * so that accumulated calltime data can be drained without race conditions using the
+ * {@link #drainAccumulatedJobs()} method.
+ */
+ @Override
+ public synchronized Set<Event> processLines(BufferedReader bufferedReader) throws IOException {
+ return super.processLines(bufferedReader);
+ }
+
+ @Override
+ protected LogEntry processPrimaryLine(Matcher matcher) throws ParseException {
+ LogEntry logEntry = super.processPrimaryLine(matcher);
+
+ //call this fragile ;) but LogEntry doesn't have much of a public API
+ String detail = matcher.group(3);
+
+ if (detail.startsWith(JobSummary.EXPECTED_LOGGER)) {
+ JobSummary summary = JobSummary.parseJobSummaryLogEntry(matcher.group());
+ if (summary != null) {
+ accumulatedJobEntries.add(summary);
+ }
+ }
+
+ return logEntry;
+ }
+}
diff --git a/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/calltime/JobSummary.java b/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/calltime/JobSummary.java
new file mode 100644
index 0000000..e8b6ee6
--- /dev/null
+++ b/modules/plugins/hadoop/src/main/java/org/rhq/plugins/hadoop/calltime/JobSummary.java
@@ -0,0 +1,177 @@
+/*
+ * RHQ Management Platform
+ * Copyright (C) 2005-2012 Red Hat, Inc.
+ * All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation version 2 of the License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+package org.rhq.plugins.hadoop.calltime;
+
+import java.io.Serializable;
+
+/**
+ * Description of a Hadoop map-reduce job.
+ *
+ * @author Lukas Krejci
+ */
+public class JobSummary implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final String EXPECTED_LOGGER = "org.apache.hadoop.mapred.JobInProgress$JobSummary";
+
+ private String jobId;
+ private String jobName;
+ private long startTime;
+ private long endTime;
+ private long submitTime;
+ private int mapTasks;
+ private int reduceTasks;
+ private boolean succeeded;
+
+ public static JobSummary parseJobSummaryLogEntry(String logEntry) {
+ //2012-08-03 13:55:45,224 INFO org.apache.hadoop.mapred.JobInProgress$JobSummary: jobId=job_201208031353_0001,submitTime=1343994869505,
+ //launchTime=1343994869801,firstMapTaskLaunchTime=1343994878611,firstReduceTaskLaunchTime=1343994884642,
+ //firstJobSetupTaskLaunchTime=1343994872548,firstJobCleanupTaskLaunchTime=1343994938904,finishTime=1343994944924,numMaps=16,
+ //numSlotsPerMap=1,numReduces=1,numSlotsPerReduce=1,user=lkrejci,queue=default,status=SUCCEEDED,mapSlotSeconds=76,reduceSlotsSeconds=53,
+ //clusterMapCapacity=2,clusterReduceCapacity=2,jobName=grep-search
+
+ JobSummary ret = new JobSummary();
+
+ int loggerIdx = logEntry.indexOf(EXPECTED_LOGGER);
+ if (loggerIdx < 0) {
+ return null;
+ }
+
+ loggerIdx += EXPECTED_LOGGER.length() + 2; //account for the colon and go 1 past it
+
+ String[] props = logEntry.substring(loggerIdx).split(",");
+
+ for(String prop : props) {
+ String[] keyValue = prop.split("=");
+
+ if (keyValue.length != 2) {
+ continue;
+ }
+
+ String key = keyValue[0];
+ String value = keyValue[1];
+
+ if ("jobId".equals(key)) {
+ ret.setJobId(value);
+ } else if ("submitTime".equals(key)) {
+ ret.setSubmitTime(Long.parseLong(value));
+ } else if ("launchTime".equals(key)) {
+ ret.setStartTime(Long.parseLong(value));
+ } else if ("finishTime".equals(key)) {
+ ret.setEndTime(Long.parseLong(value));
+ } else if ("numMaps".equals(key)) {
+ ret.setMapTasks(Integer.parseInt(value));
+ } else if ("numReduces".equals(key)) {
+ ret.setReduceTasks(Integer.parseInt(value));
+ } else if ("status".equals(key)) {
+ ret.setSucceeded("SUCCEEDED".equals(value));
+ } else if ("jobName".equals(key)) {
+ ret.setJobName(value);
+ }
+ }
+
+ return ret;
+ }
+
+ public String getJobId() {
+ return jobId;
+ }
+
+ public void setJobId(String jobId) {
+ this.jobId = jobId;
+ }
+
+ public String getJobName() {
+ return jobName;
+ }
+
+ public void setJobName(String jobName) {
+ this.jobName = jobName;
+ }
+
+ public long getStartTime() {
+ return startTime;
+ }
+
+ public void setStartTime(long startTime) {
+ this.startTime = startTime;
+ }
+
+ public long getEndTime() {
+ return endTime;
+ }
+
+ public void setEndTime(long endTime) {
+ this.endTime = endTime;
+ }
+
+ public int getMapTasks() {
+ return mapTasks;
+ }
+
+ public void setMapTasks(int mapTasks) {
+ this.mapTasks = mapTasks;
+ }
+
+ public int getReduceTasks() {
+ return reduceTasks;
+ }
+
+ public void setReduceTasks(int reduceTasks) {
+ this.reduceTasks = reduceTasks;
+ }
+
+ public boolean isSucceeded() {
+ return succeeded;
+ }
+
+ public void setSucceeded(boolean succeeded) {
+ this.succeeded = succeeded;
+ }
+
+ public long getSubmitTime() {
+ return submitTime;
+ }
+
+ public void setSubmitTime(long submitTime) {
+ this.submitTime = submitTime;
+ }
+
+ @Override
+ public int hashCode() {
+ return jobId == null ? 0 : jobId.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o == this) {
+ return true;
+ }
+
+ if (!(o instanceof JobSummary)) {
+ return false;
+ }
+
+ JobSummary other = (JobSummary) o;
+
+ return getJobId() == null ? (other.getJobId() == null) : (getJobId().equals(other.getJobId()));
+ }
+}
diff --git a/modules/plugins/hadoop/src/main/resources/META-INF/rhq-plugin.xml b/modules/plugins/hadoop/src/main/resources/META-INF/rhq-plugin.xml
index e648af9..c255ef4 100644
--- a/modules/plugins/hadoop/src/main/resources/META-INF/rhq-plugin.xml
+++ b/modules/plugins/hadoop/src/main/resources/META-INF/rhq-plugin.xml
@@ -146,7 +146,7 @@
</server>
<!-- JobTracker (http://wiki.apache.org/hadoop/JobTracker) -->
- <server name="Hadoop JobTracker" discovery="HadoopServerDiscovery" class="HadoopServerComponent">
+ <server name="Hadoop JobTracker" discovery="HadoopServerDiscovery" class="JobTrackerServerComponent">
<plugin-configuration>
<c:simple-property name="hadoop.home.dir" displayName="Home Directory" default="*** SHOULD HAVE BEEN AUTODETECTED ***"/>
<c:simple-property name="baseObjectName" defaultValue="hadoop:service=JobTracker"/>
@@ -209,6 +209,12 @@
<metric property="Hadoop:service=JobTracker,name=JobTrackerMetrics:trackers_graylisted" displayName="Graylisted Nodes"/>
<metric property="Hadoop:service=JobTracker,name=JobTrackerMetrics:trackers_decommissioned" displayName="Excluded Nodes"/>
+ <metric property="_synthetic_jobDuration" displayName="Job Duration" dataType="calltime" destinationType="Job Name" units="milliseconds"/>
+ <!-- Seems like RHQ cannot correctly handle more than 1 calltime metric per resource type?
+ <metric property="_synthetic_jobPreStartDelay" displayName="Submit To Start Duration" dataType="calltime" destinationType="Job Name" units="milliseconds"/>
+ <metric property="_synthetic_jobSuccessRate" displayName="Job Success Rate" dataType="calltime" destinationType="Job Name" units="milliseconds" />
+ -->
+
<event name="logEntry" description="an entry in a log file"/>
<resource-configuration>
commit 1f40ca91a8ad6c49666000742668afb7a83baa5b
Author: Lukas Krejci <lkrejci at redhat.com>
Date: Mon Aug 6 10:50:45 2012 +0200
bumping the required java version to 6 (just to be consistent with the min
reqs of the RHQ agent).
diff --git a/modules/plugins/hadoop/pom.xml b/modules/plugins/hadoop/pom.xml
index 7aae62d..2e508f7 100644
--- a/modules/plugins/hadoop/pom.xml
+++ b/modules/plugins/hadoop/pom.xml
@@ -20,8 +20,8 @@
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
- <source>1.5</source>
- <target>1.5</target>
+ <source>1.6</source>
+ <target>1.6</target>
</configuration>
</plugin>
More information about the rhq-commits
mailing list