openstack
/
monasca-analytics
Code Issues Proposed changes

migrated config checking library from schema to voluptuous 

Change-Id: I371db2461a7ecb216f00b99aaf8296bde42f02f7
This commit is contained in:
David Subiros Perez 2016-07-01 16:16:30 +01:00
parent b1db54f22b
commit 4c14dace26
33 changed files with 358 additions and 261 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
monasca_analytics/config/validation.py
View File

@ -18,7 +18,7 @@
import logging
import schema
import voluptuous
from monasca_analytics.config import const
@ -73,7 +73,7 @@ def validate_links(links):
if component not in all_keys:
missing.add(component.id())
if len(missing) > 0:
raise schema.SchemaError([
raise voluptuous.Invalid([
"In connections section, the following components are not "
"connected\n\t{}\n"
"please modify the configuration so that their list of "
@ -92,11 +92,11 @@ def _validate_schema(config):
:raises: SchemaError -- if the configuration, up to the
orchestration level, is not valid
"""
config_schema = schema.Schema({
config_schema = voluptuous.Schema({
"spark_config": {
"appName": basestring,
"streaming": {
"batch_interval": schema.And(int, lambda b: b > 0)
"batch_interval": voluptuous.And(int, voluptuous.Range(min=1))
}
},
"server": {
@ -104,31 +104,31 @@ def _validate_schema(config):
"debug": bool
},
"sources": {
schema.Optional(basestring): {basestring: object}
voluptuous.Optional(basestring): {basestring: object}
},
"ingestors": {
schema.Optional(basestring): {basestring: object}
voluptuous.Optional(basestring): {basestring: object}
},
"smls": {
schema.Optional(basestring): {basestring: object}
voluptuous.Optional(basestring): {basestring: object}
},
"voters": {
schema.Optional(basestring): {basestring: object}
voluptuous.Optional(basestring): {basestring: object}
},
"sinks": {
schema.Optional(basestring): {basestring: object}
voluptuous.Optional(basestring): {basestring: object}
},
"ldps": {
schema.Optional(basestring): {basestring: object}
voluptuous.Optional(basestring): {basestring: object}
},
"connections": {
schema.Optional(basestring): [basestring]
voluptuous.Optional(basestring): [basestring]
},
"feedback": {
schema.Optional(basestring): [basestring]
voluptuous.Optional(basestring): [basestring]
}
})
return config_schema.validate(config)
}, required=True)
return config_schema(config)
def _validate_only_one_voter(config):
@ -139,7 +139,7 @@ def _validate_only_one_voter(config):
:raises: SchemaError -- if there is more than one voter defined in config
"""
def _raise(comp):
raise schema.SchemaError([
raise voluptuous.Invalid([
"More than one {} found in the config, please modify " +
"it specifying only one {}".format(comp, comp)], [])
@ -158,7 +158,7 @@ def _validate_ids_uniqueness(config):
for comp_type in valid_connection_types.keys():
for com_id in config[comp_type].keys():
if com_id in all_ids:
raise schema.SchemaError(
raise voluptuous.Invalid(
["Duplicated component ID : " + com_id], [])
all_ids.add(com_id)
@ -186,7 +186,7 @@ def _validate_expected_dest_type(config, from_id, to_ids, expected_types):
valid_connection = True
break
if not valid_connection:
raise schema.SchemaError([
raise voluptuous.Invalid([
from_id + " connected to a wrong component: " + to_id +
". It should be connected only to any of : " +
str(expected_types)], [])
@ -205,7 +205,7 @@ def _validate_existing_id(config, component_id):
if component_id in config[comp_type].keys():
found_id = True
if not found_id:
raise schema.SchemaError([
raise voluptuous.Invalid([
'In "connections", component `{}` hasn\'t been defined'
.format(component_id)
], [])

4
monasca_analytics/dsl/dsl.py
View File

@ -19,7 +19,7 @@ import json
import logging
import os
import schema
import voluptuous
from monasca_analytics.config import const
from monasca_analytics.config import validation
@ -119,7 +119,7 @@ class MonanasDSL(object):
logger.debug("New validated config = " + str(new_conf))
self._config[comp_type][comp_id] = new_conf
return True
except schema.SchemaError as e:
except voluptuous.Invalid as e:
logger.debug(str(e))
continue
return False

11
monasca_analytics/ingestor/cloud.py
View File

@ -18,10 +18,11 @@ import json
import logging
import numpy as np
import schema
import voluptuous
from monasca_analytics.ingestor import base
import monasca_analytics.util.spark_func as fn
from monasca_analytics.util import validation_utils as vu
logger = logging.getLogger(__name__)
@ -34,10 +35,10 @@ class CloudIngestor(base.BaseIngestor):
@staticmethod
def validate_config(_config):
return schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i))
}).validate(_config)
cloud_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.NoSpaceCharacter())
}, required=True)
return cloud_schema(_config)
def map_dstream(self, dstream):
features_list = list(self._features)

11
monasca_analytics/ingestor/iptables.py
View File

@ -17,11 +17,12 @@
import logging
import numpy as np
import schema
import voluptuous
from monasca_analytics.ingestor import base
from monasca_analytics.source import iptables_markov_chain as src
import monasca_analytics.util.spark_func as fn
from monasca_analytics.util import validation_utils as vu
logger = logging.getLogger(__name__)
@ -43,10 +44,10 @@ class IptablesIngestor(base.BaseIngestor):
@staticmethod
def validate_config(_config):
return schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i))
}).validate(_config)
iptables_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.NoSpaceCharacter())
}, required=True)
return iptables_schema(_config)
@staticmethod
def get_default_config():

