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/10/07 02:15:45 UTC
[3/5] git commit: FALCON-687 Add hooks for extensions in Audit.
Contributed by Venkatesh Seetharam
FALCON-687 Add hooks for extensions in Audit. 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/7ab8eb3f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/7ab8eb3f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/7ab8eb3f
Branch: refs/heads/master
Commit: 7ab8eb3f5d24ff7d7e8f891c98c665a41a5779f5
Parents: b9c7ffd
Author: Venkatesh Seetharam <ve...@apache.org>
Authored: Mon Oct 6 16:06:41 2014 -0700
Committer: Venkatesh Seetharam <ve...@apache.org>
Committed: Mon Oct 6 16:06:41 2014 -0700
----------------------------------------------------------------------
CHANGES.txt | 2 +
docs/src/site/twiki/FalconDocumentation.twiki | 77 +---------
docs/src/site/twiki/Operability.twiki | 102 +++++++++++++
.../falcon/aspect/AbstractFalconAspect.java | 27 ++++
.../org/apache/falcon/aspect/AuditMessage.java | 74 ++++++++++
.../org/apache/falcon/aspect/GenericAlert.java | 16 ++-
.../org/apache/falcon/monitors/Auditable.java | 37 +++++
.../apache/falcon/plugin/AuditingPlugin.java | 30 ++++
.../falcon/plugin/DefaultMonitoringPlugin.java | 50 +++++++
.../org/apache/falcon/plugin/LoggingPlugin.java | 42 ------
.../apache/falcon/aspect/AuditMessageTest.java | 57 ++++++++
prism/pom.xml | 1 -
.../plugin/ChainableMonitoringPlugin.java | 48 ++++++-
.../falcon/resource/AbstractEntityManager.java | 12 --
.../resource/AbstractInstanceManager.java | 6 -
.../AbstractSchedulableEntityManager.java | 4 -
.../falcon/security/FalconAuditFilter.java | 91 ++++++++++++
.../security/FalconAuthenticationFilter.java | 34 +----
.../security/FalconAuthorizationFilter.java | 81 ++++++++---
.../java/org/apache/falcon/util/Servlets.java | 88 ++++++++++++
prism/src/main/webapp/WEB-INF/web.xml | 10 ++
.../plugin/ChainableMonitoringPluginTest.java | 15 +-
.../falcon/security/FalconAuditFilterTest.java | 144 +++++++++++++++++++
.../src/main/webapp/WEB-INF/distributed/web.xml | 15 ++
webapp/src/main/webapp/WEB-INF/embedded/web.xml | 10 ++
webapp/src/main/webapp/WEB-INF/web.xml | 10 ++
26 files changed, 889 insertions(+), 194 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 3f77dfb..e22bc1e 100755
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -5,6 +5,8 @@ Trunk (Unreleased)
INCOMPATIBLE CHANGES
NEW FEATURES
+ FALCON-687 Add hooks for extensions in Audit (Venkatesh Seetharam)
+
FALCON-636 Add a sample recipe for disaster recovery of hdfs dirs/files
(Sowmya Ramesh via Venkatesh Seetharam)
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/docs/src/site/twiki/FalconDocumentation.twiki
----------------------------------------------------------------------
diff --git a/docs/src/site/twiki/FalconDocumentation.twiki b/docs/src/site/twiki/FalconDocumentation.twiki
index efa9efd..722a3bf 100644
--- a/docs/src/site/twiki/FalconDocumentation.twiki
+++ b/docs/src/site/twiki/FalconDocumentation.twiki
@@ -10,13 +10,14 @@
* <a href="#Updating_process_and_feed_definition">Updating process and feed definition</a>
* <a href="#Handling_late_input_data">Handling late input data</a>
* <a href="#Idempotency">Idempotency</a>
- * <a href="#Alerting_and_Monitoring">Alerting and Monitoring</a>
* <a href="#Falcon_EL_Expressions">Falcon EL Expressions</a>
* <a href="#Lineage">Lineage</a>
* <a href="#Security">Security</a>
* <a href="#Recipes">Recipes</a>
+ * <a href="#Monitoring">Monitoring</a>
---++ Architecture
+
---+++ Introduction
Falcon is a feed and process management platform over hadoop. Falcon essentially transforms user's feed
and process configurations into repeated actions through a standard workflow engine. Falcon by itself
@@ -557,76 +558,6 @@ All the operations in Falcon are Idempotent. That is if you make same request to
Idempotency also by takes care of the condition when request is sent through prism and fails on one or more servers. For example prism is configured to send request to 3 servers. First user sends a request to SUBMIT a process on all 3 of them, and receives a response SUCCESSFUL from all of them. Then due to some issue one of the servers goes down, and user send a request to schedule the submitted process. This time he will receive a response with PARTIAL status and a FAILURE message from the server that has gone down. If the users check he will find the process would have been started and running on the 2 SUCCESSFUL servers. Now the issue with server is figured out and it is brought up. Sending the SCHEDULE request again through prism will result in a SUCCESSFUL response from prism as well as other three servers, but this time PROCESS will be SCHEDULED only on the server which had failed earlier and other two will keep running as before.
----++ Alerting and Monitoring
----+++ Alerting
-Falcon provides monitoring of various events by capturing metrics of those events.
-The metric numbers can then be used to monitor performance and health of the Falcon system and the entire processing pipelines.
-
-Users can view the logs of these events in the metric.log file, by default this file is created under ${user.dir}/logs/ directory.
-Users may also extend the Falcon monitoring framework to send events to systems like Mondemand/lwes.
-
-The following events are captured by Falcon for logging the metrics:
- 1. New cluster definitions posted to Falcon (success & failures)
- 1. New feed definition posted to Falcon (success & failures)
- 1. New process definition posted to Falcon (success & failures)
- 1. Process update events (success & failures)
- 1. Feed update events (success & failures)
- 1. Cluster update events (success & failures)
- 1. Process suspend events (success & failures)
- 1. Feed suspend events (success & failures)
- 1. Process resume events (success & failures)
- 1. Feed resume events (success & failures)
- 1. Process remove events (success & failures)
- 1. Feed remove events (success & failures)
- 1. Cluster remove events (success & failures)
- 1. Process instance kill events (success & failures)
- 1. Process instance re-run events (success & failures)
- 1. Process instance generation events
- 1. Process instance failure events
- 1. Process instance auto-retry events
- 1. Process instance retry exhaust events
- 1. Feed instance deletion event
- 1. Feed instance deletion failure event (no retries)
- 1. Feed instance replication event
- 1. Feed instance replication failure event
- 1. Feed instance replication auto-retry event
- 1. Feed instance replication retry exhaust event
- 1. Feed instance late arrival event
- 1. Feed instance post cut-off arrival event
- 1. Process re-run due to late feed event
- 1. Transaction rollback failed event
-
-The metric logged for an event has the following properties:
- 1. Action - Name of the event.
- 2. Dimensions - A list of name/value pairs of various attributes for a given action.
- 3. Status- Status of an action FAILED/SUCCEEDED.
- 4. Time-taken - Time taken in nanoseconds for a given action.
-
-An example for an event logged for a submit of a new process definition:
-
- 2012-05-04 12:23:34,026 {Action:submit, Dimensions:{entityType=process}, Status: SUCCEEDED, Time-taken:97087000 ns}
-
-Users may parse the metric.log or capture these events from custom monitoring frameworks and can plot various graphs
-or send alerts according to their requirements.
-
----+++ Notifications
-Falcon creates a JMS topic for every process/feed that is scheduled in Falcon.
-The implementation class and the broker url of the JMS engine are read from the dependent cluster's definition.
-Users may register consumers on the required topic to check the availability or status of feed instances.
-
-For a given process that is scheduled, the name of the topic is same as the process name.
-Falcon sends a Map message for every feed produced by the instance of a process to the JMS topic.
-The JMS !MapMessage sent to a topic has the following properties:
-entityName, feedNames, feedInstancePath, workflowId, runId, nominalTime, timeStamp, brokerUrl, brokerImplClass, entityType, operation, logFile, topicName, status, brokerTTL;
-
-For a given feed that is scheduled, the name of the topic is same as the feed name.
-Falcon sends a map message for every feed instance that is deleted/archived/replicated depending upon the retention policy set in the feed definition.
-The JMS !MapMessage sent to a topic has the following properties:
-entityName, feedNames, feedInstancePath, workflowId, runId, nominalTime, timeStamp, brokerUrl, brokerImplClass, entityType, operation, logFile, topicName, status, brokerTTL;
-
-The JMS messages are automatically purged after a certain period (default 3 days) by the Falcon JMS house-keeping service.TTL (Time-to-live) for JMS message
-can be configured in the Falcon's startup.properties file.
-
---++ Falcon EL Expressions
@@ -753,3 +684,7 @@ Security is detailed in [[Security][Security]].
---++ Recipes
Recipes is detailed in [[Recipes][Recipes]].
+
+--++ Monitoring
+
+Monitoring and Operationalizing Falcon is detailed in [[Operability][Operability]].
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/docs/src/site/twiki/Operability.twiki
----------------------------------------------------------------------
diff --git a/docs/src/site/twiki/Operability.twiki b/docs/src/site/twiki/Operability.twiki
new file mode 100644
index 0000000..ae4b717
--- /dev/null
+++ b/docs/src/site/twiki/Operability.twiki
@@ -0,0 +1,102 @@
+---+ Operationalizing Falcon
+
+---++ Overview
+
+Apache Falcon provides various tools to operationalize Falcon consisting of Alerts for
+unrecoverable errors, Audits of user actions, Metrics, and Notifications. They are detailed below.
+
+
+---++ Monitoring
+
+Falcon provides monitoring of various events by capturing metrics of those events.
+The metric numbers can then be used to monitor performance and health of the Falcon system and
+the entire processing pipelines.
+
+Users can view the logs of these events in the metric.log file, by default this file is created
+under ${user.dir}/logs/ directory. Users may also extend the Falcon monitoring framework to send
+events to systems like Mondemand/lwes by implementingorg.apache.falcon.plugin.MonitoringPlugin
+interface.
+
+The following events are captured by Falcon for logging the metrics:
+ 1. New cluster definitions posted to Falcon (success & failures)
+ 1. New feed definition posted to Falcon (success & failures)
+ 1. New process definition posted to Falcon (success & failures)
+ 1. Process update events (success & failures)
+ 1. Feed update events (success & failures)
+ 1. Cluster update events (success & failures)
+ 1. Process suspend events (success & failures)
+ 1. Feed suspend events (success & failures)
+ 1. Process resume events (success & failures)
+ 1. Feed resume events (success & failures)
+ 1. Process remove events (success & failures)
+ 1. Feed remove events (success & failures)
+ 1. Cluster remove events (success & failures)
+ 1. Process instance kill events (success & failures)
+ 1. Process instance re-run events (success & failures)
+ 1. Process instance generation events
+ 1. Process instance failure events
+ 1. Process instance auto-retry events
+ 1. Process instance retry exhaust events
+ 1. Feed instance deletion event
+ 1. Feed instance deletion failure event (no retries)
+ 1. Feed instance replication event
+ 1. Feed instance replication failure event
+ 1. Feed instance replication auto-retry event
+ 1. Feed instance replication retry exhaust event
+ 1. Feed instance late arrival event
+ 1. Feed instance post cut-off arrival event
+ 1. Process re-run due to late feed event
+ 1. Transaction rollback failed event
+
+The metric logged for an event has the following properties:
+ 1. Action - Name of the event.
+ 2. Dimensions - A list of name/value pairs of various attributes for a given action.
+ 3. Status- Status of an action FAILED/SUCCEEDED.
+ 4. Time-taken - Time taken in nanoseconds for a given action.
+
+An example for an event logged for a submit of a new process definition:
+
+ 2012-05-04 12:23:34,026 {Action:submit, Dimensions:{entityType=process}, Status: SUCCEEDED, Time-taken:97087000 ns}
+
+Users may parse the metric.log or capture these events from custom monitoring frameworks and can plot various graphs
+or send alerts according to their requirements.
+
+
+---++ Notifications
+
+Falcon creates a JMS topic for every process/feed that is scheduled in Falcon.
+The implementation class and the broker url of the JMS engine are read from the dependent cluster's definition.
+Users may register consumers on the required topic to check the availability or status of feed instances.
+
+For a given process that is scheduled, the name of the topic is same as the process name.
+Falcon sends a Map message for every feed produced by the instance of a process to the JMS topic.
+The JMS !MapMessage sent to a topic has the following properties:
+entityName, feedNames, feedInstancePath, workflowId, runId, nominalTime, timeStamp, brokerUrl, brokerImplClass, entityType, operation, logFile, topicName, status, brokerTTL;
+
+For a given feed that is scheduled, the name of the topic is same as the feed name.
+Falcon sends a map message for every feed instance that is deleted/archived/replicated depending upon the retention policy set in the feed definition.
+The JMS !MapMessage sent to a topic has the following properties:
+entityName, feedNames, feedInstancePath, workflowId, runId, nominalTime, timeStamp, brokerUrl, brokerImplClass, entityType, operation, logFile, topicName, status, brokerTTL;
+
+The JMS messages are automatically purged after a certain period (default 3 days) by the Falcon JMS house-keeping service.TTL (Time-to-live) for JMS message
+can be configured in the Falcon's startup.properties file.
+
+
+---++ Alerts
+
+Falcon generates alerts for unrecoverable errors into a log file by default.
+Users can view these alerts in the alerts.log file, by default this file is created
+under ${user.dir}/logs/ directory.
+
+Users may also extend the Falcon Alerting plugin to send events to systems like Nagios, etc. by
+extending org.apache.falcon.plugin.AlertingPlugin interface.
+
+
+---++ Audits
+
+Falcon audits all user activity and captures them into a log file by default.
+Users can view these audits in the audit.log file, by default this file is created
+under ${user.dir}/logs/ directory.
+
+Users may also extend the Falcon Audit plugin to send audits to systems like Apache Argus, etc. by
+extending org.apache.falcon.plugin.AuditingPlugin interface.
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/metrics/src/main/java/org/apache/falcon/aspect/AbstractFalconAspect.java
----------------------------------------------------------------------
diff --git a/metrics/src/main/java/org/apache/falcon/aspect/AbstractFalconAspect.java b/metrics/src/main/java/org/apache/falcon/aspect/AbstractFalconAspect.java
index 06bda3f..b358011 100644
--- a/metrics/src/main/java/org/apache/falcon/aspect/AbstractFalconAspect.java
+++ b/metrics/src/main/java/org/apache/falcon/aspect/AbstractFalconAspect.java
@@ -110,4 +110,31 @@ public abstract class AbstractFalconAspect {
}
public abstract void publishAlert(AlertMessage alertMessage);
+
+ @Around("@annotation(org.apache.falcon.monitors.Auditable)")
+ public Object logAroundAudit(ProceedingJoinPoint joinPoint) throws Throwable {
+ Object[] args = joinPoint.getArgs();
+ Object result;
+
+ try {
+ result = joinPoint.proceed();
+ } finally {
+ AuditMessage auditMessage = new AuditMessage(
+ getStringValue(args[0], "Unknown-User"),
+ getStringValue(args[1], "Unknown-Address"),
+ getStringValue(args[2], "Unknown-Host"),
+ args[3].toString(),
+ args[4].toString(),
+ args[5].toString());
+ publishAudit(auditMessage);
+ }
+
+ return result;
+ }
+
+ private String getStringValue(Object value, String defaultValue) {
+ return value == null ? defaultValue : value.toString();
+ }
+
+ public abstract void publishAudit(AuditMessage auditMessage);
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/metrics/src/main/java/org/apache/falcon/aspect/AuditMessage.java
----------------------------------------------------------------------
diff --git a/metrics/src/main/java/org/apache/falcon/aspect/AuditMessage.java b/metrics/src/main/java/org/apache/falcon/aspect/AuditMessage.java
new file mode 100644
index 0000000..93c30a4
--- /dev/null
+++ b/metrics/src/main/java/org/apache/falcon/aspect/AuditMessage.java
@@ -0,0 +1,74 @@
+/**
+ * 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.aspect;
+
+/**
+ * Message to be sent to the auditing system.
+ */
+public class AuditMessage {
+
+ private final String user; // who
+ private final String remoteAddress; // who
+ private final String remoteHost; // who
+ private final String requestUrl; // what
+ private final String serverAddress; // what
+ private final String requestTimeISO9601; // when
+
+ public AuditMessage(String user, String remoteAddress, String remoteHost,
+ String requestUrl, String serverAddress, String requestTimeISO9601) {
+ this.user = user;
+ this.remoteAddress = remoteAddress;
+ this.remoteHost = remoteHost;
+ this.requestUrl = requestUrl;
+ this.serverAddress = serverAddress;
+ this.requestTimeISO9601 = requestTimeISO9601;
+ }
+
+ public String getUser() {
+ return user;
+ }
+
+ public String getRemoteAddress() {
+ return remoteAddress;
+ }
+
+ public String getRemoteHost() {
+ return remoteHost;
+ }
+
+ public String getRequestUrl() {
+ return requestUrl;
+ }
+
+ public String getServerAddress() {
+ return serverAddress;
+ }
+
+ public String getRequestTimeISO9601() {
+ return requestTimeISO9601;
+ }
+
+ @Override
+ public String toString() {
+ return "Audit: "
+ + user + "@" + remoteHost // who
+ + " performed " + requestUrl + " (" + serverAddress + ")" // what
+ + " at " + requestTimeISO9601; // when
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/metrics/src/main/java/org/apache/falcon/aspect/GenericAlert.java
----------------------------------------------------------------------
diff --git a/metrics/src/main/java/org/apache/falcon/aspect/GenericAlert.java b/metrics/src/main/java/org/apache/falcon/aspect/GenericAlert.java
index c7a86d9..2973347 100644
--- a/metrics/src/main/java/org/apache/falcon/aspect/GenericAlert.java
+++ b/metrics/src/main/java/org/apache/falcon/aspect/GenericAlert.java
@@ -18,14 +18,15 @@
package org.apache.falcon.aspect;
import org.apache.falcon.monitors.Alert;
+import org.apache.falcon.monitors.Auditable;
import org.apache.falcon.monitors.Dimension;
import org.apache.falcon.monitors.Monitored;
import org.apache.falcon.monitors.TimeTaken;
import org.aspectj.lang.annotation.Aspect;
/**
- * Create a method with params you want to monitor via Aspect and log in metric
- * and iMon, invoke this method from code.
+ * Create a method with params you want to monitor/alert/audit via Aspect
+ * and log in metric, invoke this method from code.
*/
@SuppressWarnings("UnusedParameters")
@Aspect
@@ -111,4 +112,15 @@ public final class GenericAlert {
@Dimension(value = "exception") Throwable throwable) {
return "IGNORE";
}
+
+ @Auditable(operation = "record-audit")
+ public static String audit(
+ @Dimension(value = "request-user") String user, // who
+ @Dimension(value = "remote-address") String remoteAddress, // who
+ @Dimension(value = "remote-host") String remoteHost, // who
+ @Dimension(value = "request-url") String requestUrl, // what
+ @Dimension(value = "server-address") String serverAddress, // what server
+ @Dimension(value = "request-time") String time) { // when
+ return "IGNORE";
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/metrics/src/main/java/org/apache/falcon/monitors/Auditable.java
----------------------------------------------------------------------
diff --git a/metrics/src/main/java/org/apache/falcon/monitors/Auditable.java b/metrics/src/main/java/org/apache/falcon/monitors/Auditable.java
new file mode 100644
index 0000000..72cd8d8
--- /dev/null
+++ b/metrics/src/main/java/org/apache/falcon/monitors/Auditable.java
@@ -0,0 +1,37 @@
+/**
+ * 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.monitors;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Auditable annotation for audit collection.
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.METHOD)
+public @interface Auditable {
+
+ /**
+ * @return Event name associated with this monitoring
+ */
+ String operation();
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/metrics/src/main/java/org/apache/falcon/plugin/AuditingPlugin.java
----------------------------------------------------------------------
diff --git a/metrics/src/main/java/org/apache/falcon/plugin/AuditingPlugin.java b/metrics/src/main/java/org/apache/falcon/plugin/AuditingPlugin.java
new file mode 100644
index 0000000..e72e9bc
--- /dev/null
+++ b/metrics/src/main/java/org/apache/falcon/plugin/AuditingPlugin.java
@@ -0,0 +1,30 @@
+/**
+ * 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.plugin;
+
+import org.apache.falcon.aspect.AuditMessage;
+
+/**
+ * Generic interface to receiving audits.
+ */
+public interface AuditingPlugin {
+
+ void audit(AuditMessage auditMessage);
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/metrics/src/main/java/org/apache/falcon/plugin/DefaultMonitoringPlugin.java
----------------------------------------------------------------------
diff --git a/metrics/src/main/java/org/apache/falcon/plugin/DefaultMonitoringPlugin.java b/metrics/src/main/java/org/apache/falcon/plugin/DefaultMonitoringPlugin.java
new file mode 100644
index 0000000..44b1384
--- /dev/null
+++ b/metrics/src/main/java/org/apache/falcon/plugin/DefaultMonitoringPlugin.java
@@ -0,0 +1,50 @@
+/**
+ * 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.plugin;
+
+import org.apache.falcon.aspect.AlertMessage;
+import org.apache.falcon.aspect.AuditMessage;
+import org.apache.falcon.aspect.ResourceMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Plugin implementation for logging metrics/alerts/audits that logs to a file.
+ */
+public class DefaultMonitoringPlugin implements MonitoringPlugin, AlertingPlugin, AuditingPlugin {
+
+ private static final Logger METRIC = LoggerFactory.getLogger("METRIC");
+ private static final Logger ALERT = LoggerFactory.getLogger("ALERT");
+ private static final Logger AUDIT = LoggerFactory.getLogger("AUDIT");
+
+ @Override
+ public void monitor(ResourceMessage message) {
+ METRIC.info("{}", message);
+ }
+
+ @Override
+ public void alert(AlertMessage message) {
+ ALERT.info("{}", message);
+ }
+
+ @Override
+ public void audit(AuditMessage auditMessage) {
+ AUDIT.info("{}", auditMessage);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/metrics/src/main/java/org/apache/falcon/plugin/LoggingPlugin.java
----------------------------------------------------------------------
diff --git a/metrics/src/main/java/org/apache/falcon/plugin/LoggingPlugin.java b/metrics/src/main/java/org/apache/falcon/plugin/LoggingPlugin.java
deleted file mode 100644
index c959fc2..0000000
--- a/metrics/src/main/java/org/apache/falcon/plugin/LoggingPlugin.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/**
- * 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.plugin;
-
-import org.apache.falcon.aspect.AlertMessage;
-import org.apache.falcon.aspect.ResourceMessage;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Plugin for logging metrics using log4j.
- */
-public class LoggingPlugin implements MonitoringPlugin, AlertingPlugin {
- private static final Logger METRIC = LoggerFactory.getLogger("METRIC");
- private static final Logger ALERT = LoggerFactory.getLogger("ALERT");
-
- @Override
- public void monitor(ResourceMessage message) {
- METRIC.info("{}", message);
- }
-
- @Override
- public void alert(AlertMessage message) {
- ALERT.info("{}", message);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/metrics/src/test/java/org/apache/falcon/aspect/AuditMessageTest.java
----------------------------------------------------------------------
diff --git a/metrics/src/test/java/org/apache/falcon/aspect/AuditMessageTest.java b/metrics/src/test/java/org/apache/falcon/aspect/AuditMessageTest.java
new file mode 100644
index 0000000..3da3797
--- /dev/null
+++ b/metrics/src/test/java/org/apache/falcon/aspect/AuditMessageTest.java
@@ -0,0 +1,57 @@
+/**
+ * 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.aspect;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+/**
+ * Test Message to be sent to the auditing system.
+ */
+public class AuditMessageTest {
+
+ private final AuditMessage auditMessage = new AuditMessage(
+ "falcon", "127.0.0.1", "LOCALHOST", "action", "127.0.0.1", "2014-09-15T20:56Z");
+
+ @Test
+ public void testGetRequestUrl() throws Exception {
+ Assert.assertEquals(auditMessage.getRequestUrl(), "action");
+ }
+
+ @Test
+ public void testGetUser() throws Exception {
+ Assert.assertEquals(auditMessage.getUser(), "falcon");
+ }
+
+ @Test
+ public void testGetRemoteHost() throws Exception {
+ Assert.assertEquals(auditMessage.getRemoteHost(), "LOCALHOST");
+ }
+
+ @Test
+ public void testGetCurrentTimeMillis() throws Exception {
+ Assert.assertEquals(auditMessage.getRequestTimeISO9601(), "2014-09-15T20:56Z");
+ }
+
+ @Test
+ public void testToString() throws Exception {
+ Assert.assertEquals(auditMessage.toString(),
+ "Audit: falcon@LOCALHOST performed action (127.0.0.1) at 2014-09-15T20:56Z");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/prism/pom.xml
----------------------------------------------------------------------
diff --git a/prism/pom.xml b/prism/pom.xml
index 9072d1b..b02ef51 100644
--- a/prism/pom.xml
+++ b/prism/pom.xml
@@ -198,7 +198,6 @@
<include>org/apache/falcon/resource/proxy/InstanceManagerProxy.java</include>
<include>org/apache/falcon/resource/AbstractInstanceManager.java</include>
<include>org/apache/falcon/plugin/ChainableMonitoringPlugin.java</include>
- <include>org/apache/falcon/service/FalconTopicSubscriber.java</include>
<include>org/apache/falcon/aspect/GenericAlert.java</include>
<include>org/apache/falcon/aspect/GenericAlertTest.java</include>
</includes>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/prism/src/main/java/org/apache/falcon/plugin/ChainableMonitoringPlugin.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/plugin/ChainableMonitoringPlugin.java b/prism/src/main/java/org/apache/falcon/plugin/ChainableMonitoringPlugin.java
index 1a5a331..1c2c625 100644
--- a/prism/src/main/java/org/apache/falcon/plugin/ChainableMonitoringPlugin.java
+++ b/prism/src/main/java/org/apache/falcon/plugin/ChainableMonitoringPlugin.java
@@ -21,6 +21,7 @@ package org.apache.falcon.plugin;
import org.apache.falcon.FalconException;
import org.apache.falcon.aspect.AbstractFalconAspect;
import org.apache.falcon.aspect.AlertMessage;
+import org.apache.falcon.aspect.AuditMessage;
import org.apache.falcon.aspect.ResourceMessage;
import org.apache.falcon.util.ReflectionUtils;
import org.apache.falcon.util.StartupProperties;
@@ -34,24 +35,26 @@ import java.util.List;
/**
* This class implements the chain of responsibility for configured implementations
- * of {@link MonitoringPlugin}. {@link LoggingPlugin} is the default.
+ * of {@link MonitoringPlugin}. {@link DefaultMonitoringPlugin} is the default.
*/
@Aspect
public class ChainableMonitoringPlugin extends AbstractFalconAspect
- implements MonitoringPlugin, AlertingPlugin {
+ implements MonitoringPlugin, AlertingPlugin, AuditingPlugin {
private static final Logger LOG = LoggerFactory.getLogger(ChainableMonitoringPlugin.class);
private List<MonitoringPlugin> monitoringPlugins = new ArrayList<MonitoringPlugin>();
private List<AlertingPlugin> alertingPlugins = new ArrayList<AlertingPlugin>();
+ private List<AuditingPlugin> auditingPlugins = new ArrayList<AuditingPlugin>();
public ChainableMonitoringPlugin() {
initializeMonitoringPlugins();
initializeAlertingPlugins();
+ initializeAuditingPlugins();
}
private void initializeMonitoringPlugins() {
String pluginClasses = StartupProperties.get().
- getProperty("monitoring.plugins", LoggingPlugin.class.getName());
+ getProperty("monitoring.plugins", DefaultMonitoringPlugin.class.getName());
try {
for (String pluginClass : pluginClasses.split(",")) {
MonitoringPlugin plugin = ReflectionUtils.getInstanceByClassName(pluginClass.trim());
@@ -59,14 +62,14 @@ public class ChainableMonitoringPlugin extends AbstractFalconAspect
LOG.info("Registered Monitoring Plugin {}", pluginClass);
}
} catch (FalconException e) {
- monitoringPlugins = Arrays.asList((MonitoringPlugin) new LoggingPlugin());
+ monitoringPlugins = Arrays.asList((MonitoringPlugin) new DefaultMonitoringPlugin());
LOG.error("Unable to initialize monitoring.plugins: {}", pluginClasses, e);
}
}
private void initializeAlertingPlugins() {
String pluginClasses = StartupProperties.get().
- getProperty("alerting.plugins", LoggingPlugin.class.getName());
+ getProperty("alerting.plugins", DefaultMonitoringPlugin.class.getName());
try {
for (String pluginClass : pluginClasses.split(",")) {
AlertingPlugin plugin = ReflectionUtils.getInstanceByClassName(pluginClass.trim());
@@ -74,11 +77,26 @@ public class ChainableMonitoringPlugin extends AbstractFalconAspect
LOG.info("Registered Alerting Plugin {}", pluginClass);
}
} catch (FalconException e) {
- alertingPlugins = Arrays.asList((AlertingPlugin) new LoggingPlugin());
+ alertingPlugins = Arrays.asList((AlertingPlugin) new DefaultMonitoringPlugin());
LOG.error("Unable to initialize alerting.plugins: {}", pluginClasses, e);
}
}
+ private void initializeAuditingPlugins() {
+ String pluginClasses = StartupProperties.get().
+ getProperty("auditing.plugins", DefaultMonitoringPlugin.class.getName());
+ try {
+ for (String pluginClass : pluginClasses.split(",")) {
+ AuditingPlugin plugin = ReflectionUtils.getInstanceByClassName(pluginClass.trim());
+ auditingPlugins.add(plugin);
+ LOG.info("Registered Auditing Plugin {}", pluginClass);
+ }
+ } catch (FalconException e) {
+ alertingPlugins = Arrays.asList((AlertingPlugin) new DefaultMonitoringPlugin());
+ LOG.error("Unable to initialize auditing.plugins: {}", pluginClasses, e);
+ }
+ }
+
@Override
public void monitor(ResourceMessage message) {
for (MonitoringPlugin plugin : monitoringPlugins) {
@@ -106,7 +124,23 @@ public class ChainableMonitoringPlugin extends AbstractFalconAspect
try {
plugin.alert(alertMessage);
} catch (Exception e) {
- LOG.debug("Unable to publish message to {}", plugin.getClass(), e);
+ LOG.debug("Unable to publish alert to {}", plugin.getClass(), e);
+ }
+ }
+ }
+
+ @Override
+ public void publishAudit(AuditMessage auditMessage) {
+ audit(auditMessage);
+ }
+
+ @Override
+ public void audit(AuditMessage auditMessage) {
+ for (AuditingPlugin plugin : auditingPlugins) {
+ try {
+ plugin.audit(auditMessage);
+ } catch (Exception e) {
+ LOG.debug("Unable to publish auditMessage to {}", plugin.getClass(), e);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/prism/src/main/java/org/apache/falcon/resource/AbstractEntityManager.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/resource/AbstractEntityManager.java b/prism/src/main/java/org/apache/falcon/resource/AbstractEntityManager.java
index fcd7b29..41cd601 100644
--- a/prism/src/main/java/org/apache/falcon/resource/AbstractEntityManager.java
+++ b/prism/src/main/java/org/apache/falcon/resource/AbstractEntityManager.java
@@ -57,7 +57,6 @@ import java.util.*;
*/
public abstract class AbstractEntityManager {
private static final Logger LOG = LoggerFactory.getLogger(AbstractEntityManager.class);
- private static final Logger AUDIT = LoggerFactory.getLogger("AUDIT");
protected static final int XML_DEBUG_LEN = 10 * 1024;
protected static final String DEFAULT_NUM_RESULTS = "10";
@@ -154,7 +153,6 @@ public abstract class AbstractEntityManager {
checkColo(colo);
try {
- audit(request, "STREAMED_DATA", type, "SUBMIT");
Entity entity = submitInternal(request, type);
return new APIResult(APIResult.Status.SUCCEEDED, "Submit successful (" + type + ") " + entity.getName());
} catch (Throwable e) {
@@ -207,7 +205,6 @@ public abstract class AbstractEntityManager {
checkColo(colo);
try {
EntityType entityType = EntityType.valueOf(type.toUpperCase());
- audit(request, entity, type, "DELETE");
String removedFromEngine = "";
try {
Entity entityObj = EntityUtil.getEntity(type, entity);
@@ -239,7 +236,6 @@ public abstract class AbstractEntityManager {
checkColo(colo);
try {
EntityType entityType = EntityType.valueOf(type.toUpperCase());
- audit(request, entityName, type, "UPDATE");
Entity oldEntity = EntityUtil.getEntity(type, entityName);
Entity newEntity = deserializeEntity(request, entityType);
validate(newEntity);
@@ -383,14 +379,6 @@ public abstract class AbstractEntityManager {
return new String(data);
}
- protected void audit(HttpServletRequest request, String entity, String type, String action) {
- if (request == null) {
- return; // this must be internal call from Falcon
- }
- AUDIT.info("Performed {} on {} ({}) :: {}/{}",
- action, entity, type, request.getRemoteHost(), CurrentUser.getUser());
- }
-
private enum EntityStatus {
SUBMITTED, SUSPENDED, RUNNING
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/prism/src/main/java/org/apache/falcon/resource/AbstractInstanceManager.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/resource/AbstractInstanceManager.java b/prism/src/main/java/org/apache/falcon/resource/AbstractInstanceManager.java
index 5e351f6..2070713 100644
--- a/prism/src/main/java/org/apache/falcon/resource/AbstractInstanceManager.java
+++ b/prism/src/main/java/org/apache/falcon/resource/AbstractInstanceManager.java
@@ -341,7 +341,6 @@ public abstract class AbstractInstanceManager extends AbstractEntityManager {
checkType(type);
try {
lifeCycles = checkAndUpdateLifeCycle(lifeCycles, type);
- audit(request, entity, type, "INSTANCE_KILL");
validateParams(type, entity);
Entity entityObject = EntityUtil.getEntity(type, entity);
Pair<Date, Date> startAndEndDate = getStartAndEndDate(entityObject, startStr, endStr);
@@ -364,7 +363,6 @@ public abstract class AbstractInstanceManager extends AbstractEntityManager {
checkType(type);
try {
lifeCycles = checkAndUpdateLifeCycle(lifeCycles, type);
- audit(request, entity, type, "INSTANCE_SUSPEND");
validateParams(type, entity);
Entity entityObject = EntityUtil.getEntity(type, entity);
Pair<Date, Date> startAndEndDate = getStartAndEndDate(entityObject, startStr, endStr);
@@ -383,12 +381,10 @@ public abstract class AbstractInstanceManager extends AbstractEntityManager {
String type, String entity, String startStr,
String endStr, String colo,
List<LifeCycle> lifeCycles) {
-
checkColo(colo);
checkType(type);
try {
lifeCycles = checkAndUpdateLifeCycle(lifeCycles, type);
- audit(request, entity, type, "INSTANCE_RESUME");
validateParams(type, entity);
Entity entityObject = EntityUtil.getEntity(type, entity);
Pair<Date, Date> startAndEndDate = getStartAndEndDate(entityObject, startStr, endStr);
@@ -406,12 +402,10 @@ public abstract class AbstractInstanceManager extends AbstractEntityManager {
public InstancesResult reRunInstance(String type, String entity, String startStr,
String endStr, HttpServletRequest request,
String colo, List<LifeCycle> lifeCycles) {
-
checkColo(colo);
checkType(type);
try {
lifeCycles = checkAndUpdateLifeCycle(lifeCycles, type);
- audit(request, entity, type, "INSTANCE_RERUN");
validateParams(type, entity);
Entity entityObject = EntityUtil.getEntity(type, entity);
Pair<Date, Date> startAndEndDate = getStartAndEndDate(entityObject, startStr, endStr);
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/prism/src/main/java/org/apache/falcon/resource/AbstractSchedulableEntityManager.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/resource/AbstractSchedulableEntityManager.java b/prism/src/main/java/org/apache/falcon/resource/AbstractSchedulableEntityManager.java
index b3151b9..eb89f5f 100644
--- a/prism/src/main/java/org/apache/falcon/resource/AbstractSchedulableEntityManager.java
+++ b/prism/src/main/java/org/apache/falcon/resource/AbstractSchedulableEntityManager.java
@@ -60,7 +60,6 @@ public abstract class AbstractSchedulableEntityManager extends AbstractInstanceM
@Dimension("colo") @PathParam("colo") String colo) {
checkColo(colo);
try {
- audit(request, entity, type, "SCHEDULED");
scheduleInternal(type, entity);
return new APIResult(APIResult.Status.SUCCEEDED, entity + "(" + type + ") scheduled successfully");
} catch (Throwable e) {
@@ -89,7 +88,6 @@ public abstract class AbstractSchedulableEntityManager extends AbstractInstanceM
checkColo(colo);
try {
checkSchedulableEntity(type);
- audit(request, "STREAMED_DATA", type, "SUBMIT_AND_SCHEDULE");
Entity entity = submitInternal(request, type);
scheduleInternal(type, entity.getName());
return new APIResult(APIResult.Status.SUCCEEDED,
@@ -114,7 +112,6 @@ public abstract class AbstractSchedulableEntityManager extends AbstractInstanceM
checkColo(colo);
try {
checkSchedulableEntity(type);
- audit(request, entity, type, "SUSPEND");
Entity entityObj = EntityUtil.getEntity(type, entity);
if (getWorkflowEngine().isActive(entityObj)) {
getWorkflowEngine().suspend(entityObj);
@@ -143,7 +140,6 @@ public abstract class AbstractSchedulableEntityManager extends AbstractInstanceM
checkColo(colo);
try {
checkSchedulableEntity(type);
- audit(request, entity, type, "RESUME");
Entity entityObj = EntityUtil.getEntity(type, entity);
if (getWorkflowEngine().isActive(entityObj)) {
getWorkflowEngine().resume(entityObj);
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/prism/src/main/java/org/apache/falcon/security/FalconAuditFilter.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/security/FalconAuditFilter.java b/prism/src/main/java/org/apache/falcon/security/FalconAuditFilter.java
new file mode 100644
index 0000000..d8b73b2
--- /dev/null
+++ b/prism/src/main/java/org/apache/falcon/security/FalconAuditFilter.java
@@ -0,0 +1,91 @@
+/**
+ * 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.security;
+
+import org.apache.falcon.aspect.GenericAlert;
+import org.apache.falcon.entity.v0.SchemaHelper;
+import org.apache.falcon.util.Servlets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.servlet.Filter;
+import javax.servlet.FilterChain;
+import javax.servlet.FilterConfig;
+import javax.servlet.ServletException;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import javax.servlet.http.HttpServletRequest;
+import java.io.IOException;
+import java.util.Date;
+
+/**
+ * This records audit information as part of the filter after processing the request.
+ */
+public class FalconAuditFilter implements Filter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FalconAuditFilter.class);
+
+ @Override
+ public void init(FilterConfig filterConfig) throws ServletException {
+ LOG.info("FalconAuditFilter initialization started");
+ }
+
+ @Override
+ public void doFilter(ServletRequest request,
+ ServletResponse response,
+ FilterChain filterChain) throws IOException, ServletException {
+ final String requestTimeISO9601 = SchemaHelper.formatDateUTC(new Date());
+
+ try {
+ filterChain.doFilter(request, response);
+ } finally {
+ recordAudit((HttpServletRequest) request, requestTimeISO9601);
+ }
+ }
+
+ private void recordAudit(HttpServletRequest httpRequest, String whenISO9601) {
+ final String who = getUserFromRequest(httpRequest);
+ final String fromHost = httpRequest.getRemoteHost();
+ final String fromAddress = httpRequest.getRemoteAddr();
+ final String whatURL = Servlets.getRequestURL(httpRequest);
+ final String whatAddrs = httpRequest.getLocalAddr();
+
+ LOG.debug("Audit: {}/{} performed request {} ({}) at time {}",
+ who, fromAddress, whatURL, whatAddrs, whenISO9601);
+ GenericAlert.audit(who, fromHost, fromAddress, whatURL, whatAddrs, whenISO9601);
+ }
+
+ private String getUserFromRequest(HttpServletRequest httpRequest) {
+ try {
+ // get the authenticated user
+ return CurrentUser.getUser();
+ } catch (IllegalStateException ignore) {
+ // ignore since the user authentication might have failed
+ }
+
+ // look for the user in the request
+ final String userFromRequest = Servlets.getUserFromRequest(httpRequest);
+ return userFromRequest == null ? "UNKNOWN" : userFromRequest;
+ }
+
+ @Override
+ public void destroy() {
+ // do nothing
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/prism/src/main/java/org/apache/falcon/security/FalconAuthenticationFilter.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/security/FalconAuthenticationFilter.java b/prism/src/main/java/org/apache/falcon/security/FalconAuthenticationFilter.java
index 54023d1..ae9e874 100644
--- a/prism/src/main/java/org/apache/falcon/security/FalconAuthenticationFilter.java
+++ b/prism/src/main/java/org/apache/falcon/security/FalconAuthenticationFilter.java
@@ -19,6 +19,7 @@
package org.apache.falcon.security;
import org.apache.commons.lang.StringUtils;
+import org.apache.falcon.util.Servlets;
import org.apache.falcon.util.StartupProperties;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.authentication.server.KerberosAuthenticationHandler;
@@ -167,7 +168,7 @@ public class FalconAuthenticationFilter
if (httpRequest.getMethod().equals("OPTIONS")) { // option request meant only for authentication
optionsServlet.service(request, response);
} else {
- final String user = getUserFromRequest(httpRequest);
+ final String user = Servlets.getUserFromRequest(httpRequest);
if (StringUtils.isEmpty(user)) {
((HttpServletResponse) response).sendError(Response.Status.BAD_REQUEST.getStatusCode(),
"User can't be empty");
@@ -180,7 +181,8 @@ public class FalconAuthenticationFilter
NDC.push(user + ":" + httpRequest.getMethod() + "/" + httpRequest.getPathInfo());
NDC.push(requestId);
CurrentUser.authenticate(user);
- LOG.info("Request from user: {}, URL={}", user, getRequestUrl(httpRequest));
+ LOG.info("Request from user: {}, URL={}", user,
+ Servlets.getRequestURI(httpRequest));
filterChain.doFilter(servletRequest, servletResponse);
} finally {
@@ -190,34 +192,6 @@ public class FalconAuthenticationFilter
}
}
}
-
- private String getUserFromRequest(HttpServletRequest httpRequest) {
- String user = httpRequest.getRemoteUser(); // this is available from wrapper in super class
- if (!StringUtils.isEmpty(user)) {
- return user;
- }
-
- user = httpRequest.getParameter("user.name"); // available in query-param
- if (!StringUtils.isEmpty(user)) {
- return user;
- }
-
- user = httpRequest.getHeader("Remote-User"); // backwards-compatibility
- if (!StringUtils.isEmpty(user)) {
- return user;
- }
-
- return null;
- }
-
- private String getRequestUrl(HttpServletRequest request) {
- StringBuffer url = request.getRequestURL();
- if (request.getQueryString() != null) {
- url.append("?").append(request.getQueryString());
- }
-
- return url.toString();
- }
};
super.doFilter(request, response, filterChainWrapper);
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/prism/src/main/java/org/apache/falcon/security/FalconAuthorizationFilter.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/security/FalconAuthorizationFilter.java b/prism/src/main/java/org/apache/falcon/security/FalconAuthorizationFilter.java
index ceb13d0..b61360b 100644
--- a/prism/src/main/java/org/apache/falcon/security/FalconAuthorizationFilter.java
+++ b/prism/src/main/java/org/apache/falcon/security/FalconAuthorizationFilter.java
@@ -31,7 +31,6 @@ import javax.servlet.ServletResponse;
import javax.servlet.http.HttpServletRequest;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.List;
/**
* This enforces authorization as part of the filter before processing the request.
@@ -62,28 +61,35 @@ public class FalconAuthorizationFilter implements Filter {
ServletResponse response,
FilterChain filterChain) throws IOException, ServletException {
HttpServletRequest httpRequest = (HttpServletRequest) request;
-
- String pathInfo = httpRequest.getPathInfo();
- String[] paths = getResourcesAndActions(pathInfo);
- final String resource = paths[0];
- final String action = paths[1];
- final String entityType = paths.length > 2 ? paths[2] : null;
- final String entityName = paths.length > 3 ? paths[3] : null;
+ RequestParts requestParts = getUserRequest(httpRequest);
if (isAuthorizationEnabled) {
- LOG.info("Authorizing user={} against resource={}, action={}, entity name={}, "
- + "entity type={}", CurrentUser.getUser(), resource, action, entityName, entityType);
- authorizationProvider.authorizeResource(resource, action,
- entityType, entityName, CurrentUser.getProxyUgi());
+ LOG.info("Authorizing user={} against request={}", CurrentUser.getUser(), requestParts);
+ authorizationProvider.authorizeResource(requestParts.getResource(),
+ requestParts.getAction(), requestParts.getEntityType(),
+ requestParts.getEntityName(), CurrentUser.getProxyUgi());
}
filterChain.doFilter(request, response);
}
- private static String[] getResourcesAndActions(String pathInfo) {
- List<String> splits = new ArrayList<String>();
+ @Override
+ public void destroy() {
+ authorizationProvider = null;
+ }
+
+ /**
+ * Returns the resource and action for the given request.
+ *
+ * @param httpRequest an HTTP servlet request
+ * @return the parts of a path
+ */
+ private static RequestParts getUserRequest(HttpServletRequest httpRequest) {
+ String pathInfo = httpRequest.getPathInfo();
final String[] pathSplits = pathInfo.substring(1).split("/");
final String resource = pathSplits[0];
+
+ ArrayList<String> splits = new ArrayList<String>();
if (resource.equals("graphs")) {
splits.add(pathSplits[1]); // resource
splits.add(pathSplits[2]); // action
@@ -98,11 +104,50 @@ public class FalconAuthorizationFilter implements Filter {
}
}
- return splits.toArray(new String[splits.size()]);
+ final String entityType = splits.size() > 2 ? splits.get(2) : null;
+ final String entityName = splits.size() > 3 ? splits.get(3) : null;
+
+ return new RequestParts(splits.get(0), splits.get(1), entityName, entityType);
}
- @Override
- public void destroy() {
- authorizationProvider = null;
+ private static class RequestParts {
+ private final String resource;
+ private final String action;
+ private final String entityName;
+ private final String entityType;
+
+ public RequestParts(String resource, String action,
+ String entityName, String entityType) {
+ this.resource = resource;
+ this.action = action;
+ this.entityName = entityName;
+ this.entityType = entityType;
+ }
+
+ public String getResource() {
+ return resource;
+ }
+
+ public String getAction() {
+ return action;
+ }
+
+ public String getEntityName() {
+ return entityName;
+ }
+
+ public String getEntityType() {
+ return entityType;
+ }
+
+ @Override
+ public String toString() {
+ return "RequestParts{"
+ + "resource='" + resource + '\''
+ + ", action='" + action + '\''
+ + ", entityName='" + entityName + '\''
+ + ", entityType='" + entityType + '\''
+ + '}';
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/prism/src/main/java/org/apache/falcon/util/Servlets.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/util/Servlets.java b/prism/src/main/java/org/apache/falcon/util/Servlets.java
new file mode 100644
index 0000000..0df2dda
--- /dev/null
+++ b/prism/src/main/java/org/apache/falcon/util/Servlets.java
@@ -0,0 +1,88 @@
+/**
+ * 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.util;
+
+import org.apache.commons.lang.StringUtils;
+
+import javax.servlet.http.HttpServletRequest;
+
+/**
+ * Utility functions for dealing with servlets.
+ */
+public final class Servlets {
+
+ private Servlets() {
+ /* singleton */
+ }
+
+ /**
+ * Returns the user of the given request.
+ *
+ * @param httpRequest an HTTP servlet request
+ * @return the user
+ */
+ public static String getUserFromRequest(HttpServletRequest httpRequest) {
+ String user = httpRequest.getRemoteUser();
+ if (!StringUtils.isEmpty(user)) {
+ return user;
+ }
+
+ user = httpRequest.getParameter("user.name"); // available in query-param
+ if (!StringUtils.isEmpty(user)) {
+ return user;
+ }
+
+ user = httpRequest.getHeader("Remote-User"); // backwards-compatibility
+ if (!StringUtils.isEmpty(user)) {
+ return user;
+ }
+
+ return null;
+ }
+
+ /**
+ * Returns the URI of the given request.
+ *
+ * @param httpRequest an HTTP servlet request
+ * @return the URI, including the query string
+ */
+ public static String getRequestURI(HttpServletRequest httpRequest) {
+ final StringBuilder url = new StringBuilder(100).append(httpRequest.getRequestURI());
+ if (httpRequest.getQueryString() != null) {
+ url.append('?').append(httpRequest.getQueryString());
+ }
+
+ return url.toString();
+ }
+
+ /**
+ * Returns the full URL of the given request.
+ *
+ * @param httpRequest an HTTP servlet request
+ * @return the full URL, including the query string
+ */
+ public static String getRequestURL(HttpServletRequest httpRequest) {
+ final StringBuilder url = new StringBuilder(100).append(httpRequest.getRequestURL());
+ if (httpRequest.getQueryString() != null) {
+ url.append('?').append(httpRequest.getQueryString());
+ }
+
+ return url.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/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 ba139aa..551bf56 100644
--- a/prism/src/main/webapp/WEB-INF/web.xml
+++ b/prism/src/main/webapp/WEB-INF/web.xml
@@ -26,6 +26,11 @@
<description>Apache Falcon Prism</description>
<filter>
+ <filter-name>audit</filter-name>
+ <filter-class>org.apache.falcon.security.FalconAuditFilter</filter-class>
+ </filter>
+
+ <filter>
<filter-name>authentication</filter-name>
<filter-class>org.apache.falcon.security.FalconAuthenticationFilter</filter-class>
</filter>
@@ -36,6 +41,11 @@
</filter>
<filter-mapping>
+ <filter-name>audit</filter-name>
+ <servlet-name>FalconProxyAPI</servlet-name>
+ </filter-mapping>
+
+ <filter-mapping>
<filter-name>authentication</filter-name>
<servlet-name>FalconProxyAPI</servlet-name>
</filter-mapping>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/prism/src/test/java/org/apache/falcon/plugin/ChainableMonitoringPluginTest.java
----------------------------------------------------------------------
diff --git a/prism/src/test/java/org/apache/falcon/plugin/ChainableMonitoringPluginTest.java b/prism/src/test/java/org/apache/falcon/plugin/ChainableMonitoringPluginTest.java
index 9a386c6..a0f6a81 100644
--- a/prism/src/test/java/org/apache/falcon/plugin/ChainableMonitoringPluginTest.java
+++ b/prism/src/test/java/org/apache/falcon/plugin/ChainableMonitoringPluginTest.java
@@ -19,17 +19,22 @@
package org.apache.falcon.plugin;
import org.apache.falcon.aspect.AlertMessage;
+import org.apache.falcon.aspect.AuditMessage;
import org.apache.falcon.aspect.GenericAlert;
import org.apache.falcon.aspect.ResourceMessage;
+import org.apache.falcon.entity.v0.SchemaHelper;
import org.apache.falcon.util.StartupProperties;
import org.testng.Assert;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
+import java.util.Date;
+
/**
* Test for ChainableMonitoringPlugin.
*/
-public class ChainableMonitoringPluginTest implements MonitoringPlugin, AlertingPlugin {
+public class ChainableMonitoringPluginTest
+ implements MonitoringPlugin, AlertingPlugin, AuditingPlugin {
@BeforeClass
public void setUp() throws Exception {
@@ -44,6 +49,8 @@ public class ChainableMonitoringPluginTest implements MonitoringPlugin, Alerting
GenericAlert.instrumentFailedInstance("cluster", "process", "agg-coord", "120:df",
"ef-id", "wf-user", "1", "DELETE", "now", "error", "none", 1242);
GenericAlert.alertJMSMessageConsumerFailed("test-alert", new Exception("test"));
+ GenericAlert.audit("falcon", "127.0.0.1", "localhost", "test-action", "127.0.0.1",
+ SchemaHelper.formatDateUTC(new Date()));
}
@Override
@@ -57,4 +64,10 @@ public class ChainableMonitoringPluginTest implements MonitoringPlugin, Alerting
Assert.assertNotNull(alertMessage);
Assert.assertEquals(alertMessage.getEvent(), "jms-message-consumer-failed");
}
+
+ @Override
+ public void audit(AuditMessage auditMessage) {
+ Assert.assertNotNull(auditMessage);
+ Assert.assertEquals(auditMessage.getRequestUrl(), "test-action");
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/prism/src/test/java/org/apache/falcon/security/FalconAuditFilterTest.java
----------------------------------------------------------------------
diff --git a/prism/src/test/java/org/apache/falcon/security/FalconAuditFilterTest.java b/prism/src/test/java/org/apache/falcon/security/FalconAuditFilterTest.java
new file mode 100644
index 0000000..0022451
--- /dev/null
+++ b/prism/src/test/java/org/apache/falcon/security/FalconAuditFilterTest.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.security;
+
+import org.apache.falcon.util.StartupProperties;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import javax.servlet.Filter;
+import javax.servlet.FilterChain;
+import javax.servlet.FilterConfig;
+import javax.servlet.ServletException;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+
+/**
+ * Test for FalconAuditFilterTest using mock objects.
+ */
+public class FalconAuditFilterTest {
+
+ @Mock
+ private HttpServletRequest mockRequest;
+
+ @Mock
+ private HttpServletResponse mockResponse;
+
+ @Mock
+ private FilterChain mockChain;
+
+ @Mock
+ private FilterConfig mockConfig;
+
+ @Mock
+ private UserGroupInformation mockUgi;
+
+ @BeforeClass
+ public void init() {
+ MockitoAnnotations.initMocks(this);
+ }
+
+ @Test
+ public void testDoFilter() throws Exception {
+ Filter filter = new FalconAuditFilter();
+ synchronized (StartupProperties.get()) {
+ filter.init(mockConfig);
+ }
+
+ Mockito.when(mockRequest.getRemoteUser()).thenReturn("falcon");
+ Mockito.when(mockRequest.getRemoteHost()).thenReturn("http://remotehost");
+ Mockito.when(mockRequest.getRequestURI()).thenReturn("http://127.0.0.1:15000");
+ filter.doFilter(mockRequest, mockResponse, mockChain);
+ }
+
+ @Test
+ public void testDoFilterNoUserHostInRequest() throws Exception {
+ Filter filter = new FalconAuditFilter();
+ synchronized (StartupProperties.get()) {
+ filter.init(mockConfig);
+ }
+
+ Mockito.when(mockRequest.getRequestURI()).thenReturn("http://127.0.0.1:15000");
+ Mockito.when(mockRequest.getQueryString()).thenReturn("user.name=guest");
+ filter.doFilter(mockRequest, mockResponse, mockChain);
+ }
+
+ @Test
+ public void testDoFilterNoUserInRequest() throws Exception {
+ Filter filter = new FalconAuditFilter();
+ synchronized (StartupProperties.get()) {
+ filter.init(mockConfig);
+ }
+
+ Mockito.when(mockRequest.getRemoteHost()).thenReturn("http://remotehost");
+ Mockito.when(mockRequest.getRequestURI()).thenReturn("http://127.0.0.1:15000");
+ Mockito.when(mockRequest.getQueryString()).thenReturn("user.name=guest");
+ filter.doFilter(mockRequest, mockResponse, mockChain);
+ }
+
+ @Test
+ public void testDoFilterNoHostInRequest() throws Exception {
+ Filter filter = new FalconAuditFilter();
+ synchronized (StartupProperties.get()) {
+ filter.init(mockConfig);
+ }
+
+ Mockito.when(mockRequest.getRemoteUser()).thenReturn("falcon");
+ Mockito.when(mockRequest.getRequestURI()).thenReturn("http://127.0.0.1:15000");
+ Mockito.when(mockRequest.getQueryString()).thenReturn("user.name=guest");
+ filter.doFilter(mockRequest, mockResponse, mockChain);
+ }
+
+ @Test
+ public void testDoFilterEmptyRequest() throws Exception {
+ Filter filter = new FalconAuditFilter();
+ synchronized (StartupProperties.get()) {
+ filter.init(mockConfig);
+ }
+
+ filter.doFilter(mockRequest, mockResponse, mockChain);
+ }
+
+ @Test (expectedExceptions = ServletException.class)
+ public void testDoFilterErrorInChain() throws Exception {
+ Filter filter = new FalconAuditFilter();
+ synchronized (StartupProperties.get()) {
+ filter.init(mockConfig);
+ }
+
+ FilterChain chain = new FilterChain() {
+ @Override
+ public void doFilter(ServletRequest request,
+ ServletResponse response) throws IOException, ServletException {
+ throw new ServletException("Something bad happened down the road");
+ }
+ };
+ Mockito.when(mockRequest.getRemoteUser()).thenReturn("bad-user");
+ Mockito.when(mockRequest.getRequestURI()).thenReturn("http://bad-host:15000");
+ Mockito.when(mockRequest.getQueryString()).thenReturn("bad=param");
+ filter.doFilter(mockRequest, mockResponse, chain);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/webapp/src/main/webapp/WEB-INF/distributed/web.xml
----------------------------------------------------------------------
diff --git a/webapp/src/main/webapp/WEB-INF/distributed/web.xml b/webapp/src/main/webapp/WEB-INF/distributed/web.xml
index 7a4de55..a92f531 100644
--- a/webapp/src/main/webapp/WEB-INF/distributed/web.xml
+++ b/webapp/src/main/webapp/WEB-INF/distributed/web.xml
@@ -26,6 +26,11 @@
<description>Apache Falcon Distributed Server</description>
<filter>
+ <filter-name>audit</filter-name>
+ <filter-class>org.apache.falcon.security.FalconAuditFilter</filter-class>
+ </filter>
+
+ <filter>
<filter-name>authentication</filter-name>
<filter-class>org.apache.falcon.security.FalconAuthenticationFilter</filter-class>
</filter>
@@ -41,6 +46,11 @@
</filter>
<filter-mapping>
+ <filter-name>audit</filter-name>
+ <servlet-name>FalconRESTApi</servlet-name>
+ </filter-mapping>
+
+ <filter-mapping>
<filter-name>authentication</filter-name>
<servlet-name>FalconRESTApi</servlet-name>
</filter-mapping>
@@ -51,6 +61,11 @@
</filter-mapping>
<filter-mapping>
+ <filter-name>audit</filter-name>
+ <servlet-name>SecureApi</servlet-name>
+ </filter-mapping>
+
+ <filter-mapping>
<filter-name>authentication</filter-name>
<servlet-name>SecureApi</servlet-name>
</filter-mapping>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/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 7d0cb08..fa2db39 100644
--- a/webapp/src/main/webapp/WEB-INF/embedded/web.xml
+++ b/webapp/src/main/webapp/WEB-INF/embedded/web.xml
@@ -26,6 +26,11 @@
<description>Apache Falcon Embedded Server</description>
<filter>
+ <filter-name>audit</filter-name>
+ <filter-class>org.apache.falcon.security.FalconAuditFilter</filter-class>
+ </filter>
+
+ <filter>
<filter-name>authentication</filter-name>
<filter-class>org.apache.falcon.security.FalconAuthenticationFilter</filter-class>
</filter>
@@ -36,6 +41,11 @@
</filter>
<filter-mapping>
+ <filter-name>audit</filter-name>
+ <servlet-name>FalconRESTApi</servlet-name>
+ </filter-mapping>
+
+ <filter-mapping>
<filter-name>authentication</filter-name>
<servlet-name>FalconRESTApi</servlet-name>
</filter-mapping>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/7ab8eb3f/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 08c30cb..2cfd7de 100644
--- a/webapp/src/main/webapp/WEB-INF/web.xml
+++ b/webapp/src/main/webapp/WEB-INF/web.xml
@@ -26,6 +26,11 @@
<description>Apache Falcon Placeholder</description>
<filter>
+ <filter-name>audit</filter-name>
+ <filter-class>org.apache.falcon.security.FalconAuditFilter</filter-class>
+ </filter>
+
+ <filter>
<filter-name>authentication</filter-name>
<filter-class>org.apache.falcon.security.FalconAuthenticationFilter</filter-class>
</filter>
@@ -36,6 +41,11 @@
</filter>
<filter-mapping>
+ <filter-name>audit</filter-name>
+ <servlet-name>FalconRESTApi</servlet-name>
+ </filter-mapping>
+
+ <filter-mapping>
<filter-name>authentication</filter-name>
<servlet-name>FalconRESTApi</servlet-name>
</filter-mapping>