Infra
/
risotto
9
0
Fork 1
Code Issues 1 Releases Wiki Activity

Merge branch 'develop' into dist/risotto/risotto-2.8.0/develop

This commit is contained in:
Emmanuel Garette 2021-04-24 17:11:14 +02:00
parent 52209a5ebd ed51bc483d
commit 2a98575790
4 changed files with 173 additions and 178 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
sql/risotto.sql
View File

@ -1,12 +1,12 @@
CREATE TABLE RisottoLog( CREATE TABLE RisottoLog(
LogId SERIAL PRIMARY KEY, LogId SERIAL PRIMARY KEY,
ContextId INTEGER,
Msg VARCHAR(255) NOT NULL, Msg VARCHAR(255) NOT NULL,
URI VARCHAR(255), URI VARCHAR(255),
URIS VARCHAR(255), URIS VARCHAR(255),
UserLogin VARCHAR(100) NOT NULL, UserLogin VARCHAR(100) NOT NULL,
Level VARCHAR(10) NOT NULL, Level VARCHAR(10) NOT NULL,
ContextId INTEGER, Kwargs JSON,
Data JSON,
Returns JSON, Returns JSON,
StartDate timestamp DEFAULT current_timestamp, StartDate timestamp DEFAULT current_timestamp,
StopDate timestamp StopDate timestamp

208
src/risotto/dispatcher.py
View File

@ -126,60 +126,59 @@ class CallDispatcher:
decoder=loads, decoder=loads,
schema='pg_catalog' schema='pg_catalog'
) )
async with log_connection.transaction(): risotto_context.log_connection = log_connection
try: try:
risotto_context.log_connection = log_connection async with self.pool.acquire() as connection:
async with self.pool.acquire() as connection: await connection.set_type_codec(
await connection.set_type_codec( 'json',
'json', encoder=dumps,
encoder=dumps, decoder=loads,
decoder=loads, schema='pg_catalog'
schema='pg_catalog' )
) risotto_context.connection = connection
risotto_context.connection = connection async with connection.transaction():
async with connection.transaction(): try:
try: await log.start(risotto_context,
await log.start(risotto_context, kwargs,
kwargs, info_msg,
info_msg, )
) await self.check_message_type(risotto_context,
await self.check_message_type(risotto_context, kwargs,
kwargs, )
) config_arguments = await self.load_kwargs_to_config(risotto_context,
config_arguments = await self.load_kwargs_to_config(risotto_context, f'{version}.{message}',
f'{version}.{message}', kwargs,
kwargs, check_role,
check_role, internal,
internal, )
) ret = await self.launch(risotto_context,
ret = await self.launch(risotto_context, kwargs,
kwargs, config_arguments,
config_arguments, function_obj,
function_obj, )
) # log the success
# log the success await log.success(risotto_context,
await log.success(risotto_context, ret,
ret, )
) if not internal and isinstance(ret, dict):
if not internal and isinstance(ret, dict): ret['context_id'] = risotto_context.context_id
ret['context_id'] = risotto_context.context_id except CallError as err:
except CallError as err: if get_config()['global']['debug']:
if get_config()['global']['debug']: print_exc()
print_exc() await log.failed(risotto_context,
await log.failed(risotto_context, str(err),
str(err), )
) raise err from err
raise err from err except CallError as err:
except CallError as err: error = err
error = err except Exception as err:
except Exception as err: # if there is a problem with arguments, just send an error and do nothing
# if there is a problem with arguments, just send an error and do nothing if get_config()['global']['debug']:
if get_config()['global']['debug']: print_exc()
print_exc() await log.failed(risotto_context,
await log.failed(risotto_context, str(err),
str(err), )
) error = err
error = err
if error: if error:
if not internal: if not internal:
err = CallError(str(error)) err = CallError(str(error))
@ -260,14 +259,14 @@ class PublishDispatcher:
False, False,
False, False,
) )
async with self.pool.acquire() as log_connection: for function_obj in self.messages[version][message]['functions']:
await log_connection.set_type_codec( async with self.pool.acquire() as log_connection:
'json', await log_connection.set_type_codec(
encoder=dumps, 'json',
decoder=loads, encoder=dumps,
schema='pg_catalog' decoder=loads,
) schema='pg_catalog'
async with log_connection.transaction(): )
risotto_context.log_connection = log_connection risotto_context.log_connection = log_connection
async with self.pool.acquire() as connection: async with self.pool.acquire() as connection:
await connection.set_type_codec( await connection.set_type_codec(
@ -277,50 +276,49 @@ class PublishDispatcher:
schema='pg_catalog' schema='pg_catalog'
) )
risotto_context.connection = connection risotto_context.connection = connection
for function_obj in self.messages[version][message]['functions']: function_name = function_obj['function'].__name__
function_name = function_obj['function'].__name__ info_msg = _(f"call function {function_obj['full_module_name']}.{function_name}")
info_msg = _(f"call function {function_obj['full_module_name']}.{function_name}") try:
try: async with connection.transaction():
try:
await log.start(risotto_context,
kwargs,
info_msg,
)
await self.check_message_type(risotto_context,
kwargs,
)
await self.launch(risotto_context,
kwargs,
config_arguments,
function_obj,
)
# log the success
await log.success(risotto_context)
except CallError as err:
if get_config()['global']['debug']:
print_exc()
await log.failed(risotto_context,
str(err),
)
except CallError:
pass
except Exception as err:
# if there is a problem with arguments, log and do nothing
if get_config()['global']['debug']:
print_exc()
async with self.pool.acquire() as connection:
await connection.set_type_codec(
'json',
encoder=dumps,
decoder=loads,
schema='pg_catalog'
)
risotto_context.connection = connection
async with connection.transaction(): async with connection.transaction():
try: await log.failed(risotto_context,
await log.start(risotto_context, str(err),
kwargs, )
info_msg,
)
await self.check_message_type(risotto_context,
kwargs,
)
await self.launch(risotto_context,
kwargs,
config_arguments,
function_obj,
)
# log the success
await log.success(risotto_context)
except CallError as err:
if get_config()['global']['debug']:
print_exc()
await log.failed(risotto_context,
str(err),
)
except CallError:
pass
except Exception as err:
# if there is a problem with arguments, log and do nothing
if get_config()['global']['debug']:
print_exc()
async with self.pool.acquire() as connection:
await connection.set_type_codec(
'json',
encoder=dumps,
decoder=loads,
schema='pg_catalog'
)
risotto_context.connection = connection
async with connection.transaction():
await log.failed(risotto_context,
str(err),
)
class Dispatcher(register.RegisterDispatcher, class Dispatcher(register.RegisterDispatcher,

80
src/risotto/logger.py
View File

@ -13,24 +13,24 @@ class Logger:
""" """
async def insert(self, async def insert(self,
msg: str, msg: str,
uri: str,
uris: str,
risotto_context: Context, risotto_context: Context,
level: str, level: str,
data: Any=None, kwargs: Any=None,
start: bool=False, start: bool=False,
) -> None: ) -> None:
uri = self._get_last_uri(risotto_context)
uris = " ".join(risotto_context.paths)
insert = 'INSERT INTO RisottoLog(Msg, URI, URIS, UserLogin, Level' insert = 'INSERT INTO RisottoLog(Msg, URI, URIS, UserLogin, Level'
values = 'VALUES($1,$2,$3,$4,$5' values = 'VALUES($1,$2,$3,$4,$5'
args = [msg, uri, uris, risotto_context.username, level] args = [msg, uri, uris, risotto_context.username, level]
if data: if kwargs:
insert += ', Data' insert += ', Kwargs'
values += ',$6' values += ',$6'
args.append(dumps(data)) args.append(dumps(kwargs))
context_id = risotto_context.context_id context_id = risotto_context.context_id
if context_id is not None: if context_id is not None:
insert += ', ContextId' insert += ', ContextId'
if data: if kwargs:
values += ',$7' values += ',$7'
else: else:
values += ',$6' values += ',$6'
@ -38,7 +38,8 @@ class Logger:
sql = insert + ') ' + values + ') RETURNING LogId' sql = insert + ') ' + values + ') RETURNING LogId'
try: try:
log_id = await risotto_context.log_connection.fetchval(sql, *args) async with risotto_context.log_connection.transaction():
log_id = await risotto_context.log_connection.fetchval(sql, *args)
if context_id is None and start: if context_id is None and start:
risotto_context.context_id = log_id risotto_context.context_id = log_id
if start: if start:
@ -51,7 +52,7 @@ class Logger:
context_id: int, context_id: int,
uri: Optional[str], uri: Optional[str],
) -> list: ) -> list:
sql = '''SELECT Msg as msg, URI as uri_name, URIS as uris, UserLogin as user_login, Level as level, Data as data, StartDate as start_date, StopDate as stop_date sql = '''SELECT Msg as msg, URI as uri_name, URIS as uris, UserLogin as user_login, Level as level, Kwargs as kwargs, Returns as returns, StartDate as start_date, StopDate as stop_date
FROM RisottoLog FROM RisottoLog
WHERE UserLogin = $1 AND (LogId = $2 OR ContextId = $2) WHERE UserLogin = $1 AND (LogId = $2 OR ContextId = $2)
''' '''
@ -60,20 +61,23 @@ class Logger:
sql += ' AND URI = $3' sql += ' AND URI = $3'
args.append(uri) args.append(uri)
ret = [] ret = []
for row in await risotto_context.log_connection.fetch(*args): async with risotto_context.log_connection.transaction():
d = {} for row in await risotto_context.log_connection.fetch(*args):
for key, value in row.items(): d = {}
if key == 'data': for key, value in row.items():
if isinstance(value, dict): if key in ['kwargs', 'returns']:
pass if isinstance(value, dict):
elif not value: pass
value = {} elif not value:
else: value = {}
value = loads(value) else:
elif key in ['start_date', 'stop_date']: value = loads(value)
value = str(value) if key == 'uris':
d[key] = value value = value.split(' ')
ret.append(d) elif key in ['start_date', 'stop_date']:
value = str(value)
d[key] = value
ret.append(d)
return ret return ret
def _get_last_uri(self, def _get_last_uri(self,
@ -111,8 +115,6 @@ class Logger:
paths_msg = self._get_message_paths(risotto_context) paths_msg = self._get_message_paths(risotto_context)
print(_(f'{risotto_context.username}: ERROR: {error} ({paths_msg} with arguments "{arguments}": {msg})')) print(_(f'{risotto_context.username}: ERROR: {error} ({paths_msg} with arguments "{arguments}": {msg})'))
await self.insert(msg, await self.insert(msg,
self._get_last_uri(risotto_context),
paths_msg,
risotto_context, risotto_context,
'Error', 'Error',
arguments, arguments,
@ -129,8 +131,6 @@ class Logger:
if get_config()['global']['debug']: if get_config()['global']['debug']:
print(_(f'{risotto_context.username}: INFO:{paths_msg}: {msg}')) print(_(f'{risotto_context.username}: INFO:{paths_msg}: {msg}'))
await self.insert(msg, await self.insert(msg,
self._get_last_uri(risotto_context),
paths_msg,
risotto_context, risotto_context,
'Info', 'Info',
arguments, arguments,
@ -149,8 +149,6 @@ class Logger:
context = '' context = ''
print(_(f'{risotto_context.username}: START{context}:{paths_msg}: {msg}')) print(_(f'{risotto_context.username}: START{context}:{paths_msg}: {msg}'))
await self.insert(msg, await self.insert(msg,
self._get_last_uri(risotto_context),
paths_msg,
risotto_context, risotto_context,
'Start', 'Start',
arguments, arguments,
@ -175,10 +173,11 @@ class Logger:
args.append(dumps(returns)) args.append(dumps(returns))
sql += """WHERE LogId = $1 sql += """WHERE LogId = $1
""" """
await risotto_context.log_connection.execute(sql, async with risotto_context.log_connection.transaction():
risotto_context.start_id, await risotto_context.log_connection.execute(sql,
*args, risotto_context.start_id,
) *args,
)
async def failed(self, async def failed(self,
risotto_context: Context, risotto_context: Context,
@ -190,18 +189,19 @@ class Logger:
context = f'({risotto_context.context_id})' context = f'({risotto_context.context_id})'
else: else:
context = '' context = ''
print(_(f'{risotto_context.username}: FAILED({risotto_context.context_id}):{paths_msg}: err')) print(_(f'{risotto_context.username}: FAILED({risotto_context.context_id}):{paths_msg}: {err}'))
sql = """UPDATE RisottoLog sql = """UPDATE RisottoLog
SET StopDate = $2, SET StopDate = $2,
Level = 'FAILED', Level = 'FAILED',
Msg = $3 Msg = $3
WHERE LogId = $1 WHERE LogId = $1
""" """
await risotto_context.log_connection.execute(sql, async with risotto_context.log_connection.transaction():
risotto_context.start_id, await risotto_context.log_connection.execute(sql,
datetime.now(), risotto_context.start_id,
err, datetime.now(),
) err,
)
async def info(self, async def info(self,
risotto_context, risotto_context,
@ -210,8 +210,6 @@ class Logger:
if get_config()['global']['debug']: if get_config()['global']['debug']:
print(msg) print(msg)
await self.insert(msg, await self.insert(msg,
'',
None,
risotto_context, risotto_context,
'Info', 'Info',
) )

59
src/risotto/register.py
View File

@ -298,37 +298,36 @@ class RegisterDispatcher:
) -> None: ) -> None:
internal_user = get_config()['global']['internal_user'] internal_user = get_config()['global']['internal_user']
async with self.pool.acquire() as log_connection: async with self.pool.acquire() as log_connection:
async with log_connection.transaction(): async with self.pool.acquire() as connection:
async with self.pool.acquire() as connection: await connection.set_type_codec(
await connection.set_type_codec( 'json',
'json', encoder=dumps,
encoder=dumps, decoder=loads,
decoder=loads, schema='pg_catalog'
schema='pg_catalog' )
) if truncate:
if truncate:
async with connection.transaction():
await connection.execute('TRUNCATE InfraServer, InfraSite, InfraZone, Log, ProviderDeployment, ProviderFactoryCluster, ProviderFactoryClusterNode, SettingApplicationservice, SettingApplicationServiceDependency, SettingRelease, SettingServer, SettingServermodel, SettingSource, UserRole, UserRoleURI, UserURI, UserUser, InfraServermodel, ProviderZone, ProviderServer, ProviderSource, ProviderApplicationservice, ProviderServermodel')
async with connection.transaction(): async with connection.transaction():
for submodule_name, module in self.injected_self.items(): await connection.execute('TRUNCATE InfraServer, InfraSite, InfraZone, Log, ProviderDeployment, ProviderFactoryCluster, ProviderFactoryClusterNode, SettingApplicationservice, SettingApplicationServiceDependency, SettingRelease, SettingServer, SettingServermodel, SettingSource, UserRole, UserRoleURI, UserURI, UserUser, InfraServermodel, ProviderZone, ProviderServer, ProviderSource, ProviderApplicationservice, ProviderServermodel')
risotto_context = Context() async with connection.transaction():
risotto_context.username = internal_user for submodule_name, module in self.injected_self.items():
risotto_context.paths.append(f'internal.{submodule_name}.on_join') risotto_context = Context()
risotto_context.type = None risotto_context.username = internal_user
risotto_context.log_connection = log_connection risotto_context.paths.append(f'internal.{submodule_name}.on_join')
risotto_context.connection = connection risotto_context.type = None
risotto_context.module = submodule_name.split('.', 1)[0] risotto_context.log_connection = log_connection
info_msg = _(f'in function risotto_{submodule_name}.on_join') risotto_context.connection = connection
await log.info_msg(risotto_context, risotto_context.module = submodule_name.split('.', 1)[0]
None, info_msg = _(f'in function risotto_{submodule_name}.on_join')
info_msg) await log.info_msg(risotto_context,
try: None,
await module.on_join(risotto_context) info_msg)
except Exception as err: try:
if get_config()['global']['debug']: await module.on_join(risotto_context)
print_exc() except Exception as err:
msg = _(f'on_join returns an error in module {submodule_name}: {err}') if get_config()['global']['debug']:
await log.error_msg(risotto_context, {}, msg) print_exc()
msg = _(f'on_join returns an error in module {submodule_name}: {err}')
await log.error_msg(risotto_context, {}, msg)
async def load(self): async def load(self):
# valid function's arguments # valid function's arguments