josch
/
flowy
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
main
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from 'main'
${ noResults }
flowy/merger_validator.py

503 lines
23 KiB
Python
Raw Permalink Blame History

from validator_common import *
from copy import deepcopy
from merger import MergerStorage
from merger import Merger as MergerImpl
from merger import MergerBranch as MergerBranchImpl
from merger import MergerLastBranch as MergerLastBranchImpl
from merger import MergerRejectBranch as MergerRejectBranchImpl
from merger import MergerRule as MergerRuleImpl
import itertools
import allen_ops
import options
class MergerValidator(object):
def __init__(self, parser, gr_filter_validator):
self.parser = parser
self.gr_filter_validator = gr_filter_validator
self.mergers = deepcopy(parser.mergers)
# The last field returns a list of the present fields for each branch
# ('rec_id', 'Last', 'First', 'records', 'srcip', 'dstip', 'bytes', 'n', 'flags', 'srcports')
# ('rec_id', 'Last', 'First', 'records', 'srcip', 'dstip', 'bytes', 'n', 'flags', 'dstports')
self.branches_fields = gr_filter_validator.branches_fields
# A simple dictionary mapptin of branch name to a GroupFilter
# {'A': <groupfilter.GroupFilter object at 0x9c3d66c>, 'B': <groupfilter.GroupFilter object at 0x9c43ccc>}
self.br_name_to_gr_filter = gr_filter_validator.br_name_to_gr_filter
# Checks that all the defined merger modules are actually exported
# Returns a dictionay of a merger name and module implementation
self.megers_export_modules = self.find_mergers_export_modules()
# Returns the size of the field type of the 'records' field, 4 bytes
self.id_size = self.get_id_size()
self.impl = self.get_mergers_impl()
# Returns the size of the field type of the 'records' field, 4 bytess
def get_id_size(self):
rec_reader = self.gr_filter_validator.impl[0].records
field_types = dict(zip(rec_reader.group_record_fields,
rec_reader.group_record_types))
id_size = field_types['records'].itemsize
return id_size
# Check for duplicate merger names
def check_duplicate_merger_names(self):
duplicates = {}
for merger in self.mergers:
old_val = duplicates.setdefault(merger.name, 0)
duplicates[merger.name] = old_val + 1
duplicate_names = [k for k,v in duplicates.iteritems() if v > 1]
if len(duplicate_names) > 0:
msg = "Merger(s) %s"%duplicate_names
msg += " is/are all defined more than once."
raise SyntaxError(msg)
# Check for duplicate module names
def check_duplicate_module_names(self, merger):
duplicates = {}
for module in merger.modules:
old_val = duplicates.setdefault(module.name, 0)
duplicates[module.name] = old_val + 1
duplicate_names = [k for k,v in duplicates.iteritems() if v > 1]
if len(duplicate_names) > 0:
msg = "Module(s) %s"%duplicate_names
msg += " is/are all defined more than once in merger"
msg += " %s."%merger.name
raise SyntaxError(msg)
# Checks that all the defined merger modules are actually exported
# Returns a dictionay of a merger name and module implementation
def find_mergers_export_modules(self):
merger_to_export_module = {}
for merger in self.mergers:
exp = None
for module in merger.modules:
if merger.export == module.name:
exp = module
break
if exp:
merger_to_export_module[merger.name] = exp
# print merger_to_export_module
else:
msg = "Merger %s"%merger.name
msg += " export module %s is not defined."%merger.export
return merger_to_export_module
#--------------------------------------ALLEN CHECKS-------------------------------------#
#All the operations on rules are around a sample set like: {'M': Module('m1', 38, [[Rule('EQ', 40, [Field('A.srcip'), Field('B.dstip')], False)], [Rule('EQ', 41, [Field('A.srcports'), Field('B.dstports')], False)], [Rule('LT', 42, [Field('A.bytes'), Field('B.bytes')], False)], [AllenRule('oi', 43, [Field('B'), Field('A')], False), AllenRule('d', 43, [Field('B'), Field('A')], False)]], ['B', 'A'])}
#Returns only the Allen rules
def iterate_module_allen_op_groups(self, merger):
for module in merger.modules:
for rules in module.rules:
if type(rules[0]) is not AllenRule:
continue
else:
for op in rules:
yield op
# Orders allen operations and the branches that they influence in a reverse order, if not already so
def order_allen_ops_args(self, merger):
order = self.get_merger_branches_order(merger)#Orders merger branches, exported module's branches being first
arg_combinaions = tuple(itertools.combinations(order, 2))#combinations('ABCD', 2) --> AB AC AD BC BD CD
for allen_op in self.iterate_module_allen_op_groups(merger):#Returns only the Allen rules
first, second = allen_op.args[:2] # Returns Field('B') Field('A')
op = allen_op.op # operations like oi, d
if (first.name, second.name) not in arg_combinaions:
allen_op.args = [second, first] + allen_op.args[2:]# reverse names
allen_op.op = allen_ops.inv_op_str(op)# and operations
# A number of different checks of the AllenRule
def check_allen_ops(self, merger):
allen_arg_pairs = []
arg_pairs_to_line = {}
for module in merger.modules:
for rules in module.rules:
if type(rules[0]) is not AllenRule:
continue
first_arg = rules[0].args[0].name # Get the branch names influenced by the AllenRule
second_arg = rules[0].args[1].name
line = rules[0].line
order = (first_arg, second_arg)
allen_arg_pairs.append(order)# [('B', 'A')]
self.check_allen_satisfiability(arg_pairs_to_line, order, line)
self.check_allen_consistency(first_arg, second_arg, rules)
self.check_allen_deltas(rules)
self.check_allen_reachability(allen_arg_pairs, merger)
# The following 3 methods run different tests on the allen arguments and rules
def check_allen_satisfiability(self, arg_pairs_to_line, order, line):
if arg_pairs_to_line.has_key(order):
msg = "Unsatisfiable Allen op group. "
msg += "All allen ops concerning a pair of branches should"
msg += " be connected with and OR into a single group "
msg += "within a single module.\n"
msg += "Argument pair %s on line %s"%(order, line)
msg += " is also used on line %s."%arg_pairs_to_line[order]
raise SyntaxError(msg)
else:
arg_pairs_to_line[order] = line
def check_allen_consistency(self, first_arg, second_arg, rules):
for al_op in rules:
first = al_op.args[0].name
second = al_op.args[1].name
if (first != first_arg or second != second_arg):
msg = "Inconsistent group of Allen statements "
msg += "on line %s"%rules[0].line
msg += ": %s, %s.\n"%(first, second)
msg += "All branches in this group should have "
msg += "%s and %s"%(first_arg, second_arg)
msg += " as left and righthand side arguments "
msg += "respectively."
raise SyntaxError(msg)
def check_allen_deltas(self, rules):
for al_op in rules:
if al_op.op == 'LT' or al_op.op == 'GT':
if len(al_op.args) < 3:
msg = "Allen op < or > on line %s "%al_op.line
msg += " should have delta explicitly stated."
raise SyntaxError(msg)
# A check for reachability of subsequent branches from the first one
def check_allen_reachability(self, allen_arg_pairs, merger):
br_order = self.get_merger_branches_order(merger)
# check reachability through allen index from initial branch
# of export module:
reachable = br_order[0:1] # list of first branch of exp module
unreachable = br_order[1:]
change = True
while(change):
change = False
for arg1, arg2 in allen_arg_pairs:
if arg1 in reachable and arg2 in unreachable:
unreachable.remove(arg2)
reachable.append(arg2)
change = True
if len(unreachable) > 0:
msg = "Branch(es): %s"%unreachable
msg += " in merger %s"%merger.name
msg += " is/are unreachable through an allen op or chain of"
msg += " allen ops from the first branch of the exported module"
raise SyntaxError(msg)
#--------------------------------------END ALLEN CHECKS---------------------------------#
# Orders the merger modules s.t. the exported module comes first
def order_modules(self):
for merger in self.mergers:
exp_module = self.megers_export_modules[merger.name]
new_modules_order = [exp_module]
new_modules_order += [m for m in merger.modules if m != exp_module]
merger.modules = new_modules_order
# Checks that the modules are interconnected among each other with at least one branch
def check_for_disjoint_modules(self):
for merger in self.mergers:
exp_module = self.megers_export_modules[merger.name]
exp_branches = set(exp_module.branches)
for module in merger.modules:
branches = set(module.branches)
# NOTE & is set intersection
if len(exp_branches & branches) < 1:
msg = "Merger module %s.%s"%(merger.name,module.name)
msg += " in has no overlaping branches with the"
msg += " export module."
raise SyntaxError(msg)
# Check the validity of the AllenRule, by seeing if the branch names are all defined
def check_branch_id_ref(self, rule, module_branches):
for arg in rule.args:
if type(arg) is Field:
id_ref = arg.name
if id_ref not in self.br_name_to_gr_filter.keys():
msg = 'Branch %s referenced on line'%id_ref
msg += ' %s is not defined.'%rule.line
raise SyntaxError(msg)
if id_ref not in module_branches:
msg = 'Branch %s referenced on line'%id_ref
msg += " %s "%rule.line
msg += "is not in module's branches statement."
raise SyntaxError(msg)
# Check the validity of the Rule, GrouperRule and statements like A.bytes
def check_qid_field_ref(self, rule, module_branches):
for arg in rule.args:
if type(arg) is Field:
qid_field = arg.name
branch, _, field = qid_field.partition('.') #Separates statements like A.bytes
try:
if field not in self.branches_fields[branch]:
msg = 'Wrong field %s on line %s. '%(qid_field,
rule.line)
msg += 'Branch %s does not have field %s.'%(branch,
field)
raise SyntaxError(msg)
except KeyError:
msg = 'Branch %s referenced on line'%branch
msg += ' %s is not defined'%rule.line
raise SyntaxError(msg)
if branch not in module_branches:
msg = 'Branch %s referenced on line'%branch
msg += " %s "%rule.line
msg += "is not in module's branches statement."
raise SyntaxError(msg)
# Orders merger branches with the exported module's branches being first
def get_merger_branches_order(self, merger):
br_order = []
# first add export module
for module in merger.modules:
if module.name == merger.export:
for br in module.branches:
if br not in br_order:
br_order.append(br)
# add all the others:
for module in merger.modules:
for br in module.branches:
if br not in br_order:
br_order.append(br)
return br_order
#
def order_merger_rules(self, merger):
"""
Produces mapping between incrementally larger available branches tuples
(A,B,C,etc) ordered as they will appear in the implementation.
"""
br_order = self.get_merger_branches_order(merger)
needed_brs_to_rule = {}
for module in merger.modules:
replace_with_vals(module)
replace_bound_rules(module)
for rules in module.rules:
rule_branches = self.get_rule_needed_branches(rules[0])
ordered_branches = tuple(br for br in br_order
if br in rule_branches)
if len(rules) > 1:
rule = Rule('or_op', 0, rules)
else:
rule = rules[0]
needed_brs_to_rule.setdefault(ordered_branches,
[]).append(rule)
avail_to_rules = {}
tup = ()
# create sets - needed for the set intersection operation
needed_sets = map(set, needed_brs_to_rule.keys())
# incrementaly add branches to the tuple of available branches
# and check which rules have their branch needs satisfied
for br in br_order:
tup += (br,)
# find how many of the needed branches are in this tuple
# of branches. It makes elementwise intesection of the sets
# of the needed branches and the tuple of available branches
intersect = map(set(tup).intersection , needed_sets )
for el, intersection, key in zip(needed_sets , intersect,
needed_brs_to_rule.keys()):
if len(intersection) == len(el):
# Lenght is the same, which means all needed branches
# are present. Remove these elements, take the rules from
# the needed_brs_to_rule and delete the key their to
# keep the zip() in sync
needed_sets.remove(el)
avail_to_rules[tup] = needed_brs_to_rule[key]
del needed_brs_to_rule[key]
return avail_to_rules
#
def get_rule_needed_branches(self, rule):
args_list = set()
for sub_rule in iterate_subrules(rule):
for arg in sub_rule.args:
if type(arg) is Field:
args_list.add(arg.name)
for arg in rule.args:
if type(arg) is Field:
args_list.add(arg.name)
if type(rule) is AllenRule:
return list(args_list)
else:
return [qid.partition('.')[0] for qid in args_list]
# Validates the correctness of the merger stage
def validate(self):
self.check_duplicate_merger_names()
for merger in self.mergers:
self.check_duplicate_module_names(merger)
for module in merger.modules:
# Checks the whole rule list to see that all
# the rules fall into [Rule, GrouperRule, AllenRule]
# Returns the actual rules
for rule in iterate_rules(module):
# Checks that all the rule entries are correctly specified
if type(rule) is AllenRule:
self.check_branch_id_ref(rule, module.branches)
else:
self.check_qid_field_ref(rule, module.branches)
# Orders allen operations and the branches that they influence in a reverse order
self.order_allen_ops_args(merger)
# Performs several checks on the branches and the operations (consistency, reachability, etc.)
self.check_allen_ops(merger)
# Orders the merger modules s.t. the exported module comes first
self.order_modules()
# Checks that the modules are interconnected among each other with at least one branch
self.check_for_disjoint_modules()
# Get the allen indexing operations for each branch.
def get_branches_allen_index_ops(self, merger):
"""
Get the allen indexing operations for each branch.
"""
br_to_allen_ind_ops = {}
for module in merger.modules:
for rules in module.rules:
if type(rules[0]) != AllenRule:
continue
br = rules[0].args[0].name
br_to_allen_ind_ops.setdefault(br, []).append(rules)
return br_to_allen_ind_ops
#
def get_rule_impl(self, rule, br_to_record):
if type(rule) == AllenRule:
op = find_op(rule, module='allen_ops')
args = [ (arg.name, None)
if type(arg) == Field else arg
for arg in rule.args]
else:
args = []
op = find_op(rule)
for arg in rule.args:
if type(arg) == Rule:
arg_impl = self.get_rule_impl(arg, br_to_record)
elif type(arg) == Field:
branch, _, field = arg.name.partition('.')
arg_impl = (branch, field)
else:
arg_impl = arg
args.append(arg_impl)
return MergerRuleImpl(op, args, br_to_record)
# Create indexing rules implementation for AllenRules
def get_index_rule_impl(self, rules):
res = []
for or_rules in rules:
or_rules_impl = []
for rule in or_rules:
op = find_op(rule, 'allen_index')
args = [arg.name if type(arg) == Field else arg
for arg in rule.args]
# replace with values
args = [arg.value if type(arg) == Arg else arg
for arg in args]
#[<allen_index.oi object at 0x9f5adcc>, <allen_index.d object at 0x9f5ae0c>]
or_rules_impl.append(op(*args))
res.append(or_rules_impl)
return res
# Creates a file MergedM.h5 for further storage of the merged files
def get_merger_table_impl(self, merger):
fields = self.megers_export_modules[merger.name].branches
types = [UIntCol(self.id_size) for _ in fields]
field_types = dict(zip(fields,types))
recordClass = record.get_record_class(fields, types)
# TODO fix file names
fname = options.temp_path + options.merger_file_prefix
fname += merger.name + ".h5"
if options.delete_temp_files: if_exists_delete(fname)
pytables.create_table_file(fname, field_types)
mergerTable = FlowRecordsTable(fname)
return MergerStorage(merger.name, mergerTable, recordClass)
# Actual implementation of the merger stage
def get_merger_impl(self, merger):
# Create merger storage
merger_table = self.get_merger_table_impl(merger)
# Create indexing rules implementation
br_to_index_rule_impl = {}
# {'B': [[AllenRule('oi', 43, [Field('B'), Field('A')], False), AllenRule('d', 43, [Field('B'), Field('A')], False)]]}
for br, rules in self.get_branches_allen_index_ops(merger).iteritems():
br_to_index_rule_impl[br] = self.get_index_rule_impl(rules)# List of allen index rules implemented
for br in self.get_merger_branches_order(merger):#orders branches with the exported branch being first
if br not in br_to_index_rule_impl.keys():
br_to_index_rule_impl[br] = []
# some "globals" shared among branches or needed for their creation
needed_brs = self.order_merger_rules(merger) # Re-orders the rules as they will appear in the implementation
tup = () # tuple of available branches
name = merger.name
br_order = self.get_merger_branches_order(merger) # Returns reversely-ordered branch names of the merger
export_branches = self.megers_export_modules[merger.name].branches # Returns branch names contained in the export module
br_to_record = {}
name_to_branch = {}
merger_impl = None
for br_name in br_order: # For each branch in the ordered branch set
tup += (br_name,)
next_branches_names = [br for br in br_order if br not in tup]
records = self.br_name_to_gr_filter[br_name] # Group-filters associated with each branch
index_rules = br_to_index_rule_impl[br_name] # Allen index rule associated with each branch
index = records.index # Time index object
if len(tup)<2: # If tuple contains only one branch, then execute the initial Merger class
# first branch
rules = []
impl = MergerImpl(name, br_name, records, name_to_branch,
next_branches_names, export_branches,
br_to_record, index, index_rules, rules,
merger_table)
merger_impl = impl
else:
unimpl_rules = needed_brs[tup]
rules = [self.get_rule_impl(rule, br_to_record)
for rule in unimpl_rules]
if br_name not in export_branches:
# Reject branch
impl = MergerRejectBranchImpl(br_name, records,
name_to_branch, next_branches_names,
export_branches, br_to_record, index,
index_rules, rules, merger_table)
elif not next_branches_names:
# Last non-rejecting branch
impl = MergerLastBranchImpl(br_name, records,
name_to_branch, next_branches_names,
export_branches, br_to_record, index,
index_rules, rules, merger_table)
else:
# For normal middle branches execute the MergerBranch class
impl = MergerBranchImpl(br_name, records, name_to_branch,
next_branches_names, export_branches,
br_to_record, index, index_rules,
rules, merger_table)
name_to_branch[br_name] = impl
return merger_impl
def get_mergers_impl(self):
self.validate()
mergers_impl = [self.get_merger_impl(merger)
for merger in self.mergers]
return mergers_impl
Reference in New Issue View Git Blame Copy Permalink