You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spot.apache.org by ev...@apache.org on 2017/07/21 20:21:26 UTC
[27/50] [abbrv] incubator-spot git commit: Addind OA context and
configuration migration
Addind OA context and configuration migration
Project: http://git-wip-us.apache.org/repos/asf/incubator-spot/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spot/commit/401ffafe
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spot/tree/401ffafe
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spot/diff/401ffafe
Branch: refs/heads/master
Commit: 401ffafea3630a97579ab7e79ad30c2f28285c2b
Parents: b829177
Author: Moy Valdovinos <mo...@gmail.com>
Authored: Thu Jun 29 04:49:06 2017 -0500
Committer: LedaLima <ga...@intel.com>
Committed: Tue Jul 4 18:55:44 2017 -0500
----------------------------------------------------------------------
spot-setup/migration/import_ingest_summary.py | 60 -----
spot-setup/migration/import_proxy_timeline.py | 63 ------
spot-setup/migration/migrate_old_data.py | 83 +++++++
spot-setup/migration/migrate_old_dns_data.py | 231 +++++++++++++++++++
spot-setup/migration/migrate_old_dns_data.sh | 195 ----------------
spot-setup/migration/migrate_old_flow_data.py | 232 ++++++++++++++++++++
spot-setup/migration/migrate_old_flow_data.sh | 202 -----------------
spot-setup/migration/migrate_old_proxy_data.py | 222 +++++++++++++++++++
spot-setup/migration/migrate_old_proxy_data.sh | 169 --------------
spot-setup/migration/utilities.py | 89 ++++++++
10 files changed, 857 insertions(+), 689 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/401ffafe/spot-setup/migration/import_ingest_summary.py
----------------------------------------------------------------------
diff --git a/spot-setup/migration/import_ingest_summary.py b/spot-setup/migration/import_ingest_summary.py
deleted file mode 100755
index e8fb819..0000000
--- a/spot-setup/migration/import_ingest_summary.py
+++ /dev/null
@@ -1,60 +0,0 @@
-#!/bin/env python
-
-#
-# 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 os
-import sys
-import re
-import subprocess
-import pandas as pd
-import datetime
-
-filepath = sys.argv[1]
-staging_db_name = sys.argv[2]
-staging_table_name = sys.argv[3]
-dest_db_name = sys.argv[4]
-dest_table_name = sys.argv[5]
-
-df = pd.read_csv(filepath)
-
-s = df.iloc[:,0]
-l_dates = list(s.unique())
-l_dates = map(lambda x: x[0:10].strip(), l_dates)
-l_dates = filter(lambda x: re.match('\d{4}[-/]\d{2}[-/]\d{1}', x), l_dates)
-s_dates = set(l_dates)
-
-for date_str in s_dates:
- date_dt = datetime.datetime.strptime(date_str, '%Y-%m-%d')
- print 'Processing day: ', date_str, date_dt.year, date_dt.month, date_dt.day
-
- records = df[df['date'].str.contains(date_str)]
- filename = "ingest_summary_{0}{1}{2}.csv".format(date_dt.year, date_dt.month, date_dt.day)
- records.to_csv(filename, index=False)
-
- insert_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db_name, staging_table_name)
- load_to_table_cmd = "hive -e \"{0}\"".format(insert_cmd)
- print load_to_table_cmd
- subprocess.call(load_to_table_cmd, shell=True)
-
- insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT tdate, total FROM {5}.{6}".format(dest_db_name, dest_table_name, date_dt.year, date_dt.month, date_dt.day, staging_db_name, staging_table_name)
- load_to_table_cmd = "hive -e \"{0}\"".format(insert_cmd)
- print load_to_table_cmd
- subprocess.call(load_to_table_cmd, shell=True)
-
- os.remove(filename)
-
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/401ffafe/spot-setup/migration/import_proxy_timeline.py
----------------------------------------------------------------------
diff --git a/spot-setup/migration/import_proxy_timeline.py b/spot-setup/migration/import_proxy_timeline.py
deleted file mode 100755
index a503c7f..0000000
--- a/spot-setup/migration/import_proxy_timeline.py
+++ /dev/null
@@ -1,63 +0,0 @@
-#!/bin/env python
-
-#
-# 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 os
-import sys
-import subprocess
-import pandas as pd
-
-timeline_path = sys.argv[1]
-staging_db_name = sys.argv[2]
-staging_table_name = sys.argv[3]
-dest_db_name = sys.argv[4]
-dest_table_name = sys.argv[5]
-dt = sys.argv[6]
-
-dt_year, dt_month, dt_day = dt.split('-')
-
-print dt_year, dt_month, dt_day
-
-filename = os.path.splitext(os.path.basename(timeline_path))[0]
-hash_code = filename.split('-')[1]
-folder = os.path.dirname(timeline_path)
-extsearch_path = "{0}/es-{1}.csv".format(folder, hash_code)
-
-print filename, hash_code, folder, extsearch_path
-
-if os.path.isfile(extsearch_path):
-
- # Get FullURI from extended search file
- es_df = pd.read_csv(extsearch_path, sep='\t')
- fulluri = es_df.iloc[0]['fulluri']
- print fulluri
-
- # Load timeline to staging table
- insert_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(timeline_path, staging_db_name, staging_table_name)
- load_to_table_cmd = "hive -e \"{0}\"".format(insert_cmd)
- print load_to_table_cmd
- subprocess.call(load_to_table_cmd, shell=True)
-
- # Insert into new table from staging table and adding FullURI value
- insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT '{5}', tstart, tend, duration, clientip, respcode, '' FROM {6}.{7} where cast(tstart as timestamp) is not null;".format(dest_db_name, dest_table_name, dt_year, dt_month, dt_day, fulluri, staging_db_name, staging_table_name)
- load_to_table_cmd = "hive -e \"{0}\"".format(insert_cmd)
- print load_to_table_cmd
- subprocess.call(load_to_table_cmd, shell=True)
-
-else:
- print "Extended search file {0} doesn't exist. Timeline on file {1} can't be processed".format(extsearch_path, timeline_path)
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/401ffafe/spot-setup/migration/migrate_old_data.py
----------------------------------------------------------------------
diff --git a/spot-setup/migration/migrate_old_data.py b/spot-setup/migration/migrate_old_data.py
new file mode 100755
index 0000000..50ecf6c
--- /dev/null
+++ b/spot-setup/migration/migrate_old_data.py
@@ -0,0 +1,83 @@
+#!/bin/env python
+
+#
+# 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 os
+import sys
+import subprocess
+from utilities import util
+
+pipelines=sys.argv[1]
+old_oa_path=sys.argv[2]
+staging_db=sys.argv[3]
+hdfs_staging_path=sys.argv[4]
+dest_db = sys.argv[5]
+impala_daemon = sys.argv[6]
+
+
+# Execution example:
+#./migrate_old_proxy_data.py '/home/spotuser/incubator-spot_old/spot-oa' 'spot_migration' '/user/spotuser/spot_migration/' 'migrated' 'node01'
+
+def main():
+ log = util.get_logger('SPOT.MIGRATE')
+
+ cur_path = os.path.dirname(os.path.realpath(__file__))
+ new_spot_path = os.path.split(os.path.split(cur_path)[0])[0]
+ new_oa_path = '{0}/spot-oa'.format(new_spot_path)
+ log.info('New Spot OA path: {0}'.format(new_oa_path))
+ old_spot_path = os.path.split(old_oa_path)[0]
+
+
+ log.info("Copy ingest_conf.json file from old folder code to new Spot location")
+ util.execute_cmd('cp {0}/spot-ingest/ingest_conf.json {1}/spot-ingest/ingest_conf.json'.format(old_spot_path, new_spot_path),log)
+
+ log.info("Copy Spot OA configuration files to new Spot location")
+ util.execute_cmd('cp {0}/spot-oa/oa/components/iana/iana_config.json {1}/spot-oa/oa/components/iana/iana_config.json'.format(old_spot_path, new_spot_path),log)
+ util.execute_cmd('cp {0}/spot-oa/oa/components/reputation/reputation_config.json {1}/spot-oa/oa/components/reputation/reputation_config.json'.format(old_spot_path, new_spot_path),log)
+ util.execute_cmd('cp {0}/spot-oa/oa/components/nc/nc_config.json {1}/spot-oa/oa/components/nc/nc_config.json'.format(old_spot_path, new_spot_path),log)
+ util.execute_cmd('cp {0}/spot-oa/oa/components/data/engine.json {1}/spot-oa/oa/components/data/engine.json'.format(old_spot_path, new_spot_path),log)
+
+ log.info("Copy Spot OA context files to new Spot location")
+ util.execute_cmd('cp {0}/spot-oa/context/iploc.csv {1}/spot-oa/context/iploc.csv'.format(old_spot_path, new_spot_path),log)
+ util.execute_cmd('cp {0}/spot-oa/context/ipranges.csv {1}/spot-oa/context/ipranges.csv'.format(old_spot_path, new_spot_path),log)
+
+ log.info("Install browserify and uglifyjs using npm")
+ os.chdir('{0}/ui'.format(new_oa_path))
+ # util.execute_cmd('sudo -H -E npm install -g browserify uglifyjs',log)
+
+ log.info("Install requirements using pip")
+ os.chdir('{0}'.format(new_oa_path))
+ # util.execute_cmd('sudo pip install -r requirements.txt',log)
+
+
+
+ l_pipelines = pipelines.split(',')
+ valid_pipelines = ('flow', 'dns', 'proxy')
+ if len(l_pipelines) > 0:
+ for pipe in l_pipelines:
+ if pipe in valid_pipelines:
+ os.chdir('{0}/spot-setup/migration'.format(new_spot_path))
+ util.execute_cmd("python migrate_old_{0}_data.py '{1}' '{2}' '{3}' '{4}' '{5}' ".format(pipe, old_oa_path, staging_db, hdfs_staging_path, dest_db, impala_daemon),log)
+ else:
+ log.error("Pipeline {0} is not valid. Valid pipelines are flow, dns or proxy".format(pipe))
+ else:
+ log.error("Pipelines argument must be separated by commas")
+
+
+if __name__=='__main__':
+ main()
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/401ffafe/spot-setup/migration/migrate_old_dns_data.py
----------------------------------------------------------------------
diff --git a/spot-setup/migration/migrate_old_dns_data.py b/spot-setup/migration/migrate_old_dns_data.py
new file mode 100755
index 0000000..34b792a
--- /dev/null
+++ b/spot-setup/migration/migrate_old_dns_data.py
@@ -0,0 +1,231 @@
+#!/bin/env python
+
+#
+# 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 os
+import sys
+import subprocess
+import fnmatch
+import re
+import pandas as pd
+import datetime
+from utilities import util
+
+old_oa_path=sys.argv[1]
+staging_db=sys.argv[2]
+hdfs_staging_path=sys.argv[3]
+dest_db = sys.argv[4]
+impala_daemon = sys.argv[5]
+
+
+# Execution example:
+#./migrate_old_dns_data.py '/home/spotuser/incubator-spot_old/spot-oa' 'spot_migration' '/user/spotuser/spot_migration/' 'migrated' 'node01'
+
+def main():
+ log = util.get_logger('SPOT.MIGRATE.DNS')
+
+ cur_path = os.path.dirname(os.path.realpath(__file__))
+ new_spot_path = os.path.split(os.path.split(cur_path)[0])[0]
+ new_oa_path = '{0}/spot-oa'.format(new_spot_path)
+ log.info('New Spot OA path: {0}'.format(new_oa_path))
+ old_spot_path = os.path.split(old_oa_path)[0]
+
+
+ log.info("Creating HDFS paths for Impala tables")
+ util.create_hdfs_folder('{0}/dns/scores'.format(hdfs_staging_path),log)
+ util.create_hdfs_folder('{0}/dns/dendro'.format(hdfs_staging_path),log)
+ util.create_hdfs_folder('{0}/dns/edge'.format(hdfs_staging_path),log)
+ util.create_hdfs_folder('{0}/dns/summary'.format(hdfs_staging_path),log)
+ util.create_hdfs_folder('{0}/dns/storyboard'.format(hdfs_staging_path),log)
+ util.create_hdfs_folder('{0}/dns/threat_dendro'.format(hdfs_staging_path),log)
+ util.execute_cmd('hdfs dfs -setfacl -R -m user:impala:rwx {0}'.format(hdfs_staging_path),log)
+
+
+ log.info("Creating Staging tables in Impala")
+ util.execute_cmd('impala-shell -i {0} --var=hpath={1} --var=dbname={2} -c -f create_dns_migration_tables.hql'.format(impala_daemon, hdfs_staging_path, staging_db),log)
+
+
+ ## dns Ingest Summary
+ log.info('Processing Dns Ingest Summary')
+ ing_sum_path='{0}/data/dns/ingest_summary/'.format(old_oa_path)
+ pattern='is_??????.csv'
+ staging_table_name = 'dns_ingest_summary_tmp'
+ dest_table_name = 'dns_ingest_summary'
+
+ if os.path.exists(ing_sum_path):
+ for file in fnmatch.filter(os.listdir(ing_sum_path), pattern):
+ log.info('Processing file: {0}'.format(file))
+
+ filepath='{0}{1}'.format(ing_sum_path, file)
+ df = pd.read_csv(filepath)
+
+ s = df.iloc[:,0]
+ l_dates = list(s.unique())
+ l_dates = map(lambda x: x[0:10].strip(), l_dates)
+ l_dates = filter(lambda x: re.match('\d{4}[-/]\d{2}[-/]\d{1}', x), l_dates)
+ s_dates = set(l_dates)
+
+ for date_str in s_dates:
+ dt = datetime.datetime.strptime(date_str, '%Y-%m-%d')
+ log.info('Processing day: {0} {1} {2} {3}'.format(date_str, dt.year, dt.month, dt.day))
+
+ records = df[df['date'].str.contains(date_str)]
+ filename = "ingest_summary_{0}{1}{2}.csv".format(dt.year, dt.month, dt.day)
+ records.to_csv(filename, index=False)
+
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT tdate, total FROM {5}.{6}".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+ os.remove(filename)
+
+
+ ## Iterating days
+ days_path='{0}/data/dns/'.format(old_oa_path)
+
+ if os.path.exists(days_path):
+ for day_folder in fnmatch.filter(os.listdir(days_path), '2*'):
+
+ print day_folder
+ dt = datetime.datetime.strptime(day_folder, '%Y%m%d')
+ log.info('Processing day: {0} {1} {2} {3}'.format(day_folder, dt.year, dt.month, dt.day))
+ full_day_path = '{0}{1}'.format(days_path,day_folder)
+
+
+ ## dns Scores and dns Threat Investigation
+ filename = '{0}/dns_scores.csv'.format(full_day_path)
+ if os.path.isfile(filename):
+
+ log.info("Processing Dns Scores")
+ staging_table_name = 'dns_scores_tmp'
+ dest_table_name = 'dns_scores'
+
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT frame_time, unix_tstamp, frame_len, ip_dst, dns_qry_name, dns_qry_class, dns_qry_type, dns_qry_rcode, ml_score, tld, query_rep, hh, dns_qry_class_name, dns_qry_type_name, dns_qry_rcode_name, network_context FROM {5}.{6}".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+ log.info("Processing dns Threat Investigation")
+ staging_table_name = 'dns_scores_tmp'
+ dest_table_name = 'dns_threat_investigation'
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT unix_tstamp, ip_dst, dns_qry_name, ip_sev, dns_sev FROM {5}.{6} WHERE ip_sev > 0 or dns_sev > 0;".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+
+ # dns Dendro
+ log.info("Processing Dns Dendro")
+ staging_table_name = 'dns_dendro_tmp'
+ dest_table_name = 'dns_dendro'
+ pattern = 'dendro*.csv'
+ dendro_files = fnmatch.filter(os.listdir(full_day_path), pattern)
+ filename = '{0}/{1}'.format(full_day_path, pattern)
+
+ if len(dendro_files) > 0:
+
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT unix_timestamp('{5}', 'yyyyMMMdd'), dns_a, dns_qry_name, ip_dst FROM {6}.{7};".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, day_folder, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+
+ ## dns Edge
+ log.info("Processing Dns Edge")
+ staging_table_name = 'dns_edge_tmp'
+ dest_table_name = 'dns_edge'
+ pattern = 'edge*.csv'
+ edge_files = fnmatch.filter(os.listdir(full_day_path), pattern)
+ filename = '{0}/{1}'.format(full_day_path, pattern)
+
+ if len(edge_files) > 0:
+
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT unix_timestamp(regexp_replace(frame_time, '\"', ''), 'MMMMM dd yyyy H:mm:ss.SSS z'), frame_len, ip_dst, ip_src, dns_qry_name, dns_qry_class, dns_qry_type, dns_qry_rcode, dns_a, 0, '', '', '', '' FROM {5}.{6};".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+
+ ##dns_storyboard
+ log.info("Processing Dns Storyboard")
+ staging_table_name = 'dns_storyboard_tmp'
+ dest_table_name = 'dns_storyboard'
+ filename = '{0}/threats.csv'.format(full_day_path)
+
+ if os.path.isfile(filename):
+
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT ip_threat, dns_threat, title, text FROM {5}.{6};".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+
+ # dns Threat Dendro
+ log.info("Processing Dns Threat Dendro")
+ staging_table_name = 'dns_threat_dendro_tmp'
+ dest_table_name = 'dns_threat_dendro'
+ pattern = 'threat-dendro*.csv'
+ threat_dendro_files = fnmatch.filter(os.listdir(full_day_path), pattern)
+ filename = '{0}/{1}'.format(full_day_path, pattern)
+
+ for file in threat_dendro_files:
+
+ ip = re.findall("threat-dendro-(\S+).csv", file)[0]
+ log.info("Processing File: {0} with IP:{1}".format(file, ip))
+ filename = '{0}/{1}'.format(full_day_path, file)
+
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT '{5}', total, dns_qry_name, ip_dst FROM {6}.{7};".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, ip, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+
+
+ log.info("Dropping staging tables")
+ util.execute_cmd('impala-shell -i {0} --var=dbname={1} -c -f drop_dns_migration_tables.hql'.format(impala_daemon, staging_db),log)
+
+ log.info("Removing staging tables' path in HDFS")
+ util.execute_cmd('hadoop fs -rm -r {0}/dns/'.format(hdfs_staging_path),log)
+
+ log.info("Moving CSV data to backup folder")
+ util.execute_cmd('mkdir {0}/data/backup/'.format(old_oa_path),log)
+ util.execute_cmd('mv {0}/data/dns/ {0}/data/backup/'.format(old_oa_path),log)
+
+ log.info("Invalidating metadata in Impala to refresh tables content")
+ util.execute_cmd('impala-shell -i {0} -q "INVALIDATE METADATA;"'.format(impala_daemon),log)
+
+
+
+ log.info("Copying advanced mode notebooks to each existing day in new Spot location")
+ for folder in os.listdir('{0}/ipynb/dns/'.format(new_oa_path)):
+ util.execute_cmd('cp {0}/oa/dns/ipynb_templates/Advanced_Mode_master.ipynb {1}/ipynb/dns/{2}/Advanced_Mode.ipynb'.format(old_oa_path, new_oa_path, folder),log)
+
+ log.info("Copying threat investigation ipynb template to each existing day in new Spot location")
+ for folder in os.listdir('{0}/ipynb/dns/'.format(new_oa_path)):
+ util.execute_cmd('cp {0}/oa/dns/ipynb_templates/Threat_Investigation_master.ipynb {1}/ipynb/dns/{2}/Threat_Investigation.ipynb'.format(old_oa_path, new_oa_path, folder),log)
+
+
+
+if __name__=='__main__':
+ main()
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/401ffafe/spot-setup/migration/migrate_old_dns_data.sh
----------------------------------------------------------------------
diff --git a/spot-setup/migration/migrate_old_dns_data.sh b/spot-setup/migration/migrate_old_dns_data.sh
deleted file mode 100755
index 96341e8..0000000
--- a/spot-setup/migration/migrate_old_dns_data.sh
+++ /dev/null
@@ -1,195 +0,0 @@
-#!/bin/bash
-
-#
-# 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.
-#
-
-OLD_OA_PATH=$1
-NEW_OA_PATH=$2
-STAGING_DB=$3
-HDFS_STAGING_PATH=$4
-DEST_DB=$5
-IMPALA_DEM=$6
-
-# Execution example:
-#./migrate_old_dns_data.sh '/home/spotuser/incubator-spot_old/spot-oa' '/home/spotuser/incubator-spot_new/spot-oa/' 'spot_migration' '/user/spotuser/spot_migration/' 'migrated' 'node01'
-
-hadoop fs -mkdir $HDFS_STAGING_PATH
-hadoop fs -mkdir $HDFS_STAGING_PATH/dns/
-hadoop fs -mkdir $HDFS_STAGING_PATH/dns/scores/
-hadoop fs -mkdir $HDFS_STAGING_PATH/dns/dendro/
-hadoop fs -mkdir $HDFS_STAGING_PATH/dns/edge/
-hadoop fs -mkdir $HDFS_STAGING_PATH/dns/summary/
-hadoop fs -mkdir $HDFS_STAGING_PATH/dns/storyboard/
-hadoop fs -mkdir $HDFS_STAGING_PATH/dns/threat_dendro/
-hdfs dfs -setfacl -R -m user:impala:rwx $HDFS_STAGING_PATH
-
-
-#Creating Staging tables in Impala
-impala-shell -i ${IMPALA_DEM} --var=hpath=${HDFS_STAGING_PATH} --var=dbname=${STAGING_DB} -c -f create_dns_migration_tables.hql
-
-
-## DNS Ingest Summary
-echo "Processing DNS Ingest Summary"
-
-ing_sum_path=$OLD_OA_PATH/data/dns/ingest_summary/is_??????.csv
-
-for file in $ing_sum_path
-do
- echo $file
- ./import_ingest_summary.py "${file}" "${STAGING_DB}" 'dns_ingest_summary_tmp' "${DEST_DB}" 'dns_ingest_summary'
-done
-
-
-DAYS=$OLD_OA_PATH/data/dns/2*
-
-for dir in $DAYS
-do
- day="$(basename $dir)"
- echo "Processing day $day ..."
- y=${day:0:4}
- m=$(expr ${day:4:2} + 0)
- d=$(expr ${day:6:2} + 0)
- echo $y $m $d $d2
- echo $dir
-
-
- ## dns Scores and dns_threat_investigation
- echo "Processing dns Scores"
- if [ -f $dir/dns_scores.csv ]
- then
- command="LOAD DATA LOCAL INPATH '$dir/dns_scores.csv' OVERWRITE INTO TABLE $STAGING_DB.dns_scores_tmp;"
- echo $command
- hive -e "$command"
-
- command="INSERT INTO $DEST_DB.dns_scores PARTITION (y=$y, m=$m, d=$d)
-select frame_time, unix_tstamp, frame_len, ip_dst, dns_qry_name, dns_qry_class, dns_qry_type, dns_qry_rcode, ml_score, tld, query_rep, hh, dns_qry_class_name, dns_qry_type_name, dns_qry_rcode_name, network_context
-from $STAGING_DB.dns_scores_tmp;"
- echo $command
- hive -e "$command"
-
- echo "Processing dns Threat Investigation"
- command="INSERT INTO $DEST_DB.dns_threat_investigation PARTITION (y=$y, m=$m, d=$d)
-select unix_tstamp, ip_dst, dns_qry_name, ip_sev, dns_sev
-from $STAGING_DB.dns_scores_tmp
-where ip_sev > 0 or dns_sev > 0;"
- echo $command
- hive -e "$command"
-
- fi
-
- ## dns Dendro
- echo "Processing dns Dendro"
- dendro_files=`ls $dir/dendro*.csv`
-
- if [ ! -z "$dendro_files" ]
- then
-
- command="LOAD DATA LOCAL INPATH '$dir/dendro*.csv' OVERWRITE INTO TABLE $STAGING_DB.dns_dendro_tmp;"
- echo $command
- hive -e "$command"
-
- command="INSERT INTO $DEST_DB.dns_dendro PARTITION (y=$y, m=$m, d=$d)
-select unix_timestamp('$y${day:4:2}${day:6:2}', 'yyyyMMMdd'), dns_a, dns_qry_name, ip_dst
-from $STAGING_DB.dns_dendro_tmp;"
- echo $command
- hive -e "$command"
-
- fi
-
- ## dns Edge
- echo "Processing dns Edge"
- edge_files=`ls $dir/edge*.csv`
-
- if [ ! -z "$edge_files" ]
- then
-
- command="LOAD DATA LOCAL INPATH '$dir/edge*.csv' OVERWRITE INTO TABLE $STAGING_DB.dns_edge_tmp;"
- echo $command
- hive -e "$command"
-
- command="INSERT INTO $DEST_DB.dns_edge PARTITION (y=$y, m=$m, d=$d)
-select unix_timestamp(regexp_replace(frame_time, '\"', ''), 'MMMMM dd yyyy H:mm:ss.SSS z'), frame_len, ip_dst, ip_src, dns_qry_name, dns_qry_class, dns_qry_type, dns_qry_rcode, dns_a, 0, '', '', '', ''
-from $STAGING_DB.dns_edge_tmp;"
- echo $command
- hive -e "$command"
-
- fi
-
- ##dns_storyboard
- echo "Processing dns Storyboard"
- if [ -f $dir/threats.csv ]
- then
-
- command="LOAD DATA LOCAL INPATH '$dir/threats.csv' OVERWRITE INTO TABLE $STAGING_DB.dns_storyboard_tmp;"
- echo $command
- hive -e "$command"
-
- command="INSERT INTO $DEST_DB.dns_storyboard PARTITION (y=$y, m=$m, d=$d)
-select ip_threat, dns_threat, title, text from $STAGING_DB.dns_storyboard_tmp;"
- echo $command
- hive -e "$command"
-
- fi
-
- ##dns_threat_dendro
- echo "Processing dns Threat Dendro"
- threat_dendro_files=`ls $dir/threat-dendro*.csv`
-
- if [ ! -z "$threat_dendro_files" ]
- then
- for file in $threat_dendro_files
- do
- filename="$(basename $file)"
- ip="${filename%.csv}"
- ip="${ip#threat-dendro-}"
- echo $filename $ip
-
- command="LOAD DATA LOCAL INPATH '$file' OVERWRITE INTO TABLE $STAGING_DB.dns_threat_dendro_tmp;"
- echo $command
- hive -e "$command"
-
- command="INSERT INTO $DEST_DB.dns_threat_dendro PARTITION (y=$y, m=$m, d=$d)
-select '$ip', total, dns_qry_name, ip_dst
-from $STAGING_DB.dns_threat_dendro_tmp;"
- echo $command
- hive -e "$command"
-
- done
- fi
-done
-
-
-## Copying advanced mode notebooks to new paths in Spot 1.0
-for new_dir in $NEW_OA_PATH/ipynb/dns/*
-do
- cp $OLD_OA_PATH/oa/dns/ipynb_templates/Advanced_Mode_master.ipynb $new_dir/Advanced_Mode.ipynb
-done
-
-# Dropping staging tables
-impala-shell -i ${IMPALA_DEM} --var=dbname=${STAGING_DB} -c -f drop_dns_migration_tables.hql
-
-# Removing staging tables' path in HDFS
-hadoop fs -rm -r $HDFS_STAGING_PATH/dns/
-
-# Moving CSV data to backup folder
-mkdir $OLD_OA_PATH/data/backup/
-mv $OLD_OA_PATH/data/dns/ $OLD_OA_PATH/data/backup/
-
-# Invalidating metadata in Impala to refresh tables content
-impala-shell -i ${IMPALA_DEM} -q "INVALIDATE METADATA;"
-
-
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/401ffafe/spot-setup/migration/migrate_old_flow_data.py
----------------------------------------------------------------------
diff --git a/spot-setup/migration/migrate_old_flow_data.py b/spot-setup/migration/migrate_old_flow_data.py
new file mode 100755
index 0000000..eb9253f
--- /dev/null
+++ b/spot-setup/migration/migrate_old_flow_data.py
@@ -0,0 +1,232 @@
+#!/bin/env python
+
+#
+# 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 os
+import sys
+import subprocess
+import fnmatch
+import re
+import pandas as pd
+import datetime
+from utilities import util
+
+old_oa_path=sys.argv[1]
+staging_db=sys.argv[2]
+hdfs_staging_path=sys.argv[3]
+dest_db = sys.argv[4]
+impala_daemon = sys.argv[5]
+
+
+# Execution example:
+#./migrate_old_flow_data.py '/home/spotuser/incubator-spot_old/spot-oa' 'spot_migration' '/user/spotuser/spot_migration/' 'migrated' 'node01'
+
+def main():
+ log = util.get_logger('SPOT.MIGRATE.FLOW')
+
+ cur_path = os.path.dirname(os.path.realpath(__file__))
+ new_spot_path = os.path.split(os.path.split(cur_path)[0])[0]
+ new_oa_path = '{0}/spot-oa'.format(new_spot_path)
+ log.info('New Spot OA path: {0}'.format(new_oa_path))
+ old_spot_path = os.path.split(old_oa_path)[0]
+
+
+ log.info("Creating HDFS paths for Impala tables")
+ util.create_hdfs_folder('{0}/flow/scores'.format(hdfs_staging_path),log)
+ util.create_hdfs_folder('{0}/flow/chords'.format(hdfs_staging_path),log)
+ util.create_hdfs_folder('{0}/flow/edge'.format(hdfs_staging_path),log)
+ util.create_hdfs_folder('{0}/flow/summary'.format(hdfs_staging_path),log)
+ util.create_hdfs_folder('{0}/flow/storyboard'.format(hdfs_staging_path),log)
+ util.create_hdfs_folder('{0}/flow/threat_investigation'.format(hdfs_staging_path),log)
+ util.create_hdfs_folder('{0}/flow/timeline'.format(hdfs_staging_path),log)
+ util.execute_cmd('hdfs dfs -setfacl -R -m user:impala:rwx {0}'.format(hdfs_staging_path),log)
+
+
+ log.info("Creating Staging tables in Impala")
+ util.execute_cmd('impala-shell -i {0} --var=hpath={1} --var=dbname={2} -c -f create_flow_migration_tables.hql'.format(impala_daemon, hdfs_staging_path, staging_db),log)
+
+
+ ## Flow Ingest Summary
+ log.info('Processing Flow Ingest Summary')
+ ing_sum_path='{0}/data/flow/ingest_summary/'.format(old_oa_path)
+ pattern='is_??????.csv'
+ staging_table_name = 'flow_ingest_summary_tmp'
+ dest_table_name = 'flow_ingest_summary'
+
+ if os.path.exists(ing_sum_path):
+ for file in fnmatch.filter(os.listdir(ing_sum_path), pattern):
+ log.info('Processing file: {0}'.format(file))
+
+ filepath='{0}{1}'.format(ing_sum_path, file)
+ df = pd.read_csv(filepath)
+
+ s = df.iloc[:,0]
+ l_dates = list(s.unique())
+ l_dates = map(lambda x: x[0:10].strip(), l_dates)
+ l_dates = filter(lambda x: re.match('\d{4}[-/]\d{2}[-/]\d{1}', x), l_dates)
+ s_dates = set(l_dates)
+
+ for date_str in s_dates:
+ dt = datetime.datetime.strptime(date_str, '%Y-%m-%d')
+ log.info('Processing day: {0} {1} {2} {3}'.format(date_str, dt.year, dt.month, dt.day))
+
+ records = df[df['date'].str.contains(date_str)]
+ filename = "ingest_summary_{0}{1}{2}.csv".format(dt.year, dt.month, dt.day)
+ records.to_csv(filename, index=False)
+
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT tdate, total FROM {5}.{6}".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+ os.remove(filename)
+
+
+ ## Iterating days
+ days_path='{0}/data/flow/'.format(old_oa_path)
+
+ if os.path.exists(days_path):
+ for day_folder in fnmatch.filter(os.listdir(days_path), '2*'):
+
+ print day_folder
+ dt = datetime.datetime.strptime(day_folder, '%Y%m%d')
+ log.info('Processing day: {0} {1} {2} {3}'.format(day_folder, dt.year, dt.month, dt.day))
+ full_day_path = '{0}{1}'.format(days_path,day_folder)
+
+
+ ## Flow Scores and Flow Threat Investigation
+ filename = '{0}/flow_scores.csv'.format(full_day_path)
+ if os.path.isfile(filename):
+
+ log.info("Processing Flow Scores")
+ staging_table_name = 'flow_scores_tmp'
+ dest_table_name = 'flow_scores'
+
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT tstart,srcip,dstip,sport,dport,proto,ipkt,ibyt,opkt,obyt,score,rank,srcIpInternal,destIpInternal,srcGeo,dstGeo,srcDomain,dstDomain,srcIP_rep,dstIP_rep FROM {5}.{6}".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+ log.info("Processing Flow Threat Investigation")
+ staging_table_name = 'flow_scores_tmp'
+ dest_table_name = 'flow_threat_investigation'
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT tstart,srcip,dstip,sport,dport,sev FROM {5}.{6} WHERE sev > 0;".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+
+ # Flow Chords
+ log.info("Processing Flow Chords")
+ staging_table_name = 'flow_chords_tmp'
+ dest_table_name = 'flow_chords'
+
+ for file in fnmatch.filter(os.listdir(full_day_path), 'chord*.tsv'):
+
+ ip = re.findall("chord-(\S+).tsv", file)[0]
+ ip = ip.replace('_', '.')
+ log.info("Processing File: {0} with IP:{1}".format(file, ip))
+ filename = '{0}/{1}'.format(full_day_path, file)
+
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT '{5}', srcip, dstip, ibyt, ipkt FROM {6}.{7};".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, ip, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+
+ ## Flow Edge
+ log.info("Processing Flow Edge")
+ staging_table_name = 'flow_edge_tmp'
+ dest_table_name = 'flow_edge'
+ pattern = 'edge*.tsv'
+ edge_files = fnmatch.filter(os.listdir(full_day_path), pattern)
+ filename = '{0}/{1}'.format(full_day_path, pattern)
+
+ if len(edge_files) > 0:
+
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT tstart, srcip, dstip, sport, dport, proto, flags, tos, ibyt, ipkt, input, output, rip, obyt, opkt, 0, 0 FROM {5}.{6} WHERE srcip is not NULL;".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+
+ ##flow_storyboard
+ log.info("Processing Flow Storyboard")
+ staging_table_name = 'flow_storyboard_tmp'
+ dest_table_name = 'flow_storyboard'
+ filename = '{0}/threats.csv'.format(full_day_path)
+
+ if os.path.isfile(filename):
+
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT ip_threat, title, text FROM {5}.{6};".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+
+ ##flow_timeline
+ log.info("Processing Flow Timeline")
+ staging_table_name = 'flow_timeline_tmp'
+ dest_table_name = 'flow_timeline'
+
+ for file in fnmatch.filter(os.listdir(full_day_path), 'sbdet*.tsv'):
+
+ ip = re.findall("sbdet-(\S+).tsv", file)[0]
+ log.info("Processing File: {0} with IP:{1}".format(file, ip))
+ filename = '{0}/{1}'.format(full_day_path, file)
+
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT '{5}', tstart, tend, srcip, dstip, proto, sport, dport, ipkt, ibyt FROM {6}.{7};".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, ip, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+
+
+ log.info("Dropping staging tables")
+ util.execute_cmd('impala-shell -i {0} --var=dbname={1} -c -f drop_flow_migration_tables.hql'.format(impala_daemon, staging_db),log)
+
+ log.info("Removing staging tables' path in HDFS")
+ util.execute_cmd('hadoop fs -rm -r {0}/flow/'.format(hdfs_staging_path),log)
+
+ log.info("Moving CSV data to backup folder")
+ util.execute_cmd('mkdir {0}/data/backup/'.format(old_oa_path),log)
+ util.execute_cmd('mv {0}/data/flow/ {0}/data/backup/'.format(old_oa_path),log)
+
+ log.info("Invalidating metadata in Impala to refresh tables content")
+ util.execute_cmd('impala-shell -i {0} -q "INVALIDATE METADATA;"'.format(impala_daemon),log)
+
+
+
+ log.info("Copying advanced mode notebooks to each existing day in new Spot location")
+ for folder in os.listdir('{0}/ipynb/flow/'.format(new_oa_path)):
+ util.execute_cmd('cp {0}/oa/flow/ipynb_templates/Advanced_Mode_master.ipynb {1}/ipynb/flow/{2}/Advanced_Mode.ipynb'.format(old_oa_path, new_oa_path, folder),log)
+
+ log.info("Copying threat investigation ipynb template to each existing day in new Spot location")
+ for folder in os.listdir('{0}/ipynb/flow/'.format(new_oa_path)):
+ util.execute_cmd('cp {0}/oa/flow/ipynb_templates/Threat_Investigation_master.ipynb {1}/ipynb/flow/{2}/Threat_Investigation.ipynb'.format(old_oa_path, new_oa_path, folder),log)
+
+
+
+
+if __name__=='__main__':
+ main()
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/401ffafe/spot-setup/migration/migrate_old_flow_data.sh
----------------------------------------------------------------------
diff --git a/spot-setup/migration/migrate_old_flow_data.sh b/spot-setup/migration/migrate_old_flow_data.sh
deleted file mode 100755
index 9b51987..0000000
--- a/spot-setup/migration/migrate_old_flow_data.sh
+++ /dev/null
@@ -1,202 +0,0 @@
-#!/bin/bash
-
-#
-# 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.
-#
-
-OLD_OA_PATH=$1
-NEW_OA_PATH=$2
-STAGING_DB=$3
-HDFS_STAGING_PATH=$4
-DEST_DB=$5
-IMPALA_DEM=$6
-
-# Execution example:
-#./migrate_old_flow_data.sh '/home/spotuser/incubator-spot_old/spot-oa' '/home/spotuser/incubator-spot_new/spot-oa/' 'spot_migration' '/user/spotuser/spot_migration/' 'migrated' 'node01'
-
-hadoop fs -mkdir $HDFS_STAGING_PATH
-hadoop fs -mkdir $HDFS_STAGING_PATH/flow/
-hadoop fs -mkdir $HDFS_STAGING_PATH/flow/scores/
-hadoop fs -mkdir $HDFS_STAGING_PATH/flow/chords/
-hadoop fs -mkdir $HDFS_STAGING_PATH/flow/edge/
-hadoop fs -mkdir $HDFS_STAGING_PATH/flow/summary/
-hadoop fs -mkdir $HDFS_STAGING_PATH/flow/storyboard/
-hadoop fs -mkdir $HDFS_STAGING_PATH/flow/threat_investigation/
-hadoop fs -mkdir $HDFS_STAGING_PATH/flow/timeline/
-hdfs dfs -setfacl -R -m user:impala:rwx $HDFS_STAGING_PATH
-
-
-#Creating Staging tables in Impala
-impala-shell -i ${IMPALA_DEM} --var=hpath=${HDFS_STAGING_PATH} --var=dbname=${STAGING_DB} -c -f create_flow_migration_tables.hql
-
-
-## Flow Ingest Summary
-echo "Processing Flow Ingest Summary"
-
-ing_sum_path=$OLD_OA_PATH/data/flow/ingest_summary/is_??????.csv
-
-for file in $ing_sum_path
-do
- echo $file
- ./import_ingest_summary.py "${file}" "${STAGING_DB}" 'flow_ingest_summary_tmp' "${DEST_DB}" 'flow_ingest_summary'
-done
-
-
-DAYS=$OLD_OA_PATH/data/flow/2*
-
-for dir in $DAYS
-do
- day="$(basename $dir)"
- echo "Processing day $day ..."
- y=${day:0:4}
- m=$(expr ${day:4:2} + 0)
- d=$(expr ${day:6:2} + 0)
- echo $y $m $d $d2
- echo $dir
-
-
- ## Flow Scores and Flow Threat Investigation
- echo "Processing Flow Scores"
- if [ -f $dir/flow_scores.csv ]
- then
- command="LOAD DATA LOCAL INPATH '$dir/flow_scores.csv' OVERWRITE INTO TABLE $STAGING_DB.flow_scores_tmp;"
- echo $command
- hive -e "$command"
-
- command="INSERT INTO $DEST_DB.flow_scores PARTITION (y=$y, m=$m, d=$d)
-select tstart,srcip,dstip,sport,dport,proto,ipkt,ibyt,opkt,obyt,score,rank,srcIpInternal,destIpInternal,srcGeo,dstGeo,srcDomain,dstDomain,srcIP_rep,dstIP_rep
-from $STAGING_DB.flow_scores_tmp;"
- echo $command
- hive -e "$command"
-
- echo "Processing Flow Threat Investigation"
- command="INSERT INTO $DEST_DB.flow_threat_investigation PARTITION (y=$y, m=$m, d=$d)
-select tstart,srcip,dstip,sport,dport,sev
-from $STAGING_DB.flow_scores_tmp
-where sev > 0;"
- echo $command
- hive -e "$command"
-
- fi
-
- ## Flow Chords
- echo "Processing Flow Chords"
- chord_files=`ls $dir/chord*.tsv`
-
- if [ ! -z "$chord_files" ]
- then
- for file in $chord_files
- do
- filename="$(basename $file)"
- ip="${filename%.tsv}"
- ip="${ip#chord-}"
- ip="${ip//_/.}"
- echo $filename $ip
-
- command="LOAD DATA LOCAL INPATH '$file' OVERWRITE INTO TABLE $STAGING_DB.flow_chords_tmp;"
- echo $command
- hive -e "$command"
-
- command="INSERT INTO $DEST_DB.flow_chords PARTITION (y=$y, m=$m, d=$d)
- select '$ip', srcip, dstip, ibyt, ipkt from $STAGING_DB.flow_chords_tmp;"
- echo $command
- hive -e "$command"
-
- done
- fi
-
- ## Flow Edge
- echo "Processing Flow Edge"
- edge_files=`ls $dir/edge*.tsv`
-
- if [ ! -z "$edge_files" ]
- then
-
- command="LOAD DATA LOCAL INPATH '$dir/edge*.tsv' OVERWRITE INTO TABLE $STAGING_DB.flow_edge_tmp;"
- echo $command
- hive -e "$command"
-
- command="INSERT INTO $DEST_DB.flow_edge PARTITION (y=$y, m=$m, d=$d)
-select tstart, srcip, dstip, sport, dport, proto, flags, tos, ibyt, ipkt, input, output, rip, obyt, opkt, 0, 0
-from $STAGING_DB.flow_edge_tmp
-where srcip is not NULL;"
- echo $command
- hive -e "$command"
-
- fi
-
- ##flow_storyboard
- echo "Processing Flow Storyboard"
- if [ -f $dir/threats.csv ]
- then
-
- command="LOAD DATA LOCAL INPATH '$dir/threats.csv' OVERWRITE INTO TABLE $STAGING_DB.flow_storyboard_tmp;"
- echo $command
- hive -e "$command"
-
- command="INSERT INTO $DEST_DB.flow_storyboard PARTITION (y=$y, m=$m, d=$d)
-select ip_threat, title, text from $STAGING_DB.flow_storyboard_tmp;"
- echo $command
- hive -e "$command"
-
- fi
-
- ##flow_timeline
- echo "Processing Flow Timeline"
- timeline_files=`ls $dir/sbdet*.tsv`
-
- if [ ! -z "$timeline_files" ]
- then
- for file in $timeline_files
- do
- filename="$(basename $file)"
- ip="${filename%.tsv}"
- ip="${ip#sbdet-}"
- echo $filename $ip
-
- command="LOAD DATA LOCAL INPATH '$file' OVERWRITE INTO TABLE $STAGING_DB.flow_timeline_tmp;"
- echo $command
- hive -e "$command"
-
- command="INSERT INTO $DEST_DB.flow_timeline PARTITION (y=$y, m=$m, d=$d)
-select '$ip', tstart, tend, srcip, dstip, proto, sport, dport, ipkt, ibyt from $STAGING_DB.flow_timeline_tmp;"
- echo $command
- hive -e "$command"
-
- done
- fi
-done
-
-
-## Copying advanced mode notebooks to new paths in Spot 1.0
-for new_dir in $NEW_OA_PATH/ipynb/flow/*
-do
- cp $OLD_OA_PATH/oa/flow/ipynb_templates/Advanced_Mode_master.ipynb $new_dir/Advanced_Mode.ipynb
-done
-
-# Dropping staging tables
-impala-shell -i ${IMPALA_DEM} --var=dbname=${STAGING_DB} -c -f drop_flow_migration_tables.hql
-
-# Removing staging tables' path in HDFS
-hadoop fs -rm -r $HDFS_STAGING_PATH/flow/
-
-# Moving CSV data to backup folder
-mkdir $OLD_OA_PATH/data/backup/
-mv $OLD_OA_PATH/data/flow/ $OLD_OA_PATH/data/backup/
-
-# Invalidating metadata in Impala to refresh tables content
-impala-shell -i ${IMPALA_DEM} -q "INVALIDATE METADATA;"
-
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/401ffafe/spot-setup/migration/migrate_old_proxy_data.py
----------------------------------------------------------------------
diff --git a/spot-setup/migration/migrate_old_proxy_data.py b/spot-setup/migration/migrate_old_proxy_data.py
new file mode 100755
index 0000000..4298fc6
--- /dev/null
+++ b/spot-setup/migration/migrate_old_proxy_data.py
@@ -0,0 +1,222 @@
+#!/bin/env python
+
+#
+# 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 os
+import sys
+import subprocess
+import fnmatch
+import re
+import pandas as pd
+import datetime
+from utilities import util
+
+old_oa_path=sys.argv[1]
+staging_db=sys.argv[2]
+hdfs_staging_path=sys.argv[3]
+dest_db = sys.argv[4]
+impala_daemon = sys.argv[5]
+
+
+# Execution example:
+#./migrate_old_proxy_data.py '/home/spotuser/incubator-spot_old/spot-oa' 'spot_migration' '/user/spotuser/spot_migration/' 'migrated' 'node01'
+
+def main():
+ log = util.get_logger('SPOT.MIGRATE.PROXY')
+
+ cur_path = os.path.dirname(os.path.realpath(__file__))
+ new_spot_path = os.path.split(os.path.split(cur_path)[0])[0]
+ new_oa_path = '{0}/spot-oa'.format(new_spot_path)
+ log.info('New Spot OA path: {0}'.format(new_oa_path))
+ old_spot_path = os.path.split(old_oa_path)[0]
+
+
+ log.info("Creating HDFS paths for Impala tables")
+ util.create_hdfs_folder('{0}/proxy/scores'.format(hdfs_staging_path),log)
+ util.create_hdfs_folder('{0}/proxy/edge'.format(hdfs_staging_path),log)
+ util.create_hdfs_folder('{0}/proxy/summary'.format(hdfs_staging_path),log)
+ util.create_hdfs_folder('{0}/proxy/storyboard'.format(hdfs_staging_path),log)
+ util.create_hdfs_folder('{0}/proxy/timeline'.format(hdfs_staging_path),log)
+ util.execute_cmd('hdfs dfs -setfacl -R -m user:impala:rwx {0}'.format(hdfs_staging_path),log)
+
+
+ log.info("Creating Staging tables in Impala")
+ util.execute_cmd('impala-shell -i {0} --var=hpath={1} --var=dbname={2} -c -f create_proxy_migration_tables.hql'.format(impala_daemon, hdfs_staging_path, staging_db),log)
+
+
+ ## proxy Ingest Summary
+ log.info('Processing proxy Ingest Summary')
+ ing_sum_path='{0}/data/proxy/ingest_summary/'.format(old_oa_path)
+ pattern='is_??????.csv'
+ staging_table_name = 'proxy_ingest_summary_tmp'
+ dest_table_name = 'proxy_ingest_summary'
+
+ if os.path.exists(ing_sum_path):
+ for file in fnmatch.filter(os.listdir(ing_sum_path), pattern):
+ log.info('Processing file: {0}'.format(file))
+
+ filepath='{0}{1}'.format(ing_sum_path, file)
+ df = pd.read_csv(filepath)
+
+ s = df.iloc[:,0]
+ l_dates = list(s.unique())
+ l_dates = map(lambda x: x[0:10].strip(), l_dates)
+ l_dates = filter(lambda x: re.match('\d{4}[-/]\d{2}[-/]\d{1}', x), l_dates)
+ s_dates = set(l_dates)
+
+ for date_str in s_dates:
+ dt = datetime.datetime.strptime(date_str, '%Y-%m-%d')
+ log.info('Processing day: {0} {1} {2} {3}'.format(date_str, dt.year, dt.month, dt.day))
+
+ records = df[df['date'].str.contains(date_str)]
+ filename = "ingest_summary_{0}{1}{2}.csv".format(dt.year, dt.month, dt.day)
+ records.to_csv(filename, index=False)
+
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT tdate, total FROM {5}.{6}".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+ os.remove(filename)
+
+
+ ## Iterating days
+ days_path='{0}/data/proxy/'.format(old_oa_path)
+
+ if os.path.exists(days_path):
+ for day_folder in fnmatch.filter(os.listdir(days_path), '2*'):
+
+ print day_folder
+ dt = datetime.datetime.strptime(day_folder, '%Y%m%d')
+ log.info('Processing day: {0} {1} {2} {3}'.format(day_folder, dt.year, dt.month, dt.day))
+ full_day_path = '{0}{1}'.format(days_path,day_folder)
+
+
+ ## proxy Scores and proxy Threat Investigation
+ filename = '{0}/proxy_scores.tsv'.format(full_day_path)
+ if os.path.isfile(filename):
+
+ log.info("Processing Proxy Scores")
+ staging_table_name = 'proxy_scores_tmp'
+ dest_table_name = 'proxy_scores'
+
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT tdate, time, clientip, host, reqmethod, useragent, resconttype, duration, username, webcat, referer, respcode, uriport, uripath, uriquery, serverip, scbytes, csbytes, fulluri, word, ml_score, uri_rep, respcode_name, network_context FROM {5}.{6}".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+ log.info("Processing Proxy Threat Investigation")
+ staging_table_name = 'proxy_scores_tmp'
+ dest_table_name = 'proxy_threat_investigation'
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT tdate, fulluri, uri_sev FROM {5}.{6} WHERE uri_sev > 0;".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+
+ ## proxy Edge
+ log.info("Processing Proxy Edge")
+ staging_table_name = 'proxy_edge_tmp'
+ dest_table_name = 'proxy_edge'
+ pattern = 'edge*.tsv'
+ edge_files = fnmatch.filter(os.listdir(full_day_path), pattern)
+ filename = '{0}/{1}'.format(full_day_path, pattern)
+
+ if len(edge_files) > 0:
+
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT tdate, time, clientip, host, webcat, respcode, reqmethod, useragent, resconttype, referer, uriport, serverip, scbytes, csbytes, fulluri, 0, '' FROM {5}.{6};".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+
+ ##proxy_storyboard
+ log.info("Processing Proxy Storyboard")
+ staging_table_name = 'proxy_storyboard_tmp'
+ dest_table_name = 'proxy_storyboard'
+ filename = '{0}/threats.csv'.format(full_day_path)
+
+ if os.path.isfile(filename):
+
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT p_threat, title, text FROM {5}.{6};".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+
+ ##proxy Timeline
+ log.info("Processing Proxy Timeline")
+ staging_table_name = 'proxy_timeline_tmp'
+ dest_table_name = 'proxy_timeline'
+
+ for filename in fnmatch.filter(os.listdir(full_day_path), 'timeline*.tsv'):
+ print filename
+
+ hash_code = re.findall("timeline-(\S+).tsv", filename)[0]
+ extsearch_path = "{0}/es-{1}.csv".format(full_day_path, hash_code)
+ log.info('File: {0} Hash: {1} Extended Search file: {2}'.format(filename, hash_code, extsearch_path))
+
+ if os.path.isfile('{0}'.format(extsearch_path)):
+
+ log.info("Getting Full URI from extended search file")
+ es_df = pd.read_csv(extsearch_path, sep='\t')
+ fulluri = es_df.iloc[0]['fulluri']
+ log.info('Full URI found: {0}'.format(fulluri))
+
+ # Load timeline to staging table
+ load_cmd = "LOAD DATA LOCAL INPATH '{0}' OVERWRITE INTO TABLE {1}.{2};".format(filename, staging_db, staging_table_name)
+ util.execute_hive_cmd(load_cmd, log)
+
+ # Insert into new table from staging table and adding FullURI value
+ insert_cmd = "INSERT INTO {0}.{1} PARTITION (y={2}, m={3}, d={4}) SELECT '{5}', tstart, tend, duration, clientip, respcode, '' FROM {6}.{7} where cast(tstart as timestamp) is not null;".format(dest_db, dest_table_name, dt.year, dt.month, dt.day, fulluri, staging_db, staging_table_name)
+ util.execute_hive_cmd(insert_cmd, log)
+
+ else:
+ print "Extended search file {0} doesn't exist. Timeline on file {1} can't be processed".format(extsearch_path, timeline_path)
+
+
+
+ log.info("Dropping staging tables")
+ util.execute_cmd('impala-shell -i {0} --var=dbname={1} -c -f drop_proxy_migration_tables.hql'.format(impala_daemon, staging_db),log)
+
+ log.info("Removing staging tables' path in HDFS")
+ util.execute_cmd('hadoop fs -rm -r {0}/proxy/'.format(hdfs_staging_path),log)
+
+ log.info("Moving CSV data to backup folder")
+ util.execute_cmd('mkdir {0}/data/backup/'.format(old_oa_path),log)
+ util.execute_cmd('mv {0}/data/proxy/ {0}/data/backup/'.format(old_oa_path),log)
+
+ log.info("Invalidating metadata in Impala to refresh tables content")
+ util.execute_cmd('impala-shell -i {0} -q "INVALIDATE METADATA;"'.format(impala_daemon),log)
+
+
+ log.info("Copying advanced mode notebooks to each existing day in new Spot location")
+ for folder in os.listdir('{0}/ipynb/proxy/'.format(new_oa_path)):
+ util.execute_cmd('cp {0}/oa/proxy/ipynb_templates/Advanced_Mode_master.ipynb {1}/ipynb/proxy/{2}/Advanced_Mode.ipynb'.format(old_oa_path, new_oa_path, folder),log)
+
+ log.info("Copying threat investigation ipynb template to each existing day in new Spot location")
+ for folder in os.listdir('{0}/ipynb/proxy/'.format(new_oa_path)):
+ util.execute_cmd('cp {0}/oa/proxy/ipynb_templates/Threat_Investigation_master.ipynb {1}/ipynb/proxy/{2}/Threat_Investigation.ipynb'.format(old_oa_path, new_oa_path, folder),log)
+
+
+
+if __name__=='__main__':
+ main()
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/401ffafe/spot-setup/migration/migrate_old_proxy_data.sh
----------------------------------------------------------------------
diff --git a/spot-setup/migration/migrate_old_proxy_data.sh b/spot-setup/migration/migrate_old_proxy_data.sh
deleted file mode 100755
index f8eaea3..0000000
--- a/spot-setup/migration/migrate_old_proxy_data.sh
+++ /dev/null
@@ -1,169 +0,0 @@
-#!/bin/bash
-
-#
-# 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.
-#
-
-OLD_OA_PATH=$1
-NEW_OA_PATH=$2
-STAGING_DB=$3
-HDFS_STAGING_PATH=$4
-DEST_DB=$5
-IMPALA_DEM=$6
-
-# Execution example:
-#./migrate_old_proxy_data.sh '/home/spotuser/incubator-spot_old/spot-oa' '/home/spotuser/incubator-spot_new/spot-oa/' 'spot_migration' '/user/spotuser/spot_migration/' 'migrated' 'node01'
-
-hadoop fs -mkdir $HDFS_STAGING_PATH
-hadoop fs -mkdir $HDFS_STAGING_PATH/proxy/
-hadoop fs -mkdir $HDFS_STAGING_PATH/proxy/scores/
-hadoop fs -mkdir $HDFS_STAGING_PATH/proxy/edge/
-hadoop fs -mkdir $HDFS_STAGING_PATH/proxy/summary/
-hadoop fs -mkdir $HDFS_STAGING_PATH/proxy/storyboard/
-hadoop fs -mkdir $HDFS_STAGING_PATH/proxy/timeline/
-hdfs dfs -setfacl -R -m user:impala:rwx $HDFS_STAGING_PATH
-
-
-# Creating Staging tables in Impala
-impala-shell -i ${IMPALA_DEM} --var=hpath=${HDFS_STAGING_PATH} --var=dbname=${STAGING_DB} -c -f create_proxy_migration_tables.hql
-
-
-# Proxy Ingest Summary
-echo "Processing proxy Ingest Summary"
-
-ing_sum_path=$OLD_OA_PATH/data/proxy/ingest_summary/is_??????.csv
-
-for file in $ing_sum_path
-do
- echo $file
- ./import_ingest_summary.py "${file}" "${STAGING_DB}" 'proxy_ingest_summary_tmp' "${DEST_DB}" 'proxy_ingest_summary'
-done
-
-
-DAYS=$OLD_OA_PATH/data/proxy/2*
-
-for dir in $DAYS
-do
-
- day="$(basename $dir)"
- echo "Processing day $day ..."
- y=${day:0:4}
- m=$(expr ${day:4:2} + 0)
- d=$(expr ${day:6:2} + 0)
- echo $y $m $d $d2
- echo $dir
-
-
- ## proxy Scores and proxy_threat_investigation
- echo "Processing proxy Scores"
- if [ -f $dir/proxy_scores.tsv ]
- then
- command="LOAD DATA LOCAL INPATH '$dir/proxy_scores.tsv' OVERWRITE INTO TABLE $STAGING_DB.proxy_scores_tmp;"
- echo $command
- hive -e "$command"
-
- command="INSERT INTO $DEST_DB.proxy_scores PARTITION (y=$y, m=$m, d=$d)
-select tdate, time, clientip, host, reqmethod, useragent, resconttype, duration, username, webcat, referer, respcode, uriport, uripath, uriquery, serverip, scbytes, csbytes, fulluri, word, ml_score, uri_rep, respcode_name, network_context
-from $STAGING_DB.proxy_scores_tmp;"
-
- echo $command
- hive -e "$command"
-
- echo "Processing proxy Threat Investigation"
- command="INSERT INTO $DEST_DB.proxy_threat_investigation PARTITION (y=$y, m=$m, d=$d)
-select tdate, fulluri, uri_sev
-from $STAGING_DB.proxy_scores_tmp
-where uri_sev > 0;"
- echo $command
- hive -e "$command"
-
- fi
-
-
- ## proxy Edge
- echo "Processing proxy Edge"
- edge_files=`ls $dir/edge*.tsv`
- #echo $edge_files
- if [ ! -z "$edge_files" ]
- then
-
- command="LOAD DATA LOCAL INPATH '$dir/edge*.tsv' OVERWRITE INTO TABLE $STAGING_DB.proxy_edge_tmp;"
- echo $command
- hive -e "$command"
-
- command="INSERT INTO $DEST_DB.proxy_edge PARTITION (y=$y, m=$m, d=$d)
-select tdate, time, clientip, host, webcat, respcode, reqmethod, useragent, resconttype, referer, uriport, serverip, scbytes, csbytes, fulluri, 0, ''
-from $STAGING_DB.proxy_edge_tmp;"
- echo $command
- hive -e "$command"
-
- fi
-
-
- ##proxy_storyboard
- echo "Processing proxy Storyboard"
- if [ -f $dir/threats.csv ]
- then
-
- command="LOAD DATA LOCAL INPATH '$dir/threats.csv' OVERWRITE INTO TABLE $STAGING_DB.proxy_storyboard_tmp;"
- echo $command
- hive -e "$command"
-
- command="INSERT INTO $DEST_DB.proxy_storyboard PARTITION (y=$y, m=$m, d=$d)
-select p_threat, title, text
-from $STAGING_DB.proxy_storyboard_tmp;"
- echo $command
- hive -e "$command"
-
- fi
-
- ##proxy_timeline
- echo "Processing proxy Timeline"
- timeline_files=`ls $dir/timeline*.tsv`
-
- if [ ! -z "$timeline_files" ]
- then
- for file in $timeline_files
- do
-
- echo $file
- ./import_proxy_timeline.py "${file}" "${STAGING_DB}" 'proxy_timeline_tmp' "${DEST_DB}" 'proxy_timeline' "${y}-${m}-${d}"
-
- done
- fi
-done
-
-
-## Copying advanced mode notebooks to new paths in Spot 1.0
-for new_dir in $NEW_OA_PATH/ipynb/proxy/*
-do
- cp $OLD_OA_PATH/oa/proxy/ipynb_templates/Advanced_Mode_master.ipynb $new_dir/Advanced_Mode.ipynb
-done
-
-# Dropping staging tables
-impala-shell -i ${IMPALA_DEM} --var=dbname=${STAGING_DB} -c -f drop_proxy_migration_tables.hql
-
-# Removing staging tables' path in HDFS
-hadoop fs -rm -r $HDFS_STAGING_PATH/proxy/
-
-# Moving CSV data to backup folder
-mkdir $OLD_OA_PATH/data/backup/
-mv $OLD_OA_PATH/data/proxy/ $OLD_OA_PATH/data/backup/
-
-# Invalidating metadata in Impala to refresh tables content
-impala-shell -i ${IMPALA_DEM} -q "INVALIDATE METADATA;"
-
-
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/401ffafe/spot-setup/migration/utilities.py
----------------------------------------------------------------------
diff --git a/spot-setup/migration/utilities.py b/spot-setup/migration/utilities.py
new file mode 100644
index 0000000..046679e
--- /dev/null
+++ b/spot-setup/migration/utilities.py
@@ -0,0 +1,89 @@
+#!/bin/env python
+
+#
+# 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 os
+import sys
+import subprocess
+import logging
+
+
+class util(object):
+
+ @classmethod
+ def create_hdfs_folder(cls,hdfs_path,logger):
+ hadoop_create_folder="hadoop fs -mkdir -p {0}".format(hdfs_path)
+ logger.info("SPOT.Utils: Creating hdfs folder: {0}".format(hadoop_create_folder))
+ subprocess.call(hadoop_create_folder,shell=True)
+
+ @classmethod
+ def load_to_hdfs(cls,file_local_path,file_hdfs_path,logger):
+ # move file to hdfs.
+ load_to_hadoop_script = "hadoop fs -moveFromLocal {0} {1}".format(file_local_path,file_hdfs_path)
+ logger.info("SPOT.Utils: Loading file to hdfs: {0}".format(load_to_hadoop_script))
+ subprocess.call(load_to_hadoop_script,shell=True)
+
+ @classmethod
+ def get_logger(cls,logger_name,create_file=False):
+
+ # create logger for prd_ci
+ log = logging.getLogger(logger_name)
+ log.setLevel(level=logging.INFO)
+
+ # create formatter and add it to the handlers
+ formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+
+ if create_file:
+ # create file handler for logger.
+ fh = logging.FileHandler('SPOT.log')
+ fh.setLevel(level=logging.DEBUG)
+ fh.setFormatter(formatter)
+ # reate console handler for logger.
+ ch = logging.StreamHandler()
+ ch.setLevel(level=logging.DEBUG)
+ ch.setFormatter(formatter)
+
+ # add handlers to logger.
+ if create_file:
+ log.addHandler(fh)
+
+ log.addHandler(ch)
+ return log
+
+ @classmethod
+ def execute_cmd(cls,command,logger):
+
+ try:
+ logger.info("SPOT.Utils: Executing: {0}".format(command))
+ subprocess.call(command,shell=True)
+
+ except subprocess.CalledProcessError as e:
+ logger.error("SPOT.Utils: There was an error executing: {0}".format(e.cmd))
+ sys.exit(1)
+
+ @classmethod
+ def execute_hive_cmd(cls,command,logger):
+
+ try:
+ logger.info("SPOT.Utils: Executing hive command: hive -e \"{0}\"".format(command))
+ subprocess.call("hive -e \"{0}\"".format(command),shell=True)
+
+ except subprocess.CalledProcessError as e:
+ logger.error("SPOT.Utils: There was an error executing: {0}".format(e.cmd))
+ sys.exit(1)
+