11
monasca_analytics/ldp/cloud_causality.py
View File

@ -17,9 +17,10 @@
import json
import logging
import schema
import voluptuous
import monasca_analytics.ldp.base as bt
from monasca_analytics.util import validation_utils as vu
logger = logging.getLogger(__name__)
@ -30,10 +31,10 @@ class CloudCausalityLDP(bt.BaseLDP):
@staticmethod
def validate_config(_config):
return schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
}).validate(_config)
cloud_causality_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.NoSpaceCharacter())
}, required=True)
return cloud_causality_schema(_config)
@staticmethod
def get_default_config():

11
monasca_analytics/ldp/iptables_ldp.py
View File

@ -16,12 +16,13 @@
import logging
import schema
import voluptuous
import monasca_analytics.ingestor.iptables as ip_ing
import monasca_analytics.ldp.base as bt
from monasca_analytics.sml import svm_one_class
import monasca_analytics.util.spark_func as fn
from monasca_analytics.util import validation_utils as vu
logger = logging.getLogger(__name__)
@ -34,10 +35,10 @@ class IptablesLDP(bt.BaseLDP):
@staticmethod
def validate_config(_config):
return schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
}).validate(_config)
iptables_ldp_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.NoSpaceCharacter())
}, required=True)
return iptables_ldp_schema(_config)
@staticmethod
def get_default_config():

13
monasca_analytics/ldp/monasca_aggregate.py
View File

@ -15,12 +15,13 @@
# under the License.
import logging
import schema
import voluptuous
import monasca_analytics.ldp.base as bt
import monasca_analytics.ldp.monasca.helpers as helpers
import monasca_analytics.util.spark_func as fn
from monasca_analytics.util import validation_utils as vu
logger = logging.getLogger(__name__)
@ -36,12 +37,11 @@ class MonascaAggregateLDP(bt.BaseLDP):
@staticmethod
def validate_config(_config):
return schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
monasca_ag_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.NoSpaceCharacter()),
"params": {
"aggregation_period": int,
"aggregation_function": schema.Or(
"aggregation_function": voluptuous.Or(
"avg",
"max",
"sum",
@ -49,7 +49,8 @@ class MonascaAggregateLDP(bt.BaseLDP):
"cnt"
)
}
})
}, required=True)
return monasca_ag_schema(_config)
@staticmethod
def get_default_config():

13
monasca_analytics/ldp/monasca_combine.py
View File

@ -15,12 +15,13 @@
# under the License.
import logging
import schema
import voluptuous
import monasca_analytics.banana as banana
import monasca_analytics.ldp.base as bt
import monasca_analytics.ldp.monasca.helpers as helpers
import monasca_analytics.util.spark_func as fn
from monasca_analytics.util import validation_utils as vu
logger = logging.getLogger(__name__)
@ -140,15 +141,14 @@ class MonascaCombineLDP(bt.BaseLDP):
@staticmethod
def validate_config(_config):
schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
monasca_comb_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.NoSpaceCharacter()),
"params": {
"metric_name": basestring,
"combine_period": int,
"lambda": basestring,
"metric_names_binding": {
basestring: schema.Or(
basestring: voluptuous.Or(
"apache.net.kbytes_sec",
"apache.net.requests_sec",
"apache.performance.cpu_load_perc",
@ -211,7 +211,8 @@ class MonascaCombineLDP(bt.BaseLDP):
)
}
}
}).validate(_config)
}, required=True)
monasca_comb_schema(_config)
# Checks the expression and the environment
handle = banana.validate_expression(_config["params"]["lambda"])
banana.validate_name_binding(handle,

13
monasca_analytics/ldp/monasca_derivative.py
View File

@ -15,11 +15,12 @@
# under the License.
import logging
import schema
import voluptuous
import monasca_analytics.ldp.base as bt
import monasca_analytics.ldp.monasca.helpers as helpers
import monasca_analytics.util.spark_func as fn
from monasca_analytics.util import validation_utils as vu
logger = logging.getLogger(__name__)
@ -41,14 +42,14 @@ class MonascaDerivativeLDP(bt.BaseLDP):
@staticmethod
def validate_config(_config):
return schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
monasca_der_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.NoSpaceCharacter()),
"params": {
# Derivative period in multiple of batch interval
"derivative_period": int
}
})
}, required=True)
return monasca_der_schema(_config)
@staticmethod
def get_default_config():
@ -100,7 +101,7 @@ class MonascaDerivativeLDP(bt.BaseLDP):
timestamps = map(lambda m: m["metric"]["timestamp"], metric_values)
all_values = map(lambda m: m["metric"]["value"], metric_values)
# Sort values
all_values = [y for (x, y) in sorted(zip(timestamps, all_values))]
all_values = [y for (_, y) in sorted(zip(timestamps, all_values))]
timestamps = sorted(timestamps)
# Remove duplicates
last_timestamp = timestamps[0]

