Files
distcloud/distributedcloud/dcmanager/common/messaging.py
Hugo Brito c2c7ab93ef Update endpoint caches post network reconfig
After switching the management network to the admin network
we need to update the services endpoints cache of all
dcmanager, dcorch and cert-mon workers to the new admin value.

The fanout parameter is being added to the cast calls of the
RPC clients (dcorch and audit), this parameter makes the method
cast to all servers listening on a topic rather than just one
of them.

Test Plan:
PASS: dcmanager subcloud update (using admin network parameters)
1. Endpoints for the subcloud updated with admin ip value
2. subcloud availability = online
PASS: Verify that the subcloud is online shortly after succesful
completion of subcloud_update playbook
PASS: Verify that the service endpoints are updated in all workers'
endpoint caches for the subcloud
PASS: Manage the subcloud and verify that both dcmanager and dcorch
audits are working as expected
PASS: Perform a Identity sync:
1. openstack --os-region-name SystemController user create <new_user>
--domain <domain> --project <project> --password <password>
2. Log in into subcloud and verify the new user: openstack user list
PASS: Verify that the master token is refreshed successfully after an hour

Story: 2010319
Task: 47556

Depends-On: https://review.opendev.org/c/starlingx/config/+/877323

Signed-off-by: Hugo Brito <hugo.brito@windriver.com>
Change-Id: I149c864382b7c63d424f736bdb4eaac2a787b709
2023-03-14 11:44:15 -03:00

114 lines
3.6 KiB
Python

# Copyright (c) 2017, 2019, 2021 Wind River Systems, Inc.
# 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 eventlet
from oslo_config import cfg
import oslo_messaging
from oslo_serialization import jsonutils
from dcmanager.common import context
TRANSPORT = None
NOTIFIER = None
class RequestContextSerializer(oslo_messaging.Serializer):
def __init__(self, base):
self._base = base
def serialize_entity(self, ctxt, entity):
if not self._base:
return entity
return self._base.serialize_entity(ctxt, entity)
def deserialize_entity(self, ctxt, entity):
if not self._base:
return entity
return self._base.deserialize_entity(ctxt, entity)
@staticmethod
def serialize_context(ctxt):
return ctxt.to_dict()
@staticmethod
def deserialize_context(ctxt):
return context.RequestContext.from_dict(ctxt)
class JsonPayloadSerializer(oslo_messaging.NoOpSerializer):
@classmethod
def serialize_entity(cls, context, entity):
return jsonutils.to_primitive(entity, convert_instances=True)
def setup(url=None, optional=False):
"""Initialise the oslo_messaging layer."""
global TRANSPORT, NOTIFIER
if url and url.startswith("fake://"):
# NOTE: oslo_messaging fake driver uses time.sleep
# for task switch, so we need to monkey_patch it
eventlet.monkey_patch(time=True)
if not TRANSPORT:
oslo_messaging.set_transport_defaults('dcmanager')
exmods = ['dcmanager.common.exception']
try:
TRANSPORT = oslo_messaging.get_transport(
cfg.CONF, url, allowed_remote_exmods=exmods)
except oslo_messaging.InvalidTransportURL as e:
TRANSPORT = None
if not optional or e.url:
raise
if not NOTIFIER and TRANSPORT:
serializer = RequestContextSerializer(JsonPayloadSerializer())
NOTIFIER = oslo_messaging.Notifier(TRANSPORT, serializer=serializer)
def cleanup():
"""Cleanup the oslo_messaging layer."""
global TRANSPORT, NOTIFIER
if TRANSPORT:
TRANSPORT.cleanup()
TRANSPORT = NOTIFIER = None
def get_rpc_server(target, endpoint):
"""Return a configured oslo_messaging rpc server."""
serializer = RequestContextSerializer(JsonPayloadSerializer())
return oslo_messaging.get_rpc_server(TRANSPORT, target, [endpoint],
executor='eventlet',
serializer=serializer)
def get_rpc_client(timeout, **msg_target_kwargs):
"""Return a configured oslo_messaging RPCClient."""
target = oslo_messaging.Target(**msg_target_kwargs)
serializer = RequestContextSerializer(JsonPayloadSerializer())
# With timeout == None the default value will be 60 seconds
return oslo_messaging.RPCClient(TRANSPORT, target, timeout=timeout,
serializer=serializer)
def get_transport():
"""Return a configured oslo_messaging transport."""
return TRANSPORT
def get_notifier(publisher_id):
"""Return a configured oslo_messaging notifier."""
return NOTIFIER.prepare(publisher_id=publisher_id)