You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@oozie.apache.org by rk...@apache.org on 2013/08/29 20:06:58 UTC
svn commit: r1518739 [2/4] - in /oozie/trunk: ./
client/src/main/java/org/apache/oozie/cli/
client/src/main/java/org/apache/oozie/client/
client/src/main/java/org/apache/oozie/client/rest/ core/
core/src/main/java/org/apache/oozie/ core/src/main/java/o...
Added: oozie/trunk/core/src/main/java/org/apache/oozie/service/ZKXLogStreamingService.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/service/ZKXLogStreamingService.java?rev=1518739&view=auto
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/service/ZKXLogStreamingService.java (added)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/service/ZKXLogStreamingService.java Thu Aug 29 18:06:57 2013
@@ -0,0 +1,269 @@
+/**
+ * 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.oozie.service;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.FileReader;
+import org.apache.oozie.util.Instrumentable;
+import org.apache.oozie.util.Instrumentation;
+import org.apache.oozie.util.XLogStreamer;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.Writer;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import org.apache.curator.x.discovery.ServiceInstance;
+import org.apache.oozie.ErrorCode;
+import org.apache.oozie.client.OozieClient;
+import org.apache.oozie.client.rest.RestConstants;
+import org.apache.oozie.util.IOUtils;
+import org.apache.oozie.util.SimpleTimestampedMessageParser;
+import org.apache.oozie.util.TimestampedMessageParser;
+import org.apache.oozie.util.XLog;
+import org.apache.oozie.util.ZKUtils;
+
+/**
+ * Service that performs streaming of log files over Web Services if enabled in XLogService and collates logs from other Oozie
+ * servers. Requires that a ZooKeeper ensemble is available.
+ */
+public class ZKXLogStreamingService extends XLogStreamingService implements Service, Instrumentable {
+
+ private static final String ALL_SERVERS_PARAM = "allservers";
+
+ private ZKUtils zk;
+ private XLog log;
+
+ /**
+ * Initialize the log streaming service.
+ *
+ * @param services services instance.
+ * @throws ServiceException thrown if the log streaming service could not be initialized.
+ */
+ @Override
+ public void init(Services services) throws ServiceException {
+ super.init(services);
+ try {
+ zk = ZKUtils.register(this);
+ }
+ catch (Exception ex) {
+ throw new ServiceException(ErrorCode.E1700, ex.getMessage(), ex);
+ }
+ log = XLog.getLog(this.getClass());
+ }
+
+ /**
+ * Destroy the log streaming service.
+ */
+ @Override
+ public void destroy() {
+ if (zk != null) {
+ zk.unregister(this);
+ }
+ zk = null;
+ super.destroy();
+ }
+
+ /**
+ * Instruments the log streaming service.
+ *
+ * @param instr instrumentation to use.
+ */
+ @Override
+ public void instrument(Instrumentation instr) {
+ super.instrument(instr);
+ }
+
+ /**
+ * Stream the log of a job. It contacts any other running Oozie servers to collate relevant logs while streaming.
+ *
+ * @param filter log streamer filter.
+ * @param startTime start time for log events to filter.
+ * @param endTime end time for log events to filter.
+ * @param writer writer to stream the log to.
+ * @param params additional parameters from the request
+ * @throws IOException thrown if the log cannot be streamed.
+ */
+ @Override
+ public void streamLog(XLogStreamer.Filter filter, Date startTime, Date endTime, Writer writer, Map<String, String[]> params)
+ throws IOException {
+ XLogService xLogService = Services.get().get(XLogService.class);
+ if (xLogService.getLogOverWS()) {
+ // If ALL_SERVERS_PARAM is set to false, then only stream our log
+ if (params.get(ALL_SERVERS_PARAM) != null && params.get(ALL_SERVERS_PARAM).length > 0
+ && params.get(ALL_SERVERS_PARAM)[0].equals("false")) {
+ new XLogStreamer(filter, xLogService.getOozieLogPath(), xLogService.getOozieLogName(),
+ xLogService.getOozieLogRotation()).streamLog(writer, startTime, endTime);
+ }
+ // Otherwise, we have to go collate relevant logs from the other Oozie servers
+ else {
+ collateLogs(filter, startTime, endTime, writer);
+ }
+ }
+ else {
+ writer.write("Log streaming disabled!!");
+ }
+ }
+
+ /**
+ * Contacts each of the other Oozie servers, gets their logs for the job, collates them, and sends them to the user via the
+ * Writer. It will make sure to not read all of the log messages into memory at the same time to not use up the heap. If there
+ * is a problem talking to one of the other servers, it will ignore that server and prepend a message to the Writer about it.
+ * For getting the logs from this server, it won't use the REST API and instead get them directly to be more efficient.
+ *
+ * @param filter
+ * @param startTime
+ * @param endTime
+ * @param writer
+ * @throws IOException
+ */
+ private void collateLogs(XLogStreamer.Filter filter, Date startTime, Date endTime, Writer writer) throws IOException {
+ XLogService xLogService = Services.get().get(XLogService.class);
+ List<String> badOozies = new ArrayList<String>();
+ List<ServiceInstance<Map>> oozies = null;
+ try {
+ oozies = zk.getAllMetaData();
+ }
+ catch (Exception ex) {
+ throw new IOException("Issue communicating with ZooKeeper: " + ex.getMessage(), ex);
+ }
+ List<TimestampedMessageParser> parsers = new ArrayList<TimestampedMessageParser>(oozies.size());
+ List<File> localTempFiles = new ArrayList<File>(oozies.size() - 1);
+ try {
+ // Create a BufferedReader for getting the logs of each server and put them in a TimestampedMessageParser
+ for (ServiceInstance<Map> oozie : oozies) {
+ Map<String, String> oozieMeta = oozie.getPayload();
+ String otherId = oozieMeta.get(ZKUtils.ZKMetadataKeys.OOZIE_ID);
+ // If it's this server, we can just get them directly
+ if (otherId.equals(zk.getZKId())) {
+ BufferedReader reader = new XLogStreamer(filter, xLogService.getOozieLogPath(), xLogService.getOozieLogName(),
+ xLogService.getOozieLogRotation()).makeReader(startTime, endTime);
+ parsers.add(new TimestampedMessageParser(reader, filter));
+ }
+ // If it's another server, we'll have to use the REST API
+ else {
+ String otherUrl = oozieMeta.get(ZKUtils.ZKMetadataKeys.OOZIE_URL);
+ String jobId = filter.getFilterParams().get(DagXLogInfoService.JOB);
+ try {
+ File f = File.createTempFile(otherId, ".log"); // Create a temp file to stream the log to
+ localTempFiles.add(f);
+ BufferedReader reader = fetchOtherLog(otherUrl, jobId, f);
+ parsers.add(new SimpleTimestampedMessageParser(reader, filter));
+ }
+ catch(IOException ioe) {
+ log.warn("Failed to retrieve logs for job [" + jobId + "] from Oozie server with ID [" + otherId
+ + "] at [" + otherUrl + "]; log information may be incomplete", ioe);
+ badOozies.add(otherId);
+ }
+ }
+ }
+
+ // Add a message about any servers we couldn't contact
+ if (!badOozies.isEmpty()) {
+ writer.write("Unable to contact the following Oozie Servers for logs (log information may be incomplete):\n");
+ for (String badOozie : badOozies) {
+ writer.write(" ");
+ writer.write(badOozie);
+ writer.write("\n");
+ }
+ writer.write("\n");
+ writer.flush();
+ }
+
+ // If it's just the one server (this server), then we don't need to do any more processing and can just copy it directly
+ if (parsers.size() == 1) {
+ TimestampedMessageParser parser = parsers.get(0);
+ parser.processRemaining(writer);
+ }
+ else {
+ // Now that we have a Reader for each server to get the logs from that server, we have to collate them. Within each
+ // server, the logs should already be in the correct order, so we can take advantage of that. We'll use the
+ // BufferedReaders to read the messages from the logs of each server and put them in order without having to bring
+ // every message into memory at the same time.
+ TreeMap<String, TimestampedMessageParser> timestampMap = new TreeMap<String, TimestampedMessageParser>();
+ // populate timestampMap with initial values
+ for (TimestampedMessageParser parser : parsers) {
+ if (parser.increment()) {
+ timestampMap.put(parser.getLastTimestamp(), parser);
+ }
+ }
+ while (timestampMap.size() > 1) {
+ // The first entry will be the earliest based on the timestamp (also removes it) from the map
+ TimestampedMessageParser earliestParser = timestampMap.pollFirstEntry().getValue();
+ // Write the message from that parser at that timestamp
+ writer.write(earliestParser.getLastMessage());
+ // Increment that parser to read the next message
+ if (earliestParser.increment()) {
+ // If it still has messages left, put it back in the map with the new last timestamp for it
+ timestampMap.put(earliestParser.getLastTimestamp(), earliestParser);
+ }
+ }
+ // If there's only one parser left in the map, then we can simply copy the rest of its lines directly to be faster
+ if (timestampMap.size() == 1) {
+ TimestampedMessageParser parser = timestampMap.values().iterator().next();
+ writer.write(parser.getLastMessage()); // don't forget the last message read by the parser
+ parser.processRemaining(writer);
+ }
+ }
+ }
+ finally {
+ for (TimestampedMessageParser parser : parsers) {
+ parser.closeReader();
+ }
+ for (File f : localTempFiles) {
+ f.delete();
+ }
+ writer.flush();
+ }
+ }
+
+ /**
+ * Creates a connection over HTTP to another Oozie server and asks it for the logs related to the jobId. It will then stream
+ * the logs to a temporary file in the local filesystem and return a BufferedReader to that file (this prevents us from leaving
+ * an Http stream open for too long while processing line-by-line and running into timeouts).
+ *
+ * @param otherUrl The URL of the other Oozie server
+ * @param jobId The job id of the job we want logs for
+ * @param tempFile The temporary file to save the log to
+ * @return a BufferedReader of the logs from the other server for the jobId
+ * @throws IOException If there was a problem connecting to the other Oozie server
+ */
+ private BufferedReader fetchOtherLog(String otherUrl, String jobId, File tempFile) throws IOException {
+ // It's important that we specify ALL_SERVERS_PARAM=false in the GET request to prevent the other Oozie Server from trying
+ // aggregate logs from the other Oozie servers (and creating an infinite recursion)
+ URL url = new URL(otherUrl + "/v" + OozieClient.WS_PROTOCOL_VERSION + "/" + RestConstants.JOB + "/" + jobId
+ + "?" + RestConstants.JOB_SHOW_PARAM + "=" + RestConstants.JOB_SHOW_LOG + "&" + ALL_SERVERS_PARAM + "=false");
+ HttpURLConnection conn = (HttpURLConnection) url.openConnection();
+ BufferedReader reader = null;
+ if ((conn.getResponseCode() == HttpURLConnection.HTTP_OK)) {
+ InputStream is = conn.getInputStream();
+ FileOutputStream fos = new FileOutputStream(tempFile);
+ IOUtils.copyStream(is, fos);
+ is.close();
+ fos.close();
+ reader = new BufferedReader(new FileReader(tempFile));
+ }
+ return reader;
+ }
+}
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/servlet/BaseAdminServlet.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/servlet/BaseAdminServlet.java?rev=1518739&r1=1518738&r2=1518739&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/servlet/BaseAdminServlet.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/servlet/BaseAdminServlet.java Thu Aug 29 18:06:57 2013
@@ -142,7 +142,11 @@ public abstract class BaseAdminServlet e
JsonBean jmsBean = getJMSConnectionInfo(request, response);
sendJsonResponse(response, HttpServletResponse.SC_OK, jmsBean, timeZoneId);
}
-
+ else if (resource.equals(RestConstants.ADMIN_AVAILABLE_OOZIE_SERVERS_RESOURCE)) {
+ JSONObject json = new JSONObject();
+ json.putAll(getOozieURLs());
+ sendJsonResponse(response, HttpServletResponse.SC_OK, json);
+ }
}
@@ -245,4 +249,5 @@ public abstract class BaseAdminServlet e
return array;
}
+ protected abstract Map<String, String> getOozieURLs() throws XServletException;
}
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/servlet/ServicesLoader.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/servlet/ServicesLoader.java?rev=1518739&r1=1518738&r2=1518739&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/servlet/ServicesLoader.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/servlet/ServicesLoader.java Thu Aug 29 18:06:57 2013
@@ -27,6 +27,7 @@ import javax.servlet.ServletContextEvent
*/
public class ServicesLoader implements ServletContextListener {
private static Services services;
+ private static boolean sslEnabled = false;
/**
* Initialize Oozie services.
@@ -35,6 +36,11 @@ public class ServicesLoader implements S
*/
public void contextInitialized(ServletContextEvent event) {
try {
+ String ssl = event.getServletContext().getInitParameter("ssl.enabled");
+ if (ssl != null) {
+ sslEnabled = true;
+ }
+
services = new Services();
services.init();
}
@@ -62,4 +68,7 @@ public class ServicesLoader implements S
services.destroy();
}
+ public static boolean isSSLEnabled() {
+ return sslEnabled;
+ }
}
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V0AdminServlet.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V0AdminServlet.java?rev=1518739&r1=1518738&r2=1518739&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V0AdminServlet.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V0AdminServlet.java Thu Aug 29 18:06:57 2013
@@ -20,6 +20,7 @@ package org.apache.oozie.servlet;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Map;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
@@ -118,4 +119,9 @@ public class V0AdminServlet extends Base
IOException {
throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ErrorCode.E0302, "Not supported in v0");
}
+
+ @Override
+ protected Map<String, String> getOozieURLs() throws XServletException {
+ throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ErrorCode.E0302, "Not supported in v1");
+ }
}
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V0JobServlet.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V0JobServlet.java?rev=1518739&r1=1518738&r2=1518739&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V0JobServlet.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V0JobServlet.java Thu Aug 29 18:06:57 2013
@@ -186,7 +186,7 @@ public class V0JobServlet extends BaseJo
String jobId = getResourceName(request);
try {
- dagEngine.streamLog(jobId, response.getWriter());
+ dagEngine.streamLog(jobId, response.getWriter(), request.getParameterMap());
}
catch (DagEngineException ex) {
throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ex);
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V1AdminServlet.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V1AdminServlet.java?rev=1518739&r1=1518738&r2=1518739&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V1AdminServlet.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V1AdminServlet.java Thu Aug 29 18:06:57 2013
@@ -21,6 +21,7 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.Properties;
import javax.servlet.http.HttpServletRequest;
@@ -45,7 +46,7 @@ public class V1AdminServlet extends Base
private static final long serialVersionUID = 1L;
private static final String INSTRUMENTATION_NAME = "v1admin";
- private static final ResourceInfo RESOURCES_INFO[] = new ResourceInfo[9];
+ private static final ResourceInfo RESOURCES_INFO[] = new ResourceInfo[10];
static {
RESOURCES_INFO[0] = new ResourceInfo(RestConstants.ADMIN_STATUS_RESOURCE, Arrays.asList("PUT", "GET"),
@@ -67,7 +68,8 @@ public class V1AdminServlet extends Base
Collections.EMPTY_LIST);
RESOURCES_INFO[8] = new ResourceInfo((RestConstants.ADMIN_JMS_INFO), Arrays.asList("GET"),
Collections.EMPTY_LIST);
-
+ RESOURCES_INFO[9] = new ResourceInfo(RestConstants.ADMIN_AVAILABLE_OOZIE_SERVERS_RESOURCE, Arrays.asList("GET"),
+ Collections.EMPTY_LIST);
}
protected V1AdminServlet(String name) {
@@ -152,5 +154,8 @@ public class V1AdminServlet extends Base
throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ErrorCode.E0302, "Not supported in v1");
}
-
+ @Override
+ protected Map<String, String> getOozieURLs() throws XServletException {
+ throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ErrorCode.E0302, "Not supported in v1");
+ }
}
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V1JobServlet.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V1JobServlet.java?rev=1518739&r1=1518738&r2=1518739&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V1JobServlet.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V1JobServlet.java Thu Aug 29 18:06:57 2013
@@ -955,7 +955,7 @@ public class V1JobServlet extends BaseJo
DagEngine dagEngine = Services.get().get(DagEngineService.class).getDagEngine(getUser(request));
String jobId = getResourceName(request);
try {
- dagEngine.streamLog(jobId, response.getWriter());
+ dagEngine.streamLog(jobId, response.getWriter(), request.getParameterMap());
}
catch (DagEngineException ex) {
throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ex);
@@ -974,7 +974,7 @@ public class V1JobServlet extends BaseJo
BundleEngine bundleEngine = Services.get().get(BundleEngineService.class).getBundleEngine(getUser(request));
String jobId = getResourceName(request);
try {
- bundleEngine.streamLog(jobId, response.getWriter());
+ bundleEngine.streamLog(jobId, response.getWriter(), request.getParameterMap());
}
catch (BundleEngineException ex) {
throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ex);
@@ -998,7 +998,7 @@ public class V1JobServlet extends BaseJo
String logRetrievalScope = request.getParameter(RestConstants.JOB_LOG_SCOPE_PARAM);
String logRetrievalType = request.getParameter(RestConstants.JOB_LOG_TYPE_PARAM);
try {
- coordEngine.streamLog(jobId, logRetrievalScope, logRetrievalType, response.getWriter());
+ coordEngine.streamLog(jobId, logRetrievalScope, logRetrievalType, response.getWriter(), request.getParameterMap());
}
catch (BaseEngineException ex) {
throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ex);
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V2AdminServlet.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V2AdminServlet.java?rev=1518739&r1=1518738&r2=1518739&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V2AdminServlet.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V2AdminServlet.java Thu Aug 29 18:06:57 2013
@@ -18,6 +18,7 @@
package org.apache.oozie.servlet;
import java.io.IOException;
+import java.util.Map;
import java.util.Properties;
import javax.servlet.http.HttpServletRequest;
@@ -30,6 +31,7 @@ import org.apache.oozie.client.rest.Json
import org.apache.oozie.jms.JMSConnectionInfo;
import org.apache.oozie.jms.JMSJobEventListener;
import org.apache.oozie.service.JMSTopicService;
+import org.apache.oozie.service.JobsConcurrencyService;
import org.apache.oozie.service.Services;
/**
@@ -74,4 +76,14 @@ public class V2AdminServlet extends V1Ad
return jmsBean;
}
+ @Override
+ protected Map<String, String> getOozieURLs() throws XServletException {
+ Map<String, String> serverUrls = null;
+ try {
+ serverUrls = Services.get().get(JobsConcurrencyService.class).getServerUrls();
+ } catch (Exception ex) {
+ throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ErrorCode.E0307, ex.getMessage(), ex);
+ }
+ return serverUrls;
+ }
}
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/util/ConfigUtils.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/util/ConfigUtils.java?rev=1518739&r1=1518738&r2=1518739&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/util/ConfigUtils.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/util/ConfigUtils.java Thu Aug 29 18:06:57 2013
@@ -18,6 +18,7 @@
package org.apache.oozie.util;
import org.apache.hadoop.conf.Configuration;
+import org.apache.oozie.servlet.ServicesLoader;
/**
*
@@ -66,4 +67,35 @@ public class ConfigUtils {
String value = getWithDeprecatedCheck(conf, newName, oldName, Boolean.toString(defaultValue));
return Boolean.parseBoolean(value);
}
+
+ /**
+ * Returns the HTTP or HTTPS URL for this Oozie server
+ * (http://HOSTNAME:HTTP_PORT/oozie or https://HOSTNAME:HTTPS_PORT/oozie)
+ *
+ * @param secure true to return the HTTPS URL or false to return the HTTP URL
+ * @return the HTTP or HTTPS URL for this Oozie server
+ */
+ public static String getOozieURL(boolean secure) {
+ StringBuilder sb = new StringBuilder();
+ if (secure) {
+ sb.append("https://");
+ }
+ else {
+ sb.append("http://");
+ }
+ sb.append(System.getProperty("oozie.http.hostname"));
+ sb.append(":");
+ sb.append(System.getProperty("oozie.http.port"));
+ sb.append("/oozie");
+ return sb.toString();
+ }
+
+ /**
+ * Returns the HTTP or HTTPS URL for this Oozie server depending on which is actually configured
+ *
+ * @return the HTTP or HTTPS URL for this Oozie server
+ */
+ public static String getOozieEffectiveUrl() {
+ return getOozieURL(ServicesLoader.isSSLEnabled());
+ }
}
Added: oozie/trunk/core/src/main/java/org/apache/oozie/util/FixedJsonInstanceSerializer.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/util/FixedJsonInstanceSerializer.java?rev=1518739&view=auto
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/util/FixedJsonInstanceSerializer.java (added)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/util/FixedJsonInstanceSerializer.java Thu Aug 29 18:06:57 2013
@@ -0,0 +1,163 @@
+/**
+ * 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.oozie.util;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+import org.codehaus.jackson.JsonNode;
+import org.codehaus.jackson.JsonParseException;
+import org.codehaus.jackson.map.DeserializationConfig;
+import org.codehaus.jackson.map.JsonMappingException;
+import org.codehaus.jackson.map.ObjectMapper;
+
+import com.google.common.base.Preconditions;
+import org.apache.curator.x.discovery.ServiceInstance;
+import org.apache.curator.x.discovery.ServiceInstanceBuilder;
+import org.apache.curator.x.discovery.ServiceType;
+import org.apache.curator.x.discovery.UriSpec;
+import org.apache.curator.x.discovery.details.InstanceSerializer;
+
+// TODO: Workaround for CURATOR-5 (https://issues.apache.org/jira/browse/CURATOR-5)
+// Remove this class (code from pull request listed on JIRA) and use regular JsonInstanceSerializer once fixed
+// (Otherwise we can't properly serialize objects for the ZK Service Discovery)
+public class FixedJsonInstanceSerializer<T> implements InstanceSerializer<T>
+{
+
+ private final ObjectMapper mMapper;
+ private final Class<T> mPayloadClass;
+
+ /**
+ * @param payloadClass
+ * used to validate payloads when deserializing
+ */
+ public FixedJsonInstanceSerializer(final Class<T> payloadClass) {
+ this(payloadClass, new ObjectMapper());
+ }
+
+ public FixedJsonInstanceSerializer(final Class<T> pPayloadClass, final ObjectMapper pMapper) {
+ mPayloadClass = pPayloadClass;
+ mMapper = pMapper;
+ mMapper.configure(DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+ }
+
+ @Override
+ public byte[] serialize(final ServiceInstance<T> pInstance) throws Exception {
+ final ByteArrayOutputStream out = new ByteArrayOutputStream();
+ mMapper.writeValue(out, pInstance);
+ return out.toByteArray();
+
+ }
+
+ private String getTextField(final JsonNode pNode, final String pFieldName) {
+ Preconditions.checkNotNull(pNode);
+ Preconditions.checkNotNull(pFieldName);
+ return pNode.get(pFieldName) != null ? pNode.get(pFieldName).getTextValue() : null;
+ }
+
+ private Integer getIntegerField(final JsonNode pNode, final String pFieldName) {
+ Preconditions.checkNotNull(pNode);
+ Preconditions.checkNotNull(pFieldName);
+ return (pNode.get(pFieldName) != null && pNode.get(pFieldName).isNumber()) ? pNode.get(pFieldName)
+ .getIntValue() : null;
+ }
+
+ private Long getLongField(final JsonNode pNode, final String pFieldName) {
+ Preconditions.checkNotNull(pNode);
+ Preconditions.checkNotNull(pFieldName);
+ return (pNode.get(pFieldName) != null && pNode.get(pFieldName).isLong()) ? pNode.get(pFieldName).getLongValue()
+ : null;
+ }
+
+ private <O> O getObject(final JsonNode pNode, final String pFieldName, final Class<O> pObjectClass)
+ throws JsonParseException, JsonMappingException, IOException {
+ Preconditions.checkNotNull(pNode);
+ Preconditions.checkNotNull(pFieldName);
+ Preconditions.checkNotNull(pObjectClass);
+ if (pNode.get(pFieldName) != null && pNode.get(pFieldName).isObject()) {
+ return mMapper.readValue(pNode.get(pFieldName), pObjectClass);
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public ServiceInstance<T> deserialize(final byte[] pBytes) throws Exception {
+ final ByteArrayInputStream bais = new ByteArrayInputStream(pBytes);
+ final JsonNode rootNode = mMapper.readTree(bais);
+ final ServiceInstanceBuilder<T> builder = ServiceInstance.builder();
+ {
+ final String address = getTextField(rootNode, "address");
+ if (address != null) {
+ builder.address(address);
+ }
+ }
+ {
+ final String id = getTextField(rootNode, "id");
+ if (id != null) {
+ builder.id(id);
+ }
+ }
+ {
+ final String name = getTextField(rootNode, "name");
+ if (name != null) {
+ builder.name(name);
+ }
+ }
+ {
+ final Integer port = getIntegerField(rootNode, "port");
+ if (port != null) {
+ builder.port(port);
+ }
+ }
+ {
+ final Integer sslPort = getIntegerField(rootNode, "sslPort");
+ if (sslPort != null) {
+ builder.sslPort(sslPort);
+ }
+ }
+ {
+ final Long registrationTimeUTC = getLongField(rootNode, "registrationTimeUTC");
+ if (registrationTimeUTC != null) {
+ builder.registrationTimeUTC(registrationTimeUTC);
+ }
+ }
+ {
+ final T payload = getObject(rootNode, "payload", mPayloadClass);
+ if (payload != null) {
+ builder.payload(payload);
+ }
+ }
+ {
+ final ServiceType serviceType = getObject(rootNode, "serviceType", ServiceType.class);
+ if (serviceType != null) {
+ builder.serviceType(serviceType);
+ }
+ }
+ {
+ final UriSpec uriSpec = getObject(rootNode, "uriSpec", UriSpec.class);
+ if (uriSpec != null) {
+ builder.uriSpec(uriSpec);
+ }
+ }
+ return builder.build();
+ }
+
+}
\ No newline at end of file
Added: oozie/trunk/core/src/main/java/org/apache/oozie/util/MultiFileReader.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/util/MultiFileReader.java?rev=1518739&view=auto
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/util/MultiFileReader.java (added)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/util/MultiFileReader.java Thu Aug 29 18:06:57 2013
@@ -0,0 +1,93 @@
+/**
+ * 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.oozie.util;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.Reader;
+import java.util.ArrayList;
+import java.util.zip.GZIPInputStream;
+
+/**
+ * Implementation of a {@link Reader} which can be used to read in multiple files sequentially. That is, when the first file ends
+ * it will silently move to the next file and so on. If the file has a ".gz" extension, this Reader will properly handle it; all
+ * other types of files will simply be read using a {@link FileReader}.
+ */
+public class MultiFileReader extends Reader {
+
+ private ArrayList<File> files;
+ private int index;
+ private Reader reader;
+ private boolean closed;
+
+ /**
+ * Constructs the MultiFileReader with the given files. The files will be read in the order given in the ArrayList.
+ *
+ * @param files The files to read
+ * @throws IOException If there was a problem opening the first file
+ */
+ public MultiFileReader(ArrayList<File> files) throws IOException {
+ this.files = files;
+ closed = false;
+ index = 0;
+ reader = null;
+ openNextReader();
+ }
+
+ @Override
+ public int read(char[] cbuf, int off, int len) throws IOException {
+ int numRead = -1;
+ while(!closed && numRead == -1) {
+ numRead = reader.read(cbuf, off, len);
+ if (numRead == -1) {
+ reader.close();
+ openNextReader();
+ }
+ }
+ return numRead;
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (reader != null) {
+ reader.close();
+ }
+ closed = true;
+ }
+
+ private void openNextReader() throws IOException {
+ if (index < files.size()) {
+ // gzip files
+ if (files.get(index).getName().endsWith(".gz")) {
+ GZIPInputStream gzipInputStream = new GZIPInputStream(new FileInputStream(files.get(index)));
+ reader = new InputStreamReader(gzipInputStream);
+ }
+ // regular files
+ else {
+ reader = new FileReader(files.get(index));
+ }
+ index++;
+ }
+ else {
+ closed = true;
+ }
+ }
+}
Added: oozie/trunk/core/src/main/java/org/apache/oozie/util/SimpleTimestampedMessageParser.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/util/SimpleTimestampedMessageParser.java?rev=1518739&view=auto
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/util/SimpleTimestampedMessageParser.java (added)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/util/SimpleTimestampedMessageParser.java Thu Aug 29 18:06:57 2013
@@ -0,0 +1,44 @@
+/**
+ * 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.oozie.util;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+
+/**
+ * A simplified version of {@link TimestampedMessageParser} which doesn't do any filtering when reading the lines from the Reader.
+ * This is useful when processing a Reader where the lines have already been filtered, so we can be more efficient by not
+ * doing the same filtering again.
+ */
+public class SimpleTimestampedMessageParser extends TimestampedMessageParser {
+
+ public SimpleTimestampedMessageParser(BufferedReader reader, XLogStreamer.Filter filter) {
+ super(reader, filter);
+ }
+
+ /**
+ * This implementation simply returns the next line from the Reader.
+ *
+ * @return The next line from the Reader
+ * @throws IOException
+ */
+ @Override
+ protected String parseNextLine() throws IOException {
+ return reader.readLine();
+ }
+}
\ No newline at end of file
Added: oozie/trunk/core/src/main/java/org/apache/oozie/util/TimestampedMessageParser.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/util/TimestampedMessageParser.java?rev=1518739&view=auto
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/util/TimestampedMessageParser.java (added)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/util/TimestampedMessageParser.java Thu Aug 29 18:06:57 2013
@@ -0,0 +1,173 @@
+/**
+ * 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.oozie.util;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.Writer;
+import java.util.ArrayList;
+
+ /**
+ * Encapsulates the parsing and filtering of the log messages from a BufferedReader. It makes sure not to read in the entire log
+ * into memory at the same time; at most, it will have two messages (which can be multi-line in the case of exception stack traces).
+ * <p>
+ * To use this class: Calling {@link TimestampedMessageParser#increment()} will tell the parser to read the next message from the
+ * Reader. It will return true if there are more messages and false if not. Calling
+ * {@link TimestampedMessageParser#getLastMessage()} and {@link TimestampedMessageParser#getLastTimestamp()} will return the last
+ * message and timestamp, respectively, that were parsed when {@link TimestampedMessageParser#increment()} was called. Calling
+ * {@link TimestampedMessageParser#processRemaining(java.io.Writer)} will write the remaining log messages to the given Writer.
+ */
+public class TimestampedMessageParser {
+
+ protected BufferedReader reader;
+ private String nextLine = null;
+ private String lastTimestamp = null;
+ private XLogStreamer.Filter filter;
+ private boolean empty = false;
+ private String lastMessage = null;
+ private boolean patternMatched = false;
+
+ /**
+ * Creates a TimestampedMessageParser with the given BufferedReader and filter.
+ *
+ * @param reader The BufferedReader to get the log messages from
+ * @param filter The filter
+ */
+ public TimestampedMessageParser(BufferedReader reader, XLogStreamer.Filter filter) {
+ this.reader = reader;
+ this.filter = filter;
+ if (filter == null) {
+ filter = new XLogStreamer.Filter();
+ }
+ filter.constructPattern();
+ }
+
+ /**
+ * Causes the next message and timestamp to be parsed from the BufferedReader.
+ *
+ * @return true if there are more messages left; false if not
+ * @throws IOException If there was a problem reading from the Reader
+ */
+ public boolean increment() throws IOException {
+ if (empty) {
+ return false;
+ }
+
+ StringBuilder message = new StringBuilder();
+
+ if (nextLine == null) { // first time only
+ nextLine = parseNextLine();
+ if (nextLine == null) { // reader finished
+ empty = true;
+ return false;
+ }
+ }
+ lastTimestamp = parseTimestamp(nextLine);
+ String nextTimestamp = null;
+ while (nextTimestamp == null) {
+ message.append(nextLine).append("\n");
+ nextLine = parseNextLine();
+ if (nextLine != null) {
+ nextTimestamp = parseTimestamp(nextLine); // exit loop if we have a timestamp, continue if not
+ }
+ else { // reader finished
+ empty = true;
+ nextTimestamp = ""; // exit loop
+ }
+ }
+
+ lastMessage = message.toString();
+ return true;
+ }
+
+
+ /**
+ * Returns the timestamp from the last message that was parsed.
+ *
+ * @return the timestamp from the last message that was parsed
+ */
+ public String getLastTimestamp() {
+ return lastTimestamp;
+ }
+
+ /**
+ * Returns the message that was last parsed.
+ *
+ * @return the message that was last parsed
+ */
+ public String getLastMessage() {
+ return lastMessage;
+ }
+
+ /**
+ * Closes the Reader.
+ *
+ * @throws IOException
+ */
+ public void closeReader() throws IOException {
+ reader.close();
+ }
+
+ /**
+ * Reads the next line from the Reader and checks if it matches the filter. It can also handle multi-line messages (i.e.
+ * exception stack traces). If it returns null, then there are no lines left in the Reader.
+ *
+ * @return The next line, or null
+ * @throws IOException
+ */
+ protected String parseNextLine() throws IOException {
+ String line = reader.readLine();
+ if (line != null) {
+ ArrayList<String> logParts = filter.splitLogMessage(line);
+ if (logParts != null) {
+ patternMatched = filter.matches(logParts);
+ }
+ if (!patternMatched) {
+ line = parseNextLine();
+ }
+ }
+ return line;
+ }
+
+ /**
+ * Parses the timestamp out of the passed in line. If there isn't one, it returns null.
+ *
+ * @param line The line to check
+ * @return the timestamp of the line, or null
+ */
+ private String parseTimestamp(String line) {
+ String timestamp = null;
+ ArrayList<String> logParts = filter.splitLogMessage(line);
+ if (logParts != null) {
+ timestamp = logParts.get(0);
+ }
+ return timestamp;
+ }
+
+ /**
+ * Writes the remaining log messages to the passed in Writer.
+ *
+ * @param writer
+ * @throws IOException
+ */
+ public void processRemaining(Writer writer) throws IOException {
+ while(increment()) {
+ writer.write(getLastMessage());
+ }
+ }
+}
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/util/XLogStreamer.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/util/XLogStreamer.java?rev=1518739&r1=1518738&r2=1518739&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/util/XLogStreamer.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/util/XLogStreamer.java Thu Aug 29 18:06:57 2013
@@ -18,9 +18,7 @@
package org.apache.oozie.util;
import java.io.File;
-import java.io.FileInputStream;
import java.io.IOException;
-import java.io.InputStream;
import java.io.Writer;
import java.util.ArrayList;
import java.util.Calendar;
@@ -31,12 +29,12 @@ import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
-import java.util.zip.GZIPInputStream;
import com.google.common.annotations.VisibleForTesting;
+import java.io.BufferedReader;
/**
- * XLogStreamer streams the given log file to logWriter after applying the given filter.
+ * XLogStreamer streams the given log file to writer after applying the given filter.
*/
public class XLogStreamer {
private static XLog LOG = XLog.getLog(XLogStreamer.class);
@@ -116,8 +114,8 @@ public class XLogStreamer {
* @return
*/
public boolean matches(ArrayList<String> logParts) {
- String logLevel = logParts.get(0);
- String logMessage = logParts.get(1);
+ String logLevel = logParts.get(1);
+ String logMessage = logParts.get(2);
if (this.logLevels == null || this.logLevels.containsKey(logLevel.toUpperCase())) {
Matcher logMatcher = filterPattern.matcher(logMessage);
return logMatcher.matches();
@@ -128,7 +126,7 @@ public class XLogStreamer {
}
/**
- * Splits the log line into timestamp, logLevel and remaining log message. Returns array containing logLevel and
+ * Splits the log line into timestamp, logLevel and remaining log message. Returns array containing timestamp, logLevel, and
* logMessage if the pattern matches i.e A new log statement, else returns null.
*
* @param logLine
@@ -138,6 +136,7 @@ public class XLogStreamer {
Matcher splitter = SPLITTER_PATTERN.matcher(logLine);
if (splitter.matches()) {
ArrayList<String> logParts = new ArrayList<String>();
+ logParts.add(splitter.group(1));// timestamp
logParts.add(splitter.group(2));// log level
logParts.add(splitter.group(3));// Log Message
return logParts;
@@ -184,11 +183,9 @@ public class XLogStreamer {
private String logFile;
private String logPath;
private Filter logFilter;
- private Writer logWriter;
private long logRotation;
- public XLogStreamer(Filter logFilter, Writer logWriter, String logPath, String logFile, long logRotationSecs) {
- this.logWriter = logWriter;
+ public XLogStreamer(Filter logFilter, String logPath, String logFile, long logRotationSecs) {
this.logFilter = logFilter;
if (logFile == null) {
logFile = "oozie-app.log";
@@ -202,11 +199,33 @@ public class XLogStreamer {
* Gets the files that are modified between startTime and endTime in the given logPath and streams the log after
* applying the filters.
*
+ * @param writer
* @param startTime
* @param endTime
* @throws IOException
*/
- public void streamLog(Date startTime, Date endTime) throws IOException {
+ public void streamLog(Writer writer, Date startTime, Date endTime) throws IOException {
+ // Get a Reader for the log file(s)
+ BufferedReader reader = makeReader(startTime, endTime);
+ try {
+ // Process the entire logs from the reader using the logFilter
+ new TimestampedMessageParser(reader, logFilter).processRemaining(writer);
+ }
+ finally {
+ reader.close();
+ }
+ writer.flush();
+ }
+
+ /**
+ * Returns a BufferedReader configured to read the log files based on the given startTime and endTime.
+ *
+ * @param startTime
+ * @param endTime
+ * @return A BufferedReader for the log files
+ * @throws IOException
+ */
+ public BufferedReader makeReader(Date startTime, Date endTime) throws IOException {
long startTimeMillis = 0;
long endTimeMillis;
if (startTime != null) {
@@ -219,43 +238,26 @@ public class XLogStreamer {
endTimeMillis = endTime.getTime();
}
File dir = new File(logPath);
- ArrayList<FileInfo> fileList = getFileList(dir, startTimeMillis, endTimeMillis, logRotation, logFile);
- File file;
- String fileName;
- XLogReader logReader;
- for (int i = 0; i < fileList.size(); i++) {
- fileName = fileList.get(i).getFileName();
- if (fileName.endsWith(".gz")) {
- file = new File(fileName);
- GZIPInputStream gzipInputStream = null;
- gzipInputStream = new GZIPInputStream(new FileInputStream(file));
- logReader = new XLogReader(gzipInputStream, logFilter, logWriter);
- logReader.processLog();
- logReader.close();
- continue;
- }
- InputStream ifs;
- ifs = new FileInputStream(fileName);
- logReader = new XLogReader(ifs, logFilter, logWriter);
- logReader.processLog();
- ifs.close();
- }
+ ArrayList<File> files = getFileList(dir, startTimeMillis, endTimeMillis, logRotation, logFile);
+ // The MultiFileReader is a Reader that treats the files as one source so we can easily go through them all with one
+ // BufferedReader
+ return new BufferedReader(new MultiFileReader(files));
}
/**
- * File name along with the modified time which will be used to sort later.
+ * File along with the modified time which will be used to sort later.
*/
- class FileInfo implements Comparable<FileInfo> {
- String fileName;
+ public class FileInfo implements Comparable<FileInfo> {
+ File file;
long modTime;
- public FileInfo(String fileName, long modTime) {
- this.fileName = fileName;
+ public FileInfo(File file, long modTime) {
+ this.file = file;
this.modTime = modTime;
}
- public String getFileName() {
- return fileName;
+ public File getFile() {
+ return file;
}
public long getModTime() {
@@ -286,11 +288,11 @@ public class XLogStreamer {
* @param logFile
* @return List of files to be streamed
*/
- private ArrayList<FileInfo> getFileList(File dir, long startTime, long endTime, long logRotationTime, String logFile) {
+ private ArrayList<File> getFileList(File dir, long startTime, long endTime, long logRotationTime, String logFile) {
String[] children = dir.list();
ArrayList<FileInfo> fileList = new ArrayList<FileInfo>();
if (children == null) {
- return fileList;
+ return new ArrayList<File>();
}
else {
for (int i = 0; i < children.length; i++) {
@@ -302,7 +304,7 @@ public class XLogStreamer {
if (fileName.endsWith(".gz")) {
long gzFileCreationTime = getGZFileCreationTime(fileName, startTime, endTime);
if (gzFileCreationTime != -1) {
- fileList.add(new FileInfo(file.getAbsolutePath(), gzFileCreationTime));
+ fileList.add(new FileInfo(file, gzFileCreationTime));
}
continue;
}
@@ -313,11 +315,15 @@ public class XLogStreamer {
if (modTime / logRotationTime > (endTime / logRotationTime + 1)) {
continue;
}
- fileList.add(new FileInfo(file.getAbsolutePath(), modTime));
+ fileList.add(new FileInfo(file, modTime));
}
}
Collections.sort(fileList);
- return fileList;
+ ArrayList<File> files = new ArrayList<File>(fileList.size());
+ for (FileInfo info : fileList) {
+ files.add(info.getFile());
+ }
+ return files;
}
/**
Added: oozie/trunk/core/src/main/java/org/apache/oozie/util/ZKUtils.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/util/ZKUtils.java?rev=1518739&view=auto
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/util/ZKUtils.java (added)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/util/ZKUtils.java Thu Aug 29 18:06:57 2013
@@ -0,0 +1,269 @@
+/**
+ * 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.oozie.util;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.curator.RetryPolicy;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.apache.curator.utils.EnsurePath;
+import org.apache.curator.x.discovery.ServiceCache;
+import org.apache.curator.x.discovery.ServiceDiscovery;
+import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
+import org.apache.curator.x.discovery.ServiceInstance;
+import org.apache.curator.x.discovery.details.InstanceSerializer;
+import org.apache.oozie.service.Services;
+
+
+/**
+ * This class provides a singleton for interacting with ZooKeeper that other classes can use. It handles connecting to ZooKeeper,
+ * service discovery, and publishing metadata about this server.
+ * <p>
+ * Users of this class should call {@link ZKUtils#register(java.lang.Object)} to obtain the singleton. This will ensure that we're
+ * properly connected and ready to go with ZooKeeper. When the user is done (i.e. on shutdown), it should call
+ * {@link ZKUtils#unregister(java.lang.Object)} to let this class know; once there are no more users, this class will automatically
+ * remove itself from ZooKeeper.
+ * <p>
+ * Each Oozie Server provides metadata that can be shared with the other Oozie Servers. To keep things simple and to make it easy
+ * to add additional metadata in the future, we share a Map. They keys are defined in {@link ZKMetadataKeys}.
+ * <p>
+ * For the service discovery, the structure in ZooKeeper is /oozie.zookeeper.namespace/ZK_BASE_PATH/ (default is /oozie/services/).
+ * There is currently only one service, named "servers" under which each Oozie server creates a ZNode named oozie.zookeeper.oozie.id
+ * (default is the hostname) that contains the metadata payload. For example, with the default settings, an Oozie server named
+ * "foo" would create a ZNode at /oozie/services/servers/foo where the foo ZNode contains the metadata.
+ */
+public class ZKUtils {
+ /**
+ * oozie-site property for specifying the ZooKeeper connection string. Comma-separated values of host:port pairs of the
+ * ZooKeeper servers.
+ */
+ public static final String ZK_CONNECTION_STRING = "oozie.zookeeper.connection.string";
+ /**
+ * oozie-site property for specifying the ZooKeeper namespace to use (e.g. "oozie"). All of the Oozie servers that are planning
+ * on talking to each other should have the same value for this.
+ */
+ public static final String ZK_NAMESPACE = "oozie.zookeeper.namespace";
+ /**
+ * oozie-site property for specifying the ID for this Oozie Server. Each Oozie server should have a unique ID.
+ */
+ public static final String ZK_ID = "oozie.zookeeper.oozie.id";
+ private static final String ZK_OOZIE_SERVICE = "servers";
+ private static final String ZK_BASE_PATH = "/services";
+
+ private static Set<Object> users = new HashSet<Object>();
+ private CuratorFramework client = null;
+ private String zkId;
+ private long zkRegTime;
+ private ServiceDiscovery<Map> sDiscovery;
+ private ServiceCache<Map> sCache;
+ private XLog log;
+
+ private static ZKUtils zk = null;
+
+ /**
+ * Private Constructor for the singleton; it connects to ZooKeeper and advertises this Oozie Server.
+ *
+ * @throws Exception
+ */
+ private ZKUtils() throws Exception {
+ log = XLog.getLog(getClass());
+ zkId = Services.get().getConf().get(ZK_ID, System.getProperty("oozie.http.hostname"));
+ createClient();
+ advertiseService();
+ }
+
+ /**
+ * Classes that want to use ZooKeeper should call this method to get the ZKUtils singleton.
+ *
+ * @param user The calling class
+ * @return the ZKUtils singleton
+ * @throws Exception
+ */
+ public static synchronized ZKUtils register(Object user) throws Exception {
+ if (zk == null) {
+ zk = new ZKUtils();
+ }
+ // Remember the calling class so we can disconnect when everybody is done
+ users.add(user);
+ return zk;
+ }
+
+ /**
+ * Classes should call this when they are done (i.e. shutdown).
+ *
+ * @param user The calling class
+ */
+ public synchronized void unregister(Object user) {
+ // If there are no more classes using ZooKeeper, we should teardown everything.
+ users.remove(user);
+ if (users.isEmpty() && zk != null) {
+ zk.teardown();
+ zk = null;
+ }
+ }
+
+ private void createClient() throws Exception {
+ // Connect to the ZooKeeper server
+ RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
+ String zkConnectionString = Services.get().getConf().get(ZK_CONNECTION_STRING, "localhost:2181");
+ String zkNamespace = Services.get().getConf().get(ZK_NAMESPACE, "oozie");
+ client = CuratorFrameworkFactory.builder()
+ .namespace(zkNamespace)
+ .connectString(zkConnectionString)
+ .retryPolicy(retryPolicy)
+ .build();
+ client.start();
+ }
+
+ private void advertiseService() throws Exception {
+ // Advertise on the service discovery
+ new EnsurePath(ZK_BASE_PATH).ensure(client.getZookeeperClient());
+ InstanceSerializer<Map> instanceSerializer = new FixedJsonInstanceSerializer<Map>(Map.class);
+ sDiscovery = ServiceDiscoveryBuilder.builder(Map.class)
+ .basePath(ZK_BASE_PATH)
+ .client(client)
+ .serializer(instanceSerializer)
+ .build();
+ sDiscovery.start();
+ sDiscovery.registerService(getMetadataInstance());
+
+ // Create the service discovery cache
+ sCache = sDiscovery.serviceCacheBuilder().name(ZK_OOZIE_SERVICE).build();
+ sCache.start();
+
+ zkRegTime = sDiscovery.queryForInstance(ZK_OOZIE_SERVICE, zkId).getRegistrationTimeUTC();
+ }
+
+ private void unadvertiseService() throws Exception {
+ // Stop the service discovery cache
+ sCache.close();
+
+ // Unadvertise on the service discovery
+ sDiscovery.unregisterService(getMetadataInstance());
+ sDiscovery.close();
+ }
+
+ private void teardown() {
+ try {
+ zk.unadvertiseService();
+ }
+ catch (Exception ex) {
+ log.warn("Exception occurred while unadvertising: " + ex.getMessage(), ex);
+ }
+ client.close();
+ client = null;
+ }
+
+ private ServiceInstance<Map> getMetadataInstance() throws Exception {
+ // Creates the metadata that this server is providing to ZooKeeper and other Oozie Servers
+ String url = ConfigUtils.getOozieEffectiveUrl();
+ Map<String, String> map = new HashMap<String, String>();
+ map.put(ZKMetadataKeys.OOZIE_ID, zkId);
+ map.put(ZKMetadataKeys.OOZIE_URL, url);
+
+ return ServiceInstance.<Map>builder()
+ .name(ZK_OOZIE_SERVICE)
+ .id(zkId)
+ .payload(map)
+ .build();
+ }
+
+ /**
+ * Returns a list of the metadata provided by all of the Oozie Servers. Note that the metadata is cached so it may be a second
+ * or two stale.
+ *
+ * @return a List of the metadata provided by all of the Oozie Servers.
+ * @throws Exception
+ */
+ public List<ServiceInstance<Map>> getAllMetaData() {
+ List<ServiceInstance<Map>> instances = null;
+ if (sCache != null) {
+ instances = sCache.getInstances();
+ }
+ return instances;
+ }
+
+ /**
+ * Returns the ID of this Oozie Server as seen by ZooKeeper and other Oozie Servers
+ *
+ * @return the ID of this Oozie Server
+ */
+ public String getZKId() {
+ return zkId;
+ }
+
+ /**
+ * Returns the {@link CuratorFramework} used for managing the ZooKeeper connection; it can be used by calling classes to perform
+ * more direct operations on ZooKeeper. Most of the time, this shouldn't be needed.
+ * <p>
+ * Be careful not to close the connection.
+ *
+ * @return the CuratorFramework object
+ */
+ public CuratorFramework getClient() {
+ return client;
+ }
+
+ /**
+ * Returns the index of this Oozie Server in ZooKeeper's list of Oozie Servers (ordered by registration time)
+ *
+ * @param oozies The collection of metadata provided by all of the Oozie Servers (from calling {@link ZKUtils#getAllMetaData())
+ * @return the index of this Oozie Server in ZooKeeper's list of Oozie Servers (ordered by registration time)
+ */
+ public int getZKIdIndex(List<ServiceInstance<Map>> oozies) {
+ int index = 0;
+ // We don't actually have to sort all of the IDs, we can simply find out how many are before our zkId
+ for (ServiceInstance<Map> oozie : oozies) {
+ long otherRegTime = oozie.getRegistrationTimeUTC();
+ if (otherRegTime < zkRegTime) {
+ index++;
+ }
+ }
+ return index;
+ }
+
+ /**
+ * Useful for tests to get the registered classes
+ *
+ * @return the set of registered classes
+ */
+ @VisibleForTesting
+ public static Set<Object> getUsers() {
+ return users;
+ }
+
+ /**
+ * Keys used in the metadata provided by each Oozie Server to ZooKeeper and other Oozie Servers
+ */
+ public abstract class ZKMetadataKeys {
+ /**
+ * The ID of the Oozie Server
+ */
+ public static final String OOZIE_ID = "OOZIE_ID";
+ /**
+ * The URL of the Oozie Server
+ */
+ public static final String OOZIE_URL = "OOZIE_URL";
+ }
+}
Modified: oozie/trunk/core/src/main/resources/oozie-default.xml
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/resources/oozie-default.xml?rev=1518739&r1=1518738&r2=1518739&view=diff
==============================================================================
--- oozie/trunk/core/src/main/resources/oozie-default.xml (original)
+++ oozie/trunk/core/src/main/resources/oozie-default.xml Thu Aug 29 18:06:57 2013
@@ -126,7 +126,9 @@
org.apache.oozie.service.StatusTransitService,
org.apache.oozie.service.PauseTransitService,
org.apache.oozie.service.GroupsService,
- org.apache.oozie.service.ProxyUserService
+ org.apache.oozie.service.ProxyUserService,
+ org.apache.oozie.service.XLogStreamingService,
+ org.apache.oozie.service.JobsConcurrencyService
</value>
<description>
All services to be created and managed by Oozie Services singleton.
@@ -1926,4 +1928,31 @@
</description>
</property>
+ <!-- ZooKeeper configuration -->
+ <property>
+ <name>oozie.zookeeper.connection.string</name>
+ <value>localhost:2181</value>
+ <description>
+ Comma-separated values of host:port pairs of the ZooKeeper servers.
+ </description>
+ </property>
+
+ <property>
+ <name>oozie.zookeeper.namespace</name>
+ <value>oozie</value>
+ <description>
+ The namespace to use. All of the Oozie Servers that are planning on talking to each other should have the same
+ namespace.
+ </description>
+ </property>
+
+ <property>
+ <name>oozie.zookeeper.oozie.id</name>
+ <value> </value>
+ <description>
+ The ID for the Oozie server to use. Each Oozie server must have a unique ID.
+ If blank, Oozie will use the hostname.
+ </description>
+ </property>
+
</configuration>
Modified: oozie/trunk/core/src/test/java/org/apache/oozie/TestCoordinatorEngineStreamLog.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/test/java/org/apache/oozie/TestCoordinatorEngineStreamLog.java?rev=1518739&r1=1518738&r2=1518739&view=diff
==============================================================================
--- oozie/trunk/core/src/test/java/org/apache/oozie/TestCoordinatorEngineStreamLog.java (original)
+++ oozie/trunk/core/src/test/java/org/apache/oozie/TestCoordinatorEngineStreamLog.java Thu Aug 29 18:06:57 2013
@@ -26,7 +26,9 @@ import java.io.StringWriter;
import java.io.Writer;
import java.net.URI;
import java.util.Date;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -36,7 +38,7 @@ import org.apache.oozie.client.OozieClie
import org.apache.oozie.client.rest.RestConstants;
import org.apache.oozie.service.DagXLogInfoService;
import org.apache.oozie.service.Services;
-import org.apache.oozie.service.XLogService;
+import org.apache.oozie.service.XLogStreamingService;
import org.apache.oozie.test.XFsTestCase;
import org.apache.oozie.util.DateUtils;
import org.apache.oozie.util.IOUtils;
@@ -77,11 +79,12 @@ public class TestCoordinatorEngineStream
}
}
- static class DummyXLogService extends XLogService {
+ static class DummyXLogStreamingService extends XLogStreamingService {
Filter filter;
@Override
- public void streamLog(Filter filter1, Date startTime, Date endTime, Writer writer) throws IOException {
+ public void streamLog(Filter filter1, Date startTime, Date endTime, Writer writer, Map<String, String[]> params)
+ throws IOException {
filter = filter1;
}
}
@@ -100,9 +103,9 @@ public class TestCoordinatorEngineStream
public void testStreamLog2() throws Exception {
CoordinatorEngine ce = createCoordinatorEngine();
String jobId = runJobsImpl(ce);
- ce.streamLog(jobId, new StringWriter()/* writer is unused */);
+ ce.streamLog(jobId, new StringWriter(), new HashMap<String, String[]>());
- DummyXLogService service = (DummyXLogService) services.get(XLogService.class);
+ DummyXLogStreamingService service = (DummyXLogStreamingService) services.get(XLogStreamingService.class);
Filter filter = service.filter;
assertEquals(filter.getFilterParams().get(DagXLogInfoService.JOB), jobId);
@@ -115,9 +118,9 @@ public class TestCoordinatorEngineStream
public void testStreamLog4NullNull() throws Exception {
CoordinatorEngine ce = createCoordinatorEngine();
String jobId = runJobsImpl(ce);
- ce.streamLog(jobId, null, null, new StringWriter()/* writer is unused */);
+ ce.streamLog(jobId, null, null, new StringWriter(), new HashMap<String, String[]>());
- DummyXLogService service = (DummyXLogService) services.get(XLogService.class);
+ DummyXLogStreamingService service = (DummyXLogStreamingService) services.get(XLogStreamingService.class);
Filter filter = service.filter;
assertEquals(filter.getFilterParams().get(DagXLogInfoService.JOB), jobId);
@@ -132,9 +135,9 @@ public class TestCoordinatorEngineStream
CoordinatorEngine ce = createCoordinatorEngine();
String jobId = runJobsImpl(ce);
- ce.streamLog(jobId, "678, 123-127, 946", RestConstants.JOB_LOG_ACTION, new StringWriter()/* unused */);
+ ce.streamLog(jobId, "678, 123-127, 946", RestConstants.JOB_LOG_ACTION, new StringWriter(), new HashMap<String, String[]>());
- DummyXLogService service = (DummyXLogService) services.get(XLogService.class);
+ DummyXLogStreamingService service = (DummyXLogStreamingService) services.get(XLogStreamingService.class);
Filter filter = service.filter;
assertEquals(jobId, filter.getFilterParams().get(DagXLogInfoService.JOB));
@@ -160,9 +163,9 @@ public class TestCoordinatorEngineStream
ce.streamLog(jobId, DateUtils.formatDateOozieTZ(createdDate) + "::" + DateUtils.formatDateOozieTZ(middleDate) + ","
+ DateUtils.formatDateOozieTZ(middleDate) + "::" + DateUtils.formatDateOozieTZ(endDate),
- RestConstants.JOB_LOG_DATE, new StringWriter()/* unused */);
+ RestConstants.JOB_LOG_DATE, new StringWriter(), new HashMap<String, String[]>());
- DummyXLogService service = (DummyXLogService) services.get(XLogService.class);
+ DummyXLogStreamingService service = (DummyXLogStreamingService) services.get(XLogStreamingService.class);
Filter filter = service.filter;
assertEquals(jobId, filter.getFilterParams().get(DagXLogInfoService.JOB));
@@ -171,10 +174,9 @@ public class TestCoordinatorEngineStream
}
private String runJobsImpl(final CoordinatorEngine ce) throws Exception {
- services.setService(DummyXLogService.class);
+ services.setService(DummyXLogStreamingService.class);
// need to re-define the parameters that are cleared upon the service reset:
new DagXLogInfoService().init(services);
- services.init();
Configuration conf = new XConfiguration();
Modified: oozie/trunk/core/src/test/java/org/apache/oozie/command/TestXCommand.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/test/java/org/apache/oozie/command/TestXCommand.java?rev=1518739&r1=1518738&r2=1518739&view=diff
==============================================================================
--- oozie/trunk/core/src/test/java/org/apache/oozie/command/TestXCommand.java (original)
+++ oozie/trunk/core/src/test/java/org/apache/oozie/command/TestXCommand.java Thu Aug 29 18:06:57 2013
@@ -21,7 +21,7 @@ import org.apache.oozie.ErrorCode;
import org.apache.oozie.service.MemoryLocksService;
import org.apache.oozie.service.Services;
import org.apache.oozie.test.XTestCase;
-import org.apache.oozie.util.MemoryLocks;
+import org.apache.oozie.lock.LockToken;
public class TestXCommand extends XTestCase {
@@ -228,7 +228,7 @@ public class TestXCommand extends XTestC
@Override
public void run() {
try {
- MemoryLocks.LockToken lock = Services.get().get(MemoryLocksService.class).getWriteLock("key", 1);
+ LockToken lock = Services.get().get(MemoryLocksService.class).getWriteLock("key", 1);
if (lock == null) {
fail();
}
Copied: oozie/trunk/core/src/test/java/org/apache/oozie/lock/TestMemoryLocks.java (from r1518716, oozie/trunk/core/src/test/java/org/apache/oozie/util/TestMemoryLocks.java)
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/test/java/org/apache/oozie/lock/TestMemoryLocks.java?p2=oozie/trunk/core/src/test/java/org/apache/oozie/lock/TestMemoryLocks.java&p1=oozie/trunk/core/src/test/java/org/apache/oozie/util/TestMemoryLocks.java&r1=1518716&r2=1518739&rev=1518739&view=diff
==============================================================================
--- oozie/trunk/core/src/test/java/org/apache/oozie/util/TestMemoryLocks.java (original)
+++ oozie/trunk/core/src/test/java/org/apache/oozie/lock/TestMemoryLocks.java Thu Aug 29 18:06:57 2013
@@ -6,18 +6,19 @@
* 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.oozie.util;
+package org.apache.oozie.lock;
import org.apache.oozie.test.XTestCase;
+import org.apache.oozie.util.XLog;
public class TestMemoryLocks extends XTestCase {
private XLog log = XLog.getLog(getClass());
@@ -51,7 +52,7 @@ public class TestMemoryLocks extends XTe
public void run() {
try {
log.info("Getting lock [{0}]", nameIndex);
- MemoryLocks.LockToken token = getLock();
+ MemoryLocks.MemoryLockToken token = getLock();
if (token != null) {
log.info("Got lock [{0}]", nameIndex);
sb.append(nameIndex + "-L ");
@@ -78,7 +79,7 @@ public class TestMemoryLocks extends XTe
}
}
- protected abstract MemoryLocks.LockToken getLock() throws InterruptedException;
+ protected abstract MemoryLocks.MemoryLockToken getLock() throws InterruptedException;
}
@@ -89,7 +90,7 @@ public class TestMemoryLocks extends XTe
super(name, nameIndex, timeout, buffer);
}
- protected MemoryLocks.LockToken getLock() throws InterruptedException {
+ protected MemoryLocks.MemoryLockToken getLock() throws InterruptedException {
return locks.getReadLock(name, timeout);
}
}
@@ -100,7 +101,7 @@ public class TestMemoryLocks extends XTe
super(name, nameIndex, timeout, buffer);
}
- protected MemoryLocks.LockToken getLock() throws InterruptedException {
+ protected MemoryLocks.MemoryLockToken getLock() throws InterruptedException {
return locks.getWriteLock(name, timeout);
}
}
Added: oozie/trunk/core/src/test/java/org/apache/oozie/service/DummyLogStreamingServlet.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/test/java/org/apache/oozie/service/DummyLogStreamingServlet.java?rev=1518739&view=auto
==============================================================================
--- oozie/trunk/core/src/test/java/org/apache/oozie/service/DummyLogStreamingServlet.java (added)
+++ oozie/trunk/core/src/test/java/org/apache/oozie/service/DummyLogStreamingServlet.java Thu Aug 29 18:06:57 2013
@@ -0,0 +1,44 @@
+/**
+ * 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.oozie.service;
+
+import java.io.IOException;
+import java.io.Writer;
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+/**
+ * Used by {@link TestZKXLogStreamingService#testStreamingWithMultipleOozieServers() } to stream logs from another Oozie "server".
+ * Due to the way the servlet gets loaded, it has to be its own class instead of an inner class.
+ */
+public class DummyLogStreamingServlet extends HttpServlet {
+
+ public static String lastQueryString = null;
+ public static String logs = null;
+
+ @Override
+ protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
+ lastQueryString = request.getQueryString();
+ response.setStatus(HttpServletResponse.SC_OK);
+ Writer writer = response.getWriter();
+ writer.append(logs);
+ writer.close();
+ }
+}
Added: oozie/trunk/core/src/test/java/org/apache/oozie/service/TestJobsConcurrencyService.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/test/java/org/apache/oozie/service/TestJobsConcurrencyService.java?rev=1518739&view=auto
==============================================================================
--- oozie/trunk/core/src/test/java/org/apache/oozie/service/TestJobsConcurrencyService.java (added)
+++ oozie/trunk/core/src/test/java/org/apache/oozie/service/TestJobsConcurrencyService.java Thu Aug 29 18:06:57 2013
@@ -0,0 +1,90 @@
+/**
+ * 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.oozie.service;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.oozie.test.XTestCase;
+import org.apache.oozie.util.ConfigUtils;
+
+public class TestJobsConcurrencyService extends XTestCase {
+
+ @Override
+ protected void setUp() throws Exception {
+ super.setUp();
+ }
+
+ @Override
+ protected void tearDown() throws Exception {
+ super.tearDown();
+ }
+
+ public void testIsFirstServer() throws Exception {
+ JobsConcurrencyService jcs = new JobsConcurrencyService();
+ try {
+ jcs.init(null);
+ assertTrue(jcs.isFirstServer());
+ }
+ finally {
+ jcs.destroy();
+ }
+ }
+
+ public void testIsJobIdForThisServer() throws Exception {
+ JobsConcurrencyService jcs = new JobsConcurrencyService();
+ try {
+ jcs.init(null);
+ assertTrue(jcs.isJobIdForThisServer("blah"));
+ }
+ finally {
+ jcs.destroy();
+ }
+ }
+
+ public void testGetJobIdsForThisServer() throws Exception {
+ JobsConcurrencyService jcs = new JobsConcurrencyService();
+ try {
+ jcs.init(null);
+ List<String> ids = new ArrayList<String>();
+ ids.add("blah");
+ ids.add("0000002-130521183438837-oozie-rkan-W");
+ ids.add("0000001-130521155209569-oozie-rkan-W");
+ List<String> ids2 = jcs.getJobIdsForThisServer(ids);
+ assertEquals(ids.size(), ids2.size());
+ assertTrue(ids2.containsAll(ids));
+ }
+ finally {
+ jcs.destroy();
+ }
+ }
+
+ public void testGetServerUrls() throws Exception {
+ JobsConcurrencyService jcs = new JobsConcurrencyService();
+ try {
+ jcs.init(null);
+ Map<String, String> map = jcs.getServerUrls();
+ assertEquals(1, map.size());
+ assertEquals(System.getProperty("oozie.http.hostname"), map.keySet().iterator().next());
+ assertEquals(ConfigUtils.getOozieURL(false), map.get(System.getProperty("oozie.http.hostname")));
+ }
+ finally {
+ jcs.destroy();
+ }
+ }
+}