4
monasca_analytics/monanas.py
View File

@ -18,8 +18,8 @@ import logging
import os
import sys
import schema
from tornado import ioloop
import voluptuous
from monasca_analytics.exception import monanas as err
from monasca_analytics.spark import driver
@ -125,7 +125,7 @@ if __name__ == "__main__":
except ValueError as e:
raise err.MonanasMainError("`{0}` is not a \
valid json file.".format(sys.argv[1]))
except schema.SchemaError:
except voluptuous.Invalid:
raise err.MonanasMainError("`{0}` has an \
invalid schema.".format(sys.argv[1]))

19
monasca_analytics/sink/file.py
View File

@ -16,11 +16,12 @@
import json
import os.path as path
import schema
import tempfile
import time
import voluptuous
import monasca_analytics.sink.base as base
from monasca_analytics.util import validation_utils as vu
class FileSink(base.BaseSink):
@ -69,14 +70,10 @@ class FileSink(base.BaseSink):
@staticmethod
def validate_config(_config):
return schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
schema.Optional("params"): {
"path": schema.And(
basestring,
lambda i: path.exists(path.expanduser(i)) or
path.exists(path.dirname(path.expanduser(i)))
)
file_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.NoSpaceCharacter()),
voluptuous.Optional("params"): {
"path": voluptuous.And(basestring, vu.ExistingPath())
}
}).validate(_config)
}, required=True)
return file_schema(_config)

15
monasca_analytics/sink/iptables_sqlite.py
View File

@ -14,9 +14,10 @@
# License for the specific language governing permissions and limitations
# under the License.
import schema
import voluptuous
import monasca_analytics.sink.base_sqlite as base
from monasca_analytics.util import validation_utils as vu
class IptablesSQLiteSink(base.BaseSQLiteSink):
@ -42,9 +43,9 @@ class IptablesSQLiteSink(base.BaseSQLiteSink):
@staticmethod
def validate_config(_config):
return schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
schema.Optional("db_name"): schema.And(
basestring, lambda i: not any(c.isspace() for c in i)),
}).validate(_config)
iptables_sql_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.NoSpaceCharacter()),
voluptuous.Optional("db_name"): voluptuous.And(
basestring, vu.NoSpaceCharacter()),
}, required=True)
return iptables_sql_schema(_config)

27
monasca_analytics/sink/sink_config_validator.py
View File

@ -16,29 +16,22 @@
"""A list of functions for validating sink configs."""
import schema
import voluptuous
from monasca_analytics.config import const
from monasca_analytics.util import common_util as cu
from monasca_analytics.util import validation_utils as vu
def validate_kafka_sink_config(config):
"""Validates the KafkaSink configuration"""
available_sink_classes = \
cu.get_available_classes(const.SINKS)[const.SINKS]
available_sink_names = [Clazz.__name__ for Clazz in available_sink_classes]
config_schema = schema.Schema({
"module": schema.And(basestring,
lambda m: m in available_sink_names),
config_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.AvailableSink()),
"params": {
"host": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
"host": voluptuous.And(
basestring, vu.NoSpaceCharacter()),
"port": int,
"topic": schema.And(basestring,
lambda i: not any(c.isspace() for c in i))
"topic": voluptuous.And(
basestring, vu.NoSpaceCharacter())
}
})
return config_schema.validate(config)
}, required=True)
return config_schema(config)

11
monasca_analytics/sink/stdout_sink.py
View File

@ -14,9 +14,10 @@
# License for the specific language governing permissions and limitations
# under the License.
import schema
import voluptuous
from monasca_analytics.sink import base
from monasca_analytics.util import validation_utils as vu
class StdoutSink(base.BaseSink):
@ -34,7 +35,7 @@ class StdoutSink(base.BaseSink):
@staticmethod
def validate_config(_config):
return schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i))
}).validate(_config)
stdout_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.NoSpaceCharacter())
}, required=True)
stdout_schema(_config)

11
monasca_analytics/sml/lingam.py
View File

