Merge from OSM SO master
[osm/SO.git] / common / python / rift / mano / config_agent / config.py
1
2 #
3 # Copyright 2016 RIFT.IO Inc
4 #
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
8 #
9 # http://www.apache.org/licenses/LICENSE-2.0
10 #
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16 #
17
18 import asyncio
19 import rw_peas
20
21 import gi
22 gi.require_version('RwDts', '1.0')
23 import rift.tasklets
24 from rift.mano.utils.project import get_add_delete_update_cfgs
25
26 from gi.repository import (
27 RwcalYang as rwcal,
28 RwDts as rwdts,
29 RwConfigAgentYang as rwcfg_agent,
30 ProtobufC,
31 )
32
33 class ConfigAccountNotFound(Exception):
34 pass
35
36 class ConfigAccountError(Exception):
37 pass
38
39
40 class ConfigAgentCallbacks(object):
41 def __init__(self,
42 on_add_apply=None, on_add_prepare=None,
43 on_delete_apply=None, on_delete_prepare=None):
44
45 @asyncio.coroutine
46 def prepare_noop(*args, **kwargs):
47 pass
48
49 def apply_noop(*args, **kwargs):
50 pass
51
52 self.on_add_apply = on_add_apply
53 self.on_add_prepare = on_add_prepare
54 self.on_delete_apply = on_delete_apply
55 self.on_delete_prepare = on_delete_prepare
56
57 for f in ('on_add_apply', 'on_delete_apply'):
58 ref = getattr(self, f)
59 if ref is None:
60 setattr(self, f, apply_noop)
61 continue
62
63 if asyncio.iscoroutinefunction(ref):
64 raise ValueError('%s cannot be a coroutine' % (f,))
65
66 for f in ('on_add_prepare', 'on_delete_prepare'):
67 ref = getattr(self, f)
68 if ref is None:
69 setattr(self, f, prepare_noop)
70 continue
71
72 if not asyncio.iscoroutinefunction(ref):
73 raise ValueError("%s must be a coroutine" % f)
74
75
76 class ConfigAgentSubscriber(object):
77 XPATH = "C,/rw-config-agent:config-agent/account"
78
79 def __init__(self, dts, log, project, config_callbacks):
80 self._dts = dts
81 self._log = log
82 self._project = project
83 self._reg = None
84
85 self.accounts = {}
86
87 self._config_callbacks = config_callbacks
88
89 def add_account(self, account_msg):
90 self._log.info("adding config account: {}".format(account_msg))
91
92 self.accounts[account_msg.name] = account_msg
93
94 self._config_callbacks.on_add_apply(account_msg)
95
96 def delete_account(self, account_msg):
97 self._log.info("deleting config account: {}".format(account_msg.name))
98 del self.accounts[account_msg.name]
99
100 self._config_callbacks.on_delete_apply(account_msg)
101
102 def update_account(self, account_msg):
103 """ Update an existing config-agent account
104
105 In order to simplify update, turn an update into a delete followed by
106 an add. The drawback to this approach is that we will not support
107 updates of an "in-use" config-agent account, but this seems like a
108 reasonable trade-off.
109
110 Arguments:
111 account_msg - The config-agent account config message
112 """
113
114 self._log.info("updating config-agent account: {}".format(account_msg))
115 self.delete_account(account_msg)
116 self.add_account(account_msg)
117
118 def deregister(self):
119 self._log.debug("De-register config agent handler for project {}".
120 format(self._project.name))
121 if self._reg:
122 self._reg.deregister()
123 self._reg = None
124
125 def register(self):
126 def apply_config(dts, acg, xact, action, _):
127 self._log.debug("Got config account apply config (xact: %s) (action: %s)", xact, action)
128
129 if xact.xact is None:
130 # When RIFT first comes up, an INSTALL is called with the current config
131 # Since confd doesn't actally persist data this never has any data so
132 # skip this for now.
133 self._log.debug("No xact handle. Skipping apply config")
134 return
135
136 add_cfgs, delete_cfgs, update_cfgs = get_add_delete_update_cfgs(
137 dts_member_reg=self._reg,
138 xact=xact,
139 key_name="name",
140 )
141
142 # Handle Deletes
143 for cfg in delete_cfgs:
144 self.delete_account(cfg)
145
146 # Handle Adds
147 for cfg in add_cfgs:
148 self.add_account(cfg)
149
150 # Handle Updates
151 for cfg in update_cfgs:
152 self.update_account(cfg)
153
154 @asyncio.coroutine
155 def on_prepare(dts, acg, xact, xact_info, ks_path, msg, scratch):
156 """ Prepare callback from DTS for Config Account """
157
158 action = xact_info.handle.query_action
159 self._log.debug("Config account on_prepare config received (action: %s): %s",
160 xact_info.handle.query_action, msg)
161
162 if action in [rwdts.QueryAction.CREATE, rwdts.QueryAction.UPDATE]:
163 # If the account already exists, then this is an update.
164 if msg.name in self.accounts:
165 self._log.debug("Config account already exists. Invoking on_prepare update request")
166 if msg.has_field("account_type"):
167 raise ConfigAccountError("Cannot change config's account-type")
168
169 # Since updates are handled by a delete followed by an add, invoke the
170 # delete prepare callbacks to give clients an opportunity to reject.
171 yield from self._config_callbacks.on_delete_prepare(msg.name)
172
173 else:
174 self._log.debug("Config account does not already exist. Invoking on_prepare add request")
175 if not msg.has_field('account_type'):
176 raise ConfigAccountError("New Config account must contain account_type field.")
177
178 account = msg
179 yield from self._config_callbacks.on_add_prepare(account)
180
181 elif action == rwdts.QueryAction.DELETE:
182 # Check if the entire cloud account got deleted
183 fref = ProtobufC.FieldReference.alloc()
184 fref.goto_whole_message(msg.to_pbcm())
185 if fref.is_field_deleted():
186 yield from self._config_callbacks.on_delete_prepare(msg.name)
187 else:
188 self._log.error("Deleting individual fields for config account not supported")
189 xact_info.respond_xpath(rwdts.XactRspCode.NACK)
190 return
191
192 else:
193 self._log.error("Action (%s) NOT SUPPORTED", action)
194 xact_info.respond_xpath(rwdts.XactRspCode.NACK)
195
196 xact_info.respond_xpath(rwdts.XactRspCode.ACK)
197
198
199 acg_handler = rift.tasklets.AppConfGroup.Handler(
200 on_apply=apply_config,
201 )
202
203 with self._dts.appconf_group_create(acg_handler) as acg:
204 xpath = self._project.add_project(ConfigAgentSubscriber.XPATH)
205 self._log.debug("Registering for Config Account config using xpath: %s",
206 xpath)
207 self._reg = acg.register(
208 xpath=xpath,
209 flags=rwdts.Flag.SUBSCRIBER,
210 on_prepare=on_prepare,
211 )