Neutron Messaging Callback System¶

Rolling upgrades strategy¶

In this section we assume the standard Neutron upgrade process, which means upgrade the server first and then upgrade the agents:

More information about the upgrade strategy.

We provide an automatic method which avoids manual pinning and unpinning of versions by the administrator which could be prone to error.

'resource_versions': {'QosPolicy': '1.1',
                      'SecurityGroup': '1.0',
                      'Port': '1.0'}

def get_agents_resource_versions(self, tracker):
   ...

Topic names for every resource type RPC endpoint¶

neutron-vo-<resource_class_name>-<version>

In the future, we may want to get oslo messaging to support subscribing topics dynamically, then we may want to use:

neutron-vo-<resource_class_name>-<resource_id>-<version> instead,

or something equivalent which would allow fine granularity for the receivers to only get interesting information to them.

Subscribing to resources¶

Imagine that you have agent A, which just got to handle a new port, which has an associated security group, and QoS policy.

The agent code processing port updates may look like:

from neutron.api.rpc.callbacks.consumer import registry
from neutron.api.rpc.callbacks import events
from neutron.api.rpc.callbacks import resources


def process_resource_updates(context, resource_type, resource_list, event_type):

    # send to the right handler which will update any control plane
    # details related to the updated resources...


def subscribe_resources():
    registry.register(process_resource_updates, resources.SEC_GROUP)

    registry.register(process_resource_updates, resources.QOS_POLICY)

def port_update(port):

    # here we extract sg_id and qos_policy_id from port..

    sec_group = registry.pull(resources.SEC_GROUP, sg_id)
    qos_policy = registry.pull(resources.QOS_POLICY, qos_policy_id)

The relevant function is:

• register(callback, resource_type): subscribes callback to a resource type.

The callback function will receive the following arguments:

• context: the neutron context that triggered the notification.
• resource_type: the type of resource which is receiving the update.
• resource_list: list of resources which have been pushed by server.
• event_type: will be one of CREATED, UPDATED, or DELETED, see neutron.api.rpc.callbacks.events for details.

With the underlying oslo_messaging support for dynamic topics on the receiver we cannot implement a per “resource type + resource id” topic, rabbitmq seems to handle 10000’s of topics without suffering, but creating 100’s of oslo_messaging receivers on different topics seems to crash.

We may want to look into that later, to avoid agents receiving resource updates which are uninteresting to them.

Unsubscribing from resources¶

To unsubscribe registered callbacks:

• unsubscribe(callback, resource_type): unsubscribe from specific resource type.
• unsubscribe_all(): unsubscribe from all resources.

Sending resource events¶

On the server side, resource updates could come from anywhere, a service plugin, an extension, anything that updates, creates, or destroys the resources and that is of any interest to subscribed agents.

A callback is expected to receive a list of resources. When resources in the list belong to the same resource type, a single push RPC message is sent; if the list contains objects of different resource types, resources of each type are grouped and sent separately, one push RPC message per type. On the receiver side, resources in a list always belong to the same type. In other words, a server-side push of a list of heterogeneous objects will result into N messages on bus and N client-side callback invocations, where N is the number of unique resource types in the given list, e.g. L(A, A, B, C, C, C) would be fragmented into L1(A, A), L2(B), L3(C, C, C), and each list pushed separately.

Note: there is no guarantee in terms of order in which separate resource lists will be delivered to consumers.

The server/publisher side may look like:

from neutron.api.rpc.callbacks.producer import registry
from neutron.api.rpc.callbacks import events

def create_qos_policy(...):
    policy = fetch_policy(...)
    update_the_db(...)
    registry.push([policy], events.CREATED)

def update_qos_policy(...):
    policy = fetch_policy(...)
    update_the_db(...)
    registry.push([policy], events.UPDATED)

def delete_qos_policy(...):
    policy = fetch_policy(...)
    update_the_db(...)
    registry.push([policy], events.DELETED)

References¶