@ -18,10 +18,11 @@ import logging
import math
import numpy as np
import schema
from sklearn import decomposition
import voluptuous
from monasca_analytics.sml import base
from monasca_analytics.util import validation_utils as vu
logger = logging.getLogger(__name__)
@ -36,13 +37,13 @@ class LiNGAM(base.BaseSML):
@staticmethod
def validate_config(_config):
schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
lingam_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.NoSpaceCharacter()),
"params": {
"threshold": float
}
}).validate(_config)
}, required=True)
return lingam_schema(_config)
@staticmethod
def get_default_config():

11
monasca_analytics/sml/svm_one_class.py
View File

@ -17,10 +17,11 @@
import logging
import numpy as np
import schema
from sklearn import svm
import voluptuous
from monasca_analytics.sml import base
from monasca_analytics.util import validation_utils as vu
logger = logging.getLogger(__name__)
@ -38,10 +39,10 @@ class SvmOneClass(base.BaseSML):
@staticmethod
def validate_config(_config):
schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i))
}).validate(_config)
svm_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.NoSpaceCharacter())
}, required=True)
return svm_schema(_config)
@staticmethod
def get_default_config():

62
monasca_analytics/source/cloud_markov_chain.py
View File

@ -16,13 +16,14 @@
import logging
import schema
import voluptuous
from monasca_analytics.source.markov_chain import base
from monasca_analytics.source.markov_chain import events as ev
import monasca_analytics.source.markov_chain.prob_checks as pck
import monasca_analytics.source.markov_chain.state_check as dck
import monasca_analytics.source.markov_chain.transition as tr
from monasca_analytics.util import validation_utils as vu
logger = logging.getLogger(__name__)
@ -31,58 +32,57 @@ class CloudMarkovChainSource(base.MarkovChainSource):
@staticmethod
def validate_config(_config):
source_schema = schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
source_schema = voluptuous.Schema({
"module": voluptuous.And(
basestring, vu.NoSpaceCharacter()),
"params": {
"server_sleep_in_seconds": schema.And(float,
lambda v: 0 < v < 1)
"server_sleep_in_seconds": voluptuous.And(
float, voluptuous.Range(
min=0, max=1, min_included=False, max_included=False))
},
"transitions": {
"web_service": {
"run=>slow": {
schema.Use(int): schema.And(schema.Or(int, float),
lambda v: 0 <= v <= 1),
voluptuous.And(vu.NumericString()): voluptuous.And(
voluptuous.Or(int, float),
voluptuous.Range(min=0, max=1)),
},
"slow=>run": {
schema.Use(int): schema.And(schema.Or(int, float),
lambda v: 0 <= v <= 1),
voluptuous.And(vu.NumericString()): voluptuous.And(
voluptuous.Or(int, float),
voluptuous.Range(min=0, max=1)),
},
"stop=>run": schema.And(schema.Or(int, float),
lambda v: 0 <= v <= 1),
"stop=>run": voluptuous.And(
voluptuous.Or(int, float),
voluptuous.Range(min=0, max=1)),
},
"switch": {
"on=>off": schema.And(schema.Or(int, float),
lambda v: 0 <= v <= 1),
"off=>on": schema.And(schema.Or(int, float),
lambda v: 0 <= v <= 1),
"on=>off": voluptuous.And(voluptuous.Or(int, float),
voluptuous.Range(min=0, max=1)),
"off=>on": voluptuous.And(voluptuous.Or(int, float),
voluptuous.Range(min=0, max=1)),
},
"host": {
"on=>off": schema.And(schema.Or(int, float),
lambda v: 0 <= v <= 1),
"off=>on": schema.And(schema.Or(int, float),
lambda v: 0 <= v <= 1),
"on=>off": voluptuous.And(voluptuous.Or(int, float),
voluptuous.Range(min=0, max=1)),
"off=>on": voluptuous.And(voluptuous.Or(int, float),
voluptuous.Range(min=0, max=1)),
},
},
"triggers": {
"support": {
"get_called": {
schema.Use(int): schema.And(schema.Or(int, float),
lambda v: 0 <= v <= 1),
voluptuous.And(vu.NumericString()): voluptuous.And(
voluptuous.Or(int, float),
voluptuous.Range(min=0, max=1)),
},
},
},
"graph": {
schema.And(basestring,
lambda s: len(s.split(":")) == 2 and
s.split(":")[1] in
["host", "web_service", "switch"],
error="Key should be of the form 'anything:host' "
"where 'host' can be replaced "
"by 'web_service' or 'switch'."): [basestring]
voluptuous.And(basestring, vu.ValidMarkovGraph()): [basestring]
}
})
return source_schema.validate(_config)
}, required=True)
return source_schema(_config)
@staticmethod
def get_default_config():

17
monasca_analytics/source/iptables_markov_chain.py
View File

