You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@falcon.apache.org by ve...@apache.org on 2014/03/06 01:49:37 UTC
[1/6] git commit: FALCON-336 FALCON-331 introduced a check style
error - builds fail. Contributed by Venkatesh Seetharam
Repository: incubator-falcon
Updated Branches:
refs/heads/master 464442af2 -> b8b2fe45d
FALCON-336 FALCON-331 introduced a check style error - builds fail. Contributed by Venkatesh Seetharam
Project: http://git-wip-us.apache.org/repos/asf/incubator-falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-falcon/commit/42d04f37
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/42d04f37
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/42d04f37
Branch: refs/heads/master
Commit: 42d04f37feaeee26fd6239dfa275173660a331f9
Parents: 464442a
Author: Venkatesh Seetharam <ve...@hortonworks.com>
Authored: Wed Mar 5 15:45:15 2014 -0800
Committer: Venkatesh Seetharam <ve...@hortonworks.com>
Committed: Wed Mar 5 15:45:15 2014 -0800
----------------------------------------------------------------------
CHANGES.txt | 2 ++
.../org/apache/falcon/workflow/engine/OozieWorkflowEngine.java | 4 ++--
2 files changed, 4 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/42d04f37/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 237646c..6b536b8 100755
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -118,6 +118,8 @@ Trunk (Unreleased)
FALCON-226 Tests for subscribing/listening to multiple active-mq topics. (Shaik Idris
via Shwetha GS)
+ FALCON-336 Fix check style error - builds fail (Venkatesh Seetharam)
+
Release Version: 0.4-incubating
NEW FEATURES
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/42d04f37/oozie/src/main/java/org/apache/falcon/workflow/engine/OozieWorkflowEngine.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/workflow/engine/OozieWorkflowEngine.java b/oozie/src/main/java/org/apache/falcon/workflow/engine/OozieWorkflowEngine.java
index 328cd82..dee77c0 100644
--- a/oozie/src/main/java/org/apache/falcon/workflow/engine/OozieWorkflowEngine.java
+++ b/oozie/src/main/java/org/apache/falcon/workflow/engine/OozieWorkflowEngine.java
@@ -564,8 +564,8 @@ public class OozieWorkflowEngine extends AbstractWorkflowEngine {
Frequency freq = createFrequency(String.valueOf(coord.getFrequency()), coord.getTimeUnit());
TimeZone tz = EntityUtil.getTimeZone(coord.getTimeZone());
Date iterStart = EntityUtil.getNextStartTime(coord.getStartTime(), freq, tz, start);
- Date iterEnd = (coord.getLastActionTime() != null && coord.getLastActionTime().before(end) ?
- coord.getLastActionTime() : end);
+ Date iterEnd = (coord.getLastActionTime() != null && coord.getLastActionTime().before(end)
+ ? coord.getLastActionTime() : end);
if (i == (applicableCoords.size() - 1)) {
isLastCoord = true;
[6/6] git commit: FALCON-289 Provide REST APIs for discovering
lineage metadata over the store. Contributed by Venkatesh Seetharam
Posted by ve...@apache.org.
FALCON-289 Provide REST APIs for discovering lineage metadata over the store. Contributed by Venkatesh Seetharam
Project: http://git-wip-us.apache.org/repos/asf/incubator-falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-falcon/commit/b8b2fe45
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/b8b2fe45
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/b8b2fe45
Branch: refs/heads/master
Commit: b8b2fe45dc480223af1bc70ed13e70ab82b60053
Parents: ecb919c
Author: Venkatesh Seetharam <ve...@hortonworks.com>
Authored: Wed Mar 5 16:47:37 2014 -0800
Committer: Venkatesh Seetharam <ve...@hortonworks.com>
Committed: Wed Mar 5 16:47:37 2014 -0800
----------------------------------------------------------------------
CHANGES.txt | 5 +
.../java/org/apache/falcon/cli/FalconCLI.java | 123 ++++++
.../org/apache/falcon/client/FalconClient.java | 74 ++++
.../metadata/LineageMetadataResource.java | 433 +++++++++++++++++++
prism/src/main/webapp/WEB-INF/web.xml | 2 +-
process/pom.xml | 10 -
webapp/src/main/webapp/WEB-INF/embedded/web.xml | 2 +-
webapp/src/main/webapp/WEB-INF/web.xml | 2 +-
.../org/apache/falcon/cli/FalconCLISmokeIT.java | 10 +
9 files changed, 648 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b8b2fe45/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 931a868..fdb9490 100755
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -18,6 +18,11 @@ Trunk (Unreleased)
FALCON-254 Bootstrap designer module. (Srikanth Sundarrajan via Shwetha GS)
FALCON-238 Support updates at specific time. (Shwetha GS)
+
+ FALCON-285 Support Lineage information capture (Venkatesh Seetharam)
+
+ FALCON-289 Provide REST APIs for discovering lineage metadata over the store.
+ (Venkatesh Seetharam)
IMPROVEMENTS
FALCON-333 jsp-api dependency is defined twice. (Jean-Baptiste
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b8b2fe45/client/src/main/java/org/apache/falcon/cli/FalconCLI.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/falcon/cli/FalconCLI.java b/client/src/main/java/org/apache/falcon/cli/FalconCLI.java
index a414e32..1663073 100644
--- a/client/src/main/java/org/apache/falcon/cli/FalconCLI.java
+++ b/client/src/main/java/org/apache/falcon/cli/FalconCLI.java
@@ -89,6 +89,20 @@ public class FalconCLI {
public static final String CURRENT_COLO = "current.colo";
public static final String CLIENT_PROPERTIES = "/client.properties";
+ // Graph Commands
+ public static final String GRAPH_CMD = "graph";
+ public static final String VERTEX_CMD = "vertex";
+ public static final String VERTICES_CMD = "vertices";
+ public static final String VERTEX_EDGES_CMD = "edges";
+
+ // Graph Command Options
+ public static final String EDGE_CMD = "edge";
+ public static final String ID_OPT = "id";
+ public static final String NAME_OPT = "name";
+ public static final String VALUE_OPT = "value";
+ public static final String DIRECTION_OPT = "direction";
+ public static final String DUMP_OPT = "all";
+
/**
* Entry point for the Falcon CLI when invoked from the command line. Upon
* completion this method exits the JVM with '0' (success) or '-1'
@@ -132,6 +146,7 @@ public class FalconCLI {
"",
"Process instances operations like running, status, kill, suspend, resume, rerun, logs",
instanceOptions(), false);
+ parser.addCommand(GRAPH_CMD, "", "graph operations", createGraphOptions(), true);
try {
CLIParser.Command command = parser.parse(args);
@@ -149,6 +164,8 @@ public class FalconCLI {
entityCommand(commandLine, client);
} else if (command.getName().equals(INSTANCE_CMD)) {
instanceCommand(commandLine, client);
+ } else if (command.getName().equals(GRAPH_CMD)) {
+ graphCommand(commandLine, client);
}
}
@@ -572,6 +589,112 @@ public class FalconCLI {
return instanceOptions;
}
+ private Options createGraphOptions() {
+ Options graphOptions = new Options();
+ Option url = new Option(URL_OPTION, true, "Falcon URL");
+ graphOptions.addOption(url);
+
+ Option vertex = new Option(VERTEX_CMD, false, "show the vertices");
+ Option vertices = new Option(VERTICES_CMD, false, "show the vertices");
+ Option vertexEdges = new Option(VERTEX_EDGES_CMD, false, "show the edges for a given vertex");
+ Option edges = new Option(EDGE_CMD, false, "show the edges");
+
+ OptionGroup group = new OptionGroup();
+ group.addOption(vertex);
+ group.addOption(vertices);
+ group.addOption(vertexEdges);
+ group.addOption(edges);
+ graphOptions.addOptionGroup(group);
+
+ Option id = new Option(ID_OPT, true, "vertex or edge id");
+ graphOptions.addOption(id);
+
+ Option name = new Option(NAME_OPT, true, "name property");
+ graphOptions.addOption(name);
+
+ Option value = new Option(VALUE_OPT, true, "value property");
+ graphOptions.addOption(value);
+
+ Option direction = new Option(DIRECTION_OPT, true, "edge direction property");
+ graphOptions.addOption(direction);
+
+ Option dump = new Option(DUMP_OPT, false, "dump all elements");
+ graphOptions.addOption(dump);
+
+ return graphOptions;
+ }
+
+ private void graphCommand(CommandLine commandLine,
+ FalconClient client) throws FalconCLIException {
+ Set<String> optionsList = new HashSet<String>();
+ for (Option option : commandLine.getOptions()) {
+ optionsList.add(option.getOpt());
+ }
+
+ String result;
+ String id = commandLine.getOptionValue(ID_OPT);
+ String name = commandLine.getOptionValue(NAME_OPT);
+ String value = commandLine.getOptionValue(VALUE_OPT);
+ String direction = commandLine.getOptionValue(DIRECTION_OPT);
+ String dump = commandLine.getOptionValue(DUMP_OPT);
+
+ if (optionsList.contains(VERTEX_CMD)) {
+ validateId(id);
+ result = client.getVertex(id);
+ } else if (optionsList.contains(VERTICES_CMD)) {
+ if (isDump(dump)) {
+ result = client.getVertices();
+ } else {
+ validateVerticesCommand(name, value);
+ result = client.getVertices(name, value);
+ }
+ } else if (optionsList.contains(VERTEX_EDGES_CMD)) {
+ if (isDump(dump)) {
+ result = client.getEdges();
+ } else {
+ validateVertexEdgesCommand(id, direction);
+ result = client.getVertexEdges(id, direction);
+ }
+ } else if (optionsList.contains(EDGE_CMD)) {
+ validateId(id);
+ result = client.getEdge(id);
+ } else {
+ throw new FalconCLIException("Invalid command");
+ }
+
+ OUT.get().println(result);
+ }
+
+ private void validateId(String id) throws FalconCLIException {
+ if (id == null || id.length() == 0) {
+ throw new FalconCLIException("Missing argument: id");
+ }
+ }
+
+ private boolean isDump(String dump) {
+ return dump != null;
+ }
+
+ private void validateVerticesCommand(String name, String value) throws FalconCLIException {
+ if (name == null || name.length() == 0) {
+ throw new FalconCLIException("Missing argument: name");
+ }
+
+ if (value == null || value.length() == 0) {
+ throw new FalconCLIException("Missing argument: value");
+ }
+ }
+
+ private void validateVertexEdgesCommand(String id, String direction) throws FalconCLIException {
+ if (id == null || id.length() == 0) {
+ throw new FalconCLIException("Missing argument: id");
+ }
+
+ if (direction == null || direction.length() == 0) {
+ throw new FalconCLIException("Missing argument: direction");
+ }
+ }
+
protected String getFalconEndpoint(CommandLine commandLine) throws FalconCLIException, IOException {
String url = commandLine.getOptionValue(URL_OPTION);
if (url == null) {
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b8b2fe45/client/src/main/java/org/apache/falcon/client/FalconClient.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/falcon/client/FalconClient.java b/client/src/main/java/org/apache/falcon/client/FalconClient.java
index 59b2a3b..f008953 100644
--- a/client/src/main/java/org/apache/falcon/client/FalconClient.java
+++ b/client/src/main/java/org/apache/falcon/client/FalconClient.java
@@ -736,6 +736,80 @@ public class FalconClient {
return sb.toString();
}
+ protected static enum GraphOperations {
+
+ VERTICES("api/graphs/lineage/vertices", HttpMethod.GET, MediaType.APPLICATION_JSON),
+ EDGES("api/graphs/lineage/edges", HttpMethod.GET, MediaType.APPLICATION_JSON);
+
+ private String path;
+ private String method;
+ private String mimeType;
+
+ GraphOperations(String path, String method, String mimeType) {
+ this.path = path;
+ this.method = method;
+ this.mimeType = mimeType;
+ }
+ }
+
+ public String getVertex(String id) throws FalconCLIException {
+ return sendGraphRequest(GraphOperations.VERTICES, id);
+ }
+
+ public String getVertices(String key, String value) throws FalconCLIException {
+ return sendGraphRequest(GraphOperations.VERTICES, key, value);
+ }
+
+ public String getVertexEdges(String id, String direction) throws FalconCLIException {
+ return sendGraphRequestForEdges(GraphOperations.VERTICES, id, direction);
+ }
+
+ public String getEdge(String id) throws FalconCLIException {
+ return sendGraphRequest(GraphOperations.EDGES, id);
+ }
+
+ public String getVertices() throws FalconCLIException {
+ return sendGraphRequest(GraphOperations.VERTICES, "all");
+ }
+
+ public String getEdges() throws FalconCLIException {
+ return sendGraphRequest(GraphOperations.EDGES, "all");
+ }
+
+ private String sendGraphRequest(GraphOperations job, String id) throws FalconCLIException {
+ ClientResponse clientResponse = service.path(job.path)
+ .path(id)
+ .header("Cookie", AUTH_COOKIE_EQ + authenticationToken)
+ .accept(job.mimeType)
+ .type(job.mimeType)
+ .method(job.method, ClientResponse.class);
+ return parseStringResult(clientResponse);
+ }
+
+ private String sendGraphRequest(GraphOperations job, String name,
+ String value) throws FalconCLIException {
+ ClientResponse clientResponse = service.path(job.path)
+ .queryParam("name", name)
+ .queryParam("value", value)
+ .header("Cookie", AUTH_COOKIE_EQ + authenticationToken)
+ .accept(job.mimeType)
+ .type(job.mimeType)
+ .method(job.method, ClientResponse.class);
+ return parseStringResult(clientResponse);
+ }
+
+ private String sendGraphRequestForEdges(GraphOperations job, String id,
+ String direction) throws FalconCLIException {
+ ClientResponse clientResponse = service.path(job.path)
+ .path(id)
+ .path(direction)
+ .header("Cookie", AUTH_COOKIE_EQ + authenticationToken)
+ .accept(job.mimeType)
+ .type(job.mimeType)
+ .method(job.method, ClientResponse.class);
+ return parseStringResult(clientResponse);
+ }
+
private void checkIfSuccessful(ClientResponse clientResponse)
throws FalconCLIException {
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b8b2fe45/prism/src/main/java/org/apache/falcon/resource/metadata/LineageMetadataResource.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/resource/metadata/LineageMetadataResource.java b/prism/src/main/java/org/apache/falcon/resource/metadata/LineageMetadataResource.java
new file mode 100644
index 0000000..f140bca
--- /dev/null
+++ b/prism/src/main/java/org/apache/falcon/resource/metadata/LineageMetadataResource.java
@@ -0,0 +1,433 @@
+/**
+ * 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.falcon.resource.metadata;
+
+import com.tinkerpop.blueprints.Direction;
+import com.tinkerpop.blueprints.Edge;
+import com.tinkerpop.blueprints.KeyIndexableGraph;
+import com.tinkerpop.blueprints.Vertex;
+import com.tinkerpop.blueprints.VertexQuery;
+import com.tinkerpop.blueprints.util.io.graphson.GraphSONMode;
+import com.tinkerpop.blueprints.util.io.graphson.GraphSONUtility;
+import org.apache.falcon.metadata.GraphUtils;
+import org.apache.falcon.metadata.MetadataMappingService;
+import org.apache.falcon.service.Services;
+import org.apache.falcon.util.StartupProperties;
+import org.apache.log4j.Logger;
+import org.codehaus.jettison.json.JSONArray;
+import org.codehaus.jettison.json.JSONException;
+import org.codehaus.jettison.json.JSONObject;
+
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import java.util.Set;
+
+/**
+ * Jersey Resource for lineage metadata operations.
+ */
+@Path("graphs/lineage")
+public class LineageMetadataResource {
+
+ private static final Logger LOG = Logger.getLogger(LineageMetadataResource.class);
+
+ public static final String RESULTS = "results";
+ public static final String TOTAL_SIZE = "totalSize";
+
+ private final MetadataMappingService service;
+
+ public LineageMetadataResource() {
+ if (Services.get().isRegistered(MetadataMappingService.SERVICE_NAME)) {
+ service = Services.get().getService(MetadataMappingService.SERVICE_NAME);
+ } else {
+ service = null;
+ }
+ }
+
+ private KeyIndexableGraph getGraph() {
+ return service.getGraph();
+ }
+
+ private Set<String> getVertexIndexedKeys() {
+ return service.getVertexIndexedKeys();
+ }
+
+ private Set<String> getEdgeIndexedKeys() {
+ return service.getEdgeIndexedKeys();
+ }
+
+ /**
+ * Dump the graph.
+ *
+ * GET http://host/graphs/lineage/serialize
+ * graph.getVertices();
+ */
+ @GET
+ @Path("/serialize")
+ @Produces({MediaType.APPLICATION_JSON})
+ public Response serializeGraph() {
+ checkIfMetadataMappingServiceIsEnabled();
+ String file = StartupProperties.get().getProperty("falcon.graph.serialize.path")
+ + "lineage-graph-" + System.currentTimeMillis() + ".json";
+ LOG.info("Serialize Graph to: " + file);
+ try {
+ GraphUtils.dump(getGraph(), file);
+ return Response.ok().build();
+ } catch (Exception e) {
+ throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
+ .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ }
+ }
+
+ /**
+ * Get all vertices.
+ *
+ * GET http://host/graphs/lineage/vertices/all
+ * graph.getVertices();
+ */
+ @GET
+ @Path("/vertices/all")
+ @Produces({MediaType.APPLICATION_JSON})
+ public Response getVertices() {
+ checkIfMetadataMappingServiceIsEnabled();
+ LOG.info("Get All Vertices");
+ try {
+ JSONArray vertexArray = new JSONArray();
+ long counter = 0;
+ for (Vertex vertex : getGraph().getVertices()) {
+ counter++;
+ vertexArray.put(GraphSONUtility.jsonFromElement(
+ vertex, getVertexIndexedKeys(), GraphSONMode.NORMAL));
+ }
+
+ JSONObject response = new JSONObject();
+ response.put(RESULTS, vertexArray);
+ response.put(TOTAL_SIZE, counter);
+ return Response.ok(response).build();
+ } catch (JSONException e) {
+ throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
+ .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ }
+ }
+
+ /**
+ * Get a single vertex with a unique id.
+ *
+ * GET http://host/graphs/lineage/vertices/id
+ * graph.getVertex(id);
+ */
+ @GET
+ @Path("/vertices/{id}")
+ @Produces({MediaType.APPLICATION_JSON})
+ public Response getVertex(@PathParam("id") final String vertexId) {
+ checkIfMetadataMappingServiceIsEnabled();
+ LOG.info("Get vertex for vertexId= " + vertexId);
+ try {
+ Vertex vertex = getGraph().getVertex(vertexId);
+ if (vertex == null) {
+ String message = "Vertex with [" + vertexId + "] cannot be found.";
+ LOG.info(message);
+ throw new WebApplicationException(Response.status(Response.Status.NOT_FOUND)
+ .entity(JSONObject.quote(message)).build());
+ }
+
+ JSONObject response = new JSONObject();
+ response.put(RESULTS, GraphSONUtility.jsonFromElement(
+ vertex, getVertexIndexedKeys(), GraphSONMode.NORMAL));
+ return Response.ok(response).build();
+ } catch (JSONException e) {
+ throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
+ .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ }
+ }
+
+ /**
+ * Get a list of vertices matching a property name and a value.
+ * <p/>
+ * GET http://host/graphs/lineage/vertices?name=<name>&value=<value>
+ * graph.getVertices(name, value);
+ */
+ @GET
+ @Path("/vertices")
+ @Produces({MediaType.APPLICATION_JSON})
+ public Response getVertices(@QueryParam("name") final String name,
+ @QueryParam("value") final String value) {
+ checkIfMetadataMappingServiceIsEnabled();
+ LOG.info("Get vertices for property name= " + name + ", value= " + value);
+ try {
+ Iterable<Vertex> vertices = getGraph().getVertices(name, value);
+ final JSONArray vertexArray = new JSONArray();
+
+ long counter = 0;
+ for (Vertex vertex : vertices) {
+ counter++;
+ vertexArray.put(GraphSONUtility.jsonFromElement(
+ vertex, getVertexIndexedKeys(), GraphSONMode.NORMAL));
+ }
+
+ JSONObject response = new JSONObject();
+ response.put(RESULTS, vertexArray);
+ response.put(TOTAL_SIZE, counter);
+ return Response.ok(response).build();
+ } catch (JSONException e) {
+ throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
+ .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ }
+ }
+
+ /**
+ * Get a list of adjacent edges with a direction.
+ *
+ * GET http://host/graphs/lineage/vertices/id/direction
+ * graph.getVertex(id).get{Direction}Edges();
+ * direction: {(?!outE)(?!bothE)(?!inE)(?!out)(?!both)(?!in)(?!query).+}
+ */
+ @GET
+ @Path("vertices/{id}/{direction}")
+ @Produces({MediaType.APPLICATION_JSON})
+ public Response getVertexEdges(@PathParam("id") String vertexId,
+ @PathParam("direction") String direction) {
+ checkIfMetadataMappingServiceIsEnabled();
+ LOG.info("Get vertex edges for vertexId= " + vertexId + ", direction= " + direction);
+ try {
+ Vertex vertex = getGraph().getVertex(vertexId);
+ if (vertex == null) {
+ String message = "Vertex with [" + vertexId + "] cannot be found.";
+ LOG.info(message);
+ throw new WebApplicationException(Response.status(Response.Status.NOT_FOUND)
+ .entity(JSONObject.quote(message)).build());
+ }
+
+ // break out the segment into the return and the direction
+ VertexQueryArguments queryArguments = new VertexQueryArguments(direction);
+ // if this is a query and the _return is "count" then we don't bother to send back the result array
+ boolean countOnly = queryArguments.isCountOnly();
+ // what kind of data the calling client wants back (vertices, edges, count, vertex identifiers)
+ ReturnType returnType = queryArguments.getReturnType();
+ // the query direction (both, out, in)
+ Direction queryDirection = queryArguments.getQueryDirection();
+
+ VertexQuery query = vertex.query().direction(queryDirection);
+
+ JSONArray elementArray = new JSONArray();
+ long counter = 0;
+ if (returnType == ReturnType.VERTICES || returnType == ReturnType.VERTEX_IDS) {
+ Iterable<Vertex> vertexQueryResults = query.vertices();
+ for (Vertex v : vertexQueryResults) {
+ if (returnType.equals(ReturnType.VERTICES)) {
+ elementArray.put(GraphSONUtility.jsonFromElement(
+ v, getVertexIndexedKeys(), GraphSONMode.NORMAL));
+ } else {
+ elementArray.put(v.getId());
+ }
+ counter++;
+ }
+ } else if (returnType == ReturnType.EDGES) {
+ Iterable<Edge> edgeQueryResults = query.edges();
+ for (Edge e : edgeQueryResults) {
+ elementArray.put(GraphSONUtility.jsonFromElement(
+ e, getEdgeIndexedKeys(), GraphSONMode.NORMAL));
+ counter++;
+ }
+ } else if (returnType == ReturnType.COUNT) {
+ counter = query.count();
+ }
+
+ JSONObject response = new JSONObject();
+ if (!countOnly) {
+ response.put(RESULTS, elementArray);
+ }
+ response.put(TOTAL_SIZE, counter);
+ return Response.ok(response).build();
+ } catch (JSONException e) {
+ throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
+ .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ }
+ }
+
+ /**
+ * Get all edges.
+ *
+ * GET http://host/graphs/lineage/edges/all
+ * graph.getEdges();
+ */
+ @GET
+ @Path("/edges/all")
+ @Produces({MediaType.APPLICATION_JSON})
+ public Response getEdges() {
+ checkIfMetadataMappingServiceIsEnabled();
+ LOG.info("Get All Edges.");
+ try {
+ JSONArray vertexArray = new JSONArray();
+ long counter = 0;
+ for (Edge edge : getGraph().getEdges()) {
+ counter++;
+ vertexArray.put(GraphSONUtility.jsonFromElement(
+ edge, getEdgeIndexedKeys(), GraphSONMode.NORMAL));
+ }
+
+ JSONObject response = new JSONObject();
+ response.put(RESULTS, vertexArray);
+ response.put(TOTAL_SIZE, counter);
+ return Response.ok(response).build();
+ } catch (JSONException e) {
+ throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
+ .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ }
+ }
+
+ /**
+ * Get a single edge with a unique id.
+ *
+ * GET http://host/graphs/lineage/edges/id
+ * graph.getEdge(id);
+ */
+ @GET
+ @Path("/edges/{id}")
+ @Produces({MediaType.APPLICATION_JSON})
+ public Response getEdge(@PathParam("id") final String edgeId) {
+ checkIfMetadataMappingServiceIsEnabled();
+ LOG.info("Get vertex for edgeId= " + edgeId);
+ try {
+ Edge edge = getGraph().getEdge(edgeId);
+ if (edge == null) {
+ String message = "Edge with [" + edgeId + "] cannot be found.";
+ LOG.info(message);
+ throw new WebApplicationException(Response.status(Response.Status.NOT_FOUND)
+ .entity(JSONObject.quote(message)).build());
+ }
+
+ JSONObject response = new JSONObject();
+ response.put(RESULTS, GraphSONUtility.jsonFromElement(
+ edge, getEdgeIndexedKeys(), GraphSONMode.NORMAL));
+ return Response.ok(response).build();
+ } catch (JSONException e) {
+ throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
+ .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+ }
+ }
+
+ private void checkIfMetadataMappingServiceIsEnabled() {
+ if (service == null) {
+ throw new WebApplicationException(
+ Response.status(Response.Status.NOT_FOUND)
+ .entity("Lineage Metadata Service is not enabled.")
+ .type("text/plain")
+ .build());
+ }
+ }
+
+ private enum ReturnType {VERTICES, EDGES, COUNT, VERTEX_IDS}
+
+ public static final String OUT_E = "outE";
+ public static final String IN_E = "inE";
+ public static final String BOTH_E = "bothE";
+ public static final String OUT = "out";
+ public static final String IN = "in";
+ public static final String BOTH = "both";
+ public static final String OUT_COUNT = "outCount";
+ public static final String IN_COUNT = "inCount";
+ public static final String BOTH_COUNT = "bothCount";
+ public static final String OUT_IDS = "outIds";
+ public static final String IN_IDS = "inIds";
+ public static final String BOTH_IDS = "bothIds";
+
+ /**
+ * Helper class for query arguments.
+ */
+ public static final class VertexQueryArguments {
+
+ private final Direction queryDirection;
+ private final ReturnType returnType;
+ private final boolean countOnly;
+
+ public VertexQueryArguments(String directionSegment) {
+ if (directionSegment.equals(OUT_E)) {
+ returnType = ReturnType.EDGES;
+ queryDirection = Direction.OUT;
+ countOnly = false;
+ } else if (directionSegment.equals(IN_E)) {
+ returnType = ReturnType.EDGES;
+ queryDirection = Direction.IN;
+ countOnly = false;
+ } else if (directionSegment.equals(BOTH_E)) {
+ returnType = ReturnType.EDGES;
+ queryDirection = Direction.BOTH;
+ countOnly = false;
+ } else if (directionSegment.equals(OUT)) {
+ returnType = ReturnType.VERTICES;
+ queryDirection = Direction.OUT;
+ countOnly = false;
+ } else if (directionSegment.equals(IN)) {
+ returnType = ReturnType.VERTICES;
+ queryDirection = Direction.IN;
+ countOnly = false;
+ } else if (directionSegment.equals(BOTH)) {
+ returnType = ReturnType.VERTICES;
+ queryDirection = Direction.BOTH;
+ countOnly = false;
+ } else if (directionSegment.equals(BOTH_COUNT)) {
+ returnType = ReturnType.COUNT;
+ queryDirection = Direction.BOTH;
+ countOnly = true;
+ } else if (directionSegment.equals(IN_COUNT)) {
+ returnType = ReturnType.COUNT;
+ queryDirection = Direction.IN;
+ countOnly = true;
+ } else if (directionSegment.equals(OUT_COUNT)) {
+ returnType = ReturnType.COUNT;
+ queryDirection = Direction.OUT;
+ countOnly = true;
+ } else if (directionSegment.equals(BOTH_IDS)) {
+ returnType = ReturnType.VERTEX_IDS;
+ queryDirection = Direction.BOTH;
+ countOnly = false;
+ } else if (directionSegment.equals(IN_IDS)) {
+ returnType = ReturnType.VERTEX_IDS;
+ queryDirection = Direction.IN;
+ countOnly = false;
+ } else if (directionSegment.equals(OUT_IDS)) {
+ returnType = ReturnType.VERTEX_IDS;
+ queryDirection = Direction.OUT;
+ countOnly = false;
+ } else {
+ throw new WebApplicationException(Response.status(Response.Status.BAD_REQUEST)
+ .entity(JSONObject.quote(directionSegment + " segment was invalid."))
+ .build());
+ }
+ }
+
+ public Direction getQueryDirection() {
+ return queryDirection;
+ }
+
+ public ReturnType getReturnType() {
+ return returnType;
+ }
+
+ public boolean isCountOnly() {
+ return countOnly;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b8b2fe45/prism/src/main/webapp/WEB-INF/web.xml
----------------------------------------------------------------------
diff --git a/prism/src/main/webapp/WEB-INF/web.xml b/prism/src/main/webapp/WEB-INF/web.xml
index 00a6c42..e7c6234 100644
--- a/prism/src/main/webapp/WEB-INF/web.xml
+++ b/prism/src/main/webapp/WEB-INF/web.xml
@@ -49,7 +49,7 @@
<init-param>
<param-name>com.sun.jersey.config.property.packages</param-name>
<param-value>
- org.apache.falcon.resource.admin,org.apache.falcon.resource.provider,org.apache.falcon.resource.proxy
+ org.apache.falcon.resource.admin,org.apache.falcon.resource.provider,org.apache.falcon.resource.proxy,org.apache.falcon.resource.metadata
</param-value>
</init-param>
<load-on-startup>1</load-on-startup>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b8b2fe45/process/pom.xml
----------------------------------------------------------------------
diff --git a/process/pom.xml b/process/pom.xml
index 0ca554a..46f9cf0 100644
--- a/process/pom.xml
+++ b/process/pom.xml
@@ -106,16 +106,6 @@
</dependency>
<dependency>
- <groupId>org.springframework</groupId>
- <artifactId>spring-jms</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.springframework</groupId>
- <artifactId>spring-beans</artifactId>
- </dependency>
-
- <dependency>
<groupId>org.testng</groupId>
<artifactId>testng</artifactId>
</dependency>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b8b2fe45/webapp/src/main/webapp/WEB-INF/embedded/web.xml
----------------------------------------------------------------------
diff --git a/webapp/src/main/webapp/WEB-INF/embedded/web.xml b/webapp/src/main/webapp/WEB-INF/embedded/web.xml
index bc74d91..9dc371f 100644
--- a/webapp/src/main/webapp/WEB-INF/embedded/web.xml
+++ b/webapp/src/main/webapp/WEB-INF/embedded/web.xml
@@ -49,7 +49,7 @@
<init-param>
<param-name>com.sun.jersey.config.property.packages</param-name>
<param-value>
- org.apache.falcon.resource.admin,org.apache.falcon.resource.provider,org.apache.falcon.resource.proxy
+ org.apache.falcon.resource.admin,org.apache.falcon.resource.provider,org.apache.falcon.resource.proxy,org.apache.falcon.resource.metadata
</param-value>
</init-param>
<load-on-startup>1</load-on-startup>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b8b2fe45/webapp/src/main/webapp/WEB-INF/web.xml
----------------------------------------------------------------------
diff --git a/webapp/src/main/webapp/WEB-INF/web.xml b/webapp/src/main/webapp/WEB-INF/web.xml
index 3864789..971fcdd 100644
--- a/webapp/src/main/webapp/WEB-INF/web.xml
+++ b/webapp/src/main/webapp/WEB-INF/web.xml
@@ -49,7 +49,7 @@
<init-param>
<param-name>com.sun.jersey.config.property.packages</param-name>
<param-value>
- org.apache.falcon.resource.admin,org.apache.falcon.resource.proxy,org.apache.falcon.resource.provider
+ org.apache.falcon.resource.admin,org.apache.falcon.resource.proxy,org.apache.falcon.resource.provider,org.apache.falcon.resource.metadata
</param-value>
</init-param>
<load-on-startup>1</load-on-startup>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/b8b2fe45/webapp/src/test/java/org/apache/falcon/cli/FalconCLISmokeIT.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/falcon/cli/FalconCLISmokeIT.java b/webapp/src/test/java/org/apache/falcon/cli/FalconCLISmokeIT.java
index f946202..f4f72e2 100644
--- a/webapp/src/test/java/org/apache/falcon/cli/FalconCLISmokeIT.java
+++ b/webapp/src/test/java/org/apache/falcon/cli/FalconCLISmokeIT.java
@@ -18,6 +18,7 @@
package org.apache.falcon.cli;
+import org.apache.falcon.client.FalconClient;
import org.apache.falcon.resource.TestContext;
import org.apache.falcon.util.OozieTestUtils;
import org.apache.falcon.util.StartupProperties;
@@ -57,6 +58,15 @@ public class FalconCLISmokeIT {
// this is necessary for lineage
Assert.assertEquals(0, executeWithURL("entity -submit -type cluster -file " + filePath));
+ // verify
+ Assert.assertEquals(0, executeWithURL("graph -vertices -all"));
+ Assert.assertEquals(0, executeWithURL("graph -edges -all"));
+ Assert.assertEquals(0, executeWithURL("graph -vertices -name name -value " + context.getClusterName()));
+
+ // verify lineage
+ FalconClient client = new FalconClient(TestContext.BASE_URL);
+ String clusterVertex = client.getVertices("name", context.getClusterName());
+ Assert.assertTrue(clusterVertex.contains(context.getClusterName()));
filePath = TestContext.overlayParametersOverTemplate(TestContext.FEED_TEMPLATE1, overlay);
Assert.assertEquals(0,
[4/6] FALCON-288 Persist lineage information into a persistent store.
Contributed by Venkatesh Seetharam
Posted by ve...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/common/src/test/java/org/apache/falcon/metadata/MetadataMappingServiceTest.java
----------------------------------------------------------------------
diff --git a/common/src/test/java/org/apache/falcon/metadata/MetadataMappingServiceTest.java b/common/src/test/java/org/apache/falcon/metadata/MetadataMappingServiceTest.java
new file mode 100644
index 0000000..7045636
--- /dev/null
+++ b/common/src/test/java/org/apache/falcon/metadata/MetadataMappingServiceTest.java
@@ -0,0 +1,731 @@
+/**
+ * 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.falcon.metadata;
+
+import com.tinkerpop.blueprints.Direction;
+import com.tinkerpop.blueprints.Edge;
+import com.tinkerpop.blueprints.GraphQuery;
+import com.tinkerpop.blueprints.KeyIndexableGraph;
+import com.tinkerpop.blueprints.Vertex;
+import org.apache.falcon.entity.Storage;
+import org.apache.falcon.entity.store.ConfigurationStore;
+import org.apache.falcon.entity.v0.EntityType;
+import org.apache.falcon.entity.v0.Frequency;
+import org.apache.falcon.entity.v0.cluster.Cluster;
+import org.apache.falcon.entity.v0.cluster.Interface;
+import org.apache.falcon.entity.v0.cluster.Interfaces;
+import org.apache.falcon.entity.v0.cluster.Interfacetype;
+import org.apache.falcon.entity.v0.feed.CatalogTable;
+import org.apache.falcon.entity.v0.feed.Feed;
+import org.apache.falcon.entity.v0.feed.Location;
+import org.apache.falcon.entity.v0.feed.LocationType;
+import org.apache.falcon.entity.v0.feed.Locations;
+import org.apache.falcon.entity.v0.process.EngineType;
+import org.apache.falcon.entity.v0.process.Input;
+import org.apache.falcon.entity.v0.process.Inputs;
+import org.apache.falcon.entity.v0.process.Output;
+import org.apache.falcon.entity.v0.process.Outputs;
+import org.apache.falcon.entity.v0.process.Process;
+import org.apache.falcon.entity.v0.process.Workflow;
+import org.apache.falcon.security.CurrentUser;
+import org.apache.falcon.util.StartupProperties;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Test for Metadata relationship mapping service.
+ */
+public class MetadataMappingServiceTest {
+
+ public static final String FALCON_USER = "falcon-user";
+ private static final String LOGS_DIR = "target/log";
+ private static final String NOMINAL_TIME = "2014-01-01-01-00";
+ public static final String OPERATION = "GENERATE";
+
+ public static final String CLUSTER_ENTITY_NAME = "primary-cluster";
+ public static final String PROCESS_ENTITY_NAME = "sample-process";
+ public static final String COLO_NAME = "west-coast";
+ public static final String WORKFLOW_NAME = "imp-click-join-workflow";
+ public static final String WORKFLOW_VERSION = "1.0.9";
+
+ public static final String INPUT_FEED_NAMES = "impression-feed,clicks-feed";
+ public static final String INPUT_INSTANCE_PATHS =
+ "jail://global:00/falcon/impression-feed/20140101,jail://global:00/falcon/clicks-feed/20140101";
+
+ public static final String OUTPUT_FEED_NAMES = "imp-click-join1,imp-click-join2";
+ public static final String OUTPUT_INSTANCE_PATHS =
+ "jail://global:00/falcon/imp-click-join1/20140101,jail://global:00/falcon/imp-click-join2/20140101";
+
+ private ConfigurationStore configStore;
+ private MetadataMappingService service;
+
+ private Cluster clusterEntity;
+ private Cluster bcpCluster;
+ private List<Feed> inputFeeds = new ArrayList<Feed>();
+ private List<Feed> outputFeeds = new ArrayList<Feed>();
+ private Process processEntity;
+
+
+ @BeforeClass
+ public void setUp() throws Exception {
+ CurrentUser.authenticate(FALCON_USER);
+
+ configStore = ConfigurationStore.get();
+
+ StartupProperties.get().setProperty("falcon.graph.preserve.history", "true");
+ service = new MetadataMappingService();
+ service.init();
+
+ Set<String> vertexPropertyKeys = service.getVertexIndexedKeys();
+ System.out.println("Got vertex property keys: " + vertexPropertyKeys);
+
+ Set<String> edgePropertyKeys = service.getEdgeIndexedKeys();
+ System.out.println("Got edge property keys: " + edgePropertyKeys);
+ }
+
+ @AfterClass
+ public void tearDown() throws Exception {
+ GraphUtils.dump(service.getGraph(), System.out);
+
+ cleanupGraphStore(service.getGraph());
+ cleanupConfigurationStore(configStore);
+ service.destroy();
+ StartupProperties.get().setProperty("falcon.graph.preserve.history", "false");
+ }
+
+ @AfterMethod
+ public void printGraph() throws Exception {
+ GraphUtils.dump(service.getGraph());
+ }
+
+ private GraphQuery getQuery() {
+ return service.getGraph().query();
+ }
+
+ @Test
+ public void testGetName() throws Exception {
+ Assert.assertEquals(service.getName(), MetadataMappingService.SERVICE_NAME);
+ }
+
+ @Test
+ public void testOnAddClusterEntity() throws Exception {
+ clusterEntity = buildCluster(CLUSTER_ENTITY_NAME, COLO_NAME, "classification=production");
+ configStore.publish(EntityType.CLUSTER, clusterEntity);
+
+ verifyEntityWasAddedToGraph(CLUSTER_ENTITY_NAME, EntityRelationshipGraphBuilder.CLUSTER_ENTITY_TYPE);
+ verifyClusterEntityEdges();
+
+ Assert.assertEquals(getVerticesCount(service.getGraph()), 3); // +3 = cluster, colo, tag
+ Assert.assertEquals(getEdgesCount(service.getGraph()), 2); // +2 = cluster to colo and tag
+ }
+
+ @Test (dependsOnMethods = "testOnAddClusterEntity")
+ public void testOnAddFeedEntity() throws Exception {
+ Feed impressionsFeed = buildFeed("impression-feed", clusterEntity, "classified-as=Secure", "analytics",
+ Storage.TYPE.FILESYSTEM, "/falcon/impression-feed/${YEAR}${MONTH}${DAY}");
+ configStore.publish(EntityType.FEED, impressionsFeed);
+ inputFeeds.add(impressionsFeed);
+ verifyEntityWasAddedToGraph(impressionsFeed.getName(), EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE);
+ verifyFeedEntityEdges(impressionsFeed.getName());
+ Assert.assertEquals(getVerticesCount(service.getGraph()), 7); // +4 = feed, tag, group, user
+ Assert.assertEquals(getEdgesCount(service.getGraph()), 6); // +4 = cluster, tag, group, user
+
+ Feed clicksFeed = buildFeed("clicks-feed", clusterEntity, "classified-as=Secure,classified-as=Financial",
+ "analytics", Storage.TYPE.FILESYSTEM, "/falcon/clicks-feed/${YEAR}${MONTH}${DAY}");
+ configStore.publish(EntityType.FEED, clicksFeed);
+ inputFeeds.add(clicksFeed);
+ verifyEntityWasAddedToGraph(clicksFeed.getName(),
+ EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE);
+ Assert.assertEquals(getVerticesCount(service.getGraph()), 9); // feed and financial vertex
+ Assert.assertEquals(getEdgesCount(service.getGraph()), 11); // +5 = cluster + user + 2Group + Tag
+
+ Feed join1Feed = buildFeed("imp-click-join1", clusterEntity, "classified-as=Financial", "reporting,bi",
+ Storage.TYPE.FILESYSTEM, "/falcon/imp-click-join1/${YEAR}${MONTH}${DAY}");
+ configStore.publish(EntityType.FEED, join1Feed);
+ outputFeeds.add(join1Feed);
+ verifyEntityWasAddedToGraph(join1Feed.getName(),
+ EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE);
+ Assert.assertEquals(getVerticesCount(service.getGraph()), 12); // + 3 = 1 feed and 2 groups
+ Assert.assertEquals(getEdgesCount(service.getGraph()), 16); // +5 = cluster + user + Group + 2Tags
+
+ Feed join2Feed = buildFeed("imp-click-join2", clusterEntity, "classified-as=Secure,classified-as=Financial",
+ "reporting,bi", Storage.TYPE.FILESYSTEM, "/falcon/imp-click-join2/${YEAR}${MONTH}${DAY}");
+ configStore.publish(EntityType.FEED, join2Feed);
+ outputFeeds.add(join2Feed);
+ verifyEntityWasAddedToGraph(join2Feed.getName(), EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE);
+
+ Assert.assertEquals(getVerticesCount(service.getGraph()), 13); // +1 feed
+ // +6 = user + 2tags + 2Groups + Cluster
+ Assert.assertEquals(getEdgesCount(service.getGraph()), 22);
+ }
+
+ @Test (dependsOnMethods = "testOnAddFeedEntity")
+ public void testOnAddProcessEntity() throws Exception {
+ processEntity = buildProcess(PROCESS_ENTITY_NAME, clusterEntity, "classified-as=Critical");
+ addWorkflow(processEntity, WORKFLOW_NAME, WORKFLOW_VERSION);
+
+ for (Feed inputFeed : inputFeeds) {
+ addInput(processEntity, inputFeed);
+ }
+
+ for (Feed outputFeed : outputFeeds) {
+ addOutput(processEntity, outputFeed);
+ }
+
+ configStore.publish(EntityType.PROCESS, processEntity);
+
+ verifyEntityWasAddedToGraph(processEntity.getName(), EntityRelationshipGraphBuilder.PROCESS_ENTITY_TYPE);
+ verifyProcessEntityEdges();
+
+ // +2 = 1 process + 1 tag
+ Assert.assertEquals(getVerticesCount(service.getGraph()), 15);
+ // +7 = user,tag,cluster, 2 inputs,2 outputs
+ Assert.assertEquals(getEdgesCount(service.getGraph()), 29);
+ }
+
+ @Test (dependsOnMethods = "testOnAddProcessEntity")
+ public void testOnAdd() throws Exception {
+ verifyEntityGraph(EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE, "Secure");
+ }
+
+ @Test(dependsOnMethods = "testOnAdd")
+ public void testMapLineage() throws Exception {
+
+ LineageRecorder.main(getTestMessageArgs());
+
+ service.mapLineage(PROCESS_ENTITY_NAME, OPERATION, LOGS_DIR);
+
+ debug(service.getGraph());
+ GraphUtils.dump(service.getGraph());
+ verifyLineageGraph(InstanceRelationshipGraphBuilder.FEED_INSTANCE_TYPE);
+
+ // +6 = 1 process, 2 inputs,2 outputs
+ Assert.assertEquals(getVerticesCount(service.getGraph()), 20);
+ //+32 = +26 for feed instances + 6 for process instance
+ Assert.assertEquals(getEdgesCount(service.getGraph()), 61);
+ }
+
+ @Test (dependsOnMethods = "testMapLineage")
+ public void testOnChange() throws Exception {
+ // cannot modify cluster, adding a new cluster
+ bcpCluster = buildCluster("bcp-cluster", "east-coast", "classification=bcp");
+ configStore.publish(EntityType.CLUSTER, bcpCluster);
+ verifyEntityWasAddedToGraph("bcp-cluster", EntityRelationshipGraphBuilder.CLUSTER_ENTITY_TYPE);
+
+ Assert.assertEquals(getVerticesCount(service.getGraph()), 23); // +3 = cluster, colo, tag
+ // +2 edges to above, no user but only to colo and new tag
+ Assert.assertEquals(getEdgesCount(service.getGraph()), 63);
+ }
+
+ @Test(dependsOnMethods = "testOnChange")
+ public void testOnFeedEntityChange() throws Exception {
+ Feed oldFeed = inputFeeds.get(0);
+ Feed newFeed = buildFeed(oldFeed.getName(), clusterEntity,
+ "classified-as=Secured,source=data-warehouse", "reporting",
+ Storage.TYPE.FILESYSTEM, "jail://global:00/falcon/impression-feed/20140101");
+
+ try {
+ configStore.initiateUpdate(newFeed);
+
+ // add cluster
+ org.apache.falcon.entity.v0.feed.Cluster feedCluster =
+ new org.apache.falcon.entity.v0.feed.Cluster();
+ feedCluster.setName(bcpCluster.getName());
+ newFeed.getClusters().getClusters().add(feedCluster);
+
+ configStore.update(EntityType.FEED, newFeed);
+ } finally {
+ configStore.cleanupUpdateInit();
+ }
+
+ verifyUpdatedEdges(newFeed);
+ Assert.assertEquals(getVerticesCount(service.getGraph()), 25); //+2 = 2 new tags
+ Assert.assertEquals(getEdgesCount(service.getGraph()), 65); // +2 = 1 new cluster, 1 new tag
+ }
+
+ private void verifyUpdatedEdges(Feed newFeed) {
+ Vertex feedVertex = getEntityVertex(newFeed.getName(),
+ EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE);
+
+ // groups
+ Edge edge = feedVertex.getEdges(Direction.OUT, RelationshipGraphBuilder.GROUPS_LABEL).iterator().next();
+ Assert.assertEquals(edge.getVertex(Direction.IN).getProperty("name"), "reporting");
+
+ // tags
+ edge = feedVertex.getEdges(Direction.OUT, "classified-as").iterator().next();
+ Assert.assertEquals(edge.getVertex(Direction.IN).getProperty("name"), "Secured");
+ edge = feedVertex.getEdges(Direction.OUT, "source").iterator().next();
+ Assert.assertEquals(edge.getVertex(Direction.IN).getProperty("name"), "data-warehouse");
+
+ // new cluster
+ Iterator<Edge> clusterEdgeIterator = feedVertex.getEdges(Direction.OUT,
+ RelationshipGraphBuilder.FEED_CLUSTER_EDGE_LABEL).iterator();
+ edge = clusterEdgeIterator.next();
+ Assert.assertEquals(edge.getVertex(Direction.IN).getProperty("name"), clusterEntity.getName());
+ edge = clusterEdgeIterator.next();
+ Assert.assertEquals(edge.getVertex(Direction.IN).getProperty("name"), bcpCluster.getName());
+ }
+
+ @Test(dependsOnMethods = "testOnFeedEntityChange")
+ public void testOnProcessEntityChange() throws Exception {
+ Process oldProcess = processEntity;
+ Process newProcess = buildProcess(oldProcess.getName(), bcpCluster, null);
+ addWorkflow(newProcess, WORKFLOW_NAME, "2.0.0");
+ addInput(newProcess, inputFeeds.get(0));
+
+ try {
+ configStore.initiateUpdate(newProcess);
+ configStore.update(EntityType.PROCESS, newProcess);
+ } finally {
+ configStore.cleanupUpdateInit();
+ }
+
+ verifyUpdatedEdges(newProcess);
+ Assert.assertEquals(getVerticesCount(service.getGraph()), 25); // +0, no net new
+ Assert.assertEquals(getEdgesCount(service.getGraph()), 61); // -4 = -2 outputs, -1 tag, -1 cluster
+ }
+
+ private void verifyUpdatedEdges(Process newProcess) {
+ Vertex processVertex = getEntityVertex(newProcess.getName(),
+ EntityRelationshipGraphBuilder.PROCESS_ENTITY_TYPE);
+
+ // cluster
+ Edge edge = processVertex.getEdges(Direction.OUT,
+ RelationshipGraphBuilder.PROCESS_CLUSTER_EDGE_LABEL).iterator().next();
+ Assert.assertEquals(edge.getVertex(Direction.IN).getProperty("name"), bcpCluster.getName());
+
+/*
+ // workflow
+ edge = processVertex.getEdges(Direction.OUT,
+ RelationshipGraphBuilder.PROCESS_WORKFLOW_EDGE_LABEL).iterator().next();
+ Assert.assertEquals(edge.getVertex(Direction.IN).getProperty("version"),
+ newProcess.getWorkflow().getVersion());
+*/
+
+ // inputs
+ edge = processVertex.getEdges(Direction.IN,
+ RelationshipGraphBuilder.FEED_PROCESS_EDGE_LABEL).iterator().next();
+ Assert.assertEquals(edge.getVertex(Direction.OUT).getProperty("name"),
+ newProcess.getInputs().getInputs().get(0).getFeed());
+
+ // outputs
+ for (Edge e : processVertex.getEdges(Direction.OUT, RelationshipGraphBuilder.PROCESS_FEED_EDGE_LABEL)) {
+ Assert.fail("there should not be any edges to output feeds" + e);
+ }
+ }
+
+ public static void debug(final KeyIndexableGraph graph) {
+ System.out.println("*****Vertices of " + graph);
+ for (Vertex vertex : graph.getVertices()) {
+ System.out.println(GraphUtils.vertexString(vertex));
+ }
+
+ System.out.println("*****Edges of " + graph);
+ for (Edge edge : graph.getEdges()) {
+ System.out.println(GraphUtils.edgeString(edge));
+ }
+ }
+
+ private static Cluster buildCluster(String name, String colo, String tags) {
+ Cluster cluster = new Cluster();
+ cluster.setName(name);
+ cluster.setColo(colo);
+ cluster.setTags(tags);
+
+ Interfaces interfaces = new Interfaces();
+ cluster.setInterfaces(interfaces);
+
+ Interface storage = new Interface();
+ storage.setEndpoint("jail://global:00");
+ storage.setType(Interfacetype.WRITE);
+ cluster.getInterfaces().getInterfaces().add(storage);
+
+ return cluster;
+ }
+
+ private static Feed buildFeed(String feedName, Cluster cluster, String tags, String groups,
+ Storage.TYPE storageType, String uriTemplate) {
+ Feed feed = new Feed();
+ feed.setName(feedName);
+ feed.setTags(tags);
+ feed.setGroups(groups);
+ feed.setFrequency(Frequency.fromString("hours(1)"));
+
+ org.apache.falcon.entity.v0.feed.Clusters
+ clusters = new org.apache.falcon.entity.v0.feed.Clusters();
+ feed.setClusters(clusters);
+ org.apache.falcon.entity.v0.feed.Cluster feedCluster =
+ new org.apache.falcon.entity.v0.feed.Cluster();
+ feedCluster.setName(cluster.getName());
+ clusters.getClusters().add(feedCluster);
+
+ addStorage(feed, storageType, uriTemplate);
+
+ return feed;
+ }
+
+ private static void addStorage(Feed feed, Storage.TYPE storageType, String uriTemplate) {
+ if (storageType == Storage.TYPE.FILESYSTEM) {
+ Locations locations = new Locations();
+ feed.setLocations(locations);
+
+ Location location = new Location();
+ location.setType(LocationType.DATA);
+ location.setPath(uriTemplate);
+ feed.getLocations().getLocations().add(location);
+ } else {
+ CatalogTable table = new CatalogTable();
+ table.setUri(uriTemplate);
+ feed.setTable(table);
+ }
+ }
+
+ private static Process buildProcess(String processName, Cluster cluster,
+ String tags) throws Exception {
+ Process processEntity = new Process();
+ processEntity.setName(processName);
+ processEntity.setTags(tags);
+
+ org.apache.falcon.entity.v0.process.Cluster processCluster =
+ new org.apache.falcon.entity.v0.process.Cluster();
+ processCluster.setName(cluster.getName());
+ processEntity.setClusters(new org.apache.falcon.entity.v0.process.Clusters());
+ processEntity.getClusters().getClusters().add(processCluster);
+
+ return processEntity;
+ }
+
+ private static void addWorkflow(Process process, String workflowName, String version) {
+ Workflow workflow = new Workflow();
+ workflow.setName(workflowName);
+ workflow.setVersion(version);
+ workflow.setEngine(EngineType.PIG);
+ workflow.setPath("/falcon/test/workflow");
+
+ process.setWorkflow(workflow);
+ }
+
+ private static void addInput(Process process, Feed feed) {
+ if (process.getInputs() == null) {
+ process.setInputs(new Inputs());
+ }
+
+ Inputs inputs = process.getInputs();
+ Input input = new Input();
+ input.setFeed(feed.getName());
+ inputs.getInputs().add(input);
+ }
+
+ private static void addOutput(Process process, Feed feed) {
+ if (process.getOutputs() == null) {
+ process.setOutputs(new Outputs());
+ }
+
+ Outputs outputs = process.getOutputs();
+ Output output = new Output();
+ output.setFeed(feed.getName());
+ outputs.getOutputs().add(output);
+ }
+
+ private void verifyEntityWasAddedToGraph(String entityName, String entityType) {
+ Vertex entityVertex = getEntityVertex(entityName, entityType);
+ Assert.assertNotNull(entityVertex);
+ verifyEntityProperties(entityVertex, entityName, entityType);
+ }
+
+ private void verifyEntityProperties(Vertex entityVertex, String entityName, String entityType) {
+ Assert.assertEquals(entityName, entityVertex.getProperty(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY));
+ Assert.assertEquals(entityType, entityVertex.getProperty(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY));
+ Assert.assertNotNull(entityVertex.getProperty(EntityRelationshipGraphBuilder.TIMESTAMP_PROPERTY_KEY));
+ }
+
+ private void verifyClusterEntityEdges() {
+ Vertex clusterVertex = getEntityVertex(CLUSTER_ENTITY_NAME,
+ EntityRelationshipGraphBuilder.CLUSTER_ENTITY_TYPE);
+
+ // verify edge to user vertex
+ verifyVertexForEdge(clusterVertex, Direction.OUT, EntityRelationshipGraphBuilder.USER_LABEL,
+ FALCON_USER, EntityRelationshipGraphBuilder.USER_TYPE);
+ // verify edge to colo vertex
+ verifyVertexForEdge(clusterVertex, Direction.OUT, EntityRelationshipGraphBuilder.CLUSTER_COLO_LABEL,
+ COLO_NAME, EntityRelationshipGraphBuilder.COLO_TYPE);
+ // verify edge to tags vertex
+ verifyVertexForEdge(clusterVertex, Direction.OUT, "classification",
+ "production", EntityRelationshipGraphBuilder.TAGS_TYPE);
+ }
+
+ private void verifyFeedEntityEdges(String feedName) {
+ Vertex feedVertex = getEntityVertex(feedName, EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE);
+
+ // verify edge to cluster vertex
+ verifyVertexForEdge(feedVertex, Direction.OUT, EntityRelationshipGraphBuilder.FEED_CLUSTER_EDGE_LABEL,
+ CLUSTER_ENTITY_NAME, EntityRelationshipGraphBuilder.CLUSTER_ENTITY_TYPE);
+ // verify edge to user vertex
+ verifyVertexForEdge(feedVertex, Direction.OUT, EntityRelationshipGraphBuilder.USER_LABEL,
+ FALCON_USER, EntityRelationshipGraphBuilder.USER_TYPE);
+ // verify edge to tags vertex
+ verifyVertexForEdge(feedVertex, Direction.OUT, "classified-as",
+ "Secure", EntityRelationshipGraphBuilder.TAGS_TYPE);
+ // verify edge to group vertex
+ verifyVertexForEdge(feedVertex, Direction.OUT, EntityRelationshipGraphBuilder.GROUPS_LABEL,
+ "analytics", EntityRelationshipGraphBuilder.GROUPS_TYPE);
+ }
+
+ private void verifyProcessEntityEdges() {
+ Vertex processVertex = getEntityVertex(PROCESS_ENTITY_NAME,
+ EntityRelationshipGraphBuilder.PROCESS_ENTITY_TYPE);
+
+ // verify edge to cluster vertex
+ verifyVertexForEdge(processVertex, Direction.OUT, EntityRelationshipGraphBuilder.FEED_CLUSTER_EDGE_LABEL,
+ CLUSTER_ENTITY_NAME, EntityRelationshipGraphBuilder.CLUSTER_ENTITY_TYPE);
+ // verify edge to user vertex
+ verifyVertexForEdge(processVertex, Direction.OUT, EntityRelationshipGraphBuilder.USER_LABEL,
+ FALCON_USER, EntityRelationshipGraphBuilder.USER_TYPE);
+ // verify edge to tags vertex
+ verifyVertexForEdge(processVertex, Direction.OUT, "classified-as",
+ "Critical", EntityRelationshipGraphBuilder.TAGS_TYPE);
+
+ // verify edges to inputs
+ List<String> actual = new ArrayList<String>();
+ for (Edge edge : processVertex.getEdges(Direction.IN,
+ EntityRelationshipGraphBuilder.FEED_PROCESS_EDGE_LABEL)) {
+ Vertex outVertex = edge.getVertex(Direction.OUT);
+ Assert.assertEquals(EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE,
+ outVertex.getProperty(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY));
+ actual.add(outVertex.<String>getProperty(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY));
+ }
+
+ Assert.assertTrue(actual.containsAll(Arrays.asList("impression-feed", "clicks-feed")),
+ "Actual does not contain expected: " + actual);
+
+ actual.clear();
+ // verify edges to outputs
+ for (Edge edge : processVertex.getEdges(Direction.OUT,
+ EntityRelationshipGraphBuilder.PROCESS_FEED_EDGE_LABEL)) {
+ Vertex outVertex = edge.getVertex(Direction.IN);
+ Assert.assertEquals(EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE,
+ outVertex.getProperty(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY));
+ actual.add(outVertex.<String>getProperty(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY));
+ }
+ Assert.assertTrue(actual.containsAll(Arrays.asList("imp-click-join1", "imp-click-join2")),
+ "Actual does not contain expected: " + actual);
+ }
+
+ private Vertex getEntityVertex(String entityName, String entityType) {
+ GraphQuery entityQuery = getQuery()
+ .has(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY, entityName)
+ .has(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY, entityType);
+ Iterator<Vertex> iterator = entityQuery.vertices().iterator();
+ Assert.assertTrue(iterator.hasNext());
+
+ Vertex entityVertex = iterator.next();
+ Assert.assertNotNull(entityVertex);
+
+ return entityVertex;
+ }
+
+ private void verifyVertexForEdge(Vertex fromVertex, Direction direction, String label,
+ String expectedName, String expectedType) {
+ for (Edge edge : fromVertex.getEdges(direction, label)) {
+ Vertex outVertex = edge.getVertex(Direction.IN);
+ Assert.assertEquals(
+ outVertex.getProperty(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY), expectedName);
+ Assert.assertEquals(
+ outVertex.getProperty(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY), expectedType);
+ }
+ }
+
+ private void verifyEntityGraph(String feedType, String classification) {
+ // feeds owned by a user
+ List<String> feedNamesOwnedByUser = getFeedsOwnedByAUser(feedType);
+ Assert.assertEquals(feedNamesOwnedByUser,
+ Arrays.asList("impression-feed", "clicks-feed", "imp-click-join1", "imp-click-join2"));
+
+ // feeds classified as secure
+ verifyFeedsClassifiedAsSecure(feedType,
+ Arrays.asList("impression-feed", "clicks-feed", "imp-click-join2"));
+
+ // feeds owned by a user and classified as secure
+ verifyFeedsOwnedByUserAndClassification(feedType, classification,
+ Arrays.asList("impression-feed", "clicks-feed", "imp-click-join2"));
+ }
+
+ private List<String> getFeedsOwnedByAUser(String feedType) {
+ GraphQuery userQuery = getQuery()
+ .has(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY, FALCON_USER)
+ .has(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY, EntityRelationshipGraphBuilder.USER_TYPE);
+
+ List<String> feedNames = new ArrayList<String>();
+ for (Vertex userVertex : userQuery.vertices()) {
+ for (Vertex feed : userVertex.getVertices(Direction.IN, EntityRelationshipGraphBuilder.USER_LABEL)) {
+ if (feed.getProperty(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY).equals(feedType)) {
+ System.out.println(FALCON_USER + " owns -> " + GraphUtils.vertexString(feed));
+ feedNames.add(feed.<String>getProperty(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY));
+ }
+ }
+ }
+
+ return feedNames;
+ }
+
+ private void verifyFeedsClassifiedAsSecure(String feedType, List<String> expected) {
+ GraphQuery classQuery = getQuery()
+ .has(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY, "Secure")
+ .has(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY, EntityRelationshipGraphBuilder.TAGS_TYPE);
+
+ List<String> actual = new ArrayList<String>();
+ for (Vertex feedVertex : classQuery.vertices()) {
+ for (Vertex feed : feedVertex.getVertices(Direction.BOTH, "classified-as")) {
+ if (feed.getProperty(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY).equals(feedType)) {
+ System.out.println(" Secure classification -> " + GraphUtils.vertexString(feed));
+ actual.add(feed.<String>getProperty(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY));
+ }
+ }
+ }
+
+ Assert.assertTrue(actual.containsAll(expected), "Actual does not contain expected: " + actual);
+ }
+
+ private void verifyFeedsOwnedByUserAndClassification(String feedType, String classification,
+ List<String> expected) {
+ List<String> actual = new ArrayList<String>();
+ Vertex userVertex = getEntityVertex(FALCON_USER, EntityRelationshipGraphBuilder.USER_TYPE);
+ for (Vertex feed : userVertex.getVertices(Direction.IN, EntityRelationshipGraphBuilder.USER_LABEL)) {
+ if (feed.getProperty(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY).equals(feedType)) {
+ for (Vertex classVertex : feed.getVertices(Direction.OUT, "classified-as")) {
+ if (classVertex.getProperty(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY)
+ .equals(classification)) {
+ actual.add(feed.<String>getProperty(RelationshipGraphBuilder.NAME_PROPERTY_KEY));
+ System.out.println(classification + " feed owned by falcon-user -> "
+ + GraphUtils.vertexString(feed));
+ }
+ }
+ }
+ }
+ Assert.assertTrue(actual.containsAll(expected), "Actual does not contain expected: " + actual);
+ }
+
+ public long getVerticesCount(final KeyIndexableGraph graph) {
+ long count = 0;
+ for (Vertex ignored : graph.getVertices()) {
+ count++;
+ }
+
+ return count;
+ }
+
+ public long getEdgesCount(final KeyIndexableGraph graph) {
+ long count = 0;
+ for (Edge ignored : graph.getEdges()) {
+ count++;
+ }
+
+ return count;
+ }
+
+ private void verifyLineageGraph(String feedType) {
+ // feeds owned by a user
+ List<String> feedNamesOwnedByUser = getFeedsOwnedByAUser(feedType);
+ List<String> expected = Arrays.asList("impression-feed/20140101", "clicks-feed/20140101",
+ "imp-click-join1/20140101", "imp-click-join2/20140101");
+ Assert.assertTrue(feedNamesOwnedByUser.containsAll(expected));
+
+ KeyIndexableGraph graph = service.getGraph();
+
+ Iterator<Vertex> vertices = graph.getVertices("name", "impression-feed/20140101").iterator();
+ Assert.assertTrue(vertices.hasNext());
+ Vertex feedInstanceVertex = vertices.next();
+ Assert.assertEquals(
+ feedInstanceVertex.getProperty(RelationshipGraphBuilder.TYPE_PROPERTY_KEY),
+ InstanceRelationshipGraphBuilder.FEED_INSTANCE_TYPE);
+
+ Object vertexId = feedInstanceVertex.getId();
+ Vertex vertexById = graph.getVertex(vertexId);
+ Assert.assertEquals(vertexById, feedInstanceVertex);
+
+ // feeds classified as secure
+ verifyFeedsClassifiedAsSecure(feedType,
+ Arrays.asList("impression-feed/20140101", "clicks-feed/20140101", "imp-click-join2/20140101"));
+
+ // feeds owned by a user and classified as secure
+ verifyFeedsOwnedByUserAndClassification(feedType, "Financial",
+ Arrays.asList("clicks-feed/20140101", "imp-click-join1/20140101", "imp-click-join2/20140101"));
+ }
+
+ private static String[] getTestMessageArgs() {
+ return new String[]{
+ "-" + LineageArgs.NOMINAL_TIME.getOptionName(), NOMINAL_TIME,
+ "-" + LineageArgs.TIMESTAMP.getOptionName(), NOMINAL_TIME,
+
+ "-" + LineageArgs.ENTITY_NAME.getOptionName(), PROCESS_ENTITY_NAME,
+ "-" + LineageArgs.ENTITY_TYPE.getOptionName(), ("process"),
+ "-" + LineageArgs.CLUSTER.getOptionName(), CLUSTER_ENTITY_NAME,
+ "-" + LineageArgs.OPERATION.getOptionName(), OPERATION,
+
+ "-" + LineageArgs.INPUT_FEED_NAMES.getOptionName(), INPUT_FEED_NAMES,
+ "-" + LineageArgs.INPUT_FEED_PATHS.getOptionName(), INPUT_INSTANCE_PATHS,
+
+ "-" + LineageArgs.FEED_NAMES.getOptionName(), OUTPUT_FEED_NAMES,
+ "-" + LineageArgs.FEED_INSTANCE_PATHS.getOptionName(), OUTPUT_INSTANCE_PATHS,
+
+ "-" + LineageArgs.WORKFLOW_ID.getOptionName(), "workflow-01-00",
+ "-" + LineageArgs.WORKFLOW_USER.getOptionName(), FALCON_USER,
+ "-" + LineageArgs.RUN_ID.getOptionName(), "1",
+ "-" + LineageArgs.STATUS.getOptionName(), "SUCCEEDED",
+ "-" + LineageArgs.WF_ENGINE_URL.getOptionName(), "http://localhost:11000/oozie",
+ "-" + LineageArgs.USER_SUBFLOW_ID.getOptionName(), "userflow@wf-id",
+ "-" + LineageArgs.USER_WORKFLOW_NAME.getOptionName(), WORKFLOW_NAME,
+ "-" + LineageArgs.USER_WORKFLOW_VERSION.getOptionName(), WORKFLOW_VERSION,
+ "-" + LineageArgs.USER_WORKFLOW_ENGINE.getOptionName(), EngineType.PIG.name(),
+
+ "-" + LineageArgs.LOG_DIR.getOptionName(), LOGS_DIR,
+ };
+ }
+
+ private void cleanupGraphStore(KeyIndexableGraph graph) {
+ for (Edge edge : graph.getEdges()) {
+ graph.removeEdge(edge);
+ }
+
+ for (Vertex vertex : graph.getVertices()) {
+ graph.removeVertex(vertex);
+ }
+
+ graph.shutdown();
+ }
+
+ private static void cleanupConfigurationStore(ConfigurationStore store) throws Exception {
+ for (EntityType type : EntityType.values()) {
+ Collection<String> entities = store.getEntities(type);
+ for (String entity : entities) {
+ store.remove(type, entity);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/oozie/src/main/java/org/apache/falcon/service/SharedLibraryHostingService.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/service/SharedLibraryHostingService.java b/oozie/src/main/java/org/apache/falcon/service/SharedLibraryHostingService.java
index 63c16ad..11dc1e4 100644
--- a/oozie/src/main/java/org/apache/falcon/service/SharedLibraryHostingService.java
+++ b/oozie/src/main/java/org/apache/falcon/service/SharedLibraryHostingService.java
@@ -147,7 +147,7 @@ public class SharedLibraryHostingService implements ConfigurationChangeListener
}
@Override
- public void onAdd(Entity entity, boolean ignoreFailure) throws FalconException {
+ public void onAdd(Entity entity) throws FalconException {
if (entity.getEntityType() != EntityType.CLUSTER) {
return;
}
@@ -157,14 +157,7 @@ public class SharedLibraryHostingService implements ConfigurationChangeListener
return;
}
- try {
- addLibsTo(cluster);
- } catch(FalconException e) {
- if (!ignoreFailure) {
- throw e;
- }
- LOG.warn("Failed to copy shared libraries to cluster", e);
- }
+ addLibsTo(cluster);
}
@Override
@@ -186,4 +179,13 @@ public class SharedLibraryHostingService implements ConfigurationChangeListener
addLibsTo(newCluster);
}
}
+
+ @Override
+ public void onReload(Entity entity) throws FalconException {
+ try {
+ onAdd(entity);
+ } catch (FalconException ignored) {
+ LOG.warn("Failed to copy shared libraries to cluster", ignored);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 81c017e..9bb3264 100644
--- a/pom.xml
+++ b/pom.xml
@@ -970,6 +970,48 @@
<artifactId>xercesImpl</artifactId>
<version>2.10.0</version>
</dependency>
+
+ <dependency>
+ <groupId>com.tinkerpop.blueprints</groupId>
+ <artifactId>blueprints-core</artifactId>
+ <version>2.4.0</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.thinkaurelius.titan</groupId>
+ <artifactId>titan-core-jre6</artifactId>
+ <version>0.4.2</version>
+ <exclusions>
+ <!-- rexster does not work with servlet-api -->
+ <exclusion>
+ <groupId>com.tinkerpop.rexster</groupId>
+ <artifactId>rexster-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.tinkerpop.rexster</groupId>
+ <artifactId>rexster-server</artifactId>
+ </exclusion>
+ <!-- asm 4.0 does not work with jersey asm 3.1 -->
+ <exclusion>
+ <groupId>com.tinkerpop</groupId>
+ <artifactId>frames</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.esotericsoftware.reflectasm</groupId>
+ <artifactId>reflectasm</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.ow2.asm</groupId>
+ <artifactId>asm</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>com.thinkaurelius.titan</groupId>
+ <artifactId>titan-berkeleyje-jre6</artifactId>
+ <version>0.4.2</version>
+ </dependency>
</dependencies>
</dependencyManagement>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/prism/src/main/java/org/apache/falcon/service/FalconTopicSubscriber.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/service/FalconTopicSubscriber.java b/prism/src/main/java/org/apache/falcon/service/FalconTopicSubscriber.java
index c1d629d..a32d2ee 100644
--- a/prism/src/main/java/org/apache/falcon/service/FalconTopicSubscriber.java
+++ b/prism/src/main/java/org/apache/falcon/service/FalconTopicSubscriber.java
@@ -23,6 +23,7 @@ import org.apache.falcon.entity.EntityUtil;
import org.apache.falcon.entity.v0.Entity;
import org.apache.falcon.entity.v0.SchemaHelper;
import org.apache.falcon.messaging.EntityInstanceMessage.ARG;
+import org.apache.falcon.metadata.MetadataMappingService;
import org.apache.falcon.rerun.event.RerunEvent.RerunType;
import org.apache.falcon.rerun.handler.AbstractRerunHandler;
import org.apache.falcon.rerun.handler.RerunHandlerFactory;
@@ -126,6 +127,7 @@ public class FalconTopicSubscriber implements MessageListener, ExceptionListener
SchemaHelper.formatDateUTC(startTime), duration);
notifySLAService(cluster, entityName, entityType, nominalTime, duration);
+ notifyMetadataMappingService(entityName, operation, mapMessage.getString(ARG.logDir.getArgName()));
}
} catch (JMSException e) {
LOG.info("Error in onMessage for subscriber of topic: " + this.toString(), e);
@@ -150,6 +152,12 @@ public class FalconTopicSubscriber implements MessageListener, ExceptionListener
return Services.get().getService(SLAMonitoringService.SERVICE_NAME);
}
+ private void notifyMetadataMappingService(String entityName, String operation,
+ String logDir) throws FalconException {
+ MetadataMappingService service = Services.get().getService(MetadataMappingService.SERVICE_NAME);
+ service.mapLineage(entityName, operation, logDir);
+ }
+
private void debug(MapMessage mapMessage) throws JMSException {
StringBuilder buff = new StringBuilder();
buff.append("Received:{");
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/src/bin/falcon-config.sh
----------------------------------------------------------------------
diff --git a/src/bin/falcon-config.sh b/src/bin/falcon-config.sh
index 47cc1e3..36be053 100644
--- a/src/bin/falcon-config.sh
+++ b/src/bin/falcon-config.sh
@@ -112,7 +112,7 @@ case $type in
export FALCON_PID_DIR
FALCON_PID_FILE=${FALCON_PID_DIR}/${app}.pid
export FALCON_PID_FILE
- FALCON_DATA_DIR=${FALCON_DATA_DIR:-${BASEDIR}/logs/data}
+ FALCON_DATA_DIR=${FALCON_DATA_DIR:-${BASEDIR}/data}
FALCON_HOME_DIR="${FALCON_HOME_DIR:-$BASEDIR}"
export FALCON_HOME_DIR
;;
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/src/conf/startup.properties
----------------------------------------------------------------------
diff --git a/src/conf/startup.properties b/src/conf/startup.properties
index 0d0ab41..74f0d6b 100644
--- a/src/conf/startup.properties
+++ b/src/conf/startup.properties
@@ -36,9 +36,10 @@
org.apache.falcon.service.ProcessSubscriberService,\
org.apache.falcon.entity.store.ConfigurationStore,\
org.apache.falcon.rerun.service.RetryService,\
- org.apache.falcon.rerun.service.LateRunService,\
- org.apache.falcon.service.SLAMonitoringService,\
- org.apache.falcon.service.LogCleanupService
+ org.apache.falcon.rerun.service.LateRunService,\
+ org.apache.falcon.service.SLAMonitoringService,\
+ org.apache.falcon.metadata.MetadataMappingService,\
+ org.apache.falcon.service.LogCleanupService
prism.application.services=org.apache.falcon.entity.store.ConfigurationStore
*.configstore.listeners=org.apache.falcon.entity.v0.EntityGraph,\
org.apache.falcon.entity.ColoClusterRelation,\
@@ -56,7 +57,7 @@ prism.configstore.listeners=org.apache.falcon.entity.v0.EntityGraph,\
######### System startup parameters #########
# Location to store user entity configurations
-*.config.store.uri=file://${falcon.home}/store
+*.config.store.uri=file://${falcon.home}/data/store
# Location of libraries that is shipped to Hadoop
*.system.lib.location=${falcon.home}/server/webapp/falcon/WEB-INF/lib
@@ -80,6 +81,17 @@ prism.system.lib.location=${falcon.home}/server/webapp/prism/WEB-INF/lib
*.internal.queue.size=1000
+######### Graph Database Properties #########
+# Graph implementation
+*.falcon.graph.blueprints.graph=com.thinkaurelius.titan.core.TitanFactory
+
+# Graph Storage
+*.falcon.graph.storage.directory=/${falcon.home}/data/graphdb
+*.falcon.graph.storage.backend=berkeleyje
+*.falcon.graph.serialize.path=/${falcon.home}/data
+*.falcon.graph.preserve.history=false
+
+
######### Authentication Properties #########
# Authentication type must be specified: simple|kerberos
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/webapp/pom.xml
----------------------------------------------------------------------
diff --git a/webapp/pom.xml b/webapp/pom.xml
index 0c9e601..5c293e1 100644
--- a/webapp/pom.xml
+++ b/webapp/pom.xml
@@ -171,6 +171,11 @@
</dependency>
<dependency>
+ <groupId>javax.servlet.jsp</groupId>
+ <artifactId>jsp-api</artifactId>
+ </dependency>
+
+ <dependency>
<groupId>org.mortbay.jetty</groupId>
<artifactId>jetty</artifactId>
</dependency>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/webapp/src/test/java/org/apache/falcon/cli/FalconCLISmokeIT.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/falcon/cli/FalconCLISmokeIT.java b/webapp/src/test/java/org/apache/falcon/cli/FalconCLISmokeIT.java
index d503735..f946202 100644
--- a/webapp/src/test/java/org/apache/falcon/cli/FalconCLISmokeIT.java
+++ b/webapp/src/test/java/org/apache/falcon/cli/FalconCLISmokeIT.java
@@ -20,6 +20,7 @@ package org.apache.falcon.cli;
import org.apache.falcon.resource.TestContext;
import org.apache.falcon.util.OozieTestUtils;
+import org.apache.falcon.util.StartupProperties;
import org.testng.Assert;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@@ -36,6 +37,10 @@ public class FalconCLISmokeIT {
@BeforeClass
public void prepare() throws Exception {
TestContext.prepare();
+
+ String services = StartupProperties.get().getProperty("application.services");
+ StartupProperties.get().setProperty("application.services",
+ services + ",org.apache.falcon.metadata.MetadataMappingService");
}
@Test
@@ -47,18 +52,20 @@ public class FalconCLISmokeIT {
filePath = TestContext.overlayParametersOverTemplate(context.getClusterFileTemplate(), overlay);
Assert.assertEquals(-1,
- executeWithURL("entity -submitAndSchedule -type cluster -file "
- + filePath));
+ executeWithURL("entity -submitAndSchedule -type cluster -file " + filePath));
context.setCluster(overlay.get("cluster"));
+ // this is necessary for lineage
+ Assert.assertEquals(0, executeWithURL("entity -submit -type cluster -file " + filePath));
+
filePath = TestContext.overlayParametersOverTemplate(TestContext.FEED_TEMPLATE1, overlay);
Assert.assertEquals(0,
- executeWithURL("entity -submitAndSchedule -type feed -file "
- + filePath));
+ executeWithURL("entity -submitAndSchedule -type feed -file " + filePath));
+
filePath = TestContext.overlayParametersOverTemplate(TestContext.FEED_TEMPLATE2, overlay);
Assert.assertEquals(0,
- executeWithURL("entity -submitAndSchedule -type feed -file "
- + filePath));
+ executeWithURL("entity -submitAndSchedule -type feed -file " + filePath));
+
filePath = TestContext.overlayParametersOverTemplate(TestContext.FEED_TEMPLATE1, overlay);
Assert.assertEquals(0,
executeWithURL("entity -submit -type feed -file " + filePath));
@@ -69,8 +76,7 @@ public class FalconCLISmokeIT {
filePath = TestContext.overlayParametersOverTemplate(TestContext.PROCESS_TEMPLATE, overlay);
Assert.assertEquals(0,
- executeWithURL("entity -validate -type process -file "
- + filePath));
+ executeWithURL("entity -validate -type process -file " + filePath));
filePath = TestContext.overlayParametersOverTemplate(TestContext.PROCESS_TEMPLATE, overlay);
Assert.assertEquals(0,
@@ -80,17 +86,14 @@ public class FalconCLISmokeIT {
OozieTestUtils.waitForProcessWFtoStart(context);
Assert.assertEquals(0,
- executeWithURL("entity -definition -type cluster -name "
- + overlay.get("cluster")));
+ executeWithURL("entity -definition -type cluster -name " + overlay.get("cluster")));
Assert.assertEquals(0,
executeWithURL("instance -status -type feed -name "
- + overlay.get("outputFeedName")
- + " -start " + START_INSTANCE));
+ + overlay.get("outputFeedName") + " -start " + START_INSTANCE));
Assert.assertEquals(0,
- executeWithURL("instance -running -type process -name "
- + overlay.get("processName")));
+ executeWithURL("instance -running -type process -name " + overlay.get("processName")));
}
private int executeWithURL(String command) throws Exception {
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/webapp/src/test/java/org/apache/falcon/resource/EntityManagerJerseySmokeIT.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/falcon/resource/EntityManagerJerseySmokeIT.java b/webapp/src/test/java/org/apache/falcon/resource/EntityManagerJerseySmokeIT.java
index 082f541..dceb2f2 100644
--- a/webapp/src/test/java/org/apache/falcon/resource/EntityManagerJerseySmokeIT.java
+++ b/webapp/src/test/java/org/apache/falcon/resource/EntityManagerJerseySmokeIT.java
@@ -63,6 +63,7 @@ public class EntityManagerJerseySmokeIT {
contexts.remove();
}
+ @Test (dependsOnMethods = "testFeedSchedule")
public void testProcessDeleteAndSchedule() throws Exception {
//Submit process with invalid property so that coord submit fails and bundle goes to failed state
TestContext context = newContext();
@@ -101,6 +102,7 @@ public class EntityManagerJerseySmokeIT {
Assert.assertEquals(bundles.size(), 2);
}
+ @Test
public void testFeedSchedule() throws Exception {
TestContext context = newContext();
ClientResponse response;
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/webapp/src/test/java/org/apache/falcon/resource/TestContext.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/falcon/resource/TestContext.java b/webapp/src/test/java/org/apache/falcon/resource/TestContext.java
index b7a2256..a1e088d 100644
--- a/webapp/src/test/java/org/apache/falcon/resource/TestContext.java
+++ b/webapp/src/test/java/org/apache/falcon/resource/TestContext.java
@@ -342,6 +342,11 @@ public class TestContext {
// setup a logged in user
CurrentUser.authenticate(REMOTE_USER);
+ // disable recording lineage metadata if enabled
+ String services = StartupProperties.get().getProperty("application.services");
+ StartupProperties.get().setProperty("application.services",
+ services.replace("org.apache.falcon.metadata.MetadataMappingService", ""));
+
Map<String, String> overlay = new HashMap<String, String>();
overlay.put("cluster", RandomStringUtils.randomAlphabetic(5));
overlay.put("colo", "gs");
[3/6] git commit: FALCON-287 Record lineage information in post
processing. Contributed by Venkatesh Seetharam
Posted by ve...@apache.org.
FALCON-287 Record lineage information in post processing. Contributed by Venkatesh Seetharam
Project: http://git-wip-us.apache.org/repos/asf/incubator-falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-falcon/commit/59c8d500
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/59c8d500
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/59c8d500
Branch: refs/heads/master
Commit: 59c8d5001109dc8485b55bd0500ecf2d1230faf4
Parents: bb5e144
Author: Venkatesh Seetharam <ve...@hortonworks.com>
Authored: Wed Mar 5 16:00:17 2014 -0800
Committer: Venkatesh Seetharam <ve...@hortonworks.com>
Committed: Wed Mar 5 16:00:17 2014 -0800
----------------------------------------------------------------------
CHANGES.txt | 8 +-
common/pom.xml | 20 +++
.../org/apache/falcon/metadata/LineageArgs.java | 81 +++++++++++
.../apache/falcon/metadata/LineageRecorder.java | 144 +++++++++++++++++++
.../falcon/metadata/LineageRecorderTest.java | 103 +++++++++++++
.../config/workflow/replication-workflow.xml | 10 ++
.../config/workflow/retention-workflow.xml | 10 ++
.../falcon/messaging/EntityInstanceMessage.java | 3 +-
.../falcon/messaging/MessageProducer.java | 2 +
.../falcon/workflow/FalconPostProcessing.java | 33 ++++-
.../workflow/FalconPostProcessingTest.java | 95 ++++++------
.../config/workflow/process-parent-workflow.xml | 8 ++
12 files changed, 460 insertions(+), 57 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/59c8d500/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 5d7eb6f..e93d7b4 100755
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -53,6 +53,11 @@ Trunk (Unreleased)
FALCON-242 Post processing is not called in Retention workflows (Shaik Idris)
+ FALCON-286 Capture information in process entity about the user workflow
+ (Venkatesh Seetharam)
+
+ FALCON-287 Record lineage information in post processing (Venkatesh Seetharam)
+
OPTIMIZATIONS
FALCON-123 Improve build speeds in falcon. (Srikanth Sundarrajan via Shwetha GS)
@@ -222,9 +227,6 @@ Release Version: 0.4-incubating
FALCON-198 Update LICENSE.txt to contain license information for all third-party libraries
- FALCON-286 Capture information in process entity about the user workflow
- (Venkatesh Seetharam)
-
BUG FIXES
FALCON-223: hive-exec bundles protobuf-2.4.1 which is incompatible with hadoop-2 requiring protobuf-2.5
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/59c8d500/common/pom.xml
----------------------------------------------------------------------
diff --git a/common/pom.xml b/common/pom.xml
index c55c989..3705964 100644
--- a/common/pom.xml
+++ b/common/pom.xml
@@ -176,6 +176,26 @@
<groupId>org.apache.hcatalog</groupId>
<artifactId>webhcat-java-client</artifactId>
</dependency>
+
+ <dependency>
+ <groupId>com.googlecode.json-simple</groupId>
+ <artifactId>json-simple</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>com.tinkerpop.blueprints</groupId>
+ <artifactId>blueprints-core</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>com.thinkaurelius.titan</groupId>
+ <artifactId>titan-core-jre6</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>com.thinkaurelius.titan</groupId>
+ <artifactId>titan-berkeleyje-jre6</artifactId>
+ </dependency>
</dependencies>
<build>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/59c8d500/common/src/main/java/org/apache/falcon/metadata/LineageArgs.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/LineageArgs.java b/common/src/main/java/org/apache/falcon/metadata/LineageArgs.java
new file mode 100644
index 0000000..7adf5bb
--- /dev/null
+++ b/common/src/main/java/org/apache/falcon/metadata/LineageArgs.java
@@ -0,0 +1,81 @@
+/**
+ * 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.falcon.metadata;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.Option;
+
+/**
+ * Args for data Lineage.
+ */
+public enum LineageArgs {
+ // process instance
+ NOMINAL_TIME("nominalTime", "instance time"),
+ ENTITY_TYPE("entityType", "type of the entity"),
+ ENTITY_NAME("entityName", "name of the entity"),
+ TIMESTAMP("timeStamp", "current timestamp"),
+
+ // where
+ CLUSTER("cluster", "name of the current cluster"),
+ OPERATION("operation", "operation like generate, delete, replicate"),
+
+ // who
+ WORKFLOW_USER("workflowUser", "user who owns the feed instance (partition)"),
+
+ // workflow details
+ WORKFLOW_ID("workflowId", "current workflow-id of the instance"),
+ RUN_ID("runId", "current run-id of the instance"),
+ STATUS("status", "status of the user workflow isnstance"),
+ WF_ENGINE_URL("workflowEngineUrl", "url of workflow engine server, ex:oozie"),
+ USER_SUBFLOW_ID("subflowId", "external id of user workflow"),
+ USER_WORKFLOW_ENGINE("userWorkflowEngine", "user workflow engine type"),
+ USER_WORKFLOW_NAME("userWorkflowName", "user workflow name"),
+ USER_WORKFLOW_VERSION("userWorkflowVersion", "user workflow version"),
+
+ // what inputs
+ INPUT_FEED_NAMES("falconInputFeeds", "name of the feeds which are used as inputs"),
+ INPUT_FEED_PATHS("falconInputPaths", "comma separated input feed instance paths"),
+
+ // what outputs
+ FEED_NAMES("feedNames", "name of the feeds which are generated/replicated/deleted"),
+ FEED_INSTANCE_PATHS("feedInstancePaths", "comma separated feed instance paths"),
+
+ // lineage data recorded
+ LOG_DIR("logDir", "log dir where lineage can be recorded");
+
+ private String name;
+ private String description;
+
+ LineageArgs(String name, String description) {
+ this.name = name;
+ this.description = description;
+ }
+
+ public Option getOption() {
+ return new Option(this.name, true, this.description);
+ }
+
+ public String getOptionName() {
+ return this.name;
+ }
+
+ public String getOptionValue(CommandLine cmd) {
+ return cmd.getOptionValue(this.name);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/59c8d500/common/src/main/java/org/apache/falcon/metadata/LineageRecorder.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/LineageRecorder.java b/common/src/main/java/org/apache/falcon/metadata/LineageRecorder.java
new file mode 100644
index 0000000..ae9d88a
--- /dev/null
+++ b/common/src/main/java/org/apache/falcon/metadata/LineageRecorder.java
@@ -0,0 +1,144 @@
+/**
+ * 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.falcon.metadata;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.GnuParser;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.falcon.FalconException;
+import org.apache.falcon.hadoop.HadoopClientFactory;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.log4j.Logger;
+import org.json.simple.JSONValue;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Utility called in the post process of oozie workflow to record lineage information.
+ */
+public class LineageRecorder extends Configured implements Tool {
+
+ private static final Logger LOG = Logger.getLogger(LineageRecorder.class);
+
+ public static void main(String[] args) throws Exception {
+ ToolRunner.run(new LineageRecorder(), args);
+ }
+
+ @Override
+ public int run(String[] arguments) throws Exception {
+ CommandLine command = getCommand(arguments);
+
+ LOG.info("Parsing lineage metadata from: " + command);
+ Map<String, String> lineageMetadata = getLineageMetadata(command);
+ LOG.info("Lineage Metadata: " + lineageMetadata);
+
+ String lineageFile = getFilePath(command.getOptionValue(LineageArgs.LOG_DIR.getOptionName()),
+ command.getOptionValue(LineageArgs.ENTITY_NAME.getOptionName())
+ );
+
+ LOG.info("Persisting lineage metadata to: " + lineageFile);
+ persistLineageMetadata(lineageMetadata, lineageFile);
+
+ return 0;
+ }
+
+ protected static CommandLine getCommand(String[] arguments) throws ParseException {
+
+ Options options = new Options();
+
+ for (LineageArgs arg : LineageArgs.values()) {
+ addOption(options, arg);
+ }
+
+ return new GnuParser().parse(options, arguments);
+ }
+
+ private static void addOption(Options options, LineageArgs arg) {
+ addOption(options, arg, true);
+ }
+
+ private static void addOption(Options options, LineageArgs arg, boolean isRequired) {
+ Option option = arg.getOption();
+ option.setRequired(isRequired);
+ options.addOption(option);
+ }
+
+ protected Map<String, String> getLineageMetadata(CommandLine command) {
+ Map<String, String> lineageMetadata = new HashMap<String, String>();
+
+ for (LineageArgs arg : LineageArgs.values()) {
+ lineageMetadata.put(arg.getOptionName(), arg.getOptionValue(command));
+ }
+
+ return lineageMetadata;
+ }
+
+ public static String getFilePath(String logDir, String entityName) {
+ return logDir + entityName + "-lineage.json";
+ }
+
+ protected void persistLineageMetadata(Map<String, String> lineageMetadata,
+ String lineageFile) throws IOException, FalconException {
+ OutputStream out = null;
+ Path file = new Path(lineageFile);
+ try {
+ FileSystem fs = HadoopClientFactory.get().createFileSystem(file.toUri(), getConf());
+ out = fs.create(file);
+
+ // making sure falcon can read this file
+ FsPermission permission = new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL);
+ fs.setPermission(file, permission);
+
+ out.write(JSONValue.toJSONString(lineageMetadata).getBytes());
+ } finally {
+ if (out != null) {
+ try {
+ out.close();
+ } catch (IOException ignore) {
+ // ignore
+ }
+ }
+ }
+ }
+
+ public static Map<String, String> parseLineageMetadata(String lineageFile) throws FalconException {
+ try {
+ Path lineageDataPath = new Path(lineageFile); // file has 777 permissions
+ FileSystem fs = HadoopClientFactory.get().createFileSystem(lineageDataPath.toUri());
+
+ BufferedReader in = new BufferedReader(new InputStreamReader(fs.open(lineageDataPath)));
+ return (Map<String, String>) JSONValue.parse(in);
+ } catch (IOException e) {
+ throw new FalconException("Error opening lineage file", e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/59c8d500/common/src/test/java/org/apache/falcon/metadata/LineageRecorderTest.java
----------------------------------------------------------------------
diff --git a/common/src/test/java/org/apache/falcon/metadata/LineageRecorderTest.java b/common/src/test/java/org/apache/falcon/metadata/LineageRecorderTest.java
new file mode 100644
index 0000000..d686ab6
--- /dev/null
+++ b/common/src/test/java/org/apache/falcon/metadata/LineageRecorderTest.java
@@ -0,0 +1,103 @@
+/**
+ * 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.falcon.metadata;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.util.Map;
+
+/**
+ * Unit test for lineage recorder.
+ */
+public class LineageRecorderTest {
+
+ private static final String LOGS_DIR = "target/log";
+ private static final String NOMINAL_TIME = "2014-01-01-01-00";
+ private static final String ENTITY_NAME = "test-process-entity";
+
+ private String[] args;
+
+ @BeforeMethod
+ public void setUp() throws Exception {
+ args = new String[]{
+ "-" + LineageArgs.ENTITY_NAME.getOptionName(), ENTITY_NAME,
+ "-" + LineageArgs.FEED_NAMES.getOptionName(), "out-click-logs,out-raw-logs",
+ "-" + LineageArgs.FEED_INSTANCE_PATHS.getOptionName(),
+ "/out-click-logs/10/05/05/00/20,/out-raw-logs/10/05/05/00/20",
+ "-" + LineageArgs.WORKFLOW_ID.getOptionName(), "workflow-01-00",
+ "-" + LineageArgs.WORKFLOW_USER.getOptionName(), "falcon",
+ "-" + LineageArgs.RUN_ID.getOptionName(), "1",
+ "-" + LineageArgs.NOMINAL_TIME.getOptionName(), NOMINAL_TIME,
+ "-" + LineageArgs.TIMESTAMP.getOptionName(), "2014-01-01-01-00",
+ "-" + LineageArgs.ENTITY_TYPE.getOptionName(), ("process"),
+ "-" + LineageArgs.OPERATION.getOptionName(), ("GENERATE"),
+ "-" + LineageArgs.STATUS.getOptionName(), ("SUCCEEDED"),
+ "-" + LineageArgs.CLUSTER.getOptionName(), "corp",
+ "-" + LineageArgs.WF_ENGINE_URL.getOptionName(), "http://localhost:11000/oozie/",
+ "-" + LineageArgs.LOG_DIR.getOptionName(), LOGS_DIR,
+ "-" + LineageArgs.USER_SUBFLOW_ID.getOptionName(), "userflow@wf-id" + "test",
+ "-" + LineageArgs.USER_WORKFLOW_ENGINE.getOptionName(), "oozie",
+ "-" + LineageArgs.INPUT_FEED_NAMES.getOptionName(), "in-click-logs,in-raw-logs",
+ "-" + LineageArgs.INPUT_FEED_PATHS.getOptionName(),
+ "/in-click-logs/10/05/05/00/20,/in-raw-logs/10/05/05/00/20",
+ "-" + LineageArgs.USER_WORKFLOW_NAME.getOptionName(), "test-workflow",
+ "-" + LineageArgs.USER_WORKFLOW_VERSION.getOptionName(), "1.0.0",
+ };
+ }
+
+ @AfterMethod
+ public void tearDown() throws Exception {
+
+ }
+
+ @Test
+ public void testMain() throws Exception {
+ LineageRecorder lineageRecorder = new LineageRecorder();
+ CommandLine command = LineageRecorder.getCommand(args);
+ Map<String, String> lineageMetadata = lineageRecorder.getLineageMetadata(command);
+
+ LineageRecorder.main(args);
+
+ String lineageFile = LineageRecorder.getFilePath(LOGS_DIR, ENTITY_NAME);
+ Path lineageDataPath = new Path(lineageFile);
+ FileSystem fs = lineageDataPath.getFileSystem(new Configuration());
+ Assert.assertTrue(fs.exists(lineageDataPath));
+
+ Map<String, String> recordedLineageMetadata =
+ LineageRecorder.parseLineageMetadata(lineageFile);
+
+ for (Map.Entry<String, String> entry : lineageMetadata.entrySet()) {
+ Assert.assertEquals(lineageMetadata.get(entry.getKey()),
+ recordedLineageMetadata.get(entry.getKey()));
+ }
+ }
+
+ @Test
+ public void testGetFilePath() throws Exception {
+ String path = LOGS_DIR + ENTITY_NAME + "-lineage.json";
+ Assert.assertEquals(path, LineageRecorder.getFilePath(LOGS_DIR, ENTITY_NAME));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/59c8d500/feed/src/main/resources/config/workflow/replication-workflow.xml
----------------------------------------------------------------------
diff --git a/feed/src/main/resources/config/workflow/replication-workflow.xml b/feed/src/main/resources/config/workflow/replication-workflow.xml
index 3ee6f86..db1c0e5 100644
--- a/feed/src/main/resources/config/workflow/replication-workflow.xml
+++ b/feed/src/main/resources/config/workflow/replication-workflow.xml
@@ -214,12 +214,22 @@
<arg>${logDir}/instancePaths-${nominalTime}-${srcClusterName}.csv</arg>
<arg>-workflowEngineUrl</arg>
<arg>${workflowEngineUrl}</arg>
+ <arg>-userWorkflowName</arg>
+ <arg>${userWorkflowName}</arg>
+ <arg>-userWorkflowVersion</arg>
+ <arg>${userWorkflowVersion}</arg>
+ <arg>-userWorkflowEngine</arg>
+ <arg>${userWorkflowEngine}</arg>
<arg>-subflowId</arg>
<arg>${wf:id()}</arg>
<arg>-logDir</arg>
<arg>${logDir}/job-${nominalTime}/${srcClusterName}/</arg>
<arg>-workflowUser</arg>
<arg>${wf:user()}</arg>
+ <arg>-falconInputFeeds</arg>
+ <arg>${falconInputFeeds}</arg>
+ <arg>-falconInputPaths</arg>
+ <arg>${falconInPaths}</arg>
<file>${wf:conf("falcon.libpath")}/activemq-core.jar</file>
<file>${wf:conf("falcon.libpath")}/geronimo-j2ee-management.jar</file>
<file>${wf:conf("falcon.libpath")}/jms.jar</file>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/59c8d500/feed/src/main/resources/config/workflow/retention-workflow.xml
----------------------------------------------------------------------
diff --git a/feed/src/main/resources/config/workflow/retention-workflow.xml b/feed/src/main/resources/config/workflow/retention-workflow.xml
index 87fdc5a..1e10aae 100644
--- a/feed/src/main/resources/config/workflow/retention-workflow.xml
+++ b/feed/src/main/resources/config/workflow/retention-workflow.xml
@@ -117,6 +117,16 @@
<arg>${logDir}/job-${nominalTime}/</arg>
<arg>-workflowUser</arg>
<arg>${wf:user()}</arg>
+ <arg>-userWorkflowName</arg>
+ <arg>${userWorkflowName}</arg>
+ <arg>-userWorkflowVersion</arg>
+ <arg>${userWorkflowVersion}</arg>
+ <arg>-userWorkflowEngine</arg>
+ <arg>${userWorkflowEngine}</arg>
+ <arg>-falconInputFeeds</arg>
+ <arg>${falconInputFeeds}</arg>
+ <arg>-falconInputPaths</arg>
+ <arg>${falconInPaths}</arg>
<file>${wf:conf("falcon.libpath")}/activemq-core.jar</file>
<file>${wf:conf("falcon.libpath")}/geronimo-j2ee-management.jar</file>
<file>${wf:conf("falcon.libpath")}/jms.jar</file>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/59c8d500/messaging/src/main/java/org/apache/falcon/messaging/EntityInstanceMessage.java
----------------------------------------------------------------------
diff --git a/messaging/src/main/java/org/apache/falcon/messaging/EntityInstanceMessage.java b/messaging/src/main/java/org/apache/falcon/messaging/EntityInstanceMessage.java
index b6ff504..d3c1a69 100644
--- a/messaging/src/main/java/org/apache/falcon/messaging/EntityInstanceMessage.java
+++ b/messaging/src/main/java/org/apache/falcon/messaging/EntityInstanceMessage.java
@@ -71,7 +71,8 @@ public class EntityInstanceMessage {
status("status"),
brokerTTL("broker.ttlInMins"),
cluster("cluster"),
- workflowUser("workflowUser");
+ workflowUser("workflowUser"),
+ logDir("logDir");
private String propName;
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/59c8d500/messaging/src/main/java/org/apache/falcon/messaging/MessageProducer.java
----------------------------------------------------------------------
diff --git a/messaging/src/main/java/org/apache/falcon/messaging/MessageProducer.java b/messaging/src/main/java/org/apache/falcon/messaging/MessageProducer.java
index 0f9e918..2b9f573 100644
--- a/messaging/src/main/java/org/apache/falcon/messaging/MessageProducer.java
+++ b/messaging/src/main/java/org/apache/falcon/messaging/MessageProducer.java
@@ -125,6 +125,8 @@ public class MessageProducer extends Configured implements Tool {
"cluster name"));
addOption(options, new Option(ARG.workflowUser.getArgName(), true,
"workflow user id"), false);
+ addOption(options, new Option(ARG.logDir.getArgName(), true,
+ "log dir where job logs are copied"), false);
return new GnuParser().parse(options, arguments);
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/59c8d500/oozie/src/main/java/org/apache/falcon/workflow/FalconPostProcessing.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/workflow/FalconPostProcessing.java b/oozie/src/main/java/org/apache/falcon/workflow/FalconPostProcessing.java
index c19c89c..6584dfe 100644
--- a/oozie/src/main/java/org/apache/falcon/workflow/FalconPostProcessing.java
+++ b/oozie/src/main/java/org/apache/falcon/workflow/FalconPostProcessing.java
@@ -21,6 +21,8 @@ import org.apache.commons.cli.*;
import org.apache.commons.lang.StringUtils;
import org.apache.falcon.logging.LogMover;
import org.apache.falcon.messaging.MessageProducer;
+import org.apache.falcon.metadata.LineageArgs;
+import org.apache.falcon.metadata.LineageRecorder;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.security.UserGroupInformation;
@@ -48,7 +50,7 @@ public class FalconPostProcessing extends Configured implements Tool {
OPERATION("operation", "operation like generate, delete, replicate"),
WORKFLOW_ID("workflowId", "current workflow-id of the instance"),
RUN_ID("runId", "current run-id of the instance"),
- STATUS("status", "status of the user workflow isnstance"),
+ STATUS("status", "status of the user workflow instance"),
TIMESTAMP("timeStamp", "current timestamp"),
TOPIC_NAME("topicName", "name of the topic to be used to send JMS message"),
BRKR_IMPL_CLASS("brokerImplClass", "falcon message broker Implementation class"),
@@ -62,8 +64,12 @@ public class FalconPostProcessing extends Configured implements Tool {
WF_ENGINE_URL("workflowEngineUrl", "url of workflow engine server, ex:oozie"),
USER_SUBFLOW_ID("subflowId", "external id of user workflow"),
USER_WORKFLOW_ENGINE("userWorkflowEngine", "user workflow engine type"),
+ USER_WORKFLOW_NAME("userWorkflowName", "user workflow name"),
+ USER_WORKFLOW_VERSION("userWorkflowVersion", "user workflow version"),
LOG_DIR("logDir", "log dir where job logs are copied"),
- WORKFLOW_USER("workflowUser", "user who owns the feed instance (partition)");
+ WORKFLOW_USER("workflowUser", "user who owns the feed instance (partition)"),
+ INPUT_FEED_NAMES("falconInputFeeds", "name of the feeds which are used as inputs"),
+ INPUT_FEED_PATHS("falconInputPaths", "comma separated input feed instance paths");
private String name;
private String description;
@@ -105,6 +111,11 @@ public class FalconPostProcessing extends Configured implements Tool {
LOG.info("Sending falcon message " + cmd);
invokeFalconMessageProducer(cmd);
+ if ("SUCCEEDED".equals(Arg.STATUS.getOptionValue(cmd))) {
+ LOG.info("Recording lineage for " + cmd);
+ recordLineageMetadata(cmd);
+ }
+
return 0;
}
@@ -158,6 +169,7 @@ public class FalconPostProcessing extends Configured implements Tool {
addArg(args, cmd, Arg.FEED_INSTANCE_PATHS);
addArg(args, cmd, Arg.LOG_FILE);
addArg(args, cmd, Arg.WORKFLOW_USER);
+ addArg(args, cmd, Arg.LOG_DIR);
MessageProducer.main(args.toArray(new String[0]));
}
@@ -181,6 +193,19 @@ public class FalconPostProcessing extends Configured implements Tool {
LogMover.main(args.toArray(new String[0]));
}
+ private void recordLineageMetadata(CommandLine cmd) throws Exception {
+ List<String> args = new ArrayList<String>();
+
+ for (LineageArgs arg : LineageArgs.values()) {
+ if (StringUtils.isNotEmpty(arg.getOptionValue(cmd))) {
+ args.add("-" + arg.getOptionName());
+ args.add(arg.getOptionValue(cmd));
+ }
+ }
+
+ LineageRecorder.main(args.toArray(new String[args.size()]));
+ }
+
private void addArg(List<String> args, CommandLine cmd, Arg arg) {
if (StringUtils.isNotEmpty(arg.getOptionValue(cmd))) {
args.add("-" + arg.getOptionName());
@@ -211,9 +236,13 @@ public class FalconPostProcessing extends Configured implements Tool {
addOption(options, Arg.LOG_FILE);
addOption(options, Arg.WF_ENGINE_URL);
addOption(options, Arg.USER_SUBFLOW_ID);
+ addOption(options, Arg.USER_WORKFLOW_NAME, false);
+ addOption(options, Arg.USER_WORKFLOW_VERSION, false);
addOption(options, Arg.USER_WORKFLOW_ENGINE, false);
addOption(options, Arg.LOG_DIR);
addOption(options, Arg.WORKFLOW_USER);
+ addOption(options, Arg.INPUT_FEED_NAMES, false);
+ addOption(options, Arg.INPUT_FEED_PATHS, false);
return new GnuParser().parse(options, arguments);
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/59c8d500/oozie/src/test/java/org/apache/falcon/oozie/workflow/FalconPostProcessingTest.java
----------------------------------------------------------------------
diff --git a/oozie/src/test/java/org/apache/falcon/oozie/workflow/FalconPostProcessingTest.java b/oozie/src/test/java/org/apache/falcon/oozie/workflow/FalconPostProcessingTest.java
index 871c63f..561303d 100644
--- a/oozie/src/test/java/org/apache/falcon/oozie/workflow/FalconPostProcessingTest.java
+++ b/oozie/src/test/java/org/apache/falcon/oozie/workflow/FalconPostProcessingTest.java
@@ -35,8 +35,6 @@ public class FalconPostProcessingTest {
private String[] args;
private static final String BROKER_URL = "vm://localhost?broker.useJmx=false&broker.persistent=true";
- // private static final String BROKER_URL =
- // "tcp://localhost:61616?daemon=true";
private static final String BROKER_IMPL_CLASS = "org.apache.activemq.ActiveMQConnectionFactory";
private static final String FALCON_TOPIC_NAME = "FALCON.ENTITY.TOPIC";
private static final String ENTITY_NAME = "agg-coord";
@@ -46,31 +44,37 @@ public class FalconPostProcessingTest {
@BeforeClass
public void setup() throws Exception {
- args = new String[]{"-" + Arg.ENTITY_NAME.getOptionName(),
- ENTITY_NAME, "-" + Arg.FEED_NAMES.getOptionName(),
- "click-logs,raw-logs",
- "-" + Arg.FEED_INSTANCE_PATHS.getOptionName(),
- "/click-logs/10/05/05/00/20,/raw-logs/10/05/05/00/20",
- "-" + Arg.WORKFLOW_ID.getOptionName(), "workflow-01-00",
- "-" + Arg.WORKFLOW_USER.getOptionName(), "falcon",
- "-" + Arg.RUN_ID.getOptionName(), "1",
- "-" + Arg.NOMINAL_TIME.getOptionName(), "2011-01-01-01-00",
- "-" + Arg.TIMESTAMP.getOptionName(), "2012-01-01-01-00",
- "-" + Arg.BRKR_URL.getOptionName(), BROKER_URL,
- "-" + Arg.BRKR_IMPL_CLASS.getOptionName(), (BROKER_IMPL_CLASS),
- "-" + Arg.USER_BRKR_URL.getOptionName(), BROKER_URL,
- "-" + Arg.USER_BRKR_IMPL_CLASS.getOptionName(),
- (BROKER_IMPL_CLASS), "-" + Arg.ENTITY_TYPE.getOptionName(),
- ("process"), "-" + Arg.OPERATION.getOptionName(), ("GENERATE"),
- "-" + Arg.LOG_FILE.getOptionName(), ("/logFile"),
- "-" + Arg.STATUS.getOptionName(), ("SUCCEEDED"),
- "-" + Arg.BRKR_TTL.getOptionName(), "10",
- "-" + Arg.CLUSTER.getOptionName(), "corp",
- "-" + Arg.WF_ENGINE_URL.getOptionName(),
- "http://localhost:11000/oozie/",
- "-" + Arg.LOG_DIR.getOptionName(), "target/log",
- "-" + Arg.USER_SUBFLOW_ID.getOptionName(), "userflow@wf-id" + "test",
- "-" + Arg.USER_WORKFLOW_ENGINE.getOptionName(), "oozie", };
+ args = new String[]{
+ "-" + Arg.ENTITY_NAME.getOptionName(), ENTITY_NAME,
+ "-" + Arg.FEED_NAMES.getOptionName(), "out-click-logs,out-raw-logs",
+ "-" + Arg.FEED_INSTANCE_PATHS.getOptionName(),
+ "/out-click-logs/10/05/05/00/20,/out-raw-logs/10/05/05/00/20",
+ "-" + Arg.WORKFLOW_ID.getOptionName(), "workflow-01-00",
+ "-" + Arg.WORKFLOW_USER.getOptionName(), "falcon",
+ "-" + Arg.RUN_ID.getOptionName(), "1",
+ "-" + Arg.NOMINAL_TIME.getOptionName(), "2011-01-01-01-00",
+ "-" + Arg.TIMESTAMP.getOptionName(), "2012-01-01-01-00",
+ "-" + Arg.BRKR_URL.getOptionName(), BROKER_URL,
+ "-" + Arg.BRKR_IMPL_CLASS.getOptionName(), (BROKER_IMPL_CLASS),
+ "-" + Arg.USER_BRKR_URL.getOptionName(), BROKER_URL,
+ "-" + Arg.USER_BRKR_IMPL_CLASS.getOptionName(), (BROKER_IMPL_CLASS),
+ "-" + Arg.ENTITY_TYPE.getOptionName(), ("process"),
+ "-" + Arg.OPERATION.getOptionName(), ("GENERATE"),
+ "-" + Arg.LOG_FILE.getOptionName(), ("/logFile"),
+ "-" + Arg.STATUS.getOptionName(), ("SUCCEEDED"),
+ "-" + Arg.BRKR_TTL.getOptionName(), "10",
+ "-" + Arg.CLUSTER.getOptionName(), "corp",
+ "-" + Arg.WF_ENGINE_URL.getOptionName(), "http://localhost:11000/oozie/",
+ "-" + Arg.LOG_DIR.getOptionName(), "target/log",
+ "-" + Arg.USER_SUBFLOW_ID.getOptionName(), "userflow@wf-id" + "test",
+ "-" + Arg.USER_WORKFLOW_ENGINE.getOptionName(), "oozie",
+ "-" + Arg.INPUT_FEED_NAMES.getOptionName(), "in-click-logs,in-raw-logs",
+ "-" + Arg.INPUT_FEED_PATHS.getOptionName(),
+ "/in-click-logs/10/05/05/00/20,/in-raw-logs/10/05/05/00/20",
+ "-" + Arg.USER_WORKFLOW_NAME.getOptionName(), "test-workflow",
+ "-" + Arg.USER_WORKFLOW_VERSION.getOptionName(), "1.0.0",
+ };
+
broker = new BrokerService();
broker.addConnector(BROKER_URL);
broker.setDataDirectory("target/activemq");
@@ -110,13 +114,11 @@ public class FalconPostProcessingTest {
}
private void consumer(String brokerUrl, String topic) throws JMSException {
- ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
- brokerUrl);
+ ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerUrl);
Connection connection = connectionFactory.createConnection();
connection.start();
- Session session = connection.createSession(false,
- Session.AUTO_ACKNOWLEDGE);
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createTopic(topic);
MessageConsumer consumer = session.createConsumer(destination);
@@ -130,37 +132,28 @@ public class FalconPostProcessingTest {
assertMessage(m);
if (topic.equals(FALCON_TOPIC_NAME)) {
Assert.assertEquals(m.getString(Arg.FEED_NAMES.getOptionName()),
- "click-logs,raw-logs");
- Assert.assertEquals(
- m.getString(Arg.FEED_INSTANCE_PATHS.getOptionName()),
- "/click-logs/10/05/05/00/20,/raw-logs/10/05/05/00/20");
+ "out-click-logs,out-raw-logs");
+ Assert.assertEquals(m.getString(Arg.FEED_INSTANCE_PATHS.getOptionName()),
+ "/out-click-logs/10/05/05/00/20,/out-raw-logs/10/05/05/00/20");
} else {
- Assert.assertEquals(m.getString(Arg.FEED_NAMES.getOptionName()),
- "click-logs");
- Assert.assertEquals(
- m.getString(Arg.FEED_INSTANCE_PATHS.getOptionName()),
- "/click-logs/10/05/05/00/20");
-
+ Assert.assertEquals(m.getString(Arg.FEED_NAMES.getOptionName()), "out-click-logs");
+ Assert.assertEquals(m.getString(Arg.FEED_INSTANCE_PATHS.getOptionName()),
+ "/out-click-logs/10/05/05/00/20");
}
connection.close();
}
private void assertMessage(MapMessage m) throws JMSException {
- Assert.assertEquals(m.getString(Arg.ENTITY_NAME.getOptionName()),
- "agg-coord");
- Assert.assertEquals(m.getString(Arg.WORKFLOW_ID.getOptionName()),
- "workflow-01-00");
+ Assert.assertEquals(m.getString(Arg.ENTITY_NAME.getOptionName()), "agg-coord");
+ Assert.assertEquals(m.getString(Arg.WORKFLOW_ID.getOptionName()), "workflow-01-00");
String workflowUser = m.getString(Arg.WORKFLOW_USER.getOptionName());
if (workflowUser != null) { // in case of user message, its NULL
Assert.assertEquals(workflowUser, "falcon");
}
Assert.assertEquals(m.getString(Arg.RUN_ID.getOptionName()), "1");
- Assert.assertEquals(m.getString(Arg.NOMINAL_TIME.getOptionName()),
- "2011-01-01T01:00Z");
- Assert.assertEquals(m.getString(Arg.TIMESTAMP.getOptionName()),
- "2012-01-01T01:00Z");
- Assert.assertEquals(m.getString(Arg.STATUS.getOptionName()),
- "SUCCEEDED");
+ Assert.assertEquals(m.getString(Arg.NOMINAL_TIME.getOptionName()), "2011-01-01T01:00Z");
+ Assert.assertEquals(m.getString(Arg.TIMESTAMP.getOptionName()), "2012-01-01T01:00Z");
+ Assert.assertEquals(m.getString(Arg.STATUS.getOptionName()), "SUCCEEDED");
}
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/59c8d500/process/src/main/resources/config/workflow/process-parent-workflow.xml
----------------------------------------------------------------------
diff --git a/process/src/main/resources/config/workflow/process-parent-workflow.xml b/process/src/main/resources/config/workflow/process-parent-workflow.xml
index 838215f..30496a9 100644
--- a/process/src/main/resources/config/workflow/process-parent-workflow.xml
+++ b/process/src/main/resources/config/workflow/process-parent-workflow.xml
@@ -187,10 +187,18 @@
<arg>${wf:id()}${userWorkflowEngine == "oozie" ? "@user-oozie-workflow" : ""}</arg>
<arg>-userWorkflowEngine</arg>
<arg>${userWorkflowEngine}</arg>
+ <arg>-userWorkflowName</arg>
+ <arg>${userWorkflowName}</arg>
+ <arg>-userWorkflowVersion</arg>
+ <arg>${userWorkflowVersion}</arg>
<arg>-logDir</arg>
<arg>${logDir}/job-${nominalTime}/</arg>
<arg>-workflowUser</arg>
<arg>${wf:user()}</arg>
+ <arg>-falconInputFeeds</arg>
+ <arg>${falconInputFeeds}</arg>
+ <arg>-falconInputPaths</arg>
+ <arg>${falconInPaths}</arg>
<file>${wf:conf("falcon.libpath")}/activemq-core.jar</file>
<file>${wf:conf("falcon.libpath")}/geronimo-j2ee-management.jar</file>
<file>${wf:conf("falcon.libpath")}/jms.jar</file>
[2/6] git commit: FALCON-286 Capture information in process entity
about the user workflow. Contributed by Venkatesh Seetharam
Posted by ve...@apache.org.
FALCON-286 Capture information in process entity about the user workflow. Contributed by Venkatesh Seetharam
Project: http://git-wip-us.apache.org/repos/asf/incubator-falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-falcon/commit/bb5e1449
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/bb5e1449
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/bb5e1449
Branch: refs/heads/master
Commit: bb5e144970cf0336be859e5576afd8908861e687
Parents: 42d04f3
Author: Venkatesh Seetharam <ve...@hortonworks.com>
Authored: Wed Mar 5 15:56:59 2014 -0800
Committer: Venkatesh Seetharam <ve...@hortonworks.com>
Committed: Wed Mar 5 15:56:59 2014 -0800
----------------------------------------------------------------------
CHANGES.txt | 3 +
client/src/main/resources/process-0.1.xsd | 2 +
.../org/apache/falcon/entity/ProcessHelper.java | 5 ++
.../falcon/converter/OozieFeedMapper.java | 22 ++++++-
.../falcon/converter/OozieFeedMapperTest.java | 9 ++-
.../falcon/converter/OozieProcessMapper.java | 30 +++++++++-
.../converter/OozieProcessMapperTest.java | 63 +++++++++++++++++++-
.../resources/config/process/dumb-process.xml | 40 +++++++++++++
.../resources/config/process/process-0.1.xml | 2 +-
9 files changed, 166 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/bb5e1449/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 6b536b8..5d7eb6f 100755
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -222,6 +222,9 @@ Release Version: 0.4-incubating
FALCON-198 Update LICENSE.txt to contain license information for all third-party libraries
+ FALCON-286 Capture information in process entity about the user workflow
+ (Venkatesh Seetharam)
+
BUG FIXES
FALCON-223: hive-exec bundles protobuf-2.4.1 which is incompatible with hadoop-2 requiring protobuf-2.5
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/bb5e1449/client/src/main/resources/process-0.1.xsd
----------------------------------------------------------------------
diff --git a/client/src/main/resources/process-0.1.xsd b/client/src/main/resources/process-0.1.xsd
index 701fa51..f98822d 100644
--- a/client/src/main/resources/process-0.1.xsd
+++ b/client/src/main/resources/process-0.1.xsd
@@ -283,6 +283,8 @@
</xs:complexType>
<xs:complexType name="workflow">
+ <xs:attribute type="xs:string" name="name" use="optional"/>
+ <xs:attribute type="xs:string" name="version" use="optional" default="1.0"/>
<xs:attribute type="engine-type" name="engine" use="optional" default="oozie"/>
<xs:attribute type="xs:string" name="path" use="required"/>
<xs:attribute type="xs:string" name="lib" use="optional"/>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/bb5e1449/common/src/main/java/org/apache/falcon/entity/ProcessHelper.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/entity/ProcessHelper.java b/common/src/main/java/org/apache/falcon/entity/ProcessHelper.java
index dc5ae7a..46e7384 100644
--- a/common/src/main/java/org/apache/falcon/entity/ProcessHelper.java
+++ b/common/src/main/java/org/apache/falcon/entity/ProcessHelper.java
@@ -18,6 +18,7 @@
package org.apache.falcon.entity;
+import org.apache.commons.lang.StringUtils;
import org.apache.falcon.entity.v0.process.Cluster;
import org.apache.falcon.entity.v0.process.Process;
@@ -36,4 +37,8 @@ public final class ProcessHelper {
}
return null;
}
+
+ public static String getProcessWorkflowName(String workflowName, String processName) {
+ return StringUtils.isEmpty(workflowName) ? processName + "-workflow" : workflowName;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/bb5e1449/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
----------------------------------------------------------------------
diff --git a/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java b/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
index d6dee77..e589c02 100644
--- a/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
+++ b/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
@@ -46,6 +46,7 @@ import org.apache.falcon.oozie.coordinator.COORDINATORAPP;
import org.apache.falcon.oozie.coordinator.SYNCDATASET;
import org.apache.falcon.oozie.coordinator.WORKFLOW;
import org.apache.falcon.oozie.workflow.WORKFLOWAPP;
+import org.apache.falcon.util.BuildProperties;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -189,6 +190,11 @@ public class OozieFeedMapper extends AbstractOozieEntityMapper<Feed> {
props.put(ARG.feedNames.getPropName(), feed.getName());
props.put(ARG.feedInstancePaths.getPropName(), "IGNORE");
+ props.put("falconInputFeeds", feed.getName());
+ props.put("falconInPaths", "IGNORE");
+
+ propagateUserWorkflowProperties(props, "eviction");
+
retentionWorkflow.setConfiguration(getCoordConfig(props));
retentionAction.setWorkflow(retentionWorkflow);
@@ -432,6 +438,7 @@ public class OozieFeedMapper extends AbstractOozieEntityMapper<Feed> {
}
propagateLateDataProperties(feed, instancePaths, sourceStorage.getType().name(), props);
+ propagateUserWorkflowProperties(props, "replication");
replicationWF.setConfiguration(getCoordConfig(props));
replicationAction.setWorkflow(replicationWF);
@@ -547,7 +554,7 @@ public class OozieFeedMapper extends AbstractOozieEntityMapper<Feed> {
// falcon post processing
props.put(ARG.feedNames.getPropName(), feed.getName());
- props.put(ARG.feedInstancePaths.getPropName(), instancePaths);
+ props.put(ARG.feedInstancePaths.getPropName(), "${coord:dataOut('output')}");
}
}
@@ -563,4 +570,17 @@ public class OozieFeedMapper extends AbstractOozieEntityMapper<Feed> {
}
}
}
+
+ private void propagateUserWorkflowProperties(Map<String, String> props, String policy) {
+ props.put("userWorkflowName", policy + "-policy");
+ props.put("userWorkflowEngine", "falcon");
+
+ String version;
+ try {
+ version = BuildProperties.get().getProperty("build.version");
+ } catch (Exception e) { // unfortunate that this is only available in prism/webapp
+ version = "0.5";
+ }
+ props.put("userWorkflowVersion", version);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/bb5e1449/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
----------------------------------------------------------------------
diff --git a/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java b/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
index a153462..671c53c 100644
--- a/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
+++ b/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
@@ -202,7 +202,12 @@ public class OozieFeedMapperTest {
// verify the post processing params
Assert.assertEquals(props.get("feedNames"), feed.getName());
- Assert.assertEquals(props.get("feedInstancePaths"), "${coord:dataIn('input')}");
+ Assert.assertEquals(props.get("feedInstancePaths"), "${coord:dataOut('output')}");
+
+ // verify workflow params
+ Assert.assertEquals("replication-policy", props.get("userWorkflowName"));
+ Assert.assertEquals("0.5", props.get("userWorkflowVersion"));
+ Assert.assertEquals("falcon", props.get("userWorkflowEngine"));
assertLibExtensions(coord, "replication");
assertWorkflowRetries(coord);
@@ -434,7 +439,7 @@ public class OozieFeedMapperTest {
// verify the post processing params
Assert.assertEquals(props.get("feedNames"), tableFeed.getName());
- Assert.assertEquals(props.get("feedInstancePaths"), "${coord:dataIn('input')}");
+ Assert.assertEquals(props.get("feedInstancePaths"), "${coord:dataOut('output')}");
}
private void assertTableStorageProperties(Cluster cluster, CatalogStorage tableStorage,
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/bb5e1449/process/src/main/java/org/apache/falcon/converter/OozieProcessMapper.java
----------------------------------------------------------------------
diff --git a/process/src/main/java/org/apache/falcon/converter/OozieProcessMapper.java b/process/src/main/java/org/apache/falcon/converter/OozieProcessMapper.java
index 6d0297e..e638961 100644
--- a/process/src/main/java/org/apache/falcon/converter/OozieProcessMapper.java
+++ b/process/src/main/java/org/apache/falcon/converter/OozieProcessMapper.java
@@ -190,7 +190,7 @@ public class OozieProcessMapper extends AbstractOozieEntityMapper<Process> {
initializeOutputPaths(cluster, process, coord, props); // outputs
Workflow processWorkflow = process.getWorkflow();
- props.put("userWorkflowEngine", processWorkflow.getEngine().value());
+ propagateUserWorkflowProperties(processWorkflow, props, process.getName());
// create parent wf
createWorkflow(cluster, process, processWorkflow, coordName, coordPath);
@@ -247,6 +247,8 @@ public class OozieProcessMapper extends AbstractOozieEntityMapper<Process> {
private void initializeInputPaths(Cluster cluster, Process process, COORDINATORAPP coord,
Map<String, String> props) throws FalconException {
if (process.getInputs() == null) {
+ props.put("falconInputFeeds", "NONE");
+ props.put("falconInPaths", "IGNORE");
return;
}
@@ -281,7 +283,7 @@ public class OozieProcessMapper extends AbstractOozieEntityMapper<Process> {
propagateCatalogTableProperties(input, (CatalogStorage) storage, props);
}
- inputFeeds.add(input.getName());
+ inputFeeds.add(feed.getName());
inputPaths.add(inputExpr);
inputFeedStorageTypes.add(storage.getType().name());
}
@@ -303,6 +305,8 @@ public class OozieProcessMapper extends AbstractOozieEntityMapper<Process> {
private void initializeOutputPaths(Cluster cluster, Process process, COORDINATORAPP coord,
Map<String, String> props) throws FalconException {
if (process.getOutputs() == null) {
+ props.put(ARG.feedNames.getPropName(), "NONE");
+ props.put(ARG.feedInstancePaths.getPropName(), "IGNORE");
return;
}
@@ -327,7 +331,7 @@ public class OozieProcessMapper extends AbstractOozieEntityMapper<Process> {
coord.getOutputEvents().getDataOut().add(dataout);
String outputExpr = "${coord:dataOut('" + output.getName() + "')}";
- outputFeeds.add(output.getName());
+ outputFeeds.add(feed.getName());
outputPaths.add(outputExpr);
if (storage.getType() == Storage.TYPE.FILESYSTEM) {
@@ -486,6 +490,14 @@ public class OozieProcessMapper extends AbstractOozieEntityMapper<Process> {
return props;
}
+ private void propagateUserWorkflowProperties(Workflow processWorkflow,
+ Map<String, String> props, String processName) {
+ props.put("userWorkflowName", ProcessHelper.getProcessWorkflowName(
+ processWorkflow.getName(), processName));
+ props.put("userWorkflowVersion", processWorkflow.getVersion());
+ props.put("userWorkflowEngine", processWorkflow.getEngine().value());
+ }
+
protected void createWorkflow(Cluster cluster, Process process, Workflow processWorkflow,
String wfName, Path parentWfPath) throws FalconException {
WORKFLOWAPP wfApp = getWorkflowTemplate(DEFAULT_WF_TEMPLATE);
@@ -615,6 +627,10 @@ public class OozieProcessMapper extends AbstractOozieEntityMapper<Process> {
private List<String> getPrepareDeleteOutputPathList(Process process) throws FalconException {
final List<String> deleteList = new ArrayList<String>();
+ if (process.getOutputs() == null) {
+ return deleteList;
+ }
+
for (Output output : process.getOutputs().getOutputs()) {
Feed feed = EntityUtil.getEntity(EntityType.FEED, output.getFeed());
@@ -630,6 +646,10 @@ public class OozieProcessMapper extends AbstractOozieEntityMapper<Process> {
private void addInputFeedsAsParams(List<String> paramList, Process process, Cluster cluster,
String engineType) throws FalconException {
+ if (process.getInputs() == null) {
+ return;
+ }
+
for (Input input : process.getInputs().getInputs()) {
Feed feed = EntityUtil.getEntity(EntityType.FEED, input.getFeed());
Storage storage = FeedHelper.createStorage(cluster, feed);
@@ -653,6 +673,10 @@ public class OozieProcessMapper extends AbstractOozieEntityMapper<Process> {
private void addOutputFeedsAsParams(List<String> paramList, Process process,
Cluster cluster) throws FalconException {
+ if (process.getOutputs() == null) {
+ return;
+ }
+
for (Output output : process.getOutputs().getOutputs()) {
Feed feed = EntityUtil.getEntity(EntityType.FEED, output.getFeed());
Storage storage = FeedHelper.createStorage(cluster, feed);
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/bb5e1449/process/src/test/java/org/apache/falcon/converter/OozieProcessMapperTest.java
----------------------------------------------------------------------
diff --git a/process/src/test/java/org/apache/falcon/converter/OozieProcessMapperTest.java b/process/src/test/java/org/apache/falcon/converter/OozieProcessMapperTest.java
index b4c059a..22bf9fe 100644
--- a/process/src/test/java/org/apache/falcon/converter/OozieProcessMapperTest.java
+++ b/process/src/test/java/org/apache/falcon/converter/OozieProcessMapperTest.java
@@ -38,6 +38,8 @@ import org.apache.falcon.entity.v0.process.Input;
import org.apache.falcon.entity.v0.process.Output;
import org.apache.falcon.entity.v0.process.Process;
import org.apache.falcon.entity.v0.process.Validity;
+import org.apache.falcon.entity.v0.process.Workflow;
+import org.apache.falcon.messaging.EntityInstanceMessage;
import org.apache.falcon.oozie.bundle.BUNDLEAPP;
import org.apache.falcon.oozie.coordinator.CONFIGURATION.Property;
import org.apache.falcon.oozie.coordinator.COORDINATORAPP;
@@ -63,7 +65,11 @@ import javax.xml.bind.Unmarshaller;
import javax.xml.transform.stream.StreamSource;
import javax.xml.validation.Schema;
import javax.xml.validation.SchemaFactory;
-import java.io.*;
+import java.io.BufferedReader;
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStreamReader;
import java.net.URL;
import java.util.Collections;
import java.util.HashMap;
@@ -332,12 +338,12 @@ public class OozieProcessMapperTest extends AbstractTestBase {
}
// verify the late data params
- Assert.assertEquals(props.get("falconInputFeeds"), process.getInputs().getInputs().get(0).getName());
+ Assert.assertEquals(props.get("falconInputFeeds"), process.getInputs().getInputs().get(0).getFeed());
Assert.assertEquals(props.get("falconInPaths"), "${coord:dataIn('input')}");
Assert.assertEquals(props.get("falconInputFeedStorageTypes"), Storage.TYPE.TABLE.name());
// verify the post processing params
- Assert.assertEquals(props.get("feedNames"), process.getOutputs().getOutputs().get(0).getName());
+ Assert.assertEquals(props.get("feedNames"), process.getOutputs().getOutputs().get(0).getFeed());
Assert.assertEquals(props.get("feedInstancePaths"), "${coord:dataOut('output')}");
}
@@ -374,6 +380,14 @@ public class OozieProcessMapperTest extends AbstractTestBase {
}
}
+ @Test
+ public void testProcessWorkflowMapper() throws Exception {
+ Process process = ConfigurationStore.get().get(EntityType.PROCESS, "clicksummary");
+ Workflow processWorkflow = process.getWorkflow();
+ Assert.assertEquals("test", processWorkflow.getName());
+ Assert.assertEquals("1.0.0", processWorkflow.getVersion());
+ }
+
@SuppressWarnings("unchecked")
private void assertLibExtensions(COORDINATORAPP coord) throws Exception {
String wfPath = coord.getAction().getWorkflow().getAppPath().replace("${nameNode}", "");
@@ -496,5 +510,48 @@ public class OozieProcessMapperTest extends AbstractTestBase {
ConfigurationStore.get().remove(EntityType.PROCESS, "table-process");
ConfigurationStore.get().remove(EntityType.FEED, "clicks-raw-table");
ConfigurationStore.get().remove(EntityType.FEED, "clicks-summary-table");
+ ConfigurationStore.get().remove(EntityType.PROCESS, "dumb-process");
+ }
+
+ @Test
+ public void testProcessWithNoInputsAndOutputs() throws Exception {
+ Cluster cluster = ConfigurationStore.get().get(EntityType.CLUSTER, "corp");
+ ClusterHelper.getInterface(cluster, Interfacetype.WRITE).setEndpoint(hdfsUrl);
+
+ URL resource = this.getClass().getResource("/config/process/dumb-process.xml");
+ Process processEntity = (Process) EntityType.PROCESS.getUnmarshaller().unmarshal(resource);
+ ConfigurationStore.get().publish(EntityType.PROCESS, processEntity);
+
+ OozieProcessMapper mapper = new OozieProcessMapper(processEntity);
+ Path bundlePath = new Path("/falcon/staging/workflows", processEntity.getName());
+ mapper.map(cluster, bundlePath);
+ assertTrue(fs.exists(bundlePath));
+
+ BUNDLEAPP bundle = getBundle(bundlePath);
+ assertEquals(EntityUtil.getWorkflowName(processEntity).toString(), bundle.getName());
+ assertEquals(1, bundle.getCoordinator().size());
+ assertEquals(EntityUtil.getWorkflowName(Tag.DEFAULT, processEntity).toString(),
+ bundle.getCoordinator().get(0).getName());
+ String coordPath = bundle.getCoordinator().get(0).getAppPath().replace("${nameNode}", "");
+
+ COORDINATORAPP coord = getCoordinator(new Path(coordPath));
+ HashMap<String, String> props = new HashMap<String, String>();
+ for (Property prop : coord.getAction().getWorkflow().getConfiguration().getProperty()) {
+ props.put(prop.getName(), prop.getValue());
+ }
+
+ String[] expected = {
+ EntityInstanceMessage.ARG.feedNames.getPropName(),
+ EntityInstanceMessage.ARG.feedInstancePaths.getPropName(),
+ "falconInputFeeds",
+ "falconInPaths",
+ "userWorkflowName",
+ "userWorkflowVersion",
+ "userWorkflowEngine",
+ };
+
+ for (String property : expected) {
+ Assert.assertTrue(props.containsKey(property), "expected property missing: " + property);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/bb5e1449/process/src/test/resources/config/process/dumb-process.xml
----------------------------------------------------------------------
diff --git a/process/src/test/resources/config/process/dumb-process.xml b/process/src/test/resources/config/process/dumb-process.xml
new file mode 100644
index 0000000..b71f089
--- /dev/null
+++ b/process/src/test/resources/config/process/dumb-process.xml
@@ -0,0 +1,40 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+ -->
+<process name="dumb-process" xmlns="uri:falcon:process:0.1">
+ <!-- where -->
+ <clusters>
+ <cluster name="corp">
+ <validity start="2011-11-02T00:00Z" end="2011-12-30T00:00Z"/>
+ </cluster>
+ </clusters>
+
+ <!-- when -->
+ <parallel>1</parallel>
+ <order>LIFO</order>
+ <frequency>hours(1)</frequency>
+ <timezone>UTC</timezone>
+
+ <!-- what = none -->
+
+ <!-- how -->
+ <workflow engine="pig" path="/apps/pig/id.pig"/>
+
+ <retry policy="periodic" delay="hours(10)" attempts="3"/>
+
+</process>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/bb5e1449/process/src/test/resources/config/process/process-0.1.xml
----------------------------------------------------------------------
diff --git a/process/src/test/resources/config/process/process-0.1.xml b/process/src/test/resources/config/process/process-0.1.xml
index 975d1a4..6148441 100644
--- a/process/src/test/resources/config/process/process-0.1.xml
+++ b/process/src/test/resources/config/process/process-0.1.xml
@@ -34,7 +34,7 @@
<property name="mapred.job.priority" value="LOW"/>
</properties>
- <workflow engine="oozie" path="/user/guest/workflow"/>
+ <workflow name="test" version="1.0.0" engine="oozie" path="/user/guest/workflow"/>
<retry policy="periodic" delay="hours(10)" attempts="3"/>
[5/6] git commit: FALCON-288 Persist lineage information into a
persistent store. Contributed by Venkatesh Seetharam
Posted by ve...@apache.org.
FALCON-288 Persist lineage information into a persistent store. Contributed by Venkatesh Seetharam
Project: http://git-wip-us.apache.org/repos/asf/incubator-falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-falcon/commit/ecb919c6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/ecb919c6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/ecb919c6
Branch: refs/heads/master
Commit: ecb919c603ae726fa1dc6e94f1867ba699505a1b
Parents: 59c8d50
Author: Venkatesh Seetharam <ve...@hortonworks.com>
Authored: Wed Mar 5 16:13:02 2014 -0800
Committer: Venkatesh Seetharam <ve...@hortonworks.com>
Committed: Wed Mar 5 16:13:02 2014 -0800
----------------------------------------------------------------------
CHANGES.txt | 3 +
LICENSE.txt | 66 ++
NOTICE.txt | 14 +-
.../falcon/entity/ColoClusterRelation.java | 7 +-
.../falcon/entity/store/ConfigurationStore.java | 14 +-
.../apache/falcon/entity/v0/EntityGraph.java | 9 +-
.../org/apache/falcon/group/FeedGroupMap.java | 11 +-
.../EntityRelationshipGraphBuilder.java | 403 ++++++++++
.../org/apache/falcon/metadata/GraphUtils.java | 83 +++
.../InstanceRelationshipGraphBuilder.java | 215 ++++++
.../falcon/metadata/MetadataMappingService.java | 247 +++++++
.../metadata/RelationshipGraphBuilder.java | 237 ++++++
.../service/ConfigurationChangeListener.java | 29 +-
common/src/main/resources/startup.properties | 17 +-
.../entity/store/ConfigurationStoreTest.java | 7 +-
.../metadata/MetadataMappingServiceTest.java | 731 +++++++++++++++++++
.../service/SharedLibraryHostingService.java | 20 +-
pom.xml | 42 ++
.../falcon/service/FalconTopicSubscriber.java | 8 +
src/bin/falcon-config.sh | 2 +-
src/conf/startup.properties | 20 +-
webapp/pom.xml | 5 +
.../org/apache/falcon/cli/FalconCLISmokeIT.java | 31 +-
.../resource/EntityManagerJerseySmokeIT.java | 2 +
.../org/apache/falcon/resource/TestContext.java | 5 +
25 files changed, 2183 insertions(+), 45 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index e93d7b4..931a868 100755
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -5,6 +5,9 @@ Trunk (Unreleased)
INCOMPATIBLE CHANGES
FALCON-11 Add support for security in Falcon (Venkatesh Seetharam)
+ FALCON-288 Persist lineage information into a persistent store
+ (Venkatesh Seetharam)
+
NEW FEATURES
FALCON-281 Design Action Interface. (Srikanth Sundarrajan)
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/LICENSE.txt
----------------------------------------------------------------------
diff --git a/LICENSE.txt b/LICENSE.txt
index 2dc0cea..d2fe249 100644
--- a/LICENSE.txt
+++ b/LICENSE.txt
@@ -1174,3 +1174,69 @@ expressed or implied warranty.
================================
+
+For the TinkerPop blueprint libraries:
+
+Copyright (c) 2009-Infinity, TinkerPop [http://tinkerpop.com]
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+ * Redistributions of source code must retain the above copyright
+ notice, this list of conditions and the following disclaimer.
+ * Redistributions in binary form must reproduce the above copyright
+ notice, this list of conditions and the following disclaimer in the
+ documentation and/or other materials provided with the distribution.
+ * Neither the name of the TinkerPop nor the
+ names of its contributors may be used to endorse or promote products
+ derived from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL TINKERPOP BE LIABLE FOR ANY
+DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+================================
+
+For the Oracle Berkeley DB Java Edition:
+
+Copyright (c) 2002, 2012 Oracle and/or its affiliates. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions
+are met:
+1. Redistributions of source code must retain the above copyright
+ notice, this list of conditions and the following disclaimer.
+2. Redistributions in binary form must reproduce the above copyright
+ notice, this list of conditions and the following disclaimer in the
+ documentation and/or other materials provided with the distribution.
+3. Redistributions in any form must be accompanied by information on
+ how to obtain complete source code for the DB software and any
+ accompanying software that uses the DB software. The source code
+ must either be included in the distribution or be available for no
+ more than the cost of distribution plus a nominal fee, and must be
+ freely redistributable under reasonable conditions. For an
+ executable file, complete source code means the source code for all
+ modules it contains. It does not include source code for modules or
+ files that typically accompany the major components of the operating
+ system on which the executable file runs.
+
+THIS SOFTWARE IS PROVIDED BY ORACLE ``AS IS'' AND ANY EXPRESS OR
+IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, OR
+NON-INFRINGEMENT, ARE DISCLAIMED. IN NO EVENT SHALL ORACLE BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
+OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN
+IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+================================
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/NOTICE.txt
----------------------------------------------------------------------
diff --git a/NOTICE.txt b/NOTICE.txt
index 20e5932..953c8cc 100644
--- a/NOTICE.txt
+++ b/NOTICE.txt
@@ -61,10 +61,22 @@ Copyright (c) 2012 Twitter, Inc.
This product includes/uses Glyphicons (http://glyphicons.com/),
Copyright (c) 2010 - 2012 Jan KovarÃk
-This product includes/uses concurrent libraris from DOUG LEA
+This product includes/uses concurrent libraries from DOUG LEA
(http://gee.cs.oswego.edu/dl/classes/EDU/oswego/cs/dl/util/sun-u.c.license.pdf)
This product includes/uses packages cern.colt* , cern.jet*, cern.clhep
Copyright (c) 1999 CERN - European Organization for Nuclear Research.
(http://acs.lbl.gov/software/colt/license.html)
+This product includes/uses TinkerPop blueprints API
+Copyright (c) 2009-Infinity, TinkerPop [http://tinkerpop.com]
+(https://github.com/tinkerpop/blueprints/blob/master/LICENSE.txt)
+
+This product includes/uses Thinkaurelius Titan
+Open source with the liberal Apache 2 license
+(https://github.com/thinkaurelius/titan/blob/master/LICENSE.txt)
+
+This product includes/uses Oracle Berkeley DB Java Edition
+Open Source License for Oracle Berkeley DB Java Edition
+Copyright (c) 2002, 2012 Oracle and/or its affiliates.
+(http://www.oracle.com/technetwork/database/database-technologies/berkeleydb/downloads/jeoslicense-086837.html)
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/common/src/main/java/org/apache/falcon/entity/ColoClusterRelation.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/entity/ColoClusterRelation.java b/common/src/main/java/org/apache/falcon/entity/ColoClusterRelation.java
index e42efd7..e4ca91b 100644
--- a/common/src/main/java/org/apache/falcon/entity/ColoClusterRelation.java
+++ b/common/src/main/java/org/apache/falcon/entity/ColoClusterRelation.java
@@ -51,7 +51,7 @@ public final class ColoClusterRelation implements ConfigurationChangeListener {
}
@Override
- public void onAdd(Entity entity, boolean ignoreFailure) {
+ public void onAdd(Entity entity) {
if (entity.getEntityType() != EntityType.CLUSTER) {
return;
}
@@ -81,4 +81,9 @@ public final class ColoClusterRelation implements ConfigurationChangeListener {
}
throw new FalconException("change shouldn't be supported on cluster!");
}
+
+ @Override
+ public void onReload(Entity entity) throws FalconException {
+ onAdd(entity);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/common/src/main/java/org/apache/falcon/entity/store/ConfigurationStore.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/entity/store/ConfigurationStore.java b/common/src/main/java/org/apache/falcon/entity/store/ConfigurationStore.java
index 156fafe..c2f1d44 100644
--- a/common/src/main/java/org/apache/falcon/entity/store/ConfigurationStore.java
+++ b/common/src/main/java/org/apache/falcon/entity/store/ConfigurationStore.java
@@ -136,7 +136,7 @@ public final class ConfigurationStore implements FalconService {
String entityName = URLDecoder.decode(encodedEntityName, UTF_8);
Entity entity = restore(type, entityName);
entityMap.put(entityName, entity);
- onAdd(entity, true);
+ onReload(entity);
}
}
}
@@ -162,7 +162,7 @@ public final class ConfigurationStore implements FalconService {
try {
if (get(type, entity.getName()) == null) {
persist(type, entity);
- onAdd(entity, false);
+ onAdd(entity);
dictionary.get(type).put(entity.getName(), entity);
} else {
throw new EntityAlreadyExistsException(
@@ -200,9 +200,9 @@ public final class ConfigurationStore implements FalconService {
}
}
- private void onAdd(Entity entity, boolean ignoreFailure) throws FalconException {
+ private void onAdd(Entity entity) throws FalconException {
for (ConfigurationChangeListener listener : listeners) {
- listener.onAdd(entity, ignoreFailure);
+ listener.onAdd(entity);
}
}
@@ -212,6 +212,12 @@ public final class ConfigurationStore implements FalconService {
}
}
+ private void onReload(Entity entity) throws FalconException {
+ for (ConfigurationChangeListener listener : listeners) {
+ listener.onReload(entity);
+ }
+ }
+
public synchronized void initiateUpdate(Entity entity) throws FalconException {
if (get(entity.getEntityType(), entity.getName()) == null || updatesInProgress.get() != null) {
throw new FalconException(
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/common/src/main/java/org/apache/falcon/entity/v0/EntityGraph.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/entity/v0/EntityGraph.java b/common/src/main/java/org/apache/falcon/entity/v0/EntityGraph.java
index fff124b..75cab5d 100644
--- a/common/src/main/java/org/apache/falcon/entity/v0/EntityGraph.java
+++ b/common/src/main/java/org/apache/falcon/entity/v0/EntityGraph.java
@@ -69,7 +69,7 @@ public final class EntityGraph implements ConfigurationChangeListener {
}
@Override
- public void onAdd(Entity entity, boolean ignoreFailure) throws FalconException {
+ public void onAdd(Entity entity) throws FalconException {
Map<Node, Set<Node>> nodeEdges = null;
switch (entity.getEntityType()) {
case PROCESS:
@@ -124,7 +124,12 @@ public final class EntityGraph implements ConfigurationChangeListener {
@Override
public void onChange(Entity oldEntity, Entity newEntity) throws FalconException {
onRemove(oldEntity);
- onAdd(newEntity, false);
+ onAdd(newEntity);
+ }
+
+ @Override
+ public void onReload(Entity entity) throws FalconException {
+ onAdd(entity);
}
private Map<Node, Set<Node>> getEdgesFor(Process process) {
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/common/src/main/java/org/apache/falcon/group/FeedGroupMap.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/group/FeedGroupMap.java b/common/src/main/java/org/apache/falcon/group/FeedGroupMap.java
index 7fbb61a..e2dcfce 100644
--- a/common/src/main/java/org/apache/falcon/group/FeedGroupMap.java
+++ b/common/src/main/java/org/apache/falcon/group/FeedGroupMap.java
@@ -54,7 +54,7 @@ public final class FeedGroupMap implements ConfigurationChangeListener {
}
@Override
- public void onAdd(Entity entity, boolean ignoreFailure) throws FalconException {
+ public void onAdd(Entity entity) throws FalconException {
if (entity.getEntityType().equals(EntityType.FEED)) {
Feed feed = (Feed) entity;
@@ -64,7 +64,6 @@ public final class FeedGroupMap implements ConfigurationChangeListener {
Set<FeedGroup> groupSet = getGroups(feed);
addGroups(feed.getName(), groupSet);
}
-
}
@Override
@@ -83,7 +82,6 @@ public final class FeedGroupMap implements ConfigurationChangeListener {
}
}
-
}
@Override
@@ -91,7 +89,12 @@ public final class FeedGroupMap implements ConfigurationChangeListener {
throws FalconException {
onRemove(oldEntity);
- onAdd(newEntity, false);
+ onAdd(newEntity);
+ }
+
+ @Override
+ public void onReload(Entity entity) throws FalconException {
+ onAdd(entity);
}
private void addGroups(String feed, Set<FeedGroup> groups) {
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/common/src/main/java/org/apache/falcon/metadata/EntityRelationshipGraphBuilder.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/EntityRelationshipGraphBuilder.java b/common/src/main/java/org/apache/falcon/metadata/EntityRelationshipGraphBuilder.java
new file mode 100644
index 0000000..6182dbe
--- /dev/null
+++ b/common/src/main/java/org/apache/falcon/metadata/EntityRelationshipGraphBuilder.java
@@ -0,0 +1,403 @@
+/**
+ * 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.falcon.metadata;
+
+import com.tinkerpop.blueprints.KeyIndexableGraph;
+import com.tinkerpop.blueprints.Vertex;
+import org.apache.falcon.entity.ProcessHelper;
+import org.apache.falcon.entity.v0.cluster.Cluster;
+import org.apache.falcon.entity.v0.feed.Feed;
+import org.apache.falcon.entity.v0.process.Input;
+import org.apache.falcon.entity.v0.process.Inputs;
+import org.apache.falcon.entity.v0.process.Output;
+import org.apache.falcon.entity.v0.process.Outputs;
+import org.apache.falcon.entity.v0.process.Process;
+import org.apache.falcon.entity.v0.process.Workflow;
+import org.apache.log4j.Logger;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Entity Metadata relationship mapping helper.
+ */
+public class EntityRelationshipGraphBuilder extends RelationshipGraphBuilder {
+
+ private static final Logger LOG = Logger.getLogger(EntityRelationshipGraphBuilder.class);
+
+ // entity vertex types
+ public static final String CLUSTER_ENTITY_TYPE = "cluster-entity";
+ public static final String FEED_ENTITY_TYPE = "feed-entity";
+ public static final String PROCESS_ENTITY_TYPE = "process-entity";
+
+
+ public EntityRelationshipGraphBuilder(KeyIndexableGraph graph, boolean preserveHistory) {
+ super(graph, preserveHistory);
+ }
+
+ public void addClusterEntity(Cluster clusterEntity) {
+ LOG.info("Adding cluster entity: " + clusterEntity.getName());
+ Vertex clusterVertex = addVertex(clusterEntity.getName(), CLUSTER_ENTITY_TYPE);
+
+ // addUserRelation(clusterVertex, "created-by"); // audit info
+ addColoRelation(clusterEntity.getColo(), clusterVertex);
+ addDataClassification(clusterEntity.getTags(), clusterVertex);
+ }
+
+ public void addFeedEntity(Feed feed) {
+ LOG.info("Adding feed entity: " + feed.getName());
+ Vertex feedVertex = addVertex(feed.getName(), FEED_ENTITY_TYPE);
+
+ addUserRelation(feedVertex);
+ addDataClassification(feed.getTags(), feedVertex);
+ addGroups(feed.getGroups(), feedVertex);
+
+ for (org.apache.falcon.entity.v0.feed.Cluster feedCluster : feed.getClusters().getClusters()) {
+ addRelationToCluster(feedVertex, feedCluster.getName(), FEED_CLUSTER_EDGE_LABEL);
+ }
+ }
+
+ public void updateFeedEntity(Feed oldFeed, Feed newFeed) {
+ LOG.info("Updating feed entity: " + newFeed.getName());
+ Vertex feedEntityVertex = findVertex(oldFeed.getName(), FEED_ENTITY_TYPE);
+ if (feedEntityVertex == null) {
+ throw new IllegalStateException(oldFeed.getName() + " entity vertex must exist.");
+ }
+
+ updateDataClassification(oldFeed.getTags(), newFeed.getTags(), feedEntityVertex);
+ updateGroups(oldFeed.getGroups(), newFeed.getGroups(), feedEntityVertex);
+ updateFeedClusters(oldFeed.getClusters().getClusters(),
+ newFeed.getClusters().getClusters(), feedEntityVertex);
+ }
+
+ public void addProcessEntity(Process process) {
+ String processName = process.getName();
+ LOG.info("Adding process entity: " + processName);
+ Vertex processVertex = addVertex(processName, PROCESS_ENTITY_TYPE);
+ addWorkflowProperties(process.getWorkflow(), processVertex, processName);
+
+ addUserRelation(processVertex);
+ addDataClassification(process.getTags(), processVertex);
+
+ for (org.apache.falcon.entity.v0.process.Cluster cluster : process.getClusters().getClusters()) {
+ addRelationToCluster(processVertex, cluster.getName(), PROCESS_CLUSTER_EDGE_LABEL);
+ }
+
+ addInputFeeds(process.getInputs(), processVertex);
+ addOutputFeeds(process.getOutputs(), processVertex);
+ // addWorkflow(process.getWorkflow(), processVertex, processName);
+ }
+
+ public void updateProcessEntity(Process oldProcess, Process newProcess) {
+ LOG.info("Updating process entity: " + newProcess.getName());
+ Vertex processEntityVertex = findVertex(oldProcess.getName(), PROCESS_ENTITY_TYPE);
+ if (processEntityVertex == null) {
+ throw new IllegalStateException(oldProcess.getName() + " entity vertex must exist.");
+ }
+
+ updateWorkflowProperties(oldProcess.getWorkflow(), newProcess.getWorkflow(),
+ processEntityVertex, newProcess.getName());
+ updateDataClassification(oldProcess.getTags(), newProcess.getTags(), processEntityVertex);
+ updateProcessClusters(oldProcess.getClusters().getClusters(),
+ newProcess.getClusters().getClusters(), processEntityVertex);
+ updateProcessInputs(oldProcess.getInputs(), newProcess.getInputs(), processEntityVertex);
+ updateProcessOutputs(oldProcess.getOutputs(), newProcess.getOutputs(), processEntityVertex);
+ }
+
+ public void addColoRelation(String colo, Vertex fromVertex) {
+ Vertex coloVertex = addVertex(colo, COLO_TYPE);
+ addEdge(fromVertex, coloVertex, CLUSTER_COLO_LABEL);
+ }
+
+ public void addRelationToCluster(Vertex fromVertex, String clusterName, String edgeLabel) {
+ Vertex clusterVertex = findVertex(clusterName, CLUSTER_ENTITY_TYPE);
+ if (clusterVertex == null) { // cluster must exist before adding other entities
+ throw new IllegalStateException("Cluster entity vertex must exist: " + clusterName);
+ }
+
+ addEdge(fromVertex, clusterVertex, edgeLabel);
+ }
+
+ public void addInputFeeds(Inputs inputs, Vertex processVertex) {
+ if (inputs == null) {
+ return;
+ }
+
+ for (Input input : inputs.getInputs()) {
+ addProcessFeedEdge(processVertex, input.getFeed(), FEED_PROCESS_EDGE_LABEL);
+ }
+ }
+
+ public void addOutputFeeds(Outputs outputs, Vertex processVertex) {
+ if (outputs == null) {
+ return;
+ }
+
+ for (Output output : outputs.getOutputs()) {
+ addProcessFeedEdge(processVertex, output.getFeed(), PROCESS_FEED_EDGE_LABEL);
+ }
+ }
+
+ public void addProcessFeedEdge(Vertex processVertex, String feedName, String edgeLabel) {
+ Vertex feedVertex = findVertex(feedName, FEED_ENTITY_TYPE);
+ if (feedVertex == null) {
+ throw new IllegalStateException("Feed entity vertex must exist: " + feedName);
+ }
+
+ addProcessFeedEdge(processVertex, feedVertex, edgeLabel);
+ }
+
+ public void addWorkflowProperties(Workflow workflow, Vertex processVertex, String processName) {
+ processVertex.setProperty(LineageArgs.USER_WORKFLOW_NAME.getOptionName(),
+ ProcessHelper.getProcessWorkflowName(workflow.getName(), processName));
+ processVertex.setProperty(VERSION_PROPERTY_KEY, workflow.getVersion());
+ processVertex.setProperty(LineageArgs.USER_WORKFLOW_ENGINE.getOptionName(), workflow.getEngine().value());
+ }
+
+ public void updateWorkflowProperties(Workflow oldWorkflow, Workflow newWorkflow,
+ Vertex processEntityVertex, String processName) {
+ if (areSame(oldWorkflow, newWorkflow)) {
+ return;
+ }
+
+ LOG.info("Updating workflow properties for: " + processEntityVertex);
+ addWorkflowProperties(newWorkflow, processEntityVertex, processName);
+ }
+
+ public void updateDataClassification(String oldClassification, String newClassification,
+ Vertex entityVertex) {
+ if (areSame(oldClassification, newClassification)) {
+ return;
+ }
+
+ removeDataClassification(oldClassification, entityVertex);
+ addDataClassification(newClassification, entityVertex);
+ }
+
+ private void removeDataClassification(String classification, Vertex entityVertex) {
+ if (classification == null || classification.length() == 0) {
+ return;
+ }
+
+ String[] oldTags = classification.split(",");
+ for (String oldTag : oldTags) {
+ int index = oldTag.indexOf("=");
+ String tagKey = oldTag.substring(0, index);
+ String tagValue = oldTag.substring(index + 1, oldTag.length());
+
+ removeEdge(entityVertex, tagValue, tagKey);
+ }
+ }
+
+ public void updateGroups(String oldGroups, String newGroups, Vertex entityVertex) {
+ if (areSame(oldGroups, newGroups)) {
+ return;
+ }
+
+ removeGroups(oldGroups, entityVertex);
+ addGroups(newGroups, entityVertex);
+ }
+
+ private void removeGroups(String groups, Vertex entityVertex) {
+ if (groups == null || groups.length() == 0) {
+ return;
+ }
+
+ String[] oldGroupTags = groups.split(",");
+ for (String groupTag : oldGroupTags) {
+ removeEdge(entityVertex, groupTag, RelationshipGraphBuilder.GROUPS_LABEL);
+ }
+ }
+
+ public static boolean areSame(String oldValue, String newValue) {
+ return oldValue == null && newValue == null
+ || oldValue != null && newValue != null && oldValue.equals(newValue);
+ }
+
+ public void updateFeedClusters(List<org.apache.falcon.entity.v0.feed.Cluster> oldClusters,
+ List<org.apache.falcon.entity.v0.feed.Cluster> newClusters,
+ Vertex feedEntityVertex) {
+ if (areFeedClustersSame(oldClusters, newClusters)) {
+ return;
+ }
+
+ // remove edges to old clusters
+ for (org.apache.falcon.entity.v0.feed.Cluster oldCuster : oldClusters) {
+ removeEdge(feedEntityVertex, oldCuster.getName(), FEED_CLUSTER_EDGE_LABEL);
+ }
+
+ // add edges to new clusters
+ for (org.apache.falcon.entity.v0.feed.Cluster newCluster : newClusters) {
+ addRelationToCluster(feedEntityVertex, newCluster.getName(), FEED_CLUSTER_EDGE_LABEL);
+ }
+ }
+
+ public boolean areFeedClustersSame(List<org.apache.falcon.entity.v0.feed.Cluster> oldClusters,
+ List<org.apache.falcon.entity.v0.feed.Cluster> newClusters) {
+ if (oldClusters.size() != newClusters.size()) {
+ return false;
+ }
+
+ List<String> oldClusterNames = getFeedClusterNames(oldClusters);
+ List<String> newClusterNames = getFeedClusterNames(newClusters);
+
+ return oldClusterNames.size() == newClusterNames.size()
+ && oldClusterNames.containsAll(newClusterNames)
+ && newClusterNames.containsAll(oldClusterNames);
+ }
+
+ public List<String> getFeedClusterNames(List<org.apache.falcon.entity.v0.feed.Cluster> clusters) {
+ List<String> clusterNames = new ArrayList<String>(clusters.size());
+ for (org.apache.falcon.entity.v0.feed.Cluster cluster : clusters) {
+ clusterNames.add(cluster.getName());
+ }
+
+ return clusterNames;
+ }
+
+ public void updateProcessClusters(List<org.apache.falcon.entity.v0.process.Cluster> oldClusters,
+ List<org.apache.falcon.entity.v0.process.Cluster> newClusters,
+ Vertex processEntityVertex) {
+ if (areProcessClustersSame(oldClusters, newClusters)) {
+ return;
+ }
+
+ // remove old clusters
+ for (org.apache.falcon.entity.v0.process.Cluster oldCuster : oldClusters) {
+ removeEdge(processEntityVertex, oldCuster.getName(), PROCESS_CLUSTER_EDGE_LABEL);
+ }
+
+ // add new clusters
+ for (org.apache.falcon.entity.v0.process.Cluster newCluster : newClusters) {
+ addRelationToCluster(processEntityVertex, newCluster.getName(), PROCESS_CLUSTER_EDGE_LABEL);
+ }
+ }
+
+ public boolean areProcessClustersSame(List<org.apache.falcon.entity.v0.process.Cluster> oldClusters,
+ List<org.apache.falcon.entity.v0.process.Cluster> newClusters) {
+ if (oldClusters.size() != newClusters.size()) {
+ return false;
+ }
+
+ List<String> oldClusterNames = getProcessClusterNames(oldClusters);
+ List<String> newClusterNames = getProcessClusterNames(newClusters);
+
+ return oldClusterNames.size() == newClusterNames.size()
+ && oldClusterNames.containsAll(newClusterNames)
+ && newClusterNames.containsAll(oldClusterNames);
+ }
+
+ public List<String> getProcessClusterNames(List<org.apache.falcon.entity.v0.process.Cluster> clusters) {
+ List<String> clusterNames = new ArrayList<String>(clusters.size());
+ for (org.apache.falcon.entity.v0.process.Cluster cluster : clusters) {
+ clusterNames.add(cluster.getName());
+ }
+
+ return clusterNames;
+ }
+
+ public static boolean areSame(Workflow oldWorkflow, Workflow newWorkflow) {
+ return areSame(oldWorkflow.getName(), newWorkflow.getName())
+ && areSame(oldWorkflow.getVersion(), newWorkflow.getVersion())
+ && areSame(oldWorkflow.getEngine().value(), newWorkflow.getEngine().value());
+ }
+
+ private void updateProcessInputs(Inputs oldProcessInputs, Inputs newProcessInputs,
+ Vertex processEntityVertex) {
+ if (areSame(oldProcessInputs, newProcessInputs)) {
+ return;
+ }
+
+ removeInputFeeds(oldProcessInputs, processEntityVertex);
+ addInputFeeds(newProcessInputs, processEntityVertex);
+ }
+
+ public static boolean areSame(Inputs oldProcessInputs, Inputs newProcessInputs) {
+ if (oldProcessInputs == null && newProcessInputs == null
+ || oldProcessInputs == null || newProcessInputs == null
+ || oldProcessInputs.getInputs().size() != newProcessInputs.getInputs().size()) {
+ return false;
+ }
+
+ List<Input> oldInputs = oldProcessInputs.getInputs();
+ List<Input> newInputs = newProcessInputs.getInputs();
+
+ return oldInputs.size() == newInputs.size()
+ && oldInputs.containsAll(newInputs)
+ && newInputs.containsAll(oldInputs);
+ }
+
+ public void removeInputFeeds(Inputs inputs, Vertex processVertex) {
+ if (inputs == null) {
+ return;
+ }
+
+ for (Input input : inputs.getInputs()) {
+ removeProcessFeedEdge(processVertex, input.getFeed(), FEED_PROCESS_EDGE_LABEL);
+ }
+ }
+
+ public void removeOutputFeeds(Outputs outputs, Vertex processVertex) {
+ if (outputs == null) {
+ return;
+ }
+
+ for (Output output : outputs.getOutputs()) {
+ removeProcessFeedEdge(processVertex, output.getFeed(), PROCESS_FEED_EDGE_LABEL);
+ }
+ }
+
+ public void removeProcessFeedEdge(Vertex processVertex, String feedName, String edgeLabel) {
+ Vertex feedVertex = findVertex(feedName, FEED_ENTITY_TYPE);
+ if (feedVertex == null) {
+ throw new IllegalStateException("Feed entity vertex must exist: " + feedName);
+ }
+
+ if (edgeLabel.equals(FEED_PROCESS_EDGE_LABEL)) {
+ removeEdge(feedVertex, processVertex, edgeLabel);
+ } else {
+ removeEdge(processVertex, feedVertex, edgeLabel);
+ }
+ }
+
+ private void updateProcessOutputs(Outputs oldProcessOutputs, Outputs newProcessOutputs,
+ Vertex processEntityVertex) {
+ if (areSame(oldProcessOutputs, newProcessOutputs)) {
+ return;
+ }
+
+ removeOutputFeeds(oldProcessOutputs, processEntityVertex);
+ addOutputFeeds(newProcessOutputs, processEntityVertex);
+ }
+
+ public static boolean areSame(Outputs oldProcessOutputs, Outputs newProcessOutputs) {
+ if (oldProcessOutputs == null && newProcessOutputs == null
+ || oldProcessOutputs == null || newProcessOutputs == null
+ || oldProcessOutputs.getOutputs().size() != newProcessOutputs.getOutputs().size()) {
+ return false;
+ }
+
+ List<Output> oldOutputs = oldProcessOutputs.getOutputs();
+ List<Output> newOutputs = newProcessOutputs.getOutputs();
+
+ return oldOutputs.size() == newOutputs.size()
+ && oldOutputs.containsAll(newOutputs)
+ && newOutputs.containsAll(oldOutputs);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/common/src/main/java/org/apache/falcon/metadata/GraphUtils.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/GraphUtils.java b/common/src/main/java/org/apache/falcon/metadata/GraphUtils.java
new file mode 100644
index 0000000..e98e44f
--- /dev/null
+++ b/common/src/main/java/org/apache/falcon/metadata/GraphUtils.java
@@ -0,0 +1,83 @@
+/**
+ * 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.falcon.metadata;
+
+import com.tinkerpop.blueprints.Direction;
+import com.tinkerpop.blueprints.Edge;
+import com.tinkerpop.blueprints.KeyIndexableGraph;
+import com.tinkerpop.blueprints.Vertex;
+import com.tinkerpop.blueprints.util.io.graphson.GraphSONWriter;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * Utility class for graph operations.
+ */
+public final class GraphUtils {
+
+ private static final Logger LOG = Logger.getLogger(GraphUtils.class);
+
+ private GraphUtils() {
+ }
+
+ public static void dumpToLog(final KeyIndexableGraph graph) {
+ LOG.debug("Vertices of " + graph);
+ for (Vertex vertex : graph.getVertices()) {
+ LOG.debug(vertexString(vertex));
+ }
+
+ LOG.debug("Edges of " + graph);
+ for (Edge edge : graph.getEdges()) {
+ LOG.debug(edgeString(edge));
+ }
+ }
+
+ public static void dump(final KeyIndexableGraph graph) throws IOException {
+ dump(graph, System.out);
+ }
+
+ public static void dump(final KeyIndexableGraph graph, OutputStream outputStream) throws IOException {
+ GraphSONWriter.outputGraph(graph, outputStream);
+ }
+
+ public static void dump(final KeyIndexableGraph graph, String fileName) throws IOException {
+ GraphSONWriter.outputGraph(graph, fileName);
+ }
+
+ public static String vertexString(final Vertex vertex) {
+ StringBuilder properties = new StringBuilder();
+ for (String propertyKey : vertex.getPropertyKeys()) {
+ properties.append(propertyKey)
+ .append("=").append(vertex.getProperty(propertyKey))
+ .append(", ");
+ }
+
+ return "v[" + vertex.getId() + "], Properties[" + properties + "]";
+ }
+
+ public static String edgeString(final Edge edge) {
+ return "e[" + edge.getLabel() + "], ["
+ + edge.getVertex(Direction.OUT).getProperty("name")
+ + " -> " + edge.getLabel() + " -> "
+ + edge.getVertex(Direction.IN).getProperty("name")
+ + "]";
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/common/src/main/java/org/apache/falcon/metadata/InstanceRelationshipGraphBuilder.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/InstanceRelationshipGraphBuilder.java b/common/src/main/java/org/apache/falcon/metadata/InstanceRelationshipGraphBuilder.java
new file mode 100644
index 0000000..11e8f86
--- /dev/null
+++ b/common/src/main/java/org/apache/falcon/metadata/InstanceRelationshipGraphBuilder.java
@@ -0,0 +1,215 @@
+/**
+ * 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.falcon.metadata;
+
+import com.tinkerpop.blueprints.KeyIndexableGraph;
+import com.tinkerpop.blueprints.Vertex;
+import org.apache.falcon.FalconException;
+import org.apache.falcon.entity.CatalogStorage;
+import org.apache.falcon.entity.FeedHelper;
+import org.apache.falcon.entity.Storage;
+import org.apache.falcon.entity.common.FeedDataPath;
+import org.apache.falcon.entity.store.ConfigurationStore;
+import org.apache.falcon.entity.v0.EntityType;
+import org.apache.falcon.entity.v0.feed.Feed;
+import org.apache.falcon.entity.v0.cluster.Cluster;
+import org.apache.falcon.entity.v0.process.Process;
+import org.apache.falcon.entity.v0.feed.LocationType;
+import org.apache.log4j.Logger;
+
+import java.net.URISyntaxException;
+import java.util.Map;
+
+/**
+ * Instance Metadata relationship mapping helper.
+ */
+public class InstanceRelationshipGraphBuilder extends RelationshipGraphBuilder {
+
+ private static final Logger LOG = Logger.getLogger(InstanceRelationshipGraphBuilder.class);
+
+ // instance vertex types
+ public static final String FEED_INSTANCE_TYPE = "feed-instance";
+ public static final String PROCESS_INSTANCE_TYPE = "process-instance";
+
+ // process workflow properties from message
+ private static final String[] INSTANCE_WORKFLOW_PROPERTIES = {
+ LineageArgs.USER_WORKFLOW_NAME.getOptionName(),
+ LineageArgs.USER_WORKFLOW_ENGINE.getOptionName(),
+ LineageArgs.WORKFLOW_ID.getOptionName(),
+ LineageArgs.RUN_ID.getOptionName(),
+ LineageArgs.STATUS.getOptionName(),
+ LineageArgs.WF_ENGINE_URL.getOptionName(),
+ LineageArgs.USER_SUBFLOW_ID.getOptionName(),
+ };
+
+ // instance edge labels
+ public static final String INSTANCE_ENTITY_EDGE_LABEL = "instance-of";
+
+ public InstanceRelationshipGraphBuilder(KeyIndexableGraph graph, boolean preserveHistory) {
+ super(graph, preserveHistory);
+ }
+
+ public Vertex addProcessInstance(Map<String, String> lineageMetadata) throws FalconException {
+ String entityName = lineageMetadata.get(LineageArgs.ENTITY_NAME.getOptionName());
+ String processInstanceName = getProcessInstanceName(entityName,
+ lineageMetadata.get(LineageArgs.NOMINAL_TIME.getOptionName()));
+ LOG.info("Adding process instance: " + processInstanceName);
+
+ String timestamp = lineageMetadata.get(LineageArgs.TIMESTAMP.getOptionName());
+ Vertex processInstance = addVertex(processInstanceName, PROCESS_INSTANCE_TYPE, timestamp);
+ addWorkflowInstanceProperties(processInstance, lineageMetadata);
+
+ addInstanceToEntity(processInstance, entityName,
+ EntityRelationshipGraphBuilder.PROCESS_ENTITY_TYPE, INSTANCE_ENTITY_EDGE_LABEL);
+ addInstanceToEntity(processInstance, lineageMetadata.get(LineageArgs.CLUSTER.getOptionName()),
+ EntityRelationshipGraphBuilder.CLUSTER_ENTITY_TYPE, PROCESS_CLUSTER_EDGE_LABEL);
+ addInstanceToEntity(processInstance,
+ lineageMetadata.get(LineageArgs.WORKFLOW_USER.getOptionName()), USER_TYPE, USER_LABEL);
+
+ if (isPreserveHistory()) {
+ Process process = ConfigurationStore.get().get(EntityType.PROCESS, entityName);
+ addDataClassification(process.getTags(), processInstance);
+ }
+
+ return processInstance;
+ }
+
+ public void addWorkflowInstanceProperties(Vertex processInstance,
+ Map<String, String> lineageMetadata) {
+ for (String instanceWorkflowProperty : INSTANCE_WORKFLOW_PROPERTIES) {
+ addProperty(processInstance, lineageMetadata, instanceWorkflowProperty);
+ }
+ processInstance.setProperty(VERSION_PROPERTY_KEY,
+ lineageMetadata.get(LineageArgs.USER_WORKFLOW_VERSION.getOptionName()));
+ }
+
+ public String getProcessInstanceName(String entityName, String nominalTime) {
+ return entityName + "/" + nominalTime;
+ }
+
+ public void addInstanceToEntity(Vertex instanceVertex, String entityName,
+ String entityType, String edgeLabel) {
+ Vertex entityVertex = findVertex(entityName, entityType);
+ LOG.info("Vertex exists? name=" + entityName + ", type=" + entityType + ", v=" + entityVertex);
+ if (entityVertex == null) {
+ throw new IllegalStateException(entityType + " entity vertex must exist " + entityName);
+ }
+
+ addEdge(instanceVertex, entityVertex, edgeLabel);
+ }
+
+ public void addOutputFeedInstances(Map<String, String> lineageMetadata,
+ Vertex processInstance) throws FalconException {
+ String outputFeedNamesArg = lineageMetadata.get(LineageArgs.FEED_NAMES.getOptionName());
+ if ("NONE".equals(outputFeedNamesArg)) {
+ return; // there are no output feeds for this process
+ }
+
+ String[] outputFeedNames = outputFeedNamesArg.split(",");
+ String[] outputFeedInstancePaths =
+ lineageMetadata.get(LineageArgs.FEED_INSTANCE_PATHS.getOptionName()).split(",");
+
+ addFeedInstances(outputFeedNames, outputFeedInstancePaths,
+ processInstance, PROCESS_FEED_EDGE_LABEL, lineageMetadata);
+ }
+
+ public void addInputFeedInstances(Map<String, String> lineageMetadata,
+ Vertex processInstance) throws FalconException {
+ String inputFeedNamesArg = lineageMetadata.get(LineageArgs.INPUT_FEED_NAMES.getOptionName());
+ if ("NONE".equals(inputFeedNamesArg)) {
+ return; // there are no input feeds for this process
+ }
+
+ String[] inputFeedNames =
+ lineageMetadata.get(LineageArgs.INPUT_FEED_NAMES.getOptionName()).split(",");
+ String[] inputFeedInstancePaths =
+ lineageMetadata.get(LineageArgs.INPUT_FEED_PATHS.getOptionName()).split(",");
+
+ addFeedInstances(inputFeedNames, inputFeedInstancePaths,
+ processInstance, FEED_PROCESS_EDGE_LABEL, lineageMetadata);
+ }
+
+ public void addFeedInstances(String[] feedNames, String[] feedInstancePaths,
+ Vertex processInstance, String edgeLabel,
+ Map<String, String> lineageMetadata) throws FalconException {
+ String clusterName = lineageMetadata.get(LineageArgs.CLUSTER.getOptionName());
+
+ for (int index = 0; index < feedNames.length; index++) {
+ String feedName = feedNames[index];
+ String feedInstancePath = feedInstancePaths[index];
+
+ String feedInstanceName = getFeedInstanceName(feedName, clusterName, feedInstancePath);
+ LOG.info("Adding feed instance: " + feedInstanceName);
+ Vertex feedInstance = addVertex(feedInstanceName, FEED_INSTANCE_TYPE,
+ lineageMetadata.get(LineageArgs.TIMESTAMP.getOptionName()));
+
+ addProcessFeedEdge(processInstance, feedInstance, edgeLabel);
+
+ addInstanceToEntity(feedInstance, feedName,
+ EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE, INSTANCE_ENTITY_EDGE_LABEL);
+ addInstanceToEntity(feedInstance, lineageMetadata.get(LineageArgs.CLUSTER.getOptionName()),
+ EntityRelationshipGraphBuilder.CLUSTER_ENTITY_TYPE, FEED_CLUSTER_EDGE_LABEL);
+ addInstanceToEntity(feedInstance,
+ lineageMetadata.get(LineageArgs.WORKFLOW_USER.getOptionName()), USER_TYPE, USER_LABEL);
+
+ if (isPreserveHistory()) {
+ Feed feed = ConfigurationStore.get().get(EntityType.FEED, feedName);
+ addDataClassification(feed.getTags(), feedInstance);
+ addGroups(feed.getGroups(), feedInstance);
+ }
+ }
+ }
+
+ public String getFeedInstanceName(String feedName, String clusterName,
+ String feedInstancePath) throws FalconException {
+ try {
+ Feed feed = ConfigurationStore.get().get(EntityType.FEED, feedName);
+ Cluster cluster = ConfigurationStore.get().get(EntityType.CLUSTER, clusterName);
+
+ Storage.TYPE storageType = FeedHelper.getStorageType(feed, cluster);
+ return storageType == Storage.TYPE.TABLE
+ ? getTableFeedInstanceName(feed, feedInstancePath, storageType)
+ : getFileSystemFeedInstanceName(feedInstancePath, feed, cluster);
+
+ } catch (URISyntaxException e) {
+ throw new FalconException(e);
+ }
+ }
+
+ private String getTableFeedInstanceName(Feed feed, String feedInstancePath,
+ Storage.TYPE storageType) throws URISyntaxException {
+ CatalogStorage instanceStorage = (CatalogStorage) FeedHelper.createStorage(
+ storageType.name(), feedInstancePath);
+ return feed.getName() + "/" + instanceStorage.toPartitionAsPath();
+ }
+
+ private String getFileSystemFeedInstanceName(String feedInstancePath, Feed feed,
+ Cluster cluster) throws FalconException {
+ Storage rawStorage = FeedHelper.createStorage(cluster, feed);
+ String feedPathTemplate = rawStorage.getUriTemplate(LocationType.DATA);
+ String instance = feedInstancePath;
+
+ String[] elements = FeedDataPath.PATTERN.split(feedPathTemplate);
+ for (String element : elements) {
+ instance = instance.replaceFirst(element, "");
+ }
+
+ return feed.getName() + "/" + instance;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java b/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java
new file mode 100644
index 0000000..75f7f88
--- /dev/null
+++ b/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java
@@ -0,0 +1,247 @@
+/**
+ * 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.falcon.metadata;
+
+import com.tinkerpop.blueprints.Edge;
+import com.tinkerpop.blueprints.GraphFactory;
+import com.tinkerpop.blueprints.KeyIndexableGraph;
+import com.tinkerpop.blueprints.Vertex;
+import org.apache.commons.configuration.BaseConfiguration;
+import org.apache.commons.configuration.Configuration;
+import org.apache.falcon.FalconException;
+import org.apache.falcon.entity.store.ConfigurationStore;
+import org.apache.falcon.entity.v0.Entity;
+import org.apache.falcon.entity.v0.EntityType;
+import org.apache.falcon.entity.v0.cluster.Cluster;
+import org.apache.falcon.entity.v0.feed.Feed;
+import org.apache.falcon.entity.v0.process.Process;
+import org.apache.falcon.service.ConfigurationChangeListener;
+import org.apache.falcon.service.FalconService;
+import org.apache.falcon.util.StartupProperties;
+import org.apache.log4j.Logger;
+
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+/**
+ * Metadata relationship mapping service. Maps relationships into a graph database.
+ */
+public class MetadataMappingService implements FalconService, ConfigurationChangeListener {
+
+ private static final Logger LOG = Logger.getLogger(MetadataMappingService.class);
+
+ /**
+ * Entity operations.
+ */
+ public enum EntityOperations {
+ GENERATE, REPLICATE, DELETE
+ }
+
+ /**
+ * Constance for the service name.
+ */
+ public static final String SERVICE_NAME = MetadataMappingService.class.getSimpleName();
+
+ /**
+ * Constant for the configuration property that indicates the prefix.
+ */
+ private static final String FALCON_PREFIX = "falcon.graph.";
+
+ private KeyIndexableGraph graph;
+ private Set<String> vertexIndexedKeys;
+ private Set<String> edgeIndexedKeys;
+ private EntityRelationshipGraphBuilder entityGraphBuilder;
+ private InstanceRelationshipGraphBuilder instanceGraphBuilder;
+
+ @Override
+ public String getName() {
+ return SERVICE_NAME;
+ }
+
+ @Override
+ public void init() throws FalconException {
+ graph = initializeGraphDB();
+ LOG.info("Initialized graph db: " + graph);
+
+ vertexIndexedKeys = graph.getIndexedKeys(Vertex.class);
+ LOG.info("Init vertex property keys: " + vertexIndexedKeys);
+
+ edgeIndexedKeys = graph.getIndexedKeys(Edge.class);
+ LOG.info("Init edge property keys: " + edgeIndexedKeys);
+
+ boolean preserveHistory = Boolean.valueOf(StartupProperties.get().getProperty(
+ "falcon.graph.preserve.history", "false"));
+ entityGraphBuilder = new EntityRelationshipGraphBuilder(graph, preserveHistory);
+ instanceGraphBuilder = new InstanceRelationshipGraphBuilder(graph, preserveHistory);
+
+ ConfigurationStore.get().registerListener(this);
+ }
+
+ protected KeyIndexableGraph initializeGraphDB() {
+ LOG.info("Initializing graph db");
+
+ Configuration graphConfig = getConfiguration();
+ KeyIndexableGraph graphDB = (KeyIndexableGraph) GraphFactory.open(graphConfig);
+ createIndicesForVertexKeys(graphDB);
+ return graphDB;
+ }
+
+ public Configuration getConfiguration() {
+ Configuration graphConfig = new BaseConfiguration();
+
+ Properties configProperties = StartupProperties.get();
+ for (Map.Entry entry : configProperties.entrySet()) {
+ String name = (String) entry.getKey();
+ if (name.startsWith(FALCON_PREFIX)) {
+ String value = (String) entry.getValue();
+ name = name.substring(FALCON_PREFIX.length());
+ graphConfig.setProperty(name, value);
+ }
+ }
+
+ return graphConfig;
+ }
+
+ protected void createIndicesForVertexKeys(KeyIndexableGraph graphDB) {
+ LOG.info("Creating indexes for graph");
+ // todo - externalize this
+ graphDB.createKeyIndex(RelationshipGraphBuilder.NAME_PROPERTY_KEY, Vertex.class);
+ graphDB.createKeyIndex(RelationshipGraphBuilder.TYPE_PROPERTY_KEY, Vertex.class);
+ graphDB.createKeyIndex(RelationshipGraphBuilder.VERSION_PROPERTY_KEY, Vertex.class);
+ graphDB.createKeyIndex(RelationshipGraphBuilder.TIMESTAMP_PROPERTY_KEY, Vertex.class);
+ }
+
+ public KeyIndexableGraph getGraph() {
+ return graph;
+ }
+
+ public Set<String> getVertexIndexedKeys() {
+ return vertexIndexedKeys;
+ }
+
+ public Set<String> getEdgeIndexedKeys() {
+ return edgeIndexedKeys;
+ }
+
+ @Override
+ public void destroy() throws FalconException {
+ LOG.info("Shutting down graph db");
+ graph.shutdown();
+ }
+
+ @Override
+ public void onAdd(Entity entity) throws FalconException {
+ EntityType entityType = entity.getEntityType();
+ LOG.info("Adding lineage for entity: " + entity.getName() + ", type: " + entityType);
+
+ switch (entityType) {
+ case CLUSTER:
+ entityGraphBuilder.addClusterEntity((Cluster) entity);
+ break;
+
+ case FEED:
+ entityGraphBuilder.addFeedEntity((Feed) entity);
+ break;
+
+ case PROCESS:
+ entityGraphBuilder.addProcessEntity((Process) entity);
+ break;
+
+ default:
+ }
+ }
+
+ @Override
+ public void onRemove(Entity entity) throws FalconException {
+ // do nothing, we'd leave the deleted entities as-is for historical purposes
+ // should we mark 'em as deleted?
+ }
+
+ @Override
+ public void onChange(Entity oldEntity, Entity newEntity) throws FalconException {
+ EntityType entityType = newEntity.getEntityType();
+ LOG.info("Updating lineage for entity: " + newEntity.getName() + ", type: " + entityType);
+
+ switch (entityType) {
+ case CLUSTER:
+ // a cluster cannot be updated
+ break;
+
+ case FEED:
+ entityGraphBuilder.updateFeedEntity((Feed) oldEntity, (Feed) newEntity);
+ break;
+
+ case PROCESS:
+ entityGraphBuilder.updateProcessEntity((Process) oldEntity, (Process) newEntity);
+ break;
+
+ default:
+ }
+ }
+
+ @Override
+ public void onReload(Entity entity) throws FalconException {
+ // do nothing since entities are being loaded from store into memory and
+ // are already added to the graph
+ }
+
+ public void mapLineage(String entityName, String operation,
+ String logDir) throws FalconException {
+ String lineageFile = LineageRecorder.getFilePath(logDir, entityName);
+
+ LOG.info("Parsing lineage metadata from: " + lineageFile);
+ Map<String, String> lineageMetadata = LineageRecorder.parseLineageMetadata(lineageFile);
+
+ EntityOperations entityOperation = EntityOperations.valueOf(operation);
+
+ LOG.info("Adding lineage for entity: " + entityName + ", operation: " + operation);
+ switch (entityOperation) {
+ case GENERATE:
+ onProcessInstanceAdded(lineageMetadata);
+ break;
+
+ case REPLICATE:
+ onFeedInstanceReplicated(lineageMetadata);
+ break;
+
+ case DELETE:
+ onFeedInstanceEvicted(lineageMetadata);
+ break;
+
+ default:
+ }
+ }
+
+ private void onProcessInstanceAdded(Map<String, String> lineageMetadata) throws FalconException {
+ Vertex processInstance = instanceGraphBuilder.addProcessInstance(lineageMetadata);
+ instanceGraphBuilder.addOutputFeedInstances(lineageMetadata, processInstance);
+ instanceGraphBuilder.addInputFeedInstances(lineageMetadata, processInstance);
+ }
+
+ private void onFeedInstanceReplicated(Map<String, String> lineageMetadata) {
+ LOG.info("Adding replicated feed instance: " + lineageMetadata.get(LineageArgs.NOMINAL_TIME.getOptionName()));
+ // todo - tbd
+ }
+
+ private void onFeedInstanceEvicted(Map<String, String> lineageMetadata) {
+ LOG.info("Adding evicted feed instance: " + lineageMetadata.get(LineageArgs.NOMINAL_TIME.getOptionName()));
+ // todo - tbd
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/common/src/main/java/org/apache/falcon/metadata/RelationshipGraphBuilder.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/RelationshipGraphBuilder.java b/common/src/main/java/org/apache/falcon/metadata/RelationshipGraphBuilder.java
new file mode 100644
index 0000000..1f607a6
--- /dev/null
+++ b/common/src/main/java/org/apache/falcon/metadata/RelationshipGraphBuilder.java
@@ -0,0 +1,237 @@
+/**
+ * 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.falcon.metadata;
+
+import com.tinkerpop.blueprints.Direction;
+import com.tinkerpop.blueprints.Edge;
+import com.tinkerpop.blueprints.GraphQuery;
+import com.tinkerpop.blueprints.KeyIndexableGraph;
+import com.tinkerpop.blueprints.Vertex;
+import org.apache.falcon.security.CurrentUser;
+import org.apache.log4j.Logger;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.TimeZone;
+
+/**
+ * Base class for Metadata relationship mapping helper.
+ */
+public abstract class RelationshipGraphBuilder {
+
+ private static final Logger LOG = Logger.getLogger(RelationshipGraphBuilder.class);
+
+ // vertex property keys
+ public static final String NAME_PROPERTY_KEY = "name";
+ public static final String TYPE_PROPERTY_KEY = "type";
+ public static final String TIMESTAMP_PROPERTY_KEY = "timestamp";
+ public static final String VERSION_PROPERTY_KEY = "version";
+
+ // vertex types
+ public static final String USER_TYPE = "user";
+ public static final String COLO_TYPE = "data-center";
+ public static final String TAGS_TYPE = "classification";
+ public static final String GROUPS_TYPE = "group";
+
+ // edge labels
+ public static final String USER_LABEL = "owned-by";
+ public static final String CLUSTER_COLO_LABEL = "collocated";
+ public static final String GROUPS_LABEL = "grouped-as";
+
+ // entity edge labels
+ public static final String FEED_CLUSTER_EDGE_LABEL = "stored-in";
+ public static final String PROCESS_CLUSTER_EDGE_LABEL = "runs-on";
+ public static final String FEED_PROCESS_EDGE_LABEL = "input";
+ public static final String PROCESS_FEED_EDGE_LABEL = "output";
+
+ /**
+ * A blueprints graph.
+ */
+ private final KeyIndexableGraph graph;
+
+ /**
+ * If enabled, preserves history of tags and groups for instances else will only
+ * be available for entities.
+ */
+ private final boolean preserveHistory;
+
+ protected RelationshipGraphBuilder(KeyIndexableGraph graph, boolean preserveHistory) {
+ this.graph = graph;
+ this.preserveHistory = preserveHistory;
+ }
+
+ protected KeyIndexableGraph getGraph() {
+ return graph;
+ }
+
+ protected boolean isPreserveHistory() {
+ return preserveHistory;
+ }
+
+ public Vertex addVertex(String name, String type) {
+ Vertex vertex = findVertex(name, type);
+ if (vertex != null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Found an existing vertex for: name=" + name + ", type=" + type);
+ }
+
+ return vertex;
+ }
+
+ return createVertex(name, type);
+ }
+
+ protected Vertex addVertex(String name, String type, String timestamp) {
+ Vertex vertex = findVertex(name, type);
+ if (vertex != null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Found an existing vertex for: name=" + name + ", type=" + type);
+ }
+
+ return vertex;
+ }
+
+ return createVertex(name, type, timestamp);
+ }
+
+ protected Vertex findVertex(String name, String type) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Finding vertex for: name=" + name + ", type=" + type);
+ }
+
+ GraphQuery query = graph.query()
+ .has(NAME_PROPERTY_KEY, name)
+ .has(TYPE_PROPERTY_KEY, type);
+ Iterator<Vertex> results = query.vertices().iterator();
+ return results.hasNext() ? results.next() : null;
+ }
+
+ protected Vertex createVertex(String name, String type) {
+ return createVertex(name, type, getCurrentTimeStamp());
+ }
+
+ protected Vertex createVertex(String name, String type, String timestamp) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Creating a new vertex for: name=" + name + ", type=" + type);
+ }
+
+ Vertex vertex = graph.addVertex(null);
+ vertex.setProperty(NAME_PROPERTY_KEY, name);
+ vertex.setProperty(TYPE_PROPERTY_KEY, type);
+ vertex.setProperty(TIMESTAMP_PROPERTY_KEY, timestamp);
+
+ return vertex;
+ }
+
+ protected Edge addEdge(Vertex fromVertex, Vertex toVertex, String edgeLabel) {
+ Edge edge = findEdge(fromVertex, edgeLabel);
+ return edgeExists(edge, toVertex) ? edge : fromVertex.addEdge(edgeLabel, toVertex);
+ }
+
+ protected void removeEdge(Vertex fromVertex, Vertex toVertex, String edgeLabel) {
+ Edge edge = findEdge(fromVertex, edgeLabel);
+ if (edgeExists(edge, toVertex)) {
+ getGraph().removeEdge(edge);
+ }
+ }
+
+ protected void removeEdge(Vertex fromVertex, Object toVertexName, String edgeLabel) {
+ Edge edge = findEdge(fromVertex, edgeLabel);
+ if (edgeExists(edge, toVertexName)) {
+ getGraph().removeEdge(edge);
+ }
+ }
+
+ protected boolean edgeExists(Edge edge, Object toVertexName) {
+ return edge != null && edge.getVertex(Direction.IN).getProperty(NAME_PROPERTY_KEY)
+ .equals(toVertexName);
+ }
+
+ protected boolean edgeExists(Edge edge, Vertex toVertex) {
+ return edge != null && edge.getVertex(Direction.IN).getProperty(NAME_PROPERTY_KEY)
+ .equals(toVertex.getProperty(NAME_PROPERTY_KEY));
+ }
+
+ protected Edge findEdge(Vertex fromVertex, String edgeLabel) {
+ Iterator<Edge> edges = fromVertex.getEdges(Direction.OUT, edgeLabel).iterator();
+ return edges.hasNext() ? edges.next() : null;
+ }
+
+ protected void addUserRelation(Vertex fromVertex) {
+ addUserRelation(fromVertex, USER_LABEL);
+ }
+
+ protected void addUserRelation(Vertex fromVertex, String edgeLabel) {
+ Vertex relationToUserVertex = addVertex(CurrentUser.getUser(), USER_TYPE);
+ addEdge(fromVertex, relationToUserVertex, edgeLabel);
+ }
+
+ protected void addDataClassification(String classification, Vertex entityVertex) {
+ if (classification == null || classification.length() == 0) {
+ return;
+ }
+
+ String[] tags = classification.split(",");
+ for (String tag : tags) {
+ int index = tag.indexOf("=");
+ String tagKey = tag.substring(0, index);
+ String tagValue = tag.substring(index + 1, tag.length());
+
+ Vertex tagValueVertex = addVertex(tagValue, TAGS_TYPE);
+ addEdge(entityVertex, tagValueVertex, tagKey);
+ }
+ }
+
+ protected void addGroups(String groups, Vertex fromVertex) {
+ if (groups == null || groups.length() == 0) {
+ return;
+ }
+
+ String[] groupTags = groups.split(",");
+ for (String groupTag : groupTags) {
+ Vertex groupVertex = addVertex(groupTag, GROUPS_TYPE);
+ addEdge(fromVertex, groupVertex, GROUPS_LABEL);
+ }
+ }
+
+ protected void addProcessFeedEdge(Vertex processVertex, Vertex feedVertex, String edgeLabel) {
+ if (edgeLabel.equals(FEED_PROCESS_EDGE_LABEL)) {
+ addEdge(feedVertex, processVertex, edgeLabel);
+ } else {
+ addEdge(processVertex, feedVertex, edgeLabel);
+ }
+ }
+
+ protected String getCurrentTimeStamp() {
+ SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd-HH-mm");
+ dateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));
+ return dateFormat.format(new Date());
+ }
+
+ protected void addProperty(Vertex vertex, Map<String, String> lineageMetadata, String optionName) {
+ String value = lineageMetadata.get(optionName);
+ if (value == null || value.length() == 0) {
+ return;
+ }
+
+ vertex.setProperty(optionName, value);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/common/src/main/java/org/apache/falcon/service/ConfigurationChangeListener.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/service/ConfigurationChangeListener.java b/common/src/main/java/org/apache/falcon/service/ConfigurationChangeListener.java
index da790fb..e20b0b5 100644
--- a/common/src/main/java/org/apache/falcon/service/ConfigurationChangeListener.java
+++ b/common/src/main/java/org/apache/falcon/service/ConfigurationChangeListener.java
@@ -26,9 +26,36 @@ import org.apache.falcon.entity.v0.Entity;
*/
public interface ConfigurationChangeListener {
- void onAdd(Entity entity, boolean ignoreFailure) throws FalconException;
+ /**
+ * This is upon adding a new entity to Store.
+ *
+ * @param entity entity object
+ * @throws FalconException
+ */
+ void onAdd(Entity entity) throws FalconException;
+ /**
+ * This is upon removing an existing entity from the Store.
+ *
+ * @param entity entity object
+ * @throws FalconException
+ */
void onRemove(Entity entity) throws FalconException;
+ /**
+ * This is upon updating an entity to the store.
+ *
+ * @param oldEntity old entity object
+ * @param newEntity updated entity object
+ * @throws FalconException
+ */
void onChange(Entity oldEntity, Entity newEntity) throws FalconException;
+
+ /**
+ * This is when existing entities are read from the store during startup.
+ *
+ * @param entity entity object
+ * @throws FalconException
+ */
+ void onReload(Entity entity) throws FalconException;
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/common/src/main/resources/startup.properties
----------------------------------------------------------------------
diff --git a/common/src/main/resources/startup.properties b/common/src/main/resources/startup.properties
index 51b3364..0b941eb 100644
--- a/common/src/main/resources/startup.properties
+++ b/common/src/main/resources/startup.properties
@@ -32,9 +32,9 @@
org.apache.falcon.service.ProcessSubscriberService,\
org.apache.falcon.entity.store.ConfigurationStore,\
org.apache.falcon.rerun.service.RetryService,\
- org.apache.falcon.rerun.service.LateRunService,\
- org.apache.falcon.service.SLAMonitoringService,\
- org.apache.falcon.service.LogCleanupService
+ org.apache.falcon.rerun.service.LateRunService,\
+ org.apache.falcon.service.SLAMonitoringService,\
+ org.apache.falcon.service.LogCleanupService
*.configstore.listeners=org.apache.falcon.entity.v0.EntityGraph,\
org.apache.falcon.entity.ColoClusterRelation,\
org.apache.falcon.group.FeedGroupMap,\
@@ -77,6 +77,17 @@ debug.libext.process.paths=${falcon.libext}
*.internal.queue.size=1000
+######### Graph Database Properties #########
+# Graph implementation
+*.falcon.graph.blueprints.graph=com.thinkaurelius.titan.core.TitanFactory
+
+# Graph Storage
+*.falcon.graph.storage.directory=${user.dir}/target/graphdb
+*.falcon.graph.storage.backend=berkeleyje
+*.falcon.graph.serialize.path=/${falcon.home}/data
+*.falcon.graph.preserve.history=false
+
+
######### Authentication Properties #########
# Authentication type must be specified: simple|kerberos
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/ecb919c6/common/src/test/java/org/apache/falcon/entity/store/ConfigurationStoreTest.java
----------------------------------------------------------------------
diff --git a/common/src/test/java/org/apache/falcon/entity/store/ConfigurationStoreTest.java b/common/src/test/java/org/apache/falcon/entity/store/ConfigurationStoreTest.java
index 89c031d..f466711 100644
--- a/common/src/test/java/org/apache/falcon/entity/store/ConfigurationStoreTest.java
+++ b/common/src/test/java/org/apache/falcon/entity/store/ConfigurationStoreTest.java
@@ -47,7 +47,7 @@ public class ConfigurationStoreTest {
private class TestListener implements ConfigurationChangeListener {
@Override
- public void onAdd(Entity entity, boolean ignoreFailure) throws FalconException {
+ public void onAdd(Entity entity) throws FalconException {
throw new FalconException("For test");
}
@@ -60,6 +60,11 @@ public class ConfigurationStoreTest {
public void onChange(Entity oldEntity, Entity newEntity) throws FalconException {
throw new FalconException("For test");
}
+
+ @Override
+ public void onReload(Entity entity) throws FalconException {
+ throw new FalconException("For test");
+ }
}
@Test