blob: 01eada539106b10d44e144d3f183787376aad244 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
|
#!/usr/bin/python
# http://www.apache.org/licenses/LICENSE-2.0
"""Instantiation class."""
import logging
import logging.config
from onapsdk.clamp.loop_instance import LoopInstance
from onapsdk.configuration import settings
from onaptests.utils.exceptions import OnapTestException
class InstantiateLoop():
"""class instantiating a closed loop in clamp."""
def __init__(self, template: str, loop_name: str, operational_policies: list):
self.template=template
self.loop_name=loop_name
self.operational_policies=operational_policies
self._logger: logging.Logger = logging.getLogger("")
logging.config.dictConfig(settings.LOG_CONFIG)
def add_policies(self, loop: LoopInstance) -> None:
"""add necessary wanted operational policies."""
for policy in self.operational_policies:
self._logger.info("******** ADD OPERATIONAL POLICY %s *******", policy["name"])
added = loop.add_operational_policy(policy_type=policy["policy_type"],
policy_version=policy["policy_version"])
if not added:
self._logger.error("an error occured while adding an operational policy")
self._logger.info("ADD OPERATION SUCCESSFULY DONE")
def configure_policies(self, loop: LoopInstance) -> None:
"""Configure all policies."""
self._logger.info("******** UPDATE MICROSERVICE POLICY *******")
loop._update_loop_details()
loop.update_microservice_policy()
self._logger.info("******** UPDATE OPERATIONAL POLICIES CONFIG *******")
for policy in self.operational_policies:
#loop.add_op_policy_config(loop.LoopInstance.__dict__[policy["config_function"]])
#possible configurations for the moment
if policy["name"] == "MinMax":
loop.add_op_policy_config(loop.add_minmax_config)
if policy["name"] == "Drools":
loop.add_op_policy_config(loop.add_drools_conf)
if policy["name"] == "FrequencyLimiter":
loop.add_op_policy_config(loop.add_frequency_limiter)
self._logger.info("Policies are well configured")
def deploy(self, loop: LoopInstance) -> None:
"""Deploy closed loop."""
self._logger.info("******** SUBMIT POLICIES TO PE *******")
submit = loop.act_on_loop_policy(loop.submit)
self._logger.info("******** CHECK POLICIES SUBMITION *******")
if submit :
self._logger.info("Policies successfully submited to PE")
else:
self._logger.error("An error occured while submitting the loop instance")
raise OnapTestException.PolicyException
self._logger.info("******** DEPLOY LOOP INSTANCE *******")
deploy = loop.deploy_microservice_to_dcae()
if deploy:
self._logger.info("Loop instance %s successfully deployed on DCAE !!", self.loop_name)
else:
self._logger.error("An error occured while deploying the loop instance")
raise OnapTestException.DcaeException
def instantiate_loop(self):
"""Instantiate the control loop."""
loop = LoopInstance(template=self.template,
name=self.loop_name,
details={})
details = loop.create()
if details:
self._logger.info("Loop instance %s successfully created !!", self.loop_name)
else:
self._logger.error("An error occured while creating the loop instance")
self.add_policies(loop=loop)
self.configure_policies(loop=loop)
self.deploy(loop=loop)
loop.details = loop._update_loop_details()
return loop
|