@ -16,13 +16,14 @@
import logging
import schema
import voluptuous
from monasca_analytics.source.markov_chain import base
from monasca_analytics.source.markov_chain import events
from monasca_analytics.source.markov_chain import prob_checks as pck
import monasca_analytics.source.markov_chain.state_check as dck
import monasca_analytics.source.markov_chain.transition as tr
from monasca_analytics.util import validation_utils as vu
logger = logging.getLogger(__name__)
@ -62,15 +63,15 @@ class IPTablesSource(base.MarkovChainSource):
@staticmethod
def validate_config(_config):
source_schema = schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
source_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.NoSpaceCharacter()),
"params": {
"server_sleep_in_seconds": schema.And(float,
lambda v: 0 < v < 1)
"server_sleep_in_seconds": voluptuous.And(
float, voluptuous.Range(
min=0, max=1, min_included=False, max_included=False))
}
})
return source_schema.validate(_config)
}, required=True)
return source_schema(_config)
@staticmethod
def get_default_config():

25
monasca_analytics/source/kafka.py
View File

@ -17,9 +17,10 @@
import logging
from pyspark.streaming import kafka
import schema
import voluptuous
from monasca_analytics.source import base
from monasca_analytics.util import validation_utils as vu
logger = logging.getLogger(__name__)
@ -29,25 +30,19 @@ class KafkaSource(base.BaseSource):
@staticmethod
def validate_config(_config):
source_schema = schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
source_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.NoSpaceCharacter()),
"params": {
"zk_host": schema.And(basestring,
lambda i: not any(c.isspace()
for c in i)),
"zk_host": voluptuous.And(basestring, vu.NoSpaceCharacter()),
"zk_port": int,
"group_id": schema.And(basestring,
lambda i: not any(c.isspace()
for c in i)),
"group_id": voluptuous.And(basestring, vu.NoSpaceCharacter()),
"topics": {
schema.And(basestring, lambda i: not any(c.isspace()
for c in i)):
schema.And(int, lambda p: p > 0)
voluptuous.And(basestring, vu.NoSpaceCharacter()):
voluptuous.And(int, voluptuous.Range(min=1))
}
}
})
return source_schema.validate(_config)
}, required=True)
return source_schema(_config)
@staticmethod
def get_default_config():

17
monasca_analytics/source/monasca_markov_chain.py
View File

@ -16,7 +16,7 @@
import logging
import random
import schema
import voluptuous
import monasca_analytics.source.markov_chain.base as base
@ -25,6 +25,8 @@ import monasca_analytics.source.markov_chain.prob_checks as pck
import monasca_analytics.source.markov_chain.state_check as dck
import monasca_analytics.source.markov_chain.transition as tr
import monasca_analytics.util.timestamp as tp
from monasca_analytics.util import validation_utils as vu
logger = logging.getLogger(__name__)
@ -33,14 +35,15 @@ class MonascaMarkovChainSource(base.MarkovChainSource):
@staticmethod
def validate_config(_config):
return schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
markov_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.NoSpaceCharacter()),
"params": {
"server_sleep_in_seconds": schema.And(float,
lambda v: 0 < v < 1)
"server_sleep_in_seconds": voluptuous.And(
float, voluptuous.Range(
min=0, max=1, min_included=False, max_included=False))
},
}).validate(_config)
}, required=True)
return markov_schema(_config)
@staticmethod
def get_default_config():

47
monasca_analytics/source/randoms.py
View File

@ -19,16 +19,16 @@ import json
import logging
import numpy as np
import random
import schema
import six
import SocketServer
import threading as th
import time
import uuid
import voluptuous
import monasca_analytics.exception.monanas as err
from monasca_analytics.source import base
from monasca_analytics.util import validation_utils as vu
logger = logging.getLogger(__name__)
@ -73,36 +73,35 @@ class RandomSource(base.BaseSource):
@staticmethod
def validate_config(_config):
source_schema = schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
source_schema = voluptuous.Schema({
"module": voluptuous.And(basestring, vu.NoSpaceCharacter()),
"params": {
"host": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
"host": voluptuous.And(basestring, vu.NoSpaceCharacter()),
"port": int,
"model": {
"name": schema.And(basestring,
lambda i: not any(c.isspace()
for c in i)),
"name": voluptuous.And(basestring, vu.NoSpaceCharacter()),
"params": {
"origin_types": schema.And([
"origin_types": voluptuous.And([
{
"origin_type": schema.And(
basestring,
lambda i: not any(c.isspace() for c in i)),
"weight": schema.And(schema.Or(int, float),
lambda w: w > 0.0)
"origin_type": voluptuous.And(
basestring, vu.NoSpaceCharacter()),
"weight": voluptuous.And(
voluptuous.Or(int, float),
voluptuous.Range(
min=0, min_included=False)),
}
], lambda o: len(o) > 0),
schema.Optional("key_causes"): dict
], vu.NotEmptyArray()),
voluptuous.Optional("key_causes"): dict
}
},
"alerts_per_burst": schema.And(int, lambda a: a > 0),
"idle_time_between_bursts": schema.And(schema.Or(int, float),
lambda i: i > 0)
"alerts_per_burst": voluptuous.And(
int, voluptuous.Range(min=1)),
"idle_time_between_bursts": voluptuous.And(
voluptuous.Or(int, float),
voluptuous.Range(min=0, min_included=False))
}
})
return source_schema.validate(_config)
}, required=True)
return source_schema(_config)
@staticmethod
def get_default_config():
@ -375,7 +374,7 @@ class MonanasTCPHandler(SocketServer.BaseRequestHandler):
validated_alert = self.server.validate(alert)
self.request.send(json.dumps(validated_alert) + "\n")
accumulated_alerts += 1
except schema.SchemaError:
except voluptuous.Invalid:
logger.warn("Invalid schema for generated alerts.")
time.sleep(self.server.generate_idle_time_between_bursts)

