[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