EAGLE-446 convert eagle-hive app to use new app framework convert eagle-hive app to use new app framework
Author: @yonzhang2012 <[email protected]> Closes: #338 Project: http://git-wip-us.apache.org/repos/asf/incubator-eagle/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-eagle/commit/15e1c833 Tree: http://git-wip-us.apache.org/repos/asf/incubator-eagle/tree/15e1c833 Diff: http://git-wip-us.apache.org/repos/asf/incubator-eagle/diff/15e1c833 Branch: refs/heads/develop Commit: 15e1c8335ada8d6e492affc9113f8f0b39fb7d9d Parents: 502c7e3 Author: yonzhang <[email protected]> Authored: Fri Aug 12 23:15:44 2016 -0700 Committer: yonzhang <[email protected]> Committed: Fri Aug 12 23:15:44 2016 -0700 ---------------------------------------------------------------------- .../crawler/RunningJobCrawlerImpl.java | 8 +- .../eagle/jobrunning/storm/JobRunningSpout.java | 34 ++- eagle-dev/checkstyle/eagle_checks.xml | 16 ++ eagle-dev/checkstyle/suppressions.xml | 16 ++ .../security/service/HiveSensitivityEntity.java | 54 +++++ .../service/IMetadataServiceClient.java | 2 + .../security/service/ISecurityMetadataDAO.java | 4 +- .../security/service/InMemMetadataDaoImpl.java | 18 +- .../service/JDBCSecurityMetadataDAO.java | 30 ++- .../security/service/MetadataDaoFactory.java | 28 +-- .../service/MetadataServiceClientImpl.java | 16 ++ .../SecurityExternalMetadataResource.java | 2 +- .../audit/JDBCSecurityMetadataDAOTest.java | 4 +- .../hbase/HBaseAuditLogApplication.java | 4 +- ....security.hbase.HBaseAuditLogAppProvider.xml | 30 ++- .../hbase/HbaseMetadataBrowseWebResource.java | 2 +- .../resolver/HbaseSensitivityTypeResolver.java | 2 +- .../hive/dao/HiveMetadataAccessConfigDAO.java | 25 --- .../dao/HiveMetadataAccessConfigDAOImpl.java | 45 ---- .../dao/HiveSensitivityMetadataDAOImpl.java | 81 ------- .../hive/res/HiveMetadataBrowseWebResource.java | 55 ++++- .../hive/resolver/HiveCommandResolver.java | 7 + .../hive/resolver/HiveMetadataResolver.java | 22 +- .../resolver/HiveSensitivityTypeResolver.java | 29 ++- .../security/hive/TestHiveSensitivityDAO.java | 32 --- eagle-security/eagle-security-hive/pom.xml | 5 + .../hive/HiveJobRunningMonitoringTopology.java | 48 ---- .../HiveJobRunningMonitoringMain.java | 43 ---- ...HiveJobRunningSourcedStormSpoutProvider.java | 21 +- .../HiveQueryMonitoringAppProvider.java | 33 +++ .../HiveQueryMonitoringApplication.java | 80 +++++++ .../hive/jobrunning/HiveQueryParserBolt.java | 156 +++++++++++++ .../jobrunning/HiveQueryParserExecutor.java | 154 ------------- .../JobConfigurationAdaptorExecutor.java | 78 ------- .../security/hive/jobrunning/JobFilterBolt.java | 84 +++++++ .../HiveResourceSensitivityDataJoinBolt.java | 109 ++++++++++ ...HiveResourceSensitivityDataJoinExecutor.java | 100 --------- .../HiveResourceSensitivityPollingJob.java | 26 +-- ....auditlog.HiveQueryMonitoringAppProvider.xml | 218 +++++++++++++++++++ ...org.apache.eagle.app.spi.ApplicationProvider | 16 ++ .../src/main/resources/application.conf | 51 ++--- .../src/main/conf/configuration.yml | 21 ++ .../src/main/conf/configuration.yml~HEAD | 21 -- .../conf/configuration.yml~upstream_develop | 21 -- 44 files changed, 1069 insertions(+), 782 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-core/eagle-data-process/eagle-storm-jobrunning-spout/src/main/java/org/apache/eagle/jobrunning/crawler/RunningJobCrawlerImpl.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-storm-jobrunning-spout/src/main/java/org/apache/eagle/jobrunning/crawler/RunningJobCrawlerImpl.java b/eagle-core/eagle-data-process/eagle-storm-jobrunning-spout/src/main/java/org/apache/eagle/jobrunning/crawler/RunningJobCrawlerImpl.java index 72a340a..4bfa614 100644 --- a/eagle-core/eagle-data-process/eagle-storm-jobrunning-spout/src/main/java/org/apache/eagle/jobrunning/crawler/RunningJobCrawlerImpl.java +++ b/eagle-core/eagle-data-process/eagle-storm-jobrunning-spout/src/main/java/org/apache/eagle/jobrunning/crawler/RunningJobCrawlerImpl.java @@ -231,10 +231,14 @@ public class RunningJobCrawlerImpl implements RunningJobCrawler{ if (controlConfig.jobInfoEnabled) { addIntoProcessingQueueAndList(completedJobSet, queueOfCompleteJobInfo, ResourceType.JOB_COMPLETE_INFO); } - Thread.sleep(20 * 1000); } catch (Throwable t) { LOG.error("Got a throwable in fetching job completed list :", t); - } + } + try { + Thread.sleep(10 * 1000); + }catch(Exception ex){ + + } } } http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-core/eagle-data-process/eagle-storm-jobrunning-spout/src/main/java/org/apache/eagle/jobrunning/storm/JobRunningSpout.java ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-storm-jobrunning-spout/src/main/java/org/apache/eagle/jobrunning/storm/JobRunningSpout.java b/eagle-core/eagle-data-process/eagle-storm-jobrunning-spout/src/main/java/org/apache/eagle/jobrunning/storm/JobRunningSpout.java index 3dd55c1..3cc3a47 100644 --- a/eagle-core/eagle-data-process/eagle-storm-jobrunning-spout/src/main/java/org/apache/eagle/jobrunning/storm/JobRunningSpout.java +++ b/eagle-core/eagle-data-process/eagle-storm-jobrunning-spout/src/main/java/org/apache/eagle/jobrunning/storm/JobRunningSpout.java @@ -21,6 +21,7 @@ import java.util.Map; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; +import backtype.storm.tuple.Fields; import org.apache.eagle.jobrunning.callback.DefaultRunningJobInputStreamCallback; import org.apache.eagle.jobrunning.callback.RunningJobMessageId; import org.apache.eagle.jobrunning.config.RunningJobCrawlConfig; @@ -57,7 +58,7 @@ public class JobRunningSpout extends BaseRichSpout { public JobRunningSpout(RunningJobCrawlConfig config){ this(config, new JobRunningSpoutCollectorInterceptor()); } - + /** * mostly this constructor signature is for unit test purpose as you can put customized interceptor here * @param config @@ -69,8 +70,8 @@ public class JobRunningSpout extends BaseRichSpout { this.callback = new DefaultRunningJobInputStreamCallback(interceptor); this.readWriteLock = new ReentrantReadWriteLock(); } - - + + /** * TODO: just copy this part from jobHistorySpout, need to move it to a common place * @param context @@ -89,14 +90,14 @@ public class JobRunningSpout extends BaseRichSpout { } throw new IllegalStateException(); } - + @SuppressWarnings("rawtypes") @Override public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { int partitionId = calculatePartitionId(context); // sanity verify 0<=partitionId<=numTotalPartitions-1 if(partitionId < 0 || partitionId > config.controlConfig.numTotalPartitions){ - throw new IllegalStateException("partitionId should be less than numTotalPartitions with partitionId " + + throw new IllegalStateException("partitionId should be less than numTotalPartitions with partitionId " + partitionId + " and numTotalPartitions " + config.controlConfig.numTotalPartitions); } Class<? extends JobPartitioner> partitionerCls = config.controlConfig.partitionerCls; @@ -108,7 +109,7 @@ public class JobRunningSpout extends BaseRichSpout { throw new IllegalStateException(e); } JobFilter jobFilter = new JobFilterByPartition(partitioner, config.controlConfig.numTotalPartitions, partitionId); - interceptor.setSpoutOutputCollector(collector); + interceptor.setSpoutOutputCollector(collector); try { zkStateManager = new JobRunningZKStateManager(config); crawler = new RunningJobCrawlerImpl(config, zkStateManager, callback, jobFilter, readWriteLock); @@ -130,15 +131,12 @@ public class JobRunningSpout extends BaseRichSpout { }catch(Exception x){ } } - - /** - * empty because framework will take care of output fields declaration - */ + @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { - + declarer.declare(new Fields("f1", "f2")); } - + /** * add to processedJob */ @@ -161,10 +159,10 @@ public class JobRunningSpout extends BaseRichSpout { try {readWriteLock.readLock().unlock(); LOG.info("Read lock released");} catch (Throwable t) { LOG.error("Fail to release Read lock", t);} } - break; + break; default: break; - } + } } /** @@ -187,14 +185,14 @@ public class JobRunningSpout extends BaseRichSpout { } } } - + @Override public void deactivate() { - + } - + @Override public void close() { - + } } http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-dev/checkstyle/eagle_checks.xml ---------------------------------------------------------------------- diff --git a/eagle-dev/checkstyle/eagle_checks.xml b/eagle-dev/checkstyle/eagle_checks.xml index 75d4df3..5b410fe 100644 --- a/eagle-dev/checkstyle/eagle_checks.xml +++ b/eagle-dev/checkstyle/eagle_checks.xml @@ -1,4 +1,20 @@ <?xml version="1.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. +--> <!DOCTYPE module PUBLIC "-//Puppy Crawl//DTD Check Configuration 1.3//EN" "http://www.puppycrawl.com/dtds/configuration_1_3.dtd";> http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-dev/checkstyle/suppressions.xml ---------------------------------------------------------------------- diff --git a/eagle-dev/checkstyle/suppressions.xml b/eagle-dev/checkstyle/suppressions.xml index abccf77..6f47cae 100644 --- a/eagle-dev/checkstyle/suppressions.xml +++ b/eagle-dev/checkstyle/suppressions.xml @@ -1,3 +1,19 @@ +<!-- + ~ 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. + --> <!DOCTYPE suppressions PUBLIC "-//Puppy Crawl//DTD Suppressions 1.1//EN" "http://www.puppycrawl.com/dtds/suppressions_1_1.dtd";> <suppressions> http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/HiveSensitivityEntity.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/HiveSensitivityEntity.java b/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/HiveSensitivityEntity.java new file mode 100644 index 0000000..d6c6973 --- /dev/null +++ b/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/HiveSensitivityEntity.java @@ -0,0 +1,54 @@ +/* + * + * * 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.eagle.security.service; + +/** + * Since 8/12/16. + */ +public class HiveSensitivityEntity { + private String site; + private String hiveResource; + private String sensitivityType; + + public String getSite() { + return site; + } + + public void setSite(String site) { + this.site = site; + } + + public String getHiveResource() { + return hiveResource; + } + + public void setHiveResource(String hiveResource) { + this.hiveResource = hiveResource; + } + + public String getSensitivityType() { + return sensitivityType; + } + + public void setSensitivityType(String sensitivityType) { + this.sensitivityType = sensitivityType; + } +} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/IMetadataServiceClient.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/IMetadataServiceClient.java b/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/IMetadataServiceClient.java index 79db47e..d87cf13 100644 --- a/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/IMetadataServiceClient.java +++ b/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/IMetadataServiceClient.java @@ -33,4 +33,6 @@ public interface IMetadataServiceClient extends Closeable, Serializable { OpResult addHdfsSensitivity(Collection<HdfsSensitivityEntity> h); Collection<IPZoneEntity> listIPZones(); OpResult addIPZone(Collection<IPZoneEntity> h); + Collection<HiveSensitivityEntity> listHiveSensitivities(); + OpResult addHiveSensitivity(Collection<HiveSensitivityEntity> h); } http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/ISecurityMetadataDAO.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/ISecurityMetadataDAO.java b/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/ISecurityMetadataDAO.java index 6158bac..49d2ed5 100644 --- a/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/ISecurityMetadataDAO.java +++ b/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/ISecurityMetadataDAO.java @@ -23,10 +23,12 @@ import java.util.Collection; * Since 6/10/16. */ public interface ISecurityMetadataDAO { - Collection<HBaseSensitivityEntity> listHBaseSensitivies(); + Collection<HBaseSensitivityEntity> listHBaseSensitivities(); OpResult addHBaseSensitivity(Collection<HBaseSensitivityEntity> h); Collection<HdfsSensitivityEntity> listHdfsSensitivities(); OpResult addHdfsSensitivity(Collection<HdfsSensitivityEntity> h); Collection<IPZoneEntity> listIPZones(); OpResult addIPZone(Collection<IPZoneEntity> h); + Collection<HiveSensitivityEntity> listHiveSensitivities(); + OpResult addHiveSensitivity(Collection<HiveSensitivityEntity> h); } http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/InMemMetadataDaoImpl.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/InMemMetadataDaoImpl.java b/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/InMemMetadataDaoImpl.java index 1869538..2699fe9 100644 --- a/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/InMemMetadataDaoImpl.java +++ b/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/InMemMetadataDaoImpl.java @@ -39,14 +39,14 @@ public class InMemMetadataDaoImpl implements ISecurityMetadataDAO { private Map<Pair<String, String>, HBaseSensitivityEntity> hBaseSensitivityEntities = new HashMap<>(); private Map<Pair<String, String>, HdfsSensitivityEntity> hdfsSensitivityEntities = new HashMap<>(); private Map<String, IPZoneEntity> ipZones = new HashMap<>(); - + private Map<Pair<String, String>, HiveSensitivityEntity> hiveSensitivityEntities = new HashMap<>(); @Inject public InMemMetadataDaoImpl() { } @Override - public synchronized Collection<HBaseSensitivityEntity> listHBaseSensitivies() { + public synchronized Collection<HBaseSensitivityEntity> listHBaseSensitivities() { return hBaseSensitivityEntities.values(); } @@ -85,4 +85,18 @@ public class InMemMetadataDaoImpl implements ISecurityMetadataDAO { } return new OpResult(); } + + @Override + public synchronized Collection<HiveSensitivityEntity> listHiveSensitivities() { + return hiveSensitivityEntities.values(); + } + + @Override + public synchronized OpResult addHiveSensitivity(Collection<HiveSensitivityEntity> h) { + for (HiveSensitivityEntity e : h) { + Pair p = new ImmutablePair<>(e.getSite(), e.getHiveResource()); + hiveSensitivityEntities.put(p, e); + } + return new OpResult(); + } } http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/JDBCSecurityMetadataDAO.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/JDBCSecurityMetadataDAO.java b/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/JDBCSecurityMetadataDAO.java index 679f3d0..05713fc 100644 --- a/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/JDBCSecurityMetadataDAO.java +++ b/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/JDBCSecurityMetadataDAO.java @@ -114,7 +114,7 @@ public class JDBCSecurityMetadataDAO implements ISecurityMetadataDAO { } @Override - public Collection<HBaseSensitivityEntity> listHBaseSensitivies() { + public Collection<HBaseSensitivityEntity> listHBaseSensitivities() { return listEntities(HBASE_QUERY_ALL_STATEMENT, rs -> { try { HBaseSensitivityEntity entity = new HBaseSensitivityEntity(); @@ -196,6 +196,34 @@ public class JDBCSecurityMetadataDAO implements ISecurityMetadataDAO { }); } + @Override + public Collection<HiveSensitivityEntity> listHiveSensitivities() { + return listEntities(HIVE_QUERY_ALL_STATEMENT, rs -> { + try { + HiveSensitivityEntity entity = new HiveSensitivityEntity(); + entity.setSite(rs.getString(1)); + entity.setHiveResource(rs.getString(2)); + entity.setSensitivityType(rs.getString(3)); + return entity; + }catch(Exception ex){ throw new IllegalStateException(ex);} + }); + } + + @Override + public OpResult addHiveSensitivity(Collection<HiveSensitivityEntity> h) { + return addEntities(HIVE_INSERT_STATEMENT, h, (entity, statement) -> { + HiveSensitivityEntity e = (HiveSensitivityEntity)entity; + try { + statement.setString(1, e.getSite()); + statement.setString(2, e.getHiveResource()); + statement.setString(3, e.getSensitivityType()); + }catch(Exception ex){ + throw new IllegalStateException(ex); + } + return statement; + }); + } + private Connection getJdbcConnection() throws Exception { Connection connection; try { http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/MetadataDaoFactory.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/MetadataDaoFactory.java b/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/MetadataDaoFactory.java index 91240a4..383fb79 100644 --- a/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/MetadataDaoFactory.java +++ b/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/MetadataDaoFactory.java @@ -1,22 +1,18 @@ /* + * 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 * - * * - * * * 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 - * * * <p/> - * * * http://www.apache.org/licenses/LICENSE-2.0 - * * * <p/> - * * * 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. - * * + * 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.eagle.security.service; http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/MetadataServiceClientImpl.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/MetadataServiceClientImpl.java b/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/MetadataServiceClientImpl.java index cac4630..c350c2c 100644 --- a/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/MetadataServiceClientImpl.java +++ b/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/MetadataServiceClientImpl.java @@ -46,6 +46,9 @@ public class MetadataServiceClientImpl implements IMetadataServiceClient { private static final String METADATA_LIST_HDFS_SENSITIVITY_PATH = "/metadata/security/hdfsSensitivity"; private static final String METADATA_ADD_HDFS_SENSITIVITY_PATH = "/metadata/security/hdfsSensitivity"; + private static final String METADATA_LIST_HIVE_SENSITIVITY_PATH = "/metadata/security/hiveSensitivity"; + private static final String METADATA_ADD_HIVE_SENSITIVITY_PATH = "/metadata/security/hiveSensitivity"; + private static final String METADATA_LIST_IPZONE_PATH = "/metadata/security/ipzone"; private static final String METADATA_ADD_IPZONE_PATH = "/metadata/security/ipzone"; @@ -141,4 +144,17 @@ public class MetadataServiceClientImpl implements IMetadataServiceClient { r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(h); return new OpResult(); } + + @Override + public Collection<HiveSensitivityEntity> listHiveSensitivities() { + return list(METADATA_LIST_HIVE_SENSITIVITY_PATH, new GenericType<List<HiveSensitivityEntity>>() { + }); + } + + @Override + public OpResult addHiveSensitivity(Collection<HiveSensitivityEntity> h) { + WebResource r = client.resource(basePath + METADATA_ADD_HIVE_SENSITIVITY_PATH); + r.accept(MediaType.APPLICATION_JSON_TYPE).type(MediaType.APPLICATION_JSON).post(h); + return new OpResult(); + } } http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/SecurityExternalMetadataResource.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/SecurityExternalMetadataResource.java b/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/SecurityExternalMetadataResource.java index 404c10d..f0ec69e 100644 --- a/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/SecurityExternalMetadataResource.java +++ b/eagle-security/eagle-security-common/src/main/java/org/apache/eagle/security/service/SecurityExternalMetadataResource.java @@ -42,7 +42,7 @@ public class SecurityExternalMetadataResource { @GET @Produces("application/json") public Collection<HBaseSensitivityEntity> getHBaseSensitivites(@QueryParam("site") String site){ - return dao.listHBaseSensitivies(); + return dao.listHBaseSensitivities(); } @Path("/hbaseSensitivity") http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-common/src/test/java/org/apache/eagle/security/crawler/audit/JDBCSecurityMetadataDAOTest.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-common/src/test/java/org/apache/eagle/security/crawler/audit/JDBCSecurityMetadataDAOTest.java b/eagle-security/eagle-security-common/src/test/java/org/apache/eagle/security/crawler/audit/JDBCSecurityMetadataDAOTest.java index fd86c29..b9e40ba 100644 --- a/eagle-security/eagle-security-common/src/test/java/org/apache/eagle/security/crawler/audit/JDBCSecurityMetadataDAOTest.java +++ b/eagle-security/eagle-security-common/src/test/java/org/apache/eagle/security/crawler/audit/JDBCSecurityMetadataDAOTest.java @@ -48,7 +48,7 @@ public class JDBCSecurityMetadataDAOTest { entity.setSite("test_site"); entity.setHbaseResource("test_hbaseResource"); metadataDAO.addHBaseSensitivity(Collections.singletonList(entity)); - Collection<HBaseSensitivityEntity> entities = metadataDAO.listHBaseSensitivies(); + Collection<HBaseSensitivityEntity> entities = metadataDAO.listHBaseSensitivities(); Assert.assertEquals(1,entities.size()); Assert.assertEquals("test_site",entities.iterator().next().getSite()); } @@ -57,4 +57,4 @@ public class JDBCSecurityMetadataDAOTest { public void after() throws SQLException { queryService.dropTable("hbase_sensitivity_entity"); } -} \ No newline at end of file +} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hbase-auditlog/src/main/java/org/apache/eagle/security/hbase/HBaseAuditLogApplication.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hbase-auditlog/src/main/java/org/apache/eagle/security/hbase/HBaseAuditLogApplication.java b/eagle-security/eagle-security-hbase-auditlog/src/main/java/org/apache/eagle/security/hbase/HBaseAuditLogApplication.java index 4a7c2df..030212a 100644 --- a/eagle-security/eagle-security-hbase-auditlog/src/main/java/org/apache/eagle/security/hbase/HBaseAuditLogApplication.java +++ b/eagle-security/eagle-security-hbase-auditlog/src/main/java/org/apache/eagle/security/hbase/HBaseAuditLogApplication.java @@ -53,7 +53,7 @@ public class HBaseAuditLogApplication extends StormApplication { builder.setSpout("ingest", spout, numOfSpoutTasks); BoltDeclarer boltDeclarer = builder.setBolt("parserBolt", bolt, numOfParserTasks); - boltDeclarer.fieldsGrouping("ingest", new Fields(StringScheme.STRING_SCHEME_KEY)); + boltDeclarer.fieldsGrouping("ingest", new Fields("f1")); HbaseResourceSensitivityDataJoinBolt joinBolt = new HbaseResourceSensitivityDataJoinBolt(config); BoltDeclarer joinBoltDeclarer = builder.setBolt("joinBolt", joinBolt, numOfJoinTasks); @@ -70,4 +70,4 @@ public class HBaseAuditLogApplication extends StormApplication { HBaseAuditLogApplication app = new HBaseAuditLogApplication(); app.run(config); } -} \ No newline at end of file +} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hbase-auditlog/src/main/resources/META-INF/providers/org.apache.eagle.security.hbase.HBaseAuditLogAppProvider.xml ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hbase-auditlog/src/main/resources/META-INF/providers/org.apache.eagle.security.hbase.HBaseAuditLogAppProvider.xml b/eagle-security/eagle-security-hbase-auditlog/src/main/resources/META-INF/providers/org.apache.eagle.security.hbase.HBaseAuditLogAppProvider.xml index 0225a50..f8d26e5 100644 --- a/eagle-security/eagle-security-hbase-auditlog/src/main/resources/META-INF/providers/org.apache.eagle.security.hbase.HBaseAuditLogAppProvider.xml +++ b/eagle-security/eagle-security-hbase-auditlog/src/main/resources/META-INF/providers/org.apache.eagle.security.hbase.HBaseAuditLogAppProvider.xml @@ -1,21 +1,19 @@ <?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 - ~ * <p/> - ~ * http://www.apache.org/licenses/LICENSE-2.0 - ~ * <p/> - ~ * 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. - ~ */ + ~ 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. --> <application> http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hbase-web/src/main/java/org/apache/eagle/service/security/hbase/HbaseMetadataBrowseWebResource.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hbase-web/src/main/java/org/apache/eagle/service/security/hbase/HbaseMetadataBrowseWebResource.java b/eagle-security/eagle-security-hbase-web/src/main/java/org/apache/eagle/service/security/hbase/HbaseMetadataBrowseWebResource.java index 9b98649..68ea552 100644 --- a/eagle-security/eagle-security-hbase-web/src/main/java/org/apache/eagle/service/security/hbase/HbaseMetadataBrowseWebResource.java +++ b/eagle-security/eagle-security-hbase-web/src/main/java/org/apache/eagle/service/security/hbase/HbaseMetadataBrowseWebResource.java @@ -54,7 +54,7 @@ public class HbaseMetadataBrowseWebResource { private Map<String, Map<String, String>> getAllSensitivities(){ Map<String, Map<String, String>> all = new HashMap<>(); - Collection<HBaseSensitivityEntity> entities = dao.listHBaseSensitivies(); + Collection<HBaseSensitivityEntity> entities = dao.listHBaseSensitivities(); for(HBaseSensitivityEntity entity : entities){ if(!all.containsKey(entity.getSite())){ all.put(entity.getSite(), new HashMap<>()); http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hbase-web/src/main/java/org/apache/eagle/service/security/hbase/resolver/HbaseSensitivityTypeResolver.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hbase-web/src/main/java/org/apache/eagle/service/security/hbase/resolver/HbaseSensitivityTypeResolver.java b/eagle-security/eagle-security-hbase-web/src/main/java/org/apache/eagle/service/security/hbase/resolver/HbaseSensitivityTypeResolver.java index 9df322b..b5ad203 100644 --- a/eagle-security/eagle-security-hbase-web/src/main/java/org/apache/eagle/service/security/hbase/resolver/HbaseSensitivityTypeResolver.java +++ b/eagle-security/eagle-security-hbase-web/src/main/java/org/apache/eagle/service/security/hbase/resolver/HbaseSensitivityTypeResolver.java @@ -45,7 +45,7 @@ public class HbaseSensitivityTypeResolver extends AbstractSensitivityTypeResolve private Map<String, Map<String, String>> getAllSensitivities(){ Map<String, Map<String, String>> all = new HashMap<>(); - Collection<HBaseSensitivityEntity> entities = dao.listHBaseSensitivies(); + Collection<HBaseSensitivityEntity> entities = dao.listHBaseSensitivities(); for(HBaseSensitivityEntity entity : entities){ if(!all.containsKey(entity.getSite())){ all.put(entity.getSite(), new HashMap<>()); http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/dao/HiveMetadataAccessConfigDAO.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/dao/HiveMetadataAccessConfigDAO.java b/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/dao/HiveMetadataAccessConfigDAO.java deleted file mode 100644 index 414fa60..0000000 --- a/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/dao/HiveMetadataAccessConfigDAO.java +++ /dev/null @@ -1,25 +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.eagle.service.security.hive.dao; - -import java.util.Map; - -public interface HiveMetadataAccessConfigDAO { - - // HiveMetadataAccessConfig for one site - HiveMetadataAccessConfig getConfig(String site) throws Exception; -} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/dao/HiveMetadataAccessConfigDAOImpl.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/dao/HiveMetadataAccessConfigDAOImpl.java b/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/dao/HiveMetadataAccessConfigDAOImpl.java deleted file mode 100644 index f4f41e0..0000000 --- a/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/dao/HiveMetadataAccessConfigDAOImpl.java +++ /dev/null @@ -1,45 +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.eagle.service.security.hive.dao; - -import com.typesafe.config.Config; -import org.apache.eagle.alert.entity.AlertDataSourceEntity; -import org.apache.eagle.log.entity.GenericServiceAPIResponseEntity; -import org.apache.eagle.log.entity.ListQueryAPIResponseEntity; -import org.apache.eagle.policy.common.Constants; -import org.apache.eagle.security.resolver.MetadataAccessConfigRepo; -import org.apache.eagle.service.generic.GenericEntityServiceResource; -import org.apache.eagle.service.generic.ListQueryResource; -import com.fasterxml.jackson.databind.ObjectMapper; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -public class HiveMetadataAccessConfigDAOImpl implements HiveMetadataAccessConfigDAO { - private static Logger LOG = LoggerFactory.getLogger(HiveMetadataAccessConfigDAOImpl.class); - - // HiveMetadataAccessConfig for one site - @Override - public HiveMetadataAccessConfig getConfig(String site) throws Exception{ - MetadataAccessConfigRepo repo = new MetadataAccessConfigRepo(); - Config config = repo.getConfig("hiveQueryLog", site); - return HiveMetadataAccessConfig.config2Entity(config); - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/dao/HiveSensitivityMetadataDAOImpl.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/dao/HiveSensitivityMetadataDAOImpl.java b/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/dao/HiveSensitivityMetadataDAOImpl.java deleted file mode 100644 index 96f3f27..0000000 --- a/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/dao/HiveSensitivityMetadataDAOImpl.java +++ /dev/null @@ -1,81 +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.eagle.service.security.hive.dao; - -import org.apache.eagle.log.entity.GenericServiceAPIResponseEntity; -import org.apache.eagle.service.generic.GenericEntityServiceResource; -import org.apache.eagle.security.entity.HiveResourceSensitivityAPIEntity; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -public class HiveSensitivityMetadataDAOImpl implements HiveSensitivityMetadataDAO{ - private static Logger LOG = LoggerFactory.getLogger(HiveSensitivityMetadataDAOImpl.class); - - @Override - public Map<String, Map<String, String>> getAllHiveSensitivityMap(){ - GenericEntityServiceResource resource = new GenericEntityServiceResource(); - /* parameters are: query, startTime, endTime, pageSzie, startRowkey, treeAgg, timeSeries, intervalmin, top, filterIfMissing, - * parallel, metricName*/ - GenericServiceAPIResponseEntity ret = resource.search("HiveResourceSensitivityService[]{*}", null, null, Integer.MAX_VALUE, null, false, false, 0L, 0, false, - 0, null, false); - List<HiveResourceSensitivityAPIEntity> list = (List<HiveResourceSensitivityAPIEntity>) ret.getObj(); - if( list == null ) - return Collections.emptyMap(); - Map<String, Map<String, String>> res = new HashMap<String, Map<String, String>>(); - - for(HiveResourceSensitivityAPIEntity entity : list){ - String site = entity.getTags().get("site"); - if(entity.getTags().containsKey("hiveResource")) { - if(res.get(site) == null){ - res.put(site, new HashMap<String, String>()); - } - Map<String, String> resSensitivityMap = res.get(site); - resSensitivityMap.put(entity.getTags().get("hiveResource"), entity.getSensitivityType()); - } - else { - if(LOG.isDebugEnabled()) { - LOG.debug("An invalid sensitivity entity is detected" + entity); - } - } - } - return res; - } - - @Override - public Map<String, String> getHiveSensitivityMap(String site){ - GenericEntityServiceResource resource = new GenericEntityServiceResource(); - String queryFormat = "HiveResourceSensitivityService[@site=\"%s\"]{*}"; - GenericServiceAPIResponseEntity ret = resource.search(String.format(queryFormat, site), null, null, Integer.MAX_VALUE, null, false, false, 0L, 0, false, - 0, null, false); - List<HiveResourceSensitivityAPIEntity> list = (List<HiveResourceSensitivityAPIEntity>) ret.getObj(); - if( list == null ) - return Collections.emptyMap(); - Map<String, String> resSensitivityMap = new HashMap<String, String>(); - for(HiveResourceSensitivityAPIEntity entity : list){ - if(entity.getTags().containsKey("hiveResource")) { - resSensitivityMap.put(entity.getTags().get("hiveResource"), entity.getSensitivityType()); - } - } - return resSensitivityMap; - } - -} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/res/HiveMetadataBrowseWebResource.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/res/HiveMetadataBrowseWebResource.java b/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/res/HiveMetadataBrowseWebResource.java index 7139000..ef88bfd 100644 --- a/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/res/HiveMetadataBrowseWebResource.java +++ b/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/res/HiveMetadataBrowseWebResource.java @@ -16,6 +16,14 @@ */ package org.apache.eagle.service.security.hive.res; +import com.google.inject.Inject; +import com.typesafe.config.Config; +import com.typesafe.config.ConfigFactory; +import org.apache.eagle.metadata.model.ApplicationEntity; +import org.apache.eagle.metadata.service.ApplicationEntityService; +import org.apache.eagle.security.service.HiveSensitivityEntity; +import org.apache.eagle.security.service.ISecurityMetadataDAO; +import org.apache.eagle.security.service.MetadataDaoFactory; import org.apache.eagle.service.common.EagleExceptionWrapper; import org.apache.eagle.security.entity.HiveResourceEntity; import org.apache.eagle.service.security.hive.dao.*; @@ -30,8 +38,16 @@ import java.util.regex.Pattern; @Path("/hiveResource") public class HiveMetadataBrowseWebResource { private static Logger LOG = LoggerFactory.getLogger(HiveMetadataBrowseWebResource.class); - private HiveSensitivityMetadataDAOImpl dao = new HiveSensitivityMetadataDAOImpl(); - private Map<String, Map<String, String>> maps = dao.getAllHiveSensitivityMap(); + + final public static String HIVE_APPLICATION = "HiveQueryMonitoringApplication"; + private ApplicationEntityService entityService; + private ISecurityMetadataDAO dao; + + @Inject + public HiveMetadataBrowseWebResource(ApplicationEntityService entityService, Config eagleServerConfig){ + this.entityService = entityService; + dao = MetadataDaoFactory.getMetadataDAO(eagleServerConfig); + } @Path("/databases") @GET @@ -43,8 +59,10 @@ public class HiveMetadataBrowseWebResource { List<String> databases = null; List<HiveResourceEntity> values = new ArrayList<>(); try { - HiveMetadataAccessConfig config = new HiveMetadataAccessConfigDAOImpl().getConfig(site); - HiveMetadataDAO dao = new HiveMetadataDAOFactory().getHiveMetadataDAO(config); + Map<String, Object> config = getAppConfig(site, HIVE_APPLICATION); + Config typesafeConfig = ConfigFactory.parseMap(config); + HiveMetadataAccessConfig hiveConfig = HiveMetadataAccessConfig.config2Entity(typesafeConfig); + HiveMetadataDAO dao = new HiveMetadataDAOFactory().getHiveMetadataDAO(hiveConfig); databases = dao.getDatabases(); } catch(Exception ex){ LOG.error("fail getting databases", ex); @@ -74,8 +92,10 @@ public class HiveMetadataBrowseWebResource { List<String> tables = null; List<HiveResourceEntity> values = new ArrayList<>(); try { - HiveMetadataAccessConfig config = new HiveMetadataAccessConfigDAOImpl().getConfig(site); - HiveMetadataDAO dao = new HiveMetadataDAOFactory().getHiveMetadataDAO(config); + Map<String, Object> config = getAppConfig(site, HIVE_APPLICATION); + Config typesafeConfig = ConfigFactory.parseMap(config); + HiveMetadataAccessConfig hiveConfig = HiveMetadataAccessConfig.config2Entity(typesafeConfig); + HiveMetadataDAO dao = new HiveMetadataDAOFactory().getHiveMetadataDAO(hiveConfig); tables = dao.getTables(database); }catch(Exception ex){ LOG.error("fail getting databases", ex); @@ -104,8 +124,10 @@ public class HiveMetadataBrowseWebResource { List<String> columns = null; List<HiveResourceEntity> values = new ArrayList<>(); try { - HiveMetadataAccessConfig config = new HiveMetadataAccessConfigDAOImpl().getConfig(site); - HiveMetadataDAO dao = new HiveMetadataDAOFactory().getHiveMetadataDAO(config); + Map<String, Object> config = getAppConfig(site, HIVE_APPLICATION); + Config typesafeConfig = ConfigFactory.parseMap(config); + HiveMetadataAccessConfig hiveConfig = HiveMetadataAccessConfig.config2Entity(typesafeConfig); + HiveMetadataDAO dao = new HiveMetadataDAOFactory().getHiveMetadataDAO(hiveConfig); columns = dao.getColumns(database, table); }catch(Exception ex){ LOG.error("fail getting databases", ex); @@ -125,6 +147,7 @@ public class HiveMetadataBrowseWebResource { } String checkSensitivity(String site, String resource, Set<String> childSensitiveTypes) { + Map<String, Map<String, String>> maps = getAllSensitivities(); String sensitiveType = null; if (maps != null && maps.get(site) != null) { Map<String, String> map = maps.get(site); @@ -142,4 +165,20 @@ public class HiveMetadataBrowseWebResource { return sensitiveType; } + private Map<String, Map<String, String>> getAllSensitivities(){ + Map<String, Map<String, String>> all = new HashMap<>(); + Collection<HiveSensitivityEntity> entities = dao.listHiveSensitivities(); + for(HiveSensitivityEntity entity : entities){ + if(!all.containsKey(entity.getSite())){ + all.put(entity.getSite(), new HashMap<>()); + } + all.get(entity.getSite()).put(entity.getHiveResource(), entity.getSensitivityType()); + } + return all; + } + + private Map<String, Object> getAppConfig(String site, String appType){ + ApplicationEntity entity = entityService.getBySiteIdAndAppType(site, appType); + return entity.getConfiguration(); + } } http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/resolver/HiveCommandResolver.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/resolver/HiveCommandResolver.java b/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/resolver/HiveCommandResolver.java index f1b2fc4..45a79c5 100644 --- a/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/resolver/HiveCommandResolver.java +++ b/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/resolver/HiveCommandResolver.java @@ -16,7 +16,9 @@ */ package org.apache.eagle.service.security.hive.resolver; +import com.typesafe.config.Config; import org.apache.commons.lang3.StringUtils; +import org.apache.eagle.metadata.service.ApplicationEntityService; import org.apache.eagle.service.alert.resolver.AttributeResolvable; import org.apache.eagle.service.alert.resolver.AttributeResolveException; import org.apache.eagle.service.alert.resolver.BadAttributeResolveRequestException; @@ -39,6 +41,11 @@ public class HiveCommandResolver implements AttributeResolvable<GenericAttribute private final static List<String> commands = Arrays.asList(cmdStrs); + private ApplicationEntityService entityService; + + public HiveCommandResolver(ApplicationEntityService entityService, Config eagleServerConfig){ + this.entityService = entityService; + } public List<String> resolve(GenericAttributeResolveRequest request) throws AttributeResolveException { String query = request.getQuery().trim(); http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/resolver/HiveMetadataResolver.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/resolver/HiveMetadataResolver.java b/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/resolver/HiveMetadataResolver.java index 29ea183..31908d9 100644 --- a/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/resolver/HiveMetadataResolver.java +++ b/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/resolver/HiveMetadataResolver.java @@ -22,11 +22,16 @@ import java.util.List; import java.util.Map; import java.util.regex.Pattern; +import com.typesafe.config.Config; +import com.typesafe.config.ConfigFactory; +import org.apache.eagle.metadata.model.ApplicationEntity; +import org.apache.eagle.metadata.service.ApplicationEntityService; import org.apache.eagle.service.alert.resolver.AttributeResolvable; import org.apache.eagle.service.alert.resolver.AttributeResolveException; import org.apache.eagle.service.alert.resolver.BadAttributeResolveRequestException; import org.apache.eagle.service.alert.resolver.GenericAttributeResolveRequest; import org.apache.eagle.service.security.hive.dao.*; +import org.apache.eagle.service.security.hive.res.HiveMetadataBrowseWebResource; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,6 +41,12 @@ public class HiveMetadataResolver implements AttributeResolvable<GenericAttribut private final static String HIVE_ATTRIBUTE_RESOLVE_FORMAT_HINT = "hive metadata resolve must be {\"site\":\"${site}\", \"query\"=\"/{db}/{table}/{column}\"}"; + private ApplicationEntityService entityService; + + public HiveMetadataResolver(ApplicationEntityService entityService, Config eagleServerConfig){ + this.entityService = entityService; + } + @Override public List<String> resolve(GenericAttributeResolveRequest request) throws AttributeResolveException { // query should be formatted as "/{db}/{table}/{column}" with "/" as leading character @@ -44,8 +55,10 @@ public class HiveMetadataResolver implements AttributeResolvable<GenericAttribut String prefix = null; try { - HiveMetadataAccessConfig config = new HiveMetadataAccessConfigDAOImpl().getConfig(request.getSite()); - HiveMetadataDAO dao = new HiveMetadataDAOFactory().getHiveMetadataDAO(config); + Map<String, Object> config = getAppConfig(request.getSite(), HiveMetadataBrowseWebResource.HIVE_APPLICATION); + Config typesafeConfig = ConfigFactory.parseMap(config); + HiveMetadataAccessConfig hiveConfig = HiveMetadataAccessConfig.config2Entity(typesafeConfig); + HiveMetadataDAO dao = new HiveMetadataDAOFactory().getHiveMetadataDAO(hiveConfig); if (subResources.length == 0) { // query all databases with "/" return filterAndCombineAttribue("/", dao.getDatabases(), null); }else if(subResources.length == 2){ // query all tables given a database @@ -109,4 +122,9 @@ public class HiveMetadataResolver implements AttributeResolvable<GenericAttribut public Class<GenericAttributeResolveRequest> getRequestClass() { return GenericAttributeResolveRequest.class; } + + private Map<String, Object> getAppConfig(String site, String appType){ + ApplicationEntity entity = entityService.getBySiteIdAndAppType(site, appType); + return entity.getConfiguration(); + } } http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/resolver/HiveSensitivityTypeResolver.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/resolver/HiveSensitivityTypeResolver.java b/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/resolver/HiveSensitivityTypeResolver.java index 3431afa..2a4363f 100644 --- a/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/resolver/HiveSensitivityTypeResolver.java +++ b/eagle-security/eagle-security-hive-web/src/main/java/org/apache/eagle/service/security/hive/resolver/HiveSensitivityTypeResolver.java @@ -16,11 +16,15 @@ */ package org.apache.eagle.service.security.hive.resolver; +import com.typesafe.config.Config; +import org.apache.eagle.metadata.service.ApplicationEntityService; +import org.apache.eagle.security.service.HiveSensitivityEntity; +import org.apache.eagle.security.service.ISecurityMetadataDAO; +import org.apache.eagle.security.service.MetadataDaoFactory; import org.apache.eagle.service.alert.resolver.AttributeResolvable; import org.apache.eagle.service.alert.resolver.AttributeResolveException; import org.apache.eagle.service.alert.resolver.BadAttributeResolveRequestException; import org.apache.eagle.service.alert.resolver.GenericAttributeResolveRequest; -import org.apache.eagle.service.security.hive.dao.HiveSensitivityMetadataDAOImpl; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -29,17 +33,20 @@ import java.util.regex.Pattern; public class HiveSensitivityTypeResolver implements AttributeResolvable<GenericAttributeResolveRequest,String> { private final static Logger LOG = LoggerFactory.getLogger(HiveSensitivityTypeResolver.class); - private HiveSensitivityMetadataDAOImpl dao = new HiveSensitivityMetadataDAOImpl(); - private Map<String, Map<String, String>> maps = dao.getAllHiveSensitivityMap(); - + private ISecurityMetadataDAO dao; + private ApplicationEntityService entityService; + public HiveSensitivityTypeResolver(ApplicationEntityService entityService, Config eagleServerConfig){ + this.entityService = entityService; + dao = MetadataDaoFactory.getMetadataDAO(eagleServerConfig); + } private final static String SENSITIVETYPE_ATTRIBUTE_RESOLVE_FORMAT_HINT = "Sensitive type should be composed of a-z, A-Z, 0-9 or -"; public List<String> resolve(GenericAttributeResolveRequest request) throws AttributeResolveException { String query = request.getQuery().trim(); String site = request.getSite().trim(); List<String> res = new ArrayList<>(); - Map<String, String> map = maps.get(site); + Map<String, String> map = getAllSensitivities().get(site); if(map == null) { return res; @@ -70,4 +77,16 @@ public class HiveSensitivityTypeResolver implements AttributeResolvable<GenericA public Class<GenericAttributeResolveRequest> getRequestClass() { return GenericAttributeResolveRequest.class; } + + private Map<String, Map<String, String>> getAllSensitivities(){ + Map<String, Map<String, String>> all = new HashMap<>(); + Collection<HiveSensitivityEntity> entities = dao.listHiveSensitivities(); + for(HiveSensitivityEntity entity : entities){ + if(!all.containsKey(entity.getSite())){ + all.put(entity.getSite(), new HashMap<>()); + } + all.get(entity.getSite()).put(entity.getHiveResource(), entity.getSensitivityType()); + } + return all; + } } http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hive-web/src/test/java/org/apache/eagle/service/security/hive/TestHiveSensitivityDAO.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hive-web/src/test/java/org/apache/eagle/service/security/hive/TestHiveSensitivityDAO.java b/eagle-security/eagle-security-hive-web/src/test/java/org/apache/eagle/service/security/hive/TestHiveSensitivityDAO.java deleted file mode 100644 index 0bb90be..0000000 --- a/eagle-security/eagle-security-hive-web/src/test/java/org/apache/eagle/service/security/hive/TestHiveSensitivityDAO.java +++ /dev/null @@ -1,32 +0,0 @@ -package org.apache.eagle.service.security.hive;/* - * 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. - */ -import org.apache.eagle.service.security.hive.dao.HiveSensitivityMetadataDAOImpl; -import org.junit.Test; - -public class TestHiveSensitivityDAO { - //@Test - public void testHiveSensitivityDAO(){ - HiveSensitivityMetadataDAOImpl dao = new HiveSensitivityMetadataDAOImpl(); - System.out.println(dao.getAllHiveSensitivityMap()); - System.out.println(dao.getHiveSensitivityMap("sandbox")); - } - - @Test - public void test() { - - } -} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hive/pom.xml ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hive/pom.xml b/eagle-security/eagle-security-hive/pom.xml index 262d99d..1d4bdc8 100644 --- a/eagle-security/eagle-security-hive/pom.xml +++ b/eagle-security/eagle-security-hive/pom.xml @@ -90,5 +90,10 @@ <groupId>org.scala-lang</groupId> <artifactId>scala-library</artifactId> </dependency> + <dependency> + <groupId>org.apache.eagle</groupId> + <artifactId>eagle-app-base</artifactId> + <version>${project.version}</version> + </dependency> </dependencies> </project> http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/HiveJobRunningMonitoringTopology.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/HiveJobRunningMonitoringTopology.java b/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/HiveJobRunningMonitoringTopology.java deleted file mode 100644 index 81f329d..0000000 --- a/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/HiveJobRunningMonitoringTopology.java +++ /dev/null @@ -1,48 +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.eagle.security.hive; - - -import com.typesafe.config.Config; -import org.apache.eagle.datastream.ExecutionEnvironments; -import org.apache.eagle.datastream.storm.StormExecutionEnvironment; -import org.apache.eagle.security.hive.jobrunning.HiveJobRunningSourcedStormSpoutProvider; -import org.apache.eagle.security.hive.jobrunning.HiveQueryParserExecutor; -import org.apache.eagle.security.hive.jobrunning.JobConfigurationAdaptorExecutor; -import org.apache.eagle.security.hive.sensitivity.HiveResourceSensitivityDataJoinExecutor; -import org.apache.eagle.stream.application.TopologyExecutable; - -import java.util.Arrays; - -public class HiveJobRunningMonitoringTopology implements TopologyExecutable { - @Override - public void submit(String topology, Config config) { - StormExecutionEnvironment env = ExecutionEnvironments.getStorm(config); - String spoutName = "msgConsumer"; - int parallelism = env.getConfig().getInt("envContextConfig.parallelismConfig." + spoutName); - env.fromSpout(new HiveJobRunningSourcedStormSpoutProvider().getSpout(env.getConfig(), parallelism)) - .withOutputFields(4).nameAs(spoutName).groupBy(Arrays.asList(0)) - .flatMap(new JobConfigurationAdaptorExecutor()).groupBy(Arrays.asList(0)) - .flatMap(new HiveQueryParserExecutor()).groupBy(Arrays.asList(0)) - .flatMap(new HiveResourceSensitivityDataJoinExecutor()) - .alertWithConsumer("hiveAccessLogStream", "hiveAccessAlertByRunningJob"); - env.execute(); - - } -} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveJobRunningMonitoringMain.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveJobRunningMonitoringMain.java b/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveJobRunningMonitoringMain.java deleted file mode 100644 index 5b1bee6..0000000 --- a/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveJobRunningMonitoringMain.java +++ /dev/null @@ -1,43 +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.eagle.security.hive.jobrunning; - -import org.apache.eagle.datastream.ExecutionEnvironments; -import org.apache.eagle.datastream.storm.StormExecutionEnvironment; -import org.apache.eagle.security.hive.sensitivity.HiveResourceSensitivityDataJoinExecutor; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import java.util.Arrays; - -public class HiveJobRunningMonitoringMain { - private static final Logger LOG = LoggerFactory.getLogger(HiveJobRunningMonitoringMain.class); - - public static void main(String[] args) throws Exception{ - StormExecutionEnvironment env = ExecutionEnvironments.getStorm(args); - String spoutName = "msgConsumer"; - int parallelism = env.getConfig().getInt("envContextConfig.parallelismConfig." + spoutName); - env.fromSpout(new HiveJobRunningSourcedStormSpoutProvider().getSpout(env.getConfig(), parallelism)) - .withOutputFields(4).nameAs(spoutName).groupBy(Arrays.asList(0)) - .flatMap(new JobConfigurationAdaptorExecutor()).groupBy(Arrays.asList(0)) - .flatMap(new HiveQueryParserExecutor()).groupBy(Arrays.asList(0)) - .flatMap(new HiveResourceSensitivityDataJoinExecutor()) - .alertWithConsumer("hiveAccessLogStream", "hiveAccessAlertByRunningJob"); - env.execute(); - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveJobRunningSourcedStormSpoutProvider.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveJobRunningSourcedStormSpoutProvider.java b/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveJobRunningSourcedStormSpoutProvider.java index 729f519..f60d463 100644 --- a/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveJobRunningSourcedStormSpoutProvider.java +++ b/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveJobRunningSourcedStormSpoutProvider.java @@ -30,34 +30,34 @@ import org.slf4j.LoggerFactory; public class HiveJobRunningSourcedStormSpoutProvider { private static final Logger LOG = LoggerFactory.getLogger(HiveJobRunningSourcedStormSpoutProvider.class); - + public BaseRichSpout getSpout(Config config, int parallelism){ RunningJobEndpointConfig endPointConfig = new RunningJobEndpointConfig(); - String RMEndPoints = config.getString("dataSourceConfig.RMEndPoints"); + String RMEndPoints = config.getString("dataSourceConfig.RMEndPoints"); endPointConfig.RMBasePaths = RMEndPoints.split(","); - + String HSEndPoint = config.getString("dataSourceConfig.HSEndPoint"); endPointConfig.HSBasePath = HSEndPoint; - + ControlConfig controlConfig = new ControlConfig(); controlConfig.jobInfoEnabled = true; controlConfig.jobConfigEnabled = true; controlConfig.numTotalPartitions = parallelism <= 0 ? 1 : parallelism; - + boolean zkCleanupTimeSet = config.hasPath("dataSourceConfig.zkCleanupTimeInday"); //default set as two days controlConfig.zkCleanupTimeInday = zkCleanupTimeSet ? config.getInt("dataSourceConfig.zkCleanupTimeInday") : 2; - + boolean completedJobOutofDateTimeSet = config.hasPath("dataSourceConfig.completedJobOutofDateTimeInMin"); controlConfig.completedJobOutofDateTimeInMin = completedJobOutofDateTimeSet ? config.getInt("dataSourceConfig.completedJobOutofDateTimeInMin") : 120; - + boolean sizeOfJobConfigQueueSet = config.hasPath("dataSourceConfig.sizeOfJobConfigQueue"); controlConfig.sizeOfJobConfigQueue = sizeOfJobConfigQueueSet ? config.getInt("dataSourceConfig.sizeOfJobConfigQueue") : 10000; boolean sizeOfJobCompletedInfoQueue = config.hasPath("dataSourceConfig.sizeOfJobCompletedInfoQueue"); controlConfig.sizeOfJobCompletedInfoQueue = sizeOfJobCompletedInfoQueue ? config.getInt("dataSourceConfig.sizeOfJobCompletedInfoQueue") : 10000; - - //controlConfig.numTotalPartitions = parallelism == null ? 1 : parallelism; + + //controlConfig.numTotalPartitions = parallelism == null ? 1 : parallelism; ZKStateConfig zkStateConfig = new ZKStateConfig(); zkStateConfig.zkQuorum = config.getString("dataSourceConfig.zkQuorum"); zkStateConfig.zkRoot = config.getString("dataSourceConfig.zkRoot"); @@ -73,9 +73,8 @@ public class HiveJobRunningSourcedStormSpoutProvider { LOG.warn("failing find job id partitioner class " + config.getString("dataSourceConfig.partitionerCls")); //throw new IllegalStateException("jobId partitioner class does not exist " + config.getString("dataSourceConfig.partitionerCls")); controlConfig.partitionerCls = DefaultJobPartitionerImpl.class; - } - + JobRunningSpout spout = new JobRunningSpout(crawlConfig); return spout; } http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveQueryMonitoringAppProvider.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveQueryMonitoringAppProvider.java b/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveQueryMonitoringAppProvider.java new file mode 100644 index 0000000..e79b0eb --- /dev/null +++ b/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveQueryMonitoringAppProvider.java @@ -0,0 +1,33 @@ +/* + * + * * 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.eagle.security.hive.jobrunning; + +import org.apache.eagle.app.spi.AbstractApplicationProvider; + +/** + * Since 8/12/16. + */ +public class HiveQueryMonitoringAppProvider extends AbstractApplicationProvider<HiveQueryMonitoringApplication> { + @Override + public HiveQueryMonitoringApplication getApplication() { + return new HiveQueryMonitoringApplication(); + } +} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveQueryMonitoringApplication.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveQueryMonitoringApplication.java b/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveQueryMonitoringApplication.java new file mode 100644 index 0000000..f19c9a9 --- /dev/null +++ b/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveQueryMonitoringApplication.java @@ -0,0 +1,80 @@ +/* + * 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.eagle.security.hive.jobrunning; + +import backtype.storm.generated.StormTopology; +import backtype.storm.topology.BoltDeclarer; +import backtype.storm.topology.IRichSpout; +import backtype.storm.topology.TopologyBuilder; +import backtype.storm.tuple.Fields; +import com.typesafe.config.Config; +import com.typesafe.config.ConfigFactory; +import org.apache.eagle.app.StormApplication; +import org.apache.eagle.app.environment.impl.StormEnvironment; +import org.apache.eagle.app.sink.StormStreamSink; +import org.apache.eagle.security.hive.sensitivity.HiveResourceSensitivityDataJoinBolt; + +/** + * Since 8/11/16. + */ +public class HiveQueryMonitoringApplication extends StormApplication { + public final static String SPOUT_TASK_NUM = "topology.numOfSpoutTasks"; + public final static String FILTER_TASK_NUM = "topology.numOfFilterTasks"; + public final static String PARSER_TASK_NUM = "topology.numOfParserTasks"; + public final static String JOIN_TASK_NUM = "topology.numOfJoinTasks"; + public final static String SINK_TASK_NUM = "topology.numOfSinkTasks"; + + @Override + public StormTopology execute(Config config, StormEnvironment environment) { + TopologyBuilder builder = new TopologyBuilder(); + HiveJobRunningSourcedStormSpoutProvider provider = new HiveJobRunningSourcedStormSpoutProvider(); + IRichSpout spout = provider.getSpout(config, config.getInt(SPOUT_TASK_NUM)); + + + int numOfSpoutTasks = config.getInt(SPOUT_TASK_NUM); + int numOfFilterTasks = config.getInt(FILTER_TASK_NUM); + int numOfParserTasks = config.getInt(PARSER_TASK_NUM); + int numOfJoinTasks = config.getInt(JOIN_TASK_NUM); + int numOfSinkTasks = config.getInt(SINK_TASK_NUM); + + builder.setSpout("ingest", spout, numOfSpoutTasks); + JobFilterBolt bolt = new JobFilterBolt(); + BoltDeclarer boltDeclarer = builder.setBolt("filterBolt", bolt, numOfFilterTasks); + boltDeclarer.fieldsGrouping("ingest", new Fields("f1")); + + HiveQueryParserBolt parserBolt = new HiveQueryParserBolt(); + BoltDeclarer parserBoltDeclarer = builder.setBolt("parserBolt", parserBolt, numOfParserTasks); + parserBoltDeclarer.fieldsGrouping("filterBolt", new Fields("user")); + + HiveResourceSensitivityDataJoinBolt joinBolt = new HiveResourceSensitivityDataJoinBolt(config); + BoltDeclarer joinBoltDeclarer = builder.setBolt("joinBolt", joinBolt, numOfJoinTasks); + joinBoltDeclarer.fieldsGrouping("parserBolt", new Fields("user")); + + StormStreamSink sinkBolt = environment.getStreamSink("hive_query_stream",config); + BoltDeclarer kafkaBoltDeclarer = builder.setBolt("kafkaSink", sinkBolt, numOfSinkTasks); + kafkaBoltDeclarer.fieldsGrouping("joinBolt", new Fields("user")); + return builder.createTopology(); + } + + public static void main(String[] args){ + Config config = ConfigFactory.load(); + HiveQueryMonitoringApplication app = new HiveQueryMonitoringApplication(); + app.run(config); + } +} http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/15e1c833/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveQueryParserBolt.java ---------------------------------------------------------------------- diff --git a/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveQueryParserBolt.java b/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveQueryParserBolt.java new file mode 100644 index 0000000..c9be0e8 --- /dev/null +++ b/eagle-security/eagle-security-hive/src/main/java/org/apache/eagle/security/hive/jobrunning/HiveQueryParserBolt.java @@ -0,0 +1,156 @@ +/* + * 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.eagle.security.hive.jobrunning; + +import backtype.storm.task.OutputCollector; +import backtype.storm.task.TopologyContext; +import backtype.storm.topology.OutputFieldsDeclarer; +import backtype.storm.topology.base.BaseRichBolt; +import backtype.storm.tuple.Fields; +import backtype.storm.tuple.Tuple; +import org.apache.eagle.security.hive.ql.HiveQLParserContent; +import org.apache.eagle.security.hive.ql.Parser; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Arrays; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; +import java.util.TreeMap; + +/** + * parse hive query log + */ +public class HiveQueryParserBolt extends BaseRichBolt { + private static final Logger LOG = LoggerFactory.getLogger(HiveQueryParserBolt.class); + private OutputCollector collector; + + @Override + public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { + this.collector = collector; + } + + @Override + public void execute(Tuple input) { + /** + * hiveQueryLog includes the following key value pair + * "hive.current.database" -> <database name> + * "hive.query.string" -> <hive query statement> + * "mapreduce.job.user.name" -> <user name> + * TODO we need hive job start and end time + */ + String user = input.getString(0); + @SuppressWarnings("unchecked") + Map<String, Object> hiveQueryLog = (Map<String, Object>)input.getValue(1); + //if(LOG.isDebugEnabled()) LOG.debug("Receive hive query log: " + hiveQueryLog); + + String query = null; + String db = null; + String userName = null; + long timestamp = -1; + for (Entry<String, Object> entry : hiveQueryLog.entrySet()) { + switch (entry.getKey()) { + case "hive.query.string": + if (entry.getValue() != null) { + query = entry.getValue().toString(); + } + break; + case "hive.current.database": + if (entry.getValue() != null) { + db = entry.getValue().toString(); + } + break; + case "mapreduce.job.user.name": + if (entry.getValue() != null) { + userName = entry.getValue().toString(); + } + break; + case "mapreduce.job.cache.files.timestamps": + if (entry.getValue() != null) { + String timestampString = (String) entry.getValue(); + String[] timestampArray = timestampString.split("\\s*,\\s*"); + /* Get timestamp of start time. */ + timestamp = Long.parseLong(timestampArray[0]); + } + break; + } + } + + HiveQLParserContent parserContent = null; + Parser queryParser = new Parser(); + try { + parserContent = queryParser.run(query); + } catch (Exception ex) { + LOG.error("Failed running hive query parser.", ex); + //throw new IllegalStateException(ex); + } + if(parserContent == null) { + LOG.warn("Event ignored as it can't be correctly parsed, the query log is " + query); + return; + } + if(parserContent.getTableColumnMap().size() == 0) { + LOG.warn("Unsupported command for parsing " + query); + return; + } + /** + * Generate "resource" field: /db/table/column + * "resource" -> </db/table/column1,/db/table/column2,...> + */ + StringBuilder resources = new StringBuilder(); + String prefix = ","; + String connector = "/"; + for (Entry<String, Set<String>> entry : parserContent.getTableColumnMap().entrySet()) { + String table = entry.getKey(); + Set<String> colSet = entry.getValue(); + /** + * If colSet is empty, it means no column is accessed in the table. + * So column is not added to the event stream. + * Only /db/table + */ + if (colSet.isEmpty()) { + resources.append(connector).append(db).append(connector).append(table).append(prefix); + } else { + for (String col : colSet) { + resources.append(connector).append(db).append(connector).append(table); + if (col != null && col.length() > 0) { + resources.append(connector).append(col); + } + resources.append(prefix); + } + } + } + /* Remove the last prefix: "," */ + resources.setLength(resources.length() - 1); + + /* <event> has to be SortedMap. */ + Map<String, Object> event = new TreeMap<String, Object>(); + event.put("user", userName); + event.put("command", parserContent.getOperation()); + event.put("timestamp", timestamp); + event.put("resource", resources.toString()); + LOG.info("HiveQL Parser event stream. " + event); + + collector.emit(Arrays.asList(user, event)); + collector.ack(input); + } + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) { + declarer.declare(new Fields("user", "message")); + } +}