89
monasca_analytics/util/validation_utils.py Normal file
View File

@ -0,0 +1,89 @@
#!/usr/bin/env python
# Copyright (c) 2016 Hewlett Packard Enterprise Development Company, L.P.
#
# Licensed 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.path as path
import re
import voluptuous
from monasca_analytics.config import const
from monasca_analytics.util import common_util as cu
def AvailableSink(msg=None):
available_sink_classes = cu.get_available_classes(const.SINKS)[const.SINKS]
available_sink_names = [Clazz.__name__ for Clazz in available_sink_classes]
def f(v):
if str(v) in available_sink_names:
return str(v)
else:
raise voluptuous.Invalid(msg or ("Invalid Sink Name: " + str(v)))
return f
def NoSpaceCharacter(msg=None):
def f(v):
if not any(c.isspace() for c in str(v)):
return str(v)
else:
raise voluptuous.Invalid(msg or (
"White space not allowed in: " + str(v)))
return f
def ExistingPath(msg=None):
def f(v):
if path.exists(path.expanduser(str(v))) or\
path.exists(path.dirname(path.expanduser(str(v)))):
return str(v)
else:
raise voluptuous.Invalid(msg or (
"Path does not exist: " + str(v)))
return f
def NumericString(msg=None):
def f(v):
if re.match("[0-9]+", str(v)):
return str(v)
else:
raise voluptuous.Invalid(msg or (
"string does not represent a number"))
return f
def ValidMarkovGraph(msg=None):
def f(v):
if len(str(v).split(":")) == 2 and str(v).split(":")[1] in\
["host", "web_service", "switch"]:
return str(v)
else:
raise voluptuous.Invalid(msg or (
"Key should be of the form 'anything:host' "
"where 'host' can be replaced "
"by 'web_service' or 'switch'."))
return f
def NotEmptyArray(msg=None):
def f(v):
if len(v) > 0:
return v
else:
raise voluptuous.Invalid(msg or ("empty array: " + str(v)))
return f

13
monasca_analytics/voter/pick_index.py
View File

@ -16,7 +16,7 @@
import logging
import schema
import voluptuous
from monasca_analytics.voter import base
@ -32,13 +32,14 @@ class PickIndexVoter(base.BaseVoter):
@staticmethod
def validate_config(_config):
schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
pick_schema = voluptuous.Schema({
"module": voluptuous.And(
basestring, lambda i: not any(c.isspace() for c in i)),
"params": {
"index": schema.And(int, lambda i: i >= 0)
"index": voluptuous.And(int, lambda i: i >= 0)
}
}).validate(_config)
}, required=True)
return pick_schema(_config)
@staticmethod
def get_default_config():

4
monasca_analytics/web_service/request_handler.py
View File

@ -18,8 +18,8 @@ import json
import logging
import sys
import schema
from tornado import web
import voluptuous
import monasca_analytics.exception.monanas as err
from monasca_analytics.web_service import web_service_model
@ -47,7 +47,7 @@ class MonanasHandler(web.RequestHandler):
body = json.loads(self.request.body)
validated_body = getattr(web_service_model, "action_model")(body)
getattr(self._monanas, validated_body["action"])()
except (AttributeError, schema.SchemaError, ValueError):
except (AttributeError, voluptuous.Invalid, ValueError):
self.set_status(400, "The request body was malformed.")
except (err.MonanasBindSourcesError,
err.MonanasAlreadyStartedStreaming,

10
monasca_analytics/web_service/web_service_model.py
View File

@ -16,13 +16,13 @@
"""A list of functions to validate web_service models."""
import schema
import voluptuous
def action_model(value):
"""Validates the data against action_model schema."""
action_model_schema = schema.Schema({
"action": schema.And(basestring, lambda o: not o.startswith("_"))
})
action_model_schema = voluptuous.Schema({
"action": voluptuous.And(basestring, lambda o: not o.startswith("_"))
}, required=True)
return action_model_schema.validate(value)
return action_model_schema(value)

2
requirements.txt
View File

@ -2,9 +2,9 @@ docopt
findspark
libpgm
numpy
schema
scipy
tornado
sklearn
kafka-python
pyparsing
voluptuous>=0.8.9 # BSD License

4
test/dsl/test_dsl.py
View File

@ -20,7 +20,7 @@ import logging.config
import os
import unittest
import schema
import voluptuous
from monasca_analytics.config import const
from monasca_analytics.dsl.dsl import MonanasDSL
@ -126,7 +126,7 @@ class TestMonanasDSL(unittest.TestCase):
def test_add_component_wrong_config(self):
del (self.testing_new_source_config["params"]["zk_port"])
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
self.dsl.add_component,
self.testing_new_source_config)
self.assertEqual(self.original_config, self.dsl._config)

