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/01/25 18:36:55 UTC
[43/49] incubator-spot git commit: Added Ingest summary method to dns
and proxy, added validations to flows ingest summary method
Added Ingest summary method to dns and proxy, added validations to flows ingest summary method
Project: http://git-wip-us.apache.org/repos/asf/incubator-spot/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spot/commit/bea0e571
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spot/tree/bea0e571
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spot/diff/bea0e571
Branch: refs/heads/master
Commit: bea0e5712779b3f19dc86949d4d564c7718346ca
Parents: 7902857
Author: LedaLima <ga...@intel.com>
Authored: Mon Dec 12 10:56:25 2016 -0600
Committer: Everardo Lopez Sandoval (Intel) <el...@elopezsa-mac02.ra.intel.com>
Committed: Fri Jan 20 17:01:02 2017 -0800
----------------------------------------------------------------------
spot-oa/oa/dns/dns_oa.py | 74 +++++++++++++++++++++++++++++++++++++--
spot-oa/oa/flow/flow_oa.py | 73 +++++++++++++++++++++-----------------
spot-oa/oa/proxy/proxy_oa.py | 58 +++++++++++++++++++++++++++++-
3 files changed, 169 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/bea0e571/spot-oa/oa/dns/dns_oa.py
----------------------------------------------------------------------
diff --git a/spot-oa/oa/dns/dns_oa.py b/spot-oa/oa/dns/dns_oa.py
index 884d46c..4f0dab9 100644
--- a/spot-oa/oa/dns/dns_oa.py
+++ b/spot-oa/oa/dns/dns_oa.py
@@ -5,7 +5,7 @@ import json
import shutil
import sys
import datetime
-import csv
+import csv, math
from tld import get_tld
from collections import OrderedDict
@@ -14,7 +14,7 @@ from components.data.data import Data
from components.iana.iana_transform import IanaTransform
from components.nc.network_context import NetworkContext
from multiprocessing import Process
-
+import pandas as pd
import time
class OA(object):
@@ -70,6 +70,7 @@ class OA(object):
self._add_network_context()
self._create_dns_scores_csv()
self._get_oa_details()
+ self._ingest_summary()
##################
end = time.time()
@@ -151,7 +152,15 @@ class OA(object):
def _add_tld_column(self):
qry_name_col = self._conf['dns_results_fields']['dns_qry_name']
- self._dns_scores = [conn + [ get_tld("http://" + str(conn[qry_name_col]), fail_silently=True) if "http://" not in str(conn[qry_name_col]) else get_tld(str(conn[qry_name_col]), fail_silently=True)] for conn in self._dns_scores ]
+ for conn in self._dns_scores:
+ try:
+ tld = get_tld("http://" + str(conn[qry_name_col]))
+ except ValueError:
+ print conn[qry_name_col]
+
+
+ # self._dns_scores = [conn + [ get_tld("http://" + str(conn[qry_name_col]), fail_silently=True) if "http://" not in str(conn[qry_name_col]) else get_tld(str(conn[qry_name_col]), fail_silently=True)] for conn in self._dns_scores ]
+
def _add_reputation(self):
@@ -326,3 +335,62 @@ class OA(object):
dndro_qry = ("SELECT dns_a, dns_qry_name, ip_dst FROM (SELECT susp.ip_dst, susp.dns_qry_name, susp.dns_a FROM {0}.{1} as susp WHERE susp.y={2} AND susp.m={3} AND susp.d={4} AND susp.ip_dst='{5}' LIMIT {6}) AS tmp GROUP BY dns_a, dns_qry_name, ip_dst").format(db,table,year,month,day,ip_dst,limit)
# execute query
self._engine.query(dndro_qry,dendro_file)
+
+
+ def _ingest_summary(self):
+ # get date parameters.
+ yr = self._date[:4]
+ mn = self._date[4:6]
+ dy = self._date[6:]
+
+ self._logger.info("Getting ingest summary data for the day")
+
+ ingest_summary_cols = ["date","total"]
+ result_rows = []
+ df_filtered = pd.DataFrame()
+
+ ingest_summary_file = "{0}/is_{1}{2}.csv".format(self._ingest_summary_path,yr,mn)
+ ingest_summary_tmp = "{0}.tmp".format(ingest_summary_file)
+
+ if os.path.isfile(ingest_summary_file):
+ df = pd.read_csv(ingest_summary_file, delimiter=',')
+ #discards previous rows from the same date
+ df_filtered = df[df['date'].str.contains("{0}-{1}-{2}".format(yr, mn, dy)) == False]
+ else:
+ df = pd.DataFrame()
+
+ # get ingest summary.
+ ingest_summary_qry = ("SELECT frame_time, COUNT(*) as total "
+ " FROM {0}.{1}"
+ " WHERE y={2} AND m={3} AND d={4} "
+ " AND unix_tstamp IS NOT NULL AND frame_time IS NOT NULL"
+ " AND frame_len IS NOT NULL AND dns_qry_name IS NOT NULL"
+ " AND ip_src IS NOT NULL "
+ " AND (dns_qry_class IS NOT NULL AND dns_qry_type IS NOT NULL AND dns_qry_rcode IS NOT NULL ) "
+ " GROUP BY frame_time;")
+
+ ingest_summary_qry = ingest_summary_qry.format(self._db,self._table_name, yr, mn, dy)
+
+ results_file = "{0}/results_{1}.csv".format(self._ingest_summary_path,self._date)
+ self._engine.query(ingest_summary_qry,output_file=results_file,delimiter=",")
+
+
+ if os.path.isfile(results_file):
+ df_results = pd.read_csv(results_file, delimiter=',')
+
+ # Forms a new dataframe splitting the minutes from the time column
+ df_new = pd.DataFrame([["{0}-{1}-{2} {3}:{4}".format(yr, mn, dy,val['frame_time'].split(" ")[3].split(":")[0].zfill(2),val['frame_time'].split(" ")[3].split(":")[1].zfill(2)), int(val['total']) if not math.isnan(val['total']) else 0 ] for key,val in df_results.iterrows()],columns = ingest_summary_cols)
+
+ #Groups the data by minute
+ sf = df_new.groupby(by=['date'])['total'].sum()
+
+ df_per_min = pd.DataFrame({'date':sf.index, 'total':sf.values})
+
+ df_final = df_filtered.append(df_per_min, ignore_index=True)
+ df_final.to_csv(ingest_summary_tmp,sep=',', index=False)
+
+ os.remove(results_file)
+ os.rename(ingest_summary_tmp,ingest_summary_file)
+ else:
+ self._logger.info("No data found for the ingest summary")
+
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/bea0e571/spot-oa/oa/flow/flow_oa.py
----------------------------------------------------------------------
diff --git a/spot-oa/oa/flow/flow_oa.py b/spot-oa/oa/flow/flow_oa.py
index 870251b..7c1013a 100644
--- a/spot-oa/oa/flow/flow_oa.py
+++ b/spot-oa/oa/flow/flow_oa.py
@@ -7,6 +7,7 @@ import json
import numpy as np
import linecache, bisect
import csv
+import pandas as pd
from collections import OrderedDict
from multiprocessing import Process
@@ -51,7 +52,7 @@ class OA(object):
# initialize data engine
self._db = self._spot_conf.get('conf', 'DBNAME').replace("'", "").replace('"', '')
self._engine = Data(self._db, self._table_name,self._logger)
-
+
def start(self):
####################
@@ -386,47 +387,55 @@ class OA(object):
self._engine.query(ch_query.format(self._db,self._table_name,yr,mn,dy,ip,ips_filter),chord_file,delimiter="\\t")
- def _ingest_summary(self):
-
+ def _ingest_summary(self):
# get date parameters.
yr = self._date[:4]
mn = self._date[4:6]
dy = self._date[6:]
+ self._logger.info("Getting ingest summary data for the day")
+
+ ingest_summary_cols = ["date","total"]
+ result_rows = []
+ df_filtered = pd.DataFrame()
+
+ ingest_summary_file = "{0}/is_{1}{2}.csv".format(self._ingest_summary_path,yr,mn)
+ ingest_summary_tmp = "{0}.tmp".format(ingest_summary_file)
+ if os.path.isfile(ingest_summary_file):
+ df = pd.read_csv(ingest_summary_file, delimiter=',',names=ingest_summary_cols, skiprows=1)
+ df_filtered = df[df['date'].str.contains("{0}-{1}-{2}".format(yr, mn, dy)) == False]
+ else:
+ df = pd.DataFrame()
+
# get ingest summary.
- ingest_summary_qry = ("SELECT tryear, trmonth, trday, trhour, trminute, COUNT(*) flows"
- " FROM {0}.flow "
- " WHERE "
- " y={1} "
- " AND m={2} "
- " AND d={3} "
- " AND unix_tstamp IS NOT NULL "
- " GROUP BY tryear, trmonth, trday, trhour, trminute;")
-
- ingest_summary_qry = ingest_summary_qry.format(self._db, yr, mn, dy)
-
+ ingest_summary_qry = ("SELECT tryear, trmonth, trday, trhour, trminute, COUNT(*) total"
+ " FROM {0}.{1} "
+ " WHERE "
+ " y={2} "
+ " AND m={3} "
+ " AND d={4} "
+ " AND unix_tstamp IS NOT NULL AND sip IS NOT NULL "
+ " AND sport IS NOT NULL AND dip IS NOT NULL "
+ " AND dport IS NOT NULL AND ibyt IS NOT NULL "
+ " AND ipkt IS NOT NULL "
+ " GROUP BY tryear, trmonth, trday, trhour, trminute;")
+
+
+ ingest_summary_qry = ingest_summary_qry.format(self._db,self._table_name, yr, mn, dy)
results_file = "{0}/results_{1}.csv".format(self._ingest_summary_path,self._date)
self._engine.query(ingest_summary_qry,output_file=results_file,delimiter=",")
-
- result_rows = []
- with open(results_file, 'rb') as rf:
- csv_reader = csv.reader(rf, delimiter = ",")
- result_rows = list(csv_reader)
-
- result_rows = iter(result_rows)
- next(result_rows)
- ingest_summary_results = [ ["date","flows"] ]
- ingest_summary_results.extend([ ["{0}-{1}-{2} {3}:{4}".format(yr, mn, dy, row[3].zfill(2) ,row[4].zfill(2)), row[5]] for row in result_rows ])
- ingest_summay_file = "{0}/is_{1}{2}.csv".format(self._ingest_summary_path,yr,mn)
+ if os.path.isfile(results_file):
+ result_rows = pd.read_csv(results_file, delimiter=',')
+ df_new = pd.DataFrame([["{0}-{1}-{2} {3}:{4}".format(yr, mn, dy, str(val['trhour']).zfill(2), str(val['trminute']).zfill(2)), int(val[5])] for key,val in result_rows.iterrows()],columns = ingest_summary_cols)
- write_format = 'a' if os.path.isfile(ingest_summay_file) else 'w+'
- with open(ingest_summay_file, write_format) as u_file:
- writer = csv.writer(u_file, quoting=csv.QUOTE_NONE, delimiter=",")
- writer.writerows(ingest_summary_results)
+ df_filtered = df_filtered.append(df_new, ignore_index=True)
+ df_filtered.to_csv(ingest_summary_tmp,sep=',', index=False)
- rm_big_file = "rm {0}".format(results_file)
- os.remove(results_file)
-
+ os.remove(results_file)
+ os.rename(ingest_summary_tmp,ingest_summary_file)
+ else:
+ self._logger.info("No data found for the ingest summary")
+
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/bea0e571/spot-oa/oa/proxy/proxy_oa.py
----------------------------------------------------------------------
diff --git a/spot-oa/oa/proxy/proxy_oa.py b/spot-oa/oa/proxy/proxy_oa.py
index d9b0362..7359ba1 100644
--- a/spot-oa/oa/proxy/proxy_oa.py
+++ b/spot-oa/oa/proxy/proxy_oa.py
@@ -5,13 +5,14 @@ import json
import shutil
import sys
import datetime
-import csv
+import csv, math
from collections import OrderedDict
from utils import Util
from components.data.data import Data
from components.iana.iana_transform import IanaTransform
from components.nc.network_context import NetworkContext
from multiprocessing import Process
+import pandas as pd
import time
import md5
@@ -70,6 +71,7 @@ class OA(object):
self._add_hash()
self._create_proxy_scores_csv()
self._get_oa_details()
+ self._ingest_summary()
##################
@@ -292,3 +294,57 @@ class OA(object):
except OSError:
pass
+
+ def _ingest_summary(self):
+ # get date parameters.
+ yr = self._date[:4]
+ mn = self._date[4:6]
+ dy = self._date[6:]
+
+ self._logger.info("Getting ingest summary data for the day")
+
+ ingest_summary_cols = ["date","total"]
+ result_rows = []
+ df_filtered = pd.DataFrame()
+
+ ingest_summary_file = "{0}/is_{1}{2}.csv".format(self._ingest_summary_path,yr,mn)
+ ingest_summary_tmp = "{0}.tmp".format(ingest_summary_file)
+
+ if os.path.isfile(ingest_summary_file):
+ df = pd.read_csv(ingest_summary_file, delimiter=',')
+ #discards previous rows from the same date
+ df_filtered = df[df['date'].str.contains("{0}-{1}-{2}".format(yr, mn, dy)) == False]
+ else:
+ df = pd.DataFrame()
+
+ # get ingest summary.
+ ingest_summary_qry = ("SELECT p_date, p_time, COUNT(*) as total "
+ " FROM {0}.{1}"
+ " WHERE y='{2}' AND m='{3}' AND d='{4}' "
+ " AND p_date IS NOT NULL AND p_time IS NOT NULL "
+ " AND clientip IS NOT NULL AND p_time != '' "
+ " AND host IS NOT NULL AND fulluri IS NOT NULL "
+ " GROUP BY p_date, p_time;")
+
+ ingest_summary_qry = ingest_summary_qry.format(self._db,self._table_name, yr, mn, dy)
+ results_file = "{0}/results_{1}.csv".format(self._ingest_summary_path,self._date)
+ self._engine.query(ingest_summary_qry,output_file=results_file,delimiter=",")
+
+ if os.path.isfile(results_file):
+ df_results = pd.read_csv(results_file, delimiter=',')
+
+ #Forms a new dataframe splitting the minutes from the time column/
+ df_new = pd.DataFrame([["{0} {1}:{2}".format(val['p_date'], val['p_time'].split(":")[0].zfill(2), val['p_time'].split(":")[1].zfill(2)), int(val['total']) if not math.isnan(val['total']) else 0 ] for key,val in df_results.iterrows()],columns = ingest_summary_cols)
+
+ #Groups the data by minute
+ sf = df_new.groupby(by=['date'])['total'].sum()
+ df_per_min = pd.DataFrame({'date':sf.index, 'total':sf.values})
+
+ df_final = df_filtered.append(df_per_min, ignore_index=True)
+ df_final.to_csv(ingest_summary_tmp,sep=',', index=False)
+
+ os.remove(results_file)
+ os.rename(ingest_summary_tmp,ingest_summary_file)
+ else:
+ self._logger.info("No data found for the ingest summary")
+
\ No newline at end of file