17
test/sink/test_base_sqlite.py
View File

@ -20,7 +20,7 @@ import unittest
import cPickle
import numpy as np
import schema
import voluptuous
from monasca_analytics.sink import base_sqlite as bsql
from test.mocks import spark_mocks
@ -47,12 +47,13 @@ class BaseSQLiteSinkDummyExtension(bsql.BaseSQLiteSink):
@staticmethod
def validate_config(_config):
return schema.Schema({
"module": schema.And(basestring,
lambda i: not any(c.isspace() for c in i)),
schema.Optional("db_name"): schema.And(
base_schema = voluptuous.Schema({
"module": voluptuous.And(
basestring, lambda i: not any(c.isspace() for c in i)),
}).validate(_config)
voluptuous.Optional("db_name"): voluptuous.And(
basestring, lambda i: not any(c.isspace() for c in i)),
}, required=True)
return base_schema(_config)
class TestSQLiteSink(unittest.TestCase):
@ -105,12 +106,12 @@ class TestSQLiteSink(unittest.TestCase):
def test_validate_config_no_module(self):
conf = {"db_name": "mySQLite.db"}
self.assertRaises(schema.SchemaError, self.snk.validate_config, conf)
self.assertRaises(voluptuous.Invalid, self.snk.validate_config, conf)
def test_validate_config_extra_param(self):
conf = {"module": "BaseSQLiteSinkDummyExtension",
"infiltrated": "I shouldn't be here"}
self.assertRaises(schema.SchemaError, self.snk.validate_config, conf)
self.assertRaises(voluptuous.Invalid, self.snk.validate_config, conf)
def test_get_db_name(self):
conf = {"db_name": "mySQLite.db"}

21
test/sink/test_sink_config_validator.py
View File

@ -14,8 +14,8 @@
# License for the specific language governing permissions and limitations
# under the License.
import schema
import unittest
import voluptuous
import monasca_analytics.sink.sink_config_validator as validator
@ -34,24 +34,31 @@ class SinkConfigValidatorTest(unittest.TestCase):
}
}
def test_validate_kafka_sink_config(self):
def test_validate_kafka_sink_valid_config(self):
try:
kafka(self._valid_config)
except schema.SchemaError as e:
except voluptuous.Invalid as e:
self.fail(e.__str__())
def test_validate_kafka_sink_invalid_module(self):
invalid_config = self._valid_config
invalid_config["module"] = "invalid_module"
self.assertRaises(schema.SchemaError, kafka, invalid_config)
self.assertRaises(voluptuous.Invalid, kafka, invalid_config)
def test_validate_kafka_sink_invalid_host(self):
invalid_config = self._valid_config
invalid_config["params"]["host"] = "invalid host"
self.assertRaises(schema.SchemaError, kafka, invalid_config)
self.assertRaises(voluptuous.Invalid, kafka, invalid_config)
def test_validate_kafka_sink_invalid_port(self):
invalid_config = self._valid_config
invalid_config["params"]["port"] = "invalid_port"
self.assertRaises(schema.SchemaError, kafka, invalid_config)
self.assertRaises(voluptuous.Invalid, kafka, invalid_config)
def test_validate_kafka_sink_invalid_topic(self):
invalid_config = self._valid_config
invalid_config["params"]["topic"] = "invalid topic"
self.assertRaises(schema.SchemaError, kafka, invalid_config)
self.assertRaises(voluptuous.Invalid, kafka, invalid_config)
def tearDown(self):
unittest.TestCase.tearDown(self)

8
test/source/test_iptables_markov_chain.py
View File

@ -19,7 +19,7 @@ import logging.config
import os
import unittest
import schema
import voluptuous
from monasca_analytics.source import iptables_markov_chain
@ -71,17 +71,17 @@ class TestIPTablesSource(unittest.TestCase):
def test_validate_config_extra_param(self):
self.assertRaises(
schema.SchemaError, self.ips.validate_config,
voluptuous.Invalid, self.ips.validate_config,
self.config_extra_param)
def test_validate_config_missing_param(self):
self.assertRaises(
schema.SchemaError, self.ips.validate_config,
voluptuous.Invalid, self.ips.validate_config,
self.config_missing_param)
def test_validate_config_wrong_type(self):
self.assertRaises(
schema.SchemaError,
voluptuous.Invalid,
self.ips.validate_config,
self.config_wrong_type)

10
test/source/test_kafka.py
View File

@ -19,7 +19,7 @@ import logging.config
import os
import unittest
import schema
import voluptuous
from monasca_analytics.source import kafka
from test.mocks import spark_mocks
@ -88,25 +88,25 @@ class KafkaSourceTest(unittest.TestCase):
def test_validate_config_extra_param(self):
self.assertRaises(
schema.SchemaError,
voluptuous.Invalid,
self.ks.validate_config,
self.config_extra_param)
def test_validate_config_missing_dir(self):
self.assertRaises(
schema.SchemaError,
voluptuous.Invalid,
self.ks.validate_config,
self.config_missing_param)
def test_validate_config_wrong_type(self):
self.assertRaises(
schema.SchemaError,
voluptuous.Invalid,
self.ks.validate_config,
self.config_wrong_type)
def test_validate_config_missing_params(self):
self.assertRaises(
schema.SchemaError,
voluptuous.Invalid,
self.ks.validate_config,
self.config_missing_params)

8
test/source/test_markov_chain.py
View File

@ -19,7 +19,7 @@ import logging.config
import os
import unittest
import schema
import voluptuous
from monasca_analytics.source import cloud_markov_chain as cloud
from test.mocks import spark_mocks
@ -115,19 +115,19 @@ class MarkovChainSourceTest(unittest.TestCase):
def test_validate_config_extra_param(self):
self.assertRaises(
schema.SchemaError,
voluptuous.Invalid,
self.mcs.validate_config,
self.config_extra_param)
def test_validate_config_missing_param(self):
self.assertRaises(
schema.SchemaError,
voluptuous.Invalid,
self.mcs.validate_config,
self.config_missing_param)
def test_validate_config_wrong_type(self):
self.assertRaises(
schema.SchemaError,
voluptuous.Invalid,
self.mcs.validate_config,
self.config_wrong_type)

44
test/util/test_config_model.py
View File

@ -19,7 +19,7 @@ import logging.config
import os
import unittest
import schema
import voluptuous
from monasca_analytics.config import validation
from monasca_analytics.util import common_util
@ -56,97 +56,97 @@ class TestConfigModel(unittest.TestCase):
def test_validate_config_missing_key(self):
for key in self.comp_types:
del self.config[key]
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
self.config = self.get_config()
def test_validate_config_extra_key(self):
self.config = self.get_config()
self.config["infiltrated"] = "I should not exist"
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
self.config = self.get_config()
def test_validate_config_missing_spark_key(self):
for key in self.config["spark_config"].keys():
del self.config["spark_config"][key]
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
self.config = self.get_config()
del self.config["spark_config"]["streaming"]["batch_interval"]
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
def test_validate_config_missing_server_key(self):
del self.config["server"]["port"]
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
self.config = self.get_config()
del self.config["server"]["debug"]
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
def test_validate_config_spark_wrong_format(self):
self.config["spark_config"]["streaming"][
"batch_interval"] = "I should not be a string"
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
self.config = self.get_config()
self.config["spark_config"]["appName"] = 123
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
def test_validate_config_server_wrong_format(self):
self.config["server"]["port"] = "I should be an int"
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
self.config = self.get_config()
self.config["server"]["debug"] = 52
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
def test_validate_config_spark_extra_parameters(self):
self.config["spark_config"]["infiltrated"] = "I should not exist"
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
self.config = self.get_config()
self.config["spark_config"]["streaming"][
"infiltrated"] = "I should not exist"
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
def test_validate_config_server_extra_parameters(self):
self.config["server"]["infiltrated"] = "I should not exist"
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
def test_validate_config_wrong_format_components(self):
for key in self.comp_types:
self.config[key] = ["I", "should", "be", "a", "dictionary"]
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
self.config = self.get_config()
for comp_id in self.config[key].keys():
self.config[key][comp_id] = ["I", "should", "be", "a", "dict"]
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
self.config = self.get_config()
def test_validate_config_wrong_format_connections(self):
self.config["connections"] = ["I", "should", "be", "a", "dictionary"]
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
self.config = self.get_config()
for comp_id in self.config["connections"].keys():
self.config["connections"][comp_id] = {"I": "should",
"be": "a list"}
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
self.config = self.get_config()
def test_validate_connections_data_models(self):
self.config["connections"]["mod1"] = ["src1"]
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
def test_validate_connections_wrong_dest(self):
@ -173,18 +173,18 @@ class TestConfigModel(unittest.TestCase):
self.config["connections"][from_id] = [dst_id]
logger.debug("checking wrong connection: " +
from_id + " --> " + dst_id)
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
self.config = self.get_config()
def test_validate_connections_inexisteng_source(self):
self.config["connections"]["inex"] = ["sin2"]
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
self.config = self.get_config()
def test_validate_connections_inexisteng_dest(self):
self.config["connections"]["src1"] = ["inex"]
self.assertRaises(schema.SchemaError,
self.assertRaises(voluptuous.Invalid,
validation.validate_config, self.config)
self.config = self.